#!/usr/bin/env python3 """Clean-room verifier for the CAIN-42 E11 governance-proof bundle. ZERO CAIN imports. python3 verify_e11.py Recomputes the published valid/negative test vectors and re-hashes every bundle artifact. Prints {"result": "INTACT", ...} on success. Derived from verification/govproof/cleanroom/verifier.py. """ from __future__ import annotations import argparse import base64 import hashlib import json import sys from typing import Any, Dict, List, Mapping, Optional, Sequence try: from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey except Exception: # pragma: no cover Ed25519PublicKey = None # type: ignore SCHEMA = "cain.govproof.v1" STAGES = ("IDENTITY", "AUTHORITY", "INTENT", "POLICY", "RISK", "PREDICTION", "DELIBERATION", "QUORUM", "DECISION", "ENVELOPE", "LEASE", "EXECUTION", "OBSERVATION", "EVIDENCE") STAGE_INDEX = {s: i for i, s in enumerate(STAGES)} REQUIRED_STAGES = tuple(s for s in STAGES if s not in ("PREDICTION", "DELIBERATION", "QUORUM")) ALLOW, DENY = "ALLOW", "DENY" RISK_ORDER = ("LOW", "MEDIUM", "HIGH", "CRITICAL") DENIAL_REASONS = { "NO_AUTHORITY", "AUTHORITY_REVOKED", "AUTHORITY_EXPIRED", "AUTHORITY_ATTENUATION_VIOLATED", "SCOPE_EXCEEDED", "PARAMETERS_MUTATED", "TENANT_CROSS", "POLICY_INVALIDATED", "GOVERNANCE_EPOCH_STALE", "RISK_ABOVE_CEILING", "ENVELOPE_VIOLATION", "QUORUM_INSUFFICIENT", "INDEPENDENCE_INSUFFICIENT", "INTENT_MISMATCH", "DECISION_DENIED", "LEASE_EXPIRED", "LEASE_ALREADY_USED", "EXECUTION_MISMATCH", "OBSERVATION_MISMATCH", "EVIDENCE_MISSING", "REPLAY_DETECTED", "BYZANTINE_ANOMALY", "CONSTITUTIONAL_CONFLICT", "FEDERATION_CONFLICT", "IDENTITY_SUBSTITUTED", "MODEL_SUBSTITUTED", "TOOL_SUBSTITUTED", } # ------------------------------------------------------------------ canonical primitives def canon(d: Any) -> bytes: return json.dumps(d, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8") def sha_bytes(b: bytes) -> str: return hashlib.sha256(b).hexdigest() def digest(domain: str, obj: Any) -> str: return sha_bytes(canon({"d": domain, "o": obj})) def _leaf(b: bytes) -> bytes: return hashlib.sha256(b"\x00" + b).digest() def _node(a: bytes, b: bytes) -> bytes: return hashlib.sha256(b"\x01" + a + b).digest() def mth(hs: List[bytes]) -> bytes: if len(hs) == 1: return hs[0] k = 1 while k * 2 < len(hs): k *= 2 return _node(mth(hs[:k]), mth(hs[k:])) def merkle_root(items: Sequence[Any]) -> str: leaves = [_leaf(canon(x)) for x in items] return hashlib.sha256(b"").hexdigest() if not leaves else mth(leaves).hex() def verify_ed(pub_b64: str, payload: Any, sig_b64: str) -> bool: if Ed25519PublicKey is None or not pub_b64 or not sig_b64: return False try: Ed25519PublicKey.from_public_bytes(base64.b64decode(pub_b64)).verify( base64.b64decode(sig_b64), payload if isinstance(payload, bytes) else canon(payload)) return True except (InvalidSignature, ValueError, TypeError): return False def stage_digest(s: Mapping[str, Any]) -> str: return digest("cain.govproof.stage.v1", {"stage": s["stage"], "seq": s["seq"], "prev": s["prev"], "body": s["body"], "signers": s.get("signers", {})}) def genesis_prev(proof_id: str, tenant: str, subject: Mapping[str, Any]) -> str: return digest("cain.govproof.genesis.v1", {"proof_id": proof_id, "tenant": tenant, "subject": dict(subject)}) def subject_digest(subject: Mapping[str, Any]) -> str: return digest("cain.govproof.subject.v1", dict(subject)) def header_payload(proof: Mapping[str, Any]) -> Dict[str, Any]: return {k: v for k, v in proof.items() if k != "issuer_signature_b64"} def stage_payload(s: Mapping[str, Any]) -> Dict[str, Any]: return {"d": "cain.govproof.stage.sig.v1", "stage": s["stage"], "seq": s["seq"], "prev": s["prev"], "body": s["body"]} def expected_effect(resource: str, action: str, params_hash: str) -> str: return digest("cain.govproof.effect.v1", {"resource": resource, "action": action, "params_hash": params_hash}) def risk_le(a: str, b: str) -> bool: try: return RISK_ORDER.index(a) <= RISK_ORDER.index(b) except ValueError: return False def attenuation_violations(child: Mapping[str, Any], parent: Mapping[str, Any]) -> List[str]: v = [] if not set(child["capabilities"]) <= set(parent["capabilities"]): v.append("capabilities_exceed_parent") if not set(child["resources"]) <= set(parent["resources"]): v.append("resources_exceed_parent") if not set(child["actions"]) <= set(parent["actions"]): v.append("actions_exceed_parent") if child["tenant"] != parent["tenant"]: v.append("tenant_changed") if child["principal"] != parent["principal"]: v.append("principal_changed") if float(child["valid_until"]) > float(parent["valid_until"]): v.append("expiration_exceeds_parent") if int(child["governance_epoch"]) < int(parent["governance_epoch"]): v.append("epoch_regression") if parent.get("model_identity") not in (None, "", "*") and \ child.get("model_identity") not in (None, "", parent.get("model_identity")): v.append("model_substitution") if parent.get("tool_identity") not in (None, "", "*") and \ child.get("tool_identity") not in (None, "", parent.get("tool_identity")): v.append("tool_substitution") return v def leaf_scope_problems(leaf: Mapping[str, Any], subject: Mapping[str, Any]) -> List[str]: p = [] if subject["resource"] not in set(leaf["resources"]): p.append("resource_out_of_scope") if subject["action"] not in set(leaf["actions"]): p.append("action_out_of_scope") if subject.get("capability") and subject["capability"] not in set(leaf["capabilities"]): p.append("capability_mismatch") if subject.get("params_hash") and subject["params_hash"] != leaf["parameter_constraints_hash"]: p.append("parameters_mutated") if subject.get("tenant") and subject["tenant"] != leaf["tenant"]: p.append("tenant_cross") return p def _find(proof: Mapping[str, Any], name: str) -> Optional[Mapping[str, Any]]: for s in proof.get("stages", []): if s.get("stage") == name: return s return None # ------------------------------------------------------------------ verification def verify(proof: Mapping[str, Any], *, trusted_signers: Optional[Mapping[str, str]] = None, trusted_issuer: Optional[str] = None, now_ms: Optional[int] = None, revocations: Sequence[str] = (), used_nonces: Sequence[str] = (), expected_tenant: Optional[str] = None, min_epoch: Optional[int] = None, trusted_policy_hashes: Optional[Mapping[str, str]] = None, expected_state_root: Optional[str] = None, evidence_items: Optional[Sequence[Any]] = None) -> Dict[str, Any]: problems: List[str] = [] checks: Dict[str, bool] = {} if proof.get("schema") != SCHEMA: return {"ok": False, "problems": ["BAD_SCHEMA"], "assurance": "NONE"} stages = proof.get("stages") or [] if not stages: return {"ok": False, "problems": ["NO_STAGES"], "assurance": "NONE"} # chain integrity prev = genesis_prev(proof["proof_id"], proof["tenant"], proof["subject"]) last_idx = -1 for i, s in enumerate(stages): if s.get("stage") not in STAGE_INDEX or STAGE_INDEX[s["stage"]] <= last_idx or int(s.get("seq", -1)) != i + 1: problems.append(f"STAGE_ORDER_VIOLATION:{s.get('stage')}") last_idx = STAGE_INDEX.get(s.get("stage"), last_idx) if s.get("prev") != prev: problems.append(f"BROKEN_LINEAGE:{s.get('stage')}") if s.get("digest") != stage_digest(s): problems.append(f"STAGE_DIGEST_MISMATCH:{s.get('stage')}") prev = s.get("digest") if proof.get("tip") != prev: problems.append("TIP_MISMATCH") if problems: return {"ok": False, "problems": sorted(set(problems)), "assurance": "NONE"} if trusted_signers is not None and not _keys_compatible(proof, trusted_signers): problems.append("TRUSTED_KEY_MISMATCH") registry = dict(trusted_signers) if trusted_signers is not None else dict(proof.get("signers") or {}) assurance = "TRUSTED_SIGNERS" if trusted_signers is not None else "EMBEDDED_SIGNERS" issuer = proof.get("issuer") if issuer not in registry or not verify_ed(registry[issuer], header_payload(proof), proof.get("issuer_signature_b64", "")): problems.append("ISSUER_SIGNATURE_INVALID") if trusted_issuer is not None and issuer != trusted_issuer: problems.append("UNTRUSTED_ISSUER") for s in stages: for sid, sig in (s.get("signers") or {}).items(): if sid not in registry or not verify_ed(registry[sid], stage_payload(s), sig): problems.append(f"{s['stage']}:STAGE_SIGNATURE_INVALID:{sid}") if expected_tenant is not None and proof.get("tenant") != expected_tenant: problems.append("TENANT_MISMATCH") rev = set(revocations) if proof.get("result") == ALLOW: _verify_allow(proof, registry, rev, set(used_nonces), now_ms, problems) elif proof.get("result") == DENY: _verify_deny(proof, problems) else: problems.append("UNKNOWN_RESULT") if min_epoch is not None and int(proof.get("governance_epoch", 0)) < int(min_epoch): problems.append("GOVERNANCE_EPOCH_ROLLBACK") if trusted_policy_hashes is not None: ref = proof.get("policy_ref") or {} if trusted_policy_hashes.get(ref.get("policy_id")) != ref.get("hash"): problems.append("POLICY_NOT_TRUSTED") if expected_state_root is not None and proof.get("state_root") != expected_state_root: problems.append("STATE_ROOT_NOT_TRUSTED") if evidence_items is not None: ev = _find(proof, "EVIDENCE") if ev is None or merkle_root(list(evidence_items)) != ev["body"].get("evidence_root"): problems.append("EVIDENCE_ROOT_MISMATCH") return {"ok": not problems, "problems": sorted(set(problems)), "assurance": assurance} def _keys_compatible(proof, trusted) -> bool: emb = proof.get("signers") or {} for k, v in trusted.items(): if k in emb and emb[k] != v: return False return True def _verify_allow(proof, registry, rev, nonces, now_ms, problems): subject = proof["subject"] epoch = int(proof.get("governance_epoch", 0)) now = now_ms if now_ms is not None else int(proof.get("created_ms", 0)) na = proof.get("not_after_ms") if na is not None and now > int(na): problems.append("PROOF_EXPIRED") present = {s["stage"] for s in proof["stages"]} if not set(REQUIRED_STAGES) <= present: problems.append("ALLOW_REQUIRES_FULL_STAGE_SET") return if proof["stages"][-1]["stage"] != "EVIDENCE": problems.append("ALLOW_REQUIRES_FULL_STAGE_SET") ident = _find(proof, "IDENTITY")["body"] if (ident.get("agent_id") != subject["agent"] or ident.get("principal") != subject["principal"] or ident.get("tenant") != proof["tenant"] or ident.get("revoked") or ident.get("agent_key_b64") != registry.get(subject["agent"])): problems.append("IDENTITY_BINDING_FAILED") chain = _find(proof, "AUTHORITY")["body"].get("chain") or [] if not chain: problems.append("AUTHORITY_CHAIN_EMPTY") else: for i, link in enumerate(chain): if i > 0: for viol in attenuation_violations(link, chain[i - 1]): problems.append("ATTENUATION:" + viol) if link.get("parent_authority") != chain[i - 1].get("authority_id"): problems.append("PARENT_LINK_MISMATCH") if link.get("authority_id") in rev or link.get("agent") in rev or link.get("revoked"): problems.append("AUTHORITY_REVOKED") if now > float(link.get("valid_until", 0)) or now < float(link.get("valid_from", 0)): problems.append("AUTHORITY_TEMPORALLY_INVALID") if link.get("principal") != subject["principal"] or link.get("tenant") != subject["tenant"]: problems.append("AUTHORITY_PRINCIPAL_OR_TENANT_MISMATCH") if int(link.get("governance_epoch", -1)) != epoch: problems.append("AUTHORITY_EPOCH_MISMATCH") for prob in leaf_scope_problems(chain[-1], subject): problems.append("LEAF_SCOPE:" + prob) sigs = _find(proof, "AUTHORITY").get("signers") or {} for d in sorted(x for x in {l.get("delegator") for l in chain} if x): if d not in sigs: problems.append("DELEGATOR_SIGNATURE_MISSING:" + d) ib = _find(proof, "INTENT")["body"] if not chain or ib.get("authority_id") != chain[-1]["authority_id"] \ or ib.get("action_digest") != subject_digest(subject) or ib.get("tenant") != proof["tenant"] \ or not ib.get("goal_lineage"): problems.append("INTENT_BINDING_FAILED") pb = _find(proof, "POLICY")["body"] ref = proof.get("policy_ref") or {} if pb.get("policy_hash") != ref.get("hash") or pb.get("version") != ref.get("version") \ or int(pb.get("epoch", -1)) != epoch or not pb.get("rule_digest"): problems.append("POLICY_BINDING_FAILED") rb = _find(proof, "RISK")["body"] if rb.get("level") not in RISK_ORDER or not risk_le(rb.get("level"), rb.get("ceiling")): problems.append("RISK_ABOVE_CEILING") pred = _find(proof, "PREDICTION") if rb.get("level") in ("HIGH", "CRITICAL") and pred is None: problems.append("PREDICTION_REQUIRED_FOR_HIGH_RISK") if pred is not None and pred["body"].get("input_commitment") != subject_digest(subject): problems.append("PREDICTION_NOT_BOUND_TO_ACTION") quorum = _find(proof, "QUORUM") if rb.get("level") in ("MEDIUM", "HIGH", "CRITICAL") and quorum is None: problems.append("QUORUM_REQUIRED_FOR_MEDIUM_RISK") if quorum is not None: qb = quorum["body"] if int(qb.get("independent_classes", 0)) < int(qb.get("threshold_classes", 1)) or \ int(qb.get("failure_domains", 0)) < int(qb.get("threshold_failure_domains", 1)): problems.append("QUORUM_INSUFFICIENT") counted, excluded = qb.get("counted") or [], set(qb.get("excluded") or []) if set(counted) & excluded: problems.append("QUORUM_COUNTS_EXCLUDED_PARTICIPANT") keys = [registry.get(c) for c in counted] if any(k is None for k in keys) or len(set(keys)) != len(keys): problems.append("QUORUM_DUPLICATE_OR_UNKNOWN_KEYS") sigs = quorum.get("signers") or {} if not all(c in sigs for c in counted): problems.append("QUORUM_VOTER_SIGNATURE_MISSING") if int(qb.get("independent_classes", 0)) > len(set(keys)): problems.append("QUORUM_CLASSES_EXCEED_VOTERS") db = _find(proof, "DECISION")["body"] if db.get("outcome") != ALLOW: problems.append("DECISION_NOT_ALLOW") if quorum is not None and db.get("quorum_stage_digest") != quorum["digest"]: problems.append("DECISION_QUORUM_MISMATCH") if not db.get("rule_digest"): problems.append("DECISION_RULE_MISSING") eb = _find(proof, "ENVELOPE")["body"] amt = int(subject.get("amount", 0)) env_ok = (subject["resource"] not in set(eb.get("forbidden_resources") or []) and subject["resource"] in set(eb.get("allowed_resources") or []) and subject["action"] in set(eb.get("allowed_actions") or []) and amt <= int(eb.get("max_amount", 0)) and risk_le(rb.get("level"), eb.get("max_risk"))) if subject.get("irreversible") and not eb.get("irreversible_allowed"): env_ok = False if not env_ok: problems.append("ENVELOPE_VIOLATION") lb = _find(proof, "LEASE")["body"] if not (chain and lb.get("authority_id") == chain[-1]["authority_id"] and lb.get("tenant") == subject["tenant"] and lb.get("principal") == subject["principal"] and lb.get("agent") == subject["agent"] and lb.get("capability") == subject.get("capability") and lb.get("action") == subject["action"] and lb.get("resource") == subject["resource"] and lb.get("params_hash") == subject["params_hash"] and lb.get("tool_ref") == subject["tool_ref"] and lb.get("model_ref") == subject["model_ref"] and int(lb.get("governance_epoch", -1)) == epoch and lb.get("policy_version") == ref.get("version") and lb.get("state_root") == proof.get("state_root")): problems.append("LEASE_BINDING_FAILED") if lb.get("nonce") in nonces: problems.append("LEASE_NONCE_REPLAYED") xb = _find(proof, "EXECUTION")["body"] if not (xb.get("lease_id") == lb.get("lease_id") and xb.get("lease_digest") == digest("cain.govproof.lease.v1", dict(lb)) and xb.get("commitment") == lb.get("commitment") and xb.get("executed_commitment") == lb.get("commitment") and xb.get("status") == "EXECUTED" and xb.get("tool_ref") == subject["tool_ref"] and xb.get("model_ref") == subject["model_ref"] and int(xb.get("start_ms", 0)) >= int(lb.get("issued_ms", 0)) and int(xb.get("end_ms", 0)) >= int(xb.get("start_ms", 0)) and int(xb.get("end_ms", 0)) <= int(lb.get("expires_ms", 0))): problems.append("EXECUTION_MISMATCH") ob = _find(proof, "OBSERVATION")["body"] if not (ob.get("lease_id") == lb.get("lease_id") and ob.get("effect") == expected_effect(subject["resource"], subject["action"], subject["params_hash"]) and int(ob.get("amount", -1)) <= int(eb.get("max_amount", 0))): problems.append("OBSERVATION_MISMATCH") evb = _find(proof, "EVIDENCE")["body"] if int(evb.get("count", 0)) < 1 or not evb.get("evidence_root"): problems.append("EVIDENCE_MISSING") def _verify_deny(proof, problems): denial = proof.get("denial") or {} reason, failed = denial.get("reason"), denial.get("failed_stage") if reason not in DENIAL_REASONS or failed not in STAGE_INDEX: problems.append("DENIAL_MALFORMED") return subj = proof["subject"] stages = {s["stage"]: s["body"] for s in proof["stages"]} chain = (stages.get("AUTHORITY") or {}).get("chain") or [] leaf = chain[-1] if chain else {} risk = stages.get("RISK") or {} quorum = stages.get("QUORUM") or {} env = (denial.get("envelope") or {}) now = int(proof.get("created_ms", 0)) confirmed = False if reason == "NO_AUTHORITY": confirmed = not chain elif reason == "AUTHORITY_REVOKED": confirmed = any(l.get("revoked") for l in chain) elif reason == "AUTHORITY_EXPIRED": confirmed = any(float(l.get("valid_until", 0)) < now for l in chain) elif reason == "AUTHORITY_ATTENUATION_VIOLATED": confirmed = any(attenuation_violations(chain[i], chain[i - 1]) for i in range(1, len(chain))) elif reason in ("SCOPE_EXCEEDED", "PARAMETERS_MUTATED", "TENANT_CROSS") and chain: probs = set(leaf_scope_problems(leaf, subj)) key = {"SCOPE_EXCEEDED": {"resource_out_of_scope", "action_out_of_scope", "capability_mismatch"}, "PARAMETERS_MUTATED": {"parameters_mutated"}, "TENANT_CROSS": {"tenant_cross"}}[reason] confirmed = bool(probs & key) elif reason == "RISK_ABOVE_CEILING": confirmed = bool(risk) and not risk_le(risk.get("level"), risk.get("ceiling")) elif reason == "QUORUM_INSUFFICIENT": confirmed = bool(quorum) and int(quorum.get("independent_classes", 0)) < int(quorum.get("threshold_classes", 1)) elif reason == "INDEPENDENCE_INSUFFICIENT": confirmed = bool(quorum) and int(quorum.get("failure_domains", 0)) < int(quorum.get("threshold_faildomains", 1)) elif reason == "EVIDENCE_MISSING": confirmed = "EVIDENCE" not in stages elif reason == "DECISION_DENIED": confirmed = (stages.get("DECISION") or {}).get("outcome") == DENY elif reason == "ENVELOPE_VIOLATION": confirmed = bool(env) and bool(env.get("violations")) if not confirmed: problems.append("DENIAL_NOT_SUBSTANTIATED") def main(argv: Optional[Sequence[str]] = None) -> int: ap = argparse.ArgumentParser(description="Clean-room CAIN governance proof verifier") ap.add_argument("proof") ap.add_argument("--trusted-signers") ap.add_argument("--now-ms", type=int) ap.add_argument("--min-epoch", type=int) ap.add_argument("--policy", action="append", default=[]) ap.add_argument("--state-root") ap.add_argument("--revoked", action="append", default=[]) ap.add_argument("--used-nonce", action="append", default=[]) ap.add_argument("--evidence-items") args = ap.parse_args(argv) proof = json.load(open(args.proof, encoding="utf-8")) signers = json.load(open(args.trusted_signers, encoding="utf-8")) if args.trusted_signers else None policies = {} for spec in args.policy: pid, _, h = spec.partition("=") policies[pid] = h items = json.load(open(args.evidence_items, encoding="utf-8")) if args.evidence_items else None res = verify(proof, trusted_signers=signers, now_ms=args.now_ms, min_epoch=args.min_epoch, trusted_policy_hashes=(policies or None), expected_state_root=args.state_root, revocations=args.revoked, used_nonces=args.used_nonce, evidence_items=items) if res["ok"]: print(f"CAIN E11 GOVERNANCE PROOF VERIFIED (independent clean-room; assurance={res['assurance']})") return 0 print("CAIN E11 GOVERNANCE PROOF REJECTED:") for p in res["problems"]: print(" -", p) return 1 # ------------------------------------------------------------------ E11 bundle runner def _sha_file(p): return hashlib.sha256(p.read_bytes()).hexdigest() def run_bundle(bundle_dir): """Recompute every published vector and every artifact hash. Returns (passed, checks, problems).""" import pathlib b = pathlib.Path(bundle_dir) passed = 0 checks = 0 problems = [] valid = json.loads((b / "test_vectors" / "valid_allow.json").read_text()) vproof, vctx = valid["proof"], dict(valid["context"]) checks += 1 if verify(vproof, **vctx)["ok"]: passed += 1 else: problems.append("valid_allow: expected PASS, got FAIL") negatives = json.loads((b / "test_vectors" / "negative.json").read_text()) for n in negatives: proof, ctx, needle = n["proof"], dict(n["context"]), n["needle"] if n["name"] == "revocation_bypass": ctx["revocations"] = ["auth:child"] res = verify(proof, **ctx) checks += 2 if res["ok"]: problems.append(f"{n['name']}: expected FAIL, got PASS") else: passed += 1 if any(needle in p for p in res["problems"]) or needle == "QUORUM" and any("QUORUM" in p for p in res["problems"]): passed += 1 else: problems.append(f"{n['name']}: expected problem containing {needle!r}") manifest = json.loads((b / "bundle_hashes.json").read_text()) for rel, want in sorted(manifest.items()): if rel == "bundle_hashes.json": continue checks += 1 p = b / rel if not p.exists() or _sha_file(p) != want: problems.append(f"hash mismatch: {rel}") else: passed += 1 return passed, checks, problems if __name__ == "__main__": import pathlib target = sys.argv[1] if len(sys.argv) > 1 else "." p, c, pr = run_bundle(target) if pr: print(json.dumps({"result": "BROKEN", "passed": p, "checks": c, "problems": pr[:20]})) raise SystemExit(1) print(json.dumps({"result": "INTACT", "passed": p, "checks": c})) raise SystemExit(0)