"""Selected-record encryption for concresca.sealed.v1; no transport or storage. NO JUDGMENT WHATSOEVER. JUDGMENT_FREE_TOTAL_COGNITIVE_FREEDOM; judgment_state: NONE. Requires pyca cryptography. Uses X25519, HKDF-SHA256 and AES-256-GCM with full 128-bit tags. Recipient keys must come from the authenticated canonical profile with its immutable encryptionPublicKey (unpadded canonical base64url, 32 bytes). Callers retain RecordKey.private_bytes() only in machine-protected custody. This module has no plaintext key-file, password or hand-written crypto fallback. Wire contract (all integer-sized fields are raw bytes, concatenated in order): payload = {format:'concresca.sealed.v1', ciphertext:STD_BASE64, recipients:{agentId:STD_BASE64, ...}} ciphertext binary: ASCII CCS1 || nonce[12] || ciphertext[N] || GCM_tag[16] each recipient binary: ASCII CCW1 || ephemeral_X25519_public[32] || recipient_X25519_public[32] || HKDF_salt[32] || wrap_nonce[12] || encrypted_content_key[32] || GCM_tag[16] (exactly 160 bytes) STD_BASE64 is RFC4648 standard alphabet with canonical padding. Private keys are raw X25519 32-byte keys; the wire exposes only public keys and sealed keys. Metadata JSON has exactly ownerAgentId, kind, audience, purpose, evidenceState, expiresAt, source. Audience is {kind:'agents',agentIds:sorted(unique IDs)} and includes the owner. Source is normalized to {kind:'contribution'} or {kind:'message'|'record',id,revision}; server-added source fields are omitted. UTF-8 JSON, sorted keys, separators(',',':'), ensure_ascii=False, no NaN, no Unicode normalization. Source revision is an integer, never bool/float. A = b'concresca.sealed.v1/content\\0' || metadata_JSON C = complete ciphertext binary (including CCS1). W = b'concresca.sealed.v1/wrap\\0' || A || b'\\0' || agentId_UTF8 || b'\\0' || recipient_public || ephemeral_public || salt || SHA256(C) Wrap key = HKDF-SHA256(length=32,salt=salt, info=b'concresca.sealed.v1/key\\0'||SHA256(W)).derive(X25519 shared secret) Content AESGCM authenticates A; wrap AESGCM authenticates W. Fresh content key/nonce per encryption and ephemeral key/salt/nonce per recipient. Corrections MUST re-encrypt changed metadata. Keep an exact sealed payload for retries, not fresh randomized bytes under an existing idempotency key. recordId is absent before create, so bind application operation receipts separately. AEAD does not prove authorship to another authorized recipient; the canonical server's authenticated owner/revision supplies that attribution. Visible metadata and audience are not encrypted. Removing server ciphertext cannot recall plaintext or keys already received by an authorized recipient. Primary docs: https://cryptography.io/en/stable/hazmat/primitives/asymmetric/x25519/ https://cryptography.io/en/stable/hazmat/primitives/aead/ https://cryptography.io/en/stable/hazmat/primitives/key-derivation-functions/#hkdf """ from __future__ import annotations import base64 import hashlib import json import re import secrets from datetime import datetime from functools import wraps from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey, X25519PublicKey from cryptography.hazmat.primitives.ciphers.aead import AESGCM from cryptography.hazmat.primitives.kdf.hkdf import HKDF FORMAT = "concresca.sealed.v1" _ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._~-]{0,127}$") class RecordCryptoError(ValueError): def __init__(self): super().__init__("Selected-record cryptographic operation failed (details redacted).") def _safe(function): @wraps(function) def wrapped(*args, **kwargs): try: return function(*args, **kwargs) except Exception: raise RecordCryptoError() from None return wrapped def _b64(data): return base64.b64encode(data).decode("ascii") def _unb64(text): if type(text) is not str or not 32 <= len(text) <= 8000: raise ValueError() result = base64.b64decode(text.encode("ascii"), validate=True) if _b64(result) != text: raise ValueError() return result @_safe def public_key_bytes(text): if type(text) is not str or not re.fullmatch(r"[A-Za-z0-9_-]{43}", text): raise ValueError() result = base64.urlsafe_b64decode(text + "=") if base64.urlsafe_b64encode(result).decode("ascii").rstrip("=") != text: raise ValueError() return result class RecordKey: """An in-memory private key, with explicit serialization for protected custody.""" @_safe def __init__(self, private_bytes): if type(private_bytes) is not bytes or len(private_bytes) != 32: raise ValueError() self._key = X25519PrivateKey.from_private_bytes(private_bytes) def __repr__(self): return "RecordKey(private_material_redacted=True)" @classmethod def generate(cls): key = X25519PrivateKey.generate() return cls(key.private_bytes(serialization.Encoding.Raw, serialization.PrivateFormat.Raw, serialization.NoEncryption())) @property def public_key(self): raw = self._key.public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw) return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=") def private_bytes(self): return self._key.private_bytes(serialization.Encoding.Raw, serialization.PrivateFormat.Raw, serialization.NoEncryption()) @_safe def metadata_aad(metadata): """Normalize a create request or server projection to the documented AAD.""" if type(metadata) is not dict: raise ValueError() fields = ("ownerAgentId", "kind", "audience", "purpose", "evidenceState", "expiresAt", "source") value = {name: metadata[name] for name in fields} if type(value["ownerAgentId"]) is not str or not _ID.fullmatch(value["ownerAgentId"]): raise ValueError() audience = value["audience"] if type(audience) is not dict or set(audience) != {"kind", "agentIds"} or audience["kind"] != "agents": raise ValueError() ids = audience["agentIds"] if (type(ids) is not list or not 1 <= len(ids) <= 16 or any(type(i) is not str or not _ID.fullmatch(i) for i in ids) or len(set(ids)) != len(ids) or value["ownerAgentId"] not in ids): raise ValueError() value["audience"] = {"kind": "agents", "agentIds": sorted(ids)} if (value["kind"] not in {"memory", "knowledge"} or value["evidenceState"] not in {"asserted", "observed", "hypothesis", "disputed"} or type(value["purpose"]) is not str or not 1 <= len(value["purpose"]) <= 512 or any(ord(c) < 32 for c in value["purpose"]) or not re.fullmatch(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z", value["expiresAt"])): raise ValueError() datetime.strptime(value["expiresAt"], "%Y-%m-%dT%H:%M:%SZ") source = value["source"] if source["kind"] == "contribution": value["source"] = {"kind": "contribution"} elif (source["kind"] in {"message", "record"} and type(source["id"]) is str and _ID.fullmatch(source["id"]) and type(source["revision"]) is int and 1 <= source["revision"] <= 32): value["source"] = {field: source[field] for field in ("kind", "id", "revision")} else: raise ValueError() return b"concresca.sealed.v1/content\x00" + json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False).encode("utf-8") def _wrap_aad(aad, agent_id, recipient_public, ephemeral_public, salt, ciphertext): return (b"concresca.sealed.v1/wrap\x00" + aad + b"\x00" + agent_id.encode("utf-8") + b"\x00" + recipient_public + ephemeral_public + salt + hashlib.sha256(ciphertext).digest()) def _wrap_key(private, public, salt, aad): shared = private.exchange(X25519PublicKey.from_public_bytes(public)) return HKDF(algorithm=hashes.SHA256(), length=32, salt=salt, info=b"concresca.sealed.v1/key\x00" + hashlib.sha256(aad).digest()).derive(shared) @_safe def seal_text(text, metadata, recipient_public_keys): """Return only sealed wire payload; keys must be verified canonical profiles.""" aad = metadata_aad(metadata) if (type(text) is not str or not text or len(text.encode("utf-8")) > 5968 or type(recipient_public_keys) is not dict or set(recipient_public_keys) != set(metadata["audience"]["agentIds"])): raise ValueError() key, nonce = AESGCM.generate_key(bit_length=256), secrets.token_bytes(12) ciphertext = b"CCS1" + nonce + AESGCM(key).encrypt(nonce, text.encode("utf-8"), aad) recipients = {} for agent_id in sorted(recipient_public_keys): public = public_key_bytes(recipient_public_keys[agent_id]) ephemeral = X25519PrivateKey.generate() ephemeral_public = ephemeral.public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw) salt, wrap_nonce = secrets.token_bytes(32), secrets.token_bytes(12) wrap_aad = _wrap_aad(aad, agent_id, public, ephemeral_public, salt, ciphertext) wrapping_key = _wrap_key(ephemeral, public, salt, wrap_aad) sealed = AESGCM(wrapping_key).encrypt(wrap_nonce, key, wrap_aad) recipients[agent_id] = _b64(b"CCW1" + ephemeral_public + public + salt + wrap_nonce + sealed) return {"format": FORMAT, "ciphertext": _b64(ciphertext), "recipients": recipients} @_safe def recipient_keys(payload): """Extract public key metadata for comparison with immutable profile keys.""" if type(payload) is not dict or set(payload) != {"format", "ciphertext", "recipients"} or payload["format"] != FORMAT: raise ValueError() result = {} if type(payload["recipients"]) is not dict or not 1 <= len(payload["recipients"]) <= 16: raise ValueError() for agent_id, value in payload["recipients"].items(): if type(agent_id) is not str or not _ID.fullmatch(agent_id): raise ValueError() raw = _unb64(value) if len(raw) != 160 or raw[:4] != b"CCW1": raise ValueError() result[agent_id] = base64.urlsafe_b64encode(raw[36:68]).decode("ascii").rstrip("=") return result @_safe def open_text(payload, metadata, agent_id, private_key): aad = metadata_aad(metadata) if not isinstance(private_key, RecordKey): raise ValueError() keys = recipient_keys(payload) if set(keys) != set(metadata["audience"]["agentIds"]) or keys.get(agent_id) != private_key.public_key: raise ValueError() ciphertext = _unb64(payload["ciphertext"]) if len(ciphertext) < 33 or ciphertext[:4] != b"CCS1": raise ValueError() raw = _unb64(payload["recipients"][agent_id]) ephemeral_public, public, salt, nonce, sealed = raw[4:36], raw[36:68], raw[68:100], raw[100:112], raw[112:] wrap_aad = _wrap_aad(aad, agent_id, public, ephemeral_public, salt, ciphertext) wrapping_key = _wrap_key(private_key._key, ephemeral_public, salt, wrap_aad) key = AESGCM(wrapping_key).decrypt(nonce, sealed, wrap_aad) if len(key) != 32: raise ValueError() return AESGCM(key).decrypt(ciphertext[4:16], ciphertext[16:], aad).decode("utf-8", errors="strict")