"""CAIN-A75-018 offline verifier (rules embedded verbatim). CAIN-A75-018 enterprise-governance rules. Standard library only, no CAIN imports: embedded verbatim in the offline verifier. An enterprise runs a fleet of agents. The governance review recomputes whether the fleet is actually governed: inventory every agent names an owner, a purpose, an environment, its data classes and tools, and has been reviewed within the maximum age controls every agent carries the required controls (identity, logging, data_boundary, kill_switch) and each passes exceptions an exception may waive one control for one agent and one policy only; it must name a known approver, an expiry after the supplied clock, and a non-wildcard scope. An expired, blanket or unknown exception does not count policies a policy that covers a high-risk data class may not be switched off; "monitor" needs a live exception authority a named governance body and a policy revision hash are present claims every boolean the record asserts matches what this review computed Verdict: CONFORMANT, else NON_CONFORMANT with every failing check named. Malformed input is NON_CONFORMANT. It reviews the supplied governance record; it does not inspect a live fleet. """ from __future__ import annotations import math import re _ID = re.compile(r"^[A-Za-z0-9_.:\-]{1,80}$") _HEX = re.compile(r"^[0-9a-f]{64}$") REQUIRED_CONTROLS = ("identity", "logging", "data_boundary", "kill_switch") HIGH_RISK = {"pii", "phi", "financial", "credentials", "children"} 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 evaluate(record): checks, summary = [], {"agents": {}, "exceptions": []} def c(cid, ok, detail=""): checks.append({"id": cid, "ok": bool(ok), "detail": detail}) return ok try: if not isinstance(record, dict) or record.get("schema") != "cain42.a75.enterprise.governance.v1": raise Bad("not a cain42.a75.enterprise.governance.v1 record") agents, policies = record.get("agents"), record.get("policies") if not isinstance(agents, list) or not agents or len(agents) > 5000: raise Bad("agents must be a non-empty list") if not isinstance(policies, list) or not policies: raise Bad("policies must be a non-empty list") auth = record.get("authority") if isinstance(record.get("authority"), dict) else {} now = _num(auth.get("now"), "authority.now") max_age = _num(record.get("max_review_age_days", 90), "max_review_age_days", 0.0, 100000.0) c("authority.body", isinstance(auth.get("governance_body"), str) and bool(auth["governance_body"]), "a governance body is named") c("authority.policy_hash", isinstance(auth.get("policy_revision_sha256"), str) and bool(_HEX.match(auth["policy_revision_sha256"])), "the policy revision is named by hash") A = {} for a in agents: if not isinstance(a, dict) or not isinstance(a.get("agent_id"), str) or not _ID.match(a["agent_id"]): raise Bad("an agent needs an agent_id") if a["agent_id"] in A: raise Bad(f"duplicate agent {a['agent_id']}") if not isinstance(a.get("owner"), str) or not a["owner"]: raise Bad(f"{a['agent_id']}: owner required") if not isinstance(a.get("purpose"), str) or not a["purpose"].strip(): raise Bad(f"{a['agent_id']}: purpose required") if a.get("environment") not in ("prod", "staging", "dev", "edge"): raise Bad(f"{a['agent_id']}: environment must be prod, staging, dev or edge") dc = a.get("data_classes") if not isinstance(dc, list) or not all(isinstance(x, str) for x in dc): raise Bad(f"{a['agent_id']}: data_classes must be strings") tools = a.get("tools") if not isinstance(tools, list) or not all(isinstance(x, str) for x in tools): raise Bad(f"{a['agent_id']}: tools must be strings") _num(a.get("last_review_days_ago"), f"{a['agent_id']}.last_review_days_ago", 0.0, 100000.0) A[a["agent_id"]] = a P = {} for p in policies: if not isinstance(p, dict) or not isinstance(p.get("id"), str) or not _ID.match(p["id"]): raise Bad("a policy needs an id") if p["id"] in P: raise Bad(f"duplicate policy {p['id']}") if p.get("enforcement") not in ("enforce", "monitor", "off"): raise Bad(f"{p['id']}: enforcement must be enforce, monitor or off") if not isinstance(p.get("data_classes"), list): raise Bad(f"{p['id']}: data_classes must be a list") P[p["id"]] = p high = bool(set(p["data_classes"]) & HIGH_RISK) c(f"policy.{p['id']}.not_off", not (high and p["enforcement"] == "off"), "a high-risk policy is not switched off" if high else "policy does not cover a high-risk class") exceptions = record.get("exceptions") if isinstance(record.get("exceptions"), list) else [] live = [] for k, e in enumerate(exceptions): if not isinstance(e, dict): raise Bad("an exception must be an object") aid, pid = e.get("agent_id"), e.get("policy_id") c(f"exception.{k}.agent", aid in A, f"agent {aid!r}") c(f"exception.{k}.policy", pid in P, f"policy {pid!r}") c(f"exception.{k}.approver", isinstance(e.get("approved_by"), str) and bool(e["approved_by"]), "named approver") exp = _num(e.get("expires_at"), f"exception.{k}.expires_at") c(f"exception.{k}.not_expired", exp > now, f"expires {exp} > now {now}") scope = e.get("scope") c(f"exception.{k}.scoped", isinstance(scope, list) and bool(scope) and "*" not in scope, f"scope {scope!r}") if aid in A and pid in P and exp > now and isinstance(scope, list) and "*" not in scope: live.append((aid, pid, set(scope))) summary["exceptions"].append({"agent": aid, "policy": pid, "live": exp > now}) controls = record.get("controls") if isinstance(record.get("controls"), list) else [] got = {} for ct in controls: if not isinstance(ct, dict) or ct.get("agent_id") not in A: raise Bad("a control must name a governed agent") key = (ct["agent_id"], ct.get("control")) if key in got: raise Bad(f"duplicate control {key}") if ct.get("status") not in ("pass", "fail", "missing"): raise Bad(f"{key}: status must be pass, fail or missing") got[key] = ct for aid in sorted(A): for control in REQUIRED_CONTROLS: ct = got.get((aid, control)) waived = any(a == aid and control in scope for a, _p, scope in live) ok = bool(ct) and ct.get("status") == "pass" c(f"{aid}.{control}", ok or waived, "pass" if ok else ("waived by a live exception" if waived else f"{(ct or {}).get('status', 'missing')!r}")) c(f"{aid}.review_current", A[aid]["last_review_days_ago"] <= max_age, f"reviewed {A[aid]['last_review_days_ago']} days ago, max {max_age}") summary["agents"][aid] = {"environment": A[aid]["environment"], "data_classes": A[aid]["data_classes"]} claims = record.get("claims") if isinstance(record.get("claims"), dict) else {} computed = { "all_agents_reviewed": all(a["last_review_days_ago"] <= max_age for a in A.values()), "all_controls_active": all( any(g.get("status") == "pass" for (gid, ctl), g in got.items() if gid == aid and ctl == control) or any(x == aid and control in scope for x, _p, scope in live) for aid in A for control in REQUIRED_CONTROLS), "no_high_risk_policy_off": not any(p["enforcement"] == "off" and set(p["data_classes"]) & HIGH_RISK for p in P.values()), } 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}") summary["computed"] = computed except Bad as e: checks.append({"id": "input.well_formed", "ok": False, "detail": str(e)}) return "NON_CONFORMANT", checks, summary except Exception as e: # noqa: BLE001 checks.append({"id": "input.error", "ok": False, "detail": f"{type(e).__name__}: {e}"[:200]}) return "NON_CONFORMANT", checks, summary return ("CONFORMANT" if all(x["ok"] for x in checks) else "NON_CONFORMANT"), 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.enterprise.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-018, {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))