open-standards/wai/live-conformance/telemetry_aggregate_verify.py

#!/usr/bin/env python3
"""Independent verifier of the QoE aggregation corpus (extensions/telemetry-aggregation.md).

Written from the extension's text and from the texts it cites: draft-ietf-ppm-dap-20 (DAP),
draft-irtf-cfrg-vdaf-22 (the VDAF draft, whose algorithms are written in Python and are
followed here), RFC 9180 (HPKE), RFC 9861 (TurboSHAKE), RFC 7748 (X25519), RFC 5869 (HKDF),
FIPS 197 (AES) and NIST SP 800-38D (GCM). It shares no code with wai-rs, which generates the
corpus, and never runs it. Ed25519 signatures are checked with
../claim-conformance/_ed25519.py, WAI's strict rule (SPEC §7 *Signatures*). Everything else
is here:

- Keccak-p[1600, 12], TurboSHAKE128 and the VDAF draft's XofTurboShake128 (§6.2.1);
- Field64 and Field128, the Lagrange-basis polynomials, the gadgets and circuits of Prio3Count,
  Prio3Sum, Prio3SumVec and Prio3Histogram, the FLP and Prio3 itself (§6, §7, Appendix A),
  checked first against the draft's own test vectors in vdaf-22/;
- X25519, HKDF-SHA256, AES-128-GCM and HPKE base mode, checked against RFC 9180 A.1.1;
- DAP's messages, and each flow of the corpus re-run from its reports: every input share
  opened with the aggregators' test keys, each measurement recombined and required to be its
  session's (extension §3.1), both aggregators' verification, the aggregation job's request
  and response required byte for byte, the batch buckets, every refusal, the collection,
  every aggregate share and statement opened with the collector's key, and the result;
- the statement and the aggregate receipt (§6, §7), verification in §8.1's order with each
  refusal's place, and a room's cell (§9.2);
- delivery-telemetry §3.4's key classes, read from that text, and §4.3's delivery-only reports.

It requires the outcome the corpus states for everything, that every reason of §11 (read from
the extension) and every code it reuses appears, then mutates copies of the corpus, one defect
each, and requires every copy to fail; and it drops each of its own checks in turn and requires
the corpus to catch the weakened verifier.

    python3 telemetry_aggregate_verify.py <live-conformance/aggregate>

Exit status 0 only if every case verifies, every mutated copy fails and every weakened verifier
fails. Python 3.8 or later; standard library only.
"""

import copy
import hashlib
import hmac
import json
import os
import re
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance"))
import _ed25519 as ED  # noqa: E402

EXTENSION = os.path.join(HERE, "..", "extensions", "telemetry-aggregation.md")
TELEMETRY = os.path.join(HERE, "..", "extensions", "delivery-telemetry.md")

SKIP = set()


class Refuse(Exception):
    def __init__(self, reason, at=None):
        super().__init__(reason)
        self.reason, self.at = reason, at


def sha256(b):
    return hashlib.sha256(b).digest()


# ---------------------------------------------------------------------------------------
# Keccak-p[1600, 12] and TurboSHAKE128 (RFC 9861 §2).

ROUND_CONSTANTS = [
    0x0000000000000001, 0x0000000000008082, 0x800000000000808A, 0x8000000080008000,
    0x000000000000808B, 0x0000000080000001, 0x8000000080008081, 0x8000000000008009,
    0x000000000000008A, 0x0000000000000088, 0x0000000080008009, 0x000000008000000A,
    0x000000008000808B, 0x800000000000008B, 0x8000000000008089, 0x8000000000008003,
    0x8000000000008002, 0x8000000000000080, 0x000000000000800A, 0x800000008000000A,
    0x8000000080008081, 0x8000000000008080, 0x0000000080000001, 0x8000000080008008,
]
ROTATIONS = [[0, 36, 3, 41, 18], [1, 44, 10, 45, 2], [62, 6, 43, 15, 61], [28, 55, 25, 21, 56], [27, 20, 39, 8, 14]]
M64 = (1 << 64) - 1
PI = [(y + 5 * ((2 * x + 3 * y) % 5), x + 5 * y, ROTATIONS[x][y]) for x in range(5) for y in range(5)]


def keccak_p12(a):
    for rc in ROUND_CONSTANTS[12:]:
        c = [a[x] ^ a[x + 5] ^ a[x + 10] ^ a[x + 15] ^ a[x + 20] for x in range(5)]
        d = [c[(x - 1) % 5] ^ (((c[(x + 1) % 5] << 1) | (c[(x + 1) % 5] >> 63)) & M64) for x in range(5)]
        a = [a[i] ^ d[i % 5] for i in range(25)]
        b = [0] * 25
        for dst, src, r in PI:
            v = a[src]
            b[dst] = ((v << r) | (v >> (64 - r))) & M64 if r else v
        a = [b[i] ^ ((~b[(i + 1) % 5 + 5 * (i // 5)]) & b[(i + 2) % 5 + 5 * (i // 5)]) for i in range(25)]
        a[0] ^= rc
    return a


class TurboShake128:
    RATE = 168

    def __init__(self, m, d):
        a = [0] * 25
        p = bytearray(m) + bytes([d])
        p += bytes((-len(p)) % self.RATE)
        p[-1] ^= 0x80
        for off in range(0, len(p), self.RATE):
            blk = p[off:off + self.RATE]
            for i in range(21):
                a[i] ^= int.from_bytes(blk[8 * i:8 * i + 8], "little")
            a = keccak_p12(a)
        self.a, self.buf = a, b""

    def read(self, n):
        while len(self.buf) < n:
            self.buf += b"".join(x.to_bytes(8, "little") for x in self.a[:21])
            self.a = keccak_p12(self.a)
        out, self.buf = self.buf[:n], self.buf[n:]
        return out


def turboshake128(m, d, n):
    return TurboShake128(m, d).read(n)


VERSION = 18   # the VDAF draft's §2


def format_dst(algo_class, algo, usage):
    return bytes([VERSION, algo_class]) + algo.to_bytes(4, "big") + usage.to_bytes(2, "big")


class Xof:
    """XofTurboShake128 (VDAF §6.2.1)."""
    SEED_SIZE = 32

    def __init__(self, seed, dst, binder):
        self.t = TurboShake128(len(dst).to_bytes(2, "little") + dst + bytes([len(seed)]) + seed + binder, 1)

    def next(self, n):
        return self.t.read(n)

    def next_vec(self, field, length):
        mask = (1 << (8 * field.SIZE)) - 1   # next_power_of_2(MODULUS) - 1, every bit of the encoding
        out = []
        while len(out) < length:
            x = int.from_bytes(self.next(field.SIZE), "little") & mask
            if x < field.P:
                out.append(x)
        return out

    @classmethod
    def derive_seed(cls, seed, dst, binder):
        return cls(seed, dst, binder).next(cls.SEED_SIZE)

    @classmethod
    def expand_into_vec(cls, field, seed, dst, binder, length):
        return cls(seed, dst, binder).next_vec(field, length)


# ---------------------------------------------------------------------------------------
# Fields (VDAF §6.1); elements are plain integers below P.

class Field64:
    P = 2 ** 32 * 4294967295 + 1
    SIZE = 8
    GEN_ORDER = 2 ** 32
    GEN = pow(7, 4294967295, P)


class Field128:
    P = 2 ** 66 * 4611686018427387897 + 1
    SIZE = 16
    GEN_ORDER = 2 ** 66
    GEN = pow(7, 4611686018427387897, P)


def inv(f, x):
    return pow(x, f.P - 2, f.P)


def nth_root(f, n):
    return pow(f.GEN, f.GEN_ORDER // n, f.P)


def nth_root_powers(f, n):
    w = nth_root(f, n)
    out, x = [], 1
    for _ in range(n):
        out.append(x)
        x = x * w % f.P
    return out


def encode_vec(f, v):
    return b"".join(x.to_bytes(f.SIZE, "little") for x in v)


def decode_vec(f, b):
    if len(b) % f.SIZE:
        raise ValueError("length")
    v = [int.from_bytes(b[i:i + f.SIZE], "little") for i in range(0, len(b), f.SIZE)]
    if any(x >= f.P for x in v):
        raise ValueError("modulus overflow")
    return v


def vadd(f, a, b):
    if len(a) != len(b):
        raise ValueError("mismatched vector sizes")
    return [(x + y) % f.P for x, y in zip(a, b)]


def vsub(f, a, b):
    if len(a) != len(b):
        raise ValueError("mismatched vector sizes")
    return [(x - y) % f.P for x, y in zip(a, b)]


def next_power_of_2(x):
    n = 1
    while n < x:
        n *= 2
    return n


def poly_eval_mono(f, p, x):
    acc = 0
    for c in reversed(p):
        acc = (acc * x + c) % f.P
    return acc


def ntt(f, p, n, set_s=False):
    s = nth_root(f, 2 * n) if set_s else 1
    return [poly_eval_mono(f, p, s * w % f.P) for w in nth_root_powers(f, n)]


def inv_ntt(f, v, n):
    wi = inv(f, nth_root(f, n))
    ninv = inv(f, n)
    out = []
    for j in range(n):
        step = pow(wi, j, f.P)
        acc, x = 0, 1
        for vi in v:
            acc = (acc + vi * x) % f.P
            x = x * step % f.P
        out.append(acc * ninv % f.P)
    return out


def double_evaluations(f, p):
    n = len(p)
    odd = ntt(f, inv_ntt(f, p, n), n, True)
    return [x for pair in zip(p, odd) for x in pair]


def poly_mul(f, p, q):
    a, b = double_evaluations(f, p), double_evaluations(f, q)
    return [x * y % f.P for x, y in zip(a, b)]


def poly_eval_batched(f, polys, x):
    """VDAF §6.1.3.2, as the draft writes it."""
    n = len(polys[0])
    nodes = nth_root_powers(f, n)
    k = 1
    u = [p[0] for p in polys]
    d = (nodes[0] - x) % f.P
    for i in range(1, n):
        k = k * d % f.P
        d = (nodes[i] - x) % f.P
        t = k * nodes[i] % f.P
        for j, p in enumerate(polys):
            u[j] = u[j] * d % f.P
            if i < len(p):
                u[j] = (u[j] + t * p[i]) % f.P
    factor = pow(f.P - 1, n - 1, f.P) * inv(f, n) % f.P
    return [v * factor % f.P for v in u]


def extend_values_to_power_of_2(f, p, n):
    """VDAF §6.1.3.2, as the draft writes it (in place)."""
    x = nth_root_powers(f, n)
    w = [0] * n
    for i in range(len(p)):
        acc = 1
        for j in range(len(p)):
            if i != j:
                acc = acc * (x[i] - x[j]) % f.P
        w[i] = acc
    for k in range(len(p), n):
        for i in range(k):
            w[i] = w[i] * (x[i] - x[k]) % f.P
        y_num, y_den = 0, 1
        for i, v in enumerate(p):
            y_num = (y_num * w[i] + y_den * v) % f.P
            y_den = y_den * w[i] % f.P
        acc = 1
        for j in range(k):
            acc = acc * (x[k] - x[j]) % f.P
        w[k] = acc
        p.append((-w[k] * y_num * inv(f, y_den)) % f.P)


def wire_poly_len(calls):
    return next_power_of_2(1 + calls)


def gadget_poly_len(degree, wire_len):
    return degree * (wire_len - 1) + 1


# Gadgets (Appendix A).

class Mul:
    ARITY, DEGREE = 2, 2

    def eval(self, f, inp):
        return inp[0] * inp[1] % f.P

    def eval_poly(self, f, inp_poly):
        return poly_mul(f, inp_poly[0], inp_poly[1])


class PolyEval:
    ARITY = 1

    def __init__(self, p, num_calls):
        while len(p) > 1 and p[-1] == 0:
            p = p[:-1]
        self.p = p
        self.DEGREE = len(p) - 1
        self.n = next_power_of_2(gadget_poly_len(self.DEGREE, wire_poly_len(num_calls)))

    def eval(self, f, inp):
        return poly_eval_mono(f, [c % f.P for c in self.p], inp[0])

    def eval_poly(self, f, inp_poly):
        m = inv_ntt(f, inp_poly[0], len(inp_poly[0]))
        c = [x % f.P for x in self.p]
        return [poly_eval_mono(f, c, x) for x in ntt(f, m, self.n)]


class ParallelSum:
    def __init__(self, sub, count):
        self.sub, self.count = sub, count
        self.ARITY, self.DEGREE = sub.ARITY * count, sub.DEGREE

    def eval(self, f, inp):
        a = self.sub.ARITY
        return sum(self.sub.eval(f, inp[i * a:(i + 1) * a]) for i in range(self.count)) % f.P

    def eval_poly(self, f, inp_poly):
        out_len = next_power_of_2(gadget_poly_len(self.DEGREE, len(inp_poly[0])))
        out = [0] * out_len
        a = self.sub.ARITY
        for i in range(self.count):
            cur = self.sub.eval_poly(f, inp_poly[i * a:(i + 1) * a])
            for j in range(out_len):
                out[j] = (out[j] + cur[j]) % f.P
        return out


# Validity circuits (§7.4). Each has one gadget; `g` is the gadget call.

def range_encode(f, value, maxm):
    if value > maxm:
        raise ValueError("measurement is too large")
    bits = maxm.bit_length()
    rest_ones = 2 ** (bits - 1) - 1
    last_weight = maxm - rest_ones
    rest, last = (value, 0) if value <= rest_ones else (value - last_weight, 1)
    return [(rest >> l) & 1 for l in range(bits - 1)] + [last]


def range_decode(f, enc, maxm):
    bits = maxm.bit_length()
    last_weight = maxm - (2 ** (bits - 1) - 1)
    d = sum((1 << l) * b for l, b in enumerate(enc[:bits - 1]))
    return (d + last_weight * enc[bits - 1]) % f.P


class Count:
    FIELD, ID = Field64, 1

    def __init__(self):
        self.GADGET, self.CALLS = Mul(), 1
        self.MEAS_LEN, self.JOINT_RAND_LEN, self.OUTPUT_LEN, self.EVAL_OUTPUT_LEN = 1, 0, 1, 1

    def encode(self, m):
        if m not in (0, 1):
            raise ValueError("count")
        return [m]

    def eval(self, meas, jr, num_shares, g):
        return [(g([meas[0], meas[0]]) - meas[0]) % self.FIELD.P]

    def truncate(self, meas):
        return list(meas)


class Sum:
    FIELD, ID = Field64, 2

    def __init__(self, maxm):
        self.max = maxm
        bits = maxm.bit_length()
        self.GADGET, self.CALLS = PolyEval([0, -1, 1], bits), bits
        self.MEAS_LEN, self.JOINT_RAND_LEN, self.OUTPUT_LEN, self.EVAL_OUTPUT_LEN = bits, 0, 1, bits

    def encode(self, m):
        return range_encode(self.FIELD, m, self.max)

    def eval(self, meas, jr, num_shares, g):
        return [g([b]) for b in meas]

    def truncate(self, meas):
        return [range_decode(self.FIELD, meas, self.max)]


def chunked_range_check(f, meas, jr, num_shares, calls, chunk, g):
    shares_inv = inv(f, num_shares)
    out = 0
    for i in range(calls):
        r = jr[i]
        rp = r
        inputs = []
        for j in range(chunk):
            idx = i * chunk + j
            m = meas[idx] if idx < len(meas) else 0
            inputs += [rp * m % f.P, (m - shares_inv) % f.P]
            rp = rp * r % f.P
        out = (out + g(inputs)) % f.P
    return out


class SumVec:
    FIELD, ID = Field128, 3

    def __init__(self, length, maxm, chunk):
        self.length, self.max, self.chunk = length, maxm, chunk
        self.bits = maxm.bit_length()
        self.CALLS = (length * self.bits + chunk - 1) // chunk
        self.GADGET = ParallelSum(Mul(), chunk)
        self.MEAS_LEN, self.JOINT_RAND_LEN, self.OUTPUT_LEN, self.EVAL_OUTPUT_LEN = length * self.bits, self.CALLS, length, 1

    def encode(self, m):
        if len(m) != self.length:
            raise ValueError("length")
        return [b for v in m for b in range_encode(self.FIELD, v, self.max)]

    def eval(self, meas, jr, num_shares, g):
        return [chunked_range_check(self.FIELD, meas, jr, num_shares, self.CALLS, self.chunk, g)]

    def truncate(self, meas):
        return [range_decode(self.FIELD, meas[i * self.bits:(i + 1) * self.bits], self.max) for i in range(self.length)]


class Histogram:
    FIELD, ID = Field128, 4

    def __init__(self, length, chunk):
        self.length, self.chunk = length, chunk
        self.CALLS = (length + chunk - 1) // chunk
        self.GADGET = ParallelSum(Mul(), chunk)
        self.MEAS_LEN, self.JOINT_RAND_LEN, self.OUTPUT_LEN, self.EVAL_OUTPUT_LEN = length, self.CALLS, length, 2

    def encode(self, m):
        if not 0 <= m < self.length:
            raise ValueError("bucket")
        v = [0] * self.length
        v[m] = 1
        return v

    def eval(self, meas, jr, num_shares, g):
        f = self.FIELD
        rc = chunked_range_check(f, meas, jr, num_shares, self.CALLS, self.chunk, g)
        sc = (sum(meas) - inv(f, num_shares)) % f.P
        return [rc, sc]

    def truncate(self, meas):
        return list(meas)


# The FLP (§7.3).

def flp_lens(v):
    g = v.GADGET
    p = wire_poly_len(v.CALLS)
    return {
        "prove_rand": g.ARITY,
        "query_rand": 1 + (v.EVAL_OUTPUT_LEN if v.EVAL_OUTPUT_LEN > 1 else 0),
        "proof": g.ARITY + gadget_poly_len(g.DEGREE, p),
        "verifier": 1 + g.ARITY + 1,
    }


def prove(v, meas, prove_rand, joint_rand):
    f, g = v.FIELD, v.GADGET
    p = wire_poly_len(v.CALLS)
    wires = [[prove_rand[j]] + [0] * (p - 1) for j in range(g.ARITY)]
    k = [0]

    def call(inp):
        k[0] += 1
        for j, x in enumerate(inp):
            wires[j][k[0]] = x
        return g.eval(f, inp)

    v.eval(meas, joint_rand, 1, call)
    return prove_rand[:g.ARITY] + g.eval_poly(f, wires)[:gadget_poly_len(g.DEGREE, p)]


def query(v, meas, proof, query_rand, joint_rand, num_shares):
    f, g = v.FIELD, v.GADGET
    p = wire_poly_len(v.CALLS)
    gp_len = gadget_poly_len(g.DEGREE, p)
    seeds, poly = proof[:g.ARITY], list(proof[g.ARITY:g.ARITY + gp_len])
    extend_values_to_power_of_2(f, poly, next_power_of_2(len(poly)))
    size = next_power_of_2(gp_len)
    while len(poly) < size:
        poly = double_evaluations(f, poly)
    step = size // p
    wires = [[s] + [0] * (p - 1) for s in seeds]
    k = [0]

    def call(inp):
        k[0] += 1
        for j, x in enumerate(inp):
            wires[j][k[0]] = x
        return poly[k[0] * step]

    out = v.eval(meas, joint_rand, num_shares, call)
    if v.EVAL_OUTPUT_LEN > 1:
        r = query_rand[:v.EVAL_OUTPUT_LEN]
        val = sum(a * b for a, b in zip(r, out)) % f.P
        t = query_rand[v.EVAL_OUTPUT_LEN]
    else:
        val, t = out[0], query_rand[0]
    if pow(t, p, f.P) == 1:
        raise ValueError("test point is a root of unity")
    return [val] + poly_eval_batched(f, wires, t) + poly_eval_batched(f, [poly], t)


def decide(v, verifier):
    if verifier[0] != 0:
        return False
    a = v.GADGET.ARITY
    return v.GADGET.eval(v.FIELD, verifier[1:1 + a]) == verifier[1 + a]


# Prio3 (§7.2), one proof, any number of shares.

class Prio3:
    def __init__(self, valid, shares=2):
        self.v, self.shares, self.f = valid, shares, valid.FIELD
        self.L = flp_lens(valid)
        self.jr = valid.JOINT_RAND_LEN > 0
        self.rand_size = 32 * shares * (2 if self.jr else 1)

    def dst(self, usage, ctx):
        return format_dst(0, self.v.ID, usage) + ctx

    def helper_meas_share(self, ctx, agg_id, share):
        return Xof.expand_into_vec(self.f, share, self.dst(1, ctx), bytes([agg_id]), self.v.MEAS_LEN)

    def helper_proofs_share(self, ctx, agg_id, share):
        return Xof.expand_into_vec(self.f, share, self.dst(2, ctx), bytes([1, agg_id]), self.L["proof"])

    def joint_rand_part(self, ctx, agg_id, blind, meas_share, nonce):
        return Xof.derive_seed(blind, self.dst(7, ctx), bytes([agg_id]) + nonce + encode_vec(self.f, meas_share))

    def joint_rand_seed(self, ctx, parts):
        return Xof.derive_seed(bytes(32), self.dst(6, ctx), b"".join(parts))

    def joint_rands(self, ctx, seed):
        return Xof.expand_into_vec(self.f, seed, self.dst(3, ctx), bytes([1]), self.v.JOINT_RAND_LEN)

    def shard(self, ctx, meas, nonce, rand):
        """shard (§7.2.1) of an encoded measurement: (public share, input shares), encoded."""
        seeds = [rand[i:i + 32] for i in range(0, len(rand), 32)]
        n = self.shares
        if self.jr:
            hs, hb = [seeds[2 * i] for i in range(n - 1)], [seeds[2 * i + 1] for i in range(n - 1)]
            lb, ps = seeds[2 * (n - 1)], seeds[2 * (n - 1) + 1]
        else:
            hs, hb, lb, ps = seeds[:n - 1], [None] * (n - 1), None, seeds[n - 1]
        lm, parts = list(meas), []
        for j in range(n - 1):
            hm = self.helper_meas_share(ctx, j + 1, hs[j])
            lm = vsub(self.f, lm, hm)
            if self.jr:
                parts.append(self.joint_rand_part(ctx, j + 1, hb[j], hm, nonce))
        jrs = []
        if self.jr:
            parts.insert(0, self.joint_rand_part(ctx, 0, lb, lm, nonce))
            jrs = self.joint_rands(ctx, self.joint_rand_seed(ctx, parts))
        pr = Xof.expand_into_vec(self.f, ps, self.dst(4, ctx), bytes([1]), self.L["prove_rand"])
        proof = prove(self.v, meas, pr, jrs)
        for j in range(n - 1):
            proof = vsub(self.f, proof, self.helper_proofs_share(ctx, j + 1, hs[j]))
        shares = [encode_vec(self.f, lm) + encode_vec(self.f, proof) + (lb if self.jr else b"")]
        shares += [hs[j] + (hb[j] if self.jr else b"") for j in range(n - 1)]
        return (b"".join(parts) if self.jr else b""), shares

    def decode_input_share(self, agg_id, b):
        bl = 32 if self.jr else 0
        if agg_id == 0:
            ml, pl = self.v.MEAS_LEN * self.f.SIZE, self.L["proof"] * self.f.SIZE
            if len(b) != ml + pl + bl:
                raise ValueError("input share")
            return decode_vec(self.f, b[:ml]), decode_vec(self.f, b[ml:ml + pl]), b[ml + pl:] or None
        if len(b) != 32 + bl:
            raise ValueError("input share")
        return b[:32], b[32:] or None

    def meas_share(self, ctx, agg_id, input_share):
        s = self.decode_input_share(agg_id, input_share)
        return s[0] if agg_id == 0 else self.helper_meas_share(ctx, agg_id, s[0])

    def verify_init(self, vk, ctx, agg_id, nonce, public, input_share):
        """verify_init (§7.2.2): (state, encoded verifier share)."""
        s = self.decode_input_share(agg_id, input_share)
        if agg_id == 0:
            meas, proofs, blind = s
        else:
            meas, proofs, blind = self.helper_meas_share(ctx, agg_id, s[0]), self.helper_proofs_share(ctx, agg_id, s[0]), s[1]
        out = self.v.truncate(meas)
        jrs, corrected, part = [], None, None
        if self.jr:
            if len(public) != 32 * self.shares:
                raise ValueError("public share")
            parts = [public[i:i + 32] for i in range(0, len(public), 32)]
            part = self.joint_rand_part(ctx, agg_id, blind, meas, nonce)
            parts[agg_id] = part
            corrected = self.joint_rand_seed(ctx, parts)
            jrs = self.joint_rands(ctx, corrected)
        elif public:
            raise ValueError("public share")
        qr = Xof.expand_into_vec(self.f, vk, self.dst(5, ctx), bytes([1]) + nonce, self.L["query_rand"])
        ver = query(self.v, meas, proofs, qr, jrs, self.shares)
        return (out, corrected), encode_vec(self.f, ver) + (part if self.jr else b"")

    def message(self, ctx, shares):
        """verifier_shares_to_message (§7.2.2): the encoded verifier message."""
        vl = self.L["verifier"] * self.f.SIZE
        total = [0] * self.L["verifier"]
        parts = []
        for s in shares:
            if len(s) != vl + (32 if self.jr else 0):
                raise ValueError("verifier share")
            total = vadd(self.f, total, decode_vec(self.f, s[:vl]))
            parts.append(s[vl:])
        if "proof" not in SKIP and not decide(self.v, total):
            raise ValueError("proof verifier check failed")
        return self.joint_rand_seed(ctx, parts) if self.jr else b""

    def verify_next(self, state, msg):
        out, corrected = state
        if (msg if self.jr else None) != corrected or (not self.jr and msg):
            raise ValueError("joint randomness check failed")
        return encode_vec(self.f, out)

    def agg(self, shares):
        acc = [0] * self.v.OUTPUT_LEN
        for s in shares:
            v = decode_vec(self.f, s)
            if len(v) != self.v.OUTPUT_LEN:
                raise ValueError("aggregate share")
            acc = vadd(self.f, acc, v)
        return acc


def prio3_of(name, doc):
    s = doc["shares"]
    if name.startswith("Prio3Count_"):
        return Prio3(Count(), s)
    if name.startswith("Prio3SumVec_"):
        return Prio3(SumVec(doc["length"], doc["max_measurement"], doc["chunk_length"]), s)
    if name.startswith("Prio3Sum_"):
        return Prio3(Sum(doc["max_measurement"]), s)
    if name.startswith("Prio3Histogram_"):
        return Prio3(Histogram(doc["length"], doc["chunk_length"]), s)
    return None


def run_vdaf_file(name, doc):
    """One of the draft's vector files (Appendix C.1): the failures."""
    fails = []
    p = prio3_of(name, doc)
    if p is None:
        return ["%s: not a variant this verifier implements" % name]
    ctx, vk = bytes.fromhex(doc["ctx"]), bytes.fromhex(doc["verify_key"])
    reps = doc["reports"]
    states = {}
    for op in doc["operations"]:
        kind, ok = op["operation"], op["success"]
        i, j = op.get("report_index"), op.get("aggregator_id")
        try:
            if kind == "shard":
                r = reps[i]
                ps, iss = p.shard(ctx, p.v.encode(r["measurement"]), bytes.fromhex(r["nonce"]), bytes.fromhex(r["rand"]))
                got = ps.hex() == r["public_share"] and [x.hex() for x in iss] == r["input_shares"]
            elif kind == "verify_init":
                r = reps[i]
                st, vs = p.verify_init(vk, ctx, j, bytes.fromhex(r["nonce"]), bytes.fromhex(r["public_share"]), bytes.fromhex(r["input_shares"][j]))
                states[(i, j)] = st
                got = vs.hex() == r["verifier_shares"][0][j]
            elif kind == "verifier_shares_to_message":
                r = reps[i]
                got = p.message(ctx, [bytes.fromhex(x) for x in r["verifier_shares"][0]]).hex() == r["verifier_messages"][0]
            elif kind == "verify_next":
                r = reps[i]
                got = p.verify_next(states[(i, j)], bytes.fromhex(r["verifier_messages"][0])).hex() == r["out_shares"][j]
            elif kind == "aggregate":
                got = encode_vec(p.f, p.agg([bytes.fromhex(r["out_shares"][j]) for r in reps])).hex() == doc["agg_shares"][j]
            elif kind == "unshard":
                total = p.agg([bytes.fromhex(x) for x in doc["agg_shares"]])
                res = total[0] if isinstance(doc["agg_result"], int) else total
                got = res == doc["agg_result"]
            else:
                got = False
            if not (ok and got):
                fails.append("%s: %s %s %s gives other bytes" % (name, kind, i, j))
        except (ValueError, KeyError, IndexError):
            if ok:
                fails.append("%s: %s %s %s refused" % (name, kind, i, j))
    return fails


def run_xof(doc):
    seed, dst, binder = (bytes.fromhex(doc[k]) for k in ("seed", "dst", "binder"))
    fails = []
    if Xof.derive_seed(seed, dst, binder).hex() != doc["derived_seed"]:
        fails.append("XofTurboShake128: derived seed")
    if encode_vec(Field128, Xof.expand_into_vec(Field128, seed, dst, binder, doc["length"])).hex() != doc["expanded_vec_field128"]:
        fails.append("XofTurboShake128: expansion")
    return fails


# ---------------------------------------------------------------------------------------
# X25519 (RFC 7748 §5), HKDF-SHA256 (RFC 5869), AES-128 (FIPS 197), GCM (SP 800-38D), HPKE
# base mode (RFC 9180).

P25519 = 2 ** 255 - 19


def x25519(k, u):
    k = bytearray(k)
    k[0] &= 248
    k[31] &= 127
    k[31] |= 64
    k = int.from_bytes(k, "little")
    x1 = int.from_bytes(u, "little") & ((1 << 255) - 1)
    x2, z2, x3, z3, swap = 1, 0, x1, 1, 0
    for t in reversed(range(255)):
        kt = (k >> t) & 1
        swap ^= kt
        if swap:
            x2, x3, z2, z3 = x3, x2, z3, z2
        swap = kt
        a, b, c, d = x2 + z2, x2 - z2, x3 + z3, x3 - z3
        aa, bb = a * a % P25519, b * b % P25519
        e = (aa - bb) % P25519
        da, cb = d * a % P25519, c * b % P25519
        x3 = (da + cb) ** 2 % P25519
        z3 = x1 * (da - cb) ** 2 % P25519
        x2 = aa * bb % P25519
        z2 = e * (aa + 121665 * e) % P25519
    if swap:
        x2, z2 = x3, z3
    return (x2 * pow(z2, P25519 - 2, P25519) % P25519).to_bytes(32, "little")


BASE = (9).to_bytes(32, "little")


def hkdf_extract(salt, ikm):
    return hmac.new(salt, ikm, hashlib.sha256).digest()


def hkdf_expand(prk, info, n):
    out, t, i = b"", b"", 1
    while len(out) < n:
        t = hmac.new(prk, t + info + bytes([i]), hashlib.sha256).digest()
        out += t
        i += 1
    return out[:n]


def _xt(a):
    return ((a << 1) ^ 0x1B) & 0xFF if a & 0x80 else a << 1


def _mul(a, b):
    r = 0
    while b:
        if b & 1:
            r ^= a
        a, b = _xt(a), b >> 1
    return r


def _sbox():
    inv_ = [0] * 256
    for a in range(1, 256):
        for b in range(1, 256):
            if _mul(a, b) == 1:
                inv_[a] = b
                break
    s = []
    for a in range(256):
        x = inv_[a]
        y = x
        for _ in range(4):
            x = ((x << 1) | (x >> 7)) & 0xFF
            y ^= x
        s.append(y ^ 0x63)
    return s


SBOX = _sbox()


def aes_keys(key):
    w = [list(key[4 * i:4 * i + 4]) for i in range(4)]
    rcon = 1
    for i in range(4, 44):
        t = list(w[i - 1])
        if i % 4 == 0:
            t = [SBOX[x] for x in t[1:] + t[:1]]
            t[0] ^= rcon
            rcon = _xt(rcon)
        w.append([a ^ b for a, b in zip(w[i - 4], t)])
    return [sum(w[4 * r:4 * r + 4], []) for r in range(11)]


def aes_block(rk, blk):
    s = [a ^ b for a, b in zip(blk, rk[0])]
    for r in range(1, 11):
        s = [SBOX[x] for x in s]
        s = [s[(i + 4 * (i % 4)) % 16] for i in range(16)]   # ShiftRows, column-major state
        if r < 10:
            m = []
            for c in range(4):
                a = s[4 * c:4 * c + 4]
                m += [_mul(a[0], 2) ^ _mul(a[1], 3) ^ a[2] ^ a[3], a[0] ^ _mul(a[1], 2) ^ _mul(a[2], 3) ^ a[3],
                      a[0] ^ a[1] ^ _mul(a[2], 2) ^ _mul(a[3], 3), _mul(a[0], 3) ^ a[1] ^ a[2] ^ _mul(a[3], 2)]
            s = m
        s = [a ^ b for a, b in zip(s, rk[r])]
    return bytes(s)


R_GCM = 0xE1 << 120


def gf_mul(x, y):
    z, v = 0, y
    for i in range(127, -1, -1):
        if (x >> i) & 1:
            z ^= v
        v = (v >> 1) ^ R_GCM if v & 1 else v >> 1
    return z


def ghash(h, data):
    y = 0
    for i in range(0, len(data), 16):
        y = gf_mul(y ^ int.from_bytes(data[i:i + 16], "big"), h)
    return y


def _pad16(b):
    return b + bytes((-len(b)) % 16)


def gcm_open(key, nonce, aad, ct):
    if len(ct) < 16:
        raise ValueError("short")
    rk = aes_keys(key)
    h = int.from_bytes(aes_block(rk, bytes(16)), "big")
    j0 = nonce + b"\x00\x00\x00\x01"
    c, tag = ct[:-16], ct[-16:]
    s = ghash(h, _pad16(aad) + _pad16(c) + (8 * len(aad)).to_bytes(8, "big") + (8 * len(c)).to_bytes(8, "big"))
    want = (int.from_bytes(aes_block(rk, j0), "big") ^ s).to_bytes(16, "big")
    if not hmac.compare_digest(want, tag):
        raise ValueError("tag")
    out = bytearray()
    ctr = int.from_bytes(j0[12:], "big")
    for i in range(0, len(c), 16):
        ctr = (ctr + 1) & 0xFFFFFFFF
        ks = aes_block(rk, nonce + ctr.to_bytes(4, "big"))
        out += bytes(a ^ b for a, b in zip(c[i:i + 16], ks))
    return bytes(out)


KEM_SUITE = b"KEM" + (0x0020).to_bytes(2, "big")
HPKE_SUITE = b"HPKE" + (0x0020).to_bytes(2, "big") + (1).to_bytes(2, "big") + (1).to_bytes(2, "big")


def labeled_extract(suite, salt, label, ikm):
    return hkdf_extract(salt, b"HPKE-v1" + suite + label + ikm)


def labeled_expand(suite, prk, label, info, n):
    return hkdf_expand(prk, n.to_bytes(2, "big") + b"HPKE-v1" + suite + label + info, n)


def hpke_open(enc, sk, info, aad, ct):
    dh = x25519(sk, enc)
    if dh == bytes(32):
        raise ValueError("zero")
    ctx = enc + x25519(sk, BASE)
    shared = labeled_expand(KEM_SUITE, labeled_extract(KEM_SUITE, b"", b"eae_prk", dh), b"shared_secret", ctx, 32)
    ksc = b"\x00" + labeled_extract(HPKE_SUITE, b"", b"psk_id_hash", b"") + labeled_extract(HPKE_SUITE, b"", b"info_hash", info)
    secret = labeled_extract(HPKE_SUITE, shared, b"secret", b"")
    key = labeled_expand(HPKE_SUITE, secret, b"key", ksc, 16)
    nonce = labeled_expand(HPKE_SUITE, secret, b"base_nonce", ksc, 12)
    return gcm_open(key, nonce, aad, ct)


def crypto_self_test():
    """RFC 9861 §5's TurboSHAKE128 vectors, RFC 9180 A.1.1 (DeriveKeyPair, OpenBase), RFC
    7748 §5.2's first X25519 vector: the checks that fail."""
    fails = []
    if turboshake128(b"", 0x1F, 32).hex() != "1e415f1c5983aff2169217277d17bb538cd945a397ddec541f1ce41af2c1b74c":
        fails.append("TurboSHAKE128 empty")
    if turboshake128(bytes(i % 251 for i in range(17 ** 3)), 0x1F, 32).hex() != "d4976eb56bcf118520582b709f73e1d6853e001fdaf80e1b13e0d0599d5fb372":
        fails.append("TurboSHAKE128 ptn(17^3)")
    if turboshake128(b"\xff" * 3, 0x01, 32).hex() != "bf323f940494e88ee1c540fe660be8a0c93f43d15ec006998462fa994eed5dab":
        fails.append("TurboSHAKE128 D=01")
    k = bytes.fromhex("a546e36bf0527c9d3b16154b82465edd62144c0ac1fc5a18506a2244ba449ac4")
    u = bytes.fromhex("e6db6867583030db3594c1a424b15f7c726624ec26b3353b10a903a6d0ab1c4c")
    if x25519(k, u).hex() != "c3da55379de9c6908e94ea4df28d084f32eccf03491c71f754b4075577a28552":
        fails.append("X25519")
    ikm_r = bytes.fromhex("6db9df30aa07dd42ee5e8181afdb977e538f5e1fec8a06223f33f7013e525037")
    sk_r = labeled_expand(KEM_SUITE, labeled_extract(KEM_SUITE, b"", b"dkp_prk", ikm_r), b"sk", b"", 32)
    if sk_r.hex() != "4612c550263fc8ad58375df3f557aac531d26850903e55a9f23f21d8534e8ac8":
        fails.append("HPKE DeriveKeyPair")
    try:
        pt = hpke_open(bytes.fromhex("37fda3567bdbd628e88668c3c8d7e97d1d1253b6d4ea6d44c150f741f1bf4431"), sk_r,
                       bytes.fromhex("4f6465206f6e2061204772656369616e2055726e"), bytes.fromhex("436f756e742d30"),
                       bytes.fromhex("f938558b5d72f1a23810b4be2ab4f84331acc02fc97babc53a52ae8218a355a96d8770ac83d07bea87e13c512a"))
        if pt.hex() != "4265617574792069732074727574682c20747275746820626561757479":
            fails.append("HPKE OpenBase plaintext")
    except ValueError:
        fails.append("HPKE OpenBase")
    return fails


# ---------------------------------------------------------------------------------------
# DAP's messages (draft-ietf-ppm-dap-20 §4), read strictly.

class Rd:
    def __init__(self, b):
        self.b, self.i = b, 0

    def take(self, n):
        if self.i + n > len(self.b):
            raise ValueError("truncated")
        x = self.b[self.i:self.i + n]
        self.i += n
        return x

    def u(self, n):
        return int.from_bytes(self.take(n), "big")

    def opaque(self, n):
        return self.take(self.u(n))

    def done(self):
        return self.i == len(self.b)

    def end(self):
        if not self.done():
            raise ValueError("trailing")


def read_exts(r):
    body = Rd(r.opaque(2))
    out = []
    while not body.done():
        out.append((body.u(2), body.opaque(2)))
    return out


def read_task(r):
    t = {"task_info": r.opaque(1), "leader": r.opaque(2), "helper": r.opaque(2), "time_precision": r.u(8),
         "min_batch_size": r.u(8), "batch_mode": r.u(1), "batch_config": r.opaque(2), "vdaf_type": r.u(4),
         "vdaf_config": r.opaque(2), "extensions": read_exts(r)}
    if not t["leader"] or not t["helper"]:
        raise ValueError("empty URL")
    return t


def read_ct(r):
    c = {"config_id": r.u(1), "enc": r.opaque(2), "payload": r.opaque(4)}
    if not c["enc"] or not c["payload"]:
        raise ValueError("empty")
    return c


def write_ct(c):
    return bytes([c["config_id"]]) + len(c["enc"]).to_bytes(2, "big") + c["enc"] + len(c["payload"]).to_bytes(4, "big") + c["payload"]


def read_metadata(r):
    start = r.i
    m = {"id": r.take(16), "time": r.u(8), "exts": read_exts(r)}
    m["raw"] = r.b[start:r.i]
    return m


def read_report(r):
    m = read_metadata(r)
    return {"m": m, "public": r.opaque(4), "leader": read_ct(r), "helper": read_ct(r)}


def read_collection(b):
    r = Rd(b)
    c = {"batch_mode": r.u(1), "config": r.opaque(2), "agg_param": r.opaque(4), "extensions": read_exts(r)}
    r.end()
    return c


def interval_of(b):
    r = Rd(b)
    i = (r.u(8), r.u(8))
    r.end()
    return i


def within(a, b):
    return a[0] >= b[0] and a[0] + a[1] <= b[0] + b[1]


def u64(x):
    return x.to_bytes(8, "big")


# ---------------------------------------------------------------------------------------
# The extension: measures (§3), tasks (§4), statements (§6), receipts (§7), verification (§8),
# cells (§9.2).

FLOOR = 50
# Measure 5's configuration is impression-aggregates §5.3's, against a catalogue (§4.3 rule 3).
MEASURES = {1: "outcomes", 2: "stalls", 3: "durations", 4: "start_time", 5: "impressions"}
START_EDGES = [250, 500, 1000, 2000, 4000, 8000]


def measure_vdaf(code):
    return {1: lambda: Prio3(SumVec(4, 1, 2)), 2: lambda: Prio3(Sum(65535)),
            3: lambda: Prio3(SumVec(3, 4294967295, 10)), 4: lambda: Prio3(Histogram(8, 3))}[code]()


def dap_config(p):
    v = p.v
    if isinstance(v, Count):
        return b""
    if isinstance(v, Sum):
        return u64(v.max)
    if isinstance(v, SumVec):
        return v.length.to_bytes(4, "big") + u64(v.max) + v.chunk.to_bytes(4, "big")
    return v.length.to_bytes(4, "big") + v.chunk.to_bytes(4, "big")


def measurement(code, m):
    """§3.1: a session's measurement, before the VDAF encodes it."""
    if code == 1:
        return [int(m["started"]), int(m["start_failed"]), int(m["exit_before_start"]), int(m["vst_ms"] is not None)]
    if code == 2:
        return [min(m["stalls"], 65535)]
    if code == 3:
        return [min(x, 4294967295) for x in (m["stall_ms"], m["engaged_ms"], m["vst_ms"] or 0)]
    v = m["vst_ms"]
    if v is None:
        return [7]
    return [next((i for i, e in enumerate(START_EDGES) if v < e), 6)]


def vdaf_encode(code, p, meas):
    return p.v.encode(meas if code in (1, 3) else meas[0])


def task_info(b):
    if len(b) != 153 or b[:8] != b"wai-qoe1":
        return None
    return {"measure": b[8], "scope": b[9:41], "salt": b[41:57], "leader": b[57:89], "helper": b[89:121], "collector": b[121:153]}


def task_id(cfg):
    return sha256(b"wai:telemetry-aggregate/task\x00" + cfg)


def task_fields(cfg_bytes):
    """§4.3 rule 1: the task's fields and its info."""
    try:
        r = Rd(cfg_bytes)
        t = read_task(r)
        r.end()
    except ValueError:
        raise Refuse("task_invalid", "task")
    info = task_info(t["task_info"])
    if "task-rules" not in SKIP:
        if info is None or t["time_precision"] == 0 or t["batch_mode"] != 1 or t["batch_config"]:
            raise Refuse("task_invalid", "task")
        types = [e[0] for e in t["extensions"]]
        if any(e[0] != 1 or len(e[1]) != 16 for e in t["extensions"]) or types != sorted(set(types)):
            raise Refuse("task_invalid", "task")
    return t, info


def check_task(cfg_bytes, measure5=None):
    """§4.3, in order: the task's fields, its info and its Prio3. For measure 5, rule 3 is
    impression-aggregates §5.3 steps 2 to 4, given as `measure5(t, info)`, which returns the
    task's Prio3 or refuses; given none, the task is refused `catalogue_mismatch`."""
    t, info = task_fields(cfg_bytes)
    if info is None or info["measure"] not in MEASURES:
        raise Refuse("measure_unknown", "task")
    if info["measure"] == 5:
        if measure5 is None:
            raise Refuse("catalogue_mismatch", "task")
        p = measure5(t, info)
    else:
        p = measure_vdaf(info["measure"])
        if "measure" not in SKIP and (t["vdaf_type"] != p.v.ID or t["vdaf_config"] != dap_config(p)):
            raise Refuse("measure_mismatch", "task")
    if "floor-rule" not in SKIP and t["min_batch_size"] < FLOOR:
        raise Refuse("floor_too_low", "task")
    if "one-key" not in SKIP and info["leader"] == info["helper"]:
        raise Refuse("aggregator_alone", "task")
    return t, info, p, task_id(cfg_bytes)


def read_statement(b):
    try:
        r = Rd(b)
        if r.take(4) != b"WQS1" or r.u(1) != 1:
            raise ValueError("magic")
        s = {"role": r.u(1), "task_id": r.take(32), "collection": r.take(32), "batch": (r.u(8), r.u(8)),
             "count": r.u(8), "checksum": r.take(32), "agg_share": r.opaque(4), "signer": r.take(32)}
        s["body"] = b[:r.i]
        s["sig"] = r.take(64)
        r.end()
    except ValueError:
        raise Refuse("statement_invalid")
    if s["role"] not in (2, 3):
        raise Refuse("statement_invalid")
    return s


def read_receipt(b):
    r = Rd(b)
    try:
        if r.take(4) != b"WQA1":
            raise Refuse("malformed", "receipt")
        if r.u(1) != 1:
            raise Refuse("unknown_version", "receipt")
        rc = {"task": r.opaque(2), "collection": r.opaque(4), "interval": (r.u(8), r.u(8)),
              "leader": r.opaque(4), "helper": r.opaque(4)}
        rc["result"] = [int.from_bytes(r.take(16), "big") for _ in range(r.u(2))]
        rc["signer"] = r.take(32)
        rc["body"] = b[:r.i]
        rc["sig"] = r.take(64)
    except ValueError:
        raise Refuse("truncated", "receipt")
    if not r.done():
        raise Refuse("malformed", "receipt")
    return rc


def sig_check(key, domain, body, sig, at):
    e = ED.verify(key, domain + body, sig)
    if e is not None:
        raise Refuse(e, at)


def verify_receipt(b, reports, task_check=check_task):
    """§8.1, in order; step 2 is `task_check` (impression-aggregates §8.1 gives its own)."""
    rc = read_receipt(b)
    t, info, p, tid = task_check(rc["task"])
    if "collector-signer" not in SKIP and rc["signer"] != info["collector"]:
        raise Refuse("wrong_signer", "receipt")
    if "collector-sig" not in SKIP:
        sig_check(rc["signer"], b"wai:telemetry-aggregate/receipt\x01", rc["body"], rc["sig"], "receipt")
    try:
        c = read_collection(rc["collection"])
        q = interval_of(c["config"]) if c["batch_mode"] == 1 else None
    except ValueError:
        raise Refuse("collection_invalid", "collection")
    if "collection" not in SKIP and (q is None or q[1] == 0 or c["agg_param"] or c["extensions"]):
        raise Refuse("collection_invalid", "collection")
    if not rc["leader"] or not rc["helper"]:
        raise Refuse("aggregator_alone", "statements")
    sts = []
    for raw, at in ((rc["leader"], "leader"), (rc["helper"], "helper")):
        try:
            sts.append(read_statement(raw))
        except Refuse as e:
            raise Refuse(e.reason, at)
    l, h = sts
    if "two-aggregators" not in SKIP and (l["signer"] == h["signer"] or l["role"] == h["role"]):
        raise Refuse("aggregator_alone", "statements")
    for s, role, key, at in ((l, 2, info["leader"], "leader"), (h, 3, info["helper"], "helper")):
        if "statement-signer" not in SKIP and (s["role"] != role or s["signer"] != key):
            raise Refuse("wrong_signer", at)
        if "statement-sig" not in SKIP:
            sig_check(s["signer"], b"wai:telemetry-aggregate/share\x01", s["body"], s["sig"], at)
    for s, at in ((l, "leader"), (h, "helper")):
        if "binding" not in SKIP and (s["task_id"] != tid or s["collection"] != sha256(rc["collection"])):
            raise Refuse("task_mismatch", at)
    for s, at in ((l, "leader"), (h, "helper")):
        if "batch" not in SKIP and (s["batch"][1] == 0 or not within(s["batch"], q)):
            raise Refuse("batch_invalid", at)
    if "batch" not in SKIP and (rc["interval"][1] == 0 or not within(rc["interval"], l["batch"])):
        raise Refuse("batch_invalid", "interval")
    if "agreement" not in SKIP and (l["batch"], l["count"], l["checksum"]) != (h["batch"], h["count"], h["checksum"]):
        raise Refuse("statements_disagree", "statements")
    if "floor" not in SKIP and l["count"] < t["min_batch_size"]:
        raise Refuse("batch_too_small", "statements")
    shares = []
    for s, at in ((l, "leader"), (h, "helper")):
        try:
            shares.append(p.agg([s["agg_share"]]))
        except ValueError:
            if "share" not in SKIP:
                raise Refuse("share_invalid", at)
            shares.append([0] * p.v.OUTPUT_LEN)
    result = vadd(p.f, shares[0], shares[1])
    if "result" not in SKIP and result != rc["result"]:
        raise Refuse("result_mismatch", "result")
    if reports is not None and "reports" not in SKIP:
        ids, ck = set(), bytes(32)
        for g in reports:
            gid = bytes.fromhex(g["id"])
            if bytes.fromhex(g["task"]) != tid or not rc["interval"][0] <= g["time"] < sum(rc["interval"]) or gid in ids:
                raise Refuse("batch_mismatch", "reports")
            ids.add(gid)
            ck = bytes(a ^ b for a, b in zip(ck, sha256(gid)))
        if len(reports) != l["count"] or ck != l["checksum"]:
            raise Refuse("batch_mismatch", "reports")
    return {
        "outcome": "accept", "digest": sha256(rc["body"]).hex(), "task_id": tid.hex(), "measure": MEASURES[info["measure"]],
        "vdaf": {1: "Prio3Count", 2: "Prio3Sum", 3: "Prio3SumVec", 4: "Prio3Histogram"}[p.v.ID], "vdaf_config": t["vdaf_config"].hex(),
        "scope": info["scope"].hex(), "time_precision": t["time_precision"], "min_batch_size": t["min_batch_size"],
        "batch": {"start": l["batch"][0], "duration": l["batch"][1]},
        "interval": {"start": rc["interval"][0], "duration": rc["interval"][1]}, "report_count": l["count"],
        "result": rc["result"], "leader": info["leader"].hex(), "helper": info["helper"].hex(), "collector": info["collector"].hex(),
    }


def receipt_outcome(v):
    try:
        return verify_receipt(bytes.fromhex(v["receipt"]), v["reports"])
    except Refuse as e:
        return {"outcome": "refuse", "reason": e.reason, "at": e.at}


CELL_FIELDS = [("started", "outcomes", 0), ("start_failed", "outcomes", 1), ("exit_before_start", "outcomes", 2),
               ("stalls", "stalls", 0), ("stall_ms", "durations", 0), ("engaged_ms", "durations", 1),
               ("vst_count", "outcomes", 3), ("vst_ms", "durations", 2)]


def cell_outcome(doc, v):
    acc = []
    for n in v["receipts"]:
        r = next((x for x in doc["receipts"] if x["name"] == n), None)
        if r is None:
            return {"outcome": "refuse", "reason": "cell_mismatch"}
        o = receipt_outcome(r)
        if o["outcome"] != "accept":
            return {"outcome": "refuse", "reason": o["reason"]}
        acc.append(o)
    by = {}
    for a in acc:
        by.setdefault(a["measure"], []).append(a)
    if "cell" not in SKIP:
        if len(acc) != 3 or sorted(by) != ["durations", "outcomes", "stalls"] or any(len(x) != 1 for x in by.values()):
            return {"outcome": "refuse", "reason": "cell_mismatch"}
        if len({a["scope"] for a in acc}) != 1 or acc[0]["scope"] == "00" * 32 or len({json.dumps(a["batch"]) for a in acc}) != 1:
            return {"outcome": "refuse", "reason": "cell_mismatch"}
    def pick(m):
        return (by.get(m) or acc)[0]
    try:
        return {"outcome": "accept", "scope": pick("outcomes")["scope"], "batch": pick("outcomes")["batch"],
                "counts": [pick(m)["report_count"] for m in ("outcomes", "stalls", "durations")],
                "cell": {f: pick(m)["result"][i] for f, m, i in CELL_FIELDS}}
    except IndexError:
        return {"outcome": "refuse", "reason": "cell_mismatch"}


def plaintext_cell(sessions):
    """live-metering §4.7 over the sessions' metrics."""
    ms = [s["metrics"] for s in sessions]
    return {"started": sum(m["started"] for m in ms), "start_failed": sum(m["start_failed"] for m in ms),
            "exit_before_start": sum(m["exit_before_start"] for m in ms), "stalls": sum(m["stalls"] for m in ms),
            "stall_ms": sum(m["stall_ms"] for m in ms), "engaged_ms": sum(m["engaged_ms"] for m in ms),
            "vst_count": sum(m["vst_ms"] is not None for m in ms), "vst_ms": sum(m["vst_ms"] or 0 for m in ms)}


# ---------------------------------------------------------------------------------------
# A flow, re-run (§5, DAP §4.4 to §4.6).

def input_share_aad(tid, cfg, m, public):
    return tid + cfg + m["raw"] + len(public).to_bytes(4, "big") + public


def agg_share_aad(tid, cfg, coll):
    return tid + cfg + coll


def open_share(sk, role, aad, ct):
    pt = hpke_open(ct["enc"], sk, b"dap-20 input share" + bytes([1, role]), aad, ct["payload"])
    r = Rd(pt)
    exts = read_exts(r)
    payload = r.opaque(4)
    r.end()
    if exts or not payload:
        raise ValueError("extensions")
    return payload


def check_flow(doc, f, fails, quiet):
    name = f["name"]
    K = doc["keys"]

    def bad(what):
        fails.append("flow %s: %s" % (name, what))

    cfg = bytes.fromhex(f["task"])
    try:
        t, info, p, tid = check_task(cfg)
    except Refuse as e:
        return bad("its task is refused %s" % e.reason)
    code = info["measure"]
    if tid.hex() != f["task_id"] or MEASURES[code] != name:
        bad("task id or measure")
    vk = hkdf_expand(hkdf_extract(b"verify_key", bytes.fromhex(K["verify_key_seed"])), tid, 32)
    if vk.hex() != f["verify_key"]:
        bad("verification key (DAP §8.6.2)")
    ctx = b"dap-20" + tid
    lsk, hsk, csk = (bytes.fromhex(K[k]["private_key"]) for k in ("leader_hpke", "helper_hpke", "collector_hpke"))
    cfgs = {}
    for k, sk in (("leader_hpke", lsk), ("helper_hpke", hsk), ("collector_hpke", csk)):
        r = Rd(bytes.fromhex(K[k]["config"]))
        c = {"id": r.u(1), "kem": r.u(2), "kdf": r.u(2), "aead": r.u(2), "pk": r.opaque(2)}
        r.end()
        if (c["kem"], c["kdf"], c["aead"]) != (0x20, 1, 1) or c["pk"] != x25519(sk, BASE):
            bad("%s is not DAP §7's suite with its key" % k)
        cfgs[k] = c
    sessions = {s["session"]: s for s in doc["sessions"]}
    up = Rd(bytes.fromhex(f["upload"]))
    reports = []
    try:
        while not up.done():
            reports.append(read_report(up))
    except ValueError:
        return bad("the upload does not read")
    listed = f["reports"]
    if [r["m"]["id"].hex() for r in reports] != [x["id"] for x in listed] or [r["m"]["time"] for r in reports] != [x["time"] for x in listed]:
        return bad("the upload is not the listed reports")
    seen, pending, upload_errors = set(), [], []
    for r, x in zip(reports, listed):
        if r["leader"]["config_id"] != cfgs["leader_hpke"]["id"]:
            upload_errors.append({"id": x["id"], "error": "hpke_unknown_config_id"})
        elif r["m"]["id"] in seen:
            upload_errors.append({"id": x["id"], "error": "report_replayed"})
        else:
            seen.add(r["m"]["id"])
            pending.append((r, x))
    if upload_errors != f["upload_errors"]:
        bad("the reports the leader refused at upload")
    init = bytes([0]) + (0).to_bytes(4, "big") + (0).to_bytes(2, "big")
    resp, outcomes, committed = b"", [], []
    for r, x in pending:
        m = r["m"]
        laad = input_share_aad(tid, cfg, m, r["public"])
        try:
            lpay = open_share(lsk, 2, laad, r["leader"])
        except ValueError:
            outcomes.append({"id": x["id"], "outcome": "hpke_decrypt_error"})
            continue
        try:
            lstate, lvs = p.verify_init(vk, ctx, 0, m["id"], r["public"], lpay)
        except ValueError:
            outcomes.append({"id": x["id"], "outcome": "vdaf_verify_error"})
            continue
        outbound = b"\x00" + len(lvs).to_bytes(4, "big") + lvs
        init += m["raw"] + len(r["public"]).to_bytes(4, "big") + r["public"] + write_ct(r["helper"]) + len(outbound).to_bytes(4, "big") + outbound
        try:
            hpay = open_share(hsk, 3, laad, r["helper"])
        except ValueError:
            resp += m["id"] + b"\x02" + bytes([5])
            outcomes.append({"id": x["id"], "outcome": "hpke_decrypt_error"})
            if x["kind"] != "corrupt":
                bad("report %s does not open" % x["id"])
            continue
        if True:
            enc = vadd(p.f, p.meas_share(ctx, 0, lpay), p.meas_share(ctx, 1, hpay))
            if x["kind"] in ("session",):
                meas = measurement(code, sessions[x["session"]]["metrics"])
                if meas != x["measurement"] or enc != vdaf_encode(code, p, meas):
                    bad("report %s is not its session's measurement (§3.1)" % x["id"])
            elif x["kind"] == "out_of_range":
                if enc != x["encoded"]:
                    bad("the out-of-range report's encoding")
        try:
            hstate, hvs = p.verify_init(vk, ctx, 1, m["id"], r["public"], hpay)
            msg = p.message(ctx, [lvs, hvs])
            hout = p.verify_next(hstate, msg)
        except ValueError:
            resp += m["id"] + b"\x02" + bytes([6])
            outcomes.append({"id": x["id"], "outcome": "vdaf_verify_error"})
            continue
        fin = b"\x02" + len(msg).to_bytes(4, "big") + msg
        resp += m["id"] + b"\x00" + len(fin).to_bytes(4, "big") + fin
        lout = p.verify_next(lstate, msg)
        committed.append((m["id"], m["time"], lout, hout))
        outcomes.append({"id": x["id"], "outcome": "committed"})
    if init.hex() != f["job"]["init_req"]:
        bad("the AggregationJobInitReq is not DAP §4.5.3.1's")
    if resp.hex() != f["job"]["resp"]:
        bad("the AggregationJobResp is not DAP §4.5.3.2's")
    if outcomes != f["job"]["outcomes"]:
        bad("the reports' outcomes")
    if "proof" not in SKIP:
        for x in listed:
            o = next((y["outcome"] for y in outcomes if y["id"] == x["id"]), None)
            if x["kind"] == "out_of_range" and o != "vdaf_verify_error":
                bad("the out-of-range report is not refused vdaf_verify_error")
    buckets = {}
    for rid, tm, lo, ho in committed:
        b = buckets.setdefault(tm, {"l": [0] * p.v.OUTPUT_LEN, "h": [0] * p.v.OUTPUT_LEN, "n": 0, "ck": bytes(32), "ids": []})
        b["l"], b["h"] = vadd(p.f, b["l"], decode_vec(p.f, lo)), vadd(p.f, b["h"], decode_vec(p.f, ho))
        b["n"] += 1
        b["ck"] = bytes(a ^ c for a, c in zip(b["ck"], sha256(rid)))
        b["ids"].append((rid, tm))

    def merged(iv):
        out = {"l": [0] * p.v.OUTPUT_LEN, "h": [0] * p.v.OUTPUT_LEN, "n": 0, "ck": bytes(32), "times": [], "ids": []}
        for tm in range(iv[0], iv[0] + iv[1]):
            if tm in buckets:
                b = buckets[tm]
                out["l"], out["h"] = vadd(p.f, out["l"], b["l"]), vadd(p.f, out["h"], b["h"])
                out["n"] += b["n"]
                out["ck"] = bytes(a ^ c for a, c in zip(out["ck"], b["ck"]))
                out["times"].append(tm)
                out["ids"] += b["ids"]
        return out

    coll_raw = bytes.fromhex(f["collection"]["request"])
    query = interval_of(read_collection(coll_raw)["config"])
    for v in f["refusals"]:
        collected = query if v["phase"] == "after" else None
        req = bytes.fromhex(v["request"])

        def overlaps(iv):
            return collected is not None and iv[0] < collected[0] + collected[1] and collected[0] < iv[0] + iv[1]
        try:
            if v["actor"] == "leader":       # DAP §4.6.1
                c = read_collection(req)
                q = interval_of(c["config"])
                if c["agg_param"]:
                    got = "invalidAggregationParameter"
                elif q[1] == 0:
                    got = "batchInvalid"
                elif overlaps(q):
                    got = "batchOverlap"
                elif merged(q)["n"] < t["min_batch_size"]:
                    got = "invalidBatchSize"
                else:
                    got = "accept"
            else:                              # DAP §4.6.4
                r = Rd(req)
                c = {"batch_mode": r.u(1), "config": r.opaque(2), "agg_param": r.opaque(4), "extensions": read_exts(r)}
                mode, sel, n, ck = r.u(1), r.opaque(2), r.u(8), r.take(32)
                r.end()
                q, b = interval_of(c["config"]), interval_of(sel)
                mg = merged(b)
                if mode != 1 or c["batch_mode"] != 1:
                    got = "invalidMessage"
                elif b[1] == 0 or not within(b, q):
                    got = "batchInvalid"
                elif overlaps(b):
                    got = "batchOverlap"
                elif mg["n"] < t["min_batch_size"]:
                    got = "invalidBatchSize"
                elif c["agg_param"]:
                    got = "invalidMessage"
                elif (n, ck) != (mg["n"], mg["ck"]):
                    got = "batchMismatch"
                else:
                    got = "accept"
        except ValueError:
            got = "invalidMessage"
        if got != v["outcome"]:
            bad("refusal %s: stated %s, DAP gives %s" % (v["name"], v["outcome"], got))
    # The collection (DAP §4.6).
    mg = merged(query)
    if mg["n"] < t["min_batch_size"] and "floor" not in SKIP:
        bad("a batch below the minimum was collected")
    col = f["collection"]
    share_req = coll_raw + b"\x01" + (16).to_bytes(2, "big") + u64(query[0]) + u64(query[1]) + u64(mg["n"]) + mg["ck"]
    if share_req.hex() != col["share_req"]:
        bad("the AggregateShareReq is not DAP §4.6.4's")
    aad = agg_share_aad(tid, cfg, coll_raw)
    try:
        hct = read_ct(Rd(bytes.fromhex(col["helper_share"])))
        hshare = hpke_open(hct["enc"], csk, b"dap-20 aggregate share" + bytes([3, 0]), aad, hct["payload"])
        r = Rd(bytes.fromhex(col["resp"]))
        n, iv = r.u(8), (r.u(8), r.u(8))
        lct, hct2 = read_ct(r), read_ct(r)
        r.end()
        lshare = hpke_open(lct["enc"], csk, b"dap-20 aggregate share" + bytes([2, 0]), aad, lct["payload"])
        sts = []
        for k, role in (("leader_statement", 2), ("helper_statement", 3)):
            sct = read_ct(Rd(bytes.fromhex(col[k])))
            sts.append(read_statement(hpke_open(sct["enc"], csk, b"wai-telemetry-aggregate statement" + bytes([role, 0]), aad, sct["payload"])))
    except (ValueError, Refuse):
        return bad("the collection does not open with the collector's key")
    times = mg["times"]
    if n != mg["n"] or iv != (min(times), max(times) - min(times) + 1) or hct2 != hct:
        bad("the CollectionJobResp's count, interval or helper share")
    if decode_vec(p.f, lshare) != mg["l"] or decode_vec(p.f, hshare) != mg["h"]:
        bad("an aggregate share is not the batch's")
    for s, role, share, key in ((sts[0], 2, lshare, info["leader"]), (sts[1], 3, hshare, info["helper"])):
        if (s["role"], s["task_id"], s["collection"], s["batch"], s["count"], s["checksum"], s["agg_share"], s["signer"]) != \
                (role, tid, sha256(coll_raw), query, mg["n"], mg["ck"], share, key):
            bad("the %s statement is not its share's (§6.2)" % ("leader" if role == 2 else "helper"))
        if ED.verify(key, b"wai:telemetry-aggregate/share\x01" + s["body"], s["sig"]) is not None:
            bad("a statement's signature")
    result = vadd(p.f, mg["l"], mg["h"])
    if result != f["result"]:
        bad("the result")
    rc = read_receipt(bytes.fromhex(f["receipt"]))
    if (rc["leader"], rc["helper"], rc["result"], rc["task"], rc["collection"], rc["interval"]) != \
            (sts[0]["body"] + sts[0]["sig"], sts[1]["body"] + sts[1]["sig"], result, cfg, coll_raw, iv):
        bad("the receipt is not the collection's")
    given = [{"id": rid.hex(), "time": tm, "task": tid.hex()} for rid, tm in mg["ids"]]
    named = next((x for x in doc["receipts"] if x["name"] == name), None)
    if named is None or named["receipt"] != f["receipt"] or sorted(named["reports"] or [], key=json.dumps) != sorted(given, key=json.dumps):
        bad("the receipts section does not hold the flow's receipt with its reports")
    if not quiet:
        print("PASS flow %s: %d reports re-run, %d committed, result recomputed" % (name, len(reports), len(committed)))


# ---------------------------------------------------------------------------------------
# delivery-telemetry §3.4 and §4.3.

def key_classes():
    with open(TELEMETRY, encoding="utf-8") as fh:
        text = fh.read()
    at = text.index("\n§3.4")
    end = text.index("\n## ", at)
    out = {}
    for line in text[at:end].splitlines():
        if line.startswith("| `"):
            cells = [c.strip() for c in line.split("|")]
            for k in cells[1].split(", "):
                out[k.strip("`")] = cells[2]
    return out


CLASSES = key_classes()


def sf_dict_members(s):
    """Top-level members of an RFC 9651 Dictionary as (key, raw value), keeping strings whole."""
    out, i = [], 0
    while i < len(s):
        j = i
        while j < len(s) and s[j] not in "=,":
            j += 1
        key = s[i:j]
        if j < len(s) and s[j] == "=":
            k = j + 1
            if s[k] == '"':
                k += 1
                while s[k] != '"':
                    k += 2 if s[k] == "\\" else 1
                k += 1
            elif s[k] == "(":
                k = s.index(")", k) + 1
            else:
                while k < len(s) and s[k] != ",":
                    k += 1
            out.append((key, s[j + 1:k]))
            i = k
        else:
            out.append((key, "?1"))
            i = j
        if i < len(s):
            if s[i] != ",":
                raise ValueError("dictionary")
            i += 1
    return out


def shard_of_text():
    with open(TELEMETRY, encoding="utf-8") as fh:
        text = fh.read()
    at = text.index("§4.2 In Request mode")
    para = text[at:text.index("\n\n", at)]
    m = {}
    for group, shard in re.findall(r"((?:`[a-z]+`(?:, | and )?)+) to `(CMCD-[A-Za-z]+)`", para.replace("\n", " ")):
        for k in re.findall(r"`([a-z]+)`", group):
            m[k] = shard
    t41 = text[text.index("§4.1 Custom keys"):text.index("§4.2 In Request mode")]
    for line in t41.splitlines():
        if line.startswith("| `science"):
            cells = [c.strip() for c in line.split("|")]
            keys = [cells[1].split("`")[1]]
            rest = re.findall(r"`(-[a-z]+)`", cells[1])
            base = keys[0].rsplit("-", 1)[0]
            keys += [base + x for x in rest]
            if cells[4].startswith("`CMCD"):
                for k in keys:
                    m[k] = cells[4].strip("`")
    return m


def delivery_only_outcome(v):
    try:
        mem = sf_dict_members(v["delivery_only"])
    except (ValueError, IndexError):
        return {"outcome": "refuse", "reason": "measurement_key_in_delivery", "key": "?"}
    for k, _ in mem:
        if CLASSES.get(k) != "delivery":
            return {"outcome": "refuse", "reason": "measurement_key_in_delivery", "key": k}
    return {"outcome": "accept"}


def check_delivery_only(v, fails):
    got = delivery_only_outcome(v)
    if got != v["outcome"]:
        fails.append("delivery-only %s: stated %s, got %s" % (v["name"], v["outcome"], got))
    if "report" in v:
        full = sf_dict_members(v["report"])
        want = ",".join("%s=%s" % kv for kv in full if CLASSES.get(kv[0]) == "delivery")
        if want != v["delivery_only"]:
            fails.append("delivery-only %s: not the report's delivery-class members" % v["name"])
        shards = {}
        sm = shard_of_text()
        for k, x in sf_dict_members(want):
            shards.setdefault(sm.get(k, "?"), []).append("%s=%s" % (k, x))
        if {h: ",".join(ms) for h, ms in shards.items()} != v["shards"]:
            fails.append("delivery-only %s: header shards" % v["name"])


# ---------------------------------------------------------------------------------------

def read_reasons():
    with open(EXTENSION, encoding="utf-8") as fh:
        text = fh.read()
    at = text.index("\n## 11. Reasons")
    end = text.index("\n## ", at + 1)
    return [line.split("`")[1] for line in text[at:end].splitlines() if line.startswith("| `")]


REASONS = read_reasons()
REUSED = ["truncated", "malformed", "unknown_version", "bad_signature", "vdaf_verify_error", "hpke_decrypt_error",
          "report_replayed", "invalidBatchSize", "batchMismatch", "batchOverlap", "batchInvalid"]


MEMO = {}


def memo(key, fn):
    """A part's failures, computed once per input: a mutated copy re-runs only what it changed."""
    k = hashlib.sha256(json.dumps(key, sort_keys=True).encode()).hexdigest()
    if k not in MEMO:
        MEMO[k] = fn()
    return list(MEMO[k])


def check(doc, vdaf_files, quiet=False):
    fails = []
    for n, d in sorted(vdaf_files.items()):
        fails += memo(["vdaf", n, d, sorted(SKIP)], lambda: run_xof(d) if n == "XofTurboShake128.json" else run_vdaf_file(n[:-5], d))
    if not quiet:
        print(("PASS" if not fails else "FAIL") + " the draft's Prio3 and XofTurboShake128 vectors (%d files)" % len(vdaf_files))
    for f in doc["flows"]:
        named = [x for x in doc["receipts"] if x["name"] == f["name"]]

        def run(f=f):
            out = []
            check_flow(doc, f, out, quiet)
            return out
        fails += run() if not quiet else memo(["flow", f, doc["keys"], doc["sessions"], named, sorted(SKIP & {"proof", "floor"})], run)
    for v in doc["receipts"]:
        got = receipt_outcome(v)
        if got != v["outcome"]:
            fails.append("receipt %s: stated %s, got %s" % (v["name"], v["outcome"], got))
    for v in doc["cells"]:
        got = cell_outcome(doc, v)
        if got != v["outcome"]:
            fails.append("cell %s: stated %s, got %s" % (v["name"], v["outcome"], got))
        if got.get("outcome") == "accept" and (v.get("plaintext") != plaintext_cell(doc["sessions"]) or got["cell"] != v["plaintext"]):
            fails.append("cell %s: not the plaintext cell of live-metering §4.7" % v["name"])
    for v in doc["delivery_only"]:
        check_delivery_only(v, fails)
    if doc.get("codes") != REASONS:
        fails.append("the corpus's codes are not §11's")
    if not quiet:
        print(("PASS" if not fails else "FAIL") + " %d receipts, %d cells, %d delivery-only reports" % (len(doc["receipts"]), len(doc["cells"]), len(doc["delivery_only"])))
    return fails


def reasons_given(doc):
    out = set()
    for sec in ("receipts", "cells", "delivery_only"):
        out |= {v["outcome"].get("reason") for v in doc[sec]}
    for f in doc["flows"]:
        out |= {o["outcome"] for o in f["job"]["outcomes"]} | {e["error"] for e in f["upload_errors"]} | {r["outcome"] for r in f["refusals"]}
    return out


def flip_hex(h, i):
    b = bytearray(bytes.fromhex(h))
    b[i] ^= 0x01
    return b.hex()


def mutations(doc, vdaf_files):
    def named(d, sec, name):
        return next(x for x in d[sec] if x["name"] == name)

    def m(label, fn, vf=None):
        d = copy.deepcopy(doc)
        v = copy.deepcopy(vdaf_files)
        (fn or (lambda _: None))(d)
        if vf:
            vf(v)
        return label, d, v

    yield m("a report's leader share", lambda d: d["flows"][0].update(upload=flip_hex(d["flows"][0]["upload"], 300)))
    yield m("a VerifyInit's payload", lambda d: d["flows"][1]["job"].update(init_req=flip_hex(d["flows"][1]["job"]["init_req"], len(d["flows"][1]["job"]["init_req"]) // 2 - 1)))
    yield m("a VerifyResp", lambda d: d["flows"][2]["job"].update(resp=flip_hex(d["flows"][2]["job"]["resp"], 40)))
    yield m("a session's stalls", lambda d: d["sessions"][3]["metrics"].update(stalls=d["sessions"][3]["metrics"]["stalls"] + 1))
    yield m("a session's start time", lambda d: d["sessions"][5]["metrics"].update(vst_ms=d["sessions"][5]["metrics"]["vst_ms"] + 4000))
    yield m("the out-of-range report stated committed", lambda d: d["flows"][3]["job"]["outcomes"][50].update(outcome="committed"))
    yield m("a listed measurement", lambda d: d["flows"][0]["reports"][7].update(measurement=[1, 0, 0, 0]))
    yield m("a refusal's outcome", lambda d: d["flows"][0]["refusals"][0].update(outcome="batchMismatch"))
    yield m("a refusal's batch interval", lambda d: d["flows"][0]["refusals"][0].update(request=flip_hex(d["flows"][0]["refusals"][0]["request"], 18)))
    yield m("a flow's result", lambda d: d["flows"][2]["result"].__setitem__(1, d["flows"][2]["result"][1] + 1))
    yield m("the helper's sealed statement", lambda d: d["flows"][1]["collection"].update(helper_statement=flip_hex(d["flows"][1]["collection"]["helper_statement"], 60)))
    yield m("the collector's private key", lambda d: d["keys"]["collector_hpke"].update(private_key=flip_hex(d["keys"]["collector_hpke"]["private_key"], 5)))
    yield m("the verification key seed", lambda d: d["keys"].update(verify_key_seed=flip_hex(d["keys"]["verify_key_seed"], 0)))
    yield m("an accepted receipt's signature", lambda d: named(d, "receipts", "stalls").update(receipt=flip_hex(named(d, "receipts", "stalls")["receipt"], -3 + len(named(d, "receipts", "stalls")["receipt"]) // 2)))
    yield m("an accepted receipt's stated count", lambda d: named(d, "receipts", "durations")["outcome"].update(report_count=51))
    yield m("an accepted receipt's reports", lambda d: named(d, "receipts", "start_time")["reports"].pop())
    yield m("a refused receipt stated accepted", lambda d: named(d, "receipts", "batch_below_the_minimum").update(outcome={"outcome": "accept"}))
    yield m("a refusal's place", lambda d: named(d, "receipts", "helper_statement_by_another")["outcome"].update(at="leader"))
    yield m("a cell's plaintext", lambda d: named(d, "cells", "room_lesson")["plaintext"].update(stalls=0))
    yield m("a delivery-only report given a measurement key", lambda d: named(d, "delivery_only", "audio_request").update(
        delivery_only=named(d, "delivery_only", "audio_request")["delivery_only"] + ",sn=4"))
    yield m("a delivery-only refusal's key", lambda d: named(d, "delivery_only", "cohort_served_as_delivery")["outcome"].update(key="d"))
    yield m("the codes reordered", lambda d: d["codes"].reverse())
    yield m("a draft vector's result", None, lambda v: v["Prio3Sum_2.json"].update(agg_result=1522))
    yield m("a draft vector's verifier share", None, lambda v: v["Prio3Histogram_0.json"]["reports"][0]["verifier_shares"][0].__setitem__(
        0, flip_hex(v["Prio3Histogram_0.json"]["reports"][0]["verifier_shares"][0][0], 3)))
    yield m("a draft vector's helper seed", None, lambda v: v["Prio3SumVec_1.json"]["reports"][0]["input_shares"].__setitem__(
        1, flip_hex(v["Prio3SumVec_1.json"]["reports"][0]["input_shares"][1], 0)))


CHECKS = {
    "task-rules": "§4.3 rule 1: the task's fields",
    "measure": "§4.3 rule 3: the measure's VDAF",
    "floor-rule": "§4.3 rule 4: a minimum batch size of at least fifty",
    "one-key": "§4.3 rule 5: two aggregator keys",
    "collector-signer": "§8.1 step 3: the collector's key",
    "collector-sig": "§8.1 step 3: the receipt's signature",
    "collection": "§8.1 step 4: the collection job",
    "two-aggregators": "§8.1 step 5: two signers in two roles",
    "statement-signer": "§8.1 step 5: each statement's key",
    "statement-sig": "§8.1 step 5: each statement's signature",
    "binding": "§8.1 step 6: the task and collection bound",
    "batch": "§8.1 step 7: the batch within the query",
    "agreement": "§8.1 step 8: the statements agree",
    "floor": "§8.1 step 9: the batch at least the minimum",
    "share": "§8.1 step 10: each aggregate share",
    "result": "§8.1 step 10: the result unsharded",
    "reports": "§8.1 step 11: the batch's reports",
    "cell": "§9.2: one receipt of each measure, one scope, one batch",
    "proof": "§3.3: an out-of-range report is refused",
}


def main(corpus_dir):
    fails = []
    st = ED._self_test()
    print(("PASS" if not st else "FAIL") + " Ed25519 strict rule self-test " + ",".join(st))
    fails += ["ed25519 self-test"] if st else []
    ct = crypto_self_test()
    print(("PASS" if not ct else "FAIL") + " TurboSHAKE128, X25519 and HPKE self-tests " + ",".join(ct))
    fails += ct
    print(("PASS" if len(REASONS) == 17 else "FAIL") + " §11 read from the extension: %d reasons" % len(REASONS))
    fails += [] if len(REASONS) == 17 else ["§11 does not read"]
    print(("PASS" if len(CLASSES) == 42 else "FAIL") + " delivery-telemetry §3.4 read: %d keys classed" % len(CLASSES))
    fails += [] if len(CLASSES) == 42 else ["§3.4 does not read"]
    with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as fh:
        doc = json.load(fh)
    vdir = os.path.join(corpus_dir, "vdaf-22")
    vdaf_files = {}
    for n in sorted(os.listdir(vdir)):
        with open(os.path.join(vdir, n), encoding="utf-8") as fh:
            vdaf_files[n] = json.load(fh)
    fails += check(doc, vdaf_files)
    extra = sorted(set(os.listdir(corpus_dir)) - {"vectors.json", "README.md", "vdaf-22"})
    print(("PASS" if not extra else "FAIL") + " no file but vectors.json, README.md and vdaf-22/")
    fails += ["an extra file: %s" % n for n in extra]
    missing = [r for r in REASONS + REUSED if r not in reasons_given(doc)]
    print(("PASS" if not missing else "FAIL") + " every reason of §11, and every code it reuses, appears in the corpus")
    fails += ["no case gives %s" % r for r in missing]
    for label, d, v in mutations(doc, vdaf_files):
        caught = check(d, v, quiet=True)
        print(("PASS" if caught else "FAIL") + " mutated copy fails: %s%s" % (label, " (%s)" % caught[0][:90] if caught else ""))
        if not caught:
            fails.append("a mutated copy verifies: %s" % label)
    for name, what in CHECKS.items():
        SKIP.clear()
        SKIP.add(name)
        caught = check(doc, {}, quiet=True)
        SKIP.clear()
        print(("PASS" if caught else "FAIL") + " a verifier without %s fails the corpus%s" % (what, " (%s)" % caught[0][:70] if caught else ""))
        if not caught:
            fails.append("the corpus does not catch a verifier without %s" % what)
    for x in fails:
        print("FAIL " + x)
    print(("PASS" if not fails else "FAIL") + " QoE aggregation corpus")
    return 1 if fails else 0


if __name__ == "__main__":
    if len(sys.argv) != 2:
        print("usage: telemetry_aggregate_verify.py <live-conformance/aggregate>", file=sys.stderr)
        sys.exit(2)
    sys.exit(main(sys.argv[1]))