aere-research/pq-stark/challenger_reference.py
Aere Network 4a0b48588c Initial public release
Aere Network public source. Everything here can be checked against the live
chain (chain id 2800, https://rpc.aere.network).

Scope note, stated up front rather than buried: consensus on chain 2800 is
classical secp256k1 ECDSA QBFT. The post-quantum work in this repository is at
the signature, precompile, account and transport layers. Nothing here makes the
consensus post-quantum, and no document in it should be read as claiming so.
2026-07-20 01:02:30 +03:00

278 lines
12 KiB
Python

#!/usr/bin/env python3
# Independent reference for the Fiat-Shamir DUPLEX CHALLENGER (component (f)) of the PQ STARK-verify
# precompile 0x0AE8: the Poseidon2 duplex sponge over BabyBear that SP1/Plonky3 use to derive the FRI
# folding challenges (betas), the query indices, and the grinding proof-of-work acceptance from the
# proof data. See docs/AERE-STARK-VERIFIER-PORT-SPEC.md section 7.
#
# ============================ HONEST SCOPE (read first) ============================
# This module implements DuplexChallenger<BabyBear, Perm, WIDTH=16, RATE=8> and its GrindingChallenger
# check_witness EXACTLY as the pinned p3-challenger 0.4.3-succinct does, over the CONFIRMED Poseidon2
# permutation (component (b)). It is CONFORMANCE-CONFIRMED against ground truth emitted by executing the
# pinned crates:
# - a fully-scripted observe/sample transcript ("duplex_kat"): observe base elements + a digest,
# sample base + F_{p^4} ext elements, sample_bits, and a check_witness accept/reject table; this
# reference reproduces every sampled value, the full 16-lane sponge state, and every accept flag.
# - the FRI transcript closure ("fri_transcript"): for the SAME three FRI cases the confirmed FRI
# ground truth (component (d)) uses, this reference derives the betas and query indices FROM THE
# TRANSCRIPT (commit_phase_commits + final_poly + pow_witness) and reproduces them byte-for-byte,
# matching BOTH the challenger extractor AND fri_ground_truth.json. That closes the loop the FRI KAT
# left open (transcript -> betas/indices -> the CONFIRMED FRI verify_query), so the betas/indices are
# now DERIVED in-circuit, not taken as inputs.
#
# What this does NOT do: it does not verify any STARK proof and does not make the top-level 0x0AE8
# accept anything. Component (f) being confirmed is necessary but not sufficient: the SP1 recursion-AIR
# constraint evaluation (component (e)) is un-ported (StarkConstraints = UNAVAILABLE), so the precompile
# stays FAIL-CLOSED (returns EMPTY for every input). See the README and the port spec.
#
# Source (pinned, checksum-matched to the repo Cargo.lock):
# p3-challenger 0.4.3-succinct duplex_challenger.rs (DuplexChallenger observe/duplexing/sample/
# sample_bits) + grinding_challenger.rs (check_witness).
# The transcript observe/sample ORDER is p3-fri verifier.rs verify_shape_and_sample_challenges.
# Sponge permutation: the CONFIRMED Poseidon2-BabyBear width-16 permutation (component (b)).
import json
import os
import sys
HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
import poseidon2_babybear_reference as P2 # component (b): the CONFIRMED Poseidon2 permutation
P = P2.P
WIDTH = 16 # DuplexChallenger WIDTH (sponge state lanes). CONFIRMED from source.
RATE = 8 # DuplexChallenger RATE (absorb width + squeeze width). CONFIRMED from source.
CHALLENGER_GT = os.path.join(HERE, "challenger_ground_truth.json")
FRI_GT = os.path.join(HERE, "fri_ground_truth.json")
# Pure-KAT script constants (mirrored EXACTLY in the Rust extractor and the Node/Java references).
KAT_DIGEST = [101, 102, 103, 104, 105, 106, 107, 108]
KAT_SAMPLE_BITS = 10
KAT_POW_BITS = 4
class DuplexChallenger:
"""Port of p3-challenger DuplexChallenger<F, Perm, 16, 8>. Faithful to duplex_challenger.rs:
overwrite-mode absorb of up to RATE inputs, permute, squeeze the first RATE lanes; sample pops
from the END of the output buffer (Vec::pop)."""
def __init__(self):
self.sponge_state = [0] * WIDTH # F::default() == 0
self.input_buffer = [] # Vec<F>
self.output_buffer = [] # Vec<F>
def duplexing(self):
assert len(self.input_buffer) <= RATE
# Overwrite the first len(input_buffer) lanes with the buffered inputs (drain).
for i, val in enumerate(self.input_buffer):
self.sponge_state[i] = val % P
self.input_buffer = []
# Apply the permutation.
self.sponge_state = P2.permute(self.sponge_state)
# output_buffer = sponge_state[0..RATE]
self.output_buffer = list(self.sponge_state[0:RATE])
def observe(self, value):
# Any buffered output is now invalid.
self.output_buffer = []
self.input_buffer.append(value % P)
if len(self.input_buffer) == RATE:
self.duplexing()
def observe_slice(self, values):
for v in values:
self.observe(v)
def observe_ext(self, ext):
"""observe_ext_element: absorb the 4 base coords [c0, c1, c2, c3] in order."""
self.observe_slice(ext)
def sample_base(self):
"""CanSample<F>::sample (from_base_fn, D=1): duplex if inputs are buffered or outputs are
exhausted, then pop the last output element. Returns a canonical field element in [0, p)."""
if self.input_buffer or not self.output_buffer:
self.duplexing()
return self.output_buffer.pop() # Vec::pop -> last element
def sample_ext(self):
"""sample_ext_element::<F_{p^4}>(): sample_vec(4) then from_base_slice -> [c0, c1, c2, c3]."""
return [self.sample_base() for _ in range(4)]
def sample_bits(self, bits):
"""CanSampleBits::sample_bits: low `bits` bits of a base-field sample's canonical rep."""
rand_f = self.sample_base()
return rand_f & ((1 << bits) - 1)
def check_witness(self, bits, witness):
"""GrindingChallenger::check_witness: observe(witness); sample_bits(bits) == 0."""
self.observe(witness)
return self.sample_bits(bits) == 0
# ============================ pure duplex KAT (mirrors the Rust script) ============================
def run_duplex_script():
"""Run the exact scripted transcript the extractor emits outputs for. Returns a dict of outputs."""
ch = DuplexChallenger()
ch.observe(11)
ch.observe(22)
a = ch.sample_base()
b = ch.sample_base()
c = ch.sample_ext()
ch.observe(33)
ch.observe(44)
ch.observe(55)
d = ch.sample_bits(KAT_SAMPLE_BITS)
ch.observe_slice(KAT_DIGEST)
e = ch.sample_ext()
f = ch.sample_base()
g = ch.sample_base()
h = ch.sample_base()
i = ch.sample_base()
j = ch.sample_ext()
state_after = list(ch.sponge_state)
return {"ch": ch, "a": a, "b": b, "c": c, "d": d, "e": e,
"f": f, "g": g, "h": h, "i": i, "j": j, "state_after": state_after}
def duplex_witness_table(ch, witnesses):
"""Reproduce the check_witness accept/reject table from the checkpoint state (each on a clone)."""
out = []
for w in witnesses:
clone = _clone(ch)
out.append({"witness": w, "accept": clone.check_witness(KAT_POW_BITS, w)})
return out
def _clone(ch):
c = DuplexChallenger()
c.sponge_state = list(ch.sponge_state)
c.input_buffer = list(ch.input_buffer)
c.output_buffer = list(ch.output_buffer)
return c
# ============================ FRI transcript closure ============================
def derive_fri_challenges(commit_phase_commits, final_poly, pow_witness, pow_bits,
log_blowup, num_queries):
"""Reproduce p3-fri verify_shape_and_sample_challenges over the DuplexChallenger:
for each commit: observe(commit digest); beta = sample_ext()
observe_ext(final_poly)
pow_accept = check_witness(pow_bits, pow_witness)
log_max_height = len(commits) + log_blowup
query_indices = [sample_bits(log_max_height) for _ in range(num_queries)]
Returns (betas, query_indices, pow_accept). The betas/indices are DERIVED here, not taken as
inputs; this is exactly what component (f) adds on top of the confirmed FRI verify_query."""
ch = DuplexChallenger()
betas = []
for comm in commit_phase_commits:
ch.observe_slice(comm) # observe the 8-element commitment digest
betas.append(ch.sample_ext())
ch.observe_ext(final_poly) # observe the F_{p^4} final polynomial
pow_accept = ch.check_witness(pow_bits, pow_witness)
log_max_height = len(commit_phase_commits) + log_blowup
query_indices = [ch.sample_bits(log_max_height) for _ in range(num_queries)]
return betas, query_indices, pow_accept
# ============================ conformance KAT (real ground truth) ============================
def conformance_kats():
"""Return (passed, total, details). Ties to BOTH the challenger ground truth (pure duplex KAT +
FRI transcript) AND fri_ground_truth.json (closing the FRI betas/indices loop)."""
with open(CHALLENGER_GT) as fp:
gt = json.load(fp)
with open(FRI_GT) as fp:
fri_gt = json.load(fp)
passed = 0
total = 0
details = []
def check(name, cond):
nonlocal passed, total
total += 1
if cond:
passed += 1
details.append({"check": name, "pass": bool(cond)})
# (0) sanity: the extractor's perm matches the confirmed Poseidon2 permutation.
check("perm_zeros_matches_poseidon2", P2.permute([0] * 16) == gt["perm_zeros"])
# (1) pure duplex KAT.
k = gt["duplex_kat"]
r = run_duplex_script()
check("duplex.a", r["a"] == k["a"])
check("duplex.b", r["b"] == k["b"])
check("duplex.c", r["c"] == k["c"])
check("duplex.d", r["d"] == k["d"])
check("duplex.e", r["e"] == k["e"])
check("duplex.f", r["f"] == k["f"])
check("duplex.g", r["g"] == k["g"])
check("duplex.h", r["h"] == k["h"])
check("duplex.i", r["i"] == k["i"])
check("duplex.j", r["j"] == k["j"])
check("duplex.state_after", r["state_after"] == k["state_after"])
# check_witness accept/reject table
witnesses = [w["witness"] for w in k["witness_table"]]
my_table = duplex_witness_table(r["ch"], witnesses)
table_ok = all(my_table[n]["accept"] == k["witness_table"][n]["accept"]
for n in range(len(witnesses)))
check("duplex.witness_table", table_ok)
# sanity: the table exhibits BOTH accept and reject (a meaningful grinding test).
accepts = [w["accept"] for w in k["witness_table"]]
check("duplex.table_has_accept_and_reject", any(accepts) and not all(accepts))
# (2) FRI transcript closure: derive betas/indices/pow-accept and match extractor + fri_ground_truth.
fri_by_name = {c["name"]: c for c in fri_gt["cases"]}
for tc in gt["fri_transcript"]:
name = tc["name"]
betas, indices, pow_accept = derive_fri_challenges(
tc["commit_phase_commits"], tc["final_poly"], tc["pow_witness"],
tc["pow_bits"], tc["log_blowup"], tc["num_queries"])
check(f"fri[{name}].pow_accept", pow_accept is True)
check(f"fri[{name}].betas_match_extractor", betas == tc["betas"])
check(f"fri[{name}].indices_match_extractor", indices == tc["query_indices"])
# close the loop: the DERIVED betas/indices equal the ones the CONFIRMED FRI KAT consumed.
fc = fri_by_name[name]
check(f"fri[{name}].betas_close_fri_loop", betas == fc["betas"])
check(f"fri[{name}].indices_close_fri_loop", indices == [q["index"] for q in fc["queries"]])
return passed, total, details
# ============================ shared cross-language vector set ============================
def shared_vectors():
r = run_duplex_script()
with open(CHALLENGER_GT) as fp:
gt = json.load(fp)
witnesses = [w["witness"] for w in gt["duplex_kat"]["witness_table"]]
fri_out = []
for tc in gt["fri_transcript"]:
betas, indices, pow_accept = derive_fri_challenges(
tc["commit_phase_commits"], tc["final_poly"], tc["pow_witness"],
tc["pow_bits"], tc["log_blowup"], tc["num_queries"])
fri_out.append({"name": tc["name"], "betas": betas, "queryIndices": indices,
"powAccept": pow_accept})
return {
"permZeros": P2.permute([0] * 16),
"duplex": {"a": r["a"], "b": r["b"], "c": r["c"], "d": r["d"], "e": r["e"],
"f": r["f"], "g": r["g"], "h": r["h"], "i": r["i"], "j": r["j"],
"stateAfter": r["state_after"],
"witnessTable": duplex_witness_table(r["ch"], witnesses)},
"fri": fri_out,
}
if __name__ == "__main__":
if len(sys.argv) > 1 and sys.argv[1] == "kat":
p, t, det = conformance_kats()
for d in det:
print(("PASS" if d["pass"] else "FAIL"), d["check"])
print(f"conformance: PASS={p} FAIL={t - p} TOTAL={t}")
sys.exit(0 if p == t else 1)
json.dump(shared_vectors(), sys.stdout)