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@v0.1.4
+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@v0.1.4"
+pip install "envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.0"
 ```
 
 Requires `cryptography` (pulled transitively).
 
-Drop the `@v0.1.4` 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,11 +53,11 @@ 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
 ```
 
@@ -75,9 +73,7 @@ plain = crypto.decrypt_data(enc)               # -> {"ssn": "..."}
 
 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
-`{secure, iv, data}` blob (the file-storage pattern below, where the blob IS the whole
-document) — not just blobs nested under a key.
+whether any encrypted field exists.
 
 ```python
 from envelope_crypto import is_encrypted_record, decrypt_record
@@ -86,11 +82,6 @@ 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
-`decrypt_record`/`reencrypt` call — safe to use as a leftover-detecting audit after
-rotation, regardless of how deep the blob is nested.
-
 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`.
@@ -136,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`
-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.
+re-encrypts with itself. Only `{secure, ...}` fields are touched.
 
 ## Storage patterns
 
@@ -168,4 +152,4 @@ The lib never touches a database; only the caller's storage layer differs.
 
 ## 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 = "0.1.4"
+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/envelope_crypto.py

@@ -17,15 +17,15 @@ envelope-encryption pattern used by KMS-style systems.
 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
+    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("public_key.pem")
+    fp = crypto.get_rsa_key_fingerprint(cfg.local_pub)
     record = caller_lookup(fp)
-    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))
 
 authorize another system (this instance must already hold the DEK):
 
@@ -44,13 +44,6 @@ set, then re-encrypt existing records old -> new:
     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 (the file-storage pattern), not just blobs nested
-under a key. reencrypt fails loud (raises) rather than silently leaving a field
-under the old key — including when a blob sits deeper than `traversal_level`;
-is_encrypted_record falls back to an unbounded-depth scan past traversal_level so
-it reliably catches leftovers as a post-rotation audit.
-
 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.
@@ -61,14 +54,12 @@ 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
 from cryptography.hazmat.primitives.ciphers.aead import AESGCM
@@ -77,83 +68,6 @@ from cryptography.hazmat.primitives.serialization import load_ssh_public_key
 _log = logging.getLogger(__name__)
 
 
-def _load_private_key(key_data: bytes, pw: Optional[bytes]):
-    """load a PEM or OpenSSH private key, normalizing the missing-password error
-
-    cryptography raises TypeError when a key is encrypted but no password was given
-    (PEM raises it on the first call; OpenSSH raises it inside the openssh fallback).
-    both are normalized to a clear ValueError so callers see one error type.
-    """
-    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(
-                    "private key is encrypted but no password was provided"
-                ) from ssh_error
-        if pw is not None:
-            # a password was given but the PEM load still failed — most likely a wrong
-            # password; give a clearer message than cryptography's raw "Bad decrypt"
-            raise ValueError("could not load private key (wrong password or malformed key)") from error
-        raise error
-    except TypeError as error:
-        raise ValueError(
-            "private key is encrypted but no password was provided"
-        ) from error
-
-
-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.
-    """
-    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) contain a blob
-
-    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 _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.
-    """
-    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
 
@@ -233,18 +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,
-        password: Optional[str] = None,
+        self, key_path_or_data: str, is_private: bool = False, is_file: bool = True
     ) -> 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. fingerprinting always uses the public half, so a
-        private and its public key produce the same fingerprint. PEM and OpenSSH
-        private-key formats are both accepted (mirrors decrypt_aes_key_with_rsa). an
-        encrypted key with no/wrong password raises ValueError with a clear message
-        (cryptography raises TypeError for the missing-password case — normalized here).
-        """
+        """return a base64 SHA-256 fingerprint of an RSA key for identification"""
         if is_file:
             with open(key_path_or_data, "rb") as key_file:
                 key_data = key_file.read()
@@ -256,13 +161,21 @@ class EnvelopeCrypto:
             )
 
         if is_private:
-            pw = password.encode() if password else None
-            private_key = _load_private_key(key_data, pw)
+            private_key = serialization.load_pem_private_key(key_data, password=None)
             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
 
@@ -276,7 +189,10 @@ class EnvelopeCrypto:
         else:
             key_data = rsa_key.encode() if isinstance(rsa_key, str) else rsa_key
 
-        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)
 
         wrapped = public_key.encrypt(
             aes_key,
@@ -286,8 +202,7 @@ class EnvelopeCrypto:
                 label=None,
             ),
         )
-        # fingerprint from the already-loaded public_key — no second open/parse of the file
-        fingerprint = _fingerprint_of(public_key)
+        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
@@ -299,8 +214,17 @@ class EnvelopeCrypto:
         """unwrap an AES key with an RSA private key"""
         with open(rsa_private_key_path, "rb") as key_file:
             key_data = key_file.read()
-        pw = password.encode() if password else None
-        private_key = _load_private_key(key_data, pw)
+        try:
+            private_key = serialization.load_pem_private_key(
+                key_data, password=password.encode() if password else None
+            )
+        except ValueError as error:
+            if b"BEGIN OPENSSH PRIVATE KEY" in key_data:
+                private_key = serialization.load_ssh_private_key(
+                    key_data, password=password.encode() if password else None
+                )
+            else:
+                raise error
 
         wrapped = base64.b64decode(encrypted_key_base64)
         aes_key = private_key.decrypt(
@@ -346,10 +270,6 @@ class EnvelopeCrypto:
         """encrypt a dict or string under the data key with a unique IV"""
         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__}")
 
         data_str = json.dumps(data) if isinstance(data, dict) else data
         iv = os.urandom(12)
@@ -362,34 +282,18 @@ class EnvelopeCrypto:
         }
 
     def decrypt_data(self, encrypted_data: Dict[str, str]) -> Union[Dict[str, Any], str]:
-        """decrypt a {secure, iv, data} blob; returns the original dict or string
-
-        `encrypt_data` only json-encodes dicts (a string is stored verbatim), so decrypt
-        only treats a json-OBJECT plaintext as a dict and returns everything else as the
-        raw string. this keeps the type faithful for the common cases: a string that is
-        json-shaped but NOT an object ('123'->'123', 'true'->'true', '[1,2]'->'[1,2]')
-        round-trips as a STRING, not an int/bool/list. the one irreducible ambiguity: a
-        string whose exact value is a json OBJECT ('{"a":1}') decrypts to a dict, because
-        without a type marker it is indistinguishable from a stored dict — don't store a
-        bare string that is a json object if you need it back as a string. existing stored
-        blobs are unaffected — a dict was stored as a json object and still parses to a dict.
-        """
+        """decrypt a {secure, iv, data} blob; returns the original dict or string"""
         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")
 
         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
@@ -397,43 +301,16 @@ class EnvelopeCrypto:
         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 {secure, iv, data} blob (the file-storage pattern, where
-        the blob IS the whole document) it is re-encrypted directly and the result is
-        returned in place of `record` — not nested under a key.
-
-        unlike decrypt_record (which logs a failed field and leaves it encrypted), a
-        per-field decrypt failure here RAISES — rotation must fail loud, since silently
-        keeping a field under the old key would lose it once the old key is retired. for
-        the same reason, a blob nested DEEPER than `traversal_level` also RAISES instead
-        of being silently left under the old key: the caller either needs a higher
-        `traversal_level` or must flatten the record before rotation.
-
-        traversal recurses into nested DICTS only; a blob nested inside a LIST is NOT
-        re-encrypted and is NOT covered by the depth-limit raise above. records in this
-        scheme key blobs by field name, not inside arrays, so this doesn't arise in
-        practice — but if you store list-nested blobs, flatten them to dict fields before
-        rotation or they'll be silently left under the old key.
         """
         if not self.master_key:
             raise ValueError("destination not initialized with data key")
 
-        if _is_blob(record):
-            return self.encrypt_data(source_crypto.decrypt_data(record))
-
-        result = copy.deepcopy(record)
+        result = record.copy()
         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):
-                if traversal_level > 0:
-                    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"
-                    )
+            elif traversal_level > 0 and isinstance(value, dict):
+                result[key] = self.reencrypt(source_crypto, value, traversal_level - 1)
         return result
 
 
@@ -446,46 +323,28 @@ PCICrypto = EnvelopeCrypto  # deprecated legacy alias; remove after all systems
 def is_encrypted_record(record, traversal_level: int = 2) -> bool:
     """return whether a record has any encrypted ({secure, iv, data}) fields
 
-    checks `record` itself (the file-storage pattern stores the blob AS the whole
-    document, not nested under a key) as well as fields up to `traversal_level` deep.
-    beyond that bounded pass, this ALSO does an unbounded-depth scan before giving up —
-    so a blob left behind by a shallower decrypt_record/reencrypt call (nested deeper
-    than their traversal_level) is still reported as encrypted. this makes the function
-    safe to use as a leftover-detecting post-rotation audit: it never returns False for
-    a record that still contains a blob, at any depth.
-
     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):
-            return True
+        if isinstance(value, dict) and value.get("secure") is True:
+            if "iv" in value and "data" in value:
+                return True
 
     if traversal_level > 0:
         for value in record.values():
             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())
+    return False
 
 
-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 {secure, iv, data} blob (the file-storage pattern, where the
-    blob IS the whole document) it is decrypted directly and the decrypted value
-    (dict or string — see decrypt_data) is returned in place of `record`.
-
     failures on a single field are logged and that field is left encrypted, so a
-    partial failure is visible (the {secure,...} blob remains) rather than silent. a
-    failure decrypting `record` itself (the self-blob case above) is likewise logged
-    and the still-encrypted blob is returned unchanged.
+    partial failure is visible (the {secure,...} blob remains) rather than silent.
 
     aliases: decrypt_document, decrypt_dict — same function
     """
@@ -494,16 +353,9 @@ def decrypt_record(crypto: EnvelopeCrypto, record, traversal_level: int = 2) ->
     if not isinstance(record, dict):
         return record
 
-    if _is_blob(record):
-        try:
-            return crypto.decrypt_data(record)
-        except Exception:
-            _log.exception("failed to decrypt record")
-            return copy.deepcopy(record)
-
-    result = copy.deepcopy(record)
+    result = record.copy()
     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: