#!/usr/bin/env python3 """ Sealed-object contract validator (version sealed-object/1). The one wire format every Havre service speaks. The storage backend handles only sealed objects, so it is provably blind: this validator refuses any envelope that carries plaintext or a private key, at any depth. The blob the envelope points at is stored separately. Canonical schema: docs/contract_sealed_object.md. Keep the two in lockstep. Pure standard library, so it runs anywhere with no install. Self-test: python3 sealed_object.py --selftest """ from __future__ import annotations import json import re import sys SCHEMA_VERSION = "sealed-object/1" POSTURES = {"P0", "P1", "P2"} STORAGE_CLASSES = {"hot", "cold"} SCOPES = {"QC", "ON", "CA", "EU"} CONTENT_ALGS = {"AES-256-GCM", "none"} WRAP_ALGS = {"age-x25519", "x25519-sealedbox", "RSA-OAEP-256", "none"} # Keys that must never appear anywhere in the envelope. This is what makes blindness # structural: a producer cannot accidentally (or sneakily) ship readable material. FORBIDDEN_KEYS = { "plaintext", "private_key", "privatekey", "dek", "password", "passphrase", "secret_key", "secretkey", } _HEX64 = re.compile(r"^[0-9a-f]{64}$") def _deep_forbidden(value, path="") -> list[str]: """Return paths of any forbidden key found anywhere in the structure.""" found = [] if isinstance(value, dict): for k, v in value.items(): here = f"{path}.{k}" if path else k if str(k).lower() in FORBIDDEN_KEYS: found.append(here) found.extend(_deep_forbidden(v, here)) elif isinstance(value, list): for i, item in enumerate(value): found.extend(_deep_forbidden(item, f"{path}[{i}]")) return found def validate(obj) -> list[str]: """Return a list of error strings. Empty list means the object is valid.""" errors: list[str] = [] if not isinstance(obj, dict): return ["root: sealed object must be a JSON object"] # Hard prohibition first: no readable material, at any depth. for p in _deep_forbidden(obj): errors.append(f"forbidden key present: {p}") if obj.get("version") != SCHEMA_VERSION: errors.append(f"version: must be '{SCHEMA_VERSION}'") if not isinstance(obj.get("id"), str) or not obj.get("id"): errors.append("id: required non-empty string") posture = obj.get("posture") if posture not in POSTURES: errors.append(f"posture: must be one of {sorted(POSTURES)}") if obj.get("storage_class") not in STORAGE_CLASSES: errors.append(f"storage_class: must be one of {sorted(STORAGE_CLASSES)}") if obj.get("residency_scope") not in SCOPES: errors.append(f"residency_scope: must be one of {sorted(SCOPES)}") alg = obj.get("alg") content_alg = None if not isinstance(alg, dict): errors.append("alg: required object with 'content' and 'wrap'") else: content_alg = alg.get("content") if content_alg not in CONTENT_ALGS: errors.append(f"alg.content: must be one of {sorted(CONTENT_ALGS)}") if alg.get("wrap") not in WRAP_ALGS: errors.append(f"alg.wrap: must be one of {sorted(WRAP_ALGS)}") integrity = obj.get("integrity") if not isinstance(integrity, dict): errors.append("integrity: required object") else: ct = integrity.get("ciphertext_sha256") if not isinstance(ct, str) or not _HEX64.match(ct): errors.append("integrity.ciphertext_sha256: required 64-char lowercase hex") pt = integrity.get("plaintext_sha256") if pt is not None and (not isinstance(pt, str) or not _HEX64.match(pt)): errors.append("integrity.plaintext_sha256: if present, 64-char lowercase hex") if not isinstance(obj.get("payload_ref"), str) or not obj.get("payload_ref"): errors.append("payload_ref: required non-empty string") deks = obj.get("wrapped_deks") if not isinstance(deks, list): errors.append("wrapped_deks: required array (empty only for P0)") deks = [] else: for i, d in enumerate(deks): if not isinstance(d, dict): errors.append(f"wrapped_deks[{i}]: must be an object") continue if not d.get("recipient_ref"): errors.append(f"wrapped_deks[{i}].recipient_ref: required") if not d.get("wrapped_dek"): errors.append(f"wrapped_deks[{i}].wrapped_dek: required") # Posture-specific coherence. if posture == "P0": if content_alg not in (None, "none"): errors.append("P0: alg.content must be 'none' (plain payload)") if deks: errors.append("P0: wrapped_deks must be empty (no keys for plain storage)") elif posture in ("P1", "P2"): if content_alg == "none": errors.append(f"{posture}: alg.content must be a real cipher, not 'none'") if not deks: errors.append(f"{posture}: wrapped_deks must have at least one recipient") return errors def assert_valid(obj) -> None: errs = validate(obj) if errs: raise ValueError("invalid sealed object: " + "; ".join(errs)) def is_valid(obj) -> bool: return not validate(obj) def canonical_bytes(obj) -> bytes: """Deterministic bytes for cross-language byte-identical comparison and signing.""" return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") # --------------------------------------------------------------------------- selftest def _fixtures(): good_p0 = { "version": SCHEMA_VERSION, "id": "obj_p0_1", "posture": "P0", "storage_class": "hot", "residency_scope": "QC", "alg": {"content": "none", "wrap": "none"}, "integrity": {"ciphertext_sha256": "a" * 64}, "payload_ref": "blob/p0/1", "wrapped_deks": [], "metadata": {"size": 10, "created": "2026-06-21T00:00:00Z"}, } good_p2 = { "version": SCHEMA_VERSION, "id": "obj_p2_1", "posture": "P2", "storage_class": "cold", "residency_scope": "CA", "alg": {"content": "AES-256-GCM", "wrap": "age-x25519"}, "integrity": {"ciphertext_sha256": "b" * 64, "plaintext_sha256": "c" * 64}, "payload_ref": "blob/p2/1", "wrapped_deks": [{"recipient_ref": "fp:alice", "wrapped_dek": "AAAA"}], "metadata": {"size": 4096, "created": "2026-06-21T00:00:00Z"}, } bad_plaintext = dict(good_p2, id="obj_bad_pt") bad_plaintext["metadata"] = {"size": 4096, "plaintext": "the secret"} bad_privkey = dict(good_p2, id="obj_bad_pk") bad_privkey["wrapped_deks"] = [ {"recipient_ref": "fp:alice", "wrapped_dek": "AAAA", "private_key": "leak"} ] bad_p0_haskeys = dict(good_p0, id="obj_bad_p0") bad_p0_haskeys["wrapped_deks"] = [{"recipient_ref": "x", "wrapped_dek": "y"}] bad_p2_nodeks = dict(good_p2, id="obj_bad_p2", wrapped_deks=[]) bad_hash = dict(good_p0, id="obj_bad_hash") bad_hash["integrity"] = {"ciphertext_sha256": "nothex"} bad_scope = dict(good_p0, id="obj_bad_scope", residency_scope="US") # name, object, expect_valid return [ ("valid P0 plain (QC)", good_p0, True), ("valid P2 zero-knowledge (CA)", good_p2, True), ("reject plaintext in metadata", bad_plaintext, False), ("reject private_key in wrapped_deks", bad_privkey, False), ("reject P0 carrying keys", bad_p0_haskeys, False), ("reject P2 with no recipients", bad_p2_nodeks, False), ("reject bad ciphertext hash", bad_hash, False), ("reject non-Canada/EU scope (US)", bad_scope, False), ] def _selftest() -> int: passed = 0 failed = 0 for name, obj, expect_valid in _fixtures(): errs = validate(obj) ok = (not errs) == expect_valid mark = "PASS" if ok else "FAIL" if ok: passed += 1 else: failed += 1 detail = "" if expect_valid else f" (rejected for: {errs[0]})" if errs else "" print(f" [{mark}] {name}{detail}") # Canonical form is deterministic regardless of key order. a = {"b": 1, "a": 2} b = {"a": 2, "b": 1} if canonical_bytes(a) == canonical_bytes(b): print(" [PASS] canonical form is order-independent") passed += 1 else: print(" [FAIL] canonical form differs by key order") failed += 1 print(f"\n{passed} passed, {failed} failed") return 1 if failed else 0 if __name__ == "__main__": if "--selftest" in sys.argv: print(f"sealed_object.py self-test ({SCHEMA_VERSION})\n") raise SystemExit(_selftest()) print(__doc__)