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>
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.8
|
envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.10
|
||||||
```
|
```
|
||||||
|
|
||||||
Direct:
|
Direct:
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
pip install "envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.8"
|
pip install "envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.10"
|
||||||
```
|
```
|
||||||
|
|
||||||
Requires `cryptography` (pulled transitively).
|
Requires `cryptography` (pulled transitively).
|
||||||
|
|
||||||
Drop the `@v0.1.8` suffix from the line above to install the latest unpinned.
|
Drop the `@v0.1.10` suffix from the line above to install the latest unpinned.
|
||||||
|
|
||||||
## First-time setup
|
## First-time setup
|
||||||
|
|
||||||
@@ -80,10 +80,12 @@ 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
|
Dict keys must be `str`, at any nesting depth (including a dict nested inside a list
|
||||||
int-keyed dict of Discord snowflakes) instead of silently stringifying it — the
|
or tuple). `encrypt_data` raises `TypeError` on a non-str key anywhere in the payload
|
||||||
underlying JSON encoding has no other key type, so a coerced key would come back out
|
(e.g. an int-keyed dict of Discord snowflakes, even nested a few levels down) instead
|
||||||
of `decrypt_data` as a `str` and no longer match the original lookup key.
|
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
|
||||||
@@ -98,11 +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, including inside a list or
|
rotation, regardless of how deep the blob is nested (including inside a list or tuple
|
||||||
tuple at any depth.
|
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` =
|
||||||
|
|||||||
+1
-1
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
|
|||||||
|
|
||||||
[project]
|
[project]
|
||||||
name = "envelope_crypto"
|
name = "envelope_crypto"
|
||||||
version = "0.1.9"
|
version = "0.1.10"
|
||||||
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 = [
|
||||||
|
|||||||
@@ -94,6 +94,20 @@ def _has_encrypted_field(record: Any) -> bool:
|
|||||||
return False
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
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"""
|
||||||
|
found: List[str] = []
|
||||||
|
if isinstance(data, dict):
|
||||||
|
for key, value in data.items():
|
||||||
|
if not isinstance(key, str):
|
||||||
|
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:
|
def _require_rsa(key) -> None:
|
||||||
"""raise ValueError unless key is an RSA public or private key"""
|
"""raise ValueError unless key is an RSA public or private key"""
|
||||||
if not isinstance(key, (rsa.RSAPublicKey, rsa.RSAPrivateKey)):
|
if not isinstance(key, (rsa.RSAPublicKey, rsa.RSAPrivateKey)):
|
||||||
@@ -390,9 +404,10 @@ class EnvelopeCrypto:
|
|||||||
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
|
dict keys must be str, at any nesting depth (including dicts nested inside
|
||||||
keys (e.g. a snowflake-int-keyed dict), which would make decrypt_data return
|
lists/tuples): json.dumps silently stringifies int/float/bool/None keys
|
||||||
a dict that no longer matches the original by key type or identity - a
|
(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
|
non-str key is rejected here instead, so the failure is loud at encrypt
|
||||||
time rather than a silent lookup miss after decrypt.
|
time rather than a silent lookup miss after decrypt.
|
||||||
"""
|
"""
|
||||||
@@ -401,11 +416,11 @@ class EnvelopeCrypto:
|
|||||||
if not isinstance(data, (dict, str)):
|
if not isinstance(data, (dict, str)):
|
||||||
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):
|
if isinstance(data, dict):
|
||||||
non_str_keys = [key for key in data if not isinstance(key, str)]
|
non_str_key_types = _non_str_key_types(data)
|
||||||
if non_str_keys:
|
if non_str_key_types:
|
||||||
raise TypeError(
|
raise TypeError(
|
||||||
"encrypt_data requires str dict keys, got non-str key(s): "
|
"encrypt_data requires str dict keys (at any nesting depth), got non-str "
|
||||||
f"{[type(key).__name__ for key in non_str_keys]}"
|
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
|
||||||
@@ -509,11 +524,18 @@ PCICrypto = EnvelopeCrypto # deprecated legacy alias; remove after all systems
|
|||||||
|
|
||||||
|
|
||||||
def _is_encrypted_value(value: Any, traversal_level: int) -> bool:
|
def _is_encrypted_value(value: Any, traversal_level: int) -> bool:
|
||||||
"""bounded-pass check of a single field value, walking list/tuple items too"""
|
"""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):
|
if _is_blob(value):
|
||||||
return True
|
return True
|
||||||
if isinstance(value, dict):
|
if isinstance(value, dict):
|
||||||
return traversal_level > 0 and is_encrypted_record(value, traversal_level - 1)
|
if traversal_level > 0:
|
||||||
|
return is_encrypted_record(value, traversal_level - 1)
|
||||||
|
return _has_encrypted_field(value)
|
||||||
if isinstance(value, (list, tuple)):
|
if isinstance(value, (list, tuple)):
|
||||||
return any(_is_encrypted_value(item, traversal_level) for item in value)
|
return any(_is_encrypted_value(item, traversal_level) for item in value)
|
||||||
return False
|
return False
|
||||||
@@ -523,16 +545,18 @@ 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) as well as fields up to
|
checks `record` itself (the file-storage pattern) as well as fields up to
|
||||||
`traversal_level` deep, then falls back to an unbounded-depth scan - safe to
|
`traversal_level` deep, then always falls back to an unbounded-depth scan if
|
||||||
use as a leftover-detecting post-rotation audit; never returns False for a
|
the bounded pass found nothing - safe to use as a leftover-detecting
|
||||||
record that still contains a blob, at any depth, including one nested inside
|
post-rotation audit; never returns False for a record that still contains a
|
||||||
a list or tuple at any depth (both passes walk list/tuple items, not just
|
blob, at any depth (including one nested inside a list or tuple), for any
|
||||||
dict values).
|
traversal_level value, odd or even (both passes walk list/tuple items, not
|
||||||
|
just dict values).
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
record: the record (or bare blob) to check.
|
record: the record (or bare blob) to check.
|
||||||
traversal_level: bounded-pass nesting depth before the unbounded
|
traversal_level: bounded-pass nesting depth tried before the unbounded
|
||||||
fallback scan (default 2).
|
fallback scan runs (default 2); purely a fast-path - the fallback
|
||||||
|
always makes the final call when the bounded pass finds nothing.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
True if record or anything nested under it is an encrypted blob.
|
True if record or anything nested under it is an encrypted blob.
|
||||||
@@ -553,7 +577,6 @@ def is_encrypted_record(record, traversal_level: int = 2) -> bool:
|
|||||||
for value in record.values():
|
for value in record.values():
|
||||||
if _is_encrypted_value(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())
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user