open-standards/wai/jwp-conformance/chain_differential.py

#!/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 <wai_jwp_vectors binary> [--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))