open-standards/jcp/conformance/verify_replay.py

#!/usr/bin/env python3
"""A second reader of the published replay log, conformance/v1/replay.json (SPEC §8.1).

It shares no code with the reference implementation, and needs no package: det-CBOR,
Ed25519 (RFC 8032) and BLAKE3 are written out below from their specifications. For every
entry of the log it checks that

  1. the envelope is a COSE_Sign1 under the published steward key, over a canonical core
     equal to the published wire form;
  2. `seq` counts from zero and `prev` is the BLAKE3 of the previous envelope;
  3. what SPEC §8.1's rules read from the core is what the runtime's ledgers moved by:
     the joules debited, the money debited, and what a tainted sub-budget accrued. A hold
     is read against the whole log: settled by a later entry it restores nothing, and
     settled by none it restores its `held_uj`, as a restart holds it. A firing of a
     standing grant debits nothing, and counts one fire and its `minted_uj` in the
     standing grant's mint ledger (SPEC §4.6).

Then the sums must be the ledgers the runtime and the firings ended with, and each
receipt in `legacy` must verify, state none of the member it was sealed before, and read
as published.

    verify_replay.py [--selftest] [DIR]

DIR defaults to conformance/v1 beside this file. --selftest also checks that the reader
fails: a changed debit, a dropped or moved entry, a changed total, and a reader with each
planted fault must all be refused. Exit 0 when everything holds, 1 otherwise.
"""
import hashlib
import json
import os
import struct
import sys

# The two deny reasons the debit point seals (SPEC §4.9, §4.12).
DEBIT_POINT_REASONS = ("caveat-unsatisfied", "window-exceeded")


class Refused(Exception):
    pass


# ---- det-CBOR (RFC 8949 §4.2.1), the subset a JCR-1 receipt uses ------------------------

def cbor_read(b, i=0):
    head = b[i]
    major, info = head >> 5, head & 31
    i += 1
    if info < 24:
        n = info
    elif info <= 27:
        width = 1 << (info - 24)
        n = int.from_bytes(b[i:i + width], "big")
        i += width
    else:
        raise Refused(f"CBOR additional info {info}")
    if major == 0:
        return n, i
    if major == 1:
        return -1 - n, i
    if major == 2:
        return b[i:i + n], i + n
    if major == 3:
        return b[i:i + n].decode(), i + n
    if major == 4:
        out = []
        for _ in range(n):
            v, i = cbor_read(b, i)
            out.append(v)
        return out, i
    if major == 5:
        out = {}
        for _ in range(n):
            k, i = cbor_read(b, i)
            out[k], i = cbor_read(b, i)
        return out, i
    if major == 6:
        return cbor_read(b, i)  # the tag itself is checked by the caller
    raise Refused(f"CBOR major type {major}")


def cbor_head(major, n):
    if n < 24:
        return bytes([major << 5 | n])
    for info, width in ((24, 1), (25, 2), (26, 4), (27, 8)):
        if n < 1 << (8 * width):
            return bytes([major << 5 | info]) + n.to_bytes(width, "big")
    raise Refused("CBOR length past 64 bits")


def cbor_write(v):
    if isinstance(v, int):
        return cbor_head(0, v) if v >= 0 else cbor_head(1, -1 - v)
    if isinstance(v, bytes):
        return cbor_head(2, len(v)) + v
    if isinstance(v, str):
        e = v.encode()
        return cbor_head(3, len(e)) + e
    if isinstance(v, list):
        return cbor_head(4, len(v)) + b"".join(cbor_write(x) for x in v)
    if isinstance(v, dict):
        keys = sorted(v, key=lambda k: (len(k.encode()), k.encode()))
        return cbor_head(5, len(v)) + b"".join(cbor_write(k) + cbor_write(v[k]) for k in keys)
    raise Refused(f"no CBOR form for {type(v)}")


# ---- Ed25519 verification (RFC 8032 §5.1) -----------------------------------------------

P = 2**255 - 19
L = 2**252 + 27742317777372353535851937790883648493
D = -121665 * pow(121666, P - 2, P) % P
SQRT_M1 = pow(2, (P - 1) // 4, P)


def point_add(a, b):
    x1, y1, z1, t1 = a
    x2, y2, z2, t2 = b
    e = (y1 - x1) * (y2 - x2) % P
    f = (y1 + x1) * (y2 + x2) % P
    g = 2 * t1 * t2 * D % P
    h = 2 * z1 * z2 % P
    i, j, k, m = f - e, h - g, h + g, f + e
    return (i * j % P, k * m % P, j * k % P, i * m % P)


def scalar_mul(s, p):
    q = (0, 1, 1, 0)
    while s:
        if s & 1:
            q = point_add(q, p)
        p = point_add(p, p)
        s >>= 1
    return q


def decompress(b):
    if len(b) != 32:
        return None
    y = int.from_bytes(b, "little")
    sign = y >> 255
    y &= (1 << 255) - 1
    if y >= P:
        return None
    u = (y * y - 1) * pow(D * y * y + 1, P - 2, P) % P
    x = pow(u, (P + 3) // 8, P)
    if (x * x - u) % P:
        x = x * SQRT_M1 % P
    if (x * x - u) % P:
        return None
    if x == 0 and sign:
        return None
    if x & 1 != sign:
        x = P - x
    return (x, y, 1, x * y % P)


BASE = decompress((4 * pow(5, P - 2, P) % P).to_bytes(32, "little"))


def same_point(a, b):
    x1, y1, z1, _ = a
    x2, y2, z2, _ = b
    return (x1 * z2 - x2 * z1) % P == 0 and (y1 * z2 - y2 * z1) % P == 0


def ed25519_verify(key, message, signature):
    a, r = decompress(key), decompress(signature[:32])
    s = int.from_bytes(signature[32:], "little")
    if a is None or r is None or s >= L:
        return False
    h = int.from_bytes(hashlib.sha512(signature[:32] + key + message).digest(), "little") % L
    return same_point(scalar_mul(s, BASE), point_add(r, scalar_mul(h, a)))


# ---- BLAKE3, hash mode ------------------------------------------------------------------

IV = [0x6A09E667, 0xBB67AE85, 0x3C6EF372, 0xA54FF53A, 0x510E527F, 0x9B05688C, 0x1F83D9AB, 0x5BE0CD19]
PERMUTATION = [2, 6, 3, 10, 7, 0, 4, 13, 1, 11, 12, 5, 9, 14, 15, 8]
CHUNK_START, CHUNK_END, PARENT, ROOT = 1, 2, 4, 8
MASK = 0xFFFFFFFF


def rotr(x, n):
    return (x >> n | x << (32 - n)) & MASK


def mix(s, a, b, c, d, x, y):
    s[a] = (s[a] + s[b] + x) & MASK
    s[d] = rotr(s[d] ^ s[a], 16)
    s[c] = (s[c] + s[d]) & MASK
    s[b] = rotr(s[b] ^ s[c], 12)
    s[a] = (s[a] + s[b] + y) & MASK
    s[d] = rotr(s[d] ^ s[a], 8)
    s[c] = (s[c] + s[d]) & MASK
    s[b] = rotr(s[b] ^ s[c], 7)


def compress(node):
    cv, words, counter, block_len, flags = node
    s = cv[:8] + IV[:4] + [counter & MASK, counter >> 32 & MASK, block_len, flags]
    m = list(words)
    for _ in range(7):
        for i, (a, b, c, d) in enumerate(((0, 4, 8, 12), (1, 5, 9, 13), (2, 6, 10, 14), (3, 7, 11, 15),
                                          (0, 5, 10, 15), (1, 6, 11, 12), (2, 7, 8, 13), (3, 4, 9, 14))):
            mix(s, a, b, c, d, m[2 * i], m[2 * i + 1])
        m = [m[i] for i in PERMUTATION]
    return [s[i] ^ s[i + 8] for i in range(8)]


def chunk_node(chunk, index):
    blocks = [chunk[i:i + 64] for i in range(0, len(chunk), 64)] or [b""]
    cv = IV[:]
    for i, block in enumerate(blocks):
        flags = (CHUNK_START if i == 0 else 0) | (CHUNK_END if i == len(blocks) - 1 else 0)
        words = struct.unpack("<16I", block.ljust(64, b"\0"))
        if i == len(blocks) - 1:
            return (cv, words, index, len(block), flags)
        cv = compress((cv, words, index, 64, flags))


def tree_node(chunks, lo, hi):
    if hi - lo == 1:
        return chunk_node(chunks[lo], lo)
    left = 1 << ((hi - lo - 1).bit_length() - 1)
    words = compress(tree_node(chunks, lo, lo + left)) + compress(tree_node(chunks, lo + left, hi))
    return (IV[:], words, 0, 64, PARENT)


def blake3(data):
    chunks = [data[i:i + 1024] for i in range(0, len(data), 1024)] or [b""]
    cv, words, counter, block_len, flags = tree_node(chunks, 0, len(chunks))
    return struct.pack("<8I", *compress((cv, words, counter, block_len, flags | ROOT)))


# ---- The receipt ------------------------------------------------------------------------

def open_envelope(envelope, steward, key):
    """The signed core of a COSE_Sign1 envelope, or Refused."""
    if envelope[:1] != b"\xd2":
        raise Refused("not a tagged COSE_Sign1")
    (protected, unprotected, payload, signature), end = cbor_read(envelope)
    header, _ = cbor_read(protected)
    if end != len(envelope) or unprotected != {}:
        raise Refused("not the COSE_Sign1 shape")
    if header.get(1) != -8 or header.get(4) != steward.encode():
        raise Refused("not this steward's EdDSA envelope")
    if not ed25519_verify(key, cbor_write(["Signature1", protected, b"", payload]), signature):
        raise Refused("the signature does not verify")
    core, end = cbor_read(payload)
    if end != len(payload) or cbor_write(core) != payload:
        raise Refused("the core is not canonical")
    return core


def is_hold(core):
    """An allow stating `held_uj` (SPEC §8.1). A refusal debits nothing, so it holds nothing."""
    return "held_uj" in core and core["verdict"] == "allow"


def is_mint(core):
    """A firing of a standing grant: an allow stating `minted_uj` (SPEC §4.6, §8.1). A refusal
    mints nothing, so it fires nothing."""
    return "minted_uj" in core and core["verdict"] == "allow"


def read_debits(core):
    """What this receipt debited, by SPEC §8.1's rules: [(grant, uj)]. A hold debits nothing
    itself: its work is the settling entry's, or, settled by none, read_held's. A firing
    debits nothing: it runs no work, and its child's receipts debit what the child spends."""
    grants = core.get("chain_ids", [core["grant_id"]])
    if is_hold(core) or is_mint(core):
        return []
    if "chain_debited_uj" in core:
        if len(core["chain_debited_uj"]) != len(grants):
            raise Refused("chain_debited_uj does not match chain_ids")
        return list(zip(grants, core["chain_debited_uj"]))
    allowed = core["verdict"] == "allow"
    if "debited_uj" in core:
        uj = core["debited_uj"]
    elif allowed or core.get("deny_reason") in DEBIT_POINT_REASONS:
        uj = core["cost_uj"]
    else:
        uj = 0
    return [] if uj == 0 and not allowed else [(g, uj) for g in grants]


def read_money(core):
    """What this receipt debited in money, by SPEC §8.1's rules: [(grant, minor)].

    Only a stated figure counts, and a grant debited none is left out."""
    grants = core.get("chain_ids", [core["grant_id"]])
    if "chain_debited_minor" in core:
        per_link = core["chain_debited_minor"]
        if len(per_link) != len(grants):
            raise Refused("chain_debited_minor does not match chain_ids")
    elif "debited_minor" in core:
        per_link = [core["debited_minor"]] * len(grants)
    else:
        return []
    return [(g, m) for g, m in zip(grants, per_link) if m]


def read_tainted(core):
    """What this receipt accrued against a tainted sub-budget, by SPEC §8.1: [(grant, uj)].

    `tainted_uj` where stated; otherwise a core with a flow block that debited accrued its
    cost_uj, and any other core nothing."""
    grants = core.get("chain_ids", [core["grant_id"]])
    if "tainted_uj" in core:
        uj = core["tainted_uj"]
    elif "flow" in core and read_debits(core):
        uj = core["cost_uj"]
    else:
        uj = 0
    return [(g, uj) for g in grants] if uj else []


def read_mints(core):
    """What this receipt adds to a standing grant's mint ledger, by SPEC §8.1: one fire and its
    `minted_uj`, for the standing grant it names. [(grant, fires, uj)]."""
    return [(core["grant_id"], 1, core["minted_uj"])] if is_mint(core) else []


def read_settled(cores):
    """The ids of the entries some entry settles (SPEC §8.1): each `settles` member."""
    return {c["settles"] for c in cores if "settles" in c}


def read_held(core, unresolved):
    """What a hold restores, by SPEC §8.1: nothing when an entry settles it; otherwise its
    `held_uj` from every grant it names, and as much against each one's tainted
    sub-budget where it carries a flow block. ([(grant, uj)], [(grant, uj)])."""
    if not unresolved or not is_hold(core):
        return [], []
    held = [(g, core["held_uj"]) for g in core.get("chain_ids", [core["grant_id"]])]
    return held, (held if "flow" in core else [])


# The dimensions a log is read in: the entry's list, its unit, and the total it sums to.
DIMENSIONS = (("debits", "uj", "spent_after"), ("money", "minor", "money_after"), ("tainted", "uj", "tainted_after"))


def published(entry, dim, unit):
    return [(d["grant_hex"], d[unit]) for d in entry[dim]]


def published_mints(entry):
    return [(d["grant_hex"], d["fires"], d["uj"]) for d in entry["mints"]]


def check(doc, steward, key, read=read_debits, money=read_money, tainted=read_tainted, held=read_held,
          settled=read_settled, mints=read_mints):
    """Refused unless the log reads as published. Returns the entry count."""
    readers = (read, money, tainted)
    prev = "0" * 64
    cores, ids = [], []
    for i, entry in enumerate(doc["entries"]):
        envelope = bytes.fromhex(entry["envelope_hex"])
        core = open_envelope(envelope, steward, key)
        if core != entry["wire"]:
            raise Refused(f"entry {i}: the signed core is not the published wire form")
        if core["seq"] != i or core["prev"] != prev:
            raise Refused(f"entry {i}: the chain is broken here")
        prev = blake3(envelope).hex()
        cores.append(core)
        ids.append(prev)
    done = settled(cores)
    ledgers = [{}, {}, {}]
    fired = {}
    for i, (entry, core, own) in enumerate(zip(doc["entries"], cores, ids)):
        held_uj, held_tainted = held(core, own not in done)
        reads = (read(core) + held_uj, money(core), tainted(core) + held_tainted)
        for got, (dim, unit, _), ledger in zip(reads, DIMENSIONS, ledgers):
            want = published(entry, dim, unit)
            if got != want:
                raise Refused(f"entry {i}: read {dim} {got}, the ledger moved by {want}")
            for grant, n in got:
                ledger[grant] = ledger.get(grant, 0) + n
        got = mints(core)
        if got != published_mints(entry):
            raise Refused(f"entry {i}: read mints {got}, the mint ledger moved by {published_mints(entry)}")
        for grant, fires, uj in got:
            was = fired.get(grant, {"fires": 0, "uj": 0})
            fired[grant] = {"fires": was["fires"] + fires, "uj": was["uj"] + uj}
    for (_, _, total), ledger in zip(DIMENSIONS, ledgers):
        if ledger != doc[total]:
            raise Refused(f"the log replays to {ledger}, not {total} {doc[total]}")
    if fired != doc["mints_after"]:
        raise Refused(f"the log's firings rebuild {fired}, not mints_after {doc['mints_after']}")
    for i, entry in enumerate(doc["legacy"]):
        core = open_envelope(bytes.fromhex(entry["envelope_hex"]), steward, key)
        if core != entry["wire"] or entry["sealed_before"] in core:
            raise Refused(f"legacy {i}: not a receipt without {entry['sealed_before']}")
        for reader, (dim, unit, _) in zip(readers, DIMENSIONS):
            if reader(core) != published(entry, dim, unit):
                raise Refused(f"legacy {i}: read {dim} {reader(core)}")
        if mints(core) != published_mints(entry):
            raise Refused(f"legacy {i}: read mints {mints(core)}")
    return len(doc["entries"])


def check_coverage(doc):
    """Refused unless the log still holds each form the rules distinguish."""
    wires = [e["wire"] for e in doc["entries"]]
    denials = [w for w in wires if w["verdict"] == "deny"]
    ids = [blake3(bytes.fromhex(e["envelope_hex"])).hex() for e in doc["entries"]]
    done = read_settled(wires)
    open_holds = [i for i, w in enumerate(wires) if is_hold(w) and ids[i] not in done]
    forms = {
        "an allow stating nothing": any(w["verdict"] == "allow" and "debited_uj" not in w for w in wires),
        "a denial that debited": any(w.get("debited_uj", 0) > 0 for w in denials),
        "a refusal stating its zero": any(w.get("debited_uj") == 0 for w in denials),
        "a refusal stating nothing": any("debited_uj" not in w for w in denials),
        "a debit below its cost": any(w.get("debited_uj", w["cost_uj"]) < w["cost_uj"] and w["debited_uj"] for w in wires),
        "a chained receipt": any("chain_ids" in w for w in wires),
        "links debited differently": any("chain_debited_uj" in w for w in wires),
        "a money debit": any("debited_minor" in w for w in wires),
        "links debited money differently": any("chain_debited_minor" in w for w in wires),
        "an accrual its flow block says": any("flow" in w and "tainted_uj" not in w and read_tainted(w) for w in wires),
        "a tainted refusal with a cost": any("flow" in w and w["verdict"] == "deny" and w["cost_uj"] > 0 for w in wires),
        "an accrual stated": any("tainted_uj" in w and "flow" not in w for w in wires),
        "a hold an entry settles": any(is_hold(w) and ids[i] in done for i, w in enumerate(wires)),
        "a hold nothing settles": any("flow" not in wires[i] for i in open_holds),
        "a tainted hold nothing settles": any("flow" in wires[i] for i in open_holds),
        "an entry after an unsettled hold, not settling it": any(i + 1 < len(wires) for i in open_holds),
        "a firing of a standing grant": any(is_mint(w) for w in wires),
        "two firings of one standing grant": any(
            sum(is_mint(v) and v["grant_id"] == w["grant_id"] for v in wires) > 1 for w in wires if is_mint(w)),
        "a call under a child a firing minted": any(
            not is_mint(w) and w.get("chain_ids", [None])[0] in {v["grant_id"] for v in wires if is_mint(v)}
            for w in wires),
        "a legacy receipt for each member": {e["sealed_before"] for e in doc["legacy"]}
        == {"debited_uj", "debited_minor", "tainted_uj", "settles"},
    }
    missing = [name for name, held in forms.items() if not held]
    if missing:
        raise Refused(f"the log no longer holds: {', '.join(missing)}")


# ---- Self-test: the reader must be able to fail -----------------------------------------

def refuses(name, doc, steward, key, **readers):
    try:
        check(doc, steward, key, **readers)
    except Refused:
        return
    raise SystemExit(f"self-test: {name} was accepted")


def with_changed_member(doc, at, member):
    """`doc` with entry `at`'s stated `member` changed in place, wire form to match."""
    out = json.loads(json.dumps(doc))
    entry = out["entries"][at]
    key = cbor_write(member)
    envelope = bytearray.fromhex(entry["envelope_hex"])
    start = envelope.index(key) + len(key)
    value, end = cbor_read(bytes(envelope), start)
    # An integer's low bit; a hex id's last digit.
    changed = value ^ 1 if isinstance(value, int) else value[:-1] + ("1" if value[-1] == "0" else "0")
    envelope[start:end] = cbor_write(changed)
    entry["envelope_hex"] = envelope.hex()
    entry["wire"][member] = changed
    return out


def selftest(doc, steward, key):
    if blake3(b"").hex() != "af1349b9f5f9a1a6a0404dea36dcc9499bcb25c9adc112b7cc9a93cae41f3262":
        raise SystemExit("self-test: BLAKE3 of nothing")
    if blake3(bytes(i % 251 for i in range(1025))).hex() != "d00278ae47eb27b34faecf67b4fe263f82d5412916c1ffd97c8cb7fb814b8444":
        raise SystemExit("self-test: BLAKE3 past one chunk")

    for member in ("debited_uj", "debited_minor", "tainted_uj", "held_uj", "settles", "minted_uj"):
        stated = [i for i, e in enumerate(doc["entries"]) if member in e["wire"]]
        if not stated:
            raise SystemExit(f"self-test: no entry states {member}")
        for at in stated:
            refuses(f"{member} changed on signed entry {at}", with_changed_member(doc, at, member), steward, key)

    breach = next(i for i, e in enumerate(doc["entries"]) if e["wire"].get("debited_uj", 0) > 0 and e["wire"]["verdict"] == "deny")
    dropped = json.loads(json.dumps(doc))
    del dropped["entries"][breach]
    refuses("a breach dropped from the log", dropped, steward, key)

    moved = json.loads(json.dumps(doc))
    moved["entries"][0], moved["entries"][1] = moved["entries"][1], moved["entries"][0]
    refuses("two entries swapped", moved, steward, key)

    for _, _, name in DIMENSIONS:
        for delta in (1, -1):
            total = json.loads(json.dumps(doc))
            grant = next(iter(total[name]))
            total[name][grant] += delta
            refuses(f"{name} changed by {delta}", total, steward, key)
    for figure in ("fires", "uj"):
        for delta in (1, -1):
            total = json.loads(json.dumps(doc))
            total["mints_after"][next(iter(total["mints_after"]))][figure] += delta
            refuses(f"mints_after {figure} changed by {delta}", total, steward, key)

    faults = {
        "a reader dropping a breach": lambda c: read_debits(c) if c["verdict"] == "allow" else [],
        "a reader counting a refusal": lambda c: [(g, c["cost_uj"]) for g in c.get("chain_ids", [c["grant_id"]])],
        "a reader losing an ancestor's debit": lambda c: read_debits(c)[-1:],
        "a reader taking one figure for every link": lambda c: [(g, c.get("debited_uj", c["cost_uj"])) for g, _ in read_debits(c)],
        "a reader reading an old denial as zero": lambda c: read_debits(c) if "debited_uj" in c or c["verdict"] == "allow" else [],
    }
    for name, read in faults.items():
        refuses(name, doc, steward, key, read=read)

    def grants(c):
        return c.get("chain_ids", [c["grant_id"]])

    money_faults = {
        "a reader restoring no money": lambda c: [],
        "a reader taking one money figure for every link":
            lambda c: [(g, c["debited_minor"]) for g in grants(c)] if c.get("debited_minor") else [],
    }
    for name, money in money_faults.items():
        refuses(name, doc, steward, key, money=money)

    tainted_faults = {
        "a reader restoring no accrual": lambda c: [],
        "a reader ignoring a flow block": lambda c: [(g, c["tainted_uj"]) for g in grants(c)] if c.get("tainted_uj") else [],
        "a reader ignoring a stated accrual": lambda c: [] if "tainted_uj" in c else read_tainted(c),
        "a reader counting a tainted refusal": lambda c: [(g, c["cost_uj"]) for g in grants(c)] if "flow" in c else read_tainted(c),
    }
    for name, tainted in tainted_faults.items():
        refuses(name, doc, steward, key, tainted=tainted)

    held_faults = {
        "a reader ignoring a hold nothing settles": lambda c, unresolved: ([], []),
        "a reader counting a settled hold": lambda c, unresolved: read_held(c, True),
        "a reader ignoring a hold's flow block": lambda c, unresolved: (read_held(c, unresolved)[0], []),
        "a reader holding a hold's cost_uj": lambda c, unresolved:
            ([(g, c["cost_uj"]) for g, _ in read_held(c, unresolved)[0]], read_held(c, unresolved)[1]),
    }
    for name, held in held_faults.items():
        refuses(name, doc, steward, key, held=held)
    refuses("a reader taking the next entry as a hold's settle", doc, steward, key,
            settled=lambda cores: {c["prev"] for c in cores})

    refuses("a reader reading a firing as a call", doc, steward, key,
            read=lambda c: [(g, c["cost_uj"]) for g in grants(c)] if is_mint(c) else read_debits(c))
    mint_faults = {
        "a reader counting no firing": lambda c: [],
        "a reader counting a fire but not what it minted": lambda c: [(g, f, 0) for g, f, _ in read_mints(c)],
        "a reader counting a chained call as a firing":
            lambda c: read_mints(c) or ([(c["chain_ids"][0], 1, 0)] if "chain_ids" in c else []),
    }
    for name, mints in mint_faults.items():
        refuses(name, doc, steward, key, mints=mints)


def main():
    args = sys.argv[1:]
    run_selftest = "--selftest" in args
    args = [a for a in args if a != "--selftest"]
    pack = args[0] if args else os.path.join(os.path.dirname(os.path.abspath(__file__)), "v1")
    with open(os.path.join(pack, "meta.json"), encoding="utf-8") as f:
        meta = json.load(f)
    with open(os.path.join(pack, "replay.json"), encoding="utf-8") as f:
        doc = json.load(f)
    steward, key = meta["steward"], bytes.fromhex(meta["pubkey_hex"])
    try:
        count = check(doc, steward, key)
        check_coverage(doc)
    except Refused as e:
        print(f"verify_replay: {e}", file=sys.stderr)
        return 1
    if run_selftest:
        selftest(doc, steward, key)
    print(f"verify_replay: {count} entries and {len(doc['legacy'])} legacy receipts read as published"
          + (", and every fault was refused" if run_selftest else ""))
    return 0


if __name__ == "__main__":
    sys.exit(main())