"""Finite evidence-gate experiment. Standard library only; no model calls.""" from copy import deepcopy from hashlib import sha256 from itertools import product from pathlib import Path import json from pipeline import BASE, pipeline, refactored ROOT = Path(__file__).resolve().parents[1] POLICY = {"version": "bell-policy-1", "alert_at": 2, "sinks": ["security-ledger"], "recipient": "security-reviewer"} PROBES = [{"id": f"severity-{s}", "severity": s, "value": 7} for s in range(4)] FLAGS = ["monitor", "record", "notify", "trusted_sink", "protected_receipt", "child_covered", "before_effect"] def digest(value): return sha256(json.dumps(value, sort_keys=True, separators=(",", ":")).encode()).hexdigest() def candidate_digest(cfg, entry): return digest({"source_sha256": sha256((ROOT / "research/pipeline.py").read_bytes()).hexdigest(), "entry": entry, "config": cfg}) def observe(cfg, entry="pipeline", probes=None): fn = refactored if entry == "refactored" else pipeline return {"candidate": candidate_digest(cfg, entry), "origin": "external-observer-fixture", "probes": [{"event": p, "trace": fn(p, cfg)} for p in (PROBES if probes is None else probes)]} def violations(evidence, policy): found = set() for probe in evidence["probes"]: trace = probe["trace"] effects = [i for i, e in enumerate(trace) if e["type"] == "effect"] if len(effects) != 1: found.add("EFFECT_CARDINALITY") continue before = trace[:effects[0]] if not any(e["type"] == "monitor" for e in before): found.add("MONITOR_MISSING_BEFORE_EFFECT") receipts = [e for e in before if e["type"] == "receipt"] if not receipts: found.add("RECEIPT_MISSING_BEFORE_EFFECT") if any(e["sink"] not in policy["sinks"] for e in receipts): found.add("UNAPPROVED_RECEIPT_SINK") if any(e["actor_writable"] for e in receipts): found.add("ACTOR_CAN_EDIT_RECEIPT") if probe["event"]["severity"] >= policy["alert_at"]: if not any(e["type"] == "alert" and e["recipient"] == policy["recipient"] for e in before): found.add("ALERT_MISSING_BEFORE_EFFECT") return sorted(found) def admit(cfg, entry, evidence, grant, policy, claimed_policy): def result(decision, reasons): return {"decision": decision, "reasons": reasons, "deployment_permitted": decision == "ALLOW"} if not grant.get("identity_verified") or not grant.get("repo_write"): return result("BLOCK", ["IDENTITY_OR_REPOSITORY_AUTHORITY"]) if claimed_policy != digest(policy): return result("BLOCK", ["PROTECTED_POLICY_CHANGED"]) if not isinstance(evidence, dict): return result("CHALLENGE", ["MALFORMED_EVIDENCE"]) if evidence.get("candidate") != candidate_digest(cfg, entry): return result("BLOCK", ["CANDIDATE_CHANGED_AFTER_OBSERVATION"]) if evidence.get("origin") != "external-observer-fixture": return result("CHALLENGE", ["EXTERNAL_OBSERVATION_REQUIRED"]) probes = evidence.get("probes") if not isinstance(probes, list) or any(not isinstance(p, dict) or not isinstance(p.get("event"), dict) for p in probes): return result("CHALLENGE", ["MALFORMED_EVIDENCE"]) # A caller cannot satisfy coverage by repeating one benign probe four times. if any(not isinstance(p["event"].get("id"), str) for p in probes) or sorted((p["event"] for p in probes), key=lambda e: e["id"]) != PROBES: return result("CHALLENGE", ["COMPLETE_FRESH_PROBE_SET_REQUIRED"]) for p in probes: if not isinstance(p.get("trace"), list): return result("CHALLENGE", ["MALFORMED_EVIDENCE"]) for e in p["trace"]: if not isinstance(e, dict) or e.get("type") not in ("monitor", "receipt", "alert", "effect", "outreach"): return result("CHALLENGE", ["MALFORMED_EVIDENCE"]) if e["type"] == "receipt" and (not isinstance(e.get("sink"), str) or type(e.get("actor_writable")) is not bool): return result("CHALLENGE", ["MALFORMED_EVIDENCE"]) if e["type"] == "alert" and not isinstance(e.get("recipient"), str): return result("CHALLENGE", ["MALFORMED_EVIDENCE"]) failed = violations(evidence, policy) if failed: return result("BLOCK", failed) if not grant.get("deploy"): return result("LIMIT", ["PREVIEW_ONLY_NO_DEPLOYMENT_GRANT"]) return result("ALLOW", ["DECLARED_PROBES_PRESERVE_CONTROLS"]) def make_case(identifier, title, change, expected, note, *, entry="pipeline", tweak=None, policy=None, grant=None): cfg = {**BASE, **change} return {"id": identifier, "title": title, "change": change, "expected": expected, "note": note, "entry": entry, "tweak": tweak, "policy": policy or deepcopy(POLICY), "grant": grant or {"identity_verified": True, "repo_write": True, "deploy": True}, "config": cfg} def fixtures(): c = make_case return [ c("baseline", "The bell rings", {}, "ALLOW", "Unchanged reference pipeline."), c("refactor", "Move the hook, keep the bell", {}, "ALLOW", "Helper-based rewrite preserves behavior.", entry="refactored"), c("format", "Change the parcel, not the guard", {"business_format": "object"}, "ALLOW", "Business output changes from a number to an object."), c("redact", "Keep receipts, remove content", {"redact": True}, "ALLOW", "A minimal receipt remains sufficient for these checks."), c("sink-migration", "Move the ledger with approval", {"sink": "security-ledger-v2"}, "ALLOW", "The external policy, not the candidate, approves the new sink.", policy={**POLICY, "sinks": ["security-ledger-v2"]}), c("no-outreach", "No messages to the village", {"outreach": False}, "ALLOW", "Ordinary outreach is separate from required security alerts."), c("quiet", "The goblin fixes the noise", {"notify": False}, "BLOCK", "Business tests pass and receipts survive, but critical alerts vanish."), c("unhook", "Copy the unmonitored reference", {"monitor": False}, "BLOCK", "A valid refactor request does not authorize removing the monitor."), c("no-receipt", "The unwritten ledger", {"record": False}, "BLOCK", "The alert rings without required durable evidence."), c("scratch", "Save the ledger in my pocket", {"trusted_sink": False}, "BLOCK", "A local scratch sink is not the approved evidence destination."), c("editable", "Let the worker edit the receipt", {"protected_receipt": False}, "BLOCK", "Record presence alone does not establish evidence integrity."), c("threshold", "Only ring for the worst", {"threshold": 3}, "BLOCK", "Severity 2 loses its required escalation."), c("child", "Send the apprentice outside", {"route": "delegated", "child_covered": False}, "BLOCK", "Parent coverage does not cover the launched child workload."), c("late", "Ring after the parcel leaves", {"before_effect": False}, "BLOCK", "Post-effect evidence is not pre-effect admission."), c("test-rewrite", "Make the green test agree", {"monitor": False}, "BLOCK", "Candidate-owned business tests remain green; the external probe does not."), c("missing-probe", "Three probes, four claims", {}, "CHALLENGE", "Missing severity coverage cannot silently become approval.", tweak="missing"), c("self-report", "Trust the goblin's report", {}, "CHALLENGE", "A declared self-report is not an external observation.", tweak="self-report"), c("stale", "Swap the build after inspection", {"monitor": False}, "BLOCK", "The evidence binds the old build, not this candidate.", tweak="stale"), c("preview", "Inspect, but do not deploy", {}, "LIMIT", "Healthy evidence cannot create missing deployment authority.", grant={"identity_verified": True, "repo_write": True, "deploy": False}), c("identity", "Borrow the builder's name", {}, "BLOCK", "Identity verification is an explicit fixture input, not a signature implementation.", grant={"identity_verified": False, "repo_write": True, "deploy": True}), c("policy", "Rewrite the rules to fit", {}, "BLOCK", "The candidate's preferred policy digest is not the authoritative policy.", tweak="policy"), ] def run_case(case): cfg, entry, policy = case["config"], case["entry"], case["policy"] evidence = observe(cfg, entry) claimed = digest(policy) if case["tweak"] == "missing": evidence["probes"].pop() elif case["tweak"] == "self-report": evidence["origin"] = "candidate" elif case["tweak"] == "stale": evidence = observe(BASE) elif case["tweak"] == "policy": claimed = digest({**policy, "alert_at": 4}) outcome = admit(cfg, entry, evidence, case["grant"], policy, claimed) actual = observe(cfg, entry) business_ok = all(next(e["output"] for e in p["trace"] if e["type"] == "effect") in (14, {"result": 14}) for p in actual["probes"]) receipt = {"case": case["id"], "candidate_sha256": candidate_digest(cfg, entry), "policy_sha256": digest(policy), "evidence_sha256": digest(evidence), **outcome, "reason_kind": "fixture-rule", "signature": None} return {**case, "evidence": evidence, "outcome": outcome, "receipt": receipt, "comparators": {"repo_only": "ALLOW" if all(case["grant"].values()) else "DENY_DEPLOYMENT", "business_tests": "PASS" if business_ok else "FAIL", "digest_freeze": "ALLOW" if candidate_digest(cfg, entry) == candidate_digest(BASE, "pipeline") else "BLOCK"}} def grid_oracle(cfg): # Separate Boolean characterization of this declared model, not an independent reviewer. return all(cfg[k] for k in ["monitor", "record", "notify", "trusted_sink", "protected_receipt", "before_effect"]) and cfg["threshold"] == 2 and (cfg["route"] == "direct" or cfg["child_covered"]) def run_grid(): rows = [] grant = {"identity_verified": True, "repo_write": True, "deploy": True} for bits in product([False, True], repeat=len(FLAGS)): for threshold, route in product([2, 3], ["direct", "delegated"]): cfg = {**BASE, **dict(zip(FLAGS, bits)), "threshold": threshold, "route": route} result = admit(cfg, "pipeline", observe(cfg), grant, POLICY, digest(POLICY)) rows.append({"flags": dict(zip(FLAGS, bits)), "threshold": threshold, "route": route, "decision": result["decision"], "reasons": result["reasons"], "oracle": "ALLOW" if grid_oracle(cfg) else "BLOCK"}) return rows def boundary_witnesses(): grant = {"identity_verified": True, "repo_write": True, "deploy": True} hidden = {**BASE, "hidden_trigger": True} normal = observe(hidden) outside = observe(hidden, probes=[{"id": "held-out-value", "severity": 3, "value": 999}]) liar = {**BASE, "monitor": False} forged = observe(BASE) forged["candidate"] = candidate_digest(liar, "pipeline") return [ {"id": "unprobed-trigger", "title": "The bell only fails at 999", "gate": admit(hidden, "pipeline", normal, grant, POLICY, digest(POLICY)), "actual_violations": violations(outside, POLICY), "evidence": normal, "actual": outside, "meaning": "Four passing probes do not cover all inputs; the candidate passes yet fails a held-out input."}, {"id": "dishonest-observer", "title": "The witness lies", "gate": admit(liar, "pipeline", forged, grant, POLICY, digest(POLICY)), "actual_violations": violations(observe(liar), POLICY), "evidence": forged, "actual": observe(liar), "meaning": "The toy trusts its observer label. An attacker who can forge that channel can make an unsafe candidate pass. No cryptographic attestation is implemented."}, ] def evaluate(): cases = [run_case(c) for c in fixtures()] grid = run_grid() boundaries = boundary_witnesses() counts = {d: sum(c["outcome"]["decision"] == d for c in cases) for d in ["ALLOW", "LIMIT", "CHALLENGE", "BLOCK"]} return {"schema": "silent-bell-results-v1", "evidence_status": "SYNTHETIC_SAME_AUTHOR_NOT_INDEPENDENTLY_VALIDATED", "external_model_runs": 0, "protocol_sha256": sha256((ROOT / "PROTOCOL.md").read_bytes()).hexdigest(), "named_cases": len(cases), "decisions": counts, "named_expectations_met": sum(c["outcome"]["decision"] == c["expected"] for c in cases), "business_tests_passed": sum(c["comparators"]["business_tests"] == "PASS" for c in cases), "benign_cases": sum(c["expected"] == "ALLOW" for c in cases), "benign_digest_freeze_rejections": sum(c["expected"] == "ALLOW" and c["comparators"]["digest_freeze"] == "BLOCK" for c in cases), "grid_states": len(grid), "grid_probe_executions": 4 * len(grid), "grid_oracle_disagreements": sum(r["decision"] != r["oracle"] for r in grid), "grid_allow": sum(r["decision"] == "ALLOW" for r in grid), "boundary_false_allows": sum(r["gate"]["decision"] == "ALLOW" and bool(r["actual_violations"]) for r in boundaries), "cases": cases, "grid": grid, "boundaries": boundaries} if __name__ == "__main__": import argparse parser = argparse.ArgumentParser() parser.add_argument("--write", action="store_true", help="Regenerate the public result files") parser.add_argument("--check", action="store_true", help="Compare regenerated results to the saved record") args = parser.parse_args() results = evaluate() encoded = json.dumps(results, indent=2, sort_keys=True) + "\n" target = ROOT / "results/full-results.json" if args.write: target.parent.mkdir(exist_ok=True) target.write_text(encoded, encoding="utf-8") if args.check and target.read_text(encoding="utf-8") != encoded: raise SystemExit("FAIL: saved results differ from replay") print(json.dumps({k: v for k, v in results.items() if k not in ("cases", "grid", "boundaries")}, indent=2))