release: 1.1.0

Signed-off-by: disqualifier <dev@disqualifier.me>

README.md

@@ -11,18 +11,18 @@ and storage-agnostic.
 `requirements.txt`:
 
 ```
-envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.6
+envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v1.1.0
 ```
 
 Direct:
 
 ```bash
-pip install "envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.6"
+pip install "envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v1.1.0"
 ```
 
 Requires `cryptography` (pulled transitively).
 
-Drop the `@v0.1.6` suffix from the line above to install the latest unpinned.
+Drop the `@v1.1.0` suffix from the line above to install the latest unpinned.
 
 ## First-time setup
 
@@ -76,14 +76,16 @@ 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": "..."}
 ```
 
-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.
+Dict keys must be `str`, at any nesting depth (including a dict nested inside a list
+or tuple). `encrypt_data` raises `TypeError` on a non-str key anywhere in the payload
+(e.g. an int-keyed dict of Discord snowflakes, even nested a few levels down) 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}`
 field (nested up to `traversal_level`, default 2); `is_encrypted_record(doc)` reports
@@ -98,15 +100,56 @@ if is_encrypted_record(doc):
     doc = decrypt_record(crypto, doc)
 ```
 
-`is_encrypted_record` falls back to an unbounded-depth scan once `traversal_level` is
-exhausted, so it reliably reports `True` for a blob left behind by a shallower
+`is_encrypted_record` always falls back to an unbounded-depth scan whenever its bounded
+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
-rotation, regardless of how deep the blob is nested.
+rotation, regardless of how deep the blob is nested (including inside a list or tuple
+at any depth) and regardless of which `traversal_level` you pass; it never false-negatives.
 
 Naming aliases (same objects): `EnvelopeCrypto` = `DocumentCrypto` = `RecordCrypto`
 = `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

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "envelope_crypto"
-version = "0.1.6"
+version = "1.1.0"
 description = "Envelope encryption (RSA-OAEP wrapped AES-256-GCM) for dict records — config-free, storage-agnostic, installable."
 requires-python = ">=3.10"
 dependencies = [

src/envelope_crypto/__init__.py

@@ -1,8 +1,11 @@
+from importlib.metadata import version, PackageNotFoundError
+
 from .envelope_crypto import (
     EnvelopeCrypto,
     DocumentCrypto,
     RecordCrypto,
     PCICrypto,
+    KeyMismatchError,
     is_encrypted_record,
     is_encrypted_document,
     is_encrypted_dict,
@@ -12,11 +15,17 @@ from .envelope_crypto import (
     fingerprint_data,
 )
 
+try:
+    __version__ = version("envelope_crypto")
+except PackageNotFoundError:
+    __version__ = "0.0.0+unknown"
+
 __all__ = [
     "EnvelopeCrypto",
     "DocumentCrypto",
     "RecordCrypto",
     "PCICrypto",
+    "KeyMismatchError",
     "is_encrypted_record",
     "is_encrypted_document",
     "is_encrypted_dict",
@@ -24,4 +33,5 @@ __all__ = [
     "decrypt_document",
     "decrypt_dict",
     "fingerprint_data",
+    "__version__",
 ]

src/envelope_crypto/envelope_crypto.py

@@ -2,12 +2,11 @@
 envelope encryption for dict records
 
 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.
+(RSA-OAEP) per authorized system's public key (KEK) for distribution and stored by
+the caller, keyed by fingerprint. 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. config-free and
+storage-agnostic; see README for bootstrap/boot/authorize/rotate flows.
 
     from envelope_crypto import EnvelopeCrypto
 
@@ -15,55 +14,16 @@ and counts.
     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
 
-first-time setup: generate the DEK and wrap it for the first system in one call,
-then verify the pipeline before storing anything:
-
-    crypto, fingerprint, wrapped = EnvelopeCrypto.bootstrap("public_key.pem")
-    crypto.self_test("public_key.pem", "private_key.pem")    # raises if anything is wrong
-    caller_store({"_id": fingerprint, "key": wrapped}) # the only record of the DEK
-
-boot (already set up): fingerprint own pubkey, fetch the wrapped DEK, unwrap:
-
-    fp = crypto.get_rsa_key_fingerprint("public_key.pem")
-    record = caller_lookup(fp)
-    crypto.initialize(crypto.decrypt_aes_key_with_rsa(record["key"], "private_key.pem"))
-
-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. 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:
-
-    new_key, wrapped = crypto.rotate_master_key([pub_a, pub_b])
-    new_crypto = EnvelopeCrypto(); new_crypto.initialize(new_key)
-    for record in caller_iter():
-        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 (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.
+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
-variants are the same functions — use whichever fits your storage.
+variants are the same functions - use whichever fits your storage.
 """
 
 import os
@@ -71,6 +31,7 @@ import copy
 import json
 import base64
 import hashlib
+import hmac
 import logging
 from typing import Any, Dict, List, Optional, Tuple, Union
 
@@ -78,30 +39,51 @@ 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__)
 
+_OAEP_PADDING = padding.OAEP(
+    mgf=padding.MGF1(algorithm=hashes.SHA256()),
+    algorithm=hashes.SHA256(),
+    label=None,
+)
+
+
+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
-
-    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".
-    """
+    """clear ValueError text for a TypeError raised by a password/encryption mismatch"""
     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.
-    """
+    """load a PEM or OpenSSH private key, normalizing the password/encryption mismatch error"""
     try:
         return serialization.load_pem_private_key(key_data, password=pw)
     except ValueError as error:
@@ -111,8 +93,6 @@ def _load_private_key(key_data: bytes, pw: Optional[bytes]):
             except TypeError as 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:
@@ -120,11 +100,7 @@ def _load_private_key(key_data: bytes, pw: Optional[bytes]):
 
 
 def _fingerprint_of(public_key) -> str:
-    """base64 SHA-256 fingerprint of an already-loaded public key
-
-    factored so callers that already hold a loaded key (e.g. encrypt_aes_key_with_rsa)
-    don't re-open and re-parse the key file just to fingerprint it.
-    """
+    """base64 SHA-256 fingerprint of an already-loaded public key"""
     key_bytes = public_key.public_bytes(
         encoding=serialization.Encoding.DER,
         format=serialization.PublicFormat.SubjectPublicKeyInfo,
@@ -140,37 +116,38 @@ def _is_blob(value: Any) -> bool:
 
 
 def _has_encrypted_field(record: Any) -> bool:
-    """unbounded-depth scan: does record (or anything nested under it) contain a blob
+    """unbounded-depth scan: does record (or anything nested under it, incl. list/tuple items) contain a blob"""
+    if isinstance(record, dict):
+        if _is_blob(record):
+            return True
+        return any(_has_encrypted_field(value) for value in record.values())
+    if isinstance(record, (list, tuple)):
+        return any(_has_encrypted_field(item) for item in record)
+    return False
 
-    used to detect a blob left behind by a depth-limited traversal — no traversal_level
-    cutoff here, since the whole point is to catch what a bounded pass would miss.
-    """
-    if not isinstance(record, dict):
-        return False
-    if _is_blob(record):
-        return True
-    return any(_has_encrypted_field(value) for value in record.values())
+
+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:
-    """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.
-    """
+    """raise ValueError unless key is an RSA public or private key"""
     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
-
-    load_ssh_public_key raises UnsupportedAlgorithm (not ValueError) on non-SSH/garbage
-    input; normalize it so a bad public key always surfaces as a clear ValueError,
-    consistent with the private-key path.
-    """
+    """load a PEM or OpenSSH public key, normalizing non-key input to ValueError"""
     try:
         return serialization.load_pem_public_key(key_data)
     except ValueError:
@@ -195,10 +172,20 @@ class EnvelopeCrypto:
     def bootstrap(cls, rsa_public_key: str, is_file: bool = True) -> Tuple["EnvelopeCrypto", str, str]:
         """first-time setup: generate a DEK and wrap it for the first system
 
-        returns (crypto, fingerprint, wrapped_key) — an initialized instance plus
-        the record to store as the first authorization. the plaintext DEK is never
-        returned or persisted; it survives only as the wrapped copy. run self_test
-        before storing to confirm the keypair round-trips.
+        the plaintext DEK is never returned or persisted; it survives only as the
+        wrapped copy. run self_test before storing to confirm the keypair round-trips.
+
+        Args:
+            rsa_public_key: path to (or, if is_file=False, raw PEM/OpenSSH data of)
+                the first system's RSA public key.
+            is_file: True (default) treats rsa_public_key as a path.
+
+        Returns:
+            (crypto, fingerprint, wrapped_key): an initialized instance plus the
+            record to store as the first authorization.
+
+        Raises:
+            ValueError: rsa_public_key is not an RSA key, or is malformed.
         """
         crypto = cls()
         crypto.initialize(crypto.create_aes_key())
@@ -212,10 +199,23 @@ class EnvelopeCrypto:
         """verify the full pipeline against a keypair; raises on any mismatch
 
         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.
+        the DEK with the given keypair. run after bootstrap or as a health check.
+
+        Args:
+            rsa_public_key: public half of the keypair to verify (path or, if
+                is_file=False, raw PEM/OpenSSH data).
+            rsa_private_key: private half of the keypair to verify.
+            is_file: True (default) treats both keys as paths; False treats both
+                as in-memory PEM/OpenSSH data.
+            password: password for an encrypted private key, if any.
+
+        Returns:
+            True on success.
+
+        Raises:
+            ValueError: instance not initialized with a data key.
+            RuntimeError: data round-trip fails, or the keypair does not pair
+                (unwrap fails or the recovered key mismatches).
         """
         if not self.master_key:
             raise ValueError("self_test: not initialized with a key")
@@ -243,7 +243,7 @@ class EnvelopeCrypto:
 
         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.
+        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:
@@ -273,11 +273,23 @@ class EnvelopeCrypto:
     ) -> str:
         """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. 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).
+        always fingerprints the public half, so a private key and its public key
+        match. PEM and OpenSSH accepted (mirrors decrypt_aes_key_with_rsa).
+
+        Args:
+            key_path_or_data: path to (or, if is_file=False, raw PEM/OpenSSH data
+                of) the key.
+            is_private: fingerprint the public half of a private key; pass its
+                `password` if encrypted (an unencrypted key ignores it).
+            is_file: True (default) treats key_path_or_data as a path.
+            password: password for an encrypted private key, if is_private.
+
+        Returns:
+            base64 SHA-256 fingerprint.
+
+        Raises:
+            ValueError: key is malformed, or password/encryption mismatch
+                (cryptography's raw TypeError normalized here).
         """
         if is_file:
             with open(key_path_or_data, "rb") as key_file:
@@ -303,10 +315,22 @@ 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
 
-        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).
+        Args:
+            aes_key: the AES data key to wrap.
+            rsa_key: path to (or, if is_file=False, raw PEM/OpenSSH data of)
+                the RSA public key to wrap with.
+            is_file: True (default) treats rsa_key as a path.
+
+        Returns:
+            (fingerprint, wrapped_b64): the wrapping key's fingerprint and the
+            RSA-OAEP wrapped key, base64-encoded.
+
+        Raises:
+            ValueError: 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), or is malformed.
         """
         if is_file:
             with open(rsa_key, "rb") as key_file:
@@ -317,15 +341,7 @@ class EnvelopeCrypto:
         public_key = _load_public_key(key_data)
         _require_rsa(public_key)
 
-        wrapped = public_key.encrypt(
-            aes_key,
-            padding.OAEP(
-                mgf=padding.MGF1(algorithm=hashes.SHA256()),
-                algorithm=hashes.SHA256(),
-                label=None,
-            ),
-        )
-        # fingerprint from the already-loaded public_key — no second open/parse of the file
+        wrapped = public_key.encrypt(aes_key, _OAEP_PADDING)
         fingerprint = _fingerprint_of(public_key)
         wrapped_b64 = base64.b64encode(wrapped).decode()
         _log.info("wrapped data key for fingerprint %s", fingerprint[:8])
@@ -337,10 +353,20 @@ class EnvelopeCrypto:
     ) -> bytes:
         """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.
+        Args:
+            encrypted_key_base64: the RSA-OAEP wrapped key, base64-encoded.
+            rsa_private_key: path to (or, if is_file=False, raw PEM/OpenSSH data
+                of) the private key to unwrap with.
+            is_file: True (default) treats rsa_private_key as a path; pass False
+                to supply PEM/OpenSSH key data directly (e.g. vault-sourced).
+            password: password for an encrypted private key, if any.
+
+        Returns:
+            the unwrapped AES data key.
+
+        Raises:
+            ValueError: rsa_private_key is not an RSA key, is malformed, or a
+                password/encryption mismatch.
         """
         if is_file:
             with open(rsa_private_key, "rb") as key_file:
@@ -352,23 +378,28 @@ class EnvelopeCrypto:
         _require_rsa(private_key)
 
         wrapped = base64.b64decode(encrypted_key_base64)
-        aes_key = private_key.decrypt(
-            wrapped,
-            padding.OAEP(
-                mgf=padding.MGF1(algorithm=hashes.SHA256()),
-                algorithm=hashes.SHA256(),
-                label=None,
-            ),
-        )
+        aes_key = private_key.decrypt(wrapped, _OAEP_PADDING)
         _log.info("unwrapped data key with RSA private key")
         return aes_key
 
     def authorize_system(self, rsa_public_key: str, is_file: bool = True) -> Tuple[str, str]:
         """wrap the current data key for another system's public key
 
-        returns (fingerprint, wrapped_b64) for the caller to store as that
-        system's key-authorization record. requires this instance to already
-        hold the data key — only an authorized system can authorize others.
+        requires this instance to already hold the data key - only an
+        authorized system can authorize others.
+
+        Args:
+            rsa_public_key: path to (or, if is_file=False, raw PEM/OpenSSH data
+                of) the other system's RSA public key.
+            is_file: True (default) treats rsa_public_key as a path.
+
+        Returns:
+            (fingerprint, wrapped_b64) for the caller to store as that system's
+            key-authorization record.
+
+        Raises:
+            ValueError: this instance is not initialized, or rsa_public_key is
+                not an RSA key.
         """
         if not self.master_key:
             raise ValueError("cannot authorize another system: not initialized")
@@ -379,9 +410,20 @@ class EnvelopeCrypto:
     ) -> Tuple[bytes, Dict[str, str]]:
         """generate a NEW data key and wrap it for each authorized public key
 
-        returns (new_key, {fingerprint: wrapped_b64}). does NOT re-encrypt existing
-        data — build a new instance with the new key and call reencrypt() on each
-        record. systems not in the list get no wrapped copy (deauthorized).
+        does NOT re-encrypt existing data - build a new instance with the new key
+        and call reencrypt() on each record. systems not in the list get no
+        wrapped copy (deauthorized).
+
+        Args:
+            authorized_public_keys: RSA public keys (paths, or raw data if
+                is_file=False) to wrap the new key for.
+            is_file: True (default) treats each entry as a path.
+
+        Returns:
+            (new_key, {fingerprint: wrapped_b64}).
+
+        Raises:
+            ValueError: any entry in authorized_public_keys is not an RSA key.
         """
         new_key = self.create_aes_key()
         wrapped = {}
@@ -394,27 +436,23 @@ class EnvelopeCrypto:
     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
 
-        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
+        dict keys must be str, at any nesting depth (including dicts nested inside
+        lists/tuples): 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
+            non_str_key_types = _non_str_key_types(data)
+            if non_str_key_types:
                 raise TypeError(
-                    "encrypt_data requires str dict keys, got non-str key(s): "
-                    f"{[type(key).__name__ for key in non_str_keys]}"
+                    "encrypt_data requires str dict keys (at any nesting depth), got non-str "
+                    f"key(s): {non_str_key_types}"
                 )
 
         data_str = json.dumps(data) if isinstance(data, dict) else data
@@ -425,26 +463,37 @@ 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]:
         """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
-        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).
+        a json-shaped but non-object plaintext ('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
+        from a stored dict - don't store a bare json-object string if you need it
+        back as a string.
+
+        Args:
+            encrypted_data: a {secure, iv, data} blob from encrypt_data.
+
+        Returns:
+            the original dict or string.
+
+        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:
-            # a structurally-malformed blob would raise a raw KeyError/TypeError; surface
-            # a clear ValueError instead, matching the documented {secure, iv, data} shape
             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"])
@@ -456,25 +505,68 @@ 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
 
-        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.
+        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. used during rotation. if `record` itself is a bare blob (file-storage
+        pattern) it is re-encrypted directly and returned in place of `record`.
 
-        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.
+        traversal recurses into nested DICTS only - a blob nested inside a LIST is
+        not re-encrypted. the depth-limit raise below IS reached for a list-nested
+        blob when the cutoff dict scan finds it (it walks lists too), but a blob
+        one level shallower - hit during normal traversal instead of the cutoff
+        scan - is skipped silently; flatten list-nested blobs to dict fields before
+        rotation or they can 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.
+        Args:
+            source_crypto: instance holding the old (source) key.
+            record: the record to re-encrypt. not mutated; a new dict is returned.
+            traversal_level: max nesting depth to recurse into (default 2).
+
+        Returns:
+            a new dict with encrypted fields re-wrapped under this instance's key.
+
+        Raises:
+            ValueError: destination not initialized with a data key; a per-field
+                decrypt failure (unlike decrypt_record, this fails loud rather
+                than silently stranding a field under the old key); or a blob
+                nested deeper than `traversal_level` - raise traversal_level or
+                flatten the record before rotation instead.
         """
         if not self.master_key:
             raise ValueError("destination not initialized with data key")
@@ -493,28 +585,56 @@ class EnvelopeCrypto:
                     raise ValueError(
                         f"reencrypt: field {key!r} contains an encrypted blob nested deeper "
                         "than traversal_level; increase traversal_level or flatten the record "
-                        "before rotation — leaving it would strand the field under the old key"
+                        "before rotation - leaving it would strand the field under the old key"
                     )
         return result
 
 
-# naming aliases — same class
+# naming aliases - same class
 DocumentCrypto = EnvelopeCrypto
 RecordCrypto = EnvelopeCrypto
 PCICrypto = EnvelopeCrypto  # deprecated legacy alias; remove after all systems migrate
 
 
+def _is_encrypted_value(value: Any, traversal_level: int) -> bool:
+    """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):
+        return True
+    if isinstance(value, dict):
+        if traversal_level > 0:
+            return is_encrypted_record(value, traversal_level - 1)
+        return _has_encrypted_field(value)
+    if isinstance(value, (list, tuple)):
+        return any(_is_encrypted_value(item, traversal_level) for item in value)
+    return False
+
+
 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) 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.
+    checks `record` itself (the file-storage pattern) as well as fields up to
+    `traversal_level` deep, then always falls back to an unbounded-depth scan if
+    the bounded pass found nothing - safe to use as a leftover-detecting
+    post-rotation audit; never returns False for a record that still contains a
+    blob, at any depth (including one nested inside a list or tuple), for any
+    traversal_level value, odd or even (both passes walk list/tuple items, not
+    just dict values).
 
-    aliases: is_encrypted_document, is_encrypted_dict — same function
+    Args:
+        record: the record (or bare blob) to check.
+        traversal_level: bounded-pass nesting depth tried before the unbounded
+            fallback scan runs (default 2); purely a fast-path - the fallback
+            always makes the final call when the bounded pass finds nothing.
+
+    Returns:
+        True if record or anything nested under it is an encrypted blob.
+
+    aliases: is_encrypted_document, is_encrypted_dict - same function
     """
     if not isinstance(record, dict):
         return False
@@ -528,9 +648,8 @@ def is_encrypted_record(record, traversal_level: int = 2) -> bool:
 
     if traversal_level > 0:
         for value in record.values():
-            if isinstance(value, dict) and is_encrypted_record(value, traversal_level - 1):
+            if _is_encrypted_value(value, traversal_level - 1):
                 return True
-        return False
 
     return any(_has_encrypted_field(value) for value in record.values())
 
@@ -538,12 +657,25 @@ 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 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.
+    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
+    Args:
+        crypto: instance holding the data key to decrypt with.
+        record: the record (or bare blob) to decrypt. not mutated.
+        traversal_level: max nesting depth to recurse into (default 2).
+
+    Returns:
+        a new dict with encrypted fields decrypted (or record unchanged if not
+        a dict).
+
+    Raises:
+        ValueError: crypto is not initialized with a data key.
+
+    aliases: decrypt_document, decrypt_dict - same function
     """
     if not crypto.master_key:
         raise ValueError("not initialized with data key")
@@ -569,52 +701,19 @@ 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, collision-free SHA-256 hex fingerprint of a dict
+    """deterministic sha256 fingerprint of json-shaped dict data
 
-    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.
+    the input is a record sent over the wire, so it is json-serializable by contract:
+    keys are canonically sorted for an order-independent digest, separators are fixed for
+    whitespace stability, and any non-json scalar (uuid/decimal/path/datetime/objectid) is
+    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.
     """
-    normalized = _fingerprint_normalize(data)
-    encoded = json.dumps(normalized, sort_keys=True, default=_fingerprint_default)
-    return hashlib.sha256(encoded.encode()).hexdigest()
+    return hashlib.sha256(_canonical_json(data).encode()).hexdigest()
 
 
-# function aliases — same functions, naming preference only
+# function aliases - same functions, naming preference only
 is_encrypted_document = is_encrypted_record
 is_encrypted_dict = is_encrypted_record
 decrypt_document = decrypt_record