rethink-public/envelope_crypto
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: EC-5..EC-8 error-message inversion, RSA-only wrap guard, 32-byte key guard, fingerprint_data robustness

Browse Source 
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>
This commit is contained in:
dsql
2026-07-02 23:24:06 -04:00
parent d827dca30f
commit d446f50942
3 changed files with 167 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+15 -3
View File
@@ -11,18 +11,18 @@ and storage-agnostic.
`requirements.txt`:
```
envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.5
envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.6
```
Direct:
```bash
pip install "envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.5"
pip install "envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.6"
```
Requires `cryptography` (pulled transitively).
Drop the `@v0.1.5` 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
@@ -177,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).
Using it does not by itself confer PCI-DSS or any other compliance — that is a
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
pyproject.toml
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "envelope_crypto"
version = "0.1.5"
version = "0.1.6"
description = "Envelope encryption (RSA-OAEP wrapped AES-256-GCM) for dict records — config-free, storage-agnostic, installable."
requires-python = ">=3.10"
dependencies = [
src/envelope_crypto/envelope_crypto.py
+151 -92
View File
@@ -1,16 +1,18 @@
"""
envelope encryption for dict records
hybrid encryption: a random AES-256-GCM data key (DEK) encrypts the data, and
that key is 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. this is the same
envelope-encryption pattern used by KMS-style systems.
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.
from envelope_crypto import 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}
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)
caller_store({"_id": fp, "key": wrapped})
deauthorize: caller deletes that fingerprint's record. note this stops future
unwraps but does not revoke a DEK already in a running system's memory — rotate
if compromised.
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:
@@ -45,24 +46,24 @@ set, then re-encrypt existing records old -> new:
caller_update(new_crypto.reencrypt(crypto, record))
reencrypt/is_encrypted_record/decrypt_record all detect a bare {secure, iv, data}
blob used AS the whole record (the file-storage pattern), not just blobs nested
under a key. reencrypt fails loud (raises) rather than silently leaving a field
under the old key — including when a blob sits deeper than `traversal_level`;
is_encrypted_record falls back to an unbounded-depth scan past traversal_level so
it reliably catches leftovers as a post-rotation audit.
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 under the hood) 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.
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.
aliases (PCICrypto is a deprecated legacy alias). the document/record/dict function
variants are the same functions — use whichever fits your storage.
"""
import os
@@ -75,19 +76,31 @@ from typing import Any, Dict, List, Optional, Tuple, Union
from cryptography.exceptions import UnsupportedAlgorithm
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.serialization import load_ssh_public_key
_log = logging.getLogger(__name__)
def _load_private_key(key_data: bytes, pw: Optional[bytes]):
"""load a PEM or OpenSSH private key, normalizing the missing-password error
def _password_mismatch_message(pw: Optional[bytes]) -> str:
"""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
(PEM raises it on the first call; OpenSSH raises it inside the openssh fallback).
both are normalized to a clear ValueError so callers see one error type.
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:
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:
return serialization.load_pem_private_key(key_data, password=pw)
@@ -96,18 +109,14 @@ def _load_private_key(key_data: bytes, pw: Optional[bytes]):
try:
return serialization.load_ssh_private_key(key_data, password=pw)
except TypeError as ssh_error:
raise ValueError(
"private key is encrypted but no password was provided"
) from 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:
raise ValueError(
"private key is encrypted but no password was provided"
) from error
raise ValueError(_password_mismatch_message(pw)) from error
def _fingerprint_of(public_key) -> str:
@@ -143,6 +152,18 @@ def _has_encrypted_field(record: Any) -> bool:
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):
"""load a PEM or OpenSSH public key, normalizing non-key input to ValueError
@@ -190,14 +211,11 @@ class EnvelopeCrypto:
) -> bool:
"""verify the full pipeline against a keypair; raises on any mismatch
round-trips sample data through this instance's DEK, then wraps the DEK
with the public key and unwraps with the private key, confirming they
match. run after bootstrap (or anytime as a health check) to catch a bad
keypair or wrong key path before relying on it. is_file (default True) is
threaded through to both the public and private key loads, so
is_file=False correctly treats both rsa_public_key and rsa_private_key as
in-memory PEM/OpenSSH data rather than opening either as a file path.
returns True on success.
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.
"""
if not self.master_key:
raise ValueError("self_test: not initialized with a key")
@@ -221,7 +239,15 @@ class EnvelopeCrypto:
return True
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
_log.info("crypto initialized with data key")
@@ -248,11 +274,10 @@ class EnvelopeCrypto:
"""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. fingerprinting always uses the public half, so a
private and its public key produce the same fingerprint. PEM and OpenSSH
private-key formats are both accepted (mirrors decrypt_aes_key_with_rsa). an
encrypted key with no/wrong password raises ValueError with a clear message
(cryptography raises TypeError for the missing-password case — normalized here).
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).
"""
if is_file:
with open(key_path_or_data, "rb") as key_file:
@@ -278,7 +303,11 @@ class EnvelopeCrypto:
def encrypt_aes_key_with_rsa(
self, aes_key: bytes, rsa_key: str, is_file: bool = True
) -> 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:
with open(rsa_key, "rb") as key_file:
key_data = key_file.read()
@@ -286,6 +315,7 @@ class EnvelopeCrypto:
key_data = rsa_key.encode() if isinstance(rsa_key, str) else rsa_key
public_key = _load_public_key(key_data)
_require_rsa(public_key)
wrapped = public_key.encrypt(
aes_key,
@@ -319,6 +349,7 @@ class EnvelopeCrypto:
key_data = rsa_private_key.encode() if isinstance(rsa_private_key, str) else rsa_private_key
pw = password.encode() if password else None
private_key = _load_private_key(key_data, pw)
_require_rsa(private_key)
wrapped = base64.b64decode(encrypted_key_base64)
aes_key = private_key.decrypt(
@@ -399,17 +430,14 @@ 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
only treats a json-OBJECT plaintext as a dict and returns everything else as the
raw string. this keeps the type faithful for the common cases: a string that is
json-shaped but NOT an object ('123'->'123', 'true'->'true', '[1,2]'->'[1,2]')
round-trips as a STRING, not an int/bool/list. the one irreducible ambiguity: a
string whose exact value is a json OBJECT ('{"a":1}') decrypts to a dict, because
without a type marker it is indistinguishable from a stored dict — don't store a
bare string that is a json object if you need it back as a string. existing stored
blobs are unaffected — a dict was stored as a json object and still parses to a dict.
dict KEYS round-trip faithfully only because encrypt_data now requires str keys —
json has no other key type, so this is the only shape decrypt_data ever sees.
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).
"""
if not self.master_key:
raise ValueError("not initialized with data key")
@@ -431,26 +459,22 @@ 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.
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.
if `record` itself is a {secure, iv, data} blob (the file-storage pattern, where
the blob IS the whole document) it is re-encrypted directly and the result is
returned in place of `record` — not nested under a key.
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.
unlike decrypt_record (which logs a failed field and leaves it encrypted), a
per-field decrypt failure here RAISES — rotation must fail loud, since silently
keeping a field under the old key would lose it once the old key is retired. for
the same reason, a blob nested DEEPER than `traversal_level` also RAISES instead
of being silently left under the old key: the caller either needs a higher
`traversal_level` or must flatten the record before rotation.
traversal recurses into nested DICTS only; 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.
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.
"""
if not self.master_key:
raise ValueError("destination not initialized with data key")
@@ -484,12 +508,11 @@ def is_encrypted_record(record, traversal_level: int = 2) -> bool:
"""return whether a record has any encrypted ({secure, iv, data}) fields
checks `record` itself (the file-storage pattern stores the blob AS the whole
document, not nested under a key) as well as fields up to `traversal_level` deep.
beyond that bounded pass, this ALSO does an unbounded-depth scan before giving up —
so a blob left behind by a shallower decrypt_record/reencrypt call (nested deeper
than their traversal_level) is still reported as encrypted. this makes the function
safe to use as a leftover-detecting post-rotation audit: it never returns False for
a record that still contains a blob, at any depth.
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.
aliases: is_encrypted_document, is_encrypted_dict — same function
"""
@@ -515,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]:
"""decrypt a record's encrypted fields into a new dict (up to traversal_level deep)
if `record` itself is a {secure, iv, data} blob (the file-storage pattern, where the
blob IS the whole document) it is decrypted directly and the decrypted value
(dict or string — see decrypt_data) is returned in place of `record`.
failures on a single field are logged and that field is left encrypted, so a
partial failure is visible (the {secure,...} blob remains) rather than silent. a
failure decrypting `record` itself (the self-blob case above) is likewise logged
and the still-encrypted blob is returned unchanged.
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
"""
@@ -550,9 +569,49 @@ def decrypt_record(crypto: EnvelopeCrypto, record, traversal_level: int = 2) ->
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:
"""return a deterministic SHA-256 hex fingerprint of a dict"""
return hashlib.sha256(json.dumps(data, sort_keys=True).encode()).hexdigest()
"""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.
"""
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