#!/usr/bin/env python3 """Clean-room verifier for CAIN governance packages (cain.govpkg/1). Zero CAIN imports. It re-derives, from the package's embedded source and its own reading of the spec: * the canonical encoding and the source / IR / package digests * every policy's IR (extends = meet, normalization, the nine redundancy rules, the keep-last-condition rule) * every policy's lowered contract arguments and the risk score * the Ed25519 signature (if present) against an optional pinned key * tenant, expiry and version compatibility It does NOT re-run the optimizer's equivalence corpus (that uses the fabric's own request validator); it reports the compiler's claim for it as SELF_TESTED. Diagnostics are not re-derived (they are covered by the package digest only). usage: govc_verify_offline.py PACKAGE.json [--tenant T] [--now-ms N] [--pubkey B64] [--require-signature] exit 0 = VERIFIED, 1 = UNVERIFIED """ import argparse import base64 import hashlib import json import os import posixpath import sys import unicodedata US = b"\x1f" IR_V, LANG_V, PKG_F, COMPILER = "cain.gov.ir/1", "cain.gov.source/1", "cain.govpkg/1", "cain-govc/1.0.0" TRUST = {"low": 0, "medium": 1, "high": 2} STR_LIST = {"argv0_in", "namespace_in", "source_type_in", "payload_type_in"} POS_INT = {"max_argc", "max_timeout_s", "max_content_bytes", "max_top_k"} KIND_RISK = {"exec": 40, "memory_write": 25, "memory_read": 10, "evidence_register": 5} REQ_MAX = 64 * 1024 class Bad(Exception): pass def _walk(v): if v is None or isinstance(v, bool): return if isinstance(v, int): if not -(2 ** 53) < v < 2 ** 53: raise Bad("integer out of range") return if isinstance(v, str): if unicodedata.normalize("NFC", v) != v or "\x00" in v: raise Bad("string not NFC or has NUL") return if isinstance(v, list): for x in v: _walk(x) return if isinstance(v, dict): for k, x in v.items(): if not (isinstance(k, str) and 1 <= len(k) <= 64 and all(c in "abcdefghijklmnopqrstuvwxyz0123456789_" for c in k)): raise Bad(f"bad key {k!r}") _walk(x) return raise Bad(f"type {type(v).__name__}") def canon(v) -> bytes: _walk(v) return json.dumps(v, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False).encode() def dig(tag: bytes, v) -> str: return hashlib.sha256(tag + US + canon(v)).hexdigest() def under(p, root): root = root.rstrip("/") or "/" return p == root or p.startswith(root + "/") or root == "/" def normp(p): if not isinstance(p, str) or not p.startswith("/") or ".." in p.split("/"): raise Bad(f"bad path {p!r}") return posixpath.normpath(p) # what os.path.normpath is on POSIX (keeps a leading '//', per POSIX) def norm_arg(op, arg): if op in STR_LIST: return sorted(set(arg)) if op == "cwd_under": return sorted({normp(x) for x in arg}) return arg def meet(op, a, b): if op in STR_LIST: r = sorted(set(a) & set(b)) return r or None if op in POS_INT: return min(a, b) if op == "min_trust": return a if TRUST[a] >= TRUST[b] else b if op == "utc_hour_window": lo, hi = max(a[0], b[0]), min(a[1], b[1]) return [lo, hi] if lo < hi else None if op == "cwd_under": k = {x for x in a if any(under(x, y) for y in b)} | {y for y in b if any(under(y, x) for x in a)} k = {x for x in k if not any(x != y and under(x, y) for y in k)} return sorted(k) or None raise Bad(f"unknown op {op}") def effective(src, pid, depth=0): pols = {p["id"]: p for p in src["policies"]} p = pols[pid] if depth > 8: raise Bad("extends too deep") e = effective(src, p["extends"], depth + 1) if p.get("extends") else {"conditions": {}, "lineage": []} e = dict(e, conditions=dict(e["conditions"]), lineage=e["lineage"] + [pid]) for k in ("subject", "product", "kind"): if p.get(k): if e.get(k) and e[k] != p[k]: raise Bad(f"{pid}: {k} conflict") e[k] = p[k] if p.get("resource"): r = normp(p["resource"]) if e.get("product") == "openshell" else p["resource"] if e.get("resource"): ok = under(r, e["resource"]) if e.get("product") == "openshell" else r == e["resource"] if not ok: raise Bad(f"{pid}: resource widens") e["resource"] = r if p.get("purpose"): e["purpose"] = p["purpose"] for k, f in (("max_uses", min), ("ttl_s", min), ("max_risk_score", min), ("scitt_required", lambda a, b: a or b)): if k in p: e[k] = f(e[k], p[k]) if e.get(k) is not None else p[k] if p.get("pin") is not None: h = hashlib.sha256(b"CAIN/contract/v1/request" + US + canon(p["pin"])).hexdigest() if e.get("pin_hash") and e["pin_hash"] != h: raise Bad(f"{pid}: pin conflict") e["pin"], e["pin_hash"] = p["pin"], h for w in p.get("when", []): (op, arg), = w.items() a = norm_arg(op, arg) if op in e["conditions"]: m = meet(op, e["conditions"][op], a) if m is None: raise Bad(f"{pid}: empty meet on {op}") e["conditions"][op] = m else: e["conditions"][op] = a return e def redundant(kind, res, op, arg): return ((op == "cwd_under" and any(under(res, x) for x in arg)) or (op == "namespace_in" and res[len("memvault/ns/"):] in arg) or (op == "payload_type_in" and res[len("scitt/"):] in arg) or (op == "max_timeout_s" and arg >= 60) or (op == "max_top_k" and arg >= 50) or (op == "min_trust" and arg == "low") or (op == "utc_hour_window" and arg == [0, 24]) or (op == "max_argc" and arg >= 64) or (op == "max_content_bytes" and arg >= REQ_MAX)) def optimized(kind, res, conds): out = {k: v for k, v in conds.items() if not redundant(kind, res, k, v)} if conds and not out: k = sorted(conds)[0] out[k] = conds[k] return out def ir_of(src, pid, e, conds): ops = [{"op": "IDENTIFY", "agent_id": e["subject"], "tenant": src["tenant"]}, {"op": "AUTHORITY", "source": "rig", "action": f"{e['product']}.{e['kind']}", "resource": e["resource"]}] ops += [{"op": "CONSTRAIN", "cond": k, "arg": conds[k]} for k in sorted(conds)] if e.get("pin_hash"): ops.append({"op": "PIN", "request_hash": e["pin_hash"]}) ops += [{"op": "LEASE", "max_uses": e["max_uses"], "ttl_ms": e["ttl_s"] * 1000}, {"op": "RISK_CEILING", "max_score": e["max_risk_score"]}, {"op": "REQUIRE_EVIDENCE", "scitt": e["scitt_required"]}, {"op": "COMMIT", "boundary": "E8"}] return {"ir": IR_V, "policy": pid, "namespace": src["namespace"], "purpose": e["purpose"], "ops": ops, "lineage": e["lineage"]} def risk(kind, pinned, uses, ttl_ms, n, scitt): s = KIND_RISK.get(kind, 50) + (0 if pinned else 20) + min(uses, 10) + (10 if ttl_ms > 15 * 60_000 else 0) s += 10 if n == 0 else 0 s += 0 if scitt else 10 return min(s, 100) def sig_ok(pub_b64, msg, sig_b64): try: from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey Ed25519PublicKey.from_public_bytes(base64.b64decode(pub_b64, validate=True)).verify( base64.b64decode(sig_b64, validate=True), msg) return True except Exception: # noqa: BLE001 return False def verify(pkg, tenant=None, now_ms=None, pubkey=None, require_signature=False): checks = [] def mark(name, ok, why=""): checks.append({"check": name, "ok": bool(ok), "why": why}) try: m, src = pkg["manifest"], pkg["source"] mark("format", pkg.get("format") == PKG_F) mark("versions", m.get("ir_version") == IR_V and m.get("language") == LANG_V and m.get("compiler") == COMPILER, f"{m.get('language')} {m.get('ir_version')} {m.get('compiler')}") mark("source_digest", dig(b"CAIN/gov/v1/source", src) == pkg["source_digest"]) body = {k: v for k, v in pkg.items() if k not in ("package_digest", "signature")} mark("package_digest", dig(b"CAIN/gov/v1/package", body) == pkg["package_digest"]) mark("manifest_matches_source", (m.get("tenant"), m.get("namespace"), m.get("valid_until_ms")) == (src.get("tenant"), src.get("namespace"), src.get("valid_until_ms"))) concrete = sorted(p["id"] for p in src["policies"] if not p.get("abstract")) mark("policy_set", sorted(pkg["policies"]) == concrete, f"{sorted(pkg['policies'])} vs {concrete}") for pid in concrete: e = effective(src, pid) e.setdefault("max_uses", 1) e["max_uses"] = 1 if e["max_uses"] is None else e["max_uses"] e["ttl_s"] = e.get("ttl_s") or 300 e["max_risk_score"] = 100 if e.get("max_risk_score") is None else e["max_risk_score"] e["scitt_required"] = True if e.get("scitt_required") is None else e["scitt_required"] opt = optimized(e["kind"], e["resource"], e["conditions"]) got = pkg["policies"].get(pid, {}) un_ir = ir_of(src, pid, e, e["conditions"]) # the compiler may keep the unoptimized IR (optimization rejected by its differential check): accept # exactly the two IRs this spec allows, nothing else cands = {dig(b"CAIN/gov/v1/ir", ir_of(src, pid, e, opt)): opt, dig(b"CAIN/gov/v1/ir", un_ir): e["conditions"]} mark(f"{pid}:ir_rederived", got.get("ir_digest") in cands and dig(b"CAIN/gov/v1/ir", got.get("ir")) == got.get("ir_digest")) mark(f"{pid}:ir_unoptimized", got.get("ir_unoptimized_digest") == dig(b"CAIN/gov/v1/ir", un_ir)) conds = cands.get(got.get("ir_digest"), opt) r = risk(e["kind"], bool(e.get("pin_hash")), e["max_uses"], e["ttl_s"] * 1000, len(conds), e["scitt_required"]) mark(f"{pid}:risk", got.get("risk_score") == r and r <= e["max_risk_score"], f"{got.get('risk_score')} vs {r}") lw = got.get("lowered", {}) want = {"agent_id": e["subject"], "product": e["product"], "kind": e["kind"], "resource": e["resource"], "purpose": e["purpose"], "max_uses": e["max_uses"], "ttl_ms": e["ttl_s"] * 1000, "conditions": [{"op": k, "arg": conds[k]} for k in sorted(conds)], "max_risk_score": e["max_risk_score"], "scitt_required": e["scitt_required"], "annotations": {"gov_ir": got.get("ir_digest"), "gov_source": pkg["source_digest"], "gov_policy": pid}} if e.get("pin_hash"): want["request"] = e["pin"] mark(f"{pid}:lowered", lw == want) mark(f"{pid}:authority_source", all(p.get("authority", {"from": "rig"}) == {"from": "rig"} for p in src["policies"] if p["id"] in e["lineage"])) eq = got.get("equivalence", {}).get("status") checks.append({"check": f"{pid}:optimizer_equivalence", "ok": True, "why": f"SELF_TESTED by the compiler ({eq}); not re-run here"}) if tenant is not None: mark("tenant", m.get("tenant") == tenant) if now_ms is not None: mark("not_expired", now_ms < m.get("valid_until_ms", 0)) s = pkg.get("signature") if s is None: mark("signature", not require_signature, "unsigned") else: okk = sig_ok(s.get("public_key_b64", ""), b"CAIN/gov/v1/package-signature" + US + str(pkg["package_digest"]).encode(), s.get("sig_b64", "")) mark("signature", okk and (pubkey is None or s.get("public_key_b64") == pubkey), "pinned key mismatch" if okk and pubkey and s.get("public_key_b64") != pubkey else "") except (Bad, KeyError, TypeError, ValueError, AttributeError) as exc: mark("structure", False, f"{type(exc).__name__}: {exc}") ok = all(c["ok"] for c in checks) return {"verifier": "govc_verify_offline/1", "status": "VERIFIED" if ok else "UNVERIFIED", "ok": ok, "checks": checks} def main(argv=None): ap = argparse.ArgumentParser() ap.add_argument("package") ap.add_argument("--tenant") ap.add_argument("--now-ms", type=int) ap.add_argument("--pubkey") ap.add_argument("--require-signature", action="store_true") a = ap.parse_args(argv) with open(a.package, encoding="utf-8") as f: pkg = json.load(f) r = verify(pkg, a.tenant, a.now_ms, a.pubkey, a.require_signature) print(json.dumps(r, indent=1)) return 0 if r["ok"] else 1 if __name__ == "__main__": sys.exit(main())