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:<purpose>), 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 <dev@disqualifier.me>
This commit is contained in:
dsql committed 2026-10-16 02:40:52 -04:00
1 parent e24da95fc5
commit 8cda08e7cf
3 files changed
+115 -8

No files matched your search

+41 -1
View File
@@ -76,7 +76,7 @@ round-trips two in-memory PEM strings rather than treating them as paths.
## Encrypt / decrypt ## Encrypt / decrypt
```python ```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": "..."} 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` = = `PCICrypto` (deprecated legacy alias). `decrypt_record` = `decrypt_document` =
`decrypt_dict`; `is_encrypted_record` = `is_encrypted_document` = `is_encrypted_dict`. `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 ## Authorize another system
An initialized system wraps the DEK for another system's public key. Returns the An initialized system wraps the DEK for another system's public key. Returns the
+2
View File
@@ -5,6 +5,7 @@ from .envelope_crypto import (
DocumentCrypto, DocumentCrypto,
RecordCrypto, RecordCrypto,
PCICrypto, PCICrypto,
KeyMismatchError,
is_encrypted_record, is_encrypted_record,
is_encrypted_document, is_encrypted_document,
is_encrypted_dict, is_encrypted_dict,
@@ -24,6 +25,7 @@ __all__ = [
"DocumentCrypto", "DocumentCrypto",
"RecordCrypto", "RecordCrypto",
"PCICrypto", "PCICrypto",
"KeyMismatchError",
"is_encrypted_record", "is_encrypted_record",
"is_encrypted_document", "is_encrypted_document",
"is_encrypted_dict", "is_encrypted_dict",
+72 -7
View File
@@ -14,6 +14,12 @@ storage-agnostic; see README for bootstrap/boot/authorize/rotate flows.
crypto.initialize(master_key) # exactly 32 bytes (AES-256) crypto.initialize(master_key) # exactly 32 bytes (AES-256)
enc = crypto.encrypt_data({"ssn": "..."}) # -> {secure, iv, data} enc = crypto.encrypt_data({"ssn": "..."}) # -> {secure, iv, data}
plain = crypto.decrypt_data(enc) # -> original 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 naming: EnvelopeCrypto is canonical. PCICrypto / DocumentCrypto / RecordCrypto are
aliases (PCICrypto is a deprecated legacy alias). the document/record/dict function aliases (PCICrypto is a deprecated legacy alias). the document/record/dict function
@@ -25,6 +31,7 @@ import copy
import json import json
import base64 import base64
import hashlib import hashlib
import hmac
import logging import logging
from typing import Any, Dict, List, Optional, Tuple, Union 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 import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding, rsa from cryptography.hazmat.primitives.asymmetric import padding, rsa
from cryptography.hazmat.primitives.ciphers.aead import AESGCM 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 from cryptography.hazmat.primitives.serialization import load_ssh_public_key
_log = logging.getLogger(__name__) _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: def _password_mismatch_message(pw: Optional[bytes]) -> str:
"""clear ValueError text for a TypeError raised by a password/encryption mismatch""" """clear ValueError text for a TypeError raised by a password/encryption mismatch"""
if pw is not None: if pw is not None:
@@ -431,6 +463,7 @@ class EnvelopeCrypto:
"secure": True, "secure": True,
"iv": base64.b64encode(iv).decode(), "iv": base64.b64encode(iv).decode(),
"data": base64.b64encode(ciphertext).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]: def decrypt_data(self, encrypted_data: Dict[str, str]) -> Union[Dict[str, Any], str]:
@@ -451,11 +484,16 @@ class EnvelopeCrypto:
Raises: Raises:
ValueError: instance not initialized, or encrypted_data is not a ValueError: instance not initialized, or encrypted_data is not a
{secure, iv, data} blob. {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: if not self.master_key:
raise ValueError("not initialized with data 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: 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") 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"]) iv = base64.b64decode(encrypted_data["iv"])
ciphertext = base64.b64decode(encrypted_data["data"]) ciphertext = base64.b64decode(encrypted_data["data"])
@@ -467,6 +505,39 @@ class EnvelopeCrypto:
return plaintext return plaintext
return parsed if isinstance(parsed, dict) else 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: 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 """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 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. memory address, which would make the digest vary across processes. never logs the data.
""" """
def _stringify(value: object) -> str: return hashlib.sha256(_canonical_json(data).encode()).hexdigest()
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()
# function aliases - same functions, naming preference only # function aliases - same functions, naming preference only