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@v0.1.10

Direct:

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

Requires cryptography (pulled transitively).

Drop the @v0.1.10 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": ...}
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.

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.