1 Commits
Author SHA1 Message Date
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
3 changed files with 61 additions and 12 deletions
+15 -3
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@v0.1.5
``` ```
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@v0.1.5"
``` ```
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 `@v0.1.5` 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,11 @@ 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`. `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.
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
+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 = "0.1.5"
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 = [
+44 -7
View File
@@ -55,6 +55,11 @@ 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 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. encrypted blob is a plain dict; store it in mongo, a sql json column, or a file.
encrypt_data/decrypt_data round-trip a dict only when every key is a str —
json (the wire format under the hood) has no other key type, so encrypt_data
raises TypeError on a non-str key (e.g. an int-keyed dict of discord snowflakes)
rather than silently stringifying it and losing the original key on decrypt.
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 variants are the same functions — use whichever fits your storage. function variants are the same functions — use whichever fits your storage.
@@ -188,7 +193,11 @@ class EnvelopeCrypto:
round-trips sample data through this instance's DEK, then wraps the DEK round-trips sample data through this instance's DEK, then wraps the DEK
with the public key and unwraps with the private key, confirming they with the public key and unwraps with the private key, confirming they
match. run after bootstrap (or anytime as a health check) to catch a bad 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. keypair or wrong key path before relying on it. is_file (default True) is
threaded through to both the public and private key loads, so
is_file=False correctly treats both rsa_public_key and rsa_private_key as
in-memory PEM/OpenSSH data rather than opening either as a file path.
returns True on success.
""" """
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 +208,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, "
@@ -293,12 +302,21 @@ class EnvelopeCrypto:
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:
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.
"""
if is_file:
with open(rsa_private_key, "rb") as key_file:
key_data = key_file.read() 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)
@@ -343,13 +361,30 @@ 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: 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; # 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) # 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_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
raise TypeError(
"encrypt_data requires str dict keys, got non-str key(s): "
f"{[type(key).__name__ for key in non_str_keys]}"
)
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)
@@ -373,6 +408,8 @@ class EnvelopeCrypto:
without a type marker it is indistinguishable from a stored dict — don't store a without a type marker it is indistinguishable from a stored dict — don't store a
bare string that is a json object if you need it back as a string. existing stored bare string that is a json object if you need it back as a string. existing stored
blobs are unaffected — a dict was stored as a json object and still parses to a dict. blobs are unaffected — a dict was stored as a json object and still parses to a dict.
dict KEYS round-trip faithfully only because encrypt_data now requires str keys —
json has no other key type, so this is the only shape decrypt_data ever sees.
""" """
if not self.master_key: if not self.master_key:
raise ValueError("not initialized with data key") raise ValueError("not initialized with data key")