"""CAIN-A75-032 offline verifier (rules embedded verbatim). CAIN-A75-032 research-retrieval rules. Standard library only, no CAIN imports: embedded verbatim in the offline verifier. An agent searches biological archives. The review recomputes whether every retrieved record is traceable to a licensed, snapshotted source and whether controlled data was approved: archives every archive used is allowed by the policy, has a snapshot hash, and declares a licence snapshot each result carries the same snapshot hash as its archive (the record is bound to one release) licence each result carries the archive's licence and a terms URL freshness each result was retrieved within the policy's maximum age controlled a result marked controlled needs a live approval by someone other than the agent citations every citation points at a retrieved result and carries a locator; no orphan citation scope no result comes from an archive outside the query claims every boolean the record asserts matches what this review computed Verdict: SOUND, else UNSOUND with every failing check named. Malformed input is UNSOUND. It reviews the retrieval record; it does not search an archive. """ 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}$") 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 = [], {"results": {}, "citations": 0} 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.retrieval.v1": raise Bad("not a cain42.a75.retrieval.v1 record") query, pol = record.get("query"), record.get("policy") if not all(isinstance(x, dict) for x in (query, pol)): raise Bad("query and policy must be objects") now = _num(query.get("now"), "query.now") purpose = query.get("purpose") if not isinstance(purpose, str) or not purpose.strip(): raise Bad("query.purpose required") qarchives = query.get("archives") if not isinstance(qarchives, list) or not qarchives: raise Bad("query.archives must be a non-empty list") allowed = pol.get("allowed_archives") if not isinstance(allowed, list) or not allowed: raise Bad("policy.allowed_archives must be a non-empty list") max_age_days = _num(pol.get("max_age_days"), "policy.max_age_days", 0.0, 100000.0) archives = record.get("archives") if not isinstance(archives, list) or not archives: raise Bad("archives must be a non-empty list") A = {} for a in archives: if not isinstance(a, dict) or not isinstance(a.get("id"), str) or not _ID.match(a["id"]): raise Bad("an archive needs an id") if a["id"] in A: raise Bad(f"duplicate archive {a['id']}") if not isinstance(a.get("snapshot_sha256"), str) or not _HEX.match(a["snapshot_sha256"]): raise Bad(f"{a['id']}: snapshot_sha256 required") if not isinstance(a.get("license"), str) or not a["license"].strip(): raise Bad(f"{a['id']}: license required") if not isinstance(a.get("terms_url"), str) or not a["terms_url"].startswith(("http://", "https://")): raise Bad(f"{a['id']}: terms_url required") c(f"archive.{a['id']}.allowed", a["id"] in allowed, f"archive {a['id']!r} allowed") A[a["id"]] = a approvals = {} for ap in record.get("approvals") or []: if not isinstance(ap, dict) or ap.get("archive") not in A: raise Bad("an approval must name a known archive") approvals[ap["archive"]] = ap results = record.get("results") if not isinstance(results, list) or not results or len(results) > 5000: raise Bad("results must be a non-empty list") R = {} for r in results: if not isinstance(r, dict) or not isinstance(r.get("id"), str) or not _ID.match(r["id"]): raise Bad("a result needs an id") if r["id"] in R: raise Bad(f"duplicate result {r['id']}") R[r["id"]] = r archive = r.get("archive") c(f"{r['id']}.archive_known", archive in A, f"archive {archive!r}") c(f"{r['id'].replace(chr(46), '_')}.in_query", archive in qarchives, f"archive {archive!r} in query") if archive not in A: continue c(f"{r['id'].replace(chr(46), '_')}.snapshot", r.get("snapshot_sha256") == A[archive]["snapshot_sha256"], "result is bound to the archive snapshot") c(f"{r['id'].replace(chr(46), '_')}.license", r.get("license") == A[archive]["license"] and isinstance(r.get("license"), str), f"license {r.get('license')!r}") retrieved = _num(r.get("retrieved_at"), f"{r['id']}.retrieved_at") c(f"{r['id'].replace(chr(46), '_')}.fresh", (now - retrieved) / 86400.0 <= max_age_days + 1e-9, f"{(now - retrieved) / 86400.0:.2f} days <= {max_age_days}") if r.get("controlled") is True: ap = approvals.get(archive) c(f"{r['id'].replace(chr(46), '_')}.controlled_approved", ap is not None and ap.get("approved_by") not in (None, "", query.get("agent")), f"approved by {(ap or {}).get('approved_by')!r}") if ap is not None: c(f"{r['id'].replace(chr(46), '_')}.approval_live", _num(ap.get("expires_at"), "approval.expires_at") > now, "approval not expired") summary["results"][r["id"]] = {"archive": archive, "controlled": r.get("controlled") is True} citations = record.get("citations") if not isinstance(citations, list) or not citations: raise Bad("citations must be a non-empty list") for k, ct in enumerate(citations): if not isinstance(ct, dict): raise Bad("a citation must be an object") c(f"citation.{k}.result", ct.get("result_id") in R, f"result {ct.get('result_id')!r}") c(f"citation.{k}.locator", isinstance(ct.get("locator"), str) and bool(ct["locator"].strip()), "citation locator present") summary["citations"] += 1 claims = record.get("claims") if isinstance(record.get("claims"), dict) else {} computed = { "all_archives_allowed": all(a in allowed for a in qarchives), "all_results_bound": all(r.get("snapshot_sha256") == A[r.get("archive", "")]["snapshot_sha256"] for r in results if r.get("archive") in A), "no_unapproved_controlled": all( r.get("controlled") is not True or (approvals.get(r.get("archive")) is not None and approvals[r["archive"]].get("approved_by") not in (None, "", query.get("agent"))) for r in results if r.get("archive") in A), } 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 "UNSOUND", checks, summary except Exception as e: # noqa: BLE001 checks.append({"id": "input.error", "ok": False, "detail": f"{type(e).__name__}: {e}"[:200]}) return "UNSOUND", checks, summary return ("SOUND" if all(x["ok"] for x in checks) else "UNSOUND"), 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.retrieval.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-032, {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))