diff --git a/README.md b/README.md index c655a06..1672611 100644 --- a/README.md +++ b/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 diff --git a/pyproject.toml b/pyproject.toml index 78aaa14..bac72f6 100644 --- a/pyproject.toml +++ b/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 = [ diff --git a/src/envelope_crypto/envelope_crypto.py b/src/envelope_crypto/envelope_crypto.py index a7b604a..1e718c2 100644 --- a/src/envelope_crypto/envelope_crypto.py +++ b/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: