#!/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", []) v2 = b.get("schema") == "cain42.l5.unified_bundle.v2" # 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") signer = pub if (v2 and e.get("event_type") in GOVERNANCE_EVENTS) else agent_pub if not sig_ok(signer, e.get("signature", ""), D_EVENT, body): chain_problems.append(f"{i}:sig({'governance' if signer == pub else 'agent'})") 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") if v2: # Recompute the continuous-authorization decision from the signed inputs; do not trust the runtime's. max_ttl = float(b.get("limits", {}).get("max_lease_ttl_seconds", 3600)) ctx = dict(pub=pub, agent_pub=agent_pub, traj=traj, obj=obj, binding=b.get("trajectory_plan_binding", {}), authorized_context=b.get("trajectory_authorized_context", {}), max_ttl=max_ttl, events=events) why = recompute(b["lease"], b.get("trajectory_request", {}), **ctx) check("trajectory.decision_recomputed_allow", not why, ",".join(why)) check("trajectory.binding_evidenced", any( e.get("event_type") == "REAUTHORIZATION" and e.get("authority_hash") == h_obj(ctx["binding"]) for e in events), "governance-signed REAUTHORIZATION event carries the plan binding digest") adv = b.get("adversarial_requests", []) check("adversarial.present", len(adv) >= 5, str(len(adv))) for a in adv: why = recompute(a.get("lease", {}), a.get("request", {}), **ctx) check(f"adversarial.{a.get('case')}.recomputed_not_allow", bool(why), ",".join(why[:3])) check(f"adversarial.{a.get('case')}.runtime_agrees", a.get("runtime_decision") != "ALLOW", a.get("runtime_decision", "")) check("trajectory.decision_evidenced", bool(b.get("trajectory_decision", {}).get("evidence_digest"))) # invariants inv = b.get("invariants", {}) # SELF-REPORTED by the runtime (this verifier cannot re-execute them); a reported failure still invalidates. check("invariants.l5_safety.self_reported", inv.get("l5_safety", {}).get("all_hold") is True) check("invariants.authority.self_reported", inv.get("authority", {}).get("all_hold") is True) check("invariants.trajectory.self_reported", 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") GOVERNANCE_EVENTS = {"DECISION", "REAUTHORIZATION", "CONTAINMENT", "AUTHORITY", "REVOCATION"} BINDING_KEYS = ("plan_hash", "intent_hash", "action_hash", "parameters_hash", "resource_hash", "context_hash") MATERIAL_KEYS = ("objective_hash", "plan_hash", "resource_hash", "policy_hash", "authority_hash", "trust_state_hash", "risk_state_hash", "environment_state_hash", "credentials_hash", "delegation_depth", "context_hash") def h_obj(o) -> str: return hashlib.sha256(canon(o)).hexdigest() def recompute(lease_d, req, *, pub, agent_pub, traj, obj, binding, authorized_context, max_ttl, events): """Clean-room re-derivation of 'may this action execute now?'. Returns the reasons it may NOT ([] = may). Written from the published rules, not from CAIN's code: every bound must be present and satisfied.""" why = [] lease = body_of(lease_d, LEASE_FIELDS) if not sig_ok(pub, lease_d.get("signature", ""), D_LEASE, lease): why.append("lease_not_signed_by_governance") if agent_pub and sig_ok(agent_pub, lease_d.get("signature", ""), D_LEASE, lease): why.append("self_authorization") if lease.get("agent_id") != traj.get("agent_id") or not lease.get("agent_id"): why.append("lease_agent") if lease.get("trajectory_id") != traj.get("trajectory_id") or not lease.get("trajectory_id"): why.append("lease_trajectory") if lease.get("revocation_state") != "ACTIVE": why.append("revoked") iss, exp, now = float(lease.get("issued_at") or 0), float(lease.get("expires_at") or 0), float(req.get("now", 0)) if exp <= iss or exp - iss > max_ttl: why.append("lease_bounds") if not iss <= now <= exp: why.append("lease_time") if not lease.get("policy_version") or req.get("policy_version") != lease.get("policy_version"): why.append("policy") if not lease.get("context_hash") or req.get("context_hash") != lease.get("context_hash"): why.append("context") if float(req.get("trust", 0)) < float(lease.get("trust_floor", 1)): why.append("trust") if _band_index(traj.get("risk", {}).get("band", "CRITICAL")) > _band_index(lease.get("risk_ceiling", "")): why.append("risk") acts, ress = lease.get("action_scope") or [], lease.get("resource_scope") or [] if req.get("action") not in acts: why.append("action_scope") if not any(within(req.get("resource", ""), p) for p in ress): why.append("resource_scope") scope = obj.get("authorized_scope") or [] if scope and not any(within(req.get("resource", ""), p) for p in scope): why.append("objective_scope") if req.get("outcome") and req.get("outcome") in (obj.get("forbidden_outcomes") or []): why.append("forbidden_outcome") sub = req.get("binding") or {} if not binding or any(sub.get(k) != binding.get(k) for k in BINDING_KEYS): why.append("plan_binding") if binding.get("authorization_id") != lease_d.get("lease_id"): why.append("binding_lease") cur = req.get("current_context") if not isinstance(cur, dict): why.append("current_context_missing") elif not authorized_context: why.append("authorized_context_missing") elif any(cur.get(k) != authorized_context.get(k) for k in MATERIAL_KEYS): why.append("material_change") if traj.get("state") not in ("ACTIVE", "LIMITED"): why.append("state") return why def _band_index(band: str) -> int: return _BANDS.index(band) if band in _BANDS else len(_BANDS) - 1 if __name__ == "__main__": raise SystemExit(main())