fix: is_encrypted_record misses blobs nested inside a list or tuple
both the bounded pass and the unbounded _has_encrypted_field fallback descended only through dict values, so a blob nested inside a list at any depth was invisible and the function returned False. reencrypt() already skips list-nested blobs (documented gotcha), so after rotation such a blob was stranded under the old key while this audit reported the record clean - a rotation-data-loss trap once the old wrapped-key record is deleted. both traversal passes now walk list/tuple items in addition to dict values; the blob-detection predicate is unchanged. bump 0.1.7 -> 0.1.8 Signed-off-by: disqualifier <dev@disqualifier.me>
This commit is contained in:
@@ -101,7 +101,8 @@ if is_encrypted_record(doc):
|
|||||||
`is_encrypted_record` falls back to an unbounded-depth scan once `traversal_level` is
|
`is_encrypted_record` falls back to an unbounded-depth scan once `traversal_level` is
|
||||||
exhausted, so it reliably reports `True` for a blob left behind by a shallower
|
exhausted, so it reliably reports `True` for a blob left behind by a shallower
|
||||||
`decrypt_record`/`reencrypt` call — safe to use as a leftover-detecting audit after
|
`decrypt_record`/`reencrypt` call — safe to use as a leftover-detecting audit after
|
||||||
rotation, regardless of how deep the blob is nested.
|
rotation, regardless of how deep the blob is nested, including inside a list or
|
||||||
|
tuple at any depth.
|
||||||
|
|
||||||
Naming aliases (same objects): `EnvelopeCrypto` = `DocumentCrypto` = `RecordCrypto`
|
Naming aliases (same objects): `EnvelopeCrypto` = `DocumentCrypto` = `RecordCrypto`
|
||||||
= `PCICrypto` (deprecated legacy alias). `decrypt_record` = `decrypt_document` =
|
= `PCICrypto` (deprecated legacy alias). `decrypt_record` = `decrypt_document` =
|
||||||
|
|||||||
+1
-1
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
|
|||||||
|
|
||||||
[project]
|
[project]
|
||||||
name = "envelope_crypto"
|
name = "envelope_crypto"
|
||||||
version = "0.1.7"
|
version = "0.1.8"
|
||||||
description = "Envelope encryption (RSA-OAEP wrapped AES-256-GCM) for dict records — config-free, storage-agnostic, installable."
|
description = "Envelope encryption (RSA-OAEP wrapped AES-256-GCM) for dict records — config-free, storage-agnostic, installable."
|
||||||
requires-python = ">=3.10"
|
requires-python = ">=3.10"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
|
|||||||
@@ -78,12 +78,14 @@ def _is_blob(value: Any) -> bool:
|
|||||||
|
|
||||||
|
|
||||||
def _has_encrypted_field(record: Any) -> bool:
|
def _has_encrypted_field(record: Any) -> bool:
|
||||||
"""unbounded-depth scan: does record (or anything nested under it) contain a blob"""
|
"""unbounded-depth scan: does record (or anything nested under it, incl. list/tuple items) contain a blob"""
|
||||||
if not isinstance(record, dict):
|
if isinstance(record, dict):
|
||||||
return False
|
|
||||||
if _is_blob(record):
|
if _is_blob(record):
|
||||||
return True
|
return True
|
||||||
return any(_has_encrypted_field(value) for value in record.values())
|
return any(_has_encrypted_field(value) for value in record.values())
|
||||||
|
if isinstance(record, (list, tuple)):
|
||||||
|
return any(_has_encrypted_field(item) for item in record)
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
def _require_rsa(key) -> None:
|
def _require_rsa(key) -> None:
|
||||||
@@ -502,13 +504,26 @@ RecordCrypto = EnvelopeCrypto
|
|||||||
PCICrypto = EnvelopeCrypto # deprecated legacy alias; remove after all systems migrate
|
PCICrypto = EnvelopeCrypto # deprecated legacy alias; remove after all systems migrate
|
||||||
|
|
||||||
|
|
||||||
|
def _is_encrypted_value(value: Any, traversal_level: int) -> bool:
|
||||||
|
"""bounded-pass check of a single field value, walking list/tuple items too"""
|
||||||
|
if _is_blob(value):
|
||||||
|
return True
|
||||||
|
if isinstance(value, dict):
|
||||||
|
return traversal_level > 0 and is_encrypted_record(value, traversal_level - 1)
|
||||||
|
if isinstance(value, (list, tuple)):
|
||||||
|
return any(_is_encrypted_value(item, traversal_level) for item in value)
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
def is_encrypted_record(record, traversal_level: int = 2) -> bool:
|
def is_encrypted_record(record, traversal_level: int = 2) -> bool:
|
||||||
"""return whether a record has any encrypted ({secure, iv, data}) fields
|
"""return whether a record has any encrypted ({secure, iv, data}) fields
|
||||||
|
|
||||||
checks `record` itself (the file-storage pattern) as well as fields up to
|
checks `record` itself (the file-storage pattern) as well as fields up to
|
||||||
`traversal_level` deep, then falls back to an unbounded-depth scan - safe to
|
`traversal_level` deep, then falls back to an unbounded-depth scan - safe to
|
||||||
use as a leftover-detecting post-rotation audit; never returns False for a
|
use as a leftover-detecting post-rotation audit; never returns False for a
|
||||||
record that still contains a blob, at any depth.
|
record that still contains a blob, at any depth, including one nested inside
|
||||||
|
a list or tuple at any depth (both passes walk list/tuple items, not just
|
||||||
|
dict values).
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
record: the record (or bare blob) to check.
|
record: the record (or bare blob) to check.
|
||||||
@@ -532,7 +547,7 @@ def is_encrypted_record(record, traversal_level: int = 2) -> bool:
|
|||||||
|
|
||||||
if traversal_level > 0:
|
if traversal_level > 0:
|
||||||
for value in record.values():
|
for value in record.values():
|
||||||
if isinstance(value, dict) and is_encrypted_record(value, traversal_level - 1):
|
if _is_encrypted_value(value, traversal_level - 1):
|
||||||
return True
|
return True
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user