#!/usr/bin/env python3 """AGENT-CI KEYSTONE clean-room verifier: re-derive a release-gate verdict from its input, with no CAIN code. This file is a second, independent implementation of the gate rules written down in keystone/agentci.py's docstring. It is used twice: 1. as the PROVER: the service runs `derive(suite)` here, in a sandbox, and signs a record as PROVED only when this file and the engine reach exactly the same verdict, findings and per-case metrics; 2. by an auditor, offline: pip install cryptography python3 agentci_verify.py record*.json --pinned-pub # key fetched from the issuer yourself python3 agentci_verify.py record*.json --pinned-fingerprint SHA256: For each record it checks the record hash, the Ed25519 signature against the pinned key, the chain link to the previous record, the input/derivation digests, the DSSE envelope, and then re-derives the verdict from the embedded input. Exit code 0 only when every record passes every check. Only the standard library is needed to re-derive; `cryptography` is imported only to check signatures. """ from __future__ import annotations import base64 import glob import hashlib import json import math import re import sys import unicodedata DEFAULT_MAX_SCORE_DROP = 0.10 DEFAULT_MAX_COST_PCT = 25 TEXT_LIMIT = 20_000 UNDO_WORDS = frozenset(["undo", "revert", "rollback", "delete", "cancel", "unset", "restore"]) EPS = 1e-9 # ------------------------------------------------------------------------------------------------ outcome def _as_text_and_json(output): if isinstance(output, str): try: doc = json.loads(output) except ValueError: doc = None return output[:TEXT_LIMIT], doc return json.dumps(output, sort_keys=True, separators=(",", ":"), ensure_ascii=False)[:TEXT_LIMIT], output def outcome_ok(checker, run): if run.get("error"): return False text, doc = _as_text_and_json(run.get("output", "")) kind, want = checker["type"], checker["expected"] if kind == "exact": return text.strip() == str(want).strip() if kind == "contains": return str(want) in text if kind == "regex": return bool(re.search(want, text)) if kind == "json_subset": if not isinstance(doc, dict): return False for key in want: if key not in doc or doc[key] != want[key]: return False return True raise ValueError("unknown checker type %r" % kind) # ------------------------------------------------------------------------------------------------ path LOOKALIKES = {"\u0430": "a", "\u0435": "e", "\u043e": "o", "\u0440": "p", "\u0441": "c", "\u0443": "y", "\u0445": "x", "\u0456": "i", "\u0458": "j", "\u0455": "s", "\u04bb": "h", "\u0501": "d", "\u043a": "k", "\u043c": "m", "\u0442": "t", "\u0432": "b", "\u043d": "h", "\u0433": "r", "\u04cf": "l", "\u03b1": "a", "\u03bf": "o", "\u03c1": "p", "\u03b5": "e", "\u03b9": "i", "\u03ba": "k", "\u03bd": "v", "\u03c4": "t", "\u03c5": "u", "\u03c7": "x", "\u03f2": "c"} SEPARATORS = "._-/:\u2010\u2011\u2012\u2013\u2014\u2212" def _key(name): folded = unicodedata.normalize("NFKC", str(name)).casefold() out = [] for c in folded: c = LOOKALIKES.get(c, c) if c.isspace() or unicodedata.category(c) == "Cf" or c in SEPARATORS: continue out.append(c) return "".join(out) def _aj(v): return json.dumps({} if v is None else v, sort_keys=True, separators=(",", ":"), default=str) def _words(name): return set(w.lower() for w in re.findall(r"[A-Za-z][a-z]*|[0-9]+", name.replace("_", " "))) def path_metrics(trace, terminal, baseline_high=()): ordered = sorted(trace, key=lambda st: st["index"]) index = dict((st["index"], st) for st in trace) calls = {} for st in ordered: sig = st["tool"] + "\x00" + json.dumps(st.get("args", {}), sort_keys=True, separators=(",", ":"), default=str) calls.setdefault(sig, []).append(st["index"]) repeats = set() extra = 0 for positions in calls.values(): extra += len(positions) - 1 repeats.update(positions[1:]) total = math.fsum([float(st.get("cost", 0)) for st in trace]) off_path = set() if terminal is not None: keep, stack = set(), [terminal] while stack: node = stack.pop() if node not in keep: keep.add(node) stack.extend(index[node].get("depends_on", [])) off_path = set(i for i in index if i not in keep) corrections = [] groups = {} for st in ordered: if st.get("resource"): groups.setdefault(st["resource"], []).append(st) for res in groups: seq = groups[res] for pos in range(len(seq)): st = seq[pos] if st.get("risk_tier", "low") != "high": continue for nxt in seq[pos + 1:]: if nxt["tool"] != st["tool"] and (_words(nxt["tool"]) & UNDO_WORDS): corrections.append((st["tool"], nxt["tool"], res)) break wasted_cost = math.fsum([float(index[i].get("cost", 0)) for i in repeats | off_path]) if total > 0: share = wasted_cost / total else: share = (len(repeats | off_path) / len(trace)) if trace else 0.0 score = 1.0 - min(share, 1.0) * 0.7 - min(len(corrections) * 0.05, 0.3) score = round(max(0.0, min(1.0, score)), 4) return {"score": score, "cost": round(total, 4), "steps": len(trace), "extra": extra, "corrections": set(corrections), "tools": [st["tool"] for st in ordered], "keyed": [(_key(st["tool"]), st.get("args") or {}) for st in ordered], "high": [_key(st["tool"]) + "|" + _aj(st.get("args")) for st in ordered if st.get("risk_tier", "low") == "high" or _key(st["tool"]) in baseline_high], "lowered": [st for st in ordered if st.get("risk_tier", "low") != "high" and _key(st["tool"]) in baseline_high]} # ------------------------------------------------------------------------------------------------ gate def derive(suite): policy = suite.get("policy", {}) drop_limit = float(policy.get("max_score_drop", DEFAULT_MAX_SCORE_DROP)) cost_limit = float(policy.get("max_cost_increase_pct", DEFAULT_MAX_COST_PCT)) step_limit = policy.get("max_steps") banned = set(_key(t) for t in policy.get("forbidden_tools", [])) ordering = policy.get("require_before", []) def runs_of(side): return dict((r["case_id"], r) for r in suite[side]["runs"]) old_runs, new_runs = runs_of("baseline"), runs_of("candidate") baseline_high = set() for r in suite["baseline"]["runs"]: for st in r.get("trace", []): if st.get("risk_tier", "low") == "high": baseline_high.add(_key(st["tool"])) findings, metrics = [], [] for case in suite["cases"]: cid = case["case_id"] old, new = old_runs.get(cid), new_runs.get(cid) info = {} for name, run in (("old", old), ("new", new)): if run is not None: m = path_metrics(run.get("trace", []), run.get("terminal_step"), baseline_high if name == "new" else ()) m["ok"] = outcome_ok(case["checker"], run) info[name] = m hits = {} if new is None: hits["MISSING_RUN"] = "BLOCK" else: n = info["new"] if old is None: hits["NO_BASELINE"] = "WARN" if not n["ok"]: hits["NEW_CASE_FAILS"] = "BLOCK" else: o = info["old"] if o["ok"] and not n["ok"]: hits["OUTCOME_REGRESSION"] = "BLOCK" elif n["ok"] and not o["ok"]: hits["FIXED"] = "INFO" elif not n["ok"] and not o["ok"]: hits["STILL_FAILING"] = "WARN" elif _as_text_and_json(old.get("output", ""))[0] != _as_text_and_json(new.get("output", ""))[0]: hits["OUTPUT_CHANGED"] = "INFO" if o["score"] - n["score"] > drop_limit + EPS: hits["PATH_SCORE_DROP"] = "BLOCK" if n["cost"] > o["cost"] + EPS: if o["cost"] <= 0 or 100.0 * (n["cost"] - o["cost"]) / o["cost"] > cost_limit + EPS: hits["COST_INCREASE"] = "BLOCK" if n["corrections"] - o["corrections"]: hits["RISKY_CORRECTION"] = "BLOCK" remaining = list(n["high"]) for h in o["high"]: if h in remaining: remaining.remove(h) if remaining: hits["HIGH_RISK_CALL_CHANGED"] = "BLOCK" if n["extra"] > o["extra"]: hits["MORE_REDUNDANT_CALLS"] = "WARN" if o["steps"] > 0 and n["steps"] == 0: hits["PATH_MISSING"] = "BLOCK" if o["cost"] > 0 and n["steps"] > 0 and n["cost"] <= 0: hits["COST_NOT_REPORTED"] = "WARN" if n["lowered"]: hits["RISK_TIER_DOWNGRADE"] = "BLOCK" if banned.intersection(_key(t) for t in n["tools"]): hits["FORBIDDEN_TOOL"] = "BLOCK" def same_subject(x, y): if not (isinstance(x, dict) and isinstance(y, dict)): return True common = set(x) & set(y) if y and not common: return False return all(_aj(x[k]) == _aj(y[k]) for k in common) for first, later in ordering: fk, lk = _key(first), _key(later) for pos, (k, args) in enumerate(n["keyed"]): if k == lk and not any(k0 == fk and same_subject(a0, args) for k0, a0 in n["keyed"][:pos]): hits["ORDER_VIOLATION"] = "BLOCK" break if step_limit is not None and n["steps"] > step_limit: hits["STEP_LIMIT"] = "BLOCK" for code in sorted(hits): findings.append([cid, code, hits[code]]) def cell(name): m = info.get(name) return None if m is None else [m["ok"], m["steps"], m["cost"], m["score"]] metrics.append([cid, cell("old"), cell("new")]) severities = [f[2] for f in findings] verdict = "BLOCK" if "BLOCK" in severities else ("WARN" if "WARN" in severities else "PASS") return {"verdict": verdict, "findings": findings, "metrics": metrics} # ------------------------------------------------------------------------------------------------ records def _canon(obj): return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() def _sha(data): return hashlib.sha256(data.encode() if isinstance(data, str) else data).hexdigest() def _pae(kind, payload): return b"DSSEv1 %d %s %d %s" % (len(kind.encode()), kind.encode(), len(payload), payload) def check_record(rec, pinned_b64, prev=None): from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey pub = Ed25519PublicKey.from_public_bytes(base64.b64decode(pinned_b64)) problems = [] body = dict((k, v) for k, v in rec.items() if k not in ("record_hash", "signature_b64", "public_key_b64")) if _sha(_canon(body)) != rec.get("record_hash"): problems.append("content does not match record_hash") try: pub.verify(base64.b64decode(rec.get("signature_b64", "")), str(rec.get("record_hash", "")).encode()) except (InvalidSignature, ValueError): problems.append("signature does not verify against the pinned key") if prev is not None and rec.get("prev_record_hash") != prev.get("record_hash"): problems.append("prev_record_hash does not link to the previous record") src, der = rec.get("original_code"), rec.get("fixed_code") if not isinstance(src, str) or not isinstance(der, str): return problems + ["original_code / fixed_code missing"] if _sha(src) != rec.get("original_sha256") or _sha(der) != rec.get("fixed_sha256"): problems.append("input or derivation does not match its digest") try: mine = derive(json.loads(src)) if _canon(mine).decode() != der: problems.append("independent re-derivation differs from the signed derivation") if mine["verdict"] != rec.get("verdict"): problems.append("record says %s, the input gives %s" % (rec.get("verdict"), mine["verdict"])) except Exception as e: # noqa: BLE001 - an input this verifier cannot grade is a failed check problems.append("cannot re-derive from the embedded input (%s)" % type(e).__name__) env = rec.get("dsse") or {} try: payload = base64.b64decode(env["payload"]) pub.verify(base64.b64decode(env["signatures"][0]["sig"]), _pae(env["payloadType"], payload)) if json.loads(payload)["predicate"]["verdict"] != rec.get("verdict"): problems.append("DSSE statement verdict differs from the record's") except Exception: # noqa: BLE001 problems.append("DSSE envelope missing or does not verify") return problems def main(argv=None): args = list(sys.argv[1:] if argv is None else argv) pinned = None if "--pinned-pub" in args: i = args.index("--pinned-pub") pinned = args[i + 1] del args[i:i + 2] pinned_fp = None if "--pinned-fingerprint" in args: i = args.index("--pinned-fingerprint") pinned_fp = args[i + 1] del args[i:i + 2] files = args or sorted(glob.glob("record*.json")) if not files: print("no record files (pass record.json, or run where record*.json exist)") return 2 if pinned is None: # Never trust the bundle's own public_key.json on its own: whoever hands over the bundle could have swapped # it and re-signed every record (audit F-1, 2026-10-23). Accept it only if it matches a fingerprint the # auditor pinned out of band. bundled = None try: with open("public_key.json", encoding="utf-8") as fh: bundled = json.load(fh)["public_key_b64"] except (OSError, KeyError, ValueError): pass claimed = ("SHA256:" + hashlib.sha256(base64.b64decode(bundled)).hexdigest()) if bundled else None want = (pinned_fp or "").strip().lower() want = want if not want or want.startswith("sha256:") else "sha256:" + want if want and claimed and len(want) >= len("sha256:") + 32 and claimed.lower().startswith(want): pinned = bundled else: print("NOT VERIFIED: no pinned key. Pass --pinned-pub fetched from the issuer yourself, or " "--pinned-fingerprint SHA256:. The bundle claims %s." % (claimed or "no key")) print("RESULT: NOT VERIFIED (no trusted key)") return 3 all_ok, prev = True, None for f in files: try: with open(f, encoding="utf-8") as fh: rec = json.load(fh) except (OSError, ValueError) as e: print("INVALID %s (unreadable: %s)" % (f, e)) all_ok, prev = False, None continue problems = check_record(rec, pinned, prev) print("%s %s verdict=%s proof=%s" % ("VALID " if not problems else "INVALID", f, rec.get("verdict"), (rec.get("proof") or {}).get("status"))) for p in problems: print(" - " + p) all_ok = all_ok and not problems prev = rec print("RESULT:", "ALL VERDICTS RE-DERIVED AND VALID" if all_ok else "TAMPERED OR INVALID") return 0 if all_ok else 1 if __name__ == "__main__": raise SystemExit(main())