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

#!/usr/bin/env python3
"""Independent verifier of the impression aggregates corpus (extensions/impression-aggregates.md).

Written from the extension's text and the texts it cites. It shares no code with wai-rs, which
generates the corpus, and never runs it. From the second implementations of the texts this
extension builds on it takes what those texts define, by reference as the extension does:

- telemetry_aggregate_verify.py (telemetry-aggregation, DAP, VDAF, HPKE): Prio3Histogram with its
  field, XOF and FLP; X25519, HKDF-SHA256, AES-128-GCM and HPKE open; DAP's messages; a task's
  `task_info`, its id and its §4.3 rule 1; the statement and the aggregate receipt; and §8.1
  steps 1 to 11, step 2 given as §5.3 here;
- stitched_presentation_verify.py: a pod-input record under stitched-presentation §7.2 steps 1
  to 3 and the claims under §7.3 steps 1 to 10, with each presentation's quartile (§7.4);
- ad_presentation_verify.py: a token's reading (§7.1), its TokenChallenge (§7.2), the issuer
  key's form and RSASSA-PSS (§7.3);
- ../signer-conformance/signer_verify.py: key-log revisions, anchors and confirmation
  (jwp-receipts §7.3, §7.4), `aggregate/task` among its pinned-only authorities (§7.5);
- rung_price_verify.py: a region code and *holds* (rung-price §2);
- ../claim-conformance/_ed25519.py: Ed25519 as SPEC §7 *Signatures* states it.

Everything this extension adds is written here: the catalogue, its columns and cells (§4); the
task's rules and the task file with its HpkeConfigLists (§5); the contribution, the upload file,
the upload plan and the settlement (§6); the aggregators' report rule, token check and hold
(§7); a receipt's §8.1 steps 12 and 13, and the table (§8.2). Each flow is re-run end to end:
every report's two input shares opened with the aggregators' test keys, its extensions and token
checked as each aggregator checks its own share, each presentation's contribution recomputed
and its measurement recombined and required to be one at its cell, both aggregators' Prio3
verification, each upload's response and each relay's merge, the aggregation job's request and
response byte for byte, every bucket, collection (held, refused or completed), aggregate share,
statement, result and receipt.

For every case it requires the outcome the corpus states, field for field; that the corpus
lists exactly §10's codes (read from the extension), that each of them and each code §10 says
the extension reuses appears in an outcome, and that no outcome gives a code the extension does
not name; and it checks the size bounds no vector carries. Then it 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 -I impression_aggregate_verify.py <live-conformance/impressions>

Exit status 0 only if every case verifies, every mutated copy fails and every weakened verifier
is caught. Python 3.8 or later, the standard library and the signer verifier's own requirement
(the `cryptography` package).
"""

import copy
import hashlib
import json
import math
import os
import re
import struct
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
for _d in ("claim-conformance", "signer-conformance"):
    sys.path.insert(0, os.path.join(HERE, "..", _d))
sys.path.insert(0, HERE)
import _ed25519 as ED  # noqa: E402
import ad_presentation_verify as APV  # noqa: E402
import rung_price_verify as RP  # noqa: E402
import signer_verify as SG  # noqa: E402
import stitched_presentation_verify as SPV  # noqa: E402
import telemetry_aggregate_verify as TAV  # noqa: E402

EXTENSION = os.path.join(HERE, "..", "extensions", "impression-aggregates.md")
JWP = os.path.join(HERE, "..", "extensions", "jwp-receipts.md")

Refuse = TAV.Refuse
CAT_MAGIC, TASK_MAGIC, UPLOAD_MAGIC = b"WIC1", b"WIT1", b"WIU1"
TASK_DOMAIN = b"wai:impression-task\x01"                       # §5.4
ATTEST_DOMAIN = b"wai:impression-report-attest\x01"            # §7.2
TOKEN_EXT = 0x5749                                             # §7.2 `wai_impression_token`
MAX_CATALOGUE, MAX_TASK_FILE, MAX_UPLOAD = 1 << 20, 1 << 20, 1 << 24
MAX_CELLS = 4096                                               # §4.3 rule 7
FLOOR = 50                                                     # §5.3 step 5, §8.2 `floor_met`
MIN_PRECISION = 3600                                           # §5.3 step 3
LAST_S = 253402300799                                          # §5.3 step 3: the last instant, in seconds
HEADER = 41                                                    # §6.5: a WIU1's bytes before its reports
SUITE = (0x0020, 0x0001, 0x0001)                               # DAP §7: X25519, HKDF-SHA256, AES-128-GCM
CLAIM, ATTEST = 1, 2                                           # ad-presentation §8
REPORT_ERRORS = {1: "batch_collected", 2: "report_replayed", 3: "report_dropped", 4: "hpke_unknown_config_id",
                 5: "hpke_decrypt_error", 6: "vdaf_verify_error", 7: "invalid_message", 8: "report_too_early",
                 9: "unknown_verification_key_id", 10: "unsupported_extension"}   # DAP §4.1
CODE = {v: k for k, v in REPORT_ERRORS.items()}
KEEP = 8                                                       # §6.6: report_too_early is kept
PROBLEMS = ["invalidBatchSize", "batchMismatch", "batchOverlap", "batchInvalid", "invalidAggregationParameter",
            "invalidMessage"]                                  # DAP §3.6, as the collections meet them

# A weakened run drops one of these checks (see `WEAKENINGS`); empty checks everything.
SKIP = set()


def drops(name):
    return name in SKIP


def sha(*parts):
    h = hashlib.sha256()
    for x in parts:
        h.update(x)
    return h.digest()


def u32(n):
    return n.to_bytes(4, "big")


def section_codes(path, heading, column):
    """The codes of a reasons table: the `column`-th cell of each row that starts with a code."""
    with open(path, encoding="utf-8") as fh:
        text = fh.read()
    at = text.index(heading)
    end = text.find("\n## ", at + 1)
    rows = text[at:end if end >= 0 else len(text)].splitlines()
    out = []
    for r in rows:
        cells = [c.strip() for c in r.split("|")]
        if len(cells) > column + 1 and cells[column].startswith("`"):
            out.append(cells[column].strip("`"))
    return out


REASONS = section_codes(EXTENSION, "\n## 10. Reasons", 1)
JWP_REASONS = section_codes(JWP, "\n### 7.7. Reasons", 3)


# Each of these is a pure function of its inputs, so each is computed once for a set of inputs.
MEMO = {}


def memo(key, fn):
    if key not in MEMO:
        try:
            MEMO[key] = (True, fn())
        except (ValueError, Refuse) as e:
            MEMO[key] = (False, e)
    ok, v = MEMO[key]
    if not ok:
        raise v
    return v


# ------------------------------------------------------------------ §4 the catalogue

def read_terms(r):
    """stitched-presentation §4.1's measurement terms, little-endian there: (bytes, fields)."""
    start = r.i
    q = {"require": r.u(1), "rule": r.u(1), "token_type": int.from_bytes(r.take(2), "little")}
    q["issuer"] = r.take(int.from_bytes(r.take(2), "little"))
    return r.b[start:r.i], q


def terms_bytes(q):
    return struct.pack("<BBHH", q["require"], q["rule"], q["token_type"], len(q["issuer"])) + q["issuer"]


def terms_ok(q):
    """§4.3 rule 2: ad-presentation §8's rules, bit 0 (claim) set, and attest by token type 2 only."""
    r, rule, tt, iss = q["require"], q["rule"], q["token_type"], q["issuer"]
    claim = bool(r & CLAIM)
    if r > 7 or (r & 6 and not claim) or rule > 2 or (rule == 0) == claim:
        return False
    if r & ATTEST:
        if tt not in (1, 2) or not iss:
            return False
    elif tt != 0 or iss:
        return False
    return claim and (not r & ATTEST or tt == 2)


def ascending(xs):
    return all(a < b for a, b in zip(xs, xs[1:]))


def ceil_sqrt(n):
    s = math.isqrt(n)
    return s if s * s >= n else s + 1


def rsa_key_ok(k):
    try:
        APV.rsa_key(k)
        return True
    except ValueError:
        return False


def read_catalogue(b):
    """§4.3: the catalogue, or `catalogue_malformed`."""
    bad = Refuse("catalogue_malformed", "catalogue")
    if len(b) > MAX_CATALOGUE:
        raise bad
    r = TAV.Rd(b)
    try:
        magic, version = r.take(4), r.u(1)
        if magic != CAT_MAGIC or (version != 1 and not drops("cat-version")):           # 1
            raise bad
        raw, q = read_terms(r)
        hold = r.u(4)
        signers = [r.take(32) for _ in range(r.u(1))]
        creatives = [r.take(32) for _ in range(r.u(1))]
        regions = [r.opaque(1) for _ in range(r.u(1))]
        keys = [r.opaque(2) for _ in range(r.u(1))]
        r.end()
    except ValueError:
        raise bad
    attest = bool(q["require"] & ATTEST)
    cells = len(creatives) * (len(regions) + 1) * 5
    rules = [
        ("cat-terms", not terms_ok(q)),                                              # 2
        ("cat-hold", hold == 0),                                                     # 3
        ("cat-signers", not 1 <= len(signers) <= 16),                                # 4
        ("cat-order", not ascending(signers)),
        ("cat-creatives", not creatives),                                            # 5
        ("cat-order", not ascending(creatives)),
        ("cat-region", not all(RP.region_code(x.decode("latin-1")) for x in regions)),  # 6
        ("cat-order", not ascending(regions)),
        ("cat-overlap", any(len(x) == 2 and any(y.startswith(x + b"-") for y in regions) for x in regions)),
        ("cat-cells", cells > MAX_CELLS),                                            # 7
        ("cat-keys", (not attest and keys) or (attest and not 1 <= len(keys) <= 4)),  # 8
        ("cat-key-form", not all(rsa_key_ok(k) for k in keys)),
        ("cat-order", not ascending(keys)),
    ]
    if any(broken and not drops(name) for name, broken in rules):
        raise bad
    return {"bytes": b, "digest": sha(b), "terms_raw": raw, "terms": q, "attest": attest, "hold_s": hold,
            "signers": signers, "creatives": creatives, "regions": regions, "keys": keys, "cells": cells,
            "chunk": ceil_sqrt(cells) if cells else 0}


def column(cat, g):
    """§4.2: the column of a receiver's region `g` (bytes or None); `r` is `other`."""
    other = len(cat["regions"])
    if g is None:
        return other
    s = g.decode("latin-1")
    if not RP.region_code(s):
        return other
    for j, x in enumerate(cat["regions"]):
        if (x == g) if drops("region-subdivision") else RP.holds([x.decode("ascii")], s):
            return j
    return other


def column_name(cat, j):
    return cat["regions"][j].decode("ascii") if j < len(cat["regions"]) else "other"


def cell_of(cat, i, j, q):
    return (i * (len(cat["regions"]) + 1) + j) * 5 + (q - 1)


def catalogue_outcome(b):
    try:
        c = read_catalogue(b)
    except Refuse as e:
        return {"outcome": "refuse", "reason": e.reason}
    return {"outcome": "read", "digest": c["digest"].hex(), "cells": c["cells"], "chunk": c["chunk"]}


# ------------------------------------------------------------------ §5 the task

def task_rules(cfg, cat):
    """§5.3 against a catalogue that reads: the task's fields, Prio3, id, flight and retention."""
    t, info = TAV.task_fields(cfg)                                                   # 1
    if not drops("catalogue-digest") and (info["measure"] != 5 or info["scope"] != cat["digest"]):
        raise Refuse("catalogue_mismatch", "task")                                   # 2
    got = {}

    def steps_3_4(t, info):
        ex = t["extensions"]
        if t["time_precision"] < MIN_PRECISION and not drops("task-precision"):
            raise Refuse("task_invalid", "task")
        if len(ex) != 1 or ex[0][0] != 1 or len(ex[0][1]) != 16:
            raise Refuse("task_invalid", "task")
        iv = TAV.interval_of(ex[0][1])
        if iv[1] < 1 and not drops("task-interval"):
            raise Refuse("task_invalid", "task")
        retention = (iv[0] + iv[1]) * t["time_precision"] + cat["hold_s"]
        if retention > LAST_S and not drops("task-retention"):                        # a u64 overflow is past it too
            raise Refuse("task_invalid", "task")
        config = u32(cat["cells"]) + (b"" if drops("chunk-length") else u32(cat["chunk"]))
        given = t["vdaf_config"][:4] if drops("chunk-length") else t["vdaf_config"]
        if t["vdaf_type"] != 4 or given != config:                                   # 4: DAP Appendix B.4
            raise Refuse("measure_mismatch", "task")
        got.update(interval=iv, retention_s=retention)
        return TAV.Prio3(TAV.Histogram(cat["cells"], cat["chunk"]))

    t, info, p, tid = TAV.check_task(cfg, measure5=steps_3_4)                         # 3, 4; then 5, 6
    if "interval" not in got:
        raise ValueError("not a task of measure 5")
    return {"cfg": cfg, "t": t, "info": info, "p": p, "tid": tid, "interval": got["interval"],
            "retention_s": got["retention_s"]}


def task_outcome(cfg, cat_bytes):
    try:
        k = task_rules(cfg, read_catalogue(cat_bytes))
    except Refuse as e:
        return {"outcome": "refuse", "reason": e.reason}
    return {"outcome": "accept", "task_id": k["tid"].hex(), "retention_s": k["retention_s"]}


def suite_config(configs):
    """§6.3: the first configuration of DAP §7's suite with a 32-byte key."""
    if drops("hpke-first"):
        return configs[0]
    return next((c for c in configs if (c["kem"], c["kdf"], c["aead"]) == SUITE and len(c["pk"]) == 32), None)


def read_hpke_list(r):
    """DAP §4.4.1's HpkeConfigList<10..2^16-1>, with its own u16 length first (§5.4 step 1)."""
    body = TAV.Rd(r.opaque(2))
    if len(body.b) < 10:
        raise ValueError("short list")
    out = []
    while not body.done():
        c = {"id": body.u(1), "kem": body.u(2), "kdf": body.u(2), "aead": body.u(2), "pk": body.opaque(2)}
        if not c["pk"]:
            raise ValueError("empty key")
        out.append(c)
    ids = [c["id"] for c in out]
    if len(set(ids)) != len(ids) and not drops("tf-hpke-ids"):
        raise ValueError("two configurations of one id")
    suite = [c for c in out if (c["kem"], c["kdf"], c["aead"]) == SUITE and len(c["pk"]) == 32]
    if not suite and not drops("tf-hpke-suite"):
        raise ValueError("no configuration of DAP's suite")
    return out


def read_task_file(b):
    """§5.4 step 1: the file, or `task_file_malformed`."""
    bad = Refuse("task_file_malformed", "task_file")
    if len(b) > MAX_TASK_FILE and not drops("tf-bound"):
        raise bad
    r = TAV.Rd(b)
    try:
        magic, version = r.take(4), r.u(1)
        if magic != TASK_MAGIC or (version != 1 and not drops("tf-version")):
            raise bad
        tf = {"task": r.opaque(2), "catalogue": r.opaque(4), "leader_hpke": read_hpke_list(r),
              "helper_hpke": read_hpke_list(r), "signer": r.take(32)}
        sid = r.opaque(2)
        tf["body"] = b[:r.i]
        tf["sig"] = r.take(64)
        r.end()
    except ValueError:
        raise bad
    if not sid or (any(c > 0x7F for c in sid) and not drops("tf-signer-ascii")):
        raise bad
    tf["signer_id"] = sid.decode("latin-1")
    tf["digest"] = sha(tf["body"])
    return tf


def confirmation(tf, revision_texts, anchors):
    """§5.4 step 5: jwp-receipts §7.4 for `aggregate/task`, the item untimed."""
    held = SG.held_revisions(revision_texts)
    if drops("tf-revocation"):          # a receiver that takes in no revision the bearer brings
        named = {a["revision"] for a in anchors}
        held = [x for x in held if SG.rev_hash(x) in named]
    if drops("tf-pinned"):
        anchors = [dict(a, path="pinned", pinned_revision=a["revision"]) for a in anchors]
    authorities = [] if drops("tf-authority") else ["aggregate/task"]
    return SG.confirm(tf["signer"], tf["signer_id"], authorities, None, held, anchors, None)


def accept_task_file(b, revision_texts, anchors):
    """§5.4, in order: (file, catalogue, task, confirmation), or the refusal."""
    tf = read_task_file(b)                                                           # 1
    cat = read_catalogue(tf["catalogue"])                                            # 2
    task = task_rules(tf["task"], cat)                                               # 3
    if not drops("tf-signature"):                                                    # 4
        why = ED.verify(tf["signer"], TASK_DOMAIN + tf["body"], tf["sig"])
        if why:
            raise Refuse(why, "task_file")
    v = confirmation(tf, revision_texts, anchors)                                    # 5
    if not v.startswith("confirmed"):
        raise Refuse(re.search(r"reason=(\S+)", v).group(1), "task_file")
    return tf, cat, task, v


def task_file_outcome(doc, c):
    fx = doc["fixtures"]
    try:
        tf, _, task, v = accept_task_file(bytes.fromhex(c["bytes"]), [fx["key_logs"][n] for n in c["revisions"]],
                                          [fx["anchors"][n] for n in c["anchors"]])
    except Refuse as e:
        return {"outcome": "refuse", "reason": e.reason}
    return {"outcome": "accept", "digest": tf["digest"].hex(), "task_id": task["tid"].hex(), "confirmation": v}


def receiver_task_file(doc, name):
    """A task file of the fixtures, accepted under what `receiver` holds (§6.1)."""
    fx = doc["fixtures"]
    rv = fx["receiver"]
    return accept_task_file(bytes.fromhex(fx["task_files"][name]), [fx["key_logs"][n] for n in rv["revisions"]],
                            [fx["anchors"][n] for n in rv["anchors"]])


# ------------------------------------------------------------------ §7.2 the token

def pss(key, msg, sig):
    return memo(("pss", key, msg, sig), lambda: APV.pss_verify(key, msg, sig))


def token_check(cat, tid, rid, token):
    """§7.2's token: None when it is type 0x0002, answers the challenge of `tid` and `rid`, names one
    of the catalogue's keys and verifies under it; otherwise `attestation_invalid`."""
    t = APV.read_token(token)
    if t is None or t["type"] != 2:
        return "attestation_invalid"
    ctx = sha(ATTEST_DOMAIN, tid, rid)
    if not drops("token-challenge") and t["digest"] != sha(APV.token_challenge(2, cat["terms"]["issuer"], ctx)):
        return "attestation_invalid"
    key = next((k for k in cat["keys"] if sha(k) == t["key_id"]), None)
    if key is None or not pss(key, t["input"], t["auth"]):
        return "attestation_invalid"
    return None


def report_rule(cat, tid, rid, public, private, role):
    """§7.2 at DAP §4.5.3.4, an aggregator's own input share: None, or (error, WAI reason)."""
    if public and not drops("public-extension"):
        return "invalid_message", None
    if not cat["attest"]:
        return ("invalid_message", None) if private else None
    if role == 3 and drops("helper-token"):
        return None
    if len(private) != 1 or private[0][0] != TOKEN_EXT:
        return "invalid_message", "attestation_absent"
    if token_check(cat, tid, rid, private[0][1]):
        return "invalid_message", "attestation_invalid"
    return None


# ------------------------------------------------------------------ §6.2 the contribution

def held_fixtures(doc):
    fx = doc["fixtures"]
    manifests = [fx["manifests"][n].encode("utf-8") for n in sorted(fx["manifests"])]
    creatives = [[{"content_hash": bytes.fromhex(o["content_hash"]), "duration_ms": o["duration_ms"]}
                  for o in fx["creatives"][n]["objects"]] for n in sorted(fx["creatives"])]
    return manifests, creatives


def judge(doc, record, claims, signers):
    """stitched-presentation §7.2 steps 1 to 3 for the record (no other record, no response held)
    and §7.3 steps 1 to 10 for the claims, in order."""
    manifests, creatives = held_fixtures(doc)
    key = ("judge", record, tuple(claims), tuple(signers), tuple(manifests),
           json.dumps(doc["fixtures"]["creatives"], sort_keys=True))
    inp = {"records": [record], "claims": claims, "manifests": manifests, "responses": [], "receipts": [],
           "creatives": creatives, "issuers": [], "signers": signers}
    return memo(key, lambda: SPV.verify(inp, through_attestation=False))


def no_report(reason, discarded=()):
    return {"outcome": "no_report", "reason": reason, "discarded": list(discarded)}


def contribute(doc, cat, task, record, claims, region, at_ms, pool):
    """§6.2, in order."""
    v = judge(doc, record, claims, cat["signers"])
    verdict = v["record_facts"][0]["verdict"]
    if verdict != "accepted":                                                        # 1
        return no_report(verdict)
    rec = SPV.read_record(record)
    if rec["session"] != bytes(16) and not drops("shared"):                          # 2
        return no_report("record_not_shared")
    if terms_bytes(rec["terms"]) != cat["terms_raw"] and not drops("terms-equal"):   # 3
        return no_report("record_terms_differ")
    cf = v["claim_facts"][-1]
    if cf["verdict"] != "presented":                                                 # 4
        return no_report(cf["verdict"])
    ad = rec["ads"][cf["slot"]]
    if ad["creative_hash"] not in cat["creatives"] and not drops("creative-listed"):   # 5
        return no_report("creative_not_listed")
    tp = task["t"]["time_precision"]
    if at_ms is None and not drops("instant"):                                       # 6
        return no_report("instant_unknown")
    time = task["interval"][0] if at_ms is None else at_ms // (1000 * tp)
    iv = task["interval"]
    if not iv[0] <= time < iv[0] + iv[1] and not drops("flight-window"):             # 7
        return no_report("outside_task")
    discarded, rid = [], "fresh"
    if cat["attest"]:                                                                # 8
        chosen = None
        for pair in sorted(pool, key=lambda x: bytes.fromhex(x["id"])):
            if drops("token-pool") or token_check(cat, task["tid"], bytes.fromhex(pair["id"]),
                                                  bytes.fromhex(pair["token"])) is None:
                chosen = pair
                break
            discarded.append(pair["id"])
        if chosen is None:
            return no_report("token_exhausted", discarded)
        rid = chosen["id"]
    q = cf["quartile"]
    j = column(cat, region)
    i = cat["creatives"].index(ad["creative_hash"]) if ad["creative_hash"] in cat["creatives"] else None
    return {"outcome": "report", "cell": None if i is None else cell_of(cat, i, j, q), "creative": i,
            "column": column_name(cat, j), "quartile": q, "time": time, "id": rid, "discarded": discarded}


def contribution_outcome(doc, c):
    tf, cat, task, _ = receiver_task_file(doc, c["task_file"])
    record = bytes.fromhex(doc["fixtures"]["records"][c["record"]]["bytes"])
    region = None if c["region"] is None else c["region"].encode("utf-8")
    return contribute(doc, cat, task, record, [bytes.fromhex(x) for x in c["claims"]], region, c["at_ms"], c["pool"])


# ------------------------------------------------------------------ §6.4 to §6.6

def read_upload(b):
    """§6.4: (task_id, reports), each report with its bytes; or `upload_malformed`."""
    bad = Refuse("upload_malformed", "upload")
    if len(b) > MAX_UPLOAD and not drops("upload-bound"):
        raise bad
    r = TAV.Rd(b)
    try:
        magic, version = r.take(4), r.u(1)
        if magic != UPLOAD_MAGIC or (version != 1 and not drops("upload-version")):
            raise bad
        tid = r.take(32)
        n = r.u(4)
        up = TAV.Rd(r.take(min(n, len(b) - r.i) if drops("upload-length") else n))
        r.end()
        reports = []
        while not up.done():
            start = up.i
            rep = TAV.read_report(up)
            rep["raw"] = up.b[start:up.i]
            reports.append(rep)
    except ValueError:
        raise bad
    ids = [x["m"]["id"] for x in reports]
    if not reports or (not ascending(ids) and not drops("upload-order")):
        raise bad
    return tid, reports


def write_upload(tid, raws):
    body = b"".join(raws)
    return UPLOAD_MAGIC + b"\x01" + tid + u32(len(body)) + body


def upload_outcome(b):
    try:
        tid, reports = read_upload(b)
    except Refuse as e:
        return {"outcome": "refuse", "reason": e.reason}
    return {"outcome": "read", "task_id": tid.hex(), "ids": [x["m"]["id"].hex() for x in reports]}


def read_one_report(b):
    r = TAV.Rd(b)
    rep = TAV.read_report(r)
    r.end()
    rep["raw"] = b
    return rep


def plan(task, cat, stored, t_ms, budget):
    """§6.5."""
    tp, iv = task["t"]["time_precision"], task["interval"]
    timely, late, drop = [], [], []
    for b in stored:
        rep = read_one_report(b)
        rid, tau = rep["m"]["id"], rep["m"]["time"]
        if not iv[0] <= tau < iv[0] + iv[1] or (t_ms >= task["retention_s"] * 1000 and not drops("plan-retention")):
            drop.append(rid)
            continue
        deadline = ((tau + 1) * tp + cat["hold_s"]) * 1000
        (timely if t_ms < deadline or drops("plan-deadline") else late).append((deadline, rid, b))
    send, defer, total, taking = [], [], HEADER, True
    for cls, group in (("timely", sorted(timely)), ("late", sorted(late))):
        for _, rid, b in group:
            if taking and (budget is None or total + len(b) <= budget or drops("plan-budget")):
                send.append((rid, cls, b))
                total += len(b)
            else:
                taking = False
                defer.append(rid)
    raws = [b for _, _, b in sorted(send)]
    return {"send": [{"id": rid.hex(), "class": cls} for rid, cls, _ in send], "defer": [x.hex() for x in defer],
            "drop": [x.hex() for x in sorted(drop)], "file": write_upload(task["tid"], raws).hex() if send else None}


def plan_outcome(doc, c):
    _, cat, task, _ = receiver_task_file(doc, c["task_file"])
    return plan(task, cat, [bytes.fromhex(x) for x in c["stored"]], c["t_ms"], c["budget"])


DELETE = {1, 2, 3, 4, 5, 6, 7, 9, 10}     # §6.6: DAP §4.1's errors but report_too_early


def settle(sent, response):
    """§6.6."""
    _, reports = read_upload(sent)
    ids = [x["m"]["id"] for x in reports]
    entries = None
    if len(response) % 17 == 0:
        es = [(response[k:k + 16], response[k + 16]) for k in range(0, len(response), 17)]
        at, ok = 0, True
        for rid, _ in es:
            if drops("settle-subsequence"):
                continue
            while at < len(ids) and ids[at] != rid:
                at += 1
            if at == len(ids):
                ok = False
                break
            at += 1
        entries = es if ok else None
    if entries is None:
        return {"settles": False, "delete": [], "keep": [x.hex() for x in ids]}
    errors = dict(entries)
    delete, keep = [], []
    for rid in ids:
        e = errors.get(rid)
        if e is None or e in DELETE or (e == KEEP and drops("settle-too-early")) or \
                (e not in REPORT_ERRORS and drops("settle-unknown-value")):
            delete.append(rid.hex())
        else:
            keep.append(rid.hex())
    return {"settles": True, "delete": delete, "keep": keep}


# ------------------------------------------------------------------ §8 verification

def verify_receipt(receipt, cat_bytes, reports):
    """§8.1, in order: the outcome as accepted, or the refusal."""
    cat = read_catalogue(cat_bytes)
    task = {}

    def step2(cfg):
        k = task_rules(cfg, cat)
        task.update(k)
        return k["t"], k["info"], k["p"], k["tid"]

    out = TAV.verify_receipt(receipt, reports, task_check=step2)                     # 1 to 11
    batch = (out["batch"]["start"], out["batch"]["duration"])
    if not TAV.within(batch, task["interval"]) and not drops("batch-in-task"):       # 12
        raise Refuse("batch_invalid", "leader")
    res, n = out["result"], out["report_count"]                                      # 13
    total = sum(res) % TAV.Field128.P if drops("count-integer") else sum(res)
    elements = drops("count-integer") or all(x <= n for x in res)
    if not elements or (total != n and not drops("count-sum")):
        raise Refuse("count_mismatch", "result")
    out["reports_checked"] = reports is not None
    return out, cat, task


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


def rows(cat, vec):
    out = []
    for i in range(len(cat["creatives"])):
        for j in range(len(cat["regions"]) + 1):
            c = vec[cell_of(cat, i, j, 1):cell_of(cat, i, j, 1) + 5]
            out.append({"creative": i, "column": column_name(cat, j), "presentations": sum(c),
                        "quartiles": [sum(c[k:]) for k in range(5)], "floor_met": sum(c) >= FLOOR})
    return out


def table(cat_bytes, task_bytes, given):
    """§8.2."""
    try:
        cat = read_catalogue(cat_bytes)                                              # 1
        task = task_rules(task_bytes, cat)
    except Refuse as e:
        return {"outcome": "refuse", "reason": e.reason}
    tp = task["t"]["time_precision"]
    entries = []
    for g in given:                                                                  # 2
        rb = bytes.fromhex(g["receipt"])
        e = {"digest": None}
        try:
            rc = TAV.read_receipt(rb)
            e["digest"] = sha(rc["body"]).hex()
            out, _, _ = verify_receipt(rb, cat_bytes, g["reports"])
            if rc["task"] != task_bytes and not drops("task-equal"):
                raise Refuse("task_mismatch", "task")
            e["out"] = out
        except Refuse as x:
            e["verdict"] = "refuse reason=%s at=%s" % (x.reason, x.at)
        entries.append(e)
    one = {}                                                                         # 3
    for k, e in enumerate(entries):
        if "out" in e:
            one.setdefault(k if drops("dedupe") else e["digest"], e["out"]["batch"])
    over = set()
    for a, x in one.items():
        for b, y in one.items():
            if a != b and x["start"] < y["start"] + y["duration"] and y["start"] < x["start"] + x["duration"]:
                over.add(a)
    if drops("overlap"):
        over = set()
    stated, standing, totals = [], [], None
    for k, e in enumerate(entries):
        if "out" not in e:
            stated.append({"digest": e["digest"], "verdict": e["verdict"]})
            continue
        key = k if drops("dedupe") else e["digest"]
        if key in over:
            stated.append({"digest": e["digest"], "verdict": "set-aside reason=receipts_overlap"})
            continue
        out = e["out"]
        b = out["batch"]
        checked = out["reports_checked"] or drops("reports-checked")
        end = b["start"] + b["duration"]
        stated.append({"digest": e["digest"], "verdict": "stands",
                       "batch": {"start": b["start"], "duration": b["duration"], "from_ms": b["start"] * tp * 1000,
                                 "to_ms": end * tp * 1000},
                       "collectable_from_ms": (end * tp + cat["hold_s"]) * 1000, "report_count": out["report_count"],
                       "reports_checked": checked, "rows": rows(cat, out["result"])})
        standing.append((key, out, checked))                                        # 4
    # Each receipt counted once, its rows summed as exact integers; one given more than once is
    # checked when any of its occurrences was.
    counted, batches, vec = set(), [], [0] * cat["cells"]
    for key, out, _ in standing:
        if key in counted:
            continue
        counted.add(key)
        batches.append({"start": out["batch"]["start"], "duration": out["batch"]["duration"]})
        vec = [a + b for a, b in zip(vec, out["result"])]
    checked = {key for key, _, c in standing if c}
    every = all(c for _, _, c in standing) if drops("checked-any") else all(k in checked for k in counted)
    totals = {"batches": batches, "reports_checked": every, "rows": rows(cat, vec) if standing else []}
    return {"outcome": "table", "receipts": stated, "totals": totals}


# ------------------------------------------------------------------ §6.3 and §7: a flow, re-run

def open_input_share(sk, role, aad, ct):
    """DAP §4.4.2.1: the PlaintextInputShare's private extensions and payload."""
    def go():
        pt = TAV.hpke_open(ct["enc"], sk, b"dap-20 input share" + bytes([1, role]), aad, ct["payload"])
        r = TAV.Rd(pt)
        exts = TAV.read_exts(r)
        payload = r.opaque(4)
        r.end()
        if not payload:
            raise ValueError("empty payload")
        return exts, payload
    return memo(("open", sk, role, aad, ct["config_id"], ct["enc"], ct["payload"]), go)


def verify_init(p, vk, ctx, agg, nonce, public, share):
    return memo(("init", p.v.length, p.v.chunk, vk, ctx, agg, nonce, public, share),
                lambda: p.verify_init(vk, ctx, agg, nonce, public, share))


def meas_share(p, ctx, agg, share):
    return memo(("meas", p.v.length, p.v.chunk, ctx, agg, share), lambda: p.meas_share(ctx, agg, share))


def in_interval(x, iv):
    return iv[0] <= x < iv[0] + iv[1]


def run_flow(doc, f):
    """One flow of §11, re-run from its reports: the failures."""
    fails = []
    K = doc["keys"]

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

    try:
        tf, cat, task, _ = receiver_task_file(doc, f["task_file"])
    except Refuse as e:
        bad("its task file is refused %s" % e.reason)
        return fails
    t, p, tid, iv, cfg = task["t"], task["p"], task["tid"], task["interval"], task["cfg"]
    tp, hold, floor = t["time_precision"], cat["hold_s"], t["min_batch_size"]
    if tid.hex() != f["task_id"]:
        bad("task id")
    vk = TAV.hkdf_expand(TAV.hkdf_extract(b"verify_key", bytes.fromhex(K["verify_key_seed"])), tid, 32)
    if vk.hex() != f["verify_key"]:
        bad("verification key (telemetry-aggregation §4.4)")
    ctx = b"dap-20" + tid
    lsk, hsk, csk = (bytes.fromhex(K[k]["private_key"]) for k in ("leader_hpke", "helper_hpke", "collector_hpke"))
    lconf, hconf = suite_config(tf["leader_hpke"]), suite_config(tf["helper_hpke"])
    if lconf["pk"] != TAV.x25519(lsk, TAV.BASE) or hconf["pk"] != TAV.x25519(hsk, TAV.BASE):
        bad("the configuration the reports are sealed to is not the aggregator's key (§6.3)")
    leader_ids = [c["id"] for c in tf["leader_hpke"]]
    listed = {}
    for x in f["reports"] + f.get("late", []):
        listed[x["id"]] = x
    late = {x["id"] for x in f.get("late", [])}

    seen, collected, pending, first = set(), [], [], {}
    fate, buckets, outcomes_stated = {}, {}, None

    def merged(q):
        out = {"l": [0] * p.v.OUTPUT_LEN, "h": [0] * p.v.OUTPUT_LEN, "n": 0, "ck": bytes(32), "times": [], "ids": []}
        for tm in range(q[0], q[0] + q[1]):
            if tm in buckets:
                b = buckets[tm]
                out["l"], out["h"] = TAV.vadd(p.f, out["l"], b["l"]), TAV.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

    def collected_at(tm):
        return any(in_interval(tm, q) for q in collected)

    for s in f["steps"]:
        if s["step"] == "upload":                                                    # §6.4; DAP §4.4.2.2
            try:
                utid, reps = read_upload(bytes.fromhex(s["file"]))
            except Refuse:
                bad("upload %s does not read" % s["name"])
                continue
            if utid != tid:
                bad("upload %s names another task" % s["name"])
            if "sources" in s:
                byid = {}
                for src in s["sources"]:
                    try:
                        stid, srep = read_upload(bytes.fromhex(src))
                    except Refuse:
                        bad("a source of %s does not read" % s["name"])
                        continue
                    if stid != tid:
                        bad("a source of %s names another task" % s["name"])
                    for x in srep:
                        if byid.setdefault(x["m"]["id"], x["raw"]) != x["raw"]:
                            bad("two reports of one id with other bytes merged into one file")
                if write_upload(tid, [byid[k] for k in sorted(byid)]).hex() != s["file"]:
                    bad("upload %s is not its sources merged, ascending by id (§6.4)" % s["name"])
            resp, errs = b"", []
            for rep in reps:
                m, rid = rep["m"], rep["m"]["id"]
                first.setdefault(rid.hex(), rep)
                if rep["leader"]["config_id"] not in leader_ids:
                    e = "hpke_unknown_config_id"
                elif not in_interval(m["time"], iv) and not drops("upload-dropped"):
                    e = "report_dropped"
                elif (rid in seen or collected_at(m["time"])) and not drops("upload-replay"):
                    e = "report_replayed"
                else:
                    e = None
                    seen.add(rid)
                    pending.append(rep)
                if e:
                    resp += rid + bytes([CODE[e]])
                    errs.append({"id": rid.hex(), "error": e})
                    if rid not in seen:
                        fate[rid.hex()] = (e, None)
            if resp.hex() != s["response"] or errs != s["errors"]:
                bad("upload %s: the leader's UploadErrors" % s["name"])
            if "shares" in s:
                srcs = [[x["m"]["id"] for x in read_upload(bytes.fromhex(src))[1]] for src in s["sources"]]
                got = [b"".join(resp[k:k + 17] for k in range(0, len(resp), 17) if resp[k:k + 16] in ids).hex()
                       for ids in srcs]
                if got != s["shares"]:
                    bad("upload %s: each source's share of the response (§6.4)" % s["name"])
        elif s["step"] == "aggregate":                                               # DAP §4.5; §7.2
            init = bytes([0]) + u32(0) + (0).to_bytes(2, "big")
            resp, refused, job = b"", [], []
            for rep in pending:
                m, rid = rep["m"], rep["m"]["id"]
                aad = TAV.input_share_aad(tid, cfg, m, rep["public"])
                try:
                    lexts, lpay = open_input_share(lsk, 2, aad, rep["leader"])
                except ValueError:
                    refused.append((rid, "hpke_decrypt_error", None))
                    continue
                if not in_interval(m["time"], iv):
                    refused.append((rid, "report_dropped", None))
                    continue
                why = report_rule(cat, tid, rid, m["exts"], lexts, 2) if not drops("token-check") else None
                if why:
                    refused.append((rid,) + why)
                    continue
                try:
                    lstate, lvs = verify_init(p, vk, ctx, 0, rid, rep["public"], lpay)
                except ValueError:
                    refused.append((rid, "vdaf_verify_error", None))
                    continue
                out = b"\x00" + u32(len(lvs)) + lvs
                init += m["raw"] + u32(len(rep["public"])) + rep["public"] + TAV.write_ct(rep["helper"]) + \
                    u32(len(out)) + out
                try:
                    hexts, hpay = open_input_share(hsk, 3, aad, rep["helper"])
                except ValueError:
                    resp += rid + b"\x02" + bytes([CODE["hpke_decrypt_error"]])
                    job.append((rid, "hpke_decrypt_error", None))
                    continue
                why = report_rule(cat, tid, rid, m["exts"], hexts, 3) if not drops("token-check") else None
                if why:
                    resp += rid + b"\x02" + bytes([CODE[why[0]]])
                    job.append((rid,) + why)
                    continue
                try:
                    hstate, hvs = verify_init(p, vk, ctx, 1, rid, rep["public"], hpay)
                    msg = p.message(ctx, [lvs, hvs])
                    hout = p.verify_next(hstate, msg)
                except ValueError:
                    resp += rid + b"\x02" + bytes([CODE["vdaf_verify_error"]])
                    job.append((rid, "vdaf_verify_error", None))
                    continue
                if collected_at(m["time"]):                                          # DAP §4.5.4.3
                    resp += rid + b"\x02" + bytes([CODE["batch_collected"]])
                    job.append((rid, "batch_collected", None))
                    continue
                fin = b"\x02" + u32(len(msg)) + msg
                resp += rid + b"\x00" + u32(len(fin)) + fin
                lout = p.verify_next(lstate, msg)
                b = buckets.setdefault(m["time"], {"l": [0] * p.v.OUTPUT_LEN, "h": [0] * p.v.OUTPUT_LEN, "n": 0,
                                                   "ck": bytes(32), "ids": []})
                b["l"] = TAV.vadd(p.f, b["l"], TAV.decode_vec(p.f, lout))
                b["h"] = TAV.vadd(p.f, b["h"], TAV.decode_vec(p.f, hout))
                b["n"] += 1
                b["ck"] = bytes(a ^ c for a, c in zip(b["ck"], sha(rid)))
                b["ids"].append((rid, m["time"]))
                job.append((rid, "committed", None))
            pending = []
            for rid, o, why in refused + job:
                fate[rid.hex()] = o if o == "committed" else (o, why)
            outcomes = [{"id": rid.hex(), "outcome": o, "reason": why} for rid, o, why in refused + job]
            if init.hex() != s["init_req"]:
                bad("the AggregationJobInitReq is not DAP §4.5.3.1's")
            if resp.hex() != s["resp"]:
                bad("the AggregationJobResp is not DAP §4.5.3.2's")
            if outcomes != s["outcomes"]:
                bad("the reports' outcomes at aggregation")
            outcomes_stated = s["outcomes"]
        elif s["step"] == "collect":                                                 # DAP §4.6; §7.1
            req = bytes.fromhex(s["request"])
            try:
                c = TAV.read_collection(req)
                q = TAV.interval_of(c["config"])
            except ValueError:
                bad("collection %s does not read" % s["name"])
                continue
            if s["batch"] != {"start": q[0], "duration": q[1]}:
                bad("collection %s: its batch" % s["name"])
            mg = merged(q)
            due = (q[0] + q[1]) * tp + hold
            if s["now_s"] < due and not drops("hold"):
                got = "processing"
            elif c["agg_param"]:
                got = "invalidAggregationParameter"
            elif q[1] == 0:
                got = "batchInvalid"
            elif any(q[0] < x[0] + x[1] and x[0] < q[0] + q[1] for x in collected):
                got = "batchOverlap"
            elif mg["n"] < floor:
                got = "invalidBatchSize"
            else:
                got = "completed"
            if got != s["outcome"]:
                bad("collection %s: stated %s, the leader gives %s" % (s["name"], s["outcome"], got))
            if "helper_request" in s:                                                # DAP §4.6.4
                try:
                    r = TAV.Rd(bytes.fromhex(s["helper_request"]))
                    hc = {"batch_mode": r.u(1), "config": r.opaque(2), "agg_param": r.opaque(4),
                          "extensions": TAV.read_exts(r)}
                    mode, sel, n, ck = r.u(1), r.opaque(2), r.u(8), r.take(32)
                    r.end()
                    hq = TAV.interval_of(sel)
                    ok = mode == 1 and hc["batch_mode"] == 1 and TAV.interval_of(hc["config"]) == q and hq == q and \
                        (n, ck) == (mg["n"], mg["ck"])
                    if mg["n"] < floor:
                        hgot = "invalidBatchSize"
                    else:
                        hgot = "batchMismatch" if (n, ck) != (mg["n"], mg["ck"]) else "accept"
                except ValueError:
                    ok, hgot = False, "invalidMessage"
                if not ok or hgot != s["helper_outcome"]:
                    bad("collection %s: the helper's request or its refusal" % s["name"])
            if got == "completed" and s["outcome"] == "completed":
                collected.append(q)
                check_collection(doc, s, cfg, tid, req, q, mg, csk, task, bad)
    for rid, x in listed.items():
        rep = first.get(rid)
        if rep is None:
            bad("report %s is never uploaded" % rid)
            continue
        if rep["m"]["time"] != x["time"]:
            bad("report %s: its time" % rid)
        if rep["leader"]["config_id"] != lconf["id"] or rep["helper"]["config_id"] != hconf["id"]:
            bad("report %s is not sealed to each list's first configuration of DAP's suite (§6.3)" % rid)
        if rid in late:
            want = ("report_replayed", None)
        elif x["kind"] == "presentation":
            want = "committed"
        else:
            want = (x["error"], x["reason"])
        if fate.get(rid) != want:
            bad("report %s (%s): stated %s, the run gives %s" % (rid, x["kind"], want, fate.get(rid)))
        if x["kind"] == "presentation":
            check_presentation(doc, f, x, rep, tf, cat, task, lsk, hsk, ctx, bad)
    for rid, rep in first.items():
        if rid not in listed:
            bad("an uploaded report %s is not listed" % rid)
    if outcomes_stated is None:
        bad("no aggregation job")
    return fails


def check_presentation(doc, f, x, rep, tf, cat, task, lsk, hsk, ctx, bad):
    """A presentation's report: §6.2's contribution, §6.3's extensions and its one-hot measurement."""
    p, tid, m = task["p"], task["tid"], rep["m"]
    aad = TAV.input_share_aad(tid, task["cfg"], m, rep["public"])
    try:
        lexts, lpay = open_input_share(lsk, 2, aad, rep["leader"])
        hexts, hpay = open_input_share(hsk, 3, aad, rep["helper"])
    except ValueError:
        bad("presentation %s does not open" % x["id"])
        return
    if m["exts"] or lexts != hexts or (cat["attest"] and [e[0] for e in lexts] != [TOKEN_EXT]) or \
            (not cat["attest"] and lexts):
        bad("presentation %s: its extensions are not §6.3's" % x["id"])
        return
    pool = [{"id": x["id"], "token": lexts[0][1].hex()}] if cat["attest"] else []
    record = bytes.fromhex(doc["fixtures"]["records"][x["record"]]["bytes"])
    region = None if x["region"] is None else x["region"].encode("utf-8")
    got = contribute(doc, cat, task, record, [bytes.fromhex(x["claim"])], region, x["at_ms"], pool)
    want = {"outcome": "report", "cell": x["cell"], "creative": x["creative"], "column": x["column"],
            "quartile": x["quartile"], "time": x["time"], "id": x["id"] if cat["attest"] else "fresh", "discarded": []}
    if got != want:
        bad("presentation %s: its contribution (§6.2) is %s" % (x["id"], json.dumps(got)))
    meas = TAV.vadd(p.f, meas_share(p, ctx, 0, lpay), meas_share(p, ctx, 1, hpay))
    if meas != [1 if k == x["cell"] else 0 for k in range(cat["cells"])]:
        bad("presentation %s: its measurement is not one at its cell (§6.3)" % x["id"])


def check_collection(doc, s, cfg, tid, req, q, mg, csk, task, bad):
    """A completed collection: the shares, the statements, the result and the receipt."""
    p, info = task["p"], task["info"]
    share_req = req + b"\x01" + (16).to_bytes(2, "big") + q[0].to_bytes(8, "big") + q[1].to_bytes(8, "big") + \
        mg["n"].to_bytes(8, "big") + mg["ck"]
    if share_req.hex() != s["share_req"]:
        bad("collection %s: the AggregateShareReq is not DAP §4.6.4's" % s["name"])
    aad = TAV.agg_share_aad(tid, cfg, req)
    try:
        hct = TAV.read_ct(TAV.Rd(bytes.fromhex(s["helper_share"])))
        hshare = TAV.hpke_open(hct["enc"], csk, b"dap-20 aggregate share" + bytes([3, 0]), aad, hct["payload"])
        r = TAV.Rd(bytes.fromhex(s["resp"]))
        n, iv = r.u(8), (r.u(8), r.u(8))
        lct, hct2 = TAV.read_ct(r), TAV.read_ct(r)
        r.end()
        lshare = TAV.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 = TAV.read_ct(TAV.Rd(bytes.fromhex(s[k])))
            info_s = b"wai-telemetry-aggregate statement" + bytes([role, 0])
            sts.append(TAV.read_statement(TAV.hpke_open(sct["enc"], csk, info_s, aad, sct["payload"])))
    except (ValueError, Refuse):
        bad("collection %s does not open with the collector's key" % s["name"])
        return
    times = mg["times"]
    if n != mg["n"] or iv != (min(times), max(times) - min(times) + 1) or hct2 != hct:
        bad("collection %s: the CollectionJobResp's count, interval or helper share" % s["name"])
    if TAV.decode_vec(p.f, lshare) != mg["l"] or TAV.decode_vec(p.f, hshare) != mg["h"]:
        bad("collection %s: an aggregate share is not the batch's" % s["name"])
    for st, role, share, key in ((sts[0], 2, lshare, info["leader"]), (sts[1], 3, hshare, info["helper"])):
        if (st["role"], st["task_id"], st["collection"], st["batch"], st["count"], st["checksum"], st["agg_share"],
                st["signer"]) != (role, tid, sha(req), q, mg["n"], mg["ck"], share, key):
            bad("collection %s: a statement is not its share's (telemetry-aggregation §6.2)" % s["name"])
        if ED.verify(key, b"wai:telemetry-aggregate/share\x01" + st["body"], st["sig"]) is not None:
            bad("collection %s: a statement's signature" % s["name"])
    result = TAV.vadd(p.f, mg["l"], mg["h"])
    if result != s["result"]:
        bad("collection %s: the result" % s["name"])
    rc = TAV.read_receipt(bytes.fromhex(s["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, req, iv):
        bad("collection %s: the receipt is not the collection's" % s["name"])
    given = sorted(json.dumps({"id": rid.hex(), "time": tm, "task": tid.hex()}, sort_keys=True)
                   for rid, tm in mg["ids"])
    holders = [x for x in doc["receipts"] if x["receipt"] == s["receipt"]]
    with_reports = [x for x in holders if x["reports"] is not None]
    if not holders or (with_reports and not any(sorted(json.dumps(g, sort_keys=True) for g in x["reports"]) == given
                                                 for x in with_reports)):
        bad("collection %s: the receipts section does not hold its receipt with its reports" % s["name"])


# ------------------------------------------------------------------ the corpus

KINDS = ("catalogues", "tasks", "task_files", "contributions", "uploads", "plans", "settles", "receipts", "tables")


def run(doc, kind, c):
    if kind == "catalogues":
        return catalogue_outcome(bytes.fromhex(c["bytes"]))
    if kind == "tasks":
        return task_outcome(bytes.fromhex(c["task"]), bytes.fromhex(c["catalogue"]))
    if kind == "task_files":
        return task_file_outcome(doc, c)
    if kind == "contributions":
        return contribution_outcome(doc, c)
    if kind == "uploads":
        return upload_outcome(bytes.fromhex(c["bytes"]))
    if kind == "plans":
        return plan_outcome(doc, c)
    if kind == "settles":
        return settle(bytes.fromhex(c["sent"]), bytes.fromhex(c["response"]))
    if kind == "receipts":
        return receipt_outcome(c)
    return table(bytes.fromhex(c["catalogue"]), bytes.fromhex(c["task"]), c["receipts"])


def check_fixtures(doc):
    """What the fixtures say of themselves: each creative's digest (ad-presentation §4), each catalogue,
    task and task file the cases name by fixture."""
    fails = []
    fx = doc["fixtures"]
    for n, c in fx["creatives"].items():
        objs = [{"content_hash": bytes.fromhex(o["content_hash"]), "duration_ms": o["duration_ms"]}
                for o in c["objects"]]
        if APV.creative_digest(objs).hex() != c["digest"]:
            fails.append("creative %s: its digest" % n)
    for n, h in fx["task_files"].items():
        try:
            tf, cat, task, _ = receiver_task_file(doc, n)
            if tf["catalogue"].hex() != fx["catalogues"].get(n, tf["catalogue"].hex()) or \
                    tf["task"].hex() != fx["tasks"].get(n, tf["task"].hex()):
                fails.append("task file %s: not its fixtures' catalogue and task" % n)
        except Refuse as e:
            fails.append("task file %s is refused %s under what the receiver holds" % (n, e.reason))
    return fails


def strings(v, out):
    if isinstance(v, str):
        out.append(v)
    elif isinstance(v, list):
        for x in v:
            strings(x, out)
    elif isinstance(v, dict):
        for x in v.values():
            strings(x, out)


def stated_reasons(doc):
    """Every reason an outcome of the corpus gives, flows included."""
    out = []
    for kind in KINDS:
        for c in doc[kind]:
            o = c["outcome"]
            if isinstance(o, dict) and o.get("reason"):
                out.append(o["reason"])
            for s in [] if kind != "tables" or o.get("outcome") != "table" else o["receipts"]:
                m = re.search(r"reason=(\S+)", s["verdict"])
                if m:
                    out.append(m.group(1))
    for f in doc["flows"]:
        for x in f["reports"]:
            out += [y for y in (x.get("error"), x.get("reason")) if y]
        for s in f["steps"]:
            out += [e["error"] for e in s.get("errors", [])]
            out += [y for o in s.get("outcomes", []) for y in (o["outcome"], o["reason"]) if y and y != "committed"]
            if s["step"] == "collect":
                out += [y for y in (s["outcome"], s.get("helper_outcome"))
                        if y not in ("processing", "completed", "accept", None)]
    return out


ELSEWHERE = set(TAV.REASONS) | set(SPV.REASONS) | set(SPV.OTHER) | set(APV.REASONS) | set(JWP_REASONS) | \
    set(REPORT_ERRORS.values()) | set(PROBLEMS) | {"count_mismatch", "truncated", "malformed", "unknown_version",
                                                    "bad_key", "bad_signature"}
REUSED = ["count_mismatch", "attestation_absent", "attestation_invalid", "task_invalid", "measure_mismatch",
          "floor_too_low", "aggregator_alone", "batch_invalid", "task_mismatch", "batch_too_small", "batch_mismatch",
          "bad_key", "bad_signature", "pod_signer_unknown", "claim_not_viewable", "claim_duplicate",
          "claim_marker_absent", "claim_manifest_unknown", "signer_id_not_origin_bound", "origin_not_anchored",
          "authority_not_granted", "anchor_not_pinned", "key_revoked", "report_replayed", "report_dropped",
          "invalid_message", "hpke_decrypt_error", "vdaf_verify_error", "invalidBatchSize"]


def check(doc, quiet=False):
    """The failures of a corpus document."""
    fails = check_fixtures(doc)
    for kind in KINDS:
        for c in doc[kind]:
            try:
                got = run(doc, kind, c)
            except (Refuse, KeyError, TypeError, ValueError, IndexError, AttributeError) as e:
                got = {"error": repr(e)}
            if got != c["outcome"]:
                fails.append("%s %s" % (kind, c["name"]))
                if not quiet:
                    print("FAIL %s %s\n  stated %s\n  got    %s" % (kind, c["name"], json.dumps(c["outcome"])[:600],
                                                                  json.dumps(got)[:600]))
    for f in doc["flows"]:
        try:
            ff = run_flow(doc, f)
        except (Refuse, KeyError, TypeError, ValueError, IndexError, AttributeError) as e:
            ff = ["flow %s: %r" % (f["name"], e)]
        fails += ff
        if not quiet:
            for x in ff:
                print("FAIL", x)
            if not ff:
                n = len(f["reports"]) + len(f.get("late", []))
                print("PASS flow %s: %d reports re-run through %d steps" % (f["name"], n, len(f["steps"])))
    if doc.get("codes") != REASONS:
        fails.append("the corpus's codes are not §10's")
    given = stated_reasons(doc)
    for code in REASONS + REUSED:
        if code not in given:
            fails.append("no case gives " + code)
    for code in sorted(set(given) - set(REASONS) - ELSEWHERE):
        fails.append("a case gives %s, a code the extension does not name" % code)
    return fails


def bounds():
    """§5.4's and §6.4's size bounds, which no vector carries: a file of exactly the bound that
    reads in every other respect is read (a task file passes step 1 and is refused at step 2, its
    catalogue of zeros not reading), and the same file one byte longer is refused at step 1.
    §4.3's bound decides no outcome: no catalogue that meets §4.3 comes near it."""
    fails = []
    suite = struct.pack(">BHHHH", 1, *SUITE, 32) + b"\x09" * 32

    def hpke_list(cid):
        body = bytes([cid]) + suite[1:]
        return len(body).to_bytes(2, "big") + body

    def task_file(n):
        body = (TASK_MAGIC + b"\x01" + (0).to_bytes(2, "big") + u32(n) + b"\x00" * n + hpke_list(1) + hpke_list(2)
                + b"\x00" * 32 + (1).to_bytes(2, "big") + b"x")
        return body + b"\x00" * 64

    ct = b"\x01" + (1).to_bytes(2, "big") + b"\x01" + u32(1) + b"\x01"

    def upload(n):
        report = b"\x00" * 16 + (0).to_bytes(8, "big") + (0).to_bytes(2, "big") + u32(n) + b"\x00" * n + ct + ct
        return write_upload(b"\x00" * 32, [report])

    def outcome(fn, b):
        try:
            fn(b)
            return "read"
        except Refuse as e:
            return e.reason

    for what, make, bound, fn, inside, past in (
            ("task file", task_file, MAX_TASK_FILE, lambda b: accept_task_file(b, [], []), "catalogue_malformed",
             "task_file_malformed"),
            ("upload file", upload, MAX_UPLOAD, read_upload, "read", "upload_malformed")):
        at = bound - len(make(0))
        for b, want in ((make(at), inside), (make(at + 1), past)):
            got = outcome(fn, b)
            if got != want:
                fails.append("a %s of %d bytes: %s, not %s" % (what, len(b), got, want))
    return fails


# ------------------------------------------------------------------ mutations and weakenings

def flip(h, at):
    b = bytearray(bytes.fromhex(h))
    b[at % len(b)] ^= 0x01
    return b.hex()


def case(d, kind, name):
    return next(c for c in d[kind] if c["name"] == name)


def flow(d, name):
    return next(f for f in d["flows"] if f["name"] == name)


def step(f, name):
    return next(s for s in f["steps"] if s.get("name") == name or s["step"] == name)


def mutations():
    """Copies of the corpus, one defect each, every one of which must fail."""
    def m_catalogue_hold(d):
        c = case(d, "catalogues", "flight")
        b = bytearray.fromhex(c["bytes"])
        b[5 + 4 + 2 + 14 + 3] ^= 0x01            # the last byte of hold_s
        c["bytes"] = b.hex()

    def m_region_column(d):
        case(d, "contributions", "region_subdivision_of_listed_country")["outcome"].update(column="other", cell=14)

    def m_bucket_edge(d):
        case(d, "contributions", "bucket_first_ms")["at_ms"] -= 1

    def m_quartile(d):
        case(d, "contributions", "q4_three_quarters")["outcome"].update(quartile=5, cell=4)

    def m_pool_discarded(d):
        case(d, "contributions", "token_pool_skips_bad")["outcome"]["discarded"] = []

    def m_task_retention(d):
        case(d, "tasks", "flight")["outcome"]["retention_s"] += 1

    def m_task_file_bit(d):
        c = case(d, "task_files", "flight")
        c["bytes"] = flip(c["bytes"], 40)

    def m_anchor(d):
        case(d, "task_files", "flight")["anchors"] = ["fetched"]

    def m_floor_flag(d):
        case(d, "tables", "flight")["outcome"]["totals"]["rows"][0]["floor_met"] = False

    def m_row_count(d):
        case(d, "tables", "flight")["outcome"]["receipts"][1]["rows"][5]["presentations"] = 50

    def m_receipt_byte(d):
        c = case(d, "receipts", "r0")
        c["receipt"] = flip(c["receipt"], len(c["receipt"]) // 4)

    def m_receipt_reports(d):
        case(d, "receipts", "r0")["reports"].pop()

    def m_hold(d):
        s = step(flow(d, "flight"), "collect_b0_held")
        s["now_s"] += 1

    def m_late_error(d):
        step(flow(d, "flight"), "late_b0")["errors"][0]["error"] = "batch_collected"

    def m_token_outcome(d):
        o = step(flow(d, "flight"), "aggregate")["outcomes"][0]
        o.update(outcome="committed", reason=None)

    def m_init_byte(d):
        s = step(flow(d, "flight"), "aggregate")
        s["init_req"] = flip(s["init_req"], 400)

    def m_merge_share(d):
        s = step(flow(d, "flight"), "relay_merged")
        s["shares"].reverse()

    def m_presentation_cell(d):
        x = next(r for r in flow(d, "open")["reports"] if r["kind"] == "presentation")
        x["cell"] = (x["cell"] + 1) % 5

    def m_budget(d):
        case(d, "plans", "budget_exact")["budget"] -= 1

    def m_plan_retention(d):
        case(d, "plans", "retention_exact")["t_ms"] -= 1

    def m_upload_order(d):
        c = case(d, "uploads", "three_reports")
        tid, reps = read_upload(bytes.fromhex(c["bytes"]))
        c["bytes"] = write_upload(tid, [reps[1]["raw"], reps[0]["raw"], reps[2]["raw"]]).hex()

    def m_settle_too_early(d):
        o = case(d, "settles", "each_error")["outcome"]
        o["delete"] = sorted(o["delete"] + o["keep"])
        o["keep"] = []

    def m_overlap_gone(d):
        d["tables"].remove(case(d, "tables", "flight_overlap"))

    def m_codes(d):
        d["codes"].reverse()

    def m_creative_fixture(d):
        o = d["fixtures"]["creatives"]["A"]["objects"][0]
        o["duration_ms"] += 1

    return [m_catalogue_hold, m_region_column, m_bucket_edge, m_quartile, m_pool_discarded, m_task_retention,
            m_task_file_bit, m_anchor, m_floor_flag, m_row_count, m_receipt_byte, m_receipt_reports, m_hold,
            m_late_error, m_token_outcome, m_init_byte, m_merge_share, m_presentation_cell, m_budget,
            m_plan_retention, m_upload_order, m_settle_too_early, m_overlap_gone, m_codes, m_creative_fixture]


WEAKENINGS = {
    "cat-version": "§4.3 rule 1: version 1",
    "cat-terms": "§4.3 rule 2: the terms",
    "cat-hold": "§4.3 rule 3: a hold above zero",
    "cat-signers": "§4.3 rule 4: one to sixteen signers",
    "cat-creatives": "§4.3 rule 5: a creative",
    "cat-order": "§4.3 rules 4 to 8: strictly ascending lists",
    "cat-region": "§4.3 rule 6: each region a code",
    "cat-overlap": "§4.3 rule 6: no country beside its subdivision",
    "cat-cells": "§4.3 rule 7: at most 4,096 cells",
    "cat-keys": "§4.3 rule 8: the keys' count",
    "cat-key-form": "§4.3 rule 8: each key's form",
    "catalogue-digest": "§5.3 step 2: measure 5 and the catalogue's digest",
    "task-precision": "§5.3 step 3: an hour's precision",
    "task-interval": "§5.3 step 3: a flight of a duration",
    "task-retention": "§5.3 step 3: the retention instant's bound",
    "chunk-length": "§5.3 step 4: the chunk length",
    "tf-version": "§5.4 step 1: version 1",
    "tf-signer-ascii": "§5.4 step 1: an ASCII signer_id",
    "tf-hpke-ids": "§5.4 step 1: distinct configuration ids",
    "tf-hpke-suite": "§5.4 step 1: DAP's suite in each list",
    "tf-signature": "§5.4 step 4: the file's signature",
    "tf-authority": "§5.4 step 5: the authority aggregate/task",
    "tf-pinned": "§5.4 step 5: a pinned anchor",
    "tf-revocation": "§5.4 step 5: revisions the bearer brings",
    "shared": "§6.2 step 2: a shared record",
    "terms-equal": "§6.2 step 3: the catalogue's terms",
    "creative-listed": "§6.2 step 5: a listed creative",
    "instant": "§6.2 step 6: a known instant",
    "flight-window": "§6.2 step 7: a time within the flight",
    "token-pool": "§6.2 step 8: each pair's token",
    "region-subdivision": "§4.2: a subdivision's column is its listed country's",
    "hpke-first": "§6.3: the first configuration of DAP's suite",
    "token-check": "§7.2: the aggregators' report rule",
    "token-challenge": "§7.2: a token's challenge",
    "helper-token": "§7.2: the helper checks its own share",
    "public-extension": "§7.2: no public extension",
    "hold": "§7.1: the hold",
    "upload-dropped": "DAP §4.4.2.2: a report outside the flight dropped at upload",
    "upload-replay": "DAP §4.4.2.2: a report replayed at upload",
    "batch-in-task": "§8.1 step 12: the batch within the flight",
    "count-sum": "§8.1 step 13: the elements sum to the count",
    "count-integer": "§8.1 step 13: integers, each at most the count",
    "task-equal": "§8.2 step 2: the given task",
    "dedupe": "§8.2 step 3: equal receipts are one",
    "overlap": "§8.2 step 3: overlapping receipts set aside",
    "reports-checked": "§8.2: whether a receipt's reports were checked",
    "checked-any": "§8.2: a receipt given more than once checked when any occurrence was",
    "tf-bound": "§5.4 step 1: at most 1,048,576 bytes",
    "upload-bound": "§6.4: at most 16,777,216 bytes",
    "upload-version": "§6.4: version 1",
    "upload-length": "§6.4: a file that ends before its upload's length",
    "upload-order": "§6.4: reports in ascending order of id",
    "plan-deadline": "§6.5: timely and late",
    "plan-retention": "§6.5: dropped at the retention instant",
    "plan-budget": "§6.5: the budget",
    "settle-subsequence": "§6.6: a response of the file's ids in order",
    "settle-too-early": "§6.6: report_too_early kept",
    "settle-unknown-value": "§6.6: a value DAP does not define kept",
}


def main(argv):
    if len(argv) != 2:
        print(__doc__)
        return 2
    fails = []
    st = ED._self_test()
    ct = TAV.crypto_self_test()
    print(("PASS" if not st and not ct else "FAIL") + " Ed25519, TurboSHAKE128, X25519 and HPKE self-tests")
    fails += (["ed25519 self-test"] if st else []) + ct
    print(("PASS" if len(REASONS) == 11 else "FAIL") + " §10 read from the extension: %d codes" % len(REASONS))
    with open(os.path.join(argv[1], "vectors.json"), encoding="utf-8") as fh:
        doc = json.load(fh)
    got = check(doc)
    fails += got
    sizes = ", ".join("%d %s" % (len(doc[k]), k) for k in KINDS + ("flows",))
    print("%s %s: %d failure(s)" % ("PASS" if not got else "FAIL", sizes, len(got)))
    extra = sorted(set(os.listdir(argv[1])) - {"vectors.json", "README.md"})
    if extra:
        fails.append("files the corpus does not hold: %s" % extra)
    b = bounds()
    print(("PASS" if not b else "FAIL") + " the size bounds of §5.4 and §6.4, at their edge")
    fails += b
    caught = 0
    for m in mutations():
        d = copy.deepcopy(doc)
        m(d)
        if check(d, quiet=True):
            caught += 1
        else:
            fails.append("mutation %s was not caught" % m.__name__)
    n = len(mutations())
    print("%s %d of %d mutated copies fail" % ("PASS" if caught == n else "FAIL", caught, n))
    weak = 0
    for name, what in WEAKENINGS.items():
        SKIP.clear()
        SKIP.add(name)
        if check(doc, quiet=True) or bounds():
            weak += 1
        else:
            fails.append("a verifier without %s (%s) passes the corpus" % (what, name))
        SKIP.clear()
    n = len(WEAKENINGS)
    print("%s %d of %d weakened verifiers are caught" % ("PASS" if weak == n else "FAIL", weak, n))
    for x in fails:
        print("FAIL", x)
    print(("PASS" if not fails else "FAIL") + " impression aggregates corpus")
    return 1 if fails else 0


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