2 Commits
Author SHA1 Message Date
dsql d446f50942 fix: EC-5..EC-8 error-message inversion, RSA-only wrap guard, 32-byte key guard, fingerprint_data robustness
EC-5: _load_private_key branches on whether a password was given so the normalized
ValueError matches the actual cryptography TypeError case (was always claiming
"encrypted but no password" even when a password was given for an unencrypted key).
EC-6: encrypt_aes_key_with_rsa/decrypt_aes_key_with_rsa now raise a clear ValueError
via _require_rsa for a non-RSA key (e.g. Ed25519/EC), instead of crashing raw with
AttributeError at wrap/unwrap — this lib is RSA-envelope only.
EC-7: initialize() requires exactly 32 bytes (isinstance bytes, len==32), rejecting a
16/24-byte key (silent AES-128/192 downgrade) or a str instead of failing late and
opaquely at first encrypt.
EC-8: fingerprint_data gains a default= handler (datetime/date/time, bytes/bytearray,
and a type-tagged repr fallback) plus a key-type-tagging pre-pass so datetime/bytes/
ObjectId-like values no longer TypeError and int-vs-str dict keys no longer collide
to the same fingerprint. Never logs the data being fingerprinted.

Also compresses the essay-length docstrings (module + several methods) to cut
narration while keeping the load-bearing footgun notes (RSA-only, AES-256 key length,
never-log-key-material) intact — zero behavior change, re-verified after.

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-02 23:24:06 -04:00
dsql d827dca30f fix: is_file drift on decrypt_aes_key_with_rsa and non-str dict key coercion
self_test(is_file=False) only forwarded is_file to the public-key wrap;
decrypt_aes_key_with_rsa had no is_file parameter and always open()'d its
argument, so a PEM string was opened as a filename, misdiagnosing a good
keypair as non-pairing and leaking the private key PEM into the
FileNotFoundError traceback. decrypt_aes_key_with_rsa now takes is_file
(default True, preserving current callers), and self_test threads it
through to both key loads.

encrypt_data json.dumps a dict without checking key types, silently
stringifying int/float/bool/None keys (e.g. snowflake-int-keyed dicts),
so a decrypt round-trip silently lost the original key. encrypt_data now
raises TypeError on a non-str key instead of coercing it.

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-02 17:17:44 -04:00
3 changed files with 213 additions and 93 deletions
+27 -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.4 envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.6
``` ```
Direct: Direct:
```bash ```bash
pip install "envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.4" pip install "envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.6"
``` ```
Requires `cryptography` (pulled transitively). Requires `cryptography` (pulled transitively).
Drop the `@v0.1.4` suffix from the line above to install the latest unpinned. Drop the `@v0.1.6` suffix from the line above to install the latest unpinned.
## First-time setup ## First-time setup
@@ -66,6 +66,13 @@ bot.crypto = crypto
The `keys` schema (`_id` = fingerprint, `key` = wrapped) is the **caller's** choice; The `keys` schema (`_id` = fingerprint, `key` = wrapped) is the **caller's** choice;
this lib only produces `(fingerprint, wrapped_key)`. this lib only produces `(fingerprint, wrapped_key)`.
`decrypt_aes_key_with_rsa` (like `encrypt_aes_key_with_rsa` and
`get_rsa_key_fingerprint`) takes `is_file` (default `True`). Pass `is_file=False` to
hand it PEM/OpenSSH key data directly — e.g. a private key sourced from a vault —
instead of a file path. `self_test` threads the same `is_file` through to both the
public and private key it loads, so `self_test(pub_pem, priv_pem, is_file=False)`
round-trips two in-memory PEM strings rather than treating them as paths.
## Encrypt / decrypt ## Encrypt / decrypt
```python ```python
@@ -73,6 +80,11 @@ enc = crypto.encrypt_data({"ssn": "..."}) # -> {"secure": True, "iv": ...,
plain = crypto.decrypt_data(enc) # -> {"ssn": "..."} 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.
For whole records: `decrypt_record(crypto, doc)` decrypts every `{secure, iv, data}` 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 field (nested up to `traversal_level`, default 2); `is_encrypted_record(doc)` reports
whether any encrypted field exists. Both also detect `doc` itself being a bare whether any encrypted field exists. Both also detect `doc` itself being a bare
@@ -165,6 +177,18 @@ The lib never touches a database; only the caller's storage layer differs.
- The scheme is envelope/hybrid encryption (AES-256-GCM data key wrapped by RSA-OAEP). - The scheme is envelope/hybrid encryption (AES-256-GCM data key wrapped by RSA-OAEP).
Using it does not by itself confer PCI-DSS or any other compliance — that is a Using it does not by itself confer PCI-DSS or any other compliance — that is a
whole-system property. whole-system property.
- `initialize(master_key)` requires exactly 32 bytes (`bytes`, `len == 32`) — a
16/24-byte key or a `str` raises `ValueError` instead of silently downgrading to
AES-128/192 or failing late at first encrypt.
- Wrap/unwrap (`encrypt_aes_key_with_rsa`, `decrypt_aes_key_with_rsa`,
`authorize_system`, `bootstrap`, `rotate_master_key`) is **RSA-only**. A non-RSA
key (e.g. Ed25519/EC) loads and fingerprints fine but raises a clear `ValueError`
at wrap/unwrap rather than a raw `AttributeError`.
- `fingerprint_data` serializes non-JSON-native values (datetime/date/time,
bytes/bytearray, ObjectId-like objects) via a stable `default=` handler instead of
raising `TypeError`, and type-tags dict keys internally so `{1: "a"}` and
`{"1": "a"}` no longer collide to the same fingerprint. Never logs the data it
fingerprints.
## Versioning ## Versioning
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project] [project]
name = "envelope_crypto" name = "envelope_crypto"
version = "0.1.4" version = "0.1.6"
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 = [
+185 -89
View File
@@ -1,16 +1,18 @@
""" """
envelope encryption for dict records envelope encryption for dict records
hybrid encryption: a random AES-256-GCM data key (DEK) encrypts the data, and hybrid encryption: a random AES-256-GCM data key (DEK) encrypts the data, wrapped
that key is wrapped (RSA-OAEP) per authorized system's public key (KEK) for (RSA-OAEP) per authorized system's public key (KEK) for distribution. the wrapped
distribution. the wrapped key is stored by the caller, keyed by fingerprint; key is stored by the caller, keyed by fingerprint; each system unwraps its own copy
each system unwraps its own copy with its private key. this is the same with its private key. same pattern KMS-style systems use. RSA-envelope only — a
envelope-encryption pattern used by KMS-style systems. 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.
from envelope_crypto import EnvelopeCrypto from envelope_crypto import EnvelopeCrypto
crypto = EnvelopeCrypto() crypto = EnvelopeCrypto()
crypto.initialize(master_key) # 32-byte AES DEK crypto.initialize(master_key) # exactly 32 bytes (AES-256)
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
@@ -32,9 +34,8 @@ authorize another system (this instance must already hold the DEK):
fp, wrapped = crypto.authorize_system(other_pub_path) fp, wrapped = crypto.authorize_system(other_pub_path)
caller_store({"_id": fp, "key": wrapped}) caller_store({"_id": fp, "key": wrapped})
deauthorize: caller deletes that fingerprint's record. note this stops future deauthorize: caller deletes that fingerprint's record. stops future unwraps but
unwraps but does not revoke a DEK already in a running system's memory — rotate does not revoke a DEK already in a running system's memory — rotate if compromised.
if compromised.
rotate (new DEK + re-encrypt): generate a new DEK, wrap for the still-authorized rotate (new DEK + re-encrypt): generate a new DEK, wrap for the still-authorized
set, then re-encrypt existing records old -> new: set, then re-encrypt existing records old -> new:
@@ -45,19 +46,24 @@ set, then re-encrypt existing records old -> new:
caller_update(new_crypto.reencrypt(crypto, record)) caller_update(new_crypto.reencrypt(crypto, record))
reencrypt/is_encrypted_record/decrypt_record all detect a bare {secure, iv, data} 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 blob used AS the whole record (file-storage pattern), not just blobs nested under a
under a key. reencrypt fails loud (raises) rather than silently leaving a field key. reencrypt fails loud (raises) rather than silently leaving a field under the
under the old key — including when a blob sits deeper than `traversal_level`; old key — including a blob nested deeper than `traversal_level`; is_encrypted_record
is_encrypted_record falls back to an unbounded-depth scan past traversal_level so falls back to an unbounded-depth scan past traversal_level to reliably catch
it reliably catches leftovers as a post-rotation audit. leftovers as a post-rotation audit.
config-free: the host supplies the DEK and RSA key paths; this lib never imports 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 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. 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 aliases (PCICrypto is a deprecated legacy alias). the document/record/dict function
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
@@ -70,19 +76,31 @@ from typing import Any, Dict, List, Optional, Tuple, Union
from cryptography.exceptions import UnsupportedAlgorithm from cryptography.exceptions import UnsupportedAlgorithm
from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding from cryptography.hazmat.primitives.asymmetric import padding, rsa
from cryptography.hazmat.primitives.ciphers.aead import AESGCM from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.serialization import load_ssh_public_key from cryptography.hazmat.primitives.serialization import load_ssh_public_key
_log = logging.getLogger(__name__) _log = logging.getLogger(__name__)
def _load_private_key(key_data: bytes, pw: Optional[bytes]): def _password_mismatch_message(pw: Optional[bytes]) -> str:
"""load a PEM or OpenSSH private key, normalizing the missing-password error """clear ValueError text for a TypeError raised by a password/encryption mismatch
cryptography raises TypeError when a key is encrypted but no password was given cryptography raises TypeError for both directions: encrypted key + no password,
(PEM raises it on the first call; OpenSSH raises it inside the openssh fallback). and unencrypted key + a password given. branch on which the caller supplied so
both are normalized to a clear ValueError so callers see one error type. the message matches the actual case instead of always claiming "encrypted".
"""
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.
""" """
try: try:
return serialization.load_pem_private_key(key_data, password=pw) return serialization.load_pem_private_key(key_data, password=pw)
@@ -91,18 +109,14 @@ def _load_private_key(key_data: bytes, pw: Optional[bytes]):
try: try:
return serialization.load_ssh_private_key(key_data, password=pw) return serialization.load_ssh_private_key(key_data, password=pw)
except TypeError as ssh_error: except TypeError as ssh_error:
raise ValueError( raise ValueError(_password_mismatch_message(pw)) from ssh_error
"private key is encrypted but no password was provided"
) 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 # 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" # 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:
raise ValueError( raise ValueError(_password_mismatch_message(pw)) from error
"private key is encrypted but no password was provided"
) from error
def _fingerprint_of(public_key) -> str: def _fingerprint_of(public_key) -> str:
@@ -138,6 +152,18 @@ def _has_encrypted_field(record: Any) -> bool:
return any(_has_encrypted_field(value) for value in record.values()) return any(_has_encrypted_field(value) for value in record.values())
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.
"""
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): 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
@@ -185,10 +211,11 @@ class EnvelopeCrypto:
) -> bool: ) -> bool:
"""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 the DEK round-trips sample data through this instance's DEK, then wraps and unwraps
with the public key and unwraps with the private key, confirming they the DEK with the given keypair, confirming they match. run after bootstrap
match. run after bootstrap (or anytime as a health check) to catch a bad (or as a health check) to catch a bad keypair or wrong path before relying
keypair or wrong key path before relying on it. returns True on success. 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.
""" """
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")
@@ -199,7 +226,7 @@ class EnvelopeCrypto:
_, wrapped = self.encrypt_aes_key_with_rsa(self.master_key, rsa_public_key, is_file=is_file) _, wrapped = self.encrypt_aes_key_with_rsa(self.master_key, rsa_public_key, is_file=is_file)
try: try:
recovered = self.decrypt_aes_key_with_rsa(wrapped, rsa_private_key, password=password) recovered = self.decrypt_aes_key_with_rsa(wrapped, rsa_private_key, is_file=is_file, password=password)
except Exception as error: except Exception as error:
raise RuntimeError( raise RuntimeError(
"self_test: key unwrap failed (public/private keys do not pair, " "self_test: key unwrap failed (public/private keys do not pair, "
@@ -212,7 +239,15 @@ class EnvelopeCrypto:
return True return True
def initialize(self, master_key: bytes) -> None: def initialize(self, master_key: bytes) -> None:
"""arm the instance with the AES data key (DEK)""" """arm the instance with the AES data key (DEK)
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.
never logs the key material itself.
"""
if not isinstance(master_key, bytes) or len(master_key) != 32:
raise ValueError("master_key must be exactly 32 bytes (AES-256)")
self.master_key = master_key self.master_key = master_key
_log.info("crypto initialized with data key") _log.info("crypto initialized with data key")
@@ -239,11 +274,10 @@ class EnvelopeCrypto:
"""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 for an encrypted private key (is_private=True), pass its `password`; an
unencrypted key ignores it. fingerprinting always uses the public half, so a unencrypted key ignores it. always fingerprints the public half, so a
private and its public key produce the same fingerprint. PEM and OpenSSH private key and its public key match. PEM and OpenSSH accepted (mirrors
private-key formats are both accepted (mirrors decrypt_aes_key_with_rsa). an decrypt_aes_key_with_rsa). a password/encryption mismatch raises a clear
encrypted key with no/wrong password raises ValueError with a clear message ValueError (cryptography's raw TypeError normalized here).
(cryptography raises TypeError for the missing-password case — 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:
@@ -269,7 +303,11 @@ class EnvelopeCrypto:
def encrypt_aes_key_with_rsa( def encrypt_aes_key_with_rsa(
self, aes_key: bytes, rsa_key: str, is_file: bool = True self, aes_key: bytes, rsa_key: str, is_file: bool = True
) -> Tuple[str, str]: ) -> Tuple[str, str]:
"""wrap an AES key with an RSA public key; returns (fingerprint, wrapped_b64)""" """wrap an AES key with an RSA public key; returns (fingerprint, wrapped_b64)
raises ValueError if rsa_key is not an RSA key (this lib is RSA-envelope only;
e.g. an Ed25519/EC key loads and fingerprints fine but cannot wrap).
"""
if is_file: if is_file:
with open(rsa_key, "rb") as key_file: with open(rsa_key, "rb") as key_file:
key_data = key_file.read() key_data = key_file.read()
@@ -277,6 +315,7 @@ class EnvelopeCrypto:
key_data = rsa_key.encode() if isinstance(rsa_key, str) else rsa_key key_data = rsa_key.encode() if isinstance(rsa_key, str) else rsa_key
public_key = _load_public_key(key_data) public_key = _load_public_key(key_data)
_require_rsa(public_key)
wrapped = public_key.encrypt( wrapped = public_key.encrypt(
aes_key, aes_key,
@@ -293,14 +332,24 @@ class EnvelopeCrypto:
return fingerprint, wrapped_b64 return fingerprint, wrapped_b64
def decrypt_aes_key_with_rsa( def decrypt_aes_key_with_rsa(
self, encrypted_key_base64: str, rsa_private_key_path: str, self, encrypted_key_base64: str, rsa_private_key: str,
password: Optional[str] = None, is_file: bool = True, password: Optional[str] = None,
) -> bytes: ) -> bytes:
"""unwrap an AES key with an RSA private key""" """unwrap an AES key with an RSA private key
with open(rsa_private_key_path, "rb") as key_file:
key_data = key_file.read() 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.
"""
if is_file:
with open(rsa_private_key, "rb") as key_file:
key_data = key_file.read()
else:
key_data = rsa_private_key.encode() if isinstance(rsa_private_key, str) else rsa_private_key
pw = password.encode() if password else None pw = password.encode() if password else None
private_key = _load_private_key(key_data, pw) private_key = _load_private_key(key_data, pw)
_require_rsa(private_key)
wrapped = base64.b64decode(encrypted_key_base64) wrapped = base64.b64decode(encrypted_key_base64)
aes_key = private_key.decrypt( aes_key = private_key.decrypt(
@@ -343,13 +392,30 @@ class EnvelopeCrypto:
return new_key, wrapped return new_key, wrapped
def encrypt_data(self, data: Union[Dict[str, Any], str]) -> Dict[str, str]: 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""" """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
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: 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; # 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) # 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):
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]}"
)
data_str = json.dumps(data) if isinstance(data, dict) else data data_str = json.dumps(data) if isinstance(data, dict) else data
iv = os.urandom(12) iv = os.urandom(12)
@@ -364,15 +430,14 @@ 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 encrypt_data only json-encodes dicts (a string is stored verbatim), so decrypt
only treats a json-OBJECT plaintext as a dict and returns everything else as the treats a json-OBJECT plaintext as a dict and everything else as a raw string —
raw string. this keeps the type faithful for the common cases: a string that is a json-shaped but non-object string ('123', 'true', '[1,2]') round-trips as a
json-shaped but NOT an object ('123'->'123', 'true'->'true', '[1,2]'->'[1,2]') STRING, not int/bool/list. one irreducible ambiguity: a string whose exact
round-trips as a STRING, not an int/bool/list. the one irreducible ambiguity: a value is a json object ('{"a":1}') decrypts to a dict, indistinguishable
string whose exact value is a json OBJECT ('{"a":1}') decrypts to a dict, because without a type marker from a stored dict — don't store a bare json-object
without a type marker it is indistinguishable from a stored dict — don't store a string if you need it back as a string. dict keys round-trip faithfully
bare string that is a json object if you need it back as a string. existing stored because encrypt_data requires str keys (json's only key type).
blobs are unaffected — a dict was stored as a json object and still parses to a dict.
""" """
if not self.master_key: if not self.master_key:
raise ValueError("not initialized with data key") raise ValueError("not initialized with data key")
@@ -394,26 +459,22 @@ 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) self holds the destination (new) key; source_crypto holds the source (old) key.
key. only {secure, iv, data} fields are touched; plaintext fields are left only {secure, iv, data} fields are touched; plaintext fields are left as-is.
as-is. returns a new dict; the input is not mutated. used during rotation. 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.
if `record` itself is a {secure, iv, data} blob (the file-storage pattern, where fails loud, unlike decrypt_record: a per-field decrypt failure RAISES (silently
the blob IS the whole document) it is re-encrypted directly and the result is keeping a field under the old key would lose it once that key is retired), and
returned in place of `record` — not nested under a key. so does a blob nested DEEPER than `traversal_level` — raise a higher
traversal_level or flatten the record before rotation instead.
unlike decrypt_record (which logs a failed field and leaves it encrypted), a traversal recurses into nested DICTS only; a blob nested inside a LIST is not
per-field decrypt failure here RAISES — rotation must fail loud, since silently re-encrypted and not covered by the depth-limit raise. this scheme keys blobs
keeping a field under the old key would lose it once the old key is retired. for by field name, not inside arrays, so this shouldn't arise in practice — but
the same reason, a blob nested DEEPER than `traversal_level` also RAISES instead flatten list-nested blobs to dict fields before rotation or they'll be
of being silently left under the old key: the caller either needs a higher silently left under the old key.
`traversal_level` or must flatten the record before rotation.
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: if not self.master_key:
raise ValueError("destination not initialized with data key") raise ValueError("destination not initialized with data key")
@@ -447,12 +508,11 @@ 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 stores the blob AS the whole
document, not nested under a key) as well as fields up to `traversal_level` deep. document) as well as fields up to `traversal_level` deep. beyond that bounded
beyond that bounded pass, this ALSO does an unbounded-depth scan before giving up — pass, falls back to an unbounded-depth scan, so a blob left behind by a
so a blob left behind by a shallower decrypt_record/reencrypt call (nested deeper shallower decrypt_record/reencrypt call is still reported — safe to use as a
than their traversal_level) is still reported as encrypted. this makes the function leftover-detecting post-rotation audit; never returns False for a record that
safe to use as a leftover-detecting post-rotation audit: it never returns False for still contains a blob, at any depth.
a record that still contains a blob, at any depth.
aliases: is_encrypted_document, is_encrypted_dict — same function aliases: is_encrypted_document, is_encrypted_dict — same function
""" """
@@ -478,14 +538,10 @@ 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 {secure, iv, data} blob (the file-storage pattern, where the if `record` itself is a bare {secure, iv, data} blob (file-storage pattern) it is
blob IS the whole document) it is decrypted directly and the decrypted value decrypted directly and the value (dict or string — see decrypt_data) is returned
(dict or string — see decrypt_data) is returned in place of `record`. 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.
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. 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 aliases: decrypt_document, decrypt_dict — same function
""" """
@@ -513,9 +569,49 @@ def decrypt_record(crypto: EnvelopeCrypto, record, traversal_level: int = 2) ->
return result return result
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.
"""
if hasattr(value, "isoformat"):
return f"isoformat:{value.isoformat()}"
if isinstance(value, (bytes, bytearray)):
return f"hex:{value.hex()}"
return f"{type(value).__name__}:{value!r}"
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).
"""
if isinstance(value, dict):
return {
f"{type(key).__name__}:{key!r}": _fingerprint_normalize(sub_value)
for key, sub_value in value.items()
}
if isinstance(value, (list, tuple)):
return [_fingerprint_normalize(item) for item in value]
return value
def fingerprint_data(data: dict) -> str: def fingerprint_data(data: dict) -> str:
"""return a deterministic SHA-256 hex fingerprint of a dict""" """return a deterministic, collision-free SHA-256 hex fingerprint of a dict
return hashlib.sha256(json.dumps(data, sort_keys=True).encode()).hexdigest()
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.
"""
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