#!/usr/bin/env python3 """Clean-room verifier for the CAG-L5 system-governance bundle (Prompt 4, Part 41 Test N). INDEPENDENT IMPLEMENTATION: imports NOTHING from CAIN (only the Python standard library + `cryptography`). It does not take CAIN's decisions on trust. From the signed inputs it recomputes, for every recorded decision: policy precedence, capability/resource authority intersection, tool registration, resource role, model identity, governance-state quorum, blast-radius class and reversibility, and the two-operator step-up rule -- and fails if CAIN ever said ALLOW where any of them forbids it (or disagrees on the policy/blast verdict). It also verifies the manifest signature, the signed hash-linked event chain, every decision signature, every execution commitment, the MCPGate gate log, and the emergency log. NOT recomputed here (stated, not hidden): the trajectory/lease layer (covered by verify_l5_unified.py) and state-input attestations (only their certified digest is checked, through the quorum). python3 verify_system_bundle.py CAIN42_CAG_L5_SYSTEM_BUNDLE.json # exit 0 = VALID """ from __future__ import annotations import base64 import hashlib import json import sys from pathlib import Path from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric import ed25519 D_SYSTEM = "CAIN42/CAG-L5-SYSTEM/v1" D_GOV_STATE = "CAIN42/CAG-L5-GOVERNANCE-STATE/v1" D_DECISION = "CAIN42/CAG-L5-SYSTEM-DECISION/v1" D_COMMIT = "CAIN42/CAG-L5-EXECUTION-COMMITMENT/v1" D_EMERGENCY = "CAIN42/CAG-L5-EMERGENCY/v1" D_SYSEVENT = "CAIN42/CAG-L5-SYSTEM-EVENT/v1" LEVELS = ("CONSTITUTION", "SYSTEM_SAFETY", "ORGANIZATION", "ENVIRONMENT", "RESOURCE", "AGENT", "TRAJECTORY", "ACTION") EFFECTS = ("DENY", "REQUIRE_APPROVAL", "ALLOW") BLAST = ("LOCAL", "SERVICE", "MULTI-SERVICE", "ORGANIZATIONAL", "CROSS-ENVIRONMENT", "SYSTEM-WIDE", "CRITICAL") DEC_KEYS = ("decision", "reasons", "step_up", "blast_radius", "reversibility", "policy", "request_hash", "system_digest", "at", "commitment") COMMIT_KEYS = ("agent_id", "tool_id", "capability", "resource", "parameters_hash", "authority_hash", "policy_version", "trajectory_id", "context_hash", "expected_effect_hash", "nonce") def canon(o) -> bytes: return json.dumps(o, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() def H(o) -> str: return hashlib.sha256(canon(o)).hexdigest() def digest(domain, fields) -> str: return H({"domain": domain, **fields}) def sig_ok(pub, sig, domain, fields) -> 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 within(child, pat) -> bool: return pat == "*" or child == pat or (pat.endswith("*") and child.startswith(pat[:-1])) # ------------------------------------------------------------------ independent re-derivations def policy_effect(rules, agent, cap, res, env, sens): hits = [] for r in rules: if r["level"] not in LEVELS or r["effect"] not in EFFECTS: return "DENY" if r.get("max_sensitivity") is not None and sens <= r["max_sensitivity"]: continue if all(any(within(v, p) for p in r[f]) for v, f in ((agent, "agents"), (cap, "capabilities"), (res, "resources"), (env, "environments"))): hits.append(r["effect"]) for e in EFFECTS: # DENY beats REQUIRE_APPROVAL beats ALLOW, at any level if e in hits: return e return "DENY" def grants_allow(grants, cap, res) -> bool: return bool(grants) and any(g[0] == cap and within(res, g[1]) for g in grants) def blast_class(graph, target, rev) -> str: edges, meta = graph["edges"], graph["meta"] seen, frontier, nodes = {target}, [target], [target] while frontier: nxt = [] for n in frontier: for d in sorted(edges.get(n, [])): if d not in seen: seen.add(d) nodes.append(d) nxt.append(d) frontier = nxt services = [n for n in nodes if meta.get(n, {}).get("kind") == "SERVICE"] envs = {meta.get(n, {}).get("environment", "?") for n in nodes} orgs = {meta.get(n, {}).get("organization", "?") for n in nodes} total = len(meta) if len(orgs) > 1 or (total > 3 and len(nodes) >= total): c = "SYSTEM-WIDE" elif len(envs) > 1: c = "CROSS-ENVIRONMENT" elif len(nodes) >= 10: c = "ORGANIZATIONAL" elif len(services) >= 2: c = "MULTI-SERVICE" elif len(nodes) > 1: c = "SERVICE" else: c = "LOCAL" if rev in ("IRREVERSIBLE", "UNKNOWN") and BLAST.index(c) >= BLAST.index("MULTI-SERVICE"): c = "CRITICAL" return c def quorum_ok(node_keys, votes, local_digest) -> bool: n = len(node_keys) q = 2 * ((n - 1) // 3) + 1 if n else 1 by = {} for v in votes: pub = node_keys.get(v.get("node_id", "")) if pub and sig_ok(pub, v.get("signature", ""), D_GOV_STATE, {"node_id": v["node_id"], "state_digest": v.get("state_digest", "")}): by.setdefault(v["state_digest"], set()).add(v["node_id"]) return len(by) == 1 and len(next(iter(by.values()))) >= q and next(iter(by)) == local_digest def approvals_ok(operators, approvals, request_hash, agent_id) -> int: ok = set() for a in approvals: body = {"operator": a.get("operator"), "request_hash": a.get("request_hash"), "kind": "STEP_UP"} pub = operators.get(a.get("operator", "")) if pub and a.get("request_hash") == request_hash and a.get("operator") != agent_id and \ sig_ok(pub, a.get("signature", ""), D_EMERGENCY, body): ok.add(a["operator"]) return len(ok) def main() -> int: if len(sys.argv) != 2: print("usage: verify_system_bundle.py ", file=sys.stderr) return 2 b = json.loads(Path(sys.argv[1]).read_text()) cain = b.get("cain_public_key", "") checks = [] def check(name, ok, detail=""): checks.append({"check": name, "ok": bool(ok), "detail": str(detail)[:160]}) sysb = b["system"]["body"] check("system.signature", cain in sysb.get("trust_roots", []) and sig_ok(cain, b["system"].get("signature", ""), D_SYSTEM, sysb)) check("system.digest", digest(D_SYSTEM, sysb) == b["system"].get("digest")) check("system.not_sae", "NOT SAE" in b.get("designation", "")) # event chain ev, prev, probs = b.get("events", []), "", [] for i, e in enumerate(ev): body = {k: e.get(k) for k in ("seq", "kind", "payload_hash", "prev", "system_digest")} if e.get("seq") != i or e.get("prev") != prev: probs.append(f"{i}:order") if H(e.get("payload")) != e.get("payload_hash") or digest(D_SYSEVENT, body) != e.get("event_hash"): probs.append(f"{i}:hash") if not sig_ok(cain, e.get("signature", ""), D_SYSEVENT, body): probs.append(f"{i}:sig") prev = e.get("event_hash") check("events.chain", not probs and ev, ",".join(probs[:5])) agents, tools, resources = sysb["agents"], sysb["tools"], sysb["resources"] event_payloads = [e.get("payload") for e in ev] for rec in b.get("records", []): t = rec["test"].split()[0] d, req = rec["decision"], rec["request"] allow = d["decision"] == "ALLOW" check(f"{t}.decision_signature", sig_ok(cain, rec.get("signature", ""), D_DECISION, d)) check(f"{t}.request_bound", H(req) == d["request_hash"] and d["system_digest"] == b["system"]["digest"]) check(f"{t}.in_event_chain", d in event_payloads) check(f"{t}.expected", d["decision"] == rec.get("expected"), d["decision"]) a_id, cap, res = req["agent_id"], req["capability"], req["resource"] agent = agents.get(a_id) resource = resources.get(res) # recompute each governing layer pol = policy_effect(b["policy_rules"], a_id, cap, res, rec["environment"], rec["sensitivity"]) check(f"{t}.policy_recomputed", pol == d["policy"]["effect"], f"{pol} vs {d['policy']['effect']}") rev = (resource or {}).get("reversibility", {}).get(cap, "UNKNOWN") bc = blast_class(b["dependency_graph"], res, rev) check(f"{t}.blast_recomputed", bc == d["blast_radius"] and rev == d["reversibility"], f"{bc}/{rev}") forbid = [] if pol == "DENY": forbid.append("policy") if agent is None or not grants_allow(agent["grants"], cap, res): forbid.append("agent_authority") if not grants_allow(sysb["system_grants"], cap, res): forbid.append("system_authority") if agent and agent.get("parent") and not grants_allow(agents.get(agent["parent"], {}).get("grants"), cap, res): forbid.append("delegator_authority") tool = tools.get(req["tool_id"]) if tool is None or cap not in tool["capabilities"] or not any(within(res, p) for p in tool["resources"]): forbid.append("tool") roles = (resource or {}).get("authority_requirements", {}).get(cap) if resource is None or roles is None or (agent and agent["role"] not in roles): forbid.append("resource_role") if agent and req["model_digest"] != agent["model_digest"]: forbid.append("model") if not quorum_ok(sysb["node_keys"], rec.get("cluster_votes", []), rec.get("local_state_digest")): forbid.append("quorum") needs_step_up = pol == "REQUIRE_APPROVAL" or rev == "IRREVERSIBLE" or bc == "CRITICAL" n_ok = approvals_ok(b.get("operators", {}), rec.get("approvals", []), d["request_hash"], a_id) if needs_step_up and n_ok < 2: forbid.append("step_up") check(f"{t}.allow_is_justified", not (allow and forbid), ",".join(forbid)) check(f"{t}.refusal_is_explained", allow or bool(d["reasons"] or d["step_up"])) # commitment + gate c = rec.get("commitment") if allow: ok = bool(c) and sig_ok(cain, c.get("signature", ""), D_COMMIT, {k: c[k] for k in COMMIT_KEYS}) and \ digest(D_COMMIT, {k: c[k] for k in COMMIT_KEYS}) == c.get("digest") == d["commitment"] and \ (c["agent_id"], c["tool_id"], c["capability"], c["resource"]) == (a_id, req["tool_id"], cap, res) \ and c["parameters_hash"] == H(req["parameters"]) check(f"{t}.commitment", ok) else: check(f"{t}.no_commitment_without_allow", c is None and d["commitment"] == "") if "gate" in rec: check(f"{t}.gate_executes_only_on_allow", rec["gate"].get("executed", False) == allow, rec["gate"]) gb = b.get("gate_bypass_attempts", {}) check("gate.bypass_attempts_blocked", gb and all(v for v in gb.values()), gb) for i, e in enumerate(b.get("emergency_log", [])): if e.get("event") == "ENGAGE": check(f"emergency.{i}.scoped_and_expiring", bool(e.get("scope")) and 0 < e.get("ttl", 0) <= 86400) else: check(f"emergency.{i}.two_operator_recovery", len(set(e.get("operators", []))) >= 2) check("invariants.self_reported", b.get("invariants_self_reported", {}).get("all_hold") is True) failed = [c["check"] for c in checks if not c["ok"]] verdict = "VALID" if not failed else "INVALID" print(json.dumps({"bundle": Path(sys.argv[1]).name, "schema": b.get("schema"), "verdict": verdict, "checks_passed": len(checks) - len(failed), "checks_failed": len(failed), "failed": failed, "not_recomputed": ["trajectory/lease layer (see verify_l5_unified.py)", "state-input attestations (only the certified digest)"], "clean_room": True, "imports_cain": False, "checks": checks}, indent=2)) return 0 if verdict == "VALID" else 1 if __name__ == "__main__": raise SystemExit(main())