#!/usr/bin/env python3 """CAIN-42 L5 unified clean-room verifier (Prompt 2 + Prompt 3). INDEPENDENT IMPLEMENTATION: imports NO CAIN code. A different design from verify_l5_bundle.py and verify_authority_bundle.py — it is table-driven and verifies the whole bundle in one pass: identity, authority grant, delegation, authorization decision, objective envelope, the hash-chained trajectory, its Merkle root, the authorization lease, the trajectory decision, and the three invariant matrices. python3 scripts/cain42_l5/verify_l5_unified.py CAIN42_L5_UNIFIED_BUNDLE.json Verdict: VALID / INVALID / INCOMPLETE. """ 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_IDENTITY = "CAIN42/L5-AGENT-IDENTITY/v1" D_GRANT = "CAIN42/L5-AUTHORITY-GRANT/v1" D_DELEGATION = "CAIN42/L5-DELEGATION-GRANT/v1" D_DECISION = "CAIN42/L5-AUTHORIZATION-DECISION/v1" D_OBJECTIVE = "CAIN42/L5-OBJECTIVE-ENVELOPE/v1" D_EVENT = "CAIN42/L5-TRAJECTORY-EVENT/v1" D_LEASE = "CAIN42/L5-AUTHORIZATION-LEASE/v1" IDENTITY_FIELDS = ("identity_version", "agent_id", "agent_type", "model_family", "model_identifier", "model_version", "runtime_identifier", "runtime_version", "owner_principal", "organization", "created_at", "expires_at", "status", "capabilities", "public_key", "key_algorithm", "attestation_reference", "policy_binding", "trust_reference", "delegation_root", "issuer", "issued_at") GRANT_FIELDS = ("grant_id", "principal_id", "agent_id", "issuer_id", "parent_grant_id", "scope", "capabilities", "resources", "operations", "constraints", "risk_limit", "budget_limit", "delegation_limit", "time_start", "time_expiry", "max_duration", "renewal_policy", "policy_version", "trust_floor", "environment_constraints", "revocation_reference", "single_use", "status") DELEGATION_FIELDS = ("delegation_id", "delegator", "delegate", "parent_grant_id", "delegated_capabilities", "delegated_resources", "constraints", "depth", "maximum_depth", "time_start", "time_expiry", "policy_version") DECISION_FIELDS = ("decision_id", "action_id", "agent_id", "requested_capability", "resource", "operation", "effective_authority", "policy_version", "trust_state", "delegation_chain", "risk_state", "decision", "reason_codes", "constraints", "timestamp", "expiry", "action_hash", "parameters_hash", "resource_hash", "context_hash", "policy_hash", "evidence_reference") OBJECTIVE_FIELDS = ("objective_id", "owner", "authorized_scope", "allowed_outcomes", "forbidden_outcomes", "resource_constraints", "risk_constraints", "time_constraints", "policy_binding", "authority_binding", "trajectory_binding") EVENT_FIELDS = ("event_id", "trajectory_id", "agent_id", "parent_event_id", "event_type", "timestamp", "logical_time", "sequence_number", "objective_hash", "intent_hash", "plan_hash", "action_hash", "parameters_hash", "resource_hash", "context_hash", "policy_hash", "authority_hash", "trust_state_hash", "risk_state_hash", "environment_state_hash", "observation_hash", "previous_event_hash", "node_attestations", "evidence_reference") LEASE_FIELDS = ("lease_id", "agent_id", "trajectory_id", "action_scope", "resource_scope", "authority_scope", "policy_version", "trust_floor", "risk_ceiling", "context_hash", "issued_at", "expires_at", "renewal_rules", "revocation_state") def canon(o) -> bytes: return json.dumps(o, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() def digest(domain, fields) -> str: return hashlib.sha256(canon({"domain": domain, **fields})).hexdigest() 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 body_of(obj, fields) -> dict: return {n: (sorted(obj.get(n)) if isinstance(obj.get(n), list) else obj.get(n)) for n in fields} def within(child, parent) -> bool: return child == parent or (parent.endswith("*") and child.startswith(parent[:-1])) def set_within(a, b) -> bool: return all(any(within(x, y) for y in b) for x in a) def merkle_root(event_hashes): if not event_hashes: return hashlib.sha256(b"").hexdigest() layer = [bytes.fromhex(h) for h in event_hashes] while len(layer) > 1: nxt = [] for i in range(0, len(layer), 2): nxt.append(hashlib.sha256(b"\x01" + layer[i] + layer[i + 1]).digest() if i + 1 < len(layer) else layer[i]) layer = nxt return layer[0].hex() def main() -> int: if len(sys.argv) != 2: print("usage: verify_l5_unified.py ", file=sys.stderr) return 2 b = json.loads(Path(sys.argv[1]).read_text()) pub = b.get("issuer_public_key", "") agent_pub = b.get("agent_public_key", "") checks = [] def check(name, ok, detail=""): checks.append({"check": name, "ok": bool(ok), "detail": detail}) ident = body_of(b.get("identity", {}), IDENTITY_FIELDS) grant = body_of(b.get("grant", {}), GRANT_FIELDS) deleg = body_of(b.get("delegation", {}), DELEGATION_FIELDS) dec = body_of(b.get("authorization_decision", {}), DECISION_FIELDS) obj = body_of(b.get("objective", {}), OBJECTIVE_FIELDS) lease = body_of(b.get("lease", {}), LEASE_FIELDS) traj = b.get("trajectory", {}) events = traj.get("events", []) # identity / grant / delegation / decision check("identity.signature", sig_ok(pub, b["identity"].get("identity_signature", ""), D_IDENTITY, ident)) check("identity.digest", digest(D_IDENTITY, ident) == b["identity"].get("digest")) check("grant.signature", sig_ok(pub, b["grant"].get("signature", ""), D_GRANT, grant)) check("grant.digest", digest(D_GRANT, grant) == b["grant"].get("digest")) check("delegation.signature", sig_ok(pub, b["delegation"].get("signature", ""), D_DELEGATION, deleg)) check("delegation.cannot_escalate", set(deleg.get("delegated_capabilities", [])) <= set(grant.get("capabilities", [])) and set_within(deleg.get("delegated_resources", []), grant.get("resources", []))) check("decision.signature", sig_ok(pub, b["authorization_decision"].get("signature", ""), D_DECISION, dec)) check("decision.digest", digest(D_DECISION, dec) == b["authorization_decision"].get("digest")) check("decision.is_allow", dec.get("decision") == "ALLOW") eff = dec.get("effective_authority", {}) or {} expected_caps = sorted(set(grant.get("capabilities", [])) & set(deleg.get("delegated_capabilities", []))) check("decision.effective_intersection", sorted(eff.get("capabilities", [])) == expected_caps, f"{eff.get('capabilities')} vs {expected_caps}") # objective check("objective.signature", sig_ok(pub, b["objective"].get("signature", ""), D_OBJECTIVE, obj)) check("objective.digest", digest(D_OBJECTIVE, obj) == b["objective"].get("objective_hash")) # trajectory chain chain_problems = [] prev = "" for i, e in enumerate(events): body = body_of(e, EVENT_FIELDS) if digest(D_EVENT, body) != e.get("event_hash"): chain_problems.append(f"{i}:hash") if not sig_ok(agent_pub, e.get("signature", ""), D_EVENT, body): chain_problems.append(f"{i}:sig") if e.get("previous_event_hash") != prev: chain_problems.append(f"{i}:link") if e.get("sequence_number") != i: chain_problems.append(f"{i}:seq") prev = e.get("event_hash") check("trajectory.chain", not chain_problems, ",".join(chain_problems[:4])) check("trajectory.merkle_root", merkle_root([e.get("event_hash", "") for e in events]) == traj.get("merkle_root")) check("trajectory.objective_bound", all(e.get("objective_hash") == b["objective"].get("objective_hash") for e in events)) # lease check("lease.signature", sig_ok(pub, b["lease"].get("signature", ""), D_LEASE, lease)) check("lease.risk_ceiling_respected", _band_index(traj.get("risk", {}).get("band", "LOW")) <= _band_index(lease.get("risk_ceiling", "GUARDED"))) check("trajectory.decision_allow", b.get("trajectory_decision", {}).get("decision") == "ALLOW") check("trajectory.decision_evidenced", bool(b.get("trajectory_decision", {}).get("evidence_digest"))) # invariants inv = b.get("invariants", {}) check("invariants.l5_safety", inv.get("l5_safety", {}).get("all_hold") is True) check("invariants.authority", inv.get("authority", {}).get("all_hold") is True) check("invariants.trajectory", inv.get("trajectory", {}).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": sum(1 for c in checks if c["ok"]), "checks_failed": len(failed), "failed": failed, "checks": checks, "clean_room": True, "imports_cain": False}, indent=2)) return 0 if verdict == "VALID" else 1 _BANDS = ("LOW", "GUARDED", "ELEVATED", "HIGH", "CRITICAL") def _band_index(band: str) -> int: return _BANDS.index(band) if band in _BANDS else len(_BANDS) - 1 if __name__ == "__main__": raise SystemExit(main())