#!/usr/bin/env python3 """Independent re-computation of every PCAT 702 draw from chain data. verify.py --config sale.json --pot pot.json --source explorer [--published ledger.json] verify.py --config sale.json --pot pot.json --source dir:testdata --tip 290300 Pipeline 1. find every plain NOID transfer to the founder address (coinbase, development payout, own transactions, native/contract transactions are never payments), 2. put them into canonical order (height, position in block), 3. for each payment resolve the entropy: hash of block P+1 (checked by following prev_hash links from P through the whole finality window) plus the drand quicknet round published 1800 s after that block's timestamp, 4. stop at the first payment whose entropy is not final yet (the draw is sequential, so everything behind it waits as well), 5. run draw.run_draw and (optionally) compare with a published ledger. The published parameters live in draw.py (SEED_OFFSET, FINALITY_DEPTH, BEACON, BEACON_MARGIN_S, Rules, PCAT702_FINGERPRINT). A sale config that deviates from them is refused unless --allow-nonspec is given, and the pot list must carry the published fingerprint. Data sources are exchangeable (class with block / tx / tip / address_txs / beacon); block hashes come from the data source, so for an independent anchor run the tool against your own node's explorer as well. Standard library only. Exit codes: 0 everything verified, no open items; 1 published ledger differs; 2 error / refused; 3 verified so far but payments are pending or unattributable. """ from __future__ import annotations import argparse import hashlib import json import os import re import sys import time import urllib.error import urllib.request from dataclasses import dataclass, field import draw NATIVE_AUTHORITY_PREFIX = "o1qqqqqqqqqqqqqqqqqqqqqq" # synthetic input_owner of native transactions DRAND_CHAIN = "52db9ba70e0cc0f6eaf7803dd07447a1f5477735fd3f661792ba94600c84e971" # quicknet DRAND_GENESIS = 1692803367 DRAND_PERIOD = 3 DRAND_RELAYS = ("https://api.drand.sh", "https://api2.drand.sh", "https://api3.drand.sh", "https://drand.cloudflare.com") DRAND_MIN_AGREEING = 2 # a round counts as published when at least this many relays return the same value EXPLORER_API = "https://noidexplorer.org/api/v1" class VerifyError(RuntimeError): pass def load_json(path): with open(path, encoding="utf-8") as fh: return json.load(fh) # -------------------------------------------------------------------------- # Configuration # -------------------------------------------------------------------------- CONFIG_KEYS = ("founder", "start_height", "seed_offset", "finality_depth", "beacon", "beacon_margin_s", "rules", "note") @dataclass class SaleConfig: founder: str start_height: int = 0 # payments in earlier blocks are not part of the sale seed_offset: int = draw.SEED_OFFSET # K: decisive block = payment block + K finality_depth: int = draw.FINALITY_DEPTH # consensus is_final = 18; 6 blocks safety margin beacon: str = draw.BEACON # "drand" | "none" (none = experiments only) beacon_margin_s: int = draw.BEACON_MARGIN_S # beacon round = first round at or after ts(D) + margin rules: draw.Rules = field(default_factory=draw.Rules) note: str = "" def __post_init__(self): if not isinstance(self.founder, str) or not self.founder.startswith("o1") or len(self.founder) < 20: raise VerifyError(f"config: founder address looks wrong: {self.founder!r}") for name in ("start_height", "seed_offset", "finality_depth", "beacon_margin_s"): v = getattr(self, name) if not isinstance(v, int) or isinstance(v, bool) or v < 0: raise VerifyError(f"config: {name} must be a non-negative integer") if self.seed_offset < 1: raise VerifyError("config: seed_offset must be >= 1 (the payment block itself is never the seed)") if self.beacon not in ("drand", "none"): raise VerifyError(f"config: unknown beacon mode {self.beacon!r}") @staticmethod def from_json(doc: dict) -> "SaleConfig": if not isinstance(doc, dict): raise VerifyError("config must be a JSON object") unknown = set(doc) - set(CONFIG_KEYS) if unknown: raise VerifyError(f"config: unknown keys {sorted(unknown)}") doc = dict(doc) try: rules = draw.Rules(**doc.pop("rules", {})) return SaleConfig(rules=rules, **doc) except (TypeError, draw.DrawError) as exc: raise VerifyError(f"config malformed: {exc}") from None def spec_deviations(self) -> list: """Human-readable list of parameters that differ from the published PCAT 702 rules.""" out = [] want = {"seed_offset": draw.SEED_OFFSET, "finality_depth": draw.FINALITY_DEPTH, "beacon": draw.BEACON, "beacon_margin_s": draw.BEACON_MARGIN_S} for k, v in want.items(): if getattr(self, k) != v: out.append(f"{k}={getattr(self, k)!r} (spec {v!r})") ref = draw.Rules().to_json() for k, v in self.rules.to_json().items(): if v != ref[k]: out.append(f"rules.{k}={v!r} (spec {ref[k]!r})") return out def draw_rules(self) -> draw.Rules: """Rules handed to run_draw: the beacon is mandatory exactly when the config uses one.""" if self.beacon == "none": return draw.Rules(**{**self.rules.to_json(), "require_beacon": False}) return self.rules # -------------------------------------------------------------------------- # Data sources # -------------------------------------------------------------------------- class FixtureSource: """Offline source: dicts in Explorer API format.""" def __init__(self, blocks=None, txs=None, tip=0, beacons=None, address_index=None, now=None): self.blocks = dict(blocks or {}) self.txs = dict(txs or {}) self._tip = tip self.beacons = dict(beacons or {}) self.address_index = address_index # {address: [ {height, position, txid} ]} or None (= scan) self._now = now def block(self, height): return self.blocks.get(height) def tx(self, txid): return self.txs.get(txid) def tip(self): return self._tip def now(self): return self._now if self._now is not None else int(time.time()) def address_txs(self, address): if self.address_index is None: return None return list(self.address_index.get(address, [])) def beacon(self, round_no): return self.beacons.get(round_no) @staticmethod def from_dir(path, tip, **kw): blocks, txs = {}, {} for name in sorted(os.listdir(path)): full = os.path.join(path, name) if re.fullmatch(r"block_\d+\.json", name): doc = load_json(full) blocks[doc["height"]] = doc elif re.fullmatch(r"tx_[0-9a-f]+\.json", name): doc = load_json(full) txs[doc["txid"]] = doc return FixtureSource(blocks, txs, tip, **kw) class FixtureBeaconSource: """Testnet experiments only: wraps a data source and replaces drand by a deterministic fixture beacon. value(round) = sha256(":") hex. Rounds are treated as already published (the clock check is skipped). Selected on the command line with `--beacon-source fixture:` and only together with `--allow-nonspec`; it never matches the published rules. """ def __init__(self, inner, seed): if not seed: raise VerifyError("fixture beacon needs a seed (--beacon-source fixture:)") self.inner = inner self.seed = seed def __getattr__(self, name): return getattr(self.inner, name) def now(self): return 1 << 62 def beacon(self, round_no): return hashlib.sha256(f"{self.seed}:{round_no}".encode()).hexdigest() class ExplorerSource: """Live read-only source: Explorer API (sparse, cached) plus drand relays.""" def __init__(self, base=EXPLORER_API, delay=0.25, relays=DRAND_RELAYS): self.base = base.rstrip("/") self.delay = delay self.relays = relays self.cache = {} self.calls = 0 def _get(self, url): if url in self.cache: return self.cache[url] time.sleep(self.delay) self.calls += 1 try: with urllib.request.urlopen(urllib.request.Request(url, headers={"User-Agent": "pcat-verify/2"}), timeout=20) as r: doc = json.load(r) except urllib.error.HTTPError as exc: if exc.code == 404: doc = None else: raise VerifyError(f"{url}: HTTP {exc.code}") from None except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc: raise VerifyError(f"{url}: {exc}") from None self.cache[url] = doc return doc def block(self, height): return self._get(f"{self.base}/block/height/{height}") def tx(self, txid): return self._get(f"{self.base}/tx/{txid}") def tip(self): return int(self._get(f"{self.base}/stats")["last_processed_height"]) def now(self): return int(time.time()) def address_txs(self, address): out, page = [], 1 while True: doc = self._get(f"{self.base}/address/{address}?page={page}&page_size=100") if not doc: return out rows = doc.get("transactions", []) out.extend(rows) if page * doc.get("page_size", 100) >= doc.get("total", 0) or not rows: return out page += 1 def beacon(self, round_no): """Value of a drand round, or None while fewer than DRAND_MIN_AGREEING relays have it. Any relay that answers with a different value or a malformed document is an error (fail closed).""" values = [] for relay in self.relays: try: doc = self._get(f"{relay}/{DRAND_CHAIN}/public/{round_no}") except VerifyError: continue # relay down: ask the others if doc is None: continue # round not published there yet values.append(check_drand_doc(doc, round_no)) if len(set(values)) > 1: raise VerifyError(f"drand relays disagree on round {round_no}: {sorted(set(values))}") if len(values) < DRAND_MIN_AGREEING: return None return values[0] def check_drand_doc(doc, round_no): """Returns the randomness (lowercase hex). randomness = SHA256(signature) is part of the drand spec. The BLS signature itself is NOT verified here (needs pairing crypto, not in the standard library); that is why several independent relays have to agree.""" try: if int(doc["round"]) != round_no: raise VerifyError(f"drand returned round {doc.get('round')} instead of {round_no}") sig = bytes.fromhex(str(doc["signature"])) rnd = str(doc["randomness"]).lower() except (KeyError, TypeError, ValueError, AttributeError) as exc: raise VerifyError(f"drand round {round_no}: malformed document ({exc!r})") from None if len(rnd) != 64 or hashlib.sha256(sig).hexdigest() != rnd: raise VerifyError(f"drand round {round_no}: randomness != sha256(signature)") return rnd def drand_round_for(timestamp): """First drand round whose time is >= timestamp.""" if timestamp <= DRAND_GENESIS: return 1 return -(-(timestamp - DRAND_GENESIS) // DRAND_PERIOD) + 1 def drand_round_time(round_no): return DRAND_GENESIS + (round_no - 1) * DRAND_PERIOD # -------------------------------------------------------------------------- # Step 1: payments # -------------------------------------------------------------------------- @dataclass class Scan: payments: list = field(default_factory=list) # candidate draw.Payment without entropy unattributable: list = field(default_factory=list) # to the founder, but not an attributable plain payment ignored: int = 0 def _is_plain(summary, detail): if summary.get("native_operation") or detail.get("native_operation"): return False if summary.get("trade_count") or detail.get("trade_count"): return False if summary.get("contract") or detail.get("contract"): return False return True def classify_tx(summary, detail, founder): """Return ('payment', amount) | ('unattributable', amount) | ('ignore', 0).""" if summary.get("coinbase") or summary.get("development_payout"): return "ignore", 0 sender = summary.get("input_owner") if not sender or sender == founder: return "ignore", 0 amount = sum(int(o["amount_micronoid"]) for o in detail.get("outputs", []) if o["owner"] == founder and not o.get("native")) if amount == 0: return "ignore", 0 if _is_own_trade_settlement(summary, detail, founder): return "ignore", 0 # the miner's settlement of the founder's own DEX trades (buyback change) if not _is_plain(summary, detail) or sender.startswith(NATIVE_AUTHORITY_PREFIX): return "unattributable", amount return "payment", amount def _is_own_trade_settlement(summary, detail, founder): """A native trade-batch transaction whose trades ALL belong to the founder: that is the settlement of the founder's own buyback orders (it returns the NOID change), never a payment. Only the founder's wallet can create trades owned by the founder, so this cannot be forged.""" trades = detail.get("trades") or summary.get("trades") or [] if not trades: return False return all(isinstance(t, dict) and t.get("owner") == founder for t in trades) def check_detail(detail, summary, block, txid): """The tx detail must describe the same transaction as the block's summary row.""" if detail.get("txid", txid) != txid: raise VerifyError(f"tx detail for {txid} carries txid {detail.get('txid')}") if "position" in detail and detail["position"] != summary["position"]: raise VerifyError(f"tx {txid}: position {detail['position']} in detail vs {summary['position']} in block") if "input_owner" in detail and detail["input_owner"] != summary.get("input_owner"): raise VerifyError(f"tx {txid}: input_owner differs between detail and block summary") blk = detail.get("block") or {} if blk.get("hash") and blk["hash"] != block["hash"]: raise VerifyError(f"tx {txid}: detail belongs to block {blk['hash'][:12]}.., not {block['hash'][:12]}..") if blk.get("height") is not None and int(blk["height"]) != int(block["height"]): raise VerifyError(f"tx {txid}: detail height {blk['height']} vs block {block['height']}") def scan_payments(source, cfg, to_height): scan = Scan() wanted = None listing = source.address_txs(cfg.founder) if listing is not None: wanted = {} for row in listing: if cfg.start_height <= row["height"] <= to_height: wanted.setdefault(row["height"], {})[row["txid"]] = row.get("position") heights = sorted(wanted) if wanted is not None else range(cfg.start_height, to_height + 1) for h in heights: block = source.block(h) if block is None: raise VerifyError(f"block {h} unavailable") if block.get("canonical", True) is not True: raise VerifyError(f"block {h} is not canonical") for summary in block["transactions"]: txid = summary["txid"] if wanted is not None: if txid not in wanted[h]: continue listed = wanted[h][txid] if listed is not None and listed != summary["position"]: raise VerifyError(f"position mismatch for {txid}: address list {listed} vs block {summary['position']}") elif summary.get("coinbase"): continue detail = source.tx(txid) if detail is None: raise VerifyError(f"tx {txid} unavailable") check_detail(detail, summary, block, txid) kind, amount = classify_tx(summary, detail, cfg.founder) if kind == "ignore": scan.ignored += 1 elif kind == "unattributable": scan.unattributable.append({"height": h, "position": summary["position"], "txid": txid, "amount": amount, "input_owner": summary.get("input_owner")}) else: scan.payments.append(draw.Payment(height=h, block_hash=block["hash"], txid=txid, position=summary["position"], amount=amount, sender=summary["input_owner"])) if wanted is not None: missing = set(wanted[h]) - {t["txid"] for t in block["transactions"]} if missing: raise VerifyError(f"block {h} does not contain listed tx {sorted(missing)[0]}") scan.payments.sort(key=lambda p: (p.height, p.position)) return scan # -------------------------------------------------------------------------- # Step 2/3: entropy, finality # -------------------------------------------------------------------------- def chain_link_ok(source, low, high): """Follow prev_hash from block `high` down to `low`; return the blocks low..high or None while a block is missing / not canonical. A broken link is an error.""" hashes = {} for h in range(low, high + 1): b = source.block(h) if b is None: return None if b.get("canonical", True) is not True: return None hashes[h] = b for h in range(low + 1, high + 1): if hashes[h]["prev_hash"] != hashes[h - 1]["hash"]: raise VerifyError(f"hash chain broken between {h - 1} and {h}") return [hashes[h] for h in range(low, high + 1)] def resolve_entropy(source, cfg, payment, tip): """Return (payment with entropy, None) or (None, reason) while not decidable yet.""" d_height = payment.height + cfg.seed_offset final_height = d_height + cfg.finality_depth if tip < final_height: return None, f"decisive block {d_height} not final (tip {tip}, need {final_height})" # the whole window P .. D+depth must exist and link up; the depth is not taken on trust from the tip blocks = chain_link_ok(source, payment.height, final_height) if blocks is None: return None, f"blocks {payment.height}..{final_height} unavailable or not canonical" if blocks[0]["hash"] != payment.block_hash: raise VerifyError(f"payment block hash changed for {payment.txid}") decisive = blocks[cfg.seed_offset] beacon_hex, round_no = "", 0 if cfg.beacon == "drand": round_no = drand_round_for(int(decisive["timestamp"]) + cfg.beacon_margin_s) if drand_round_time(round_no) > source.now(): return None, f"drand round {round_no} not published yet" beacon_hex = source.beacon(round_no) if beacon_hex is None: return None, f"drand round {round_no} unavailable" draw._hex32(beacon_hex, f"drand round {round_no}") return draw.Payment(**{**payment.__dict__, "seed_block_hash": decisive["hash"], "beacon": beacon_hex, "beacon_round": round_no}), None # -------------------------------------------------------------------------- # Whole run # -------------------------------------------------------------------------- @dataclass class Report: result: dict pending: list unattributable: list n_payments_seen: int n_drawn: int def verify_all(source, cfg, pot, to_height=None): tip = source.tip() to_height = tip if to_height is None else min(to_height, tip) scan = scan_payments(source, cfg, to_height) ready, pending = [], [] blocked = None for p in scan.payments: if blocked is not None: pending.append({"txid": p.txid, "height": p.height, "reason": f"behind {blocked} (draw is sequential)"}) continue resolved, reason = resolve_entropy(source, cfg, p, tip) if resolved is None: blocked = p.txid pending.append({"txid": p.txid, "height": p.height, "reason": reason}) else: ready.append(resolved) result = draw.run_draw(pot, ready, cfg.draw_rules()).to_json() return Report(result, pending, scan.unattributable, len(scan.payments), len(ready)) def compare_ledgers(computed, published): """List of human-readable differences (empty list = identical). Order matters, duplicates matter, every section is compared, and the published digest must be the digest of the published body (a copied digest does not hide an edited body).""" diffs = [] if not isinstance(published, dict): return ["published ledger is not a JSON object"] if computed["pot_fingerprint"] != published.get("pot_fingerprint"): diffs.append("pot fingerprint differs") for key in ("assignments", "refunds", "entropy"): a, b = computed.get(key, []), published.get(key, []) if not isinstance(b, list): diffs.append(f"published {key} is not a list") continue if a == b: continue sa = {json.dumps(x, sort_keys=True) for x in a} sb = {json.dumps(x, sort_keys=True) for x in b} for x in sorted(sa - sb): diffs.append(f"only recomputed in {key}: {x}") for x in sorted(sb - sa): diffs.append(f"only published in {key}: {x}") if sa == sb: diffs.append(f"{key}: same rows but different order or duplicates ({len(a)} vs {len(b)} rows)") if computed.get("remaining") != published.get("remaining"): diffs.append(f"remaining differs: {computed.get('remaining')} vs {published.get('remaining')}") extra = set(published) - set(computed) if extra: diffs.append(f"published ledger has unknown sections {sorted(extra)}") if published.get("ledger_digest") != draw.ledger_digest(published): diffs.append("published ledger_digest is not the digest of the published body") if not diffs and computed.get("ledger_digest") != published.get("ledger_digest"): diffs.append("ledger digest differs") return diffs def main(argv=None): ap = argparse.ArgumentParser(description="Re-compute all PCAT 702 draws from chain data") ap.add_argument("--config", required=True, help="sale config JSON (founder, start_height; spec parameters)") ap.add_argument("--pot", required=True, help="published pot list (canonical JSON)") ap.add_argument("--source", default="explorer", help="explorer | dir:PATH (explorer-format JSON files)") ap.add_argument("--explorer", default=EXPLORER_API, help="explorer API base (use your own node for a second anchor)") ap.add_argument("--tip", type=int, help="tip height for dir sources") ap.add_argument("--to-height", type=int) ap.add_argument("--published", help="published ledger JSON to compare against") ap.add_argument("--expect-fingerprint", default=draw.PCAT702_FINGERPRINT, metavar="HEX", help="fingerprint the pot list must have (default: the published PCAT 702 value)") ap.add_argument("--allow-nonspec", action="store_true", help="accept a config/pot that deviates from the published PCAT 702 rules (experiments only)") ap.add_argument("--beacon-source", default="drand", metavar="SRC", help="drand (default, the published rules) | fixture: (testnet experiments: deterministic " "beacon sha256(':'); requires --allow-nonspec)") args = ap.parse_args(argv) try: fixture_seed = None if args.beacon_source.startswith("fixture:"): if not args.allow_nonspec: raise VerifyError("--beacon-source fixture: is not the published beacon (use --allow-nonspec)") fixture_seed = args.beacon_source[len("fixture:"):] if not fixture_seed: raise VerifyError("fixture beacon needs a seed (--beacon-source fixture:)") elif args.beacon_source != "drand": raise VerifyError(f"unknown beacon source {args.beacon_source!r}") cfg = SaleConfig.from_json(load_json(args.config)) if fixture_seed is not None and cfg.beacon != "drand": raise VerifyError("a fixture beacon replaces drand: the sale config must use beacon 'drand'") pot = draw.PotList.from_json(load_json(args.pot)) deviations = cfg.spec_deviations() fp = pot.fingerprint() if fp != args.expect_fingerprint.lower(): deviations.append(f"pot fingerprint {fp} != expected {args.expect_fingerprint}") try: pot.check_pcat702() except draw.DrawError as exc: deviations.append(str(exc)) if fixture_seed is not None: deviations.append(f"fixture beacon {fixture_seed!r} instead of drand") if deviations: msg = "; ".join(deviations) if not args.allow_nonspec: raise VerifyError(f"NOT THE PUBLISHED PCAT 702 RULES: {msg} (use --allow-nonspec for experiments)") print(f"WARNING: running outside the published rules: {msg}", file=sys.stderr) if args.source == "explorer": source = ExplorerSource(base=args.explorer) elif args.source.startswith("dir:"): if args.tip is None: raise VerifyError("--tip is required for dir sources") source = FixtureSource.from_dir(args.source[4:], args.tip, now=int(time.time())) else: raise VerifyError(f"unknown source {args.source!r}") if fixture_seed is not None: source = FixtureBeaconSource(source, fixture_seed) report = verify_all(source, cfg, pot, args.to_height) except (VerifyError, draw.DrawError, OSError, json.JSONDecodeError, KeyError, TypeError, ValueError) as exc: print(f"ERROR: {exc}", file=sys.stderr) return 2 print(f"rules: K={cfg.seed_offset} depth={cfg.finality_depth} beacon={cfg.beacon} margin={cfg.beacon_margin_s}s " f"fingerprint={fp[:16]}.. {'(SPEC)' if not deviations else '(NON-SPEC)'}") print(f"payments found: {report.n_payments_seen}, processed: {report.n_drawn}, pending: {len(report.pending)}, " f"unattributable: {len(report.unattributable)}") for row in report.pending: print(" pending:", row) for row in report.unattributable: print(" UNATTRIBUTABLE (handle manually):", row) for a in report.result["assignments"]: print(f" {a['txid'][:12]} #{a['height']}/{a['position']} {a['sender'][:14]}.. -> {a['nft_id']} ({a['rarity']})") for r in report.result["refunds"]: print(f" refund {r['txid'][:12]} {r['sender'][:14]}.. {r['refund']} uNOID ({r['reason']})") print("remaining:", report.result["remaining"], "ledger_digest:", report.result["ledger_digest"]) if args.published: try: diffs = compare_ledgers(report.result, load_json(args.published)) except (OSError, json.JSONDecodeError) as exc: print(f"ERROR: {exc}", file=sys.stderr) return 2 if diffs: print("MISMATCH with published ledger:") for d in diffs: print(" ", d) return 1 print("OK: published ledger matches the recomputation") if report.pending or report.unattributable: print("OPEN ITEMS: pending or unattributable payments exist (see above)") return 3 return 0 if __name__ == "__main__": sys.exit(main())