chore: ignore .claude/ dir (CLAUDE.md now lives under .claude/)

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

.gitignore

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

README.md

@@ -11,17 +11,19 @@ and storage-agnostic.
 `requirements.txt`:
 
 ```
-envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.0
+envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.3
 ```
 
 Direct:
 
 ```bash
-pip install "envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.0"
+pip install "envelope_crypto @ git+ssh://git@git.rethinkstudios.io/rethink-public/envelope_crypto.git@v0.1.3"
 ```
 
 Requires `cryptography` (pulled transitively).
 
+Drop the `@v0.1.3` 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
@@ -152,4 +154,4 @@ The lib never touches a database; only the caller's storage layer differs.
 
 ## Versioning
 
-Tagged `vX.Y.Z`. Pin the tag in `requirements.txt`.
+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.

pyproject.toml

@@ -0,0 +1,15 @@
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[project]
+name = "envelope_crypto"
+version = "0.1.3"
+description = "Envelope encryption (RSA-OAEP wrapped AES-256-GCM) for dict records — config-free, storage-agnostic, installable."
+requires-python = ">=3.10"
+dependencies = [
+    "cryptography>=42.0",
+]
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/envelope_crypto"]

src/envelope_crypto/__init__.py

@@ -0,0 +1,27 @@
+from .envelope_crypto import (
+    EnvelopeCrypto,
+    DocumentCrypto,
+    RecordCrypto,
+    PCICrypto,
+    is_encrypted_record,
+    is_encrypted_document,
+    is_encrypted_dict,
+    decrypt_record,
+    decrypt_document,
+    decrypt_dict,
+    fingerprint_data,
+)
+
+__all__ = [
+    "EnvelopeCrypto",
+    "DocumentCrypto",
+    "RecordCrypto",
+    "PCICrypto",
+    "is_encrypted_record",
+    "is_encrypted_document",
+    "is_encrypted_dict",
+    "decrypt_record",
+    "decrypt_document",
+    "decrypt_dict",
+    "fingerprint_data",
+]

src/envelope_crypto/envelope_crypto.py

@@ -0,0 +1,458 @@
+"""
+envelope encryption for dict records
+
+hybrid encryption: a random AES-256-GCM data key (DEK) encrypts the data, and
+that key is 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. this is the same
+envelope-encryption pattern used by KMS-style systems.
+
+    from envelope_crypto import EnvelopeCrypto
+
+    crypto = EnvelopeCrypto()
+    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("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. 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 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
+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 _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
+
+    holds an AES data key (DEK), injected via initialize. encrypts/decrypts record
+    fields, wraps/unwraps the DEK with RSA keys for distribution, and re-encrypts
+    records across key rotations. config-free and storage-agnostic.
+    """
+
+    def __init__(self):
+        self.master_key: Optional[bytes] = None
+
+    @classmethod
+    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.
+        """
+        crypto = cls()
+        crypto.initialize(crypto.create_aes_key())
+        fingerprint, wrapped = crypto.authorize_system(rsa_public_key, is_file=is_file)
+        return crypto, fingerprint, wrapped
+
+    def self_test(
+        self, rsa_public_key: str, rsa_private_key: str, *,
+        is_file: bool = True, password: Optional[str] = None,
+    ) -> bool:
+        """verify the full pipeline against a keypair; raises on any mismatch
+
+        round-trips sample data through this instance's DEK, then wraps the DEK
+        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
+        keypair or wrong key path before relying on it. returns True on success.
+        """
+        if not self.master_key:
+            raise ValueError("self_test: not initialized with a key")
+
+        sample = {"_selftest": "ok", "n": 12345}
+        if self.decrypt_data(self.encrypt_data(sample)) != sample:
+            raise RuntimeError("self_test: data round-trip failed")
+
+        _, 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, password=password)
+        except Exception as error:
+            raise RuntimeError(
+                "self_test: key unwrap failed (public/private keys do not pair, "
+                "or wrong password)"
+            ) from error
+        if recovered != self.master_key:
+            raise RuntimeError("self_test: key wrap/unwrap mismatch (public/private keys do not pair)")
+
+        _log.info("self_test passed")
+        return True
+
+    def initialize(self, master_key: bytes) -> None:
+        """arm the instance with the AES data key (DEK)"""
+        self.master_key = master_key
+        _log.info("crypto initialized with data key")
+
+    def shutdown(self) -> None:
+        """drop the data key reference
+
+        note: python cannot guarantee zeroing of immutable bytes in memory; this
+        only releases the reference for garbage collection. do not rely on it to
+        scrub the key from RAM.
+        """
+        self.master_key = None
+        _log.info("data key reference cleared")
+
+    def create_aes_key(self) -> bytes:
+        """generate a random AES-256 data key"""
+        key = os.urandom(32)
+        _log.info("generated new AES data key")
+        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,
+    ) -> 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).
+        """
+        if is_file:
+            with open(key_path_or_data, "rb") as key_file:
+                key_data = key_file.read()
+        else:
+            key_data = (
+                key_path_or_data.encode()
+                if isinstance(key_path_or_data, str)
+                else key_path_or_data
+            )
+
+        if is_private:
+            pw = password.encode() if password else None
+            private_key = _load_private_key(key_data, pw)
+            public_key = private_key.public_key()
+        else:
+            public_key = _load_public_key(key_data)
+
+        fingerprint = _fingerprint_of(public_key)
+        _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; returns (fingerprint, wrapped_b64)"""
+        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)
+
+        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
+        fingerprint = _fingerprint_of(public_key)
+        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_path: str,
+        password: Optional[str] = None,
+    ) -> bytes:
+        """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)
+
+        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,
+            ),
+        )
+        _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.
+        """
+        if not self.master_key:
+            raise ValueError("cannot authorize another system: not initialized")
+        return self.encrypt_aes_key_with_rsa(self.master_key, rsa_public_key, is_file=is_file)
+
+    def rotate_master_key(
+        self, authorized_public_keys: List[str], is_file: bool = True
+    ) -> 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).
+        """
+        new_key = self.create_aes_key()
+        wrapped = {}
+        for pub in authorized_public_keys:
+            fingerprint, wrapped_b64 = self.encrypt_aes_key_with_rsa(new_key, pub, is_file=is_file)
+            wrapped[fingerprint] = wrapped_b64
+        _log.info("rotated data key, wrapped for %d systems", len(wrapped))
+        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"""
+        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)
+        aesgcm = AESGCM(self.master_key)
+        ciphertext = aesgcm.encrypt(iv, data_str.encode(), None)
+        return {
+            "secure": True,
+            "iv": base64.b64encode(iv).decode(),
+            "data": base64.b64encode(ciphertext).decode(),
+        }
+
+    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.
+        """
+        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)
+        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. returns a new dict; the input is not mutated. used during rotation.
+
+        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.
+
+        traversal recurses into nested DICTS only (up to `traversal_level`); a blob nested
+        inside a LIST is NOT re-encrypted. 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 left
+        under the old key.
+        """
+        if not self.master_key:
+            raise ValueError("destination not initialized with data key")
+
+        result = copy.deepcopy(record)
+        for key, value in record.items():
+            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 traversal_level > 0 and isinstance(value, dict):
+                result[key] = self.reencrypt(source_crypto, value, traversal_level - 1)
+        return result
+
+
+# naming aliases — same class
+DocumentCrypto = EnvelopeCrypto
+RecordCrypto = EnvelopeCrypto
+PCICrypto = EnvelopeCrypto  # deprecated legacy alias; remove after all systems migrate
+
+
+def is_encrypted_record(record, traversal_level: int = 2) -> bool:
+    """return whether a record has any encrypted ({secure, iv, data}) fields
+
+    aliases: is_encrypted_document, is_encrypted_dict — same function
+    """
+    if not isinstance(record, dict):
+        return False
+
+    for value in record.values():
+        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
+
+
+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)
+
+    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.
+
+    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
+
+    result = copy.deepcopy(record)
+    for key, value in record.items():
+        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:
+                _log.exception("failed to decrypt field %s", key)
+        elif traversal_level > 0 and isinstance(value, dict):
+            result[key] = decrypt_record(crypto, value, traversal_level - 1)
+    return result
+
+
+def fingerprint_data(data: dict) -> str:
+    """return a deterministic SHA-256 hex fingerprint of a dict"""
+    return hashlib.sha256(json.dumps(data, sort_keys=True).encode()).hexdigest()
+
+
+# function aliases — same functions, naming preference only
+is_encrypted_document = is_encrypted_record
+is_encrypted_dict = is_encrypted_record
+decrypt_document = decrypt_record
+decrypt_dict = decrypt_record