""" SNIN Emission Paper Pilot — append-only SIGNED event ledger v0.4.3 (FROZEN). Gate #25132 (astranaut01 accepting nadir-codex #25098 as the operationalization): - assignments and attestations are append-only Ed25519-SIGNED events - issuer authority is verified: a valid signature proves key ownership, the frozen authority set proves the RIGHT to issue - assignment budget is consumed atomically inside append: spent_sum(assignment_id) + amount <= cap_micro UNIQUE(assignment_id, emission_id) - attestation binds (task_id, evidence_digest, instrument_version, assignee, verdict) and is consumed exactly once - append itself enforces classifier verdict (flag -> verify_accept rejected) and conservation - idempotent append: retry with the same idempotency key returns the ORIGINAL receipt byte-for-byte and writes nothing (crash points #25132) Pure python, no network. Keys are deterministic from seeds (test keyring); in production the private keys live outside the ledger. """ import hashlib import json import jsonschema from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey, Ed25519PublicKey from cryptography.exceptions import InvalidSignature import emission as E WINDOW_POLICY_VERSION = "0.4.3" REVIEW_CLASSES = {"verify_accept", "verify_reject", "recheck_assigned", "recheck_confirm", "recheck_overturn"} with open("schema_v0.4.3.json") as f: SCHEMA = json.load(f) class LedgerError(Exception): pass def _sig_payload(obj): return {k: v for k, v in obj.items() if k not in ("sig", "sig_key", "seq")} def canonical(obj): """Canonical JSON for signing: content only (seq is journal position, assigned after signing; signature never covers it).""" return json.dumps(_sig_payload(obj), sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode() def deterministic_key(seed: str) -> Ed25519PrivateKey: """Test keyring: deterministic Ed25519 keys from seeds. Private keys are NEVER published; only key_ids appear in vectors.""" d = hashlib.sha256(("snin-v0.4.3:" + seed).encode()).digest() return Ed25519PrivateKey.from_private_bytes(d) class LedgerStore: def __init__(self, policy=None): pol = policy or { "policy_version": WINDOW_POLICY_VERSION, "authority": {"root": {"issue": True, "attest": False}, "ctrl": {"issue": False, "attest": True}}, "revoked": {}, } self.policy_version = pol["policy_version"] self.authority = pol["authority"] self.revoked = pol.get("revoked", {}) self._events = [] self._by_seq = {} self._spent = {} # assignment_id -> total spent self._consumed_atts = {} # attestation_id -> emission seq self._receipts = {} # idempotency_key -> (seq, receipt_bytes) self._verify_all = True # ---- signature helpers ---------------------------------------------- def _check_authority(self, key_id, right, seq): auth = self.authority.get(key_id) if not auth or not auth.get(right, False): raise LedgerError(f"key {key_id} has no authority to {right} (authority set is frozen for {self.policy_version})") eff = self.revoked.get(key_id) if eff is not None and seq >= eff: raise LedgerError(f"key {key_id} revoked effective seq {eff}") def sign(self, event, key_id, priv): """Sign an event with the issuer/controller private key. Called BEFORE append assigns seq: the signature covers content, not journal position.""" sig = priv.sign(canonical(event)) event["sig"] = sig.hex() event["sig_key"] = key_id def verify_signature(self, event, pub): try: pub.verify(bytes.fromhex(event["sig"]), canonical(event)) return True except (InvalidSignature, KeyError, ValueError): return False # ---- schema + seq + policy gate ------------------------------------- def _base_gate(self, ev): ev["policy_version"] = ev.get("policy_version", self.policy_version) if ev["policy_version"] != self.policy_version: raise LedgerError(f"policy_version {ev['policy_version']} != frozen {self.policy_version}") last = max([x["seq"] for x in self._events] or [0]) if ev.get("seq") is None: ev["seq"] = last + 1 if ev["seq"] != last + 1: raise LedgerError(f"non-monotonic seq: got {ev['seq']}, expected {last + 1}") try: jsonschema.validate(ev, SCHEMA) except jsonschema.ValidationError as e: raise LedgerError(f"schema violation: {e.message}") return ev def _append_event(self, ev, operator_priv=None): ev = self._base_gate(ev) et = ev["event_type"] if et == "assignment_issued": self._check_authority(ev["issuer_key_id"], "issue", ev["seq"]) self._check_assignment(ev) elif et == "attestation_signed": self._check_authority(ev["controller_key_id"], "attest", ev["seq"]) self._check_attestation(ev) elif et == "emission": if operator_priv is None: raise LedgerError("operator key required to sign emission") self._check_emission(ev) elif et == "reversal": if operator_priv is None: raise LedgerError("operator key required to sign reversal") self._check_reversal(ev) self._events.append(ev) self._by_seq[ev["seq"]] = ev if et == "emission" and not ev.get("is_fixture"): aid = ev["assignment_id"] self._spent[aid] = self._spent.get(aid, 0) + int(ev["amount_micro"]) if ev.get("attestation_id"): self._consumed_atts[ev["attestation_id"]] = ev["seq"] return ev["seq"] # ---- semantic checks ------------------------------------------------- def _assignment(self, assignment_id): for ev in self._events: if ev.get("event_type") == "assignment_issued" and ev.get("assignment_id") == assignment_id: return ev raise LedgerError(f"assignment {assignment_id} not issued (capability must be issued BEFORE use)") def _check_assignment(self, ev): """Capability issuance is itself a signed event; nothing mutable is kept.""" if ev.get("is_fixture"): return if ev["emission_class"] in REVIEW_CLASSES and not ev.get("review_of"): raise LedgerError(f"review assignment {ev['assignment_id']} missing review_of") if ev["emission_class"] in REVIEW_CLASSES and ev["assignee"] == ev.get("review_of"): raise LedgerError(f"self-award: assignee == review_of for {ev['assignment_id']}") # issuer must not be the assignee for review tasks (a reviewer never # issues its own capability) — operator issues, assignee performs. if ev["emission_class"] in REVIEW_CLASSES and ev["assignee"] == ev["issuer_key_id"]: raise LedgerError("self-issued capability: issuer == assignee, forbidden") def _check_attestation(self, ev): if ev["controller_key_id"] == ev["assignee"]: raise LedgerError("attestation controller must differ from assignee (independence)") def _attestation(self, attestation_id): for ev in self._events: if ev.get("event_type") == "attestation_signed" and ev.get("attestation_id") == attestation_id: return ev raise LedgerError(f"attestation {attestation_id} does not exist") def _check_emission(self, ev): if ev.get("is_fixture") and int(ev.get("amount_micro", 0)) != 0: raise LedgerError("fixture gate: is_fixture=true => amount_micro MUST be 0") if ev["emission_class"] in ("presence", "task_receipt") and int(ev.get("amount_micro", 0)) != 0: raise LedgerError(f"{ev['emission_class']} never emits: amount MUST be 0") # capability: pre-issued signed assignment ass = self._assignment(ev["assignment_id"]) if ass["emission_class"] != ev["emission_class"]: raise LedgerError(f"assignment class {ass['emission_class']} != row class {ev['emission_class']}") if ass["assignee"] != ev["actor"]: raise LedgerError(f"assignment assignee {ass['assignee']} != row actor {ev['actor']}") # atomic budget consumption: spent_sum + amount <= cap spent = self._spent.get(ev["assignment_id"], 0) cap = int(ass["cap_micro"]) if spent + int(ev["amount_micro"]) > cap: raise LedgerError(f"budget exhausted: spent {spent} + {ev['amount_micro']} > cap {cap} " f"(consumption is atomic inside append)") # UNIQUE(assignment_id, emission_id) for old in self._events: if old.get("event_type") == "emission" and old.get("emission_id") == ev["emission_id"]: raise LedgerError(f"duplicate emission_id {ev['emission_id']}") # attestation binding + verdict gate if ev["emission_class"] in REVIEW_CLASSES: att = self._attestation(ev.get("attestation_id") or "") if att["task_id"] != ass["task_id"]: raise LedgerError("attestation task_id does not match assignment task_id (replay on another task)") if att["evidence_digest"] != ev.get("evidence_digest"): raise LedgerError("attestation evidence_digest does not match row (forged evidence)") if att["assignee"] != ev["actor"]: raise LedgerError("attestation assignee does not match row actor") required = E.required_verdict(ev["emission_class"]) if att["verdict"] != required: raise LedgerError(f"verdict gate: attestation verdict {att['verdict']} != required {required} " f"for {ev['emission_class']}") if att["attestation_id"] in self._consumed_atts: raise LedgerError(f"attestation {att['attestation_id']} already consumed by emission " f"seq {self._consumed_atts[att['attestation_id']]}") if att["controller_key_id"] == ev["actor"]: raise LedgerError("attestation controller == row actor, not independent") def _check_reversal(self, ev): target = self._by_seq.get(ev["reverses_event_id"]) if target is None: raise LedgerError(f"reversal target seq {ev['reverses_event_id']} does not exist") if target.get("event_type") != "emission": raise LedgerError("reversal target must be an emission event") if target.get("is_fixture"): raise LedgerError("reversal target must not be a fixture row") if ev.get("actor") != target.get("actor"): raise LedgerError(f"reversal actor {ev.get('actor')} != target actor {target.get('actor')}") basis, effect = ev.get("basis"), ev.get("effect") if effect not in E.REVERSAL_BASIS_EFFECTS.get(basis, set()): raise LedgerError(f"effect {effect} not permitted for basis {basis}") for r in self._events: if r.get("event_type") == "reversal" and \ r.get("reverses_event_id") == ev["reverses_event_id"] and \ r.get("effect") == effect: raise LedgerError(f"duplicate reversal ({ev['reverses_event_id']}, {effect})") q = E.quoted_emission(target) cap = q.get("balance", 0) if effect == "balance_clawback" else q.get("rating", 0) if int(ev.get("amount_micro", 0)) > cap: raise LedgerError(f"over-{effect}: {ev.get('amount_micro')} > target credit {cap}") # ---- public append API ---------------------------------------------- def issue_capability(self, assignment, issuer_key_id, issuer_priv, idem_key=None): ev = {"event_type": "assignment_issued", "actor": issuer_key_id, "ts": assignment.get("ts", 0), "policy_version": self.policy_version, "assignment_id": assignment["assignment_id"], "task_id": assignment["task_id"], "assignee": assignment["assignee"], "emission_class": assignment["emission_class"], "cap_micro": int(assignment["cap_micro"]), "issuer_key_id": issuer_key_id, "review_of": assignment.get("review_of")} self.sign(ev, "root", issuer_priv) self.sign(ev, issuer_key_id, issuer_priv) return self._append_signed(ev, idem_key) def add_attestation(self, att, controller_key_id, controller_priv, idem_key=None): ev = {"event_type": "attestation_signed", "actor": controller_key_id, "ts": att.get("ts", 0), "policy_version": self.policy_version, "attestation_id": att["attestation_id"], "task_id": att["task_id"], "evidence_digest": att["evidence_digest"], "instrument_version": att.get("instrument_version", "n-classifier-0.4.3"), "assignee": att["assignee"], "verdict": att["verdict"], "controller_key_id": controller_key_id, "nonce": att.get("nonce", "")} self.sign(ev, controller_key_id, controller_priv) return self._append_signed(ev, idem_key) def append_emission(self, ev, operator_priv, idem_key=None): full = {"event_type": "emission", "actor": ev["actor"], "ts": ev.get("ts", 0), "policy_version": self.policy_version, "emission_id": ev["emission_id"], "emission_class": ev["emission_class"], "amount_micro": int(ev["amount_micro"]), "outcome": ev["outcome"], "role": ev.get("role", "review"), "assignment_id": ev["assignment_id"], "evidence_digest": ev["evidence_digest"], "attestation_id": ev.get("attestation_id"), "is_fixture": ev.get("is_fixture", False)} self.sign(full, "root", operator_priv) return self._append_signed(full, idem_key, operator_priv) def append_reversal(self, ev, operator_priv, idem_key=None): full = {"event_type": "reversal", "actor": ev["actor"], "ts": ev.get("ts", 0), "policy_version": self.policy_version, "reverses_event_id": ev["reverses_event_id"], "effect": ev["effect"], "basis": ev["basis"], "amount_micro": int(ev["amount_micro"]), "assignment_id": ev.get("assignment_id", ""), "is_fixture": ev.get("is_fixture", False)} self.sign(full, "root", operator_priv) return self._append_signed(full, idem_key, operator_priv) def _append_signed(self, ev, idem_key=None, operator_priv=None): """Idempotent append: same idempotency key -> original receipt, byte-for-byte.""" if idem_key is not None and idem_key in self._receipts: return self._receipts[idem_key] seq = self._append_event(ev, operator_priv) receipt = json.dumps({"ok": True, "seq": seq, "event_type": ev["event_type"], "id": ev.get("emission_id") or ev.get("assignment_id") or ev.get("attestation_id") or "", "policy_version": self.policy_version}, sort_keys=True) if idem_key is not None: self._receipts[idem_key] = receipt return receipt # ---- introspection -------------------------------------------------- def events(self): return [json.loads(json.dumps(e)) for e in self._events] def verify_all(self, pubs): """Re-verify every stored signature against public keys.""" for ev in self._events: pk = pubs.get(ev.get("sig_key")) if pk is None: return False, f"seq {ev['seq']}: no pubkey for {ev.get('sig_key')}" if not self.verify_signature(ev, pk): return False, f"seq {ev['seq']}: signature invalid (event mutated?)" return True, "all signatures valid" def receipts(self): return json.loads(json.dumps(self._receipts))