docs: compress prose/module docstrings, em-dash->hyphen (de-bloat wave 1)

Signed-off-by: disqualifier <dev@disqualifier.me>
This commit is contained in:
2026-07-03 00:16:42 -04:00
parent d446f50942
commit 7287a52947
3 changed files with 193 additions and 180 deletions
+3 -3
View File
@@ -11,18 +11,18 @@ and storage-agnostic.
`requirements.txt`: `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: Direct:
```bash ```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). 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 ## First-time setup
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project] [project]
name = "envelope_crypto" 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." 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 = [
+189 -176
View File
@@ -2,12 +2,11 @@
envelope encryption for dict records envelope encryption for dict records
hybrid encryption: a random AES-256-GCM data key (DEK) encrypts the data, wrapped 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 (RSA-OAEP) per authorized system's public key (KEK) for distribution and stored by
key is stored by the caller, keyed by fingerprint; each system unwraps its own copy the caller, keyed by fingerprint. RSA-envelope only - a non-RSA key (e.g. Ed25519/EC)
with its private key. same pattern KMS-style systems use. RSA-envelope only — a loads and fingerprints fine but raises ValueError at wrap/unwrap. never logs key
non-RSA key (e.g. Ed25519/EC) loads and fingerprints fine but raises ValueError at material (DEK, PEM, wrapped key) - only fingerprints and counts. config-free and
wrap/unwrap. never logs key material (DEK, PEM, wrapped key) — only fingerprints storage-agnostic; see README for bootstrap/boot/authorize/rotate flows.
and counts.
from envelope_crypto import EnvelopeCrypto from envelope_crypto import EnvelopeCrypto
@@ -16,54 +15,9 @@ and counts.
enc = crypto.encrypt_data({"ssn": "..."}) # -> {secure, iv, data} enc = crypto.encrypt_data({"ssn": "..."}) # -> {secure, iv, data}
plain = crypto.decrypt_data(enc) # -> original 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 naming: EnvelopeCrypto is canonical. PCICrypto / DocumentCrypto / RecordCrypto are
aliases (PCICrypto is a deprecated legacy alias). the document/record/dict function 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 import os
@@ -84,24 +38,14 @@ _log = logging.getLogger(__name__)
def _password_mismatch_message(pw: Optional[bytes]) -> str: def _password_mismatch_message(pw: Optional[bytes]) -> str:
"""clear ValueError text for a TypeError raised by a password/encryption mismatch """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".
"""
if pw is not None: if pw is not None:
return "password was given but private key is not encrypted" return "password was given but private key is not encrypted"
return "private key is encrypted but no password was provided" return "private key is encrypted but no password was provided"
def _load_private_key(key_data: bytes, pw: Optional[bytes]): def _load_private_key(key_data: bytes, pw: Optional[bytes]):
"""load a PEM or OpenSSH private key, normalizing the password/encryption mismatch error """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.
"""
try: try:
return serialization.load_pem_private_key(key_data, password=pw) return serialization.load_pem_private_key(key_data, password=pw)
except ValueError as error: except ValueError as error:
@@ -111,8 +55,6 @@ def _load_private_key(key_data: bytes, pw: Optional[bytes]):
except TypeError as ssh_error: except TypeError as ssh_error:
raise ValueError(_password_mismatch_message(pw)) from ssh_error raise ValueError(_password_mismatch_message(pw)) from ssh_error
if pw is not None: 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 ValueError("could not load private key (wrong password or malformed key)") from error
raise error raise error
except TypeError as 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: def _fingerprint_of(public_key) -> str:
"""base64 SHA-256 fingerprint of an already-loaded public key """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.
"""
key_bytes = public_key.public_bytes( key_bytes = public_key.public_bytes(
encoding=serialization.Encoding.DER, encoding=serialization.Encoding.DER,
format=serialization.PublicFormat.SubjectPublicKeyInfo, format=serialization.PublicFormat.SubjectPublicKeyInfo,
@@ -140,11 +78,7 @@ 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) 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): if not isinstance(record, dict):
return False return False
if _is_blob(record): if _is_blob(record):
@@ -153,24 +87,13 @@ def _has_encrypted_field(record: Any) -> bool:
def _require_rsa(key) -> None: def _require_rsa(key) -> None:
"""raise ValueError unless key is an RSA public or private key """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.
"""
if not isinstance(key, (rsa.RSAPublicKey, rsa.RSAPrivateKey)): if not isinstance(key, (rsa.RSAPublicKey, rsa.RSAPrivateKey)):
raise ValueError(f"RSA key required for envelope wrap/unwrap, got {type(key).__name__}") raise ValueError(f"RSA key required for envelope wrap/unwrap, got {type(key).__name__}")
def _load_public_key(key_data: bytes): def _load_public_key(key_data: bytes):
"""load a PEM or OpenSSH public key, normalizing non-key input to ValueError """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.
"""
try: try:
return serialization.load_pem_public_key(key_data) return serialization.load_pem_public_key(key_data)
except ValueError: except ValueError:
@@ -195,10 +118,20 @@ class EnvelopeCrypto:
def bootstrap(cls, rsa_public_key: str, is_file: bool = True) -> Tuple["EnvelopeCrypto", str, str]: 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 """first-time setup: generate a DEK and wrap it for the first system
returns (crypto, fingerprint, wrapped_key) — an initialized instance plus the plaintext DEK is never returned or persisted; it survives only as the
the record to store as the first authorization. the plaintext DEK is never wrapped copy. run self_test before storing to confirm the keypair round-trips.
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 = cls()
crypto.initialize(crypto.create_aes_key()) crypto.initialize(crypto.create_aes_key())
@@ -212,10 +145,23 @@ class EnvelopeCrypto:
"""verify the full pipeline against a keypair; raises on any mismatch """verify the full pipeline against a keypair; raises on any mismatch
round-trips sample data through this instance's DEK, then wraps and unwraps 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 the DEK with the given keypair. run after bootstrap or as a health check.
(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 Args:
as in-memory PEM/OpenSSH data, not paths). returns True on success. 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: if not self.master_key:
raise ValueError("self_test: not initialized with a 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 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 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. never logs the key material itself.
""" """
if not isinstance(master_key, bytes) or len(master_key) != 32: if not isinstance(master_key, bytes) or len(master_key) != 32:
@@ -273,11 +219,23 @@ class EnvelopeCrypto:
) -> str: ) -> str:
"""return a base64 SHA-256 fingerprint of an RSA key for identification """return a base64 SHA-256 fingerprint of an RSA key for identification
for an encrypted private key (is_private=True), pass its `password`; an always fingerprints the public half, so a private key and its public key
unencrypted key ignores it. always fingerprints the public half, so a match. PEM and OpenSSH accepted (mirrors decrypt_aes_key_with_rsa).
private key and its public key match. PEM and OpenSSH accepted (mirrors
decrypt_aes_key_with_rsa). a password/encryption mismatch raises a clear Args:
ValueError (cryptography's raw TypeError normalized here). 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: if is_file:
with open(key_path_or_data, "rb") as key_file: with open(key_path_or_data, "rb") as key_file:
@@ -325,7 +283,6 @@ class EnvelopeCrypto:
label=None, label=None,
), ),
) )
# fingerprint from the already-loaded public_key — no second open/parse of the file
fingerprint = _fingerprint_of(public_key) fingerprint = _fingerprint_of(public_key)
wrapped_b64 = base64.b64encode(wrapped).decode() wrapped_b64 = base64.b64encode(wrapped).decode()
_log.info("wrapped data key for fingerprint %s", fingerprint[:8]) _log.info("wrapped data key for fingerprint %s", fingerprint[:8])
@@ -337,10 +294,20 @@ class EnvelopeCrypto:
) -> bytes: ) -> bytes:
"""unwrap an AES key with an RSA private key """unwrap an AES key with an RSA private key
is_file defaults to True (rsa_private_key is a path), matching Args:
encrypt_aes_key_with_rsa / get_rsa_key_fingerprint; pass is_file=False to encrypted_key_base64: the RSA-OAEP wrapped key, base64-encoded.
supply the PEM/OpenSSH key data directly (e.g. an in-memory or vault-sourced rsa_private_key: path to (or, if is_file=False, raw PEM/OpenSSH data
key) instead of a file path. 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: if is_file:
with open(rsa_private_key, "rb") as key_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]: 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 """wrap the current data key for another system's public key
returns (fingerprint, wrapped_b64) for the caller to store as that requires this instance to already hold the data key - only an
system's key-authorization record. requires this instance to already authorized system can authorize others.
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: if not self.master_key:
raise ValueError("cannot authorize another system: not initialized") raise ValueError("cannot authorize another system: not initialized")
@@ -379,9 +358,20 @@ class EnvelopeCrypto:
) -> Tuple[bytes, Dict[str, str]]: ) -> Tuple[bytes, Dict[str, str]]:
"""generate a NEW data key and wrap it for each authorized public key """generate a NEW data key and wrap it for each authorized public key
returns (new_key, {fingerprint: wrapped_b64}). does NOT re-encrypt existing does NOT re-encrypt existing data - build a new instance with the new key
data — build a new instance with the new key and call reencrypt() on each and call reencrypt() on each record. systems not in the list get no
record. systems not in the list get no wrapped copy (deauthorized). 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() new_key = self.create_aes_key()
wrapped = {} wrapped = {}
@@ -396,22 +386,17 @@ class EnvelopeCrypto:
dict keys must be str: json.dumps silently stringifies int/float/bool/None 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 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 non-str key is rejected here instead, so the failure is loud at encrypt
time rather than a silent lookup miss after decrypt. time rather than a silent lookup miss after decrypt.
""" """
if not self.master_key: if not self.master_key:
raise ValueError("not initialized with data key") raise ValueError("not initialized with data key")
if not isinstance(data, (dict, str)): 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__}") raise TypeError(f"encrypt_data expects a dict or str, got {type(data).__name__}")
if isinstance(data, dict): if isinstance(data, dict):
non_str_keys = [key for key in data if not isinstance(key, str)] non_str_keys = [key for key in data if not isinstance(key, str)]
if non_str_keys: 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( raise TypeError(
"encrypt_data requires str dict keys, got non-str key(s): " "encrypt_data requires str dict keys, got non-str key(s): "
f"{[type(key).__name__ for key in non_str_keys]}" 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]: 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 """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 a json-shaped but non-object plaintext ('123', 'true', '[1,2]') round-trips
treats a json-OBJECT plaintext as a dict and everything else as a raw string as a STRING, not int/bool/list. one irreducible ambiguity: a string whose
a json-shaped but non-object string ('123', 'true', '[1,2]') round-trips as a exact value is a json object ('{"a":1}') decrypts to a dict, indistinguishable
STRING, not int/bool/list. one irreducible ambiguity: a string whose exact from a stored dict - don't store a bare json-object string if you need it
value is a json object ('{"a":1}') decrypts to a dict, indistinguishable back as a string.
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 Args:
because encrypt_data requires str keys (json's only key type). 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: if not self.master_key:
raise ValueError("not initialized with data 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: 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") raise ValueError("decrypt_data expects a {secure, iv, data} blob")
iv = base64.b64decode(encrypted_data["iv"]) 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: 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 """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. self holds the destination (new) key; source_crypto holds the source (old)
only {secure, iv, data} fields are touched; plaintext fields are left as-is. key. only {secure, iv, data} fields are touched; plaintext fields are left
returns a new dict; the input is not mutated. used during rotation. if `record` as-is. used during rotation. if `record` itself is a bare blob (file-storage
itself is a bare blob (file-storage pattern) it is re-encrypted directly and pattern) it is re-encrypted directly and returned in place of `record`.
returned in place of `record`, not nested under a key.
fails loud, unlike decrypt_record: a per-field decrypt failure RAISES (silently traversal recurses into nested DICTS only - a blob nested inside a LIST is
keeping a field under the old key would lose it once that key is retired), and not re-encrypted and not covered by the depth-limit raise below; flatten
so does a blob nested DEEPER than `traversal_level` — raise a higher list-nested blobs to dict fields before rotation or they'll be silently
traversal_level or flatten the record before rotation instead. left under the old key.
traversal recurses into nested DICTS only; a blob nested inside a LIST is not Args:
re-encrypted and not covered by the depth-limit raise. this scheme keys blobs source_crypto: instance holding the old (source) key.
by field name, not inside arrays, so this shouldn't arise in practice — but record: the record to re-encrypt. not mutated; a new dict is returned.
flatten list-nested blobs to dict fields before rotation or they'll be traversal_level: max nesting depth to recurse into (default 2).
silently left under the old key.
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: if not self.master_key:
raise ValueError("destination not initialized with data key") raise ValueError("destination not initialized with data key")
@@ -493,12 +491,12 @@ class EnvelopeCrypto:
raise ValueError( raise ValueError(
f"reencrypt: field {key!r} contains an encrypted blob nested deeper " f"reencrypt: field {key!r} contains an encrypted blob nested deeper "
"than traversal_level; increase traversal_level or flatten the record " "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 return result
# naming aliases same class # naming aliases - same class
DocumentCrypto = EnvelopeCrypto DocumentCrypto = EnvelopeCrypto
RecordCrypto = EnvelopeCrypto RecordCrypto = EnvelopeCrypto
PCICrypto = EnvelopeCrypto # deprecated legacy alias; remove after all systems migrate 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: 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 stores the blob AS the whole checks `record` itself (the file-storage pattern) as well as fields up to
document) as well as fields up to `traversal_level` deep. beyond that bounded `traversal_level` deep, then falls back to an unbounded-depth scan - safe to
pass, falls back to an unbounded-depth scan, so a blob left behind by a use as a leftover-detecting post-rotation audit; never returns False for a
shallower decrypt_record/reencrypt call is still reported — safe to use as a record that still contains a blob, at any depth.
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): if not isinstance(record, dict):
return False 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]: 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) """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 if `record` itself is a bare {secure, iv, data} blob (file-storage pattern) it
decrypted directly and the value (dict or string see decrypt_data) is returned is decrypted directly and the value (dict or string, see decrypt_data) is
in place of `record`. a failure on a single field (or on `record` itself) is returned in place of `record`. a failure on a single field (or on `record`
logged and left encrypted, so a partial failure stays visible rather than silent. 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: if not crypto.master_key:
raise ValueError("not initialized with data 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: def _fingerprint_default(value: Any) -> str:
"""json.dumps default= handler for values fingerprint_data can't natively serialize """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.
"""
if hasattr(value, "isoformat"): if hasattr(value, "isoformat"):
return f"isoformat:{value.isoformat()}" return f"isoformat:{value.isoformat()}"
if isinstance(value, (bytes, bytearray)): if isinstance(value, (bytes, bytearray)):
@@ -583,13 +596,7 @@ def _fingerprint_default(value: Any) -> str:
def _fingerprint_normalize(value: Any) -> Any: def _fingerprint_normalize(value: Any) -> Any:
"""recursively tag dict keys with their type to avoid cross-type key collisions """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).
"""
if isinstance(value, dict): if isinstance(value, dict):
return { return {
f"{type(key).__name__}:{key!r}": _fingerprint_normalize(sub_value) 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: def fingerprint_data(data: dict) -> str:
"""return a deterministic, collision-free SHA-256 hex fingerprint of a dict """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 dict keys of different types that would otherwise coerce to the same JSON
(e.g. {1: "a"} vs {"1": "a"}) are kept distinct via a type-tagged pre-pass. values string (e.g. {1: "a"} vs {"1": "a"}) are kept distinct via a type-tagged
that aren't JSON-native (datetime/date/time, bytes/bytearray, ObjectId-like pre-pass. values that aren't JSON-native (datetime/date/time, bytes/bytearray,
objects) are serialized via a stable default= handler instead of raising. never ObjectId-like objects) are serialized via a stable default= handler instead of
logs the data being fingerprinted. raising. never logs the data being fingerprinted.
Args:
data: the dict to fingerprint.
Returns:
a deterministic SHA-256 hex digest.
""" """
normalized = _fingerprint_normalize(data) normalized = _fingerprint_normalize(data)
encoded = json.dumps(normalized, sort_keys=True, default=_fingerprint_default) encoded = json.dumps(normalized, sort_keys=True, default=_fingerprint_default)
return hashlib.sha256(encoded.encode()).hexdigest() 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_document = is_encrypted_record
is_encrypted_dict = is_encrypted_record is_encrypted_dict = is_encrypted_record
decrypt_document = decrypt_record decrypt_document = decrypt_record