diff --git a/README.md b/README.md index 6ac2cb3..4510707 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.8 +envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.10 ``` Direct: ```bash -pip install "envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.8" +pip install "envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.10" ``` Requires `cryptography` (pulled transitively). -Drop the `@v0.1.8` suffix from the line above to install the latest unpinned. +Drop the `@v0.1.10` suffix from the line above to install the latest unpinned. ## First-time setup @@ -80,10 +80,12 @@ enc = crypto.encrypt_data({"ssn": "..."}) # -> {"secure": True, "iv": ..., plain = crypto.decrypt_data(enc) # -> {"ssn": "..."} ``` -Dict keys must be `str`. `encrypt_data` raises `TypeError` on a non-str key (e.g. an -int-keyed dict of Discord snowflakes) instead of silently stringifying it — the -underlying JSON encoding has no other key type, so a coerced key would come back out -of `decrypt_data` as a `str` and no longer match the original lookup key. +Dict keys must be `str`, at any nesting depth (including a dict nested inside a list +or tuple). `encrypt_data` raises `TypeError` on a non-str key anywhere in the payload +(e.g. an int-keyed dict of Discord snowflakes, even nested a few levels down) instead +of silently stringifying it — the underlying JSON encoding has no other key type, so a +coerced key would come back out of `decrypt_data` as a `str` and no longer match the +original lookup key. 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 @@ -98,11 +100,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 +`is_encrypted_record` always falls back to an unbounded-depth scan whenever its bounded +pass finds nothing, 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, including inside a list or -tuple at any depth. +rotation, regardless of how deep the blob is nested (including inside a list or tuple +at any depth) and regardless of which `traversal_level` you pass; it never false-negatives. Naming aliases (same objects): `EnvelopeCrypto` = `DocumentCrypto` = `RecordCrypto` = `PCICrypto` (deprecated legacy alias). `decrypt_record` = `decrypt_document` = diff --git a/pyproject.toml b/pyproject.toml index 4ec1b15..730ab9c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "envelope_crypto" -version = "0.1.9" +version = "0.1.10" 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 a189afd..be4d29c 100644 --- a/src/envelope_crypto/envelope_crypto.py +++ b/src/envelope_crypto/envelope_crypto.py @@ -94,6 +94,20 @@ def _has_encrypted_field(record: Any) -> bool: return False +def _non_str_key_types(data: Any) -> List[str]: + """unbounded-depth scan collecting type names of non-str dict keys, incl. dicts nested inside list/tuple items""" + found: List[str] = [] + if isinstance(data, dict): + for key, value in data.items(): + if not isinstance(key, str): + found.append(type(key).__name__) + found.extend(_non_str_key_types(value)) + elif isinstance(data, (list, tuple)): + for item in data: + found.extend(_non_str_key_types(item)) + return found + + def _require_rsa(key) -> None: """raise ValueError unless key is an RSA public or private key""" if not isinstance(key, (rsa.RSAPublicKey, rsa.RSAPrivateKey)): @@ -390,9 +404,10 @@ class EnvelopeCrypto: def encrypt_data(self, data: Union[Dict[str, Any], str]) -> Dict[str, str]: """encrypt a dict or string under the data key with a unique IV - dict keys must be str: json.dumps silently stringifies int/float/bool/None - keys (e.g. a snowflake-int-keyed dict), which would make decrypt_data return - a dict that no longer matches the original by key type or identity - a + dict keys must be str, at any nesting depth (including dicts nested inside + lists/tuples): json.dumps silently stringifies int/float/bool/None keys + (e.g. a snowflake-int-keyed dict), which would make decrypt_data return a + dict that no longer matches the original by key type or identity - a non-str key is rejected here instead, so the failure is loud at encrypt time rather than a silent lookup miss after decrypt. """ @@ -401,11 +416,11 @@ class EnvelopeCrypto: if not isinstance(data, (dict, str)): raise TypeError(f"encrypt_data expects a dict or str, got {type(data).__name__}") if isinstance(data, dict): - non_str_keys = [key for key in data if not isinstance(key, str)] - if non_str_keys: + non_str_key_types = _non_str_key_types(data) + if non_str_key_types: raise TypeError( - "encrypt_data requires str dict keys, got non-str key(s): " - f"{[type(key).__name__ for key in non_str_keys]}" + "encrypt_data requires str dict keys (at any nesting depth), got non-str " + f"key(s): {non_str_key_types}" ) data_str = json.dumps(data) if isinstance(data, dict) else data @@ -509,11 +524,18 @@ PCICrypto = EnvelopeCrypto # deprecated legacy alias; remove after all systems def _is_encrypted_value(value: Any, traversal_level: int) -> bool: - """bounded-pass check of a single field value, walking list/tuple items too""" + """bounded-pass check of a single field value, walking list/tuple items too + + falls back to the unbounded _has_encrypted_field scan once traversal_level is + exhausted, so this can never return False for a value that still contains a + blob at any depth + """ if _is_blob(value): return True if isinstance(value, dict): - return traversal_level > 0 and is_encrypted_record(value, traversal_level - 1) + if traversal_level > 0: + return is_encrypted_record(value, traversal_level - 1) + return _has_encrypted_field(value) if isinstance(value, (list, tuple)): return any(_is_encrypted_value(item, traversal_level) for item in value) return False @@ -523,16 +545,18 @@ 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) as well as fields up 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 - 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). + `traversal_level` deep, then always falls back to an unbounded-depth scan if + the bounded pass found nothing - safe to use as a leftover-detecting + post-rotation audit; never returns False for a record that still contains a + blob, at any depth (including one nested inside a list or tuple), for any + traversal_level value, odd or even (both passes walk list/tuple items, not + just dict values). Args: record: the record (or bare blob) to check. - traversal_level: bounded-pass nesting depth before the unbounded - fallback scan (default 2). + traversal_level: bounded-pass nesting depth tried before the unbounded + fallback scan runs (default 2); purely a fast-path - the fallback + always makes the final call when the bounded pass finds nothing. Returns: True if record or anything nested under it is an encrypted blob. @@ -553,7 +577,6 @@ def is_encrypted_record(record, traversal_level: int = 2) -> bool: for value in record.values(): if _is_encrypted_value(value, traversal_level - 1): return True - return False return any(_has_encrypted_field(value) for value in record.values())