envelope_crypto

Envelope encryption for dict records. A random AES-256-GCM data key (DEK) encrypts the data; that key is wrapped (RSA-OAEP) per authorized system's public key (KEK) and stored by the caller, keyed by fingerprint. Each system unwraps its own copy with its private key. The same envelope pattern KMS-style systems use — config-free and storage-agnostic.

Install

requirements.txt:

envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v1.1.0

Direct:

pip install "envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v1.1.0"

Requires cryptography (pulled transitively).

Drop the @v1.1.0 suffix from the line above to install the latest unpinned.

First-time setup

Run once, ever, to create the data key and authorize the first system. You need an RSA keypair first:

# generate an RSA keypair for the first system (PEM)
openssl genrsa -out local_priv.pem 4096
openssl rsa -in local_priv.pem -pubout -out local_pub.pem

from envelope_crypto import EnvelopeCrypto

# generate the DEK and wrap it for this system in one call
crypto, fingerprint, wrapped = EnvelopeCrypto.bootstrap("public_key.pem")

# verify the keypair actually round-trips BEFORE storing anything
crypto.self_test("public_key.pem", "private_key.pem")        # raises if keys don't pair

# store the wrapped key — this is now the ONLY record of the DEK
await db.create_document("keys", {"_id": fingerprint, "key": wrapped})

The plaintext DEK is never stored. It survives only as the RSA-wrapped copy, and is re-derived each boot by unwrapping. Never persist the plaintext key.

Boot (already set up)

crypto = EnvelopeCrypto()
fingerprint = crypto.get_rsa_key_fingerprint("public_key.pem")
record = await db.get_document("keys", {"_id": fingerprint})
if not record:
    raise RuntimeError("this system is not authorized")
crypto.initialize(crypto.decrypt_aes_key_with_rsa(record["key"], "private_key.pem"))
bot.crypto = crypto

The keys schema (_id = fingerprint, key = wrapped) is the caller's choice; 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

enc = crypto.encrypt_data({"ssn": "..."})      # -> {"secure": True, "iv": ..., "data": ..., "kid": ...}
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} 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 {secure, iv, data} blob (the file-storage pattern below, where the blob IS the whole document) — not just blobs nested under a key.

from envelope_crypto import is_encrypted_record, decrypt_record

if is_encrypted_record(doc):
    doc = decrypt_record(crypto, doc)

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 (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 = decrypt_dict; is_encrypted_record = is_encrypted_document = is_encrypted_dict.

Key ids

Every blob encrypt_data writes carries a "kid": the first 16 hex chars of sha256(DEK). It identifies which data key sealed the blob without revealing it.

• decrypt_data on a blob whose kid doesn't match the loaded key raises KeyMismatchError (a ValueError subclass, exported) naming both kids, e.g. blob was sealed under key id 3f9a..., but the loaded key is 81c0... — instead of the opaque InvalidTag you'd get from AES-GCM. Catch KeyMismatchError to tell "sealed under another key" apart from tampering (InvalidTag).
• decrypt_record keeps its contract: a mismatched field is logged (the log line names both kids) and left encrypted; it does not raise.
• reencrypt writes the destination key's kid automatically (it goes through encrypt_data).
• Blobs without kid (written before v1.1.0) decrypt exactly as before. v1.0.0 readers ignore the extra field, so mixed versions reading the same store are safe.

Keyed fingerprints

fingerprint_data is a plain SHA-256: fine for change detection on ordinary records, but for a low-entropy secret (card number, email, phone) anyone with the digest can brute-force the value back. Use keyed_fingerprint to dedup or look up records by a secret value without storing it:

fp = crypto.keyed_fingerprint(card_number, "card")       # HMAC-SHA256 hex
await db.create_document("cards", {"card_fp": fp, "card": crypto.encrypt_data(card_number)})

The digest is an HMAC keyed by an HKDF subkey of the DEK, so it is only computable by a system that holds the DEK. purpose is required and domain-separates digests: the same email under "email" and "login" gives unrelated values, so a digest from one table can't be matched against another. data may be str (UTF-8), bytes, or dict (the same canonical JSON as fingerprint_data); "abc" and b"abc" give the same digest, as do a dict and its canonical JSON string.

A DEK rotation changes every keyed fingerprint. After rotate_master_key, recompute stored keyed fingerprints with the new instance (decrypt the field, re-fingerprint) alongside reencrypt, or lookups against old digests will miss.

Authorize another system

An initialized system wraps the DEK for another system's public key. Returns the record to store.

fingerprint, wrapped = crypto.authorize_system(other_pub_path)
await db.create_document("keys", {"_id": fingerprint, "key": wrapped})

Only a system that already holds the DEK can authorize others.

Deauthorize

Delete that fingerprint's key record — the system can no longer unwrap at boot. This does not revoke a DEK already held in memory by a running system; rotate if a system is compromised.

Rotate (new key + re-encrypt)

Generate a new DEK, wrap for the still-authorized set, then re-encrypt existing data.

new_key, wrapped = crypto.rotate_master_key([pub_a, pub_b])   # omit a system to drop it

new_crypto = EnvelopeCrypto()
new_crypto.initialize(new_key)

# re-encrypt every record (caller owns the DB loop)
for doc in await db.get_documents("settings", {}):
    fresh = new_crypto.reencrypt(crypto, doc)                 # decrypt(old) -> encrypt(new)
    await db.update_document("settings", {"_id": doc["_id"]}, fresh)

# replace the key records
await db.delete_documents("keys", {})
for fingerprint, wrapped_key in wrapped.items():
    await db.create_document("keys", {"_id": fingerprint, "key": wrapped_key})

reencrypt(source_crypto, record) is a method on the destination (new-key) instance: it decrypts each encrypted field with source_crypto (old key) and re-encrypts with itself. Only {secure, ...} fields are touched — including record itself if it IS a {secure, iv, data} blob (the file-storage pattern).

Rotation must fail loud: a per-field decrypt failure raises, and so does a blob nested deeper than traversal_level — silently leaving it under the old key would strand it once the old key's wrapped-key record is deleted below. If you nest blobs deeper than the default traversal_level=2, pass a higher traversal_level or flatten the record first.

Storage patterns

The encrypted blob is just a dict — store it wherever:

• Mongo — store the dict directly (Mongo is dict-native).
• MariaDB / Postgres — json.dumps(enc) into a JSON (or TEXT) column; json.loads on read, then decrypt_data.
• File — json.dump(enc, f).

The lib never touches a database; only the caller's storage layer differs.

Notes

• shutdown() drops the key reference but cannot guarantee zeroing it from RAM (Python immutable bytes).
• A failed field decryption in decrypt_record is logged and left encrypted (the blob stays visible) rather than silently dropped.
• 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

Releases are tagged vX.Y.Z. The install line above pins a release; drop the @vX.Y.Z suffix to install the latest unpinned. Pin deliberately for reproducible installs.