fix: reencrypt/is_encrypted_record/decrypt_record miss self-blob and deep-nested records

Two silent-data-loss paths in the record-level functions: (1) reencrypt's
traversal_level cutoff silently left blobs nested deeper than the default
under the old key with no signal, contradicting its own documented fail-loud
rotation contract, and is_encrypted_record shared the cutoff so a
post-rotation audit couldn't detect the leftover; (2) all three record
functions inspected only record.values(), never the record itself, so a bare
{secure, iv, data} blob used as the whole document (the README's file-storage
pattern) was invisible to is_encrypted_record and passed through reencrypt
unchanged under the old key.

reencrypt now raises when a blob sits deeper than traversal_level instead of
silently truncating, and detects/handles the record-itself-is-a-blob case;
is_encrypted_record falls back to an unbounded-depth scan past
traversal_level so it reliably flags leftovers regardless of nesting depth;
decrypt_record likewise handles a record that is itself a blob. Bumped to
v0.1.4.

Signed-off-by: disqualifier <dev@disqualifier.me>

README.md

@@ -11,18 +11,18 @@ and storage-agnostic.
 `requirements.txt`:
 
 ```
-envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.3
+envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.4
 ```
 
 Direct:
 
 ```bash
-pip install "envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.3"
+pip install "envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.4"
 ```
 
 Requires `cryptography` (pulled transitively).
 
-Drop the `@v0.1.3` suffix from the line above to install the latest unpinned.
+Drop the `@v0.1.4` suffix from the line above to install the latest unpinned.
 
 ## First-time setup
 
@@ -75,7 +75,9 @@ plain = crypto.decrypt_data(enc)               # -> {"ssn": "..."}
 
 For whole records: `decrypt_record(crypto, doc)` decrypts every `{secure, iv, data}`
 field (nested up to `traversal_level`, default 2); `is_encrypted_record(doc)` reports
-whether any encrypted field exists.
+whether any encrypted field exists. Both also detect `doc` itself being a bare
+`{secure, iv, data}` blob (the file-storage pattern below, where the blob IS the whole
+document) — not just blobs nested under a key.
 
 ```python
 from envelope_crypto import is_encrypted_record, decrypt_record
@@ -84,6 +86,11 @@ if is_encrypted_record(doc):
     doc = decrypt_record(crypto, doc)
 ```
 
+`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
+`decrypt_record`/`reencrypt` call — safe to use as a leftover-detecting audit after
+rotation, regardless of how deep the blob is nested.
+
 Naming aliases (same objects): `EnvelopeCrypto` = `DocumentCrypto` = `RecordCrypto`
 = `PCICrypto` (deprecated legacy alias). `decrypt_record` = `decrypt_document` =
 `decrypt_dict`; `is_encrypted_record` = `is_encrypted_document` = `is_encrypted_dict`.
@@ -129,7 +136,14 @@ for fingerprint, wrapped_key in wrapped.items():
 
 `reencrypt(source_crypto, record)` is a method on the **destination** (new-key)
 instance: it decrypts each encrypted field with `source_crypto` (old key) and
-re-encrypts with itself. Only `{secure, ...}` fields are touched.
+re-encrypts with itself. Only `{secure, ...}` fields are touched — including `record`
+itself if it IS a `{secure, iv, data}` blob (the file-storage pattern).
+
+Rotation must fail loud: a per-field decrypt failure raises, and so does a blob nested
+deeper than `traversal_level` — silently leaving it under the old key would strand it
+once the old key's wrapped-key record is deleted below. If you nest blobs deeper than
+the default `traversal_level=2`, pass a higher `traversal_level` or flatten the record
+first.
 
 ## Storage patterns
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "envelope_crypto"
-version = "0.1.3"
+version = "0.1.4"
 description = "Envelope encryption (RSA-OAEP wrapped AES-256-GCM) for dict records — config-free, storage-agnostic, installable."
 requires-python = ">=3.10"
 dependencies = [

src/envelope_crypto/envelope_crypto.py

@@ -44,6 +44,13 @@ set, then re-encrypt existing records old -> new:
     for record in caller_iter():
         caller_update(new_crypto.reencrypt(crypto, record))
 
+reencrypt/is_encrypted_record/decrypt_record all detect a bare {secure, iv, data}
+blob used AS the whole record (the file-storage pattern), not just blobs nested
+under a key. reencrypt fails loud (raises) rather than silently leaving a field
+under the old key — including when a blob sits deeper than `traversal_level`;
+is_encrypted_record falls back to an unbounded-depth scan past traversal_level so
+it reliably catches leftovers as a post-rotation audit.
+
 config-free: the host supplies the DEK and RSA key paths; this lib never imports
 config, configures logging, or touches a database. storage-agnostic — the
 encrypted blob is a plain dict; store it in mongo, a sql json column, or a file.
@@ -113,6 +120,24 @@ def _fingerprint_of(public_key) -> str:
     return base64.b64encode(digest.finalize()).decode()
 
 
+def _is_blob(value: Any) -> bool:
+    """return whether value has the {secure, iv, data} encrypted-blob shape"""
+    return isinstance(value, dict) and value.get("secure") is True and "iv" in value and "data" in value
+
+
+def _has_encrypted_field(record: Any) -> bool:
+    """unbounded-depth scan: does record (or anything nested under it) contain a blob
+
+    used to detect a blob left behind by a depth-limited traversal — no traversal_level
+    cutoff here, since the whole point is to catch what a bounded pass would miss.
+    """
+    if not isinstance(record, dict):
+        return False
+    if _is_blob(record):
+        return True
+    return any(_has_encrypted_field(value) for value in record.values())
+
+
 def _load_public_key(key_data: bytes):
     """load a PEM or OpenSSH public key, normalizing non-key input to ValueError
 
@@ -373,25 +398,42 @@ class EnvelopeCrypto:
         key. only {secure, iv, data} fields are touched; plaintext fields are left
         as-is. returns a new dict; the input is not mutated. used during rotation.
 
+        if `record` itself is a {secure, iv, data} blob (the file-storage pattern, where
+        the blob IS the whole document) it is re-encrypted directly and the result is
+        returned in place of `record` — not nested under a key.
+
         unlike decrypt_record (which logs a failed field and leaves it encrypted), a
         per-field decrypt failure here RAISES — rotation must fail loud, since silently
-        keeping a field under the old key would lose it once the old key is retired.
+        keeping a field under the old key would lose it once the old key is retired. for
+        the same reason, a blob nested DEEPER than `traversal_level` also RAISES instead
+        of being silently left under the old key: the caller either needs a higher
+        `traversal_level` or must flatten the record before rotation.
 
-        traversal recurses into nested DICTS only (up to `traversal_level`); a blob nested
-        inside a LIST is NOT re-encrypted. records in this scheme key blobs by field name,
-        not inside arrays, so this doesn't arise in practice — but if you store
-        list-nested blobs, flatten them to dict fields before rotation or they'll be left
-        under the old key.
+        traversal recurses into nested DICTS only; a blob nested inside a LIST is NOT
+        re-encrypted and is NOT covered by the depth-limit raise above. records in this
+        scheme key blobs by field name, not inside arrays, so this doesn't arise in
+        practice — but if you store list-nested blobs, flatten them to dict fields before
+        rotation or they'll be silently left under the old key.
         """
         if not self.master_key:
             raise ValueError("destination not initialized with data key")
 
+        if _is_blob(record):
+            return self.encrypt_data(source_crypto.decrypt_data(record))
+
         result = copy.deepcopy(record)
         for key, value in record.items():
-            if isinstance(value, dict) and value.get("secure") is True and "iv" in value and "data" in value:
+            if _is_blob(value):
                 result[key] = self.encrypt_data(source_crypto.decrypt_data(value))
-            elif traversal_level > 0 and isinstance(value, dict):
-                result[key] = self.reencrypt(source_crypto, value, traversal_level - 1)
+            elif isinstance(value, dict):
+                if traversal_level > 0:
+                    result[key] = self.reencrypt(source_crypto, value, traversal_level - 1)
+                elif _has_encrypted_field(value):
+                    raise ValueError(
+                        f"reencrypt: field {key!r} contains an encrypted blob nested deeper "
+                        "than traversal_level; increase traversal_level or flatten the record "
+                        "before rotation — leaving it would strand the field under the old key"
+                    )
         return result
 
 
@@ -404,28 +446,46 @@ PCICrypto = EnvelopeCrypto  # deprecated legacy alias; remove after all systems
 def is_encrypted_record(record, traversal_level: int = 2) -> bool:
     """return whether a record has any encrypted ({secure, iv, data}) fields
 
+    checks `record` itself (the file-storage pattern stores the blob AS the whole
+    document, not nested under a key) as well as fields up to `traversal_level` deep.
+    beyond that bounded pass, this ALSO does an unbounded-depth scan before giving up —
+    so a blob left behind by a shallower decrypt_record/reencrypt call (nested deeper
+    than their traversal_level) is still reported as encrypted. this makes the function
+    safe to use as a leftover-detecting post-rotation audit: it never returns False for
+    a record that still contains a blob, at any depth.
+
     aliases: is_encrypted_document, is_encrypted_dict — same function
     """
     if not isinstance(record, dict):
         return False
 
+    if _is_blob(record):
+        return True
+
     for value in record.values():
-        if isinstance(value, dict) and value.get("secure") is True:
-            if "iv" in value and "data" in value:
-                return True
+        if _is_blob(value):
+            return True
 
     if traversal_level > 0:
         for value in record.values():
             if isinstance(value, dict) and is_encrypted_record(value, traversal_level - 1):
                 return True
-    return False
+        return False
+
+    return any(_has_encrypted_field(value) for value in record.values())
 
 
-def decrypt_record(crypto: EnvelopeCrypto, record, traversal_level: int = 2) -> dict:
+def decrypt_record(crypto: EnvelopeCrypto, record, traversal_level: int = 2) -> Union[dict, Any]:
     """decrypt a record's encrypted fields into a new dict (up to traversal_level deep)
 
+    if `record` itself is a {secure, iv, data} blob (the file-storage pattern, where the
+    blob IS the whole document) it is decrypted directly and the decrypted value
+    (dict or string — see decrypt_data) is returned in place of `record`.
+
     failures on a single field are logged and that field is left encrypted, so a
-    partial failure is visible (the {secure,...} blob remains) rather than silent.
+    partial failure is visible (the {secure,...} blob remains) rather than silent. a
+    failure decrypting `record` itself (the self-blob case above) is likewise logged
+    and the still-encrypted blob is returned unchanged.
 
     aliases: decrypt_document, decrypt_dict — same function
     """
@@ -434,9 +494,16 @@ def decrypt_record(crypto: EnvelopeCrypto, record, traversal_level: int = 2) ->
     if not isinstance(record, dict):
         return record
 
+    if _is_blob(record):
+        try:
+            return crypto.decrypt_data(record)
+        except Exception:
+            _log.exception("failed to decrypt record")
+            return copy.deepcopy(record)
+
     result = copy.deepcopy(record)
     for key, value in record.items():
-        if isinstance(value, dict) and value.get("secure") is True and "iv" in value and "data" in value:
+        if _is_blob(value):
             try:
                 result[key] = crypto.decrypt_data(value)
             except Exception: