diff --git a/README.md b/README.md index ada05ea..cefdd79 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.6 +envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.7 ``` Direct: ```bash -pip install "envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.6" +pip install "envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.7" ``` Requires `cryptography` (pulled transitively). -Drop the `@v0.1.6` suffix from the line above to install the latest unpinned. +Drop the `@v0.1.7` suffix from the line above to install the latest unpinned. ## First-time setup diff --git a/pyproject.toml b/pyproject.toml index 9483705..cacad3c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "envelope_crypto" -version = "0.1.6" +version = "0.1.7" 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 38a7cc2..30e8cd3 100644 --- a/src/envelope_crypto/envelope_crypto.py +++ b/src/envelope_crypto/envelope_crypto.py @@ -2,12 +2,11 @@ envelope encryption for dict records hybrid encryption: a random AES-256-GCM data key (DEK) encrypts the data, wrapped -(RSA-OAEP) per authorized system's public key (KEK) for distribution. the wrapped -key is stored by the caller, keyed by fingerprint; each system unwraps its own copy -with its private key. same pattern KMS-style systems use. RSA-envelope only — a -non-RSA key (e.g. Ed25519/EC) loads and fingerprints fine but raises ValueError at -wrap/unwrap. never logs key material (DEK, PEM, wrapped key) — only fingerprints -and counts. +(RSA-OAEP) per authorized system's public key (KEK) for distribution and stored by +the caller, keyed by fingerprint. RSA-envelope only - a non-RSA key (e.g. Ed25519/EC) +loads and fingerprints fine but raises ValueError at wrap/unwrap. never logs key +material (DEK, PEM, wrapped key) - only fingerprints and counts. config-free and +storage-agnostic; see README for bootstrap/boot/authorize/rotate flows. from envelope_crypto import EnvelopeCrypto @@ -16,54 +15,9 @@ and counts. enc = crypto.encrypt_data({"ssn": "..."}) # -> {secure, iv, data} plain = crypto.decrypt_data(enc) # -> original -first-time setup: generate the DEK and wrap it for the first system in one call, -then verify the pipeline before storing anything: - - crypto, fingerprint, wrapped = EnvelopeCrypto.bootstrap("public_key.pem") - crypto.self_test("public_key.pem", "private_key.pem") # raises if anything is wrong - caller_store({"_id": fingerprint, "key": wrapped}) # the only record of the DEK - -boot (already set up): fingerprint own pubkey, fetch the wrapped DEK, unwrap: - - fp = crypto.get_rsa_key_fingerprint("public_key.pem") - record = caller_lookup(fp) - crypto.initialize(crypto.decrypt_aes_key_with_rsa(record["key"], "private_key.pem")) - -authorize another system (this instance must already hold the DEK): - - fp, wrapped = crypto.authorize_system(other_pub_path) - caller_store({"_id": fp, "key": wrapped}) - -deauthorize: caller deletes that fingerprint's record. stops future unwraps but -does not revoke a DEK already in a running system's memory — rotate if compromised. - -rotate (new DEK + re-encrypt): generate a new DEK, wrap for the still-authorized -set, then re-encrypt existing records old -> new: - - new_key, wrapped = crypto.rotate_master_key([pub_a, pub_b]) - new_crypto = EnvelopeCrypto(); new_crypto.initialize(new_key) - 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 (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 a blob nested deeper than `traversal_level`; is_encrypted_record -falls back to an unbounded-depth scan past traversal_level to reliably catch -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. - -encrypt_data/decrypt_data round-trip a dict only when every key is a str — json -(the wire format) has no other key type, so encrypt_data raises TypeError on a -non-str key (e.g. an int-keyed dict of discord snowflakes) rather than silently -stringifying it and losing the original key on decrypt. - naming: EnvelopeCrypto is canonical. PCICrypto / DocumentCrypto / RecordCrypto are aliases (PCICrypto is a deprecated legacy alias). the document/record/dict function -variants are the same functions — use whichever fits your storage. +variants are the same functions - use whichever fits your storage. """ import os @@ -84,24 +38,14 @@ _log = logging.getLogger(__name__) def _password_mismatch_message(pw: Optional[bytes]) -> str: - """clear ValueError text for a TypeError raised by a password/encryption mismatch - - cryptography raises TypeError for both directions: encrypted key + no password, - and unencrypted key + a password given. branch on which the caller supplied so - the message matches the actual case instead of always claiming "encrypted". - """ + """clear ValueError text for a TypeError raised by a password/encryption mismatch""" if pw is not None: return "password was given but private key is not encrypted" return "private key is encrypted but no password was provided" def _load_private_key(key_data: bytes, pw: Optional[bytes]): - """load a PEM or OpenSSH private key, normalizing the password/encryption mismatch error - - cryptography raises TypeError for a password/encryption mismatch (PEM raises it on - the first call; OpenSSH raises it inside the openssh fallback), normalized here to a - clear ValueError so callers see one error type with a message matching the actual case. - """ + """load a PEM or OpenSSH private key, normalizing the password/encryption mismatch error""" try: return serialization.load_pem_private_key(key_data, password=pw) except ValueError as error: @@ -111,8 +55,6 @@ def _load_private_key(key_data: bytes, pw: Optional[bytes]): except TypeError as ssh_error: raise ValueError(_password_mismatch_message(pw)) from ssh_error if pw is not None: - # a password was given but the PEM load still failed — most likely a wrong - # password; give a clearer message than cryptography's raw "Bad decrypt" raise ValueError("could not load private key (wrong password or malformed key)") from error raise error except TypeError as error: @@ -120,11 +62,7 @@ def _load_private_key(key_data: bytes, pw: Optional[bytes]): def _fingerprint_of(public_key) -> str: - """base64 SHA-256 fingerprint of an already-loaded public key - - factored so callers that already hold a loaded key (e.g. encrypt_aes_key_with_rsa) - don't re-open and re-parse the key file just to fingerprint it. - """ + """base64 SHA-256 fingerprint of an already-loaded public key""" key_bytes = public_key.public_bytes( encoding=serialization.Encoding.DER, format=serialization.PublicFormat.SubjectPublicKeyInfo, @@ -140,11 +78,7 @@ def _is_blob(value: Any) -> bool: 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. - """ + """unbounded-depth scan: does record (or anything nested under it) contain a blob""" if not isinstance(record, dict): return False if _is_blob(record): @@ -153,24 +87,13 @@ def _has_encrypted_field(record: Any) -> bool: def _require_rsa(key) -> None: - """raise ValueError unless key is an RSA public or private key - - this lib is RSA-envelope only: get_rsa_key_fingerprint accepts any key type - (fingerprinting is algorithm-agnostic), but wrap/unwrap calls RSA-OAEP methods - that don't exist on e.g. Ed25519/EC keys and would otherwise crash raw with an - AttributeError far from a clear cause. - """ + """raise ValueError unless key is an RSA public or private key""" if not isinstance(key, (rsa.RSAPublicKey, rsa.RSAPrivateKey)): raise ValueError(f"RSA key required for envelope wrap/unwrap, got {type(key).__name__}") def _load_public_key(key_data: bytes): - """load a PEM or OpenSSH public key, normalizing non-key input to ValueError - - load_ssh_public_key raises UnsupportedAlgorithm (not ValueError) on non-SSH/garbage - input; normalize it so a bad public key always surfaces as a clear ValueError, - consistent with the private-key path. - """ + """load a PEM or OpenSSH public key, normalizing non-key input to ValueError""" try: return serialization.load_pem_public_key(key_data) except ValueError: @@ -195,10 +118,20 @@ class EnvelopeCrypto: def bootstrap(cls, rsa_public_key: str, is_file: bool = True) -> Tuple["EnvelopeCrypto", str, str]: """first-time setup: generate a DEK and wrap it for the first system - returns (crypto, fingerprint, wrapped_key) — an initialized instance plus - the record to store as the first authorization. the plaintext DEK is never - returned or persisted; it survives only as the wrapped copy. run self_test - before storing to confirm the keypair round-trips. + the plaintext DEK is never returned or persisted; it survives only as the + wrapped copy. run self_test before storing to confirm the keypair round-trips. + + Args: + rsa_public_key: path to (or, if is_file=False, raw PEM/OpenSSH data of) + the first system's RSA public key. + is_file: True (default) treats rsa_public_key as a path. + + Returns: + (crypto, fingerprint, wrapped_key): an initialized instance plus the + record to store as the first authorization. + + Raises: + ValueError: rsa_public_key is not an RSA key, or is malformed. """ crypto = cls() crypto.initialize(crypto.create_aes_key()) @@ -212,10 +145,23 @@ class EnvelopeCrypto: """verify the full pipeline against a keypair; raises on any mismatch round-trips sample data through this instance's DEK, then wraps and unwraps - the DEK with the given keypair, confirming they match. run after bootstrap - (or as a health check) to catch a bad keypair or wrong path before relying - on it. is_file threads through to both key loads (is_file=False treats both - as in-memory PEM/OpenSSH data, not paths). returns True on success. + the DEK with the given keypair. run after bootstrap or as a health check. + + Args: + rsa_public_key: public half of the keypair to verify (path or, if + is_file=False, raw PEM/OpenSSH data). + rsa_private_key: private half of the keypair to verify. + is_file: True (default) treats both keys as paths; False treats both + as in-memory PEM/OpenSSH data. + password: password for an encrypted private key, if any. + + Returns: + True on success. + + Raises: + ValueError: instance not initialized with a data key. + RuntimeError: data round-trip fails, or the keypair does not pair + (unwrap fails or the recovered key mismatches). """ if not self.master_key: raise ValueError("self_test: not initialized with a key") @@ -243,7 +189,7 @@ class EnvelopeCrypto: requires exactly 32 bytes (AES-256): a shorter key would silently downgrade to AES-128/192 with no warning, and a non-bytes value would otherwise fail - late and opaquely at first encrypt/decrypt — both rejected here instead. + late and opaquely at first encrypt/decrypt - both rejected here instead. never logs the key material itself. """ if not isinstance(master_key, bytes) or len(master_key) != 32: @@ -273,11 +219,23 @@ class EnvelopeCrypto: ) -> str: """return a base64 SHA-256 fingerprint of an RSA key for identification - for an encrypted private key (is_private=True), pass its `password`; an - unencrypted key ignores it. always fingerprints the public half, so a - private key and its public key match. PEM and OpenSSH accepted (mirrors - decrypt_aes_key_with_rsa). a password/encryption mismatch raises a clear - ValueError (cryptography's raw TypeError normalized here). + always fingerprints the public half, so a private key and its public key + match. PEM and OpenSSH accepted (mirrors decrypt_aes_key_with_rsa). + + Args: + key_path_or_data: path to (or, if is_file=False, raw PEM/OpenSSH data + of) the key. + is_private: fingerprint the public half of a private key; pass its + `password` if encrypted (an unencrypted key ignores it). + is_file: True (default) treats key_path_or_data as a path. + password: password for an encrypted private key, if is_private. + + Returns: + base64 SHA-256 fingerprint. + + Raises: + ValueError: key is malformed, or password/encryption mismatch + (cryptography's raw TypeError normalized here). """ if is_file: with open(key_path_or_data, "rb") as key_file: @@ -325,7 +283,6 @@ class EnvelopeCrypto: label=None, ), ) - # fingerprint from the already-loaded public_key — no second open/parse of the file fingerprint = _fingerprint_of(public_key) wrapped_b64 = base64.b64encode(wrapped).decode() _log.info("wrapped data key for fingerprint %s", fingerprint[:8]) @@ -337,10 +294,20 @@ class EnvelopeCrypto: ) -> bytes: """unwrap an AES key with an RSA private key - is_file defaults to True (rsa_private_key is a path), matching - encrypt_aes_key_with_rsa / get_rsa_key_fingerprint; pass is_file=False to - supply the PEM/OpenSSH key data directly (e.g. an in-memory or vault-sourced - key) instead of a file path. + Args: + encrypted_key_base64: the RSA-OAEP wrapped key, base64-encoded. + rsa_private_key: path to (or, if is_file=False, raw PEM/OpenSSH data + of) the private key to unwrap with. + is_file: True (default) treats rsa_private_key as a path; pass False + to supply PEM/OpenSSH key data directly (e.g. vault-sourced). + password: password for an encrypted private key, if any. + + Returns: + the unwrapped AES data key. + + Raises: + ValueError: rsa_private_key is not an RSA key, is malformed, or a + password/encryption mismatch. """ if is_file: with open(rsa_private_key, "rb") as key_file: @@ -366,9 +333,21 @@ class EnvelopeCrypto: def authorize_system(self, rsa_public_key: str, is_file: bool = True) -> Tuple[str, str]: """wrap the current data key for another system's public key - returns (fingerprint, wrapped_b64) for the caller to store as that - system's key-authorization record. requires this instance to already - hold the data key — only an authorized system can authorize others. + requires this instance to already hold the data key - only an + authorized system can authorize others. + + Args: + rsa_public_key: path to (or, if is_file=False, raw PEM/OpenSSH data + of) the other system's RSA public key. + is_file: True (default) treats rsa_public_key as a path. + + Returns: + (fingerprint, wrapped_b64) for the caller to store as that system's + key-authorization record. + + Raises: + ValueError: this instance is not initialized, or rsa_public_key is + not an RSA key. """ if not self.master_key: raise ValueError("cannot authorize another system: not initialized") @@ -379,9 +358,20 @@ class EnvelopeCrypto: ) -> Tuple[bytes, Dict[str, str]]: """generate a NEW data key and wrap it for each authorized public key - returns (new_key, {fingerprint: wrapped_b64}). does NOT re-encrypt existing - data — build a new instance with the new key and call reencrypt() on each - record. systems not in the list get no wrapped copy (deauthorized). + does NOT re-encrypt existing data - build a new instance with the new key + and call reencrypt() on each record. systems not in the list get no + wrapped copy (deauthorized). + + Args: + authorized_public_keys: RSA public keys (paths, or raw data if + is_file=False) to wrap the new key for. + is_file: True (default) treats each entry as a path. + + Returns: + (new_key, {fingerprint: wrapped_b64}). + + Raises: + ValueError: any entry in authorized_public_keys is not an RSA key. """ new_key = self.create_aes_key() wrapped = {} @@ -396,22 +386,17 @@ class EnvelopeCrypto: 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 + 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. """ if not self.master_key: raise ValueError("not initialized with data key") if not isinstance(data, (dict, str)): - # a non-dict/non-str would .encode()-fail with an opaque AttributeError below; - # reject it clearly (decrypt_data only round-trips dict or str anyway) 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: - # json.dumps would silently stringify these (int/float/bool/None keys), - # corrupting the round-trip (decrypt_data would return a dict keyed by - # the stringified value) — fail loud instead of coercing raise TypeError( "encrypt_data requires str dict keys, got non-str key(s): " f"{[type(key).__name__ for key in non_str_keys]}" @@ -430,20 +415,25 @@ class EnvelopeCrypto: def decrypt_data(self, encrypted_data: Dict[str, str]) -> Union[Dict[str, Any], str]: """decrypt a {secure, iv, data} blob; returns the original dict or string - encrypt_data only json-encodes dicts (a string is stored verbatim), so decrypt - treats a json-OBJECT plaintext as a dict and everything else as a raw string — - a json-shaped but non-object string ('123', 'true', '[1,2]') round-trips as a - STRING, not int/bool/list. one irreducible ambiguity: a string whose exact - value is a json object ('{"a":1}') decrypts to a dict, indistinguishable - without a type marker from a stored dict — don't store a bare json-object - string if you need it back as a string. dict keys round-trip faithfully - because encrypt_data requires str keys (json's only key type). + a json-shaped but non-object plaintext ('123', 'true', '[1,2]') round-trips + as a STRING, not int/bool/list. one irreducible ambiguity: a string whose + exact value is a json object ('{"a":1}') decrypts to a dict, indistinguishable + from a stored dict - don't store a bare json-object string if you need it + back as a string. + + Args: + encrypted_data: a {secure, iv, data} blob from encrypt_data. + + Returns: + the original dict or string. + + Raises: + ValueError: instance not initialized, or encrypted_data is not a + {secure, iv, data} blob. """ if not self.master_key: raise ValueError("not initialized with data key") if not isinstance(encrypted_data, dict) or "iv" not in encrypted_data or "data" not in encrypted_data: - # a structurally-malformed blob would raise a raw KeyError/TypeError; surface - # a clear ValueError instead, matching the documented {secure, iv, data} shape raise ValueError("decrypt_data expects a {secure, iv, data} blob") iv = base64.b64decode(encrypted_data["iv"]) @@ -459,22 +449,30 @@ class EnvelopeCrypto: def reencrypt(self, source_crypto: "EnvelopeCrypto", record: dict, traversal_level: int = 2) -> dict: """re-encrypt a record's encrypted fields from source_crypto's key to this one's - self holds the destination (new) key; source_crypto holds the source (old) 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 bare blob (file-storage pattern) it is re-encrypted directly and - returned in place of `record`, not nested under a key. + self holds the destination (new) key; source_crypto holds the source (old) + key. only {secure, iv, data} fields are touched; plaintext fields are left + as-is. used during rotation. if `record` itself is a bare blob (file-storage + pattern) it is re-encrypted directly and returned in place of `record`. - fails loud, unlike decrypt_record: a per-field decrypt failure RAISES (silently - keeping a field under the old key would lose it once that key is retired), and - so does a blob nested DEEPER than `traversal_level` — raise a higher - traversal_level or flatten the record before rotation instead. + traversal recurses into nested DICTS only - a blob nested inside a LIST is + not re-encrypted and not covered by the depth-limit raise below; flatten + list-nested blobs to dict fields before rotation or they'll be silently + left under the old key. - traversal recurses into nested DICTS only; a blob nested inside a LIST is not - re-encrypted and not covered by the depth-limit raise. this scheme keys blobs - by field name, not inside arrays, so this shouldn't arise in practice — but - flatten list-nested blobs to dict fields before rotation or they'll be - silently left under the old key. + Args: + source_crypto: instance holding the old (source) key. + record: the record to re-encrypt. not mutated; a new dict is returned. + traversal_level: max nesting depth to recurse into (default 2). + + Returns: + a new dict with encrypted fields re-wrapped under this instance's key. + + Raises: + ValueError: destination not initialized with a data key; a per-field + decrypt failure (unlike decrypt_record, this fails loud rather + than silently stranding a field under the old key); or a blob + nested deeper than `traversal_level` - raise traversal_level or + flatten the record before rotation instead. """ if not self.master_key: raise ValueError("destination not initialized with data key") @@ -493,12 +491,12 @@ class EnvelopeCrypto: 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" + "before rotation - leaving it would strand the field under the old key" ) return result -# naming aliases — same class +# naming aliases - same class DocumentCrypto = EnvelopeCrypto RecordCrypto = EnvelopeCrypto PCICrypto = EnvelopeCrypto # deprecated legacy alias; remove after all systems migrate @@ -507,14 +505,20 @@ 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) as well as fields up to `traversal_level` deep. beyond that bounded - pass, falls back to an unbounded-depth scan, so a blob left behind by a - shallower decrypt_record/reencrypt call is still reported — safe to use as a - leftover-detecting post-rotation audit; never returns False for a record that - still contains a blob, at any depth. + 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. - aliases: is_encrypted_document, is_encrypted_dict — same function + Args: + record: the record (or bare blob) to check. + traversal_level: bounded-pass nesting depth before the unbounded + fallback scan (default 2). + + Returns: + True if record or anything nested under it is an encrypted blob. + + aliases: is_encrypted_document, is_encrypted_dict - same function """ if not isinstance(record, dict): return False @@ -538,12 +542,25 @@ def is_encrypted_record(record, traversal_level: int = 2) -> bool: 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 bare {secure, iv, data} blob (file-storage pattern) it is - decrypted directly and the value (dict or string — see decrypt_data) is returned - in place of `record`. a failure on a single field (or on `record` itself) is - logged and left encrypted, so a partial failure stays visible rather than silent. + if `record` itself is a bare {secure, iv, data} blob (file-storage pattern) it + is decrypted directly and the value (dict or string, see decrypt_data) is + returned in place of `record`. a failure on a single field (or on `record` + itself) is logged and left encrypted, so a partial failure stays visible + rather than silent. - aliases: decrypt_document, decrypt_dict — same function + Args: + crypto: instance holding the data key to decrypt with. + record: the record (or bare blob) to decrypt. not mutated. + traversal_level: max nesting depth to recurse into (default 2). + + Returns: + a new dict with encrypted fields decrypted (or record unchanged if not + a dict). + + Raises: + ValueError: crypto is not initialized with a data key. + + aliases: decrypt_document, decrypt_dict - same function """ if not crypto.master_key: raise ValueError("not initialized with data key") @@ -570,11 +587,7 @@ def decrypt_record(crypto: EnvelopeCrypto, record, traversal_level: int = 2) -> def _fingerprint_default(value: Any) -> str: - """json.dumps default= handler for values fingerprint_data can't natively serialize - - covers datetime/date/time (isoformat), bytes/bytearray (hex), and anything else - (e.g. ObjectId) via a type-tagged repr — never raises, never logs the value. - """ + """json.dumps default= handler for values fingerprint_data can't natively serialize""" if hasattr(value, "isoformat"): return f"isoformat:{value.isoformat()}" if isinstance(value, (bytes, bytearray)): @@ -583,13 +596,7 @@ def _fingerprint_default(value: Any) -> str: def _fingerprint_normalize(value: Any) -> Any: - """recursively tag dict keys with their type to avoid cross-type key collisions - - json object keys are always strings, so {1: "a"} and {"1": "a"} would otherwise - serialize identically and collide to the same fingerprint; prefixing each key with - its type name keeps them distinct. non-dict containers/values pass through - untouched (lists recurse; leaf values are handled by _fingerprint_default). - """ + """recursively tag dict keys with their type to avoid cross-type key collisions""" if isinstance(value, dict): return { f"{type(key).__name__}:{key!r}": _fingerprint_normalize(sub_value) @@ -603,18 +610,24 @@ def _fingerprint_normalize(value: Any) -> Any: def fingerprint_data(data: dict) -> str: """return a deterministic, collision-free SHA-256 hex fingerprint of a dict - dict keys of different types that would otherwise coerce to the same JSON string - (e.g. {1: "a"} vs {"1": "a"}) are kept distinct via a type-tagged pre-pass. values - that aren't JSON-native (datetime/date/time, bytes/bytearray, ObjectId-like - objects) are serialized via a stable default= handler instead of raising. never - logs the data being fingerprinted. + dict keys of different types that would otherwise coerce to the same JSON + string (e.g. {1: "a"} vs {"1": "a"}) are kept distinct via a type-tagged + pre-pass. values that aren't JSON-native (datetime/date/time, bytes/bytearray, + ObjectId-like objects) are serialized via a stable default= handler instead of + raising. never logs the data being fingerprinted. + + Args: + data: the dict to fingerprint. + + Returns: + a deterministic SHA-256 hex digest. """ normalized = _fingerprint_normalize(data) encoded = json.dumps(normalized, sort_keys=True, default=_fingerprint_default) return hashlib.sha256(encoded.encode()).hexdigest() -# function aliases — same functions, naming preference only +# function aliases - same functions, naming preference only is_encrypted_document = is_encrypted_record is_encrypted_dict = is_encrypted_record decrypt_document = decrypt_record