12 Commits
Author SHA1 Message Date
dsql e24da95fc5 release: 1.0.0
first stable release. pre-1.0.0 verification complete: all surviving MED regressions and
gaps resolved and independently re-fired, tree audited clean across the suite.

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-09 18:53:15 -04:00
dsql 1edd7ddfea revert: fingerprint_data back to canonical-json hash + stringify + callable guard
the custom normalization layer (_fingerprint_normalize / _fingerprint_key /
_fingerprint_default + type-tagged markers + determinism allowlist) guarded against
non-json inputs that never arrive by contract, and in exchange crashed on ordinary
value types (uuid/decimal/path/objectid) and still collided (bytes vs ["bytes",hex]).
worse than the four-liner on both axes, and it regressed across four iterations.

fingerprint_data takes json-shaped record data, so json.dumps(sort_keys=True,
separators, default=str) is the correct tool: order-independent, whitespace-stable,
non-json scalars stringified rather than crashing. the one real footgun default=str
introduces - a callable's str() embeds a memory address, non-deterministic across
processes - is guarded with an explicit TypeError. non-str dict keys still raise from
json natively (str-key contract, matching encrypt_data's own guard and the original).

deletes the whole machinery. closes the fingerprint regressions R1/R2/R3 and the
marker-encoding docstring nit at the root. digest VALUE changes vs the marker-based
version - any persisted fingerprints must re-baseline.

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-06 20:55:03 -04:00
dsql 4b9820b18e fix: keep the concrete type in the set-members fingerprint marker
The 6d370a4 determinism rewrite collapsed the set-value marker from
f'{type(value).__name__}:members' to a bare '\x00set:members', so a set value and a
frozenset value with the same members fingerprinted identically - a fix-introduced
collision that contradicts the same commit's collision-resistance claim. Restore the
concrete type in the marker (keeping the \x00 prefix so it stays uncollidable with a
user string key). Determinism across processes is unaffected.

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-06 20:10:32 -04:00
dsql 6d370a45e1 fix: fingerprint_data is deterministic and collision-resistant across the whole input surface
Route dict keys through the same determinism guard as values (a frozenset or
identity-repr key previously used bare repr(), yielding a different digest per
process). Replace the repr-regex identity-repr reject with type-based rejection so
callables, lambdas, bound methods, and generators fail loud instead of fingerprinting
a memory address. Encode bytes/datetime as a tagged JSON array so a genuine str value
of the same shape can never collide (the old 'hex:'/'isoformat:' string tags could).

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-06 19:20:18 -04:00
dsql 581ff712e0 fix: fingerprint_data is deterministic for sets and rejects identity-repr objects
set/frozenset values fell through to repr() (hash-randomized member order), and objects
with the default <X at 0x..> repr embedded a per-process memory address - both made
fingerprint_data return a different digest across restarts despite its 'deterministic'
contract (a fix-wave d446f50 turned a loud TypeError into a silent unstable hash).
_fingerprint_normalize now canonically sorts set/frozenset members, and the default=
handler raises TypeError on an identity-based repr instead of hashing an address.

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-06 16:39:34 -04:00
dsql 8dffb6bc61 docs: correct reencrypt's list-nested-blob docstring to match current behavior
the docstring said a list-nested blob is "not covered by the depth-limit raise
below", but the cutoff check uses _has_encrypted_field, which does walk lists -
so a list-nested blob reached via the cutoff dict scan DOES raise, while the
same blob one level shallower (hit during normal traversal instead) is what's
actually skipped silently. doc-only, no behavior change.

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-06 00:17:18 -04:00
dsql dc8f80c690 fix: is_encrypted_record never false-negatives at odd traversal_level; encrypt_data rejects nested non-str keys
Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-03 19:03:53 -04:00
dsql dfd794159e refactor: derive __version__ from package metadata (single source)
Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-03 17:00:26 -04:00
dsql d0c3bcd7c8 refactor: hoist shared RSA-OAEP padding to a constant; restore encrypt_aes_key_with_rsa docstring
hoists the byte-identical OAEP(MGF1(SHA256), SHA256, label=None) construction out
of encrypt_aes_key_with_rsa and decrypt_aes_key_with_rsa into a module-level
_OAEP_PADDING constant so a future scheme change lands in one place instead of
two in lockstep. also restores encrypt_aes_key_with_rsa's Args/Returns/Raises
docstring block, matching its four siblings, after Wave-1 over-stripped it to
bare prose.

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-03 16:46:20 -04:00
dsql 857e9380c6 docs: bump install pin to v0.1.8
Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-03 16:24:02 -04:00
dsql abf7491d26 fix: is_encrypted_record misses blobs nested inside a list or tuple
both the bounded pass and the unbounded _has_encrypted_field fallback
descended only through dict values, so a blob nested inside a list at
any depth was invisible and the function returned False. reencrypt()
already skips list-nested blobs (documented gotcha), so after rotation
such a blob was stranded under the old key while this audit reported
the record clean - a rotation-data-loss trap once the old wrapped-key
record is deleted. both traversal passes now walk list/tuple items in
addition to dict values; the blob-detection predicate is unchanged.

bump 0.1.7 -> 0.1.8

Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-03 16:14:34 -04:00
dsql 7287a52947 docs: compress prose/module docstrings, em-dash->hyphen (de-bloat wave 1)
Signed-off-by: disqualifier <dev@disqualifier.me>
2026-07-03 00:16:42 -04:00
4 changed files with 285 additions and 240 deletions
+13 -10
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.6
envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v1.0.0
```
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@v1.0.0"
```
Requires `cryptography` (pulled transitively).
Drop the `@v0.1.6` suffix from the line above to install the latest unpinned.
Drop the `@v1.0.0` suffix from the line above to install the latest unpinned.
## First-time setup
@@ -80,10 +80,12 @@ enc = crypto.encrypt_data({"ssn": "..."}) # -> {"secure": True, "iv": ...,
plain = crypto.decrypt_data(enc) # -> {"ssn": "..."}
```
Dict keys must be `str`. `encrypt_data` raises `TypeError` on a non-str key (e.g. an
int-keyed dict of Discord snowflakes) instead of silently stringifying it — the
underlying JSON encoding has no other key type, so a coerced key would come back out
of `decrypt_data` as a `str` and no longer match the original lookup key.
Dict keys must be `str`, at any nesting depth (including a dict nested inside a list
or tuple). `encrypt_data` raises `TypeError` on a non-str key anywhere in the payload
(e.g. an int-keyed dict of Discord snowflakes, even nested a few levels down) instead
of silently stringifying it — the underlying JSON encoding has no other key type, so a
coerced key would come back out of `decrypt_data` as a `str` and no longer match the
original lookup key.
For whole records: `decrypt_record(crypto, doc)` decrypts every `{secure, iv, data}`
field (nested up to `traversal_level`, default 2); `is_encrypted_record(doc)` reports
@@ -98,10 +100,11 @@ if is_encrypted_record(doc):
doc = decrypt_record(crypto, doc)
```
`is_encrypted_record` falls back to an unbounded-depth scan once `traversal_level` is
exhausted, so it reliably reports `True` for a blob left behind by a shallower
`is_encrypted_record` always falls back to an unbounded-depth scan whenever its bounded
pass finds nothing, so it reliably reports `True` for a blob left behind by a shallower
`decrypt_record`/`reencrypt` call — safe to use as a leftover-detecting audit after
rotation, regardless of how deep the blob is nested.
rotation, regardless of how deep the blob is nested (including inside a list or tuple
at any depth) and regardless of which `traversal_level` you pass; it never false-negatives.
Naming aliases (same objects): `EnvelopeCrypto` = `DocumentCrypto` = `RecordCrypto`
= `PCICrypto` (deprecated legacy alias). `decrypt_record` = `decrypt_document` =
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
[project]
name = "envelope_crypto"
version = "0.1.6"
version = "1.0.0"
description = "Envelope encryption (RSA-OAEP wrapped AES-256-GCM) for dict records — config-free, storage-agnostic, installable."
requires-python = ">=3.10"
dependencies = [
+8
View File
@@ -1,3 +1,5 @@
from importlib.metadata import version, PackageNotFoundError
from .envelope_crypto import (
EnvelopeCrypto,
DocumentCrypto,
@@ -12,6 +14,11 @@ from .envelope_crypto import (
fingerprint_data,
)
try:
__version__ = version("envelope_crypto")
except PackageNotFoundError:
__version__ = "0.0.0+unknown"
__all__ = [
"EnvelopeCrypto",
"DocumentCrypto",
@@ -24,4 +31,5 @@ __all__ = [
"decrypt_document",
"decrypt_dict",
"fingerprint_data",
"__version__",
]
+263 -229
View File
@@ -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
@@ -82,26 +36,22 @@ from cryptography.hazmat.primitives.serialization import load_ssh_public_key
_log = logging.getLogger(__name__)
_OAEP_PADDING = padding.OAEP(
mgf=padding.MGF1(algorithm=hashes.SHA256()),
algorithm=hashes.SHA256(),
label=None,
)
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 +61,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 +68,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,37 +84,38 @@ 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
"""unbounded-depth scan: does record (or anything nested under it, incl. list/tuple items) contain a blob"""
if isinstance(record, dict):
if _is_blob(record):
return True
return any(_has_encrypted_field(value) for value in record.values())
if isinstance(record, (list, tuple)):
return any(_has_encrypted_field(item) for item in record)
return False
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):
return False
if _is_blob(record):
return True
return any(_has_encrypted_field(value) for value in record.values())
def _non_str_key_types(data: Any) -> List[str]:
"""unbounded-depth scan collecting type names of non-str dict keys, incl. dicts nested inside list/tuple items"""
found: List[str] = []
if isinstance(data, dict):
for key, value in data.items():
if not isinstance(key, str):
found.append(type(key).__name__)
found.extend(_non_str_key_types(value))
elif isinstance(data, (list, tuple)):
for item in data:
found.extend(_non_str_key_types(item))
return found
def _require_rsa(key) -> None:
"""raise ValueError unless key is an RSA public or private key
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 +140,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 +167,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 +211,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 +241,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:
@@ -303,10 +283,22 @@ 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
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).
Args:
aes_key: the AES data key to wrap.
rsa_key: path to (or, if is_file=False, raw PEM/OpenSSH data of)
the RSA public key to wrap with.
is_file: True (default) treats rsa_key as a path.
Returns:
(fingerprint, wrapped_b64): the wrapping key's fingerprint and the
RSA-OAEP wrapped key, base64-encoded.
Raises:
ValueError: 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), or is malformed.
"""
if is_file:
with open(rsa_key, "rb") as key_file:
@@ -317,15 +309,7 @@ class EnvelopeCrypto:
public_key = _load_public_key(key_data)
_require_rsa(public_key)
wrapped = public_key.encrypt(
aes_key,
padding.OAEP(
mgf=padding.MGF1(algorithm=hashes.SHA256()),
algorithm=hashes.SHA256(),
label=None,
),
)
# fingerprint from the already-loaded public_key — no second open/parse of the file
wrapped = public_key.encrypt(aes_key, _OAEP_PADDING)
fingerprint = _fingerprint_of(public_key)
wrapped_b64 = base64.b64encode(wrapped).decode()
_log.info("wrapped data key for fingerprint %s", fingerprint[:8])
@@ -337,10 +321,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:
@@ -352,23 +346,28 @@ class EnvelopeCrypto:
_require_rsa(private_key)
wrapped = base64.b64decode(encrypted_key_base64)
aes_key = private_key.decrypt(
wrapped,
padding.OAEP(
mgf=padding.MGF1(algorithm=hashes.SHA256()),
algorithm=hashes.SHA256(),
label=None,
),
)
aes_key = private_key.decrypt(wrapped, _OAEP_PADDING)
_log.info("unwrapped data key with RSA private key")
return aes_key
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 +378,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 = {}
@@ -394,27 +404,23 @@ class EnvelopeCrypto:
def encrypt_data(self, data: Union[Dict[str, Any], str]) -> Dict[str, str]:
"""encrypt a dict or string under the data key with a unique IV
dict keys must be str: json.dumps silently stringifies int/float/bool/None
keys (e.g. a snowflake-int-keyed dict), which would make decrypt_data return
a dict that no longer matches the original by key type or identity — a
dict keys must be str, at any nesting depth (including dicts nested inside
lists/tuples): json.dumps silently stringifies int/float/bool/None keys
(e.g. a snowflake-int-keyed dict), which would make decrypt_data return a
dict that no longer matches the original by key type or identity - a
non-str key is rejected here instead, so the failure is loud at encrypt
time rather than a silent lookup miss after decrypt.
"""
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
non_str_key_types = _non_str_key_types(data)
if non_str_key_types:
raise TypeError(
"encrypt_data requires str dict keys, got non-str key(s): "
f"{[type(key).__name__ for key in non_str_keys]}"
"encrypt_data requires str dict keys (at any nesting depth), got non-str "
f"key(s): {non_str_key_types}"
)
data_str = json.dumps(data) if isinstance(data, dict) else data
@@ -430,20 +436,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 +470,32 @@ 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. the depth-limit raise below IS reached for a list-nested
blob when the cutoff dict scan finds it (it walks lists too), but a blob
one level shallower - hit during normal traversal instead of the cutoff
scan - is skipped silently; flatten list-nested blobs to dict fields before
rotation or they can 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,28 +514,56 @@ 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
def _is_encrypted_value(value: Any, traversal_level: int) -> bool:
"""bounded-pass check of a single field value, walking list/tuple items too
falls back to the unbounded _has_encrypted_field scan once traversal_level is
exhausted, so this can never return False for a value that still contains a
blob at any depth
"""
if _is_blob(value):
return True
if isinstance(value, dict):
if traversal_level > 0:
return is_encrypted_record(value, traversal_level - 1)
return _has_encrypted_field(value)
if isinstance(value, (list, tuple)):
return any(_is_encrypted_value(item, traversal_level) for item in value)
return False
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 always falls back to an unbounded-depth scan if
the bounded pass found nothing - safe to use as a leftover-detecting
post-rotation audit; never returns False for a record that still contains a
blob, at any depth (including one nested inside a list or tuple), for any
traversal_level value, odd or even (both passes walk list/tuple items, not
just dict values).
aliases: is_encrypted_document, is_encrypted_dict — same function
Args:
record: the record (or bare blob) to check.
traversal_level: bounded-pass nesting depth tried before the unbounded
fallback scan runs (default 2); purely a fast-path - the fallback
always makes the final call when the bounded pass finds nothing.
Returns:
True if record or anything nested under it is an encrypted blob.
aliases: is_encrypted_document, is_encrypted_dict - same function
"""
if not isinstance(record, dict):
return False
@@ -528,9 +577,8 @@ def is_encrypted_record(record, traversal_level: int = 2) -> bool:
if traversal_level > 0:
for value in record.values():
if isinstance(value, dict) and is_encrypted_record(value, traversal_level - 1):
if _is_encrypted_value(value, traversal_level - 1):
return True
return False
return any(_has_encrypted_field(value) for value in record.values())
@@ -538,12 +586,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")
@@ -569,52 +630,25 @@ 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, collision-free SHA-256 hex fingerprint of a dict
"""deterministic sha256 fingerprint of json-shaped dict data
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.
the input is a record sent over the wire, so it is json-serializable by contract:
keys are canonically sorted for an order-independent digest, separators are fixed for
whitespace stability, and any non-json scalar (uuid/decimal/path/datetime/objectid) is
stringified rather than crashing the encode. a callable value raises - its str() embeds a
memory address, which would make the digest vary across processes. never logs the data.
"""
normalized = _fingerprint_normalize(data)
encoded = json.dumps(normalized, sort_keys=True, default=_fingerprint_default)
def _stringify(value: object) -> str:
if callable(value):
raise TypeError(f"fingerprint_data: cannot fingerprint a callable: {value!r}")
return str(value)
encoded = json.dumps(data, sort_keys=True, separators=(",", ":"), default=_stringify)
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