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>
This commit is contained in:
@@ -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
@@ -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 = [
|
||||||
|
|||||||
@@ -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")
|
||||||
|
|||||||
Reference in New Issue
Block a user