"""CAIN-A75-010 offline verifier (rules embedded verbatim). CAIN-A75-010 failure-driven-improvement rules. Standard library only, no CAIN imports: embedded verbatim in the offline verifier. An agent changes itself after failures. The audit closes the loop from evidence, not from the change log: coverage every HIGH or CRITICAL incident is addressed by at least one fix; every fix addresses real incidents fix proof each fix names a changed artifact hash, a rollback plan, and a test that FAILED before and PASSES after (a test that already passed proves nothing about this fix) recurrence no incident is marked as having recurred after its fix suite the full regression suite ran and every test passed after the fixes honesty every boolean the record asserts matches what this audit computed Verdict: CLOSED, else OPEN with every failing check named. Malformed input is OPEN. It audits the supplied failure/fix record; it does not reproduce the failures or run the suite. """ from __future__ import annotations import re _ID = re.compile(r"^[A-Za-z0-9_.:\-]{1,80}$") _HEX = re.compile(r"^[0-9a-f]{64}$") _SEV = {"LOW": 0, "MEDIUM": 1, "HIGH": 2, "CRITICAL": 3} class Bad(ValueError): pass def evaluate(record): checks, summary = [], {"coverage": {}, "fixes": {}} 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.failurefix.record.v1": raise Bad("not a cain42.a75.failurefix.record.v1 record") incidents, fixes = record.get("incidents"), record.get("fixes") if not isinstance(incidents, list) or not (1 <= len(incidents) <= 500): raise Bad("incidents must be a list of 1-500 entries") if not isinstance(fixes, list) or not fixes or len(fixes) > 500: raise Bad("fixes must be a non-empty list") I = {} for inc in incidents: if not isinstance(inc, dict) or not isinstance(inc.get("id"), str) or not _ID.match(inc["id"]): raise Bad("an incident needs an id") if inc["id"] in I: raise Bad(f"duplicate incident id {inc['id']}") if inc.get("severity") not in _SEV: raise Bad(f"{inc['id']}: severity must be one of {sorted(_SEV)}") if not isinstance(inc.get("trace_sha256"), str) or not _HEX.match(inc["trace_sha256"]): raise Bad(f"{inc['id']}: trace_sha256 must be 64 hex characters") I[inc["id"]] = inc F = {} for fix in fixes: if not isinstance(fix, dict) or not isinstance(fix.get("id"), str) or not _ID.match(fix["id"]): raise Bad("a fix needs an id") if fix["id"] in F: raise Bad(f"duplicate fix id {fix['id']}") F[fix["id"]] = fix addressed = set() for fid, fix in F.items(): targets = fix.get("addresses") c(f"{fid}.addresses_real_incidents", isinstance(targets, list) and bool(targets) and all(t in I for t in targets), f"addresses {targets!r}") if isinstance(targets, list): addressed |= {t for t in targets if t in I} c(f"{fid}.change_hash", isinstance(fix.get("change_sha256"), str) and bool(_HEX.match(fix["change_sha256"])), "the changed artifact is named by hash") c(f"{fid}.rollback", isinstance(fix.get("rollback_plan"), str) and len(fix["rollback_plan"].strip()) >= 12, "a rollback plan is stated") t = fix.get("test") if isinstance(fix.get("test"), dict) else {} c(f"{fid}.test_failed_before", t.get("before") == "fail", f"before {t.get('before')!r}") c(f"{fid}.test_passes_after", t.get("after") == "pass", f"after {t.get('after')!r}") summary["fixes"][fid] = {"before": t.get("before"), "after": t.get("after"), "addresses": targets} for iid in sorted(I): inc = I[iid] needs = _SEV[inc["severity"]] >= _SEV["HIGH"] covered = iid in addressed summary["coverage"][iid] = {"severity": inc["severity"], "addressed": covered, "required": needs} if needs: c(f"{iid}.high_severity_addressed", covered, f"{inc['severity']} incident {'addressed' if covered else 'has no fix'}") c(f"{iid}.no_recurrence", inc.get("recurred_after_fix") is not True, "did not recur after the fix" if inc.get("recurred_after_fix") is not True else "recurred after the fix") suite = record.get("regression_suite") if isinstance(record.get("regression_suite"), dict) else {} total, failed = suite.get("tests"), suite.get("failed") c("suite.ran", suite.get("ran") is True, "the regression suite ran") c("suite.counts", isinstance(total, int) and not isinstance(total, bool) and isinstance(failed, int) and not isinstance(failed, bool) and total > 0 and 0 <= failed <= total, f"{failed!r} failed of {total!r}") c("suite.all_pass", failed == 0, f"{failed!r} failing test(s)") claims = record.get("claims") if isinstance(record.get("claims"), dict) else {} high = [i for i in I if _SEV[I[i]["severity"]] >= _SEV["HIGH"]] computed = { "all_high_severity_addressed": all(i in addressed for i in high), "no_recurrence": all(I[i].get("recurred_after_fix") is not True for i in I), "suite_passes": failed == 0, } for k in sorted(claims): c(f"claims.{k}", k in computed and claims[k] == computed[k], f"claimed {claims[k]!r}, computed {computed.get(k)!r}") summary["computed"] = computed except Bad as e: checks.append({"id": "input.well_formed", "ok": False, "detail": str(e)}) return "OPEN", checks, summary except Exception as e: # noqa: BLE001 checks.append({"id": "input.error", "ok": False, "detail": f"{type(e).__name__}: {e}"[:200]}) return "OPEN", checks, summary return ("CLOSED" if all(x["ok"] for x in checks) else "OPEN"), 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.failurefix.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-010, {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))