add package: pyproject + src

EnvelopeCrypto: hybrid envelope encryption for dict records — a random
AES-256-GCM data key (DEK) encrypts the data, wrapped per-system via
RSA-OAEP (SHA-256) for distribution. config-free (DEK + key paths
injected), storage-agnostic, object-only. covers bootstrap/self_test,
authorize/deauthorize, rotate + reencrypt, and record-level decrypt.
src/ layout, hatchling build, cryptography backend.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: disqualifier <dev@disqualifier.me>

.gitignore

@@ -1,5 +1,5 @@
 # claude
-.claude/
+CLAUDE.md
 
 # python
 __pycache__/

README.md

@@ -11,19 +11,17 @@ and storage-agnostic.
 `requirements.txt`:
 
 ```
-envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v1.0.0
+envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.0
 ```
 
 Direct:
 
 ```bash
-pip install "envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v1.0.0"
+pip install "envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.0"
 ```
 
 Requires `cryptography` (pulled transitively).
 
-Drop the `@v1.0.0` suffix from the line above to install the latest unpinned.
-
 ## First-time setup
 
 Run once, ever, to create the data key and authorize the first system. You need an
@@ -39,10 +37,10 @@ openssl rsa -in local_priv.pem -pubout -out local_pub.pem
 from envelope_crypto import EnvelopeCrypto
 
 # generate the DEK and wrap it for this system in one call
-crypto, fingerprint, wrapped = EnvelopeCrypto.bootstrap("public_key.pem")
+crypto, fingerprint, wrapped = EnvelopeCrypto.bootstrap(cfg.local_pub)
 
 # verify the keypair actually round-trips BEFORE storing anything
-crypto.self_test("public_key.pem", "private_key.pem")        # raises if keys don't pair
+crypto.self_test(cfg.local_pub, cfg.local_priv)        # raises if keys don't pair
 
 # store the wrapped key — this is now the ONLY record of the DEK
 await db.create_document("keys", {"_id": fingerprint, "key": wrapped})
@@ -55,24 +53,17 @@ re-derived each boot by unwrapping. **Never persist the plaintext key.**
 
 ```python
 crypto = EnvelopeCrypto()
-fingerprint = crypto.get_rsa_key_fingerprint("public_key.pem")
+fingerprint = crypto.get_rsa_key_fingerprint(cfg.local_pub)
 record = await db.get_document("keys", {"_id": fingerprint})
 if not record:
     raise RuntimeError("this system is not authorized")
-crypto.initialize(crypto.decrypt_aes_key_with_rsa(record["key"], "private_key.pem"))
+crypto.initialize(crypto.decrypt_aes_key_with_rsa(record["key"], cfg.local_priv))
 bot.crypto = crypto
 ```
 
 The `keys` schema (`_id` = fingerprint, `key` = wrapped) is the **caller's** choice;
 this lib only produces `(fingerprint, wrapped_key)`.
 
-`decrypt_aes_key_with_rsa` (like `encrypt_aes_key_with_rsa` and
-`get_rsa_key_fingerprint`) takes `is_file` (default `True`). Pass `is_file=False` to
-hand it PEM/OpenSSH key data directly — e.g. a private key sourced from a vault —
-instead of a file path. `self_test` threads the same `is_file` through to both the
-public and private key it loads, so `self_test(pub_pem, priv_pem, is_file=False)`
-round-trips two in-memory PEM strings rather than treating them as paths.
-
 ## Encrypt / decrypt
 
 ```python
@@ -80,18 +71,9 @@ enc = crypto.encrypt_data({"ssn": "..."})      # -> {"secure": True, "iv": ...,
 plain = crypto.decrypt_data(enc)               # -> {"ssn": "..."}
 ```
 
-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
-whether any encrypted field exists. Both also detect `doc` itself being a bare
+whether any encrypted field exists.
-`{secure, iv, data}` blob (the file-storage pattern below, where the blob IS the whole
-document) — not just blobs nested under a key.
 
 ```python
 from envelope_crypto import is_encrypted_record, decrypt_record
@@ -100,12 +82,6 @@ if is_encrypted_record(doc):
     doc = decrypt_record(crypto, doc)
 ```
 
-`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 (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`.
@@ -151,14 +127,7 @@ for fingerprint, wrapped_key in wrapped.items():
 
 `reencrypt(source_crypto, record)` is a method on the **destination** (new-key)
 instance: it decrypts each encrypted field with `source_crypto` (old key) and
-re-encrypts with itself. Only `{secure, ...}` fields are touched — including `record`
+re-encrypts with itself. Only `{secure, ...}` fields are touched.
-itself if it IS a `{secure, iv, data}` blob (the file-storage pattern).
-
-Rotation must fail loud: a per-field decrypt failure raises, and so does a blob nested
-deeper than `traversal_level` — silently leaving it under the old key would strand it
-once the old key's wrapped-key record is deleted below. If you nest blobs deeper than
-the default `traversal_level=2`, pass a higher `traversal_level` or flatten the record
-first.
 
 ## Storage patterns
 
@@ -180,19 +149,7 @@ The lib never touches a database; only the caller's storage layer differs.
 - The scheme is envelope/hybrid encryption (AES-256-GCM data key wrapped by RSA-OAEP).
   Using it does not by itself confer PCI-DSS or any other compliance — that is a
   whole-system property.
-- `initialize(master_key)` requires exactly 32 bytes (`bytes`, `len == 32`) — a
-  16/24-byte key or a `str` raises `ValueError` instead of silently downgrading to
-  AES-128/192 or failing late at first encrypt.
-- Wrap/unwrap (`encrypt_aes_key_with_rsa`, `decrypt_aes_key_with_rsa`,
-  `authorize_system`, `bootstrap`, `rotate_master_key`) is **RSA-only**. A non-RSA
-  key (e.g. Ed25519/EC) loads and fingerprints fine but raises a clear `ValueError`
-  at wrap/unwrap rather than a raw `AttributeError`.
-- `fingerprint_data` serializes non-JSON-native values (datetime/date/time,
-  bytes/bytearray, ObjectId-like objects) via a stable `default=` handler instead of
-  raising `TypeError`, and type-tags dict keys internally so `{1: "a"}` and
-  `{"1": "a"}` no longer collide to the same fingerprint. Never logs the data it
-  fingerprints.
 
 ## Versioning
 
-Releases are tagged `vX.Y.Z`. The install line above pins a release; drop the `@vX.Y.Z` suffix to install the latest unpinned. Pin deliberately for reproducible installs.
+Tagged `vX.Y.Z`. Pin the tag in `requirements.txt`.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "envelope_crypto"
-version = "1.0.0"
+version = "0.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,5 +1,3 @@
-from importlib.metadata import version, PackageNotFoundError
-
 from .envelope_crypto import (
     EnvelopeCrypto,
     DocumentCrypto,
@@ -14,11 +12,6 @@ from .envelope_crypto import (
     fingerprint_data,
 )
 
-try:
-    __version__ = version("envelope_crypto")
-except PackageNotFoundError:
-    __version__ = "0.0.0+unknown"
-
 __all__ = [
     "EnvelopeCrypto",
     "DocumentCrypto",
@@ -31,5 +24,4 @@ __all__ = [
     "decrypt_document",
     "decrypt_dict",
     "fingerprint_data",
-    "__version__",
 ]

src/envelope_crypto/envelope_crypto.py

@@ -1,129 +1,72 @@
 """
 envelope encryption for dict records
 
-hybrid encryption: a random AES-256-GCM data key (DEK) encrypts the data, wrapped
+hybrid encryption: a random AES-256-GCM data key (DEK) encrypts the data, and
-(RSA-OAEP) per authorized system's public key (KEK) for distribution and stored by
+that key is wrapped (RSA-OAEP) per authorized system's public key (KEK) for
-the caller, keyed by fingerprint. RSA-envelope only - a non-RSA key (e.g. Ed25519/EC)
+distribution. the wrapped key is stored by the caller, keyed by fingerprint;
-loads and fingerprints fine but raises ValueError at wrap/unwrap. never logs key
+each system unwraps its own copy with its private key. this is the same
-material (DEK, PEM, wrapped key) - only fingerprints and counts. config-free and
+envelope-encryption pattern used by KMS-style systems.
-storage-agnostic; see README for bootstrap/boot/authorize/rotate flows.
 
     from envelope_crypto import EnvelopeCrypto
 
     crypto = EnvelopeCrypto()
-    crypto.initialize(master_key)                      # exactly 32 bytes (AES-256)
+    crypto.initialize(master_key)                      # 32-byte AES DEK
     enc = crypto.encrypt_data({"ssn": "..."})          # -> {secure, iv, data}
     plain = crypto.decrypt_data(enc)                   # -> original
 
+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(cfg.local_pub)
+    crypto.self_test(cfg.local_pub, cfg.local_priv)    # 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(cfg.local_pub)
+    record = caller_lookup(fp)
+    crypto.initialize(crypto.decrypt_aes_key_with_rsa(record["key"], cfg.local_priv))
+
+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.
+
+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))
+
+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.
+
 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
-variants are the same functions - use whichever fits your storage.
+function variants are the same functions — use whichever fits your storage.
 """
 
 import os
-import copy
 import json
 import base64
 import hashlib
 import logging
 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, rsa
+from cryptography.hazmat.primitives.asymmetric import padding
 from cryptography.hazmat.primitives.ciphers.aead import AESGCM
 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,
-)
-
-
-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:
-        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"""
-    try:
-        return serialization.load_pem_private_key(key_data, password=pw)
-    except ValueError as error:
-        if b"BEGIN OPENSSH PRIVATE KEY" in key_data:
-            try:
-                return serialization.load_ssh_private_key(key_data, password=pw)
-            except TypeError as ssh_error:
-                raise ValueError(_password_mismatch_message(pw)) from ssh_error
-        if pw is not None:
-            raise ValueError("could not load private key (wrong password or malformed key)") from error
-        raise error
-    except TypeError as error:
-        raise ValueError(_password_mismatch_message(pw)) from error
-
-
-def _fingerprint_of(public_key) -> str:
-    """base64 SHA-256 fingerprint of an already-loaded public key"""
-    key_bytes = public_key.public_bytes(
-        encoding=serialization.Encoding.DER,
-        format=serialization.PublicFormat.SubjectPublicKeyInfo,
-    )
-    digest = hashes.Hash(hashes.SHA256())
-    digest.update(key_bytes)
-    return base64.b64encode(digest.finalize()).decode()
-
-
-def _is_blob(value: Any) -> bool:
-    """return whether value has the {secure, iv, data} encrypted-blob shape"""
-    return isinstance(value, dict) and value.get("secure") is True and "iv" in value and "data" in value
-
-
-def _has_encrypted_field(record: Any) -> bool:
-    """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
-
-
-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"""
-    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"""
-    try:
-        return serialization.load_pem_public_key(key_data)
-    except ValueError:
-        try:
-            return load_ssh_public_key(key_data)
-        except (UnsupportedAlgorithm, ValueError) as error:
-            raise ValueError("not a valid PEM or OpenSSH public key") from error
-
 
 class EnvelopeCrypto:
     """hybrid RSA/AES-256-GCM envelope encryption for dict records
@@ -140,20 +83,10 @@ 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
 
-        the plaintext DEK is never returned or persisted; it survives only as the
+        returns (crypto, fingerprint, wrapped_key) — an initialized instance plus
-        wrapped copy. run self_test before storing to confirm the keypair round-trips.
+        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
-        Args:
+        before storing to confirm the keypair round-trips.
-            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())
@@ -166,24 +99,10 @@ 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 and unwraps
+        round-trips sample data through this instance's DEK, then wraps the DEK
-        the DEK with the given keypair. run after bootstrap or as a health check.
+        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
-        Args:
+        keypair or wrong key path before relying on it. returns True on success.
-            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")
@@ -194,7 +113,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, is_file=is_file, password=password)
+            recovered = self.decrypt_aes_key_with_rsa(wrapped, rsa_private_key, password=password)
         except Exception as error:
             raise RuntimeError(
                 "self_test: key unwrap failed (public/private keys do not pair, "
@@ -207,15 +126,7 @@ 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")
 
@@ -236,29 +147,9 @@ class EnvelopeCrypto:
         return key
 
     def get_rsa_key_fingerprint(
-        self, key_path_or_data: str, is_private: bool = False, is_file: bool = True,
+        self, key_path_or_data: str, is_private: bool = False, is_file: bool = True
-        password: Optional[str] = None,
     ) -> str:
-        """return a base64 SHA-256 fingerprint of an RSA key for identification
+        """return a base64 SHA-256 fingerprint of an RSA key for identification"""
-
-        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:
                 key_data = key_file.read()
@@ -270,104 +161,89 @@ class EnvelopeCrypto:
             )
 
         if is_private:
-            pw = password.encode() if password else None
+            private_key = serialization.load_pem_private_key(key_data, password=None)
-            private_key = _load_private_key(key_data, pw)
             public_key = private_key.public_key()
         else:
-            public_key = _load_public_key(key_data)
+            try:
+                public_key = serialization.load_pem_public_key(key_data)
+            except ValueError:
+                public_key = load_ssh_public_key(key_data)
 
-        fingerprint = _fingerprint_of(public_key)
+        key_bytes = public_key.public_bytes(
+            encoding=serialization.Encoding.DER,
+            format=serialization.PublicFormat.SubjectPublicKeyInfo,
+        )
+        digest = hashes.Hash(hashes.SHA256())
+        digest.update(key_bytes)
+        fingerprint = base64.b64encode(digest.finalize()).decode()
         _log.info("generated %s key fingerprint", "private" if is_private else "public")
         return fingerprint
 
     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
+        """wrap an AES key with an RSA public key; returns (fingerprint, wrapped_b64)"""
-
-        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:
                 key_data = key_file.read()
         else:
             key_data = rsa_key.encode() if isinstance(rsa_key, str) else rsa_key
 
-        public_key = _load_public_key(key_data)
+        try:
-        _require_rsa(public_key)
+            public_key = serialization.load_pem_public_key(key_data)
+        except ValueError:
+            public_key = load_ssh_public_key(key_data)
 
-        wrapped = public_key.encrypt(aes_key, _OAEP_PADDING)
+        wrapped = public_key.encrypt(
-        fingerprint = _fingerprint_of(public_key)
+            aes_key,
+            padding.OAEP(
+                mgf=padding.MGF1(algorithm=hashes.SHA256()),
+                algorithm=hashes.SHA256(),
+                label=None,
+            ),
+        )
+        fingerprint = self.get_rsa_key_fingerprint(rsa_key, is_private=False, is_file=is_file)
         wrapped_b64 = base64.b64encode(wrapped).decode()
         _log.info("wrapped data key for fingerprint %s", fingerprint[:8])
         return fingerprint, wrapped_b64
 
     def decrypt_aes_key_with_rsa(
-        self, encrypted_key_base64: str, rsa_private_key: str,
+        self, encrypted_key_base64: str, rsa_private_key_path: str,
-        is_file: bool = True, password: Optional[str] = None,
+        password: Optional[str] = None,
     ) -> bytes:
-        """unwrap an AES key with an RSA private key
+        """unwrap an AES key with an RSA private key"""
-
+        with open(rsa_private_key_path, "rb") as key_file:
-        Args:
+            key_data = key_file.read()
-            encrypted_key_base64: the RSA-OAEP wrapped key, base64-encoded.
+        try:
-            rsa_private_key: path to (or, if is_file=False, raw PEM/OpenSSH data
+            private_key = serialization.load_pem_private_key(
-                of) the private key to unwrap with.
+                key_data, password=password.encode() if password else None
-            is_file: True (default) treats rsa_private_key as a path; pass False
+            )
-                to supply PEM/OpenSSH key data directly (e.g. vault-sourced).
+        except ValueError as error:
-            password: password for an encrypted private key, if any.
+            if b"BEGIN OPENSSH PRIVATE KEY" in key_data:
-
+                private_key = serialization.load_ssh_private_key(
-        Returns:
+                    key_data, password=password.encode() if password else None
-            the unwrapped AES data key.
+                )
-
+            else:
-        Raises:
+                raise error
-            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:
-                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(wrapped, _OAEP_PADDING)
+        aes_key = private_key.decrypt(
+            wrapped,
+            padding.OAEP(
+                mgf=padding.MGF1(algorithm=hashes.SHA256()),
+                algorithm=hashes.SHA256(),
+                label=None,
+            ),
+        )
         _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
 
-        requires this instance to already hold the data key - only an
+        returns (fingerprint, wrapped_b64) for the caller to store as that
-        authorized system can authorize others.
+        system's key-authorization record. 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")
@@ -378,20 +254,9 @@ class EnvelopeCrypto:
     ) -> Tuple[bytes, Dict[str, str]]:
         """generate a NEW data key and wrap it for each authorized public key
 
-        does NOT re-encrypt existing data - build a new instance with the new key
+        returns (new_key, {fingerprint: wrapped_b64}). does NOT re-encrypt existing
-        and call reencrypt() on each record. systems not in the list get no
+        data — build a new instance with the new key and call reencrypt() on each
-        wrapped copy (deauthorized).
+        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 = {}
@@ -402,26 +267,9 @@ 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, 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)):
-            raise TypeError(f"encrypt_data expects a dict or str, got {type(data).__name__}")
-        if isinstance(data, dict):
-            non_str_key_types = _non_str_key_types(data)
-            if non_str_key_types:
-                raise TypeError(
-                    "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
         iv = os.urandom(12)
@@ -434,193 +282,80 @@ 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
+        """decrypt a {secure, iv, data} blob; returns the original dict or string"""
-
-        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.
-        """
         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")
 
         iv = base64.b64decode(encrypted_data["iv"])
         ciphertext = base64.b64decode(encrypted_data["data"])
         aesgcm = AESGCM(self.master_key)
         plaintext = aesgcm.decrypt(iv, ciphertext, None).decode()
         try:
-            parsed = json.loads(plaintext)
+            return json.loads(plaintext)
         except json.JSONDecodeError:
             return plaintext
-        return parsed if isinstance(parsed, dict) else plaintext
 
     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. used during rotation. if `record` itself is a bare blob (file-storage
+        as-is. returns a new dict; the input is not mutated. used during rotation.
-        pattern) it is re-encrypted directly and returned in place of `record`.
-
-        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.
-
-        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")
 
-        if _is_blob(record):
+        result = record.copy()
-            return self.encrypt_data(source_crypto.decrypt_data(record))
-
-        result = copy.deepcopy(record)
         for key, value in record.items():
-            if _is_blob(value):
+            if isinstance(value, dict) and value.get("secure") is True and "iv" in value and "data" in value:
                 result[key] = self.encrypt_data(source_crypto.decrypt_data(value))
-            elif isinstance(value, dict):
+            elif traversal_level > 0 and isinstance(value, dict):
-                if traversal_level > 0:
+                result[key] = self.reencrypt(source_crypto, value, traversal_level - 1)
-                    result[key] = self.reencrypt(source_crypto, value, traversal_level - 1)
-                elif _has_encrypted_field(value):
-                    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"
-                    )
         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) as well as fields up to
+    aliases: is_encrypted_document, is_encrypted_dict — same function
-    `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).
-
-    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
 
-    if _is_blob(record):
-        return True
-
     for value in record.values():
-        if _is_blob(value):
+        if isinstance(value, dict) and value.get("secure") is True:
-            return True
+            if "iv" in value and "data" in value:
+                return True
 
     if traversal_level > 0:
         for value in record.values():
-            if _is_encrypted_value(value, traversal_level - 1):
+            if isinstance(value, dict) and is_encrypted_record(value, traversal_level - 1):
                 return True
-
+    return False
-    return any(_has_encrypted_field(value) for value in record.values())
 
 
-def decrypt_record(crypto: EnvelopeCrypto, record, traversal_level: int = 2) -> Union[dict, Any]:
+def decrypt_record(crypto: EnvelopeCrypto, record, traversal_level: int = 2) -> dict:
     """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
+    failures on a single field are logged and that field is left encrypted, so a
-    is decrypted directly and the value (dict or string, see decrypt_data) is
+    partial failure is visible (the {secure,...} blob remains) rather than silent.
-    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.
 
-    Args:
+    aliases: decrypt_document, decrypt_dict — same function
-        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")
     if not isinstance(record, dict):
         return record
 
-    if _is_blob(record):
+    result = record.copy()
-        try:
-            return crypto.decrypt_data(record)
-        except Exception:
-            _log.exception("failed to decrypt record")
-            return copy.deepcopy(record)
-
-    result = copy.deepcopy(record)
     for key, value in record.items():
-        if _is_blob(value):
+        if isinstance(value, dict) and value.get("secure") is True and "iv" in value and "data" in value:
             try:
                 result[key] = crypto.decrypt_data(value)
             except Exception:
@@ -631,24 +366,11 @@ def decrypt_record(crypto: EnvelopeCrypto, record, traversal_level: int = 2) ->
 
 
 def fingerprint_data(data: dict) -> str:
-    """deterministic sha256 fingerprint of json-shaped dict data
+    """return a deterministic SHA-256 hex fingerprint of a dict"""
-
+    return hashlib.sha256(json.dumps(data, sort_keys=True).encode()).hexdigest()
-    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.
-    """
-    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()
 
 
-# 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