|
|
|
@@ -1,16 +1,18 @@
|
|
|
|
|
"""
|
|
|
|
|
envelope encryption for dict records
|
|
|
|
|
|
|
|
|
|
hybrid encryption: a random AES-256-GCM data key (DEK) encrypts the data, and
|
|
|
|
|
that key is wrapped (RSA-OAEP) per authorized system's public key (KEK) for
|
|
|
|
|
distribution. the wrapped key is stored by the caller, keyed by fingerprint;
|
|
|
|
|
each system unwraps its own copy with its private key. this is the same
|
|
|
|
|
envelope-encryption pattern used by KMS-style systems.
|
|
|
|
|
hybrid encryption: a random AES-256-GCM data key (DEK) encrypts the data, wrapped
|
|
|
|
|
(RSA-OAEP) per authorized system's public key (KEK) for distribution. the wrapped
|
|
|
|
|
key is stored by the caller, keyed by fingerprint; each system unwraps its own copy
|
|
|
|
|
with its private key. same pattern KMS-style systems use. RSA-envelope only — a
|
|
|
|
|
non-RSA key (e.g. Ed25519/EC) loads and fingerprints fine but raises ValueError at
|
|
|
|
|
wrap/unwrap. never logs key material (DEK, PEM, wrapped key) — only fingerprints
|
|
|
|
|
and counts.
|
|
|
|
|
|
|
|
|
|
from envelope_crypto import EnvelopeCrypto
|
|
|
|
|
|
|
|
|
|
crypto = EnvelopeCrypto()
|
|
|
|
|
crypto.initialize(master_key) # 32-byte AES DEK
|
|
|
|
|
crypto.initialize(master_key) # exactly 32 bytes (AES-256)
|
|
|
|
|
enc = crypto.encrypt_data({"ssn": "..."}) # -> {secure, iv, data}
|
|
|
|
|
plain = crypto.decrypt_data(enc) # -> original
|
|
|
|
|
|
|
|
|
@@ -32,9 +34,8 @@ authorize another system (this instance must already hold the DEK):
|
|
|
|
|
fp, wrapped = crypto.authorize_system(other_pub_path)
|
|
|
|
|
caller_store({"_id": fp, "key": wrapped})
|
|
|
|
|
|
|
|
|
|
deauthorize: caller deletes that fingerprint's record. note this stops future
|
|
|
|
|
unwraps but does not revoke a DEK already in a running system's memory — rotate
|
|
|
|
|
if compromised.
|
|
|
|
|
deauthorize: caller deletes that fingerprint's record. stops future unwraps but
|
|
|
|
|
does not revoke a DEK already in a running system's memory — rotate if compromised.
|
|
|
|
|
|
|
|
|
|
rotate (new DEK + re-encrypt): generate a new DEK, wrap for the still-authorized
|
|
|
|
|
set, then re-encrypt existing records old -> new:
|
|
|
|
@@ -45,19 +46,24 @@ set, then re-encrypt existing records old -> new:
|
|
|
|
|
caller_update(new_crypto.reencrypt(crypto, record))
|
|
|
|
|
|
|
|
|
|
reencrypt/is_encrypted_record/decrypt_record all detect a bare {secure, iv, data}
|
|
|
|
|
blob used AS the whole record (the file-storage pattern), not just blobs nested
|
|
|
|
|
under a key. reencrypt fails loud (raises) rather than silently leaving a field
|
|
|
|
|
under the old key — including when a blob sits deeper than `traversal_level`;
|
|
|
|
|
is_encrypted_record falls back to an unbounded-depth scan past traversal_level so
|
|
|
|
|
it reliably catches leftovers as a post-rotation audit.
|
|
|
|
|
blob used AS the whole record (file-storage pattern), not just blobs nested under a
|
|
|
|
|
key. reencrypt fails loud (raises) rather than silently leaving a field under the
|
|
|
|
|
old key — including a blob nested deeper than `traversal_level`; is_encrypted_record
|
|
|
|
|
falls back to an unbounded-depth scan past traversal_level to reliably catch
|
|
|
|
|
leftovers as a post-rotation audit.
|
|
|
|
|
|
|
|
|
|
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
|
|
|
|
|
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) 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
|
|
|
|
|
aliases (PCICrypto is a deprecated legacy alias). the document/record/dict
|
|
|
|
|
function variants are the same functions — use whichever fits your storage.
|
|
|
|
|
aliases (PCICrypto is a deprecated legacy alias). the document/record/dict function
|
|
|
|
|
variants are the same functions — use whichever fits your storage.
|
|
|
|
|
"""
|
|
|
|
|
|
|
|
|
|
import os
|
|
|
|
@@ -70,19 +76,31 @@ from typing import Any, Dict, List, Optional, Tuple, Union
|
|
|
|
|
|
|
|
|
|
from cryptography.exceptions import UnsupportedAlgorithm
|
|
|
|
|
from cryptography.hazmat.primitives import hashes, serialization
|
|
|
|
|
from cryptography.hazmat.primitives.asymmetric import padding
|
|
|
|
|
from cryptography.hazmat.primitives.asymmetric import padding, rsa
|
|
|
|
|
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
|
|
|
|
from cryptography.hazmat.primitives.serialization import load_ssh_public_key
|
|
|
|
|
|
|
|
|
|
_log = logging.getLogger(__name__)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _load_private_key(key_data: bytes, pw: Optional[bytes]):
|
|
|
|
|
"""load a PEM or OpenSSH private key, normalizing the missing-password error
|
|
|
|
|
def _password_mismatch_message(pw: Optional[bytes]) -> str:
|
|
|
|
|
"""clear ValueError text for a TypeError raised by a password/encryption mismatch
|
|
|
|
|
|
|
|
|
|
cryptography raises TypeError when a key is encrypted but no password was given
|
|
|
|
|
(PEM raises it on the first call; OpenSSH raises it inside the openssh fallback).
|
|
|
|
|
both are normalized to a clear ValueError so callers see one error type.
|
|
|
|
|
cryptography raises TypeError for both directions: encrypted key + no password,
|
|
|
|
|
and unencrypted key + a password given. branch on which the caller supplied so
|
|
|
|
|
the message matches the actual case instead of always claiming "encrypted".
|
|
|
|
|
"""
|
|
|
|
|
if pw is not None:
|
|
|
|
|
return "password was given but private key is not encrypted"
|
|
|
|
|
return "private key is encrypted but no password was provided"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _load_private_key(key_data: bytes, pw: Optional[bytes]):
|
|
|
|
|
"""load a PEM or OpenSSH private key, normalizing the password/encryption mismatch error
|
|
|
|
|
|
|
|
|
|
cryptography raises TypeError for a password/encryption mismatch (PEM raises it on
|
|
|
|
|
the first call; OpenSSH raises it inside the openssh fallback), normalized here to a
|
|
|
|
|
clear ValueError so callers see one error type with a message matching the actual case.
|
|
|
|
|
"""
|
|
|
|
|
try:
|
|
|
|
|
return serialization.load_pem_private_key(key_data, password=pw)
|
|
|
|
@@ -91,18 +109,14 @@ def _load_private_key(key_data: bytes, pw: Optional[bytes]):
|
|
|
|
|
try:
|
|
|
|
|
return serialization.load_ssh_private_key(key_data, password=pw)
|
|
|
|
|
except TypeError as ssh_error:
|
|
|
|
|
raise ValueError(
|
|
|
|
|
"private key is encrypted but no password was provided"
|
|
|
|
|
) from ssh_error
|
|
|
|
|
raise ValueError(_password_mismatch_message(pw)) from ssh_error
|
|
|
|
|
if pw is not None:
|
|
|
|
|
# a password was given but the PEM load still failed — most likely a wrong
|
|
|
|
|
# password; give a clearer message than cryptography's raw "Bad decrypt"
|
|
|
|
|
raise ValueError("could not load private key (wrong password or malformed key)") from error
|
|
|
|
|
raise error
|
|
|
|
|
except TypeError as error:
|
|
|
|
|
raise ValueError(
|
|
|
|
|
"private key is encrypted but no password was provided"
|
|
|
|
|
) from error
|
|
|
|
|
raise ValueError(_password_mismatch_message(pw)) from error
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _fingerprint_of(public_key) -> str:
|
|
|
|
@@ -138,6 +152,18 @@ def _has_encrypted_field(record: Any) -> bool:
|
|
|
|
|
return any(_has_encrypted_field(value) for value in record.values())
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _require_rsa(key) -> None:
|
|
|
|
|
"""raise ValueError unless key is an RSA public or private key
|
|
|
|
|
|
|
|
|
|
this lib is RSA-envelope only: get_rsa_key_fingerprint accepts any key type
|
|
|
|
|
(fingerprinting is algorithm-agnostic), but wrap/unwrap calls RSA-OAEP methods
|
|
|
|
|
that don't exist on e.g. Ed25519/EC keys and would otherwise crash raw with an
|
|
|
|
|
AttributeError far from a clear cause.
|
|
|
|
|
"""
|
|
|
|
|
if not isinstance(key, (rsa.RSAPublicKey, rsa.RSAPrivateKey)):
|
|
|
|
|
raise ValueError(f"RSA key required for envelope wrap/unwrap, got {type(key).__name__}")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _load_public_key(key_data: bytes):
|
|
|
|
|
"""load a PEM or OpenSSH public key, normalizing non-key input to ValueError
|
|
|
|
|
|
|
|
|
@@ -185,10 +211,11 @@ class EnvelopeCrypto:
|
|
|
|
|
) -> bool:
|
|
|
|
|
"""verify the full pipeline against a keypair; raises on any mismatch
|
|
|
|
|
|
|
|
|
|
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
|
|
|
|
|
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.
|
|
|
|
|
round-trips sample data through this instance's DEK, then wraps and unwraps
|
|
|
|
|
the DEK with the given keypair, confirming they match. run after bootstrap
|
|
|
|
|
(or as a health check) to catch a bad keypair or wrong path before relying
|
|
|
|
|
on it. is_file threads through to both key loads (is_file=False treats both
|
|
|
|
|
as in-memory PEM/OpenSSH data, not paths). returns True on success.
|
|
|
|
|
"""
|
|
|
|
|
if not self.master_key:
|
|
|
|
|
raise ValueError("self_test: not initialized with a key")
|
|
|
|
@@ -199,7 +226,7 @@ class EnvelopeCrypto:
|
|
|
|
|
|
|
|
|
|
_, wrapped = self.encrypt_aes_key_with_rsa(self.master_key, rsa_public_key, is_file=is_file)
|
|
|
|
|
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:
|
|
|
|
|
raise RuntimeError(
|
|
|
|
|
"self_test: key unwrap failed (public/private keys do not pair, "
|
|
|
|
@@ -212,7 +239,15 @@ class EnvelopeCrypto:
|
|
|
|
|
return True
|
|
|
|
|
|
|
|
|
|
def initialize(self, master_key: bytes) -> None:
|
|
|
|
|
"""arm the instance with the AES data key (DEK)"""
|
|
|
|
|
"""arm the instance with the AES data key (DEK)
|
|
|
|
|
|
|
|
|
|
requires exactly 32 bytes (AES-256): a shorter key would silently downgrade
|
|
|
|
|
to AES-128/192 with no warning, and a non-bytes value would otherwise fail
|
|
|
|
|
late and opaquely at first encrypt/decrypt — both rejected here instead.
|
|
|
|
|
never logs the key material itself.
|
|
|
|
|
"""
|
|
|
|
|
if not isinstance(master_key, bytes) or len(master_key) != 32:
|
|
|
|
|
raise ValueError("master_key must be exactly 32 bytes (AES-256)")
|
|
|
|
|
self.master_key = master_key
|
|
|
|
|
_log.info("crypto initialized with data key")
|
|
|
|
|
|
|
|
|
@@ -239,11 +274,10 @@ class EnvelopeCrypto:
|
|
|
|
|
"""return a base64 SHA-256 fingerprint of an RSA key for identification
|
|
|
|
|
|
|
|
|
|
for an encrypted private key (is_private=True), pass its `password`; an
|
|
|
|
|
unencrypted key ignores it. fingerprinting always uses the public half, so a
|
|
|
|
|
private and its public key produce the same fingerprint. PEM and OpenSSH
|
|
|
|
|
private-key formats are both accepted (mirrors decrypt_aes_key_with_rsa). an
|
|
|
|
|
encrypted key with no/wrong password raises ValueError with a clear message
|
|
|
|
|
(cryptography raises TypeError for the missing-password case — normalized here).
|
|
|
|
|
unencrypted key ignores it. always fingerprints the public half, so a
|
|
|
|
|
private key and its public key match. PEM and OpenSSH accepted (mirrors
|
|
|
|
|
decrypt_aes_key_with_rsa). a password/encryption mismatch raises a clear
|
|
|
|
|
ValueError (cryptography's raw TypeError normalized here).
|
|
|
|
|
"""
|
|
|
|
|
if is_file:
|
|
|
|
|
with open(key_path_or_data, "rb") as key_file:
|
|
|
|
@@ -269,7 +303,11 @@ class EnvelopeCrypto:
|
|
|
|
|
def encrypt_aes_key_with_rsa(
|
|
|
|
|
self, aes_key: bytes, rsa_key: str, is_file: bool = True
|
|
|
|
|
) -> Tuple[str, str]:
|
|
|
|
|
"""wrap an AES key with an RSA public key; returns (fingerprint, wrapped_b64)"""
|
|
|
|
|
"""wrap an AES key with an RSA public key; returns (fingerprint, wrapped_b64)
|
|
|
|
|
|
|
|
|
|
raises ValueError if rsa_key is not an RSA key (this lib is RSA-envelope only;
|
|
|
|
|
e.g. an Ed25519/EC key loads and fingerprints fine but cannot wrap).
|
|
|
|
|
"""
|
|
|
|
|
if is_file:
|
|
|
|
|
with open(rsa_key, "rb") as key_file:
|
|
|
|
|
key_data = key_file.read()
|
|
|
|
@@ -277,6 +315,7 @@ class EnvelopeCrypto:
|
|
|
|
|
key_data = rsa_key.encode() if isinstance(rsa_key, str) else rsa_key
|
|
|
|
|
|
|
|
|
|
public_key = _load_public_key(key_data)
|
|
|
|
|
_require_rsa(public_key)
|
|
|
|
|
|
|
|
|
|
wrapped = public_key.encrypt(
|
|
|
|
|
aes_key,
|
|
|
|
@@ -293,14 +332,24 @@ class EnvelopeCrypto:
|
|
|
|
|
return fingerprint, wrapped_b64
|
|
|
|
|
|
|
|
|
|
def decrypt_aes_key_with_rsa(
|
|
|
|
|
self, encrypted_key_base64: str, rsa_private_key_path: str,
|
|
|
|
|
password: Optional[str] = None,
|
|
|
|
|
self, encrypted_key_base64: str, rsa_private_key: str,
|
|
|
|
|
is_file: bool = True, password: Optional[str] = None,
|
|
|
|
|
) -> bytes:
|
|
|
|
|
"""unwrap an AES key with an RSA private key"""
|
|
|
|
|
with open(rsa_private_key_path, "rb") as key_file:
|
|
|
|
|
key_data = key_file.read()
|
|
|
|
|
"""unwrap an AES key with an RSA private key
|
|
|
|
|
|
|
|
|
|
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()
|
|
|
|
|
else:
|
|
|
|
|
key_data = rsa_private_key.encode() if isinstance(rsa_private_key, str) else rsa_private_key
|
|
|
|
|
pw = password.encode() if password else None
|
|
|
|
|
private_key = _load_private_key(key_data, pw)
|
|
|
|
|
_require_rsa(private_key)
|
|
|
|
|
|
|
|
|
|
wrapped = base64.b64decode(encrypted_key_base64)
|
|
|
|
|
aes_key = private_key.decrypt(
|
|
|
|
@@ -343,13 +392,30 @@ class EnvelopeCrypto:
|
|
|
|
|
return new_key, wrapped
|
|
|
|
|
|
|
|
|
|
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:
|
|
|
|
|
raise ValueError("not initialized with data key")
|
|
|
|
|
if not isinstance(data, (dict, str)):
|
|
|
|
|
# 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)
|
|
|
|
|
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
|
|
|
|
|
iv = os.urandom(12)
|
|
|
|
@@ -364,15 +430,14 @@ class EnvelopeCrypto:
|
|
|
|
|
def decrypt_data(self, encrypted_data: Dict[str, str]) -> Union[Dict[str, Any], str]:
|
|
|
|
|
"""decrypt a {secure, iv, data} blob; returns the original dict or string
|
|
|
|
|
|
|
|
|
|
`encrypt_data` only json-encodes dicts (a string is stored verbatim), so decrypt
|
|
|
|
|
only treats a json-OBJECT plaintext as a dict and returns everything else as the
|
|
|
|
|
raw string. this keeps the type faithful for the common cases: a string that is
|
|
|
|
|
json-shaped but NOT an object ('123'->'123', 'true'->'true', '[1,2]'->'[1,2]')
|
|
|
|
|
round-trips as a STRING, not an int/bool/list. the one irreducible ambiguity: a
|
|
|
|
|
string whose exact value is a json OBJECT ('{"a":1}') decrypts to a dict, because
|
|
|
|
|
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
|
|
|
|
|
blobs are unaffected — a dict was stored as a json object and still parses to a dict.
|
|
|
|
|
encrypt_data only json-encodes dicts (a string is stored verbatim), so decrypt
|
|
|
|
|
treats a json-OBJECT plaintext as a dict and everything else as a raw string —
|
|
|
|
|
a json-shaped but non-object string ('123', 'true', '[1,2]') round-trips as a
|
|
|
|
|
STRING, not int/bool/list. one irreducible ambiguity: a string whose exact
|
|
|
|
|
value is a json object ('{"a":1}') decrypts to a dict, indistinguishable
|
|
|
|
|
without a type marker from a stored dict — don't store a bare json-object
|
|
|
|
|
string if you need it back as a string. dict keys round-trip faithfully
|
|
|
|
|
because encrypt_data requires str keys (json's only key type).
|
|
|
|
|
"""
|
|
|
|
|
if not self.master_key:
|
|
|
|
|
raise ValueError("not initialized with data key")
|
|
|
|
@@ -394,26 +459,22 @@ class EnvelopeCrypto:
|
|
|
|
|
def reencrypt(self, source_crypto: "EnvelopeCrypto", record: dict, traversal_level: int = 2) -> dict:
|
|
|
|
|
"""re-encrypt a record's encrypted fields from source_crypto's key to this one's
|
|
|
|
|
|
|
|
|
|
self holds the destination (new) key; source_crypto holds the source (old)
|
|
|
|
|
key. only {secure, iv, data} fields are touched; plaintext fields are left
|
|
|
|
|
as-is. returns a new dict; the input is not mutated. used during rotation.
|
|
|
|
|
self holds the destination (new) key; source_crypto holds the source (old) key.
|
|
|
|
|
only {secure, iv, data} fields are touched; plaintext fields are left as-is.
|
|
|
|
|
returns a new dict; the input is not mutated. used during rotation. if `record`
|
|
|
|
|
itself is a bare blob (file-storage pattern) it is re-encrypted directly and
|
|
|
|
|
returned in place of `record`, not nested under a key.
|
|
|
|
|
|
|
|
|
|
if `record` itself is a {secure, iv, data} blob (the file-storage pattern, where
|
|
|
|
|
the blob IS the whole document) it is re-encrypted directly and the result is
|
|
|
|
|
returned in place of `record` — not nested under a key.
|
|
|
|
|
fails loud, unlike decrypt_record: a per-field decrypt failure RAISES (silently
|
|
|
|
|
keeping a field under the old key would lose it once that key is retired), and
|
|
|
|
|
so does a blob nested DEEPER than `traversal_level` — raise a higher
|
|
|
|
|
traversal_level or flatten the record before rotation instead.
|
|
|
|
|
|
|
|
|
|
unlike decrypt_record (which logs a failed field and leaves it encrypted), a
|
|
|
|
|
per-field decrypt failure here RAISES — rotation must fail loud, since silently
|
|
|
|
|
keeping a field under the old key would lose it once the old key is retired. for
|
|
|
|
|
the same reason, a blob nested DEEPER than `traversal_level` also RAISES instead
|
|
|
|
|
of being silently left under the old key: the caller either needs a higher
|
|
|
|
|
`traversal_level` or must flatten the record before rotation.
|
|
|
|
|
|
|
|
|
|
traversal recurses into nested DICTS only; a blob nested inside a LIST is NOT
|
|
|
|
|
re-encrypted and is NOT covered by the depth-limit raise above. records in this
|
|
|
|
|
scheme key blobs by field name, not inside arrays, so this doesn't arise in
|
|
|
|
|
practice — but if you store list-nested blobs, flatten them to dict fields before
|
|
|
|
|
rotation or they'll be silently left under the old key.
|
|
|
|
|
traversal recurses into nested DICTS only; a blob nested inside a LIST is not
|
|
|
|
|
re-encrypted and not covered by the depth-limit raise. this scheme keys blobs
|
|
|
|
|
by field name, not inside arrays, so this shouldn't arise in practice — but
|
|
|
|
|
flatten list-nested blobs to dict fields before rotation or they'll be
|
|
|
|
|
silently left under the old key.
|
|
|
|
|
"""
|
|
|
|
|
if not self.master_key:
|
|
|
|
|
raise ValueError("destination not initialized with data key")
|
|
|
|
@@ -447,12 +508,11 @@ def is_encrypted_record(record, traversal_level: int = 2) -> bool:
|
|
|
|
|
"""return whether a record has any encrypted ({secure, iv, data}) fields
|
|
|
|
|
|
|
|
|
|
checks `record` itself (the file-storage pattern stores the blob AS the whole
|
|
|
|
|
document, not nested under a key) as well as fields up to `traversal_level` deep.
|
|
|
|
|
beyond that bounded pass, this ALSO does an unbounded-depth scan before giving up —
|
|
|
|
|
so a blob left behind by a shallower decrypt_record/reencrypt call (nested deeper
|
|
|
|
|
than their traversal_level) is still reported as encrypted. this makes the function
|
|
|
|
|
safe to use as a leftover-detecting post-rotation audit: it never returns False for
|
|
|
|
|
a record that still contains a blob, at any depth.
|
|
|
|
|
document) as well as fields up to `traversal_level` deep. beyond that bounded
|
|
|
|
|
pass, falls back to an unbounded-depth scan, so a blob left behind by a
|
|
|
|
|
shallower decrypt_record/reencrypt call is still reported — safe to use as a
|
|
|
|
|
leftover-detecting post-rotation audit; never returns False for a record that
|
|
|
|
|
still contains a blob, at any depth.
|
|
|
|
|
|
|
|
|
|
aliases: is_encrypted_document, is_encrypted_dict — same function
|
|
|
|
|
"""
|
|
|
|
@@ -478,14 +538,10 @@ def is_encrypted_record(record, traversal_level: int = 2) -> bool:
|
|
|
|
|
def decrypt_record(crypto: EnvelopeCrypto, record, traversal_level: int = 2) -> Union[dict, Any]:
|
|
|
|
|
"""decrypt a record's encrypted fields into a new dict (up to traversal_level deep)
|
|
|
|
|
|
|
|
|
|
if `record` itself is a {secure, iv, data} blob (the file-storage pattern, where the
|
|
|
|
|
blob IS the whole document) it is decrypted directly and the decrypted value
|
|
|
|
|
(dict or string — see decrypt_data) is returned in place of `record`.
|
|
|
|
|
|
|
|
|
|
failures on a single field are logged and that field is left encrypted, so a
|
|
|
|
|
partial failure is visible (the {secure,...} blob remains) rather than silent. a
|
|
|
|
|
failure decrypting `record` itself (the self-blob case above) is likewise logged
|
|
|
|
|
and the still-encrypted blob is returned unchanged.
|
|
|
|
|
if `record` itself is a bare {secure, iv, data} blob (file-storage pattern) it is
|
|
|
|
|
decrypted directly and the value (dict or string — see decrypt_data) is returned
|
|
|
|
|
in place of `record`. a failure on a single field (or on `record` itself) is
|
|
|
|
|
logged and left encrypted, so a partial failure stays visible rather than silent.
|
|
|
|
|
|
|
|
|
|
aliases: decrypt_document, decrypt_dict — same function
|
|
|
|
|
"""
|
|
|
|
@@ -513,9 +569,49 @@ def decrypt_record(crypto: EnvelopeCrypto, record, traversal_level: int = 2) ->
|
|
|
|
|
return result
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _fingerprint_default(value: Any) -> str:
|
|
|
|
|
"""json.dumps default= handler for values fingerprint_data can't natively serialize
|
|
|
|
|
|
|
|
|
|
covers datetime/date/time (isoformat), bytes/bytearray (hex), and anything else
|
|
|
|
|
(e.g. ObjectId) via a type-tagged repr — never raises, never logs the value.
|
|
|
|
|
"""
|
|
|
|
|
if hasattr(value, "isoformat"):
|
|
|
|
|
return f"isoformat:{value.isoformat()}"
|
|
|
|
|
if isinstance(value, (bytes, bytearray)):
|
|
|
|
|
return f"hex:{value.hex()}"
|
|
|
|
|
return f"{type(value).__name__}:{value!r}"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def _fingerprint_normalize(value: Any) -> Any:
|
|
|
|
|
"""recursively tag dict keys with their type to avoid cross-type key collisions
|
|
|
|
|
|
|
|
|
|
json object keys are always strings, so {1: "a"} and {"1": "a"} would otherwise
|
|
|
|
|
serialize identically and collide to the same fingerprint; prefixing each key with
|
|
|
|
|
its type name keeps them distinct. non-dict containers/values pass through
|
|
|
|
|
untouched (lists recurse; leaf values are handled by _fingerprint_default).
|
|
|
|
|
"""
|
|
|
|
|
if isinstance(value, dict):
|
|
|
|
|
return {
|
|
|
|
|
f"{type(key).__name__}:{key!r}": _fingerprint_normalize(sub_value)
|
|
|
|
|
for key, sub_value in value.items()
|
|
|
|
|
}
|
|
|
|
|
if isinstance(value, (list, tuple)):
|
|
|
|
|
return [_fingerprint_normalize(item) for item in value]
|
|
|
|
|
return value
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def fingerprint_data(data: dict) -> str:
|
|
|
|
|
"""return a deterministic SHA-256 hex fingerprint of a dict"""
|
|
|
|
|
return hashlib.sha256(json.dumps(data, sort_keys=True).encode()).hexdigest()
|
|
|
|
|
"""return a deterministic, collision-free SHA-256 hex fingerprint of a dict
|
|
|
|
|
|
|
|
|
|
dict keys of different types that would otherwise coerce to the same JSON string
|
|
|
|
|
(e.g. {1: "a"} vs {"1": "a"}) are kept distinct via a type-tagged pre-pass. values
|
|
|
|
|
that aren't JSON-native (datetime/date/time, bytes/bytearray, ObjectId-like
|
|
|
|
|
objects) are serialized via a stable default= handler instead of raising. never
|
|
|
|
|
logs the data being fingerprinted.
|
|
|
|
|
"""
|
|
|
|
|
normalized = _fingerprint_normalize(data)
|
|
|
|
|
encoded = json.dumps(normalized, sort_keys=True, default=_fingerprint_default)
|
|
|
|
|
return hashlib.sha256(encoded.encode()).hexdigest()
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# function aliases — same functions, naming preference only
|
|
|
|
|