#!/usr/bin/env python3 """cain-mesh-verify — CLEAN-ROOM independent verifier for the CAIN-42 SUPERIOR AI AGENTIC MESH. ZERO CAIN IMPORTS. Reimplements canonical hashing + Ed25519, recomputes delegation-chain attenuation, the mesh verdict, and the observability overlay from serialized evidence. A perfectly signed wrong verdict is still wrong. Usage: python verify.py root [--pinned-pub B64] python verify.py chain [--pinned-pub B64] [--root ROOT.json] python verify.py decision [--pinned-pub B64] [--fingerprint F] [--nonce N] python verify.py overlay python verify.py bundle [--pinned-pub B64] Exit 0 iff the artifact verifies (bundle: chain monotone AND verdict recomputed). """ from __future__ import annotations import argparse import base64 import hashlib import json import sys import time from pathlib import Path from typing import Any, Dict, List, Mapping, Optional, Sequence from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric import ed25519 D_HOP = "CAIN42/MESH-ANCESTRY-DELEGATION-HOP/v1" D_CHAIN = "CAIN42/MESH-ANCESTRY-DELEGATION-CHAIN/v1" D_ROOT = "CAIN42/MESH-ANCESTRY-ROOT-AUTHORITY/v1" D_DECISION = "CAIN42/MESH-DECISION/v1" D_MANIFEST = "CAIN42/MESH-ORIGIN-MANIFEST/v1" D_ATTRIB = "CAIN42/MESH-TRIAL-ATTRIBUTION-CERTIFICATE/v1" D_MANDATE = "CAIN42/MESH-PACT-MANDATE/v1" D_NODE = "CAIN42/MESH-FORUM-NODE-IDENTITY/v1" D_PROOF_ACTION = "CAIN42/SOVEREIGN-PROOF-ACTION/v1" D_NATIVE_ARTIFACT = "CAIN42/SOVEREIGN-NATIVE-ARTIFACT/v1" D_ENFORCEMENT_IR = "CAIN42/SOVEREIGN-ENFORCEMENT-IR/v1" D_SWARM_MSG = "CAIN42/SOVEREIGN-SWARM-MESSAGE/v1" D_SWARM_CAPABILITY = "CAIN42/SOVEREIGN-SWARM-CAPABILITY/v1" D_ASSURANCE_CONTRACT = "CAIN42/SOVEREIGN-ASSURANCE-CONTRACT/v1" D_BOUND_PCA = "CAIN42/SOVEREIGN-ATTESTATION-BOUND-PCA/v1" D_EFFECT_CONTRACT = "CAIN42/SOVEREIGN-EFFECT-CONTRACT/v1" D_RECONCILIATION_RECORD = "CAIN42/SOVEREIGN-RECONCILIATION-RECORD/v1" D_AUTONOMY_TRANSITION = "CAIN42/SOVEREIGN-AUTONOMY-TRANSITION/v1" D_AUTONOMY_INCIDENT = "CAIN42/SOVEREIGN-AUTONOMY-INCIDENT/v1" D_TRIAL_SCENARIO = "CAIN42/SOVEREIGN-TRIAL-SCENARIO/v1" D_ORACLE_RESULT = "CAIN42/SOVEREIGN-ORACLE-RESULT/v1" D_ASSURANCE_SCORECARD = "CAIN42/SOVEREIGN-ASSURANCE-SCORECARD/v1" D_TRIAL_BUNDLE = "CAIN42/SOVEREIGN-TRIAL-EVIDENCE-BUNDLE/v1" SUBSET_DIMS = ("scope", "resources", "actions") NUMERIC_DIMS = ("budget", "time_s", "consequence", "depth_budget") DIMENSIONS = SUBSET_DIMS + NUMERIC_DIMS RESTRICTION_STATES = ("ABSTAIN", "RESTRICT", "DENY") PRECEDENCE = {"ABSTAIN": 0, "RESTRICT": 1, "DENY": 2} GENAI_ATTRS = ("gen_ai.operation.name", "gen_ai.system", "gen_ai.agent.name") CAIN_ATTRS = ("cain.authority_version", "cain.change_fingerprint", "cain.decision", "cain.effect_digest", "cain.outcome", "cain.evidence_roots") def _canon(o: Any) -> bytes: return json.dumps(o, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() def _digest(domain: str, fields: Mapping[str, Any]) -> str: return hashlib.sha256(_canon({"domain": domain, **dict(fields)})).hexdigest() def _verify_sig(pub: str, sig: str, domain: str, fields: Mapping[str, Any]) -> bool: try: ed25519.Ed25519PublicKey.from_public_bytes(base64.b64decode(pub)).verify( base64.b64decode(sig), _digest(domain, fields).encode()) return True except (InvalidSignature, ValueError, TypeError): return False def _res(problems: Sequence[str]) -> Dict[str, Any]: return {"verified": not problems, "problems": list(problems), "trusts_cain_flag": False, "grants_authority": False} def _canon_grant(granted: Mapping[str, Any]) -> Dict[str, Any]: out: Dict[str, Any] = {} for d in DIMENSIONS: v = granted.get(d) if d in SUBSET_DIMS: out[d] = sorted(str(x) for x in (v or ())) elif v is not None: out[d] = float(v) return out def verify_root(rec: Mapping[str, Any], *, pinned_pub: Optional[str] = None) -> Dict[str, Any]: problems: List[str] = [] body = rec.get("body") if not isinstance(body, Mapping): return _res(["ANCESTRY_ROOT_BODY_MISSING"]) body = dict(body) if rec.get("digest") != _digest(D_ROOT, body): problems.append("ANCESTRY_ROOT_DIGEST_MISMATCH") pub = rec.get("signer_pub") if pinned_pub is not None and pub != pinned_pub: problems.append("ANCESTRY_ROOT_SIGNER_NOT_PINNED") if not isinstance(pub, str) or not _verify_sig(pub, rec.get("signature", ""), D_ROOT, body): problems.append("ANCESTRY_ROOT_SIGNATURE_INVALID") return _res(problems) def _widen(parent: Mapping[str, Any], child: Mapping[str, Any]) -> List[str]: bad: List[str] = [] p, ch = _canon_grant(parent), _canon_grant(child) for d in SUBSET_DIMS: if not set(ch.get(d, [])) <= set(p.get(d, [])): bad.append(d) for d in NUMERIC_DIMS: if d in p and d in ch and float(ch[d]) > float(p[d]): bad.append(d) if d in p and d not in ch and d == "depth_budget": bad.append(d) return bad def verify_chain(rec: Mapping[str, Any], *, pinned_pub: Optional[str] = None, root: Optional[Mapping[str, Any]] = None) -> Dict[str, Any]: problems: List[str] = [] body = rec.get("body") if not isinstance(body, Mapping): return _res(["ANCESTRY_CHAIN_BODY_MISSING"]) body = dict(body) if rec.get("digest") != _digest(D_CHAIN, body): problems.append("ANCESTRY_CHAIN_DIGEST_MISMATCH") pub = rec.get("signer_pub") if pinned_pub is not None and pub != pinned_pub: problems.append("ANCESTRY_CHAIN_SIGNER_NOT_PINNED") if not isinstance(pub, str) or not _verify_sig(pub, rec.get("signature", ""), D_CHAIN, body): problems.append("ANCESTRY_CHAIN_SIGNATURE_INVALID") root_principal = str(body.get("root_principal", "")) hops = body.get("hops") or [] if not root_principal: problems.append("ANCESTRY_CHAIN_ROOT_PRINCIPAL_MISSING") if not hops: problems.append("ANCESTRY_CHAIN_EMPTY") return _res(problems) authority = None if root is not None: problems += verify_root(root, pinned_pub=pinned_pub)["problems"] authority = _canon_grant(dict(root.get("body") or {}).get("granted") or {}) prev = root_principal for i, hop in enumerate(hops): if not isinstance(hop, Mapping): problems.append(f"ANCESTRY_HOP_MALFORMED:{i}") continue if int(hop.get("depth", -1)) != i: problems.append(f"ANCESTRY_HOP_DEPTH_GAP:{i}") if str(hop.get("delegator", "")) != prev: problems.append(f"ANCESTRY_HOP_LINK_BROKEN:{i}") if not hop.get("nonce"): problems.append(f"ANCESTRY_HOP_NONCE_MISSING:{i}") if authority is not None: for d in _widen(authority, hop.get("granted") or {}): problems.append(f"ANCESTRY_AUTHORITY_WIDENED:{i}:{d}") authority = _canon_grant(hop.get("granted") or {}) prev = str(hop.get("delegatee", "")) return _res(problems) def _norm(d: str) -> str: return d if d in RESTRICTION_STATES else "RESTRICT" def compose(signals: Sequence[Mapping[str, Any]]) -> str: worst = "ABSTAIN" for s in signals: d = _norm(str(s.get("decision", ""))) if PRECEDENCE[d] > PRECEDENCE[worst]: worst = d return "NO_RESTRICTION" if worst == "ABSTAIN" else worst def verify_decision(rec: Mapping[str, Any], *, pinned_pub: Optional[str] = None, fingerprint: Optional[str] = None, nonce: Optional[str] = None) -> Dict[str, Any]: problems: List[str] = [] body = rec.get("body") if not isinstance(body, Mapping): return _res(["MESH_DECISION_BODY_MISSING"]) body = dict(body) if rec.get("digest") != _digest(D_DECISION, body): problems.append("MESH_DECISION_DIGEST_MISMATCH") pub = rec.get("signer_pub") if pinned_pub is not None and pub != pinned_pub: problems.append("MESH_DECISION_SIGNER_NOT_PINNED") if not isinstance(pub, str) or not _verify_sig(pub, rec.get("signature", ""), D_DECISION, body): problems.append("MESH_DECISION_SIGNATURE_INVALID") recomputed = compose(body.get("signals") or []) if recomputed != str(body.get("verdict", "")).upper(): problems.append(f"MESH_VERDICT_NOT_REPRODUCED:{recomputed}") if str(body.get("verdict", "")).upper() not in ("NO_RESTRICTION", "RESTRICT", "DENY"): problems.append("MESH_VERDICT_UNKNOWN") if fingerprint and body.get("change_fingerprint") != fingerprint: problems.append("MESH_FINGERPRINT_MISMATCH") if nonce is not None and body.get("nonce") != nonce: problems.append("MESH_NONCE_MISMATCH") out = _res(problems) out["recomputed_verdict"] = recomputed return out def verify_overlay(span: Mapping[str, Any]) -> Dict[str, Any]: problems: List[str] = [] attrs = span.get("attributes") if not isinstance(attrs, Mapping): return _res(["LUMEN_ATTRIBUTES_MISSING"]) for k in GENAI_ATTRS + CAIN_ATTRS: if attrs.get(k) in (None, ""): problems.append(f"LUMEN_ATTR_MISSING:{k}") for s in attrs.get("cain.cluster_signals", []) or []: if str(s.get("decision", "")).upper() not in RESTRICTION_STATES: problems.append("LUMEN_SIGNAL_NOT_RESTRICTION_TYPED") return _res(problems) def verify_manifest(rec: Mapping[str, Any], *, pinned_pub: Optional[str] = None) -> Dict[str, Any]: problems: List[str] = [] body = rec.get("body") if not isinstance(body, Mapping): return _res(["ORIGIN_MANIFEST_BODY_MISSING"]) body = dict(body) if rec.get("digest") != _digest(D_MANIFEST, body): problems.append("ORIGIN_MANIFEST_DIGEST_MISMATCH") pub = rec.get("signer_pub") if pinned_pub is not None and pub != pinned_pub: problems.append("ORIGIN_MANIFEST_SIGNER_NOT_PINNED") if not isinstance(pub, str) or not _verify_sig(pub, rec.get("signature", ""), D_MANIFEST, body): problems.append("ORIGIN_MANIFEST_SIGNATURE_INVALID") if not body.get("digest") or not body.get("name"): problems.append("ORIGIN_MANIFEST_UNBOUND") return _res(problems) def verify_attribution(rec: Mapping[str, Any], *, pinned_pub: Optional[str] = None) -> Dict[str, Any]: problems: List[str] = [] body = rec.get("body") if not isinstance(body, Mapping): return _res(["TRIAL_CERT_BODY_MISSING"]) body = dict(body) if rec.get("digest") != _digest(D_ATTRIB, body): problems.append("TRIAL_CERT_DIGEST_MISMATCH") pub = rec.get("signer_pub") if pinned_pub is not None and pub != pinned_pub: problems.append("TRIAL_CERT_SIGNER_NOT_PINNED") if not isinstance(pub, str) or not _verify_sig(pub, rec.get("signature", ""), D_ATTRIB, body): problems.append("TRIAL_CERT_SIGNATURE_INVALID") state = str(body.get("state", "")) if state == "ATTRIBUTED" and not body.get("primary"): problems.append("TRIAL_ATTRIBUTED_WITHOUT_PRIMARY") if state != "ATTRIBUTED" and str(body.get("primary")): problems.append("TRIAL_PRIMARY_WITHOUT_ATTRIBUTION") return _res(problems) def verify_ledger(ledger: Mapping[str, Any]) -> Dict[str, Any]: problems: List[str] = [] entries = ledger.get("entries") or [] prev = "0" * 64 for i, node in enumerate(entries): e = node.get("entry") or {} if str(node.get("prev_hash")) != prev: problems.append(f"NOTARY_CHAIN_BROKEN:{i}") h = hashlib.sha256((prev + hashlib.sha256(_canon(dict(e))).hexdigest()).encode()).hexdigest() if h != str(node.get("hash")): problems.append(f"NOTARY_ENTRY_TAMPERED:{i}") prev = h if str(ledger.get("head")) != prev: problems.append("NOTARY_HEAD_MISMATCH") return _res(problems) def verify_mandate(rec: Mapping[str, Any], *, pinned_pub: Optional[str] = None, now: float = 0.0) -> Dict[str, Any]: problems: List[str] = [] body = rec.get("body") if not isinstance(body, Mapping): return _res(["PACT_MANDATE_BODY_MISSING"]) body = dict(body) if rec.get("digest") != _digest(D_MANDATE, body): problems.append("PACT_MANDATE_DIGEST_MISMATCH") pub = rec.get("signer_pub") if pinned_pub is not None and pub != pinned_pub: problems.append("PACT_MANDATE_SIGNER_NOT_PINNED") if not isinstance(pub, str) or not _verify_sig(pub, rec.get("signature", ""), D_MANDATE, body): problems.append("PACT_MANDATE_SIGNATURE_INVALID") if now and body.get("expiry") and now >= float(body["expiry"]): problems.append("PACT_MANDATE_EXPIRED") return _res(problems) def verify_node(rec: Mapping[str, Any], *, pinned_pub: Optional[str] = None) -> Dict[str, Any]: problems: List[str] = [] body = rec.get("body") if not isinstance(body, Mapping): return _res(["FORUM_NODE_BODY_MISSING"]) body = dict(body) if rec.get("digest") != _digest(D_NODE, body): problems.append("FORUM_NODE_DIGEST_MISMATCH") pub = rec.get("signer_pub") if pinned_pub is not None and pub != pinned_pub: problems.append("FORUM_SELF_DECLARED_IDENTITY") if not isinstance(pub, str) or not _verify_sig(pub, rec.get("signature", ""), D_NODE, body): problems.append("FORUM_NODE_SIGNATURE_INVALID") if not body.get("node_id") or not body.get("org"): problems.append("FORUM_NODE_UNBOUND") return _res(problems) def verify_proof_action(rec: Mapping[str, Any], *, pinned_pub: Optional[str] = None, expected_action: Optional[str] = None, expected_tool: Optional[str] = None, now: Optional[float] = None) -> Dict[str, Any]: problems: List[str] = [] body = rec.get("body") if not isinstance(body, Mapping): return _res(["PROOF_ACTION_BODY_MISSING"]) body = dict(body) if rec.get("digest") != _digest(D_PROOF_ACTION, body): problems.append("PROOF_ACTION_DIGEST_MISMATCH") pub = rec.get("signer_pub") if pinned_pub is not None and pub != pinned_pub: problems.append("PROOF_ACTION_SIGNER_NOT_PINNED") if not isinstance(pub, str) or not _verify_sig(pub, rec.get("signature", ""), D_PROOF_ACTION, body): problems.append("PROOF_ACTION_SIGNATURE_INVALID") ts = now if now is not None else time.time() v_until = float(body.get("valid_until", 0.0)) v_from = float(body.get("valid_from", 0.0)) if v_until > 0 and ts > v_until: problems.append("PROOF_ACTION_EXPIRED") if v_from > 0 and ts < v_from: problems.append("PROOF_ACTION_PREMATURE") if not body.get("nonce"): problems.append("PROOF_ACTION_NONCE_MISSING") proof = body.get("proof") if not isinstance(proof, Mapping): problems.append("PROOF_ACTION_PROOF_TERM_MISSING") else: status = str(proof.get("status", "")).upper() if status != "PROVED": problems.append(f"PROOF_ACTION_NOT_PROVED:{status}") if not proof.get("formula_digest"): problems.append("PROOF_ACTION_FORMULA_DIGEST_MISSING") if expected_action and body.get("action_name") != expected_action: problems.append("PROOF_ACTION_ACTION_MISMATCH") if expected_tool and body.get("tool_id") != expected_tool: problems.append("PROOF_ACTION_TOOL_MISMATCH") if not body.get("policy_id"): problems.append("PROOF_ACTION_POLICY_ID_MISSING") if not body.get("agent_id"): problems.append("PROOF_ACTION_AGENT_ID_MISSING") if not body.get("evidence_root"): problems.append("PROOF_ACTION_EVIDENCE_ROOT_MISSING") return _res(problems) def verify_enforcement_artifact(rec: Mapping[str, Any], *, pinned_pub: Optional[str] = None, expected_action: Optional[str] = None, expected_tool: Optional[str] = None, now: Optional[float] = None) -> Dict[str, Any]: problems: List[str] = [] body = rec.get("body") if not isinstance(body, Mapping): return _res(["NATIVE_ARTIFACT_BODY_MISSING"]) body = dict(body) if rec.get("digest") != _digest(D_NATIVE_ARTIFACT, body): problems.append("NATIVE_ARTIFACT_DIGEST_MISMATCH") pub = rec.get("signer_pub") if pinned_pub is not None and pub != pinned_pub: problems.append("NATIVE_ARTIFACT_SIGNER_NOT_PINNED") if not isinstance(pub, str) or not _verify_sig(pub, rec.get("signature", ""), D_NATIVE_ARTIFACT, body): problems.append("NATIVE_ARTIFACT_SIGNATURE_INVALID") ir = body.get("enforcement_ir") if not isinstance(ir, Mapping): problems.append("ENFORCEMENT_IR_MISSING") else: ir_pub = ir.get("signer_pub") ir_sig = ir.get("signature", "") ir_fields = {k: v for k, v in ir.items() if k not in ("signature", "signer_pub")} if not isinstance(ir_pub, str) or not _verify_sig(ir_pub, ir_sig, D_ENFORCEMENT_IR, ir_fields): problems.append("ENFORCEMENT_IR_SIGNATURE_INVALID") ts_ns = int((now if now is not None else time.time()) * 1_000_000_000) v_until_ns = int(ir.get("valid_until_ns", 0)) v_from_ns = int(ir.get("valid_from_ns", 0)) if v_until_ns > 0 and ts_ns > v_until_ns: problems.append("ENFORCEMENT_IR_EXPIRED") if v_from_ns > 0 and ts_ns < v_from_ns: problems.append("ENFORCEMENT_IR_PREMATURE") if not ir.get("nonce"): problems.append("ENFORCEMENT_IR_NONCE_MISSING") if expected_action and ir.get("action_id") != expected_action: problems.append("ENFORCEMENT_IR_ACTION_MISMATCH") if int(ir.get("currency_max_cents", 0)) < 0 or int(ir.get("resource_units_max", 0)) < 0: problems.append("ENFORCEMENT_IR_NEGATIVE_CEILINGS") ex_id = body.get("execution_identity") if ex_id is not None and isinstance(ex_id, Mapping): exe_hash = str(ex_id.get("executable_sha256", "")) if not exe_hash or len(exe_hash) != 64: problems.append("EXECUTION_IDENTITY_INVALID_BINARY_HASH") return _res(problems) def verify_swarm_message(rec: Mapping[str, Any], *, pinned_pub: Optional[str] = None, expected_tenant: Optional[str] = None, now: Optional[float] = None) -> Dict[str, Any]: problems: List[str] = [] body = rec.get("body") if body is None or not isinstance(body, Mapping): if "schema" in rec: body = {k: v for k, v in rec.items() if k not in ("signature", "signer_pub", "digest")} else: return _res(["SWARM_MESSAGE_BODY_MISSING"]) body = dict(body) expected_dig = _digest(D_SWARM_MSG, body) actual_dig = rec.get("digest") or body.get("digest") if actual_dig and actual_dig != expected_dig: problems.append("SWARM_MESSAGE_DIGEST_MISMATCH") pub = rec.get("signer_pub") or body.get("signer_pub") sig = rec.get("signature") or body.get("signature") or "" if pinned_pub is not None and pub != pinned_pub: problems.append("SWARM_MESSAGE_SIGNER_NOT_PINNED") if not isinstance(pub, str) or not _verify_sig(pub, sig, D_SWARM_MSG, body): problems.append("SWARM_MESSAGE_SIGNATURE_INVALID") if expected_tenant and body.get("tenant_id") != expected_tenant: problems.append("SWARM_MESSAGE_TENANT_MISMATCH") ts_ns = int((now if now is not None else time.time()) * 1_000_000_000) v_until_ns = int(body.get("valid_until_ns", 0)) v_from_ns = int(body.get("valid_from_ns", 0)) if v_until_ns > 0 and ts_ns > v_until_ns: problems.append("SWARM_MESSAGE_EXPIRED") if v_from_ns > 0 and ts_ns < v_from_ns: problems.append("SWARM_MESSAGE_PREMATURE") if not body.get("nonce"): problems.append("SWARM_MESSAGE_NONCE_MISSING") payload = body.get("payload") if payload and isinstance(payload, Mapping): comp_dig = hashlib.sha256(_canon(payload)).hexdigest() if body.get("payload_digest") and body.get("payload_digest") != comp_dig: problems.append("SWARM_MESSAGE_PAYLOAD_DIGEST_MISMATCH") return _res(problems) def verify_swarm_capability(rec: Mapping[str, Any], *, pinned_pub: Optional[str] = None, expected_tenant: Optional[str] = None, now: Optional[float] = None) -> Dict[str, Any]: problems: List[str] = [] body = rec.get("body") if body is None or not isinstance(body, Mapping): if "schema" in rec: body = {k: v for k, v in rec.items() if k not in ("signature", "signer_pub", "digest")} else: return _res(["SWARM_CAPABILITY_BODY_MISSING"]) body = dict(body) expected_dig = _digest(D_SWARM_CAPABILITY, body) actual_dig = rec.get("digest") or body.get("digest") if actual_dig and actual_dig != expected_dig: problems.append("SWARM_CAPABILITY_DIGEST_MISMATCH") pub = rec.get("signer_pub") or body.get("signer_pub") sig = rec.get("signature") or body.get("signature") or "" if pinned_pub is not None and pub != pinned_pub: problems.append("SWARM_CAPABILITY_SIGNER_NOT_PINNED") if not isinstance(pub, str) or not _verify_sig(pub, sig, D_SWARM_CAPABILITY, body): problems.append("SWARM_CAPABILITY_SIGNATURE_INVALID") if expected_tenant and body.get("tenant_id") != expected_tenant: problems.append("SWARM_CAPABILITY_TENANT_MISMATCH") if int(body.get("currency_max_cents", 0)) < 0 or int(body.get("resource_units_max", 0)) < 0: problems.append("SWARM_CAPABILITY_NEGATIVE_CEILINGS") return _res(problems) def verify_runtime_assurance(rec: Mapping[str, Any], *, pinned_pub: Optional[str] = None, now: Optional[float] = None) -> Dict[str, Any]: problems: List[str] = [] body = rec.get("body") if body is None or not isinstance(body, Mapping): if "schema" in rec: body = {k: v for k, v in rec.items() if k not in ("signature", "signer_pub", "digest")} else: return _res(["ASSURANCE_CONTRACT_BODY_MISSING"]) body = dict(body) expected_dig = _digest(D_ASSURANCE_CONTRACT, body) actual_dig = rec.get("digest") or body.get("digest") if actual_dig and actual_dig != expected_dig: problems.append("ASSURANCE_CONTRACT_DIGEST_MISMATCH") pub = rec.get("signer_pub") or body.get("signer_pub") sig = rec.get("signature") or body.get("signature") or "" if pinned_pub is not None and pub != pinned_pub: problems.append("ASSURANCE_CONTRACT_SIGNER_NOT_PINNED") if not isinstance(pub, str) or not _verify_sig(pub, sig, D_ASSURANCE_CONTRACT, body): problems.append("ASSURANCE_CONTRACT_SIGNATURE_INVALID") if int(body.get("consequence_ceiling", 0)) < 0 or int(body.get("blast_radius_ceiling", 0)) < 0: problems.append("ASSURANCE_CONTRACT_NEGATIVE_CEILINGS") ts_ns = int((now if now is not None else time.time()) * 1_000_000_000) created_at = int(body.get("created_at_ns", 0)) freshness = int(body.get("freshness_max_ns", 0)) if freshness > 0 and (ts_ns - created_at) > freshness: problems.append("ASSURANCE_CONTRACT_STALE") return _res(problems) def verify_bound_pca(rec: Mapping[str, Any], *, pinned_pub: Optional[str] = None) -> Dict[str, Any]: problems: List[str] = [] body = rec.get("body") if body is None or not isinstance(body, Mapping): if "schema" in rec: body = {k: v for k, v in rec.items() if k not in ("signature", "signer_pub", "digest")} else: return _res(["BOUND_PCA_BODY_MISSING"]) body = dict(body) expected_dig = _digest(D_BOUND_PCA, body) actual_dig = rec.get("digest") or body.get("digest") if actual_dig and actual_dig != expected_dig: problems.append("BOUND_PCA_DIGEST_MISMATCH") pub = rec.get("signer_pub") or body.get("signer_pub") sig = rec.get("signature") or body.get("signature") or "" if pinned_pub is not None and pub != pinned_pub: problems.append("BOUND_PCA_SIGNER_NOT_PINNED") if not isinstance(pub, str) or not _verify_sig(pub, sig, D_BOUND_PCA, body): problems.append("BOUND_PCA_SIGNATURE_INVALID") for field_name in ("pca_digest", "appraisal_digest", "workload_binary_digest"): d_val = str(body.get(field_name, "")) if not d_val or len(d_val) != 64: problems.append(f"BOUND_PCA_INVALID_{field_name.upper()}") return _res(problems) def verify_effect_contract(rec: Mapping[str, Any], *, pinned_pub: Optional[str] = None) -> Dict[str, Any]: problems: List[str] = [] body = rec.get("body") if body is None or not isinstance(body, Mapping): if "schema" in rec: body = {k: v for k, v in rec.items() if k not in ("signature", "signer_pub", "digest")} else: return _res(["EFFECT_CONTRACT_BODY_MISSING"]) body = dict(body) expected_dig = _digest(D_EFFECT_CONTRACT, body) actual_dig = rec.get("digest") or body.get("digest") if actual_dig and actual_dig != expected_dig: problems.append("EFFECT_CONTRACT_DIGEST_MISMATCH") pub = rec.get("signer_pub") or body.get("signer_pub") sig = rec.get("signature") or body.get("signature") or "" if pinned_pub is not None and pub != pinned_pub: problems.append("EFFECT_CONTRACT_SIGNER_NOT_PINNED") if not isinstance(pub, str) or not _verify_sig(pub, sig, D_EFFECT_CONTRACT, body): problems.append("EFFECT_CONTRACT_SIGNATURE_INVALID") if int(body.get("resource_modifications_max", 0)) < 0 or int(body.get("bytes_affected_max", 0)) < 0: problems.append("EFFECT_CONTRACT_NEGATIVE_LIMITS") obs = body.get("required_observers") if not obs or not isinstance(obs, list) or len(obs) == 0: problems.append("EFFECT_CONTRACT_MISSING_REQUIRED_OBSERVERS") return _res(problems) def verify_reconciliation_record(rec: Mapping[str, Any], *, pinned_pub: Optional[str] = None) -> Dict[str, Any]: problems: List[str] = [] body = rec.get("body") if body is None or not isinstance(body, Mapping): if "schema" in rec: body = {k: v for k, v in rec.items() if k not in ("signature", "signer_pub", "digest", "record_digest")} else: return _res(["RECONCILIATION_RECORD_BODY_MISSING"]) body = dict(body) expected_dig = _digest(D_RECONCILIATION_RECORD, body) actual_dig = rec.get("digest") or body.get("digest") or rec.get("record_digest") if actual_dig and actual_dig != expected_dig: problems.append("RECONCILIATION_RECORD_DIGEST_MISMATCH") pub = rec.get("signer_pub") or body.get("signer_pub") sig = rec.get("signature") or body.get("signature") or "" if pinned_pub is not None and pub != pinned_pub: problems.append("RECONCILIATION_RECORD_SIGNER_NOT_PINNED") if not isinstance(pub, str) or not _verify_sig(pub, sig, D_RECONCILIATION_RECORD, body): problems.append("RECONCILIATION_RECORD_SIGNATURE_INVALID") valid_results = {"CONVERGENT", "DIVERGENT", "SUSPENDED_UNVERIFIABLE", "MATCH", "PARTIAL_MATCH", "VIOLATION"} if str(body.get("result", "")) not in valid_results: problems.append("RECONCILIATION_RECORD_INVALID_RESULT") score = float(body.get("corroboration_score", 0.0)) if score < 0.0 or score > 1.0: problems.append("RECONCILIATION_RECORD_INVALID_SCORE") return _res(problems) def verify_autonomy_transition(rec: Mapping[str, Any], *, pinned_pub: Optional[str] = None) -> Dict[str, Any]: problems: List[str] = [] body = rec.get("body") if body is None or not isinstance(body, Mapping): if "schema" in rec: body = {k: v for k, v in rec.items() if k not in ("signature", "signer_pub", "digest")} else: return _res(["AUTONOMY_TRANSITION_BODY_MISSING"]) body = dict(body) expected_dig = _digest(D_AUTONOMY_TRANSITION, body) actual_dig = rec.get("digest") or body.get("digest") if actual_dig and actual_dig != expected_dig: problems.append("AUTONOMY_TRANSITION_DIGEST_MISMATCH") pub = rec.get("signer_pub") or body.get("signer_pub") sig = rec.get("signature") or body.get("signature") or "" if pinned_pub is not None and pub != pinned_pub: problems.append("AUTONOMY_TRANSITION_SIGNER_NOT_PINNED") if not isinstance(pub, str) or not _verify_sig(pub, sig, D_AUTONOMY_TRANSITION, body): problems.append("AUTONOMY_TRANSITION_SIGNATURE_INVALID") valid_states = {"RESTRICTED", "VERIFIED_BOUNDED", "MONITORED_OPERATION", "SUSPENDED", "REVOKED"} if str(body.get("from_state", "")) not in valid_states or str(body.get("to_state", "")) not in valid_states: problems.append("AUTONOMY_TRANSITION_INVALID_STATE") return _res(problems) def verify_autonomy_incident(rec: Mapping[str, Any], *, pinned_pub: Optional[str] = None) -> Dict[str, Any]: problems: List[str] = [] body = rec.get("body") if body is None or not isinstance(body, Mapping): if "schema" in rec: body = {k: v for k, v in rec.items() if k not in ("signature", "signer_pub", "digest")} else: return _res(["AUTONOMY_INCIDENT_BODY_MISSING"]) body = dict(body) expected_dig = _digest(D_AUTONOMY_INCIDENT, body) actual_dig = rec.get("digest") or body.get("digest") if actual_dig and actual_dig != expected_dig: problems.append("AUTONOMY_INCIDENT_DIGEST_MISMATCH") pub = rec.get("signer_pub") or body.get("signer_pub") sig = rec.get("signature") or body.get("signature") or "" if pinned_pub is not None and pub != pinned_pub: problems.append("AUTONOMY_INCIDENT_SIGNER_NOT_PINNED") if not isinstance(pub, str) or not _verify_sig(pub, sig, D_AUTONOMY_INCIDENT, body): problems.append("AUTONOMY_INCIDENT_SIGNATURE_INVALID") if str(body.get("severity", "")) not in {"LOW", "MEDIUM", "HIGH", "CRITICAL"}: problems.append("AUTONOMY_INCIDENT_INVALID_SEVERITY") return _res(problems) def verify_trial_scenario(rec: Mapping[str, Any], *, pinned_pub: Optional[str] = None) -> Dict[str, Any]: problems: List[str] = [] body = rec.get("body") if body is None or not isinstance(body, Mapping): if "schema" in rec: body = {k: v for k, v in rec.items() if k not in ("signature", "signer_pub", "digest")} else: return _res(["TRIAL_SCENARIO_BODY_MISSING"]) body = dict(body) expected_dig = _digest(D_TRIAL_SCENARIO, body) actual_dig = rec.get("digest") or body.get("digest") if actual_dig and actual_dig != expected_dig: problems.append("TRIAL_SCENARIO_DIGEST_MISMATCH") pub = rec.get("signer_pub") or body.get("signer_pub") sig = rec.get("signature") or body.get("signature") or "" if pinned_pub is not None and pub != pinned_pub: problems.append("TRIAL_SCENARIO_SIGNER_NOT_PINNED") if not isinstance(pub, str) or not _verify_sig(pub, sig, D_TRIAL_SCENARIO, body): problems.append("TRIAL_SCENARIO_SIGNATURE_INVALID") if not body.get("scenario_id") or not body.get("stages"): problems.append("TRIAL_SCENARIO_MALFORMED") return _res(problems) def verify_trial_outcome(rec: Mapping[str, Any], *, pinned_pub: Optional[str] = None) -> Dict[str, Any]: problems: List[str] = [] body = rec.get("body") if body is None or not isinstance(body, Mapping): if "schema" in rec: body = {k: v for k, v in rec.items() if k not in ("signature", "signer_pub", "digest")} else: return _res(["TRIAL_OUTCOME_BODY_MISSING"]) body = dict(body) expected_dig = _digest(D_ORACLE_RESULT, body) actual_dig = rec.get("digest") or body.get("digest") if actual_dig and actual_dig != expected_dig: problems.append("TRIAL_OUTCOME_DIGEST_MISMATCH") pub = rec.get("signer_pub") or body.get("signer_pub") sig = rec.get("signature") or body.get("signature") or "" if pinned_pub is not None and pub != pinned_pub: problems.append("TRIAL_OUTCOME_SIGNER_NOT_PINNED") if not isinstance(pub, str) or not _verify_sig(pub, sig, D_ORACLE_RESULT, body): problems.append("TRIAL_OUTCOME_SIGNATURE_INVALID") if str(body.get("verdict", "")) not in {"PASS", "FAIL", "VIOLATION", "INCONCLUSIVE"}: problems.append("TRIAL_OUTCOME_INVALID_VERDICT") return _res(problems) def verify_trial_scorecard(rec: Mapping[str, Any], *, pinned_pub: Optional[str] = None) -> Dict[str, Any]: problems: List[str] = [] body = rec.get("body") if body is None or not isinstance(body, Mapping): if "schema" in rec: body = {k: v for k, v in rec.items() if k not in ("signature", "signer_pub", "digest")} else: return _res(["TRIAL_SCORECARD_BODY_MISSING"]) body = dict(body) expected_dig = _digest(D_ASSURANCE_SCORECARD, body) actual_dig = rec.get("digest") or body.get("digest") if actual_dig and actual_dig != expected_dig: problems.append("TRIAL_SCORECARD_DIGEST_MISMATCH") pub = rec.get("signer_pub") or body.get("signer_pub") sig = rec.get("signature") or body.get("signature") or "" if pinned_pub is not None and pub != pinned_pub: problems.append("TRIAL_SCORECARD_SIGNER_NOT_PINNED") if not isinstance(pub, str) or not _verify_sig(pub, sig, D_ASSURANCE_SCORECARD, body): problems.append("TRIAL_SCORECARD_SIGNATURE_INVALID") if str(body.get("actuarial_grade", "")) not in {"A_PRIME", "A", "B", "FAIL"}: problems.append("TRIAL_SCORECARD_INVALID_GRADE") return _res(problems) def verify_trial_bundle(rec: Mapping[str, Any], *, pinned_pub: Optional[str] = None) -> Dict[str, Any]: problems: List[str] = [] body = rec.get("body") if body is None or not isinstance(body, Mapping): if "schema" in rec: body = { "schema": rec.get("schema"), "bundle_id": rec.get("bundle_id"), "tenant_id": rec.get("tenant_id"), "agent_id": rec.get("agent_id"), "scenario_digests": rec.get("scenario_digests") or sorted([s.get("digest") for s in rec.get("scenarios", []) if isinstance(s, dict)]), "trial_digests": rec.get("trial_digests") or sorted([t.get("digest") for t in rec.get("trials", []) if isinstance(t, dict)]), "scorecard_digest": rec.get("scorecard_digest") or (rec.get("scorecard", {}).get("digest") if isinstance(rec.get("scorecard"), dict) else None), "harness_seal_digest": rec.get("harness_seal_digest") or (rec.get("harness_seal", {}).get("digest") if isinstance(rec.get("harness_seal"), dict) else None), "created_at_ns": rec.get("created_at_ns"), } else: return _res(["TRIAL_BUNDLE_BODY_MISSING"]) body = dict(body) expected_dig = _digest(D_TRIAL_BUNDLE, body) actual_dig = rec.get("digest") or body.get("digest") if actual_dig and actual_dig != expected_dig: problems.append("TRIAL_BUNDLE_DIGEST_MISMATCH") pub = rec.get("signer_pub") or body.get("signer_pub") sig = rec.get("signature") or body.get("signature") or "" if pinned_pub is not None and pub != pinned_pub: problems.append("TRIAL_BUNDLE_SIGNER_NOT_PINNED") if not isinstance(pub, str) or not _verify_sig(pub, sig, D_TRIAL_BUNDLE, body): problems.append("TRIAL_BUNDLE_SIGNATURE_INVALID") # verify inner scorecard scorecard = rec.get("scorecard") if isinstance(scorecard, Mapping): problems += verify_trial_scorecard(scorecard, pinned_pub=pinned_pub)["problems"] else: problems.append("TRIAL_BUNDLE_SCORECARD_MISSING") # verify inner scenarios for sc in rec.get("scenarios", []): if isinstance(sc, Mapping): problems += verify_trial_scenario(sc, pinned_pub=pinned_pub)["problems"] return _res(problems) def verify_bundle(bundle: Mapping[str, Any], *, pinned_pub: Optional[str] = None) -> Dict[str, Any]: problems: List[str] = [] root = bundle.get("root") chain = bundle.get("chain") attempted_chain = bundle.get("attempted_chain") decision = bundle.get("decision") span = bundle.get("span") manifest = bundle.get("manifest") attribution = bundle.get("attribution") ledger = bundle.get("ledger") mandate = bundle.get("mandate") node = bundle.get("node") proof_action = bundle.get("proof_action") enforcement_artifact = bundle.get("enforcement_artifact") swarm_message = bundle.get("swarm_message") swarm_capability = bundle.get("swarm_capability") assurance_contract = bundle.get("assurance_contract") bound_pca = bundle.get("bound_pca") effect_contract = bundle.get("effect_contract") reconciliation_record = bundle.get("reconciliation_record") autonomy_transition = bundle.get("autonomy_transition") autonomy_incident = bundle.get("autonomy_incident") if chain is not None: problems += verify_chain(chain, pinned_pub=pinned_pub, root=root)["problems"] if attempted_chain is not None: problems += verify_chain(attempted_chain, pinned_pub=pinned_pub, root=root)["problems"] if decision is not None: problems += verify_decision(decision, pinned_pub=pinned_pub)["problems"] if span is not None: problems += verify_overlay(span)["problems"] if manifest is not None: problems += verify_manifest(manifest, pinned_pub=pinned_pub)["problems"] if attribution is not None: problems += verify_attribution(attribution, pinned_pub=pinned_pub)["problems"] if ledger is not None: problems += verify_ledger(ledger)["problems"] if mandate is not None: problems += verify_mandate(mandate, pinned_pub=pinned_pub)["problems"] if node is not None: problems += verify_node(node, pinned_pub=pinned_pub)["problems"] if proof_action is not None: problems += verify_proof_action(proof_action, pinned_pub=pinned_pub)["problems"] if enforcement_artifact is not None: problems += verify_enforcement_artifact(enforcement_artifact, pinned_pub=pinned_pub)["problems"] if swarm_message is not None: problems += verify_swarm_message(swarm_message, pinned_pub=pinned_pub)["problems"] if swarm_capability is not None: problems += verify_swarm_capability(swarm_capability, pinned_pub=pinned_pub)["problems"] if assurance_contract is not None: problems += verify_runtime_assurance(assurance_contract, pinned_pub=pinned_pub)["problems"] if bound_pca is not None: problems += verify_bound_pca(bound_pca, pinned_pub=pinned_pub)["problems"] if effect_contract is not None: problems += verify_effect_contract(effect_contract, pinned_pub=pinned_pub)["problems"] if reconciliation_record is not None: problems += verify_reconciliation_record(reconciliation_record, pinned_pub=pinned_pub)["problems"] if autonomy_transition is not None: problems += verify_autonomy_transition(autonomy_transition, pinned_pub=pinned_pub)["problems"] if autonomy_incident is not None: problems += verify_autonomy_incident(autonomy_incident, pinned_pub=pinned_pub)["problems"] if all(x is None for x in (chain, attempted_chain, decision, manifest, attribution, ledger, mandate, node, proof_action, enforcement_artifact, swarm_message, swarm_capability, assurance_contract, bound_pca, effect_contract, reconciliation_record, autonomy_transition, autonomy_incident)): problems.append("BUNDLE_MISSING_MESH_EVIDENCE") out = _res(problems) if isinstance(decision, Mapping): out["recomputed_verdict"] = compose((decision.get("body") or {}).get("signals") or []) return out def main(argv: Optional[List[str]] = None) -> int: ap = argparse.ArgumentParser(description="clean-room verifier for the CAIN-42 SUPERIOR AI AGENTIC MESH") ap.add_argument("kind", choices=["root", "chain", "decision", "overlay", "manifest", "attribution", "ledger", "mandate", "node", "proof-action", "enforcement-artifact", "swarm-message", "swarm-capability", "runtime-assurance", "bound-pca", "effect-contract", "reconciliation-record", "autonomy-transition", "autonomy-incident", "trial-scenario", "trial-outcome", "trial-scorecard", "trial-bundle", "bundle"]) ap.add_argument("path") ap.add_argument("--pinned-pub", default=None) ap.add_argument("--root", default=None) ap.add_argument("--fingerprint", default=None) ap.add_argument("--nonce", default=None) ap.add_argument("--action", default=None) ap.add_argument("--tool", default=None) ap.add_argument("--now", type=float, default=None) args = ap.parse_args(argv) rec = json.loads(Path(args.path).read_text()) if args.kind == "root": out = verify_root(rec, pinned_pub=args.pinned_pub) elif args.kind == "chain": root = json.loads(Path(args.root).read_text()) if args.root else None out = verify_chain(rec, pinned_pub=args.pinned_pub, root=root) elif args.kind == "decision": out = verify_decision(rec, pinned_pub=args.pinned_pub, fingerprint=args.fingerprint, nonce=args.nonce) elif args.kind == "overlay": out = verify_overlay(rec) elif args.kind == "manifest": out = verify_manifest(rec, pinned_pub=args.pinned_pub) elif args.kind == "attribution": out = verify_attribution(rec, pinned_pub=args.pinned_pub) elif args.kind == "ledger": out = verify_ledger(rec) elif args.kind == "mandate": out = verify_mandate(rec, pinned_pub=args.pinned_pub, now=args.now) elif args.kind == "node": out = verify_node(rec, pinned_pub=args.pinned_pub) elif args.kind == "proof-action": out = verify_proof_action(rec, pinned_pub=args.pinned_pub, expected_action=args.action, expected_tool=args.tool, now=args.now) elif args.kind == "enforcement-artifact": out = verify_enforcement_artifact(rec, pinned_pub=args.pinned_pub, expected_action=args.action, expected_tool=args.tool, now=args.now) elif args.kind == "swarm-message": out = verify_swarm_message(rec, pinned_pub=args.pinned_pub, now=args.now) elif args.kind == "swarm-capability": out = verify_swarm_capability(rec, pinned_pub=args.pinned_pub, now=args.now) elif args.kind == "runtime-assurance": out = verify_runtime_assurance(rec, pinned_pub=args.pinned_pub, now=args.now) elif args.kind == "bound-pca": out = verify_bound_pca(rec, pinned_pub=args.pinned_pub) elif args.kind == "effect-contract": out = verify_effect_contract(rec, pinned_pub=args.pinned_pub) elif args.kind == "reconciliation-record": out = verify_reconciliation_record(rec, pinned_pub=args.pinned_pub) elif args.kind == "autonomy-transition": out = verify_autonomy_transition(rec, pinned_pub=args.pinned_pub) elif args.kind == "autonomy-incident": out = verify_autonomy_incident(rec, pinned_pub=args.pinned_pub) elif args.kind == "trial-scenario": out = verify_trial_scenario(rec, pinned_pub=args.pinned_pub) elif args.kind == "trial-outcome": out = verify_trial_outcome(rec, pinned_pub=args.pinned_pub) elif args.kind == "trial-scorecard": out = verify_trial_scorecard(rec, pinned_pub=args.pinned_pub) elif args.kind == "trial-bundle": out = verify_trial_bundle(rec, pinned_pub=args.pinned_pub) else: out = verify_bundle(rec, pinned_pub=args.pinned_pub) print(json.dumps({"file": Path(args.path).name, **out}, indent=1)) return 0 if out["verified"] else 1 if __name__ == "__main__": raise SystemExit(main())