From 8cda08e7cf197537d5605fa3dd19c72e1769cff4 Mon Sep 17 00:00:00 2001 From: disqualifier Date: Fri, 16 Oct 2026 02:40:52 -0400 Subject: [PATCH] feat: keyed_fingerprint and key ids keyed_fingerprint(data, purpose): HMAC-SHA256 under an HKDF-SHA256 subkey of the DEK (info = envelope_crypto:keyed_fingerprint:), for dedup on secret values. dict input shares fingerprint_data's canonical json helper. encrypt_data stamps a kid (first 16 hex of sha256(DEK)); decrypt_data raises KeyMismatchError (ValueError subclass) naming both kids on a mismatch. blobs without kid decrypt as before; decrypt_record keeps log-and-leave. Signed-off-by: disqualifier --- README.md | 42 +++++++++++++- src/envelope_crypto/__init__.py | 2 + src/envelope_crypto/envelope_crypto.py | 79 +++++++++++++++++++++++--- 3 files changed, 115 insertions(+), 8 deletions(-) diff --git a/README.md b/README.md index a16ed6c..ca4fa20 100644 --- a/README.md +++ b/README.md @@ -76,7 +76,7 @@ round-trips two in-memory PEM strings rather than treating them as paths. ## Encrypt / decrypt ```python -enc = crypto.encrypt_data({"ssn": "..."}) # -> {"secure": True, "iv": ..., "data": ...} +enc = crypto.encrypt_data({"ssn": "..."}) # -> {"secure": True, "iv": ..., "data": ..., "kid": ...} plain = crypto.decrypt_data(enc) # -> {"ssn": "..."} ``` @@ -110,6 +110,46 @@ Naming aliases (same objects): `EnvelopeCrypto` = `DocumentCrypto` = `RecordCryp = `PCICrypto` (deprecated legacy alias). `decrypt_record` = `decrypt_document` = `decrypt_dict`; `is_encrypted_record` = `is_encrypted_document` = `is_encrypted_dict`. +## Key ids + +Every blob `encrypt_data` writes carries a `"kid"`: the first 16 hex chars of +`sha256(DEK)`. It identifies which data key sealed the blob without revealing it. + +- `decrypt_data` on a blob whose `kid` doesn't match the loaded key raises + `KeyMismatchError` (a `ValueError` subclass, exported) naming both kids, e.g. + `blob was sealed under key id 3f9a..., but the loaded key is 81c0...` — instead of + the opaque `InvalidTag` you'd get from AES-GCM. Catch `KeyMismatchError` to tell + "sealed under another key" apart from tampering (`InvalidTag`). +- `decrypt_record` keeps its contract: a mismatched field is logged (the log line + names both kids) and left encrypted; it does not raise. +- `reencrypt` writes the destination key's `kid` automatically (it goes through + `encrypt_data`). +- Blobs without `kid` (written before v1.1.0) decrypt exactly as before. v1.0.0 + readers ignore the extra field, so mixed versions reading the same store are safe. + +## Keyed fingerprints + +`fingerprint_data` is a plain SHA-256: fine for change detection on ordinary records, +but for a low-entropy secret (card number, email, phone) anyone with the digest can +brute-force the value back. Use `keyed_fingerprint` to dedup or look up records by a +secret value without storing it: + +```python +fp = crypto.keyed_fingerprint(card_number, "card") # HMAC-SHA256 hex +await db.create_document("cards", {"card_fp": fp, "card": crypto.encrypt_data(card_number)}) +``` + +The digest is an HMAC keyed by an HKDF subkey of the DEK, so it is only computable by a +system that holds the DEK. `purpose` is required and domain-separates digests: the same +email under `"email"` and `"login"` gives unrelated values, so a digest from one table +can't be matched against another. `data` may be `str` (UTF-8), `bytes`, or `dict` (the +same canonical JSON as `fingerprint_data`); `"abc"` and `b"abc"` give the same digest, +as do a dict and its canonical JSON string. + +**A DEK rotation changes every keyed fingerprint.** After `rotate_master_key`, recompute +stored keyed fingerprints with the new instance (decrypt the field, re-fingerprint) +alongside `reencrypt`, or lookups against old digests will miss. + ## Authorize another system An initialized system wraps the DEK for another system's public key. Returns the diff --git a/src/envelope_crypto/__init__.py b/src/envelope_crypto/__init__.py index 3e99a1d..3597559 100644 --- a/src/envelope_crypto/__init__.py +++ b/src/envelope_crypto/__init__.py @@ -5,6 +5,7 @@ from .envelope_crypto import ( DocumentCrypto, RecordCrypto, PCICrypto, + KeyMismatchError, is_encrypted_record, is_encrypted_document, is_encrypted_dict, @@ -24,6 +25,7 @@ __all__ = [ "DocumentCrypto", "RecordCrypto", "PCICrypto", + "KeyMismatchError", "is_encrypted_record", "is_encrypted_document", "is_encrypted_dict", diff --git a/src/envelope_crypto/envelope_crypto.py b/src/envelope_crypto/envelope_crypto.py index 1a3f16c..fce9ffb 100644 --- a/src/envelope_crypto/envelope_crypto.py +++ b/src/envelope_crypto/envelope_crypto.py @@ -14,6 +14,12 @@ storage-agnostic; see README for bootstrap/boot/authorize/rotate flows. crypto.initialize(master_key) # exactly 32 bytes (AES-256) enc = crypto.encrypt_data({"ssn": "..."}) # -> {secure, iv, data} plain = crypto.decrypt_data(enc) # -> original + fp = crypto.keyed_fingerprint("a@x.com", "email") # -> HMAC hex, dedup on secrets + +key ids: new blobs carry "kid" (first 16 hex of sha256(DEK)). decrypt_data raises +KeyMismatchError (a ValueError) naming both kids when a blob's kid doesn't match the +loaded key; blobs without "kid" (pre-1.1.0) decrypt as before. decrypt_record keeps its +log-and-leave-encrypted contract on a mismatch. naming: EnvelopeCrypto is canonical. PCICrypto / DocumentCrypto / RecordCrypto are aliases (PCICrypto is a deprecated legacy alias). the document/record/dict function @@ -25,6 +31,7 @@ import copy import json import base64 import hashlib +import hmac import logging from typing import Any, Dict, List, Optional, Tuple, Union @@ -32,6 +39,7 @@ from cryptography.exceptions import UnsupportedAlgorithm from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import padding, rsa from cryptography.hazmat.primitives.ciphers.aead import AESGCM +from cryptography.hazmat.primitives.kdf.hkdf import HKDF from cryptography.hazmat.primitives.serialization import load_ssh_public_key _log = logging.getLogger(__name__) @@ -43,6 +51,30 @@ _OAEP_PADDING = padding.OAEP( ) +class KeyMismatchError(ValueError): + """a blob's kid doesn't match the loaded data key: it was sealed under another key""" + + def __init__(self, blob_kid: str, key_kid: str): + self.blob_kid = blob_kid + self.key_kid = key_kid + super().__init__(f"blob was sealed under key id {blob_kid}, but the loaded key is {key_kid}") + + +def _kid(key: bytes) -> str: + """public key id of a data key: first 16 hex chars of sha256(key)""" + return hashlib.sha256(key).hexdigest()[:16] + + +def _canonical_json(data: Any) -> str: + """sorted-key, fixed-separator json; non-json scalars stringified, callables rejected""" + def _stringify(value: object) -> str: + if callable(value): + raise TypeError(f"cannot fingerprint a callable: {value!r}") + return str(value) + + return json.dumps(data, sort_keys=True, separators=(",", ":"), default=_stringify) + + def _password_mismatch_message(pw: Optional[bytes]) -> str: """clear ValueError text for a TypeError raised by a password/encryption mismatch""" if pw is not None: @@ -431,6 +463,7 @@ class EnvelopeCrypto: "secure": True, "iv": base64.b64encode(iv).decode(), "data": base64.b64encode(ciphertext).decode(), + "kid": _kid(self.master_key), } def decrypt_data(self, encrypted_data: Dict[str, str]) -> Union[Dict[str, Any], str]: @@ -451,11 +484,16 @@ class EnvelopeCrypto: Raises: ValueError: instance not initialized, or encrypted_data is not a {secure, iv, data} blob. + KeyMismatchError: the blob's "kid" doesn't match the loaded key + (a ValueError subclass; blobs without "kid" skip this check). """ if not self.master_key: raise ValueError("not initialized with data key") if not isinstance(encrypted_data, dict) or "iv" not in encrypted_data or "data" not in encrypted_data: raise ValueError("decrypt_data expects a {secure, iv, data} blob") + blob_kid = encrypted_data.get("kid") + if blob_kid is not None and blob_kid != _kid(self.master_key): + raise KeyMismatchError(blob_kid, _kid(self.master_key)) iv = base64.b64decode(encrypted_data["iv"]) ciphertext = base64.b64decode(encrypted_data["data"]) @@ -467,6 +505,39 @@ class EnvelopeCrypto: return plaintext return parsed if isinstance(parsed, dict) else plaintext + def keyed_fingerprint(self, data: Union[str, bytes, Dict[str, Any]], purpose: str) -> str: + """HMAC-SHA256 hex digest of data, keyed by an HKDF subkey of the DEK per purpose + + for dedup/lookup on secret values (card numbers, emails) where a plain hash + could be brute-forced. purpose domain-separates digests: the same value under + two purposes gives unrelated digests. a DEK rotation changes every keyed + fingerprint. str is utf-8 encoded, bytes used as-is, a dict uses the same + canonical json as fingerprint_data - so "abc" and b"abc" give the same digest, + as do a dict and its canonical json string. never logs the data or the key. + + Raises: + ValueError: instance not initialized, or purpose is empty / not a str. + TypeError: data is not str, bytes or dict. + """ + if not self.master_key: + raise ValueError("not initialized with data key") + if not isinstance(purpose, str) or not purpose: + raise ValueError("keyed_fingerprint requires a non-empty str purpose") + if isinstance(data, str): + raw = data.encode() + elif isinstance(data, bytes): + raw = data + elif isinstance(data, dict): + raw = _canonical_json(data).encode() + else: + raise TypeError(f"keyed_fingerprint expects str, bytes or dict, got {type(data).__name__}") + + subkey = HKDF( + algorithm=hashes.SHA256(), length=32, salt=None, + info=b"envelope_crypto:keyed_fingerprint:" + purpose.encode(), + ).derive(self.master_key) + return hmac.new(subkey, raw, hashlib.sha256).hexdigest() + 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 @@ -639,13 +710,7 @@ def fingerprint_data(data: dict) -> str: stringified rather than crashing the encode. a callable value raises - its str() embeds a memory address, which would make the digest vary across processes. never logs the data. """ - def _stringify(value: object) -> str: - if callable(value): - raise TypeError(f"fingerprint_data: cannot fingerprint a callable: {value!r}") - return str(value) - - encoded = json.dumps(data, sort_keys=True, separators=(",", ":"), default=_stringify) - return hashlib.sha256(encoded.encode()).hexdigest() + return hashlib.sha256(_canonical_json(data).encode()).hexdigest() # function aliases - same functions, naming preference only