""" SNIN Emission Paper Pilot — v0.4.3 runner. Gate #25132: signed capability/attestation events, issuer authority, atomic budget consumption, attestation binding+consumption, verdict gate, idempotent receipts (crash points), reversal integrity, P1 > reds. """ import json import emission as E import ledger_store as LS V = json.load(open("test_vectors_v0.4.3.json")) fails = [] def check(name, ok, detail=""): print(("PASS " if ok else "FAIL ") + name + (f" [{detail}]" if detail else "")) if not ok: fails.append(name + " " + detail) # ---- deterministic keyring (private keys NEVER leave this runner) ------- KEYS = {"root": LS.deterministic_key("root-seed"), "ctrl": LS.deterministic_key("ctrl-seed"), "alice": LS.deterministic_key("alice-seed")} PUBS = {kid: KEYS[kid].public_key() for kid in KEYS} DIG = V["digests"] def fresh_store(): s = LS.LedgerStore(V["policy"]) return s def base_fixtures(s, history_name, root=KEYS["root"], ctrl=KEYS["ctrl"]): """Load one fixture history: issues+attestations of its capabilities, then its events.""" fx = V["fixture_histories"][history_name] caps = V["capabilities"] atts = V["attestations"] used_ass = [] for ev in fx["events"]: if ev.get("kind") == "emission": cap = next(c for c in caps if c["assignment_id"] == ev["assignment_id"]) used_ass.append(cap["assignment_id"]) att = next((a for a in atts if a["attestation_id"] == ev.get("attestation_id")), None) # issue capabilities used by this history (in deterministic order) for cap in caps: if cap["assignment_id"] in used_ass: s.issue_capability(cap, "root", root) for ev in fx["events"]: if ev.get("kind") != "emission": continue att_id = ev.get("attestation_id") if att_id: a = next(a for a in atts if a["attestation_id"] == att_id) s.add_attestation({"attestation_id": a["attestation_id"], "task_id": a["task_id"], "evidence_digest": DIG[a["evidence_digest"]], "assignee": a["assignee"], "verdict": a["verdict"], "ts": 20}, "ctrl", ctrl) seq_of_pos = {} pos = 0 for ev in fx["events"]: if ev["kind"] == "emission": pos += 1 seq = json.loads(s.append_emission({"actor": ev["actor"], "emission_id": ev["emission_id"], "emission_class": ev["emission_class"], "amount_micro": ev["amount_micro"], "outcome": ev["outcome"], "role": ev["role"], "assignment_id": ev["assignment_id"], "evidence_digest": DIG[ev["evidence_digest"]], "attestation_id": ev.get("attestation_id")}, root))["seq"] seq_of_pos[pos] = seq else: # reversal target = seq_of_pos[ev["reverses_event_id"]] s.append_reversal({"actor": ev["actor"], "reverses_event_id": target, "effect": ev["effect"], "basis": ev["basis"], "amount_micro": ev["amount_micro"], "assignment_id": ev["assignment_id"]}, root) return s # ---- 1. P1 > reds ------------------------------------------------------- reds = {} for h in ("R2_plausible_incorrect", "R3_self_circle"): s = fresh_store() base_fixtures(s, h) nets = E.net_balance(s.events(), V["fixture_histories"][h]["actor"]) reds[h] = nets check(f"{h} net == 0", nets == 0, f"got {nets}") s = fresh_store() base_fixtures(s, "P1_corrective") p1 = E.net_balance(s.events(), "V") check("P1 > max(R1..R3): 100M > 0", p1 == 100_000_000, f"P1={p1}, reds={reds}") # ---- 2. falsifiers from #25132 ------------------------------------------ for f in V["falsifiers"]: s = fresh_store() try: if f["kind"] == "issue": s.issue_capability(f["capability"], f["issuer"], KEYS[f["issuer"]]) elif f["kind"] == "attest": a = f["attestation"] s.add_attestation({"attestation_id": a["attestation_id"], "task_id": a["task_id"], "evidence_digest": DIG.get(a["evidence_digest"], a["evidence_digest"]), "assignee": a["assignee"], "verdict": a["verdict"], "ts": 20}, f["issuer"], KEYS[f["issuer"]]) elif f["kind"] == "emission": ev = f["emission"] # pre-issue capability + attestation referenced by the attack cap = next((c for c in V["capabilities"] if c["assignment_id"] == ev["assignment_id"]), None) if cap and f["name"] not in ("flag_to_verify_accept",): s.issue_capability(cap, "root", KEYS["root"]) att_id = ev.get("attestation_id") if att_id: a = next((a for a in V["attestations"] if a["attestation_id"] == att_id), None) if a and f["name"] not in ("flag_to_verify_accept",): s.add_attestation({"attestation_id": a["attestation_id"], "task_id": a["task_id"], "evidence_digest": DIG[a["evidence_digest"]], "assignee": a["assignee"], "verdict": a["verdict"], "ts": 20}, "ctrl", KEYS["ctrl"]) if f.get("after"): # load history that consumes the attestation first base_fixtures(s, "R2_plausible_incorrect") # add a fresh sibling assignment so budget is not the reason s.issue_capability({"assignment_id": "A2001B", "task_id": "T2001", "assignee": "V", "emission_class": "verify_accept", "cap_micro": 100_000_000, "review_of": "W2"}, "root", KEYS["root"]) ev = dict(ev) ev["assignment_id"] = "A2001B" if f.get("pre_attest"): pa = f["pre_attest"] s.issue_capability({"assignment_id": "A2001B", "task_id": "T2001", "assignee": "V", "emission_class": "verify_accept", "cap_micro": 100_000_000, "review_of": "W2"}, "root", KEYS["root"]) s.add_attestation({"attestation_id": pa["attestation_id"], "task_id": pa["task_id"], "evidence_digest": DIG[pa["evidence_digest"]], "assignee": pa["assignee"], "verdict": pa["verdict"], "ts": 20}, "ctrl", KEYS["ctrl"]) s.append_emission({"actor": ev["actor"], "emission_id": ev["emission_id"], "emission_class": ev["emission_class"], "amount_micro": ev["amount_micro"], "outcome": ev["outcome"], "role": ev["role"], "assignment_id": ev["assignment_id"], "evidence_digest": DIG.get(ev["evidence_digest"], ev["evidence_digest"]), "attestation_id": ev.get("attestation_id")}, KEYS["root"]) if f.get("then"): t = f["then"] s.append_emission({"actor": t["actor"], "emission_id": t["emission_id"], "emission_class": t["emission_class"], "amount_micro": t["amount_micro"], "outcome": t["outcome"], "role": t["role"], "assignment_id": t["assignment_id"], "evidence_digest": DIG.get(t["evidence_digest"], t["evidence_digest"]), "attestation_id": t.get("attestation_id")}, KEYS["root"]) elif f["kind"] == "raw": s._base_gate(dict(f["event"])) check(f"falsifier:{f['name']} rejected", False, "accepted!") except LS.LedgerError as e: check(f"falsifier:{f['name']} rejected", True, str(e)[:90]) # ---- 3. adversarial reversals ------------------------------------------- s = fresh_store() s.issue_capability({"assignment_id": "A2001", "task_id": "T2001", "assignee": "V", "emission_class": "verify_accept", "cap_micro": 100_000_000, "review_of": "W2"}, "root", KEYS["root"]) s.add_attestation({"attestation_id": "AT2001", "task_id": "T2001", "evidence_digest": DIG["D_W2"], "assignee": "V", "verdict": "accept", "ts": 20}, "ctrl", KEYS["ctrl"]) E2001_SEQ = json.loads(s.append_emission({"actor": "V", "emission_id": "E2001", "emission_class": "verify_accept", "amount_micro": 100_000_000, "outcome": "accepted", "role": "review", "assignment_id": "A2001", "evidence_digest": DIG["D_W2"], "attestation_id": "AT2001"}, KEYS["root"]))["seq"] for adv in V["adversarial_reversals"]: s2 = fresh_store() s2.issue_capability({"assignment_id": "A2001", "task_id": "T2001", "assignee": "V", "emission_class": "verify_accept", "cap_micro": 100_000_000, "review_of": "W2"}, "root", KEYS["root"]) s2.add_attestation({"attestation_id": "AT2001", "task_id": "T2001", "evidence_digest": DIG["D_W2"], "assignee": "V", "verdict": "accept", "ts": 20}, "ctrl", KEYS["ctrl"]) s2.append_emission({"actor": "V", "emission_id": "E2001", "emission_class": "verify_accept", "amount_micro": 100_000_000, "outcome": "accepted", "role": "review", "assignment_id": "A2001", "evidence_digest": DIG["D_W2"], "attestation_id": "AT2001"}, KEYS["root"]) if adv.get("setup_first"): sf = adv["setup_first"] s2.append_reversal({"actor": sf["actor"], "reverses_event_id": E2001_SEQ, "effect": sf["effect"], "basis": sf["basis"], "amount_micro": sf["amount_micro"], "assignment_id": sf["assignment_id"]}, KEYS["root"]) rv = adv["reversal"] try: s2.append_reversal({"actor": rv["actor"], "reverses_event_id": rv["reverses_event_id"], "effect": rv["effect"], "basis": rv["basis"], "amount_micro": rv["amount_micro"], "assignment_id": rv["assignment_id"]}, KEYS["root"]) check(f"adv:{adv['name']} rejected", False, "accepted!") except LS.LedgerError: check(f"adv:{adv['name']} rejected", True) # ---- 4. verify_all signatures + mutation detection ----------------------- s = fresh_store() base_fixtures(s, "P1_corrective") okv, msg = s.verify_all(PUBS) check("all events have valid Ed25519 signatures", okv, msg) # mutation of a stored event must break signature verification copy_events = s.events() copy_events[0]["cap_micro"] = 999_999_999 okm = all(s.verify_signature(e, PUBS[e["sig_key"]]) for e in copy_events) check("mutated event detected (sig mismatch)", not okm) # ---- 5. crash points (idempotent receipts) ------------------------------ s = fresh_store() s.issue_capability({"assignment_id": "A_CR", "task_id": "T_CR", "assignee": "V", "emission_class": "verify_accept", "cap_micro": 100_000_000, "review_of": "W2"}, "root", KEYS["root"]) s.add_attestation({"attestation_id": "AT_CR", "task_id": "T_CR", "evidence_digest": DIG["D_W2"], "assignee": "V", "verdict": "accept", "ts": 20}, "ctrl", KEYS["ctrl"]) em = {"actor": "V", "emission_id": "E_CRASH1", "emission_class": "verify_accept", "amount_micro": 100_000_000, "outcome": "accepted", "role": "review", "assignment_id": "A_CR", "evidence_digest": DIG["D_W2"], "attestation_id": "AT_CR"} r1 = s.append_emission(em, KEYS["root"], idem_key="k1") n_events_1 = len(s.events()) r2 = s.append_emission(em, KEYS["root"], idem_key="k1") check("retry same idem key: byte-identical receipt", r1 == r2) check("retry same idem key: no second event", len(s.events()) == n_events_1) try: s.append_emission({"actor": "V", "emission_id": "E_CRASH2", "emission_class": "verify_accept", "amount_micro": 1, "outcome": "accepted", "role": "review", "assignment_id": "A_CR", "evidence_digest": DIG["D_W2"], "attestation_id": "AT_CR"}, KEYS["root"], idem_key="k2") check("new emission_id after full spend rejected", False, "accepted!") except LS.LedgerError: check("new emission_id after full spend rejected", True) # ---- 6. partial rating_adjustment ---------------------------------------- s = fresh_store() s.issue_capability({"assignment_id": "A_P", "task_id": "T_P", "assignee": "A", "emission_class": "verify_accept", "cap_micro": 100, "review_of": "B"}, "root", KEYS["root"]) s.add_attestation({"attestation_id": "AT_P", "task_id": "T_P", "evidence_digest": "dg-p", "assignee": "A", "verdict": "accept", "ts": 20}, "ctrl", KEYS["ctrl"]) p1_seq = json.loads(s.append_emission({"actor": "A", "emission_id": "E_P1", "emission_class": "verify_accept", "amount_micro": 100, "outcome": "accepted", "role": "review", "assignment_id": "A_P", "evidence_digest": "dg-p", "attestation_id": "AT_P"}, KEYS["root"]))["seq"] s.append_reversal({"actor": "A", "reverses_event_id": p1_seq, "effect": "rating_adjustment", "basis": "misclassification", "amount_micro": 10, "assignment_id": "A_P"}, KEYS["root"]) rat = E.derived_rating(s.events()) check("partial rating_adjustment: 100-10=90", rat.get("A") == 90, f"got {rat.get('A')}") # ---- 7. conservation with signed caps ------------------------------------ s = fresh_store() for h in ("R2_plausible_incorrect", "R3_self_circle", "P1_corrective"): base_fixtures(s, h) used = ["A2001", "A2002", "A3001", "A3002", "A4001", "A4002"] caps = [c["cap_micro"] for c in V["capabilities"] if c["assignment_id"] in used] okc, dc = E.conservation(s.events(), caps) check("conservation pool=sum(caps) net+carry=pool", okc, str(dc)) # ---- 8. presence gate ---------------------------------------------------- s = fresh_store() s.issue_capability({"assignment_id": "A_PRES", "task_id": "T_PRES", "assignee": "A", "emission_class": "presence", "cap_micro": 100}, "root", KEYS["root"]) try: s.append_emission({"actor": "A", "emission_id": "E_PRES1", "emission_class": "presence", "amount_micro": 100, "outcome": "recorded", "role": "create", "assignment_id": "A_PRES", "evidence_digest": "dg-live"}, KEYS["root"]) check("presence gate (amount>0)", False) except LS.LedgerError: check("presence gate (amount>0) rejected", True) print("-" * 60) if fails: print(f"FAILURES: {len(fails)}") for f in fails: print(" FAIL", f) raise SystemExit(1) print("ALL TESTS PASSED")