#!/usr/bin/env python3 """Adversarial differential for jwp-receipts §3 step 3 / §6.1 receipt chains per (signer_pubkey, track_alias): wai-rs (`wai_jwp_vectors verdicts`) against jwp_verify.py (`verdicts`). Generates `chain` vectors with no declared outcome: two keys, two tracks, legacy and labelled payloads, genuine links, replays, replays with a rewritten (unsigned, in legacy) parent, forks, cross-track and cross-key parents, fresh genesis beside a held head, bad signatures, group_id up/equal/down, group_end at, before and after the clock, offsets, fractions, leap second, year 0000 and non-dates; guard on/off; freshness on/off. Every chain is signed with the corpus's published seeds, so a run is reproducible from its seed. Exit status 0 only if the two verifiers give the same verdict on every chain. usage: chain_differential.py [--n N] [--seed S] """ import json, os, random, subprocess, sys, tempfile HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, HERE) sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance")) import jwp_verify as J # noqa: E402 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey # noqa: E402 from cryptography.hazmat.primitives import serialization # noqa: E402 KEYS = {s: Ed25519PrivateKey.from_private_bytes(bytes([s]) * 32) for s in (0x11, 0x22)} def pk(k): return k.public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw).hex() NOW = "2026-09-30T12:00:01Z" ENDS = [NOW, "2026-09-30T12:00:01Z", "2026-09-30T11:59:01Z", "2026-09-30T11:59:00Z", "2026-09-30T11:00:00Z", "2026-09-30T12:00:06Z", "2026-09-30T12:00:07Z", "2026-09-30T14:00:01+02:00", "2026-09-30T11:59:01.999Z", "2026-09-30T11:59:60Z", "0000-03-01T00:00:00Z", "9999-12-31T23:59:59Z", "2026-02-29T00:00:00Z", "now", "2026-09-30T12:00:01-00:00"] def make(seed, track, gid, root_tag, parent, end, labelled, bad_sig=False): g = {"kind": "group", "track_alias": track, "group_id": gid, "root_hash": J.blake3(b"root %d" % root_tag).hex(), "leaf_count": 1, "joules_total": 0, "group_start": "2026-09-30T12:00:00Z", "group_end": end, "signer_pubkey": pk(KEYS[seed]), "signer_id": "did:example:publisher", "sig": "00" * 64, "parent_receipt_hash": parent.hex() if parent is not None else None} if labelled: g["sign_version"] = 2 r = J.read_group(g) sig = KEYS[seed].sign(J.payload(r)) if bad_sig: sig = bytes([sig[0] ^ 1]) + sig[1:] g["sig"] = sig.hex() return g def link(g): return J.link(J.read_group(g)) def gen(rng, i): guard = rng.random() < 0.5 fresh = rng.choice([None, None, {"now": NOW, "max_age_s": 60, "max_skew_s": 5}]) genuine = {} # (seed, track) -> [groups] emitted = [] out = [] tag = 1000 * i for _ in range(rng.randint(1, 9)): tag += 1 seed = rng.choice([0x11, 0x11, 0x22]) track = rng.choice([7, 7, 8]) labelled = rng.random() < 0.5 chain = genuine.setdefault((seed, track), []) end = rng.choice([NOW] * 6 + ENDS) kind = rng.choice(["next"] * 6 + ["replay", "replay_rewritten", "fork", "cross_track", "cross_key", "genesis", "bad_sig", "random_parent"]) gid_next = (chain[-1]["group_id"] + 1) if chain else rng.choice([0, 0, 5]) gid = rng.choice([gid_next] * 5 + [max(0, gid_next - 1), max(0, gid_next - 2), gid_next + 3]) if kind == "replay" and emitted: g = rng.choice(emitted) elif kind == "replay_rewritten" and emitted: g = dict(rng.choice(emitted)) tgt = rng.choice(emitted) g["parent_receipt_hash"] = link(tgt).hex() # unsigned in a legacy group elif kind == "fork" and len(chain) >= 2: g = make(seed, track, gid, tag, link(rng.choice(chain[:-1])), end, labelled) elif kind == "cross_track": other = genuine.get((seed, 15 - track)) or [] g = make(seed, track, gid, tag, link(other[-1]) if other else None, end, labelled) elif kind == "cross_key": other = genuine.get((0x33 - seed, track)) or [] g = make(seed, track, gid, tag, link(other[-1]) if other else None, end, labelled) elif kind == "genesis": g = make(seed, track, gid, tag, None, end, labelled) elif kind == "bad_sig": g = make(seed, track, gid, tag, link(chain[-1]) if chain else None, end, labelled, bad_sig=True) elif kind == "random_parent": g = make(seed, track, gid, tag, J.blake3(b"nothing %d" % tag), end, labelled) else: g = make(seed, track, gid, tag, link(chain[-1]) if chain else None, end, labelled) chain.append(g) emitted.append(g) out.append(g) return {"name": f"chain_diff_{i}", "procedure": "chain", "input": {"groups": out, "guard": guard, "freshness": fresh}, "expected": {"outcome": "chain", "results": []}} def main(argv): rust_bin = argv[1] n = int(argv[argv.index("--n") + 1]) if "--n" in argv else 3000 seed = argv[argv.index("--seed") + 1] if "--seed" in argv else "1" rng = random.Random("chains " + seed) vs = [gen(rng, i) for i in range(n)] with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False, encoding="utf-8") as f: json.dump({"vectors": vs}, f) path = f.name try: rs = subprocess.run([rust_bin, "verdicts", path], capture_output=True, text=True, check=True).stdout py = subprocess.run([sys.executable, os.path.join(HERE, "jwp_verify.py"), "verdicts", path], capture_output=True, text=True, check=True).stdout finally: os.unlink(path) r = dict(l.split("\t", 1) for l in rs.splitlines()) p = dict(l.split("\t", 1) for l in py.splitlines()) if len(r) != n or len(p) != n: print(f"verdict counts differ: {n} chains, wai-rs {len(r)}, python {len(p)}") return 1 differ = [k for k in r if r[k] != p.get(k)] crashes = [k for k in r if r[k] == "error" or p.get(k) == "error"] tally = {} for v in r.values(): for one in v.split(": ", 1)[-1].split(" / "): tally[one] = tally.get(one, 0) + 1 print(f"{n} chains (seed {seed}); per group: " + ", ".join(f"{k} x{c}" for k, c in sorted(tally.items()))) print(f"{len(differ)} differ, {len(crashes)} error") for k in differ[:10]: v = next(x for x in vs if x["name"] == k) ends = [g["group_end"] for g in v["input"]["groups"]] print(f" {k}: rust={r[k]!r}\n python={p.get(k)!r}\n freshness={v['input']['freshness']} ends={ends}") return 1 if differ or crashes else 0 if __name__ == "__main__": sys.exit(main(sys.argv))