"""CAIN-A75-037 offline verifier (rules embedded verbatim). CAIN-A75-037 execution-governance rules. Standard library only, no CAIN imports: embedded verbatim in the offline verifier. An enterprise agent executed an action through a governance boundary. The review recomputes whether the execution was authorized and evidenced: decision a signed-looking policy decision (revision hash) exists; DENY and UNKNOWN are never executed approval REQUIRE_APPROVAL and high-risk executions need approval before the start, by someone other than the agent enforcement the boundary was applied and its depth is stated; an application-level claim cannot masquerade as OS-level execution the run starts after the decision and approval, and a completed run carries an effect hash audit the audit record links to the previous hash and its own chain hash covers the record claims every boolean the record asserts matches what this review computed Verdict: COMPLIANT, else NON_COMPLIANT with every failing check named. Malformed input is NON_COMPLIANT. It reviews the execution record; it does not re-run the action. """ from __future__ import annotations import hashlib import json import math import re _ID = re.compile(r"^[A-Za-z0-9_.:\-]{1,80}$") _HEX = re.compile(r"^[0-9a-f]{64}$") DECISIONS = {"ALLOW", "DENY", "REQUIRE_APPROVAL", "UNKNOWN"} DEPTHS = {"app", "runtime", "os", "hardware"} class Bad(ValueError): pass def _num(v, what, lo=0.0, hi=1e12): if isinstance(v, bool) or not isinstance(v, (int, float)) or not math.isfinite(v) or not (lo <= v <= hi): raise Bad(f"{what} must be a finite number in [{lo}, {hi}]") return float(v) def chain_hash(audit): body = {k: v for k, v in audit.items() if k != "chain_hash"} return hashlib.sha256(json.dumps(body, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode()).hexdigest() def evaluate(record): checks, summary = [], {} def c(cid, ok, detail=""): checks.append({"id": cid, "ok": bool(ok), "detail": detail}) try: if not isinstance(record, dict) or record.get("schema") != "cain42.a75.execution.gov.v1": raise Bad("not a cain42.a75.execution.gov.v1 record") req, pol, enf, exe, audit = record.get("request"), record.get("policy"), record.get("enforcement"), record.get("execution"), record.get("audit") if not all(isinstance(x, dict) for x in (req, pol, enf, exe, audit)): raise Bad("request, policy, enforcement, execution and audit must be objects") agent = req.get("agent") c("request.agent", isinstance(agent, str) and bool(agent), "an agent is named") c("request.tenant", isinstance(req.get("tenant"), str) and bool(req["tenant"]), "a tenant is named") c("request.params_hash", isinstance(req.get("params_sha256"), str) and bool(_HEX.match(req["params_sha256"])), "the exact parameters are hashed") decision = pol.get("decision") c("policy.decision", decision in DECISIONS, f"decision {decision!r}") c("policy.revision_hash", isinstance(pol.get("revision_sha256"), str) and bool(_HEX.match(pol["revision_sha256"])), "the policy revision is hashed") risk = _num(pol.get("risk_score"), "policy.risk_score", 0.0, 100.0) decided = _num(pol.get("decided_at"), "policy.decided_at") started = _num(exe.get("started_at"), "execution.started_at") c("execution.after_decision", started >= decided - 1e-9, "execution starts after the decision") executed = exe.get("status") in ("completed", "failed") c("decision.executed", executed and decision == "ALLOW", f"executed={executed}, decision={decision!r}") high = risk >= 70.0 if decision == "REQUIRE_APPROVAL" or high: ap = record.get("approval") if isinstance(record.get("approval"), dict) else {} at = _num(ap.get("at", 0.0), "approval.at") c("approval.present", bool(ap), "approval present for a gated execution") c("approval.independent", ap.get("by") not in (None, "", agent), f"approved by {ap.get('by')!r}") c("approval.before_execution", at <= started + 1e-9, "approval predates execution") c("approval.role", isinstance(ap.get("role"), str) and bool(ap["role"]), f"role {ap.get('role')!r}") else: c("approval.not_required", True, "not a gated or high-risk execution") c("enforcement.applied", enf.get("applied") is True, "the boundary applied the decision") depth = enf.get("enforcement_depth") c("enforcement.depth", depth in DEPTHS, f"depth {depth!r}") c("enforcement.boundary", enf.get("boundary") in ("mcp", "api", "db", "shell", "network"), f"boundary {enf.get('boundary')!r}") if executed: c("execution.effect_hash", isinstance(exe.get("effect_sha256"), str) and bool(_HEX.match(exe["effect_sha256"])), "a completed run carries an effect hash") finished = _num(exe.get("finished_at"), "execution.finished_at") c("execution.ordered", finished >= started, "the run finishes after it starts") summary["status"] = exe.get("status") prev_ok = isinstance(audit.get("prev_record_hash"), str) and bool(_HEX.match(audit["prev_record_hash"])) c("audit.prev_link", prev_ok, "the audit links to a previous record") c("audit.chain_hash", audit.get("chain_hash") == chain_hash(audit), "the chain hash covers this audit record") claims = record.get("claims") if isinstance(record.get("claims"), dict) else {} computed = { "policy_applied": decision == "ALLOW" and executed, "enforced_at_boundary": enf.get("applied") is True, "evidence_chained": prev_ok and audit.get("chain_hash") == chain_hash(audit), } for key in sorted(claims): c(f"claims.{key}", key in computed and claims[key] == computed[key], f"claimed {claims[key]!r}, computed {computed.get(key)!r}") if claims.get("os_level_enforcement") is True: c("claims.os_level_enforcement", depth == "os", f"claimed OS-level enforcement, depth is {depth!r}") summary["computed"] = computed summary["decision"] = decision summary["risk_score"] = risk summary["enforcement_depth"] = depth except Bad as e: checks.append({"id": "input.well_formed", "ok": False, "detail": str(e)}) return "NON_COMPLIANT", checks, summary except Exception as e: # noqa: BLE001 checks.append({"id": "input.error", "ok": False, "detail": f"{type(e).__name__}: {e}"[:200]}) return "NON_COMPLIANT", checks, summary return ("COMPLIANT" if all(x["ok"] for x in checks) else "NON_COMPLIANT"), checks, summary # ---------------------------------------------------------------------------- offline verifier (no CAIN imports) # Every name below is prefixed _kit_verifier_ / _b64 / _hl / _json / _sys so it cannot shadow a name in the rules above. import base64 as _b64, hashlib as _hl, json as _json, sys as _sys _KIT_VERIFIER_GENESIS = _hl.sha256(b"cain42.a75.executiongov.genesis").hexdigest() def _kit_verifier_canon(o): return _json.dumps(o, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() def _kit_verifier_sha(o): return _hl.sha256(_kit_verifier_canon(o)).hexdigest() def _kit_verifier_main(argv): from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey if len(argv) < 2: print("usage: verify_offline.py run.json --pinned-key BASE64"); return 2 run = _json.load(open(argv[1])) pin = argv[argv.index("--pinned-key") + 1] if "--pinned-key" in argv else None problems, prev, recs = [], _KIT_VERIFIER_GENESIS, run.get("records") or [] for i, r in enumerate(recs): body = {k: v for k, v in r.items() if k not in ("record_hash", "signature_b64")} if _kit_verifier_sha(body) != r.get("record_hash"): problems.append(f"record {i}: record_hash does not match") if body.get("subject_sha256") != _kit_verifier_sha(body.get("subject")): problems.append(f"record {i}: subject_sha256 does not match") if pin and r.get("public_key_b64") != pin: problems.append(f"record {i}: different key than the pinned one") try: Ed25519PublicKey.from_public_bytes(_b64.b64decode(pin or r.get("public_key_b64"))).verify( _b64.b64decode(r.get("signature_b64", "")), str(r.get("record_hash")).encode()) except Exception: problems.append(f"record {i}: signature does not verify") if r.get("prev_record_hash") != prev: problems.append(f"record {i}: chain link broken") v, checks, _ = evaluate(r.get("subject")) if v != r.get("verdict") or checks != r.get("checks"): problems.append(f"record {i}: re-running the rules gives {v}, the record says {r.get('verdict')}") prev = r.get("record_hash") if run.get("run_hash") != _kit_verifier_sha([r.get("record_hash") for r in recs]): problems.append("run_hash does not match") for p in problems: print(" -", p) if not pin: print("WARNING: no --pinned-key given; the signature was checked against the key inside the file") print("VERDICT:", "VERIFIED" if not problems else "TAMPERED OR INVALID", f"(CAIN-A75-037, {len(recs)} record(s), verdicts {run.get('verdicts')})") return 0 if not problems else 1 if __name__ == "__main__": _sys.exit(_kit_verifier_main(_sys.argv))