251 lines
21 KiB
Python
Executable file
251 lines
21 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""The shadow runner (mission item 9, docs/analysis/mission/mission.md 2.9; 7 October 2026): on one box, every
|
|
segment a node executed is proven by BOTH proof systems behind the ProofSystem trait, SP1 (proof system 1, the pinned
|
|
shard program) and RISC Zero (proof system 2, the pinned guest), and the two are compared. Agreement means: both hosts
|
|
proved the shard, each proof verifies under its own pinned id, and the two statements (keccak256 of the committed
|
|
shard public values) are one and the same. Anything else is a disagreement, which is the alert.
|
|
|
|
Read-only on the node: segments come from `igneum_exportSegments` over a tunnel (tools/shadow/tunnel.sh); nothing is
|
|
submitted anywhere. The alert (a disagreement) is written to out/alerts.jsonl, printed as an ALERT line, posted to a
|
|
node's `igneum_proofDisagreement` when --alert-rpc names one (that node's pool then stops carrying proof records until
|
|
one third of weight signals which system to trust, igneum/exec/src/proving.rs), and handed to the Discord incident
|
|
hook when --discord names the script (dry run unless --discord-live).
|
|
|
|
One pass: the node's tip, a batch of chain blocks below it (inside the node's 2,048-block account-dump ring), one
|
|
export with the account dump at the block before the batch, one fixture per block (igneum-prove-export), and per shard
|
|
of the fixture's plan both hosts in --mode compressed, then each proof re-verified by its own host in --mode verify (the
|
|
node's verifier path), then the row. Rows: out/rows.jsonl (one per shard). Summary: out/summary.json after every row
|
|
(agree, disagree, prove-time quantiles per system, proof bytes, verify time, cycles). Stops at --segments agreeing
|
|
segments, at --max-hours, or when out/STOP exists; its pid is out/shadow.pid.
|
|
|
|
The injected bad proof (the gate's second line): --inject soundness --inject-at K replaces the SP1 statement of the
|
|
K-th shard row with another value while keeping its verifier claim, which is what a soundness bug looks like from
|
|
outside (a proof that verifies for a statement native execution did not produce); --inject bytes flips a byte of the
|
|
SP1 proof file instead, which its own verifier refuses. Either way the row disagrees, pays nothing (a record carrying
|
|
it is vetoed natively, and a stopped pool carries nothing) and raises the alert. The injected row is labelled and is
|
|
never counted as agreement.
|
|
|
|
Usage (on the shadow box):
|
|
python3 shadow.py --sp1-host <igneum-prove-host> --r0-host <igneum-prove-r0-host> --export <igneum-prove-export>
|
|
--rpc http://127.0.0.1:26790 --out /root/sp-shadow/out --segments 1000 [--batch 40] [--lag 20] [--po2 20]
|
|
[--sp1-env "HOME=/opt/igneum-floor/home SP1_PROVER=cuda SP1_GPU_ELEMENT_THRESHOLD=67108864"]
|
|
[--r0-env "RISC0_PROVER=local"] [--parallel] [--alert-rpc http://127.0.0.1:29390] [--discord <hooks.mjs>]
|
|
[--inject soundness|bytes --inject-at K] [--max-hours 11] [--keep 5] [--self-test]
|
|
"""
|
|
import argparse, json, os, shutil, subprocess, sys, time, urllib.request, datetime, statistics, signal
|
|
|
|
def now(): return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
|
|
def say(line, out=None):
|
|
print(line, flush=True)
|
|
if out:
|
|
with open(os.path.join(out, "shadow.log"), "a") as f: f.write(line + "\n")
|
|
|
|
def rpc(url, method, params=None, timeout=120):
|
|
body = json.dumps({"jsonrpc": "2.0", "id": 1, "method": method, "params": params or []}).encode()
|
|
req = urllib.request.Request(url, data=body, headers={"content-type": "application/json"})
|
|
with urllib.request.urlopen(req, timeout=timeout) as r:
|
|
d = json.loads(r.read())
|
|
if "error" in d: raise RuntimeError(f"{method}: {d['error']}")
|
|
return d["result"]
|
|
|
|
def run(cmd, env_extra, log_path, timeout):
|
|
"""Runs a host; returns (rc, last RESULT line or last stderr line)."""
|
|
env = dict(os.environ)
|
|
for kv in (env_extra or "").split():
|
|
k, _, v = kv.partition("="); env[k] = v
|
|
with open(log_path, "w") as log:
|
|
try:
|
|
p = subprocess.run(cmd, env=env, stdout=log, stderr=subprocess.STDOUT, timeout=timeout)
|
|
rc = p.returncode
|
|
except subprocess.TimeoutExpired:
|
|
rc = 124
|
|
last = ""
|
|
try:
|
|
lines = open(log_path, errors="replace").read().splitlines()
|
|
res = [l for l in lines if l.startswith("RESULT")]
|
|
last = (res[-1] if res else (lines[-1] if lines else ""))[:400]
|
|
if rc == 124: last = f"timed out after {timeout} s; " + last
|
|
except OSError:
|
|
pass
|
|
return rc, last
|
|
|
|
def prove_one(label, host, env_extra, fixture, shard, out_json, log_path, extra, timeout):
|
|
"""One host in --mode compressed, then its own --mode verify on the proof it wrote (the node's path)."""
|
|
cmd = [host, fixture, "--mode", "compressed", "--shard", str(shard), "--prover", "0xCAfc6e74000000000000000000000000000000c2", "--out", out_json] + extra
|
|
t0 = time.time(); rc, last = run(cmd, env_extra, log_path, timeout); wall = round(time.time() - t0, 2)
|
|
r = {"system": label, "ok": False, "wall_s": wall, "last": last}
|
|
if rc != 0 or not os.path.exists(out_json):
|
|
r["error"] = f"prove exit {rc}: {last}"; return r
|
|
res = json.load(open(out_json))
|
|
r.update({"statement": res.get("statement"), "prove_s": res.get("compressed_prove_seconds"), "proof_bytes": res.get("compressed_proof_bytes"),
|
|
"cycles": res.get("cycles"), "proof_file": res.get("proof_file"), "program_id": res.get("image_id") or res.get("shard_program_id"), "proof_sha256": res.get("proof_sha256")})
|
|
t0 = time.time(); vrc, vlast = run([host, "--mode", "verify", "--proof", r["proof_file"], "--statement", r["statement"]], env_extra, log_path + ".verify", 300)
|
|
r["verify_s"] = round(time.time() - t0, 3); r["verified"] = vrc == 0; r["verify_last"] = vlast
|
|
r["ok"] = r["verified"] and bool(r["statement"])
|
|
if not r["verified"]: r["error"] = f"its own verifier refuses the proof: {vlast}"
|
|
return r
|
|
|
|
def quantiles(xs):
|
|
xs = sorted(x for x in xs if isinstance(x, (int, float)))
|
|
if not xs: return None
|
|
q = lambda p: xs[min(len(xs) - 1, int(p * (len(xs) - 1)))]
|
|
return {"n": len(xs), "min": xs[0], "median": statistics.median(xs), "p90": q(0.9), "max": xs[-1], "mean": round(sum(xs) / len(xs), 3)}
|
|
|
|
def summarize(rows, started, args, note=""):
|
|
agree = [r for r in rows if r["agree"]]
|
|
dis = [r for r in rows if not r["agree"]]
|
|
segs_agree = sorted({r["number"] for r in agree if not r.get("injected")})
|
|
# a segment agrees when every shard of it agreed and none disagreed
|
|
bad_segs = {r["number"] for r in dis}
|
|
segs_agree = [n for n in segs_agree if n not in bad_segs]
|
|
s = {"at": now(), "started": started, "elapsed_s": round(time.time() - time.mktime(time.strptime(started, "%Y-%m-%dT%H:%M:%SZ")) + time.timezone, 1),
|
|
"rows": len(rows), "shards_agree": len(agree), "shards_disagree": len(dis), "segments_agree": len(segs_agree), "segments_disagree": len(bad_segs),
|
|
"injected": [r["number"] for r in rows if r.get("injected")], "target_segments": args.segments, "note": note,
|
|
"sp1": {"prove_s": quantiles([r["sp1"].get("prove_s") for r in rows]), "wall_s": quantiles([r["sp1"].get("wall_s") for r in rows]), "proof_bytes": quantiles([r["sp1"].get("proof_bytes") for r in rows]), "verify_s": quantiles([r["sp1"].get("verify_s") for r in rows]), "cycles": quantiles([r["sp1"].get("cycles") for r in rows]), "program_id": next((r["sp1"].get("program_id") for r in rows if r["sp1"].get("program_id")), None)},
|
|
"r0": {"prove_s": quantiles([r["r0"].get("prove_s") for r in rows]), "wall_s": quantiles([r["r0"].get("wall_s") for r in rows]), "proof_bytes": quantiles([r["r0"].get("proof_bytes") for r in rows]), "verify_s": quantiles([r["r0"].get("verify_s") for r in rows]), "cycles": quantiles([r["r0"].get("cycles") for r in rows]), "program_id": next((r["r0"].get("program_id") for r in rows if r["r0"].get("program_id")), None)},
|
|
"txs": quantiles([r.get("txs") for r in rows]), "shards_per_segment": quantiles([r.get("shards") for r in rows]),
|
|
"first_number": min((r["number"] for r in rows), default=None), "last_number": max((r["number"] for r in rows), default=None)}
|
|
json.dump(s, open(os.path.join(args.out, "summary.json"), "w"), indent=1)
|
|
return s
|
|
|
|
def alert(row, args):
|
|
line = f"ALERT {now()} proof systems DISAGREE on chain block {row['number']} shard {row['shard']}: {row['reason']} (sp1 {row['sp1'].get('statement')} r0 {row['r0'].get('statement')}){' [INJECTED ' + row['injected'] + ']' if row.get('injected') else ''}"
|
|
say(line, args.out)
|
|
with open(os.path.join(args.out, "alerts.jsonl"), "a") as f: f.write(json.dumps({"at": now(), "line": line, "row": row}) + "\n")
|
|
json.dump({"at": now(), "line": line, "row": row}, open(os.path.join(args.out, "alert.json"), "w"), indent=1)
|
|
if args.alert_rpc:
|
|
try:
|
|
r = rpc(args.alert_rpc, "igneum_proofDisagreement", [{"number": row["number"], "blockHash": row.get("hash"), "shard": row["shard"], "systems": {"1": row["sp1"].get("statement"), "2": row["r0"].get("statement")}, "reason": row["reason"], "source": "tools/shadow/shadow.py", "injected": row.get("injected")}], timeout=30)
|
|
say(f"RESULT alert_rpc {now()} {args.alert_rpc}: {json.dumps(r)[:300]}", args.out)
|
|
except Exception as e:
|
|
say(f"RESULT alert_rpc {now()} FAILED: {e}", args.out)
|
|
if args.discord and os.path.exists(args.discord):
|
|
cmd = ["node", args.discord, "incident", "open", "--what", f"Proof systems disagree on chain block {row['number']} shard {row['shard']}: {row['reason']}", "--affected", "the proving pool: no proof record is carried until one third of weight signals which proof system to trust; full nodes keep the native-execution veto", "--doing", "the shadow runner's row is in out/alerts.jsonl; the disagreeing proof is kept for the review", "--id", f"proof-disagreement-{row['number']}-{row['shard']}"] + (["--live"] if args.discord_live else [])
|
|
try:
|
|
p = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
|
|
say(f"RESULT discord {now()} exit {p.returncode}: {(p.stdout or p.stderr).strip()[:300]}", args.out)
|
|
except Exception as e:
|
|
say(f"RESULT discord {now()} FAILED: {e}", args.out)
|
|
|
|
def self_test():
|
|
"""The gate's two shapes on synthetic rows: an agreeing row counts, an injected disagreeing row raises and never counts."""
|
|
class A: pass
|
|
a = A(); a.out = "/tmp/sp-shadow-selftest"; a.segments = 2; a.alert_rpc = None; a.discord = None; a.discord_live = False
|
|
shutil.rmtree(a.out, ignore_errors=True); os.makedirs(a.out)
|
|
started = now()
|
|
good = {"number": 1, "shard": 0, "sp1": {"statement": "0xaa", "verified": True, "ok": True, "prove_s": 1.0, "proof_bytes": 10, "verify_s": 0.1, "cycles": 5}, "r0": {"statement": "0xaa", "verified": True, "ok": True, "prove_s": 2.0, "proof_bytes": 5, "verify_s": 0.2, "cycles": 5}, "agree": True, "reason": "", "txs": 0, "shards": 1}
|
|
bad = dict(good); bad = json.loads(json.dumps(good)); bad["number"] = 2; bad["sp1"]["statement"] = "0xab"; bad["agree"] = False; bad["reason"] = "statements differ"; bad["injected"] = "soundness"
|
|
alert(bad, a)
|
|
s = summarize([good, bad], started, a)
|
|
assert s["segments_agree"] == 1 and s["segments_disagree"] == 1 and s["injected"] == [2], s
|
|
assert os.path.exists(os.path.join(a.out, "alert.json"))
|
|
assert json.loads(open(os.path.join(a.out, "alerts.jsonl")).readline())["row"]["injected"] == "soundness"
|
|
print("self-test: an agreeing row counts, an injected disagreeing row raises the alert and never counts; summary", {k: s[k] for k in ("segments_agree", "segments_disagree", "injected")})
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--sp1-host"); ap.add_argument("--r0-host"); ap.add_argument("--export"); ap.add_argument("--rpc", default="http://127.0.0.1:26790")
|
|
ap.add_argument("--out", default="/root/sp-shadow/out"); ap.add_argument("--segments", type=int, default=1000); ap.add_argument("--batch", type=int, default=40); ap.add_argument("--lag", type=int, default=20)
|
|
ap.add_argument("--po2", type=int, default=20); ap.add_argument("--sp1-env", default="SP1_PROVER=cuda RUST_LOG=off"); ap.add_argument("--r0-env", default="RISC0_PROVER=local RUST_LOG=off")
|
|
ap.add_argument("--parallel", action="store_true", help="run the two hosts at once (two GPU residents; fits a 24 GB card at the 2^26 SP1 threshold and po2 19)")
|
|
ap.add_argument("--alert-rpc"); ap.add_argument("--discord"); ap.add_argument("--discord-live", action="store_true")
|
|
ap.add_argument("--inject", choices=["soundness", "bytes"]); ap.add_argument("--inject-at", type=int, default=1)
|
|
ap.add_argument("--max-hours", type=float, default=11.0); ap.add_argument("--keep", type=int, default=5); ap.add_argument("--timeout", type=int, default=900)
|
|
ap.add_argument("--self-test", action="store_true")
|
|
args = ap.parse_args()
|
|
if args.self_test: return self_test()
|
|
for k in ("sp1_host", "r0_host", "export"):
|
|
if not getattr(args, k) or not os.access(getattr(args, k), os.X_OK): sys.exit(f"--{k.replace('_', '-')} must name an executable")
|
|
os.makedirs(args.out, exist_ok=True)
|
|
open(os.path.join(args.out, "shadow.pid"), "w").write(str(os.getpid()))
|
|
started = now(); t_start = time.time()
|
|
rows = []
|
|
rows_path = os.path.join(args.out, "rows.jsonl")
|
|
if os.path.exists(rows_path):
|
|
rows = [json.loads(l) for l in open(rows_path) if l.strip()]
|
|
done = {(r["number"], r["shard"]) for r in rows}
|
|
ids = {}
|
|
for label, host in (("sp1", args.sp1_host), ("r0", args.r0_host)):
|
|
rc, last = run([host, "--mode", "id"], "", os.path.join(args.out, f"id-{label}.log"), 60)
|
|
ids[label] = last; say(f"RESULT id {label}: {last}", args.out)
|
|
say(f"RESULT start {started} rpc {args.rpc} target {args.segments} segments, batch {args.batch}, lag {args.lag}, po2 {args.po2}, parallel {args.parallel}, inject {args.inject} at {args.inject_at}, resumed rows {len(rows)}", args.out)
|
|
shard_counter = 0
|
|
stop_reason = ""
|
|
while True:
|
|
s = summarize(rows, started, args)
|
|
if s["segments_agree"] >= args.segments: stop_reason = f"target reached: {s['segments_agree']} segments agree"; break
|
|
if time.time() - t_start > args.max_hours * 3600: stop_reason = f"max hours {args.max_hours} reached"; break
|
|
if os.path.exists(os.path.join(args.out, "STOP")): stop_reason = "STOP file"; break
|
|
try:
|
|
tip = int(rpc(args.rpc, "eth_blockNumber"), 16)
|
|
except Exception as e:
|
|
say(f"RESULT rpc {now()} tip unreachable: {e}; waiting 30 s", args.out); time.sleep(30); continue
|
|
last = tip - args.lag; first = last - args.batch + 1
|
|
# the batch below the last proven block when the chain has not moved past it (never a block twice)
|
|
if rows and first <= max(r["number"] for r in rows): first = max(r["number"] for r in rows) + 1
|
|
if first > last: say(f"RESULT wait {now()} tip {tip}: no new blocks under the lag; 20 s", args.out); time.sleep(20); continue
|
|
d = os.path.join(args.out, f"batch-{first}-{last}"); os.makedirs(d, exist_ok=True)
|
|
t0 = time.time()
|
|
try:
|
|
export = rpc(args.rpc, "igneum_exportSegments", [hex(first - 1), hex(last)], timeout=600)
|
|
except Exception as e:
|
|
say(f"RESULT export {now()} {first - 1}..{last} FAILED: {e}; 30 s", args.out); time.sleep(30); continue
|
|
json.dump(export, open(os.path.join(d, "export.json"), "w"))
|
|
say(f"RESULT export {now()} {first - 1}..{last} ({len(export.get('segments', []))} segments, {len(export.get('preState') or [])} accounts in the dump, {round(time.time() - t0, 1)} s)", args.out)
|
|
for n in range(first, last + 1):
|
|
if (n, 0) in done: continue
|
|
fx = os.path.join(d, f"block-{n}.json")
|
|
rc, lastl = run([args.export, os.path.join(d, "export.json"), str(n), fx, "--source", "shadow runner, Devnet 2"], "", os.path.join(d, f"export-{n}.log"), 600)
|
|
if rc != 0 or not os.path.exists(fx):
|
|
say(f"RESULT fixture {now()} block {n} FAILED (exit {rc}): {lastl}", args.out); continue
|
|
fixture = json.load(open(fx))
|
|
shards = len(fixture["plan"]["shards"]); txs = sum(len(b["txs"]) for b in fixture["block"]["blocks"]); bhash = fixture["block"]["env"]["hash"]
|
|
for i in range(shards):
|
|
if (n, i) in done: continue
|
|
shard_counter += 1
|
|
sp1_json = os.path.join(d, f"sp1-{n}-{i}.json"); r0_json = os.path.join(d, f"r0-{n}-{i}.json")
|
|
if args.parallel:
|
|
import concurrent.futures
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as ex:
|
|
fa = ex.submit(prove_one, "sp1", args.sp1_host, args.sp1_env, fx, i, sp1_json, os.path.join(d, f"sp1-{n}-{i}.log"), [], args.timeout)
|
|
fb = ex.submit(prove_one, "r0", args.r0_host, args.r0_env, fx, i, r0_json, os.path.join(d, f"r0-{n}-{i}.log"), ["--po2", str(args.po2)], args.timeout)
|
|
a, b = fa.result(), fb.result()
|
|
else:
|
|
a = prove_one("sp1", args.sp1_host, args.sp1_env, fx, i, sp1_json, os.path.join(d, f"sp1-{n}-{i}.log"), [], args.timeout)
|
|
b = prove_one("r0", args.r0_host, args.r0_env, fx, i, r0_json, os.path.join(d, f"r0-{n}-{i}.log"), ["--po2", str(args.po2)], args.timeout)
|
|
injected = None
|
|
if args.inject and shard_counter == args.inject_at:
|
|
injected = args.inject
|
|
if args.inject == "soundness" and a.get("statement"):
|
|
# a proof that verifies for a statement native execution did not produce: the last nibble changed
|
|
st = a["statement"]; a["statement_original"] = st; a["statement"] = st[:-1] + ("0" if st[-1] != "0" else "1"); a["injected"] = "soundness-bug simulation: the statement replaced, the verifier's claim kept"
|
|
elif args.inject == "bytes" and a.get("proof_file"):
|
|
pf = a["proof_file"]; data = bytearray(open(pf, "rb").read()); mid = len(data) // 2; data[mid] ^= 0x01; open(pf, "wb").write(data)
|
|
vrc, vlast = run([args.sp1_host, "--mode", "verify", "--proof", pf, "--statement", a["statement"]], args.sp1_env, os.path.join(d, f"sp1-{n}-{i}.log.verify-injected"), 300)
|
|
a["verified"] = vrc == 0; a["ok"] = a["verified"]; a["verify_last"] = vlast; a["injected"] = "a byte of the proof flipped; its own verifier's answer recorded"
|
|
agree = bool(a.get("ok") and b.get("ok") and a.get("statement") == b.get("statement"))
|
|
reason = "" if agree else ("sp1: " + a.get("error", "") if not a.get("ok") else "") + (" r0: " + b.get("error", "") if not b.get("ok") else "") + (" statements differ" if a.get("ok") and b.get("ok") and a.get("statement") != b.get("statement") else "")
|
|
row = {"at": now(), "number": n, "hash": bhash, "shard": i, "shards": shards, "txs": txs, "sp1": a, "r0": b, "agree": agree, "reason": reason.strip(), "injected": injected}
|
|
rows.append(row); done.add((n, i))
|
|
with open(rows_path, "a") as f: f.write(json.dumps(row) + "\n")
|
|
say(f"RESULT row {row['at']} block {n} shard {i}/{shards} txs {txs}: sp1 {a.get('prove_s')} s {a.get('proof_bytes')} B verify {a.get('verify_s')} s {'ok' if a.get('ok') else 'FAIL'}; r0 {b.get('prove_s')} s {b.get('proof_bytes')} B verify {b.get('verify_s')} s {'ok' if b.get('ok') else 'FAIL'}; {'AGREE' if agree else 'DISAGREE ' + reason}{' INJECTED' if injected else ''}", args.out)
|
|
if not agree: alert(row, args)
|
|
# disk: proofs and fixtures of the first --keep rows and of every disagreement are kept, the rest go
|
|
keep = len(rows) <= args.keep or not agree
|
|
if not keep:
|
|
for p in (a.get("proof_file"), b.get("proof_file")):
|
|
if p and os.path.exists(p): os.remove(p)
|
|
s = summarize(rows, started, args)
|
|
if s["segments_agree"] >= args.segments or os.path.exists(os.path.join(args.out, "STOP")): break
|
|
if not (len(rows) <= args.keep) and os.path.exists(fx) and all(r["agree"] for r in rows if r["number"] == n): os.remove(fx)
|
|
s = summarize(rows, started, args)
|
|
if s["segments_agree"] >= args.segments or os.path.exists(os.path.join(args.out, "STOP")) or time.time() - t_start > args.max_hours * 3600: break
|
|
if os.path.exists(os.path.join(d, "export.json")) and len(rows) > args.keep: os.remove(os.path.join(d, "export.json"))
|
|
s = summarize(rows, started, args, stop_reason)
|
|
say(f"RESULT end {now()} {stop_reason}: rows {s['rows']}, segments agree {s['segments_agree']}, disagree {s['segments_disagree']}, injected {s['injected']}; sp1 prove median {s['sp1']['prove_s'] and s['sp1']['prove_s']['median']} s p90 {s['sp1']['prove_s'] and s['sp1']['prove_s']['p90']} s; r0 median {s['r0']['prove_s'] and s['r0']['prove_s']['median']} s p90 {s['r0']['prove_s'] and s['r0']['prove_s']['p90']} s", args.out)
|
|
try: os.remove(os.path.join(args.out, "shadow.pid"))
|
|
except OSError: pass
|
|
|
|
if __name__ == "__main__":
|
|
main()
|