14 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
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
4 changed files with 353 additions and 188 deletions
+33 -6
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@v1.0.0
``` ```
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@v1.0.0"
``` ```
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 `@v1.0.0` 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,13 @@ 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`, 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}` 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
@@ -86,10 +100,11 @@ if is_encrypted_record(doc):
doc = decrypt_record(crypto, doc) doc = decrypt_record(crypto, doc)
``` ```
`is_encrypted_record` falls back to an unbounded-depth scan once `traversal_level` is `is_encrypted_record` always falls back to an unbounded-depth scan whenever its bounded
exhausted, so it reliably reports `True` for a blob left behind by a shallower 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 `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` Naming aliases (same objects): `EnvelopeCrypto` = `DocumentCrypto` = `RecordCrypto`
= `PCICrypto` (deprecated legacy alias). `decrypt_record` = `decrypt_document` = = `PCICrypto` (deprecated legacy alias). `decrypt_record` = `decrypt_document` =
@@ -165,6 +180,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 = "1.0.0"
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 = [
+8
View File
@@ -1,3 +1,5 @@
from importlib.metadata import version, PackageNotFoundError
from .envelope_crypto import ( from .envelope_crypto import (
EnvelopeCrypto, EnvelopeCrypto,
DocumentCrypto, DocumentCrypto,
@@ -12,6 +14,11 @@ from .envelope_crypto import (
fingerprint_data, fingerprint_data,
) )
try:
__version__ = version("envelope_crypto")
except PackageNotFoundError:
__version__ = "0.0.0+unknown"
__all__ = [ __all__ = [
"EnvelopeCrypto", "EnvelopeCrypto",
"DocumentCrypto", "DocumentCrypto",
@@ -24,4 +31,5 @@ __all__ = [
"decrypt_document", "decrypt_document",
"decrypt_dict", "decrypt_dict",
"fingerprint_data", "fingerprint_data",
"__version__",
] ]
+311 -181
View File
@@ -1,63 +1,23 @@
""" """
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 and stored by
distribution. the wrapped key is stored by the caller, keyed by fingerprint; the caller, keyed by fingerprint. RSA-envelope only - a non-RSA key (e.g. Ed25519/EC)
each system unwraps its own copy with its private key. this is the same loads and fingerprints fine but raises ValueError at wrap/unwrap. never logs key
envelope-encryption pattern used by KMS-style systems. 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 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
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. note this 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 (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.
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.
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,20 +30,28 @@ 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__)
_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"""
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]): def _load_private_key(key_data: bytes, pw: Optional[bytes]):
"""load a PEM or OpenSSH private key, normalizing the missing-password error """load a PEM or OpenSSH private key, normalizing the password/encryption mismatch error"""
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.
"""
try: try:
return serialization.load_pem_private_key(key_data, password=pw) return serialization.load_pem_private_key(key_data, password=pw)
except ValueError as error: except ValueError as error:
@@ -91,26 +59,16 @@ 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
# 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:
"""base64 SHA-256 fingerprint of an already-loaded public key """base64 SHA-256 fingerprint of an already-loaded public key"""
factored so callers that already hold a loaded key (e.g. encrypt_aes_key_with_rsa)
don't re-open and re-parse the key file just to fingerprint it.
"""
key_bytes = public_key.public_bytes( key_bytes = public_key.public_bytes(
encoding=serialization.Encoding.DER, encoding=serialization.Encoding.DER,
format=serialization.PublicFormat.SubjectPublicKeyInfo, format=serialization.PublicFormat.SubjectPublicKeyInfo,
@@ -126,25 +84,38 @@ def _is_blob(value: Any) -> bool:
def _has_encrypted_field(record: Any) -> bool: def _has_encrypted_field(record: Any) -> bool:
"""unbounded-depth scan: does record (or anything nested under it) contain a blob """unbounded-depth scan: does record (or anything nested under it, 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. 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"""
if not isinstance(record, dict): found: List[str] = []
return False if isinstance(data, dict):
if _is_blob(record): for key, value in data.items():
return True if not isinstance(key, str):
return any(_has_encrypted_field(value) for value in record.values()) 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"""
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"""
load_ssh_public_key raises UnsupportedAlgorithm (not ValueError) on non-SSH/garbage
input; normalize it so a bad public key always surfaces as a clear ValueError,
consistent with the private-key path.
"""
try: try:
return serialization.load_pem_public_key(key_data) return serialization.load_pem_public_key(key_data)
except ValueError: except ValueError:
@@ -169,10 +140,20 @@ class EnvelopeCrypto:
def bootstrap(cls, rsa_public_key: str, is_file: bool = True) -> Tuple["EnvelopeCrypto", str, str]: def bootstrap(cls, rsa_public_key: str, is_file: bool = True) -> Tuple["EnvelopeCrypto", str, str]:
"""first-time setup: generate a DEK and wrap it for the first system """first-time setup: generate a DEK and wrap it for the first system
returns (crypto, fingerprint, wrapped_key) — an initialized instance plus the plaintext DEK is never returned or persisted; it survives only as the
the record to store as the first authorization. the plaintext DEK is never wrapped copy. run self_test before storing to confirm the keypair round-trips.
returned or persisted; it survives only as the wrapped copy. run self_test
before storing to confirm the keypair round-trips. Args:
rsa_public_key: path to (or, if is_file=False, raw PEM/OpenSSH data of)
the first system's RSA public key.
is_file: True (default) treats rsa_public_key as a path.
Returns:
(crypto, fingerprint, wrapped_key): an initialized instance plus the
record to store as the first authorization.
Raises:
ValueError: rsa_public_key is not an RSA key, or is malformed.
""" """
crypto = cls() crypto = cls()
crypto.initialize(crypto.create_aes_key()) crypto.initialize(crypto.create_aes_key())
@@ -185,10 +166,24 @@ 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. run after bootstrap or as a health check.
match. run after bootstrap (or anytime as a health check) to catch a bad
keypair or wrong key path before relying on it. returns True on success. 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: 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 +194,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 +207,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")
@@ -238,12 +241,23 @@ class EnvelopeCrypto:
) -> str: ) -> str:
"""return a base64 SHA-256 fingerprint of an RSA key for identification """return a base64 SHA-256 fingerprint of an RSA key for identification
for an encrypted private key (is_private=True), pass its `password`; an always fingerprints the public half, so a private key and its public key
unencrypted key ignores it. fingerprinting always uses the public half, so a match. PEM and OpenSSH accepted (mirrors decrypt_aes_key_with_rsa).
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 Args:
encrypted key with no/wrong password raises ValueError with a clear message key_path_or_data: path to (or, if is_file=False, raw PEM/OpenSSH data
(cryptography raises TypeError for the missing-password case — normalized here). of) the key.
is_private: fingerprint the public half of a private key; pass its
`password` if encrypted (an unencrypted key ignores it).
is_file: True (default) treats key_path_or_data as a path.
password: password for an encrypted private key, if is_private.
Returns:
base64 SHA-256 fingerprint.
Raises:
ValueError: key is malformed, or password/encryption mismatch
(cryptography's raw TypeError normalized here).
""" """
if is_file: if is_file:
with open(key_path_or_data, "rb") as key_file: with open(key_path_or_data, "rb") as key_file:
@@ -269,7 +283,23 @@ 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
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: 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,49 +307,67 @@ 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, _OAEP_PADDING)
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
fingerprint = _fingerprint_of(public_key) fingerprint = _fingerprint_of(public_key)
wrapped_b64 = base64.b64encode(wrapped).decode() wrapped_b64 = base64.b64encode(wrapped).decode()
_log.info("wrapped data key for fingerprint %s", fingerprint[:8]) _log.info("wrapped data key for fingerprint %s", fingerprint[:8])
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() 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:
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(wrapped, _OAEP_PADDING)
wrapped,
padding.OAEP(
mgf=padding.MGF1(algorithm=hashes.SHA256()),
algorithm=hashes.SHA256(),
label=None,
),
)
_log.info("unwrapped data key with RSA private key") _log.info("unwrapped data key with RSA private key")
return aes_key return aes_key
def authorize_system(self, rsa_public_key: str, is_file: bool = True) -> Tuple[str, str]: def authorize_system(self, rsa_public_key: str, is_file: bool = True) -> Tuple[str, str]:
"""wrap the current data key for another system's public key """wrap the current data key for another system's public key
returns (fingerprint, wrapped_b64) for the caller to store as that requires this instance to already hold the data key - only an
system's key-authorization record. requires this instance to already authorized system can authorize others.
hold the data key — only an authorized system can authorize others.
Args:
rsa_public_key: path to (or, if is_file=False, raw PEM/OpenSSH data
of) the other system's RSA public key.
is_file: True (default) treats rsa_public_key as a path.
Returns:
(fingerprint, wrapped_b64) for the caller to store as that system's
key-authorization record.
Raises:
ValueError: this instance is not initialized, or rsa_public_key is
not an RSA key.
""" """
if not self.master_key: if not self.master_key:
raise ValueError("cannot authorize another system: not initialized") raise ValueError("cannot authorize another system: not initialized")
@@ -330,9 +378,20 @@ class EnvelopeCrypto:
) -> Tuple[bytes, Dict[str, str]]: ) -> Tuple[bytes, Dict[str, str]]:
"""generate a NEW data key and wrap it for each authorized public key """generate a NEW data key and wrap it for each authorized public key
returns (new_key, {fingerprint: wrapped_b64}). does NOT re-encrypt existing does NOT re-encrypt existing data - build a new instance with the new key
data — build a new instance with the new key and call reencrypt() on each and call reencrypt() on each record. systems not in the list get no
record. systems not in the list get no wrapped copy (deauthorized). wrapped copy (deauthorized).
Args:
authorized_public_keys: RSA public keys (paths, or raw data if
is_file=False) to wrap the new key for.
is_file: True (default) treats each entry as a path.
Returns:
(new_key, {fingerprint: wrapped_b64}).
Raises:
ValueError: any entry in authorized_public_keys is not an RSA key.
""" """
new_key = self.create_aes_key() new_key = self.create_aes_key()
wrapped = {} wrapped = {}
@@ -343,13 +402,26 @@ 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, 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: if not self.master_key:
raise ValueError("not initialized with data key") raise ValueError("not initialized with data key")
if not isinstance(data, (dict, str)): if not isinstance(data, (dict, str)):
# a non-dict/non-str would .encode()-fail with an opaque AttributeError below;
# reject it clearly (decrypt_data only round-trips dict or str anyway)
raise TypeError(f"encrypt_data expects a dict or str, got {type(data).__name__}") raise TypeError(f"encrypt_data expects a dict or str, got {type(data).__name__}")
if isinstance(data, dict):
non_str_key_types = _non_str_key_types(data)
if non_str_key_types:
raise TypeError(
"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 data_str = json.dumps(data) if isinstance(data, dict) else data
iv = os.urandom(12) iv = os.urandom(12)
@@ -364,21 +436,25 @@ class EnvelopeCrypto:
def decrypt_data(self, encrypted_data: Dict[str, str]) -> Union[Dict[str, Any], str]: def decrypt_data(self, encrypted_data: Dict[str, str]) -> Union[Dict[str, Any], str]:
"""decrypt a {secure, iv, data} blob; returns the original dict or string """decrypt a {secure, iv, data} blob; returns the original dict or string
`encrypt_data` only json-encodes dicts (a string is stored verbatim), so decrypt a json-shaped but non-object plaintext ('123', 'true', '[1,2]') round-trips
only treats a json-OBJECT plaintext as a dict and returns everything else as the as a STRING, not int/bool/list. one irreducible ambiguity: a string whose
raw string. this keeps the type faithful for the common cases: a string that is exact value is a json object ('{"a":1}') decrypts to a dict, indistinguishable
json-shaped but NOT an object ('123'->'123', 'true'->'true', '[1,2]'->'[1,2]') from a stored dict - don't store a bare json-object string if you need it
round-trips as a STRING, not an int/bool/list. the one irreducible ambiguity: a back as a string.
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 Args:
bare string that is a json object if you need it back as a string. existing stored encrypted_data: a {secure, iv, data} blob from encrypt_data.
blobs are unaffected — a dict was stored as a json object and still parses to a dict.
Returns:
the original dict or string.
Raises:
ValueError: instance not initialized, or encrypted_data is not a
{secure, iv, data} blob.
""" """
if not self.master_key: if not self.master_key:
raise ValueError("not initialized with data key") raise ValueError("not initialized with data key")
if not isinstance(encrypted_data, dict) or "iv" not in encrypted_data or "data" not in encrypted_data: if not isinstance(encrypted_data, dict) or "iv" not in encrypted_data or "data" not in encrypted_data:
# a structurally-malformed blob would raise a raw KeyError/TypeError; surface
# a clear ValueError instead, matching the documented {secure, iv, data} shape
raise ValueError("decrypt_data expects a {secure, iv, data} blob") raise ValueError("decrypt_data expects a {secure, iv, data} blob")
iv = base64.b64decode(encrypted_data["iv"]) iv = base64.b64decode(encrypted_data["iv"])
@@ -396,24 +472,30 @@ class EnvelopeCrypto:
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. only {secure, iv, data} fields are touched; plaintext fields are left 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. 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`.
if `record` itself is a {secure, iv, data} blob (the file-storage pattern, where traversal recurses into nested DICTS only - a blob nested inside a LIST is
the blob IS the whole document) it is re-encrypted directly and the result is not re-encrypted. the depth-limit raise below IS reached for a list-nested
returned in place of `record` — not nested under a key. 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.
unlike decrypt_record (which logs a failed field and leaves it encrypted), a Args:
per-field decrypt failure here RAISES — rotation must fail loud, since silently source_crypto: instance holding the old (source) key.
keeping a field under the old key would lose it once the old key is retired. for record: the record to re-encrypt. not mutated; a new dict is returned.
the same reason, a blob nested DEEPER than `traversal_level` also RAISES instead traversal_level: max nesting depth to recurse into (default 2).
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 Returns:
re-encrypted and is NOT covered by the depth-limit raise above. records in this a new dict with encrypted fields re-wrapped under this instance's key.
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 Raises:
rotation or they'll be silently left under the old key. ValueError: destination not initialized with a data key; a per-field
decrypt failure (unlike decrypt_record, this fails loud rather
than silently stranding a field under the old key); or a blob
nested deeper than `traversal_level` - raise traversal_level or
flatten the record before rotation instead.
""" """
if not self.master_key: if not self.master_key:
raise ValueError("destination not initialized with data key") raise ValueError("destination not initialized with data key")
@@ -432,29 +514,56 @@ class EnvelopeCrypto:
raise ValueError( raise ValueError(
f"reencrypt: field {key!r} contains an encrypted blob nested deeper " f"reencrypt: field {key!r} contains an encrypted blob nested deeper "
"than traversal_level; increase traversal_level or flatten the record " "than traversal_level; increase traversal_level or flatten the record "
"before rotation leaving it would strand the field under the old key" "before rotation - leaving it would strand the field under the old key"
) )
return result return result
# naming aliases same class # naming aliases - same class
DocumentCrypto = EnvelopeCrypto DocumentCrypto = EnvelopeCrypto
RecordCrypto = EnvelopeCrypto RecordCrypto = EnvelopeCrypto
PCICrypto = EnvelopeCrypto # deprecated legacy alias; remove after all systems migrate PCICrypto = EnvelopeCrypto # deprecated legacy alias; remove after all systems migrate
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: def is_encrypted_record(record, traversal_level: int = 2) -> bool:
"""return whether a record has any encrypted ({secure, iv, data}) fields """return whether a record has any encrypted ({secure, iv, data}) fields
checks `record` itself (the file-storage pattern stores the blob AS the whole checks `record` itself (the file-storage pattern) as well as fields up to
document, not nested under a key) as well as fields up to `traversal_level` deep. `traversal_level` deep, then always falls back to an unbounded-depth scan if
beyond that bounded pass, this ALSO does an unbounded-depth scan before giving up — the bounded pass found nothing - safe to use as a leftover-detecting
so a blob left behind by a shallower decrypt_record/reencrypt call (nested deeper post-rotation audit; never returns False for a record that still contains a
than their traversal_level) is still reported as encrypted. this makes the function blob, at any depth (including one nested inside a list or tuple), for any
safe to use as a leftover-detecting post-rotation audit: it never returns False for traversal_level value, odd or even (both passes walk list/tuple items, not
a record that still contains a blob, at any depth. 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): if not isinstance(record, dict):
return False return False
@@ -468,9 +577,8 @@ def is_encrypted_record(record, traversal_level: int = 2) -> bool:
if traversal_level > 0: if traversal_level > 0:
for value in record.values(): 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 True
return False
return any(_has_encrypted_field(value) for value in record.values()) return any(_has_encrypted_field(value) for value in record.values())
@@ -478,16 +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]: 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
blob IS the whole document) it is decrypted directly and the decrypted value is decrypted directly and the value (dict or string, see decrypt_data) is
(dict or string — see decrypt_data) is returned in place of `record`. 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.
failures on a single field are logged and that field is left encrypted, so a Args:
partial failure is visible (the {secure,...} blob remains) rather than silent. a crypto: instance holding the data key to decrypt with.
failure decrypting `record` itself (the self-blob case above) is likewise logged record: the record (or bare blob) to decrypt. not mutated.
and the still-encrypted blob is returned unchanged. traversal_level: max nesting depth to recurse into (default 2).
aliases: decrypt_document, decrypt_dict — same function Returns:
a new dict with encrypted fields decrypted (or record unchanged if not
a dict).
Raises:
ValueError: crypto is not initialized with a data key.
aliases: decrypt_document, decrypt_dict - same function
""" """
if not crypto.master_key: if not crypto.master_key:
raise ValueError("not initialized with data key") raise ValueError("not initialized with data key")
@@ -514,11 +631,24 @@ def decrypt_record(crypto: EnvelopeCrypto, record, traversal_level: int = 2) ->
def fingerprint_data(data: dict) -> str: def fingerprint_data(data: dict) -> str:
"""return a deterministic SHA-256 hex fingerprint of a dict""" """deterministic sha256 fingerprint of json-shaped dict data
return hashlib.sha256(json.dumps(data, sort_keys=True).encode()).hexdigest()
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.
"""
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_document = is_encrypted_record
is_encrypted_dict = is_encrypted_record is_encrypted_dict = is_encrypted_record
decrypt_document = decrypt_record decrypt_document = decrypt_record