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

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

Written from the extension's text. It shares no code with wai-rs, which generates the corpus,
and never runs it. It takes from the second implementations of the texts this extension builds
on what they define: group and object receipts, leaves and link hashes from the JWP corpus's
(`../jwp-conformance/jwp_verify.py`, jwp-receipts §2.2, §3); a claim's steps 1 to 5 from the
claim corpus's (`../claim-conformance/claim_verify.py`, jwp-receipts §3.1, whose steps 2 and 4
hold §5.1 and §5.2 here); a transmission's and a reconciliation's step 6 and the root over
leaves from the bearer corpus's (`bearer_verify.py`, bearer-receipts §4.3, §5.3); an
acknowledgement and its item lists from the settlement corpus's (`settlement_verify.py`,
settlement §3.6); a key log revision held under rules 1 to 3 from the signer corpus's
(`../signer-conformance/signer_verify.py`, jwp-receipts §7.2); the held-at statement's rule, a
holdings record held to its group, a timeliness claim against its group and timeliness lines
from the timeliness corpus's (`timeliness_verify.py`, timeliness §2, §4.3, §6); and region codes
from the rung price corpus's (`rung_price_verify.py`, rung-price §2). Everything this extension
adds is written here:

- §4.4: placing a server key by the heads a relying party names;
- §5.3: a cache-transmission claim against its group and the object receipts held;
- §7: the bearer verdict with the caches' instances beside the publisher's, the credit per
  instance, the status, a statement offered, a sink's own instance and the energy bound;
- §8: the population, the tally's checks, the held-at statements of acknowledgements and counted
  reconciliations, the count, its servers and the outcome;
- §9: timeliness lines that read acknowledgements and counted reconciliations too.

For every case it requires the outcome the corpus states, field for field; that every reason of
§11's table appears in the corpus; 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. A bound no vector carries is checked directly.

    python3 cache_authority_verify.py <live-conformance/cache-authority>

Exit status 0 only if every case verifies, every mutated copy fails, every weakened verifier is
caught and every direct check holds. Python 3.8 or later; the `cryptography` package, which the
claim verifier imports.
"""

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

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
for d in ("claim-conformance", "jwp-conformance", "signer-conformance"):
    sys.path.insert(0, os.path.join(HERE, "..", d))
import claim_verify as CV  # noqa: E402
import jwp_verify as JV  # noqa: E402
import signer_verify as SG  # noqa: E402
import bearer_verify as BV  # noqa: E402
import settlement_verify as SET  # noqa: E402
import timeliness_verify as TV  # noqa: E402
import rung_price_verify as RP  # noqa: E402

HV = TV.HV  # the holdings corpus's reader, which the timeliness verifier loads

EXTENSION = os.path.join(HERE, "..", "extensions", "cache-authority.md")
JWP_RECEIPTS = os.path.join(HERE, "..", "extensions", "jwp-receipts.md")
SETTLEMENT = os.path.join(HERE, "..", "extensions", "settlement.md")
LAST_INSTANT = 253_402_300_799_999       # §2, timeliness §2: 9999-12-31T23:59:59.999Z
MAX_TARGET = 999_999                     # §8.2 step 2
MAX_KEYS = 1 << 20                       # §8.1: 1,048,576 keys
PPM = 1_000_000
POPULATION = b"wai:cache-population\x01"  # §8.1
SERVING = ("cache/serve", "live/relay", "live/compositor", "live/ingest")  # §4.4, in order
DECLARATION = 2                          # settlement §3.1 `mode`

# A weakened run drops one of these checks (see `WEAKENINGS`); None checks everything.
DROPPED = None


def drops(name):
    return DROPPED == name


def reasons_table():
    """§11's codes, in order, each with where the table says it is raised."""
    text = TV.text_of(EXTENSION)
    at = text.index("\n## 11. Reasons")
    end = text.find("\n## ", at + 1)
    rows = text[at:end].splitlines()
    return [(r.split("`")[1], r.split("|")[2].strip()) for r in rows if r.startswith("| `")]


def need(ok, reason):
    """jwp-receipts §3.1 step 6."""
    if not ok:
        raise CV.Reject(6, reason)


def positions(hexbits):
    """§2 *Bitmap*: the positions whose bits are set, object i being bit 0x80 >> (i mod 8) of
    byte ⌊i / 8⌋."""
    bits = bytes.fromhex(hexbits)
    return [i for i in range(8 * len(bits)) if bits[i // 8] & (0x80 >> (i % 8))]


def ceil_div(a, b):
    return -(-a // b)


# ------------------------------------------------------------------ §8.3: instants

def issued_ms(t):
    """§8.3: a claim's `issued_at` as jwp-receipts §2.4.3 reads it, rounded up to the millisecond
    (1000 × its seconds + ⌈f × 1000⌉, f its written fraction), and its whole seconds."""
    s = JV.seconds(t)
    m = re.match(r"\.([0-9]+)", t[19:])
    digits = m.group(1) if m else ""
    num, den = int(digits or "0"), 10 ** len(digits)
    if drops("rec_second"):
        ms = 1000 if num else 0
    elif drops("rec_round"):
        ms = num * 1000 // den
    else:
        ms = ceil_div(num * 1000, den)
    return 1000 * s + ms, s


def ack_ms(end):
    """§8.3: an acknowledgement's window's last microsecond, `end` − 1, rounded up to the
    millisecond, and its whole seconds."""
    last = end if drops("ack_last_us") else end - 1
    return (last // 1000 if drops("ack_round") else ceil_div(last, 1000)), last // 1_000_000


def before(word_s, start, check):
    """timeliness §2's rule for a statement before its group."""
    return not drops(check) and TV.before_group(word_s, start)


# ------------------------------------------------------------------ §4.4: placing a server key

_HEADS = {}


def head(origin, text):
    """A head held for `origin` (jwp-receipts §7.2 rules 1 to 3), or None."""
    k = (origin, text)
    if k not in _HEADS:
        try:
            r = SG.parse_revision(text)
            SG.check_held(r, origin)
        except (SG.Stop, TypeError):
            r = None
        _HEADS[k] = r
    return _HEADS[k]


def later(at_ms, until):
    """§4.4: `at_ms` is later than a retired entry's `until` when `at_ms` > `until` × 1000."""
    u = SG.secs(until) * 1000
    return at_ms >= u if drops("until_inclusive") else at_ms > u


def roles_of(e):
    """The serving roles an entry's `may_sign` holds, in §4.4's order."""
    if drops("serving"):
        return list(e["may_sign"])
    return [r for r in SERVING if r in e["may_sign"]]


def place(key, at_ms, heads):
    """§4.4: the placement of the key `key` (hexadecimal) for an item at `at_ms`, by the heads
    named, keyed by origin, as §4.4 writes it."""
    origins = sorted(heads, key=lambda o: o.encode(), reverse=drops("least_origin"))
    held = [(o, head(o, heads[o])) for o in origins]
    listed = []
    for o, r in held:
        e = None if r is None else next((e for e in r["keys"] if e["key"] == key), None)
        if e is not None:
            listed.append((o, e))
    first = []
    revoked = next(((o, e) for o, e in listed if e["status"] == "revoked"), None)
    if revoked is not None and not drops("revoked"):
        first.append(f"refused reason=key_revoked origin={revoked[0]} "
                     f"compromised_since={revoked[1]['compromised_since']}")
    unheld = next((o for o, r in held if r is None), None)
    if unheld is not None and not drops("not_held"):
        first.append(f"unconfirmed reason=key_log_not_held origin={unheld}")
    if first:
        return first[-1] if drops("revoked_first") else first[0]
    for o, e in listed:
        roles = roles_of(e)
        in_force = e["status"] == "active" or (e["status"] == "retired"
                                               and (drops("until") or not later(at_ms, e["until"])))
        if roles and in_force:
            out = f"confirmed origin={o} key_id={e['id']} role={roles[0]}"
            return out + (f" until={e['until']}" if e["status"] == "retired" else "")
    retired = next(((o, e) for o, e in listed if roles_of(e) and e["status"] == "retired"), None)
    if retired is not None:
        return f"refused reason=key_retired origin={retired[0]} until={retired[1]['until']}"
    bare = next((o for o, e in listed if not roles_of(e)), None)
    if bare is not None:
        return f"refused reason=authority_not_granted origin={bare}"
    return "unconfirmed reason=server_not_listed"


# ------------------------------------------------------------------ §5.3: step 6

def check_cache_transmission(c, g, objects):
    """§5.3 against a group receipt that passed jwp-receipts §3 steps 1 and 2 (as jwp_verify reads
    it), and the object receipts held under it, of which those that pass §3 step 5 are used."""
    b = c["body"]
    need(drops("ct_subject") or JV.link(g).hex() == c["subject"], "cache_transmission_subject_mismatch")
    need(drops("ct_publisher") or c["signer_pubkey"] != g["signer_pubkey"], "cache_transmission_by_publisher")
    need(drops("ct_leaf_count") or b["leaf_count"] == g["leaf_count"], "cache_transmission_leaf_count_mismatch")
    sent = positions(b["sent"])
    if len(sent) == g["leaf_count"]:
        root = bytes.fromhex(g["root_hash"])
    else:
        leaves = []
        for i in sent:
            o = next((o for o in objects if o["object_id"] == i and BV.included(o, g)), None)
            if o is None and drops("ct_held"):
                continue
            need(o is not None, "cache_transmission_objects_not_held")
            leaves.append(JV.leaf(o))
        root = BV.merkle_root(leaves)
    need(drops("ct_root") or root.hex() == b["sent_root"], "cache_transmission_root_mismatch")


def read_objects(receipts):
    """The object receipts given that read (jwp-receipts §2.4.2)."""
    out = []
    for o in receipts:
        try:
            out.append(JV.read_object(o))
        except JV.Refuse:
            pass
    return out


def context_group(ctx):
    """§5.3: the group receipt a claim is checked against, after jwp-receipts §3 steps 1 and 2."""
    if not (isinstance(ctx, dict) and ctx.get("kind") == "group") and not drops("context_kind"):
        raise CV.Reject(6, "context_mismatch")
    try:
        g = JV.read_group(ctx)
        JV.verify_group(g)
    except JV.Refuse:
        raise CV.Reject(6, "context_invalid")
    return g


def claim_outcome(case):
    """A claim through jwp-receipts §3.1 steps 1 to 6 against the context the case gives, its
    signer's previous claim the last of the earlier ones held."""
    try:
        held = {}
        for wire in case["earlier"]:
            try:
                e = TV.claim_steps(wire)
                held[e["signer_pubkey"]] = e
            except CV.Reject:
                pass
        c = TV.claim_steps(case["claim"])
        CV.check_chain(c, held.get(c["signer_pubkey"]))
        if case["context"] is None and not drops("unchecked"):
            return "unchecked"
        if c["class"] != "cache-transmission":
            raise CV.Reject(6, "context_mismatch")
        check_cache_transmission(c, context_group(case["context"]), read_objects(case["objects"]))
        return "accept"
    except CV.Reject as r:
        return f"reject step={r.step} reason={r.reason}"


# ------------------------------------------------------------------ §7: the verdict

class Verdict:
    """§7.2 over a case's inputs, with what §7.3, §8 and §9 read from it."""

    def __init__(self, case, count_only):
        self.g = TV.read_groups([{"group": case["group"], "objects": case["objects"]}])[0]
        gr = self.g.receipt
        objs = [ob for ob, _ in self.g.pairs]
        # Step 2: the claims.
        labels, accepted, held = [], [], {}
        for wire in case["claims"]:
            try:
                c = TV.claim_steps(wire)
            except CV.Reject as r:
                labels.append(f"reject step={r.step} reason={r.reason}")
                accepted.append(None)
                continue
            previous = held.get(c["signer_pubkey"])
            held[c["signer_pubkey"]] = c
            try:
                CV.check_chain(c, previous)
                if c["class"] == "transmission":
                    BV.check_transmission(c, gr)
                elif c["class"] == "reconciliation":
                    BV.check_reconciliation(c, gr, objs)
                elif c["class"] == "cache-transmission":
                    check_cache_transmission(c, gr, objs)
                else:
                    raise CV.Reject(6, "context_mismatch")
            except CV.Reject as r:
                labels.append(f"reject step={r.step} reason={r.reason}")
                accepted.append(None)
                continue
            labels.append(None)
            accepted.append(c)

        def by_instance(cls):
            out = {}
            for i, c in enumerate(accepted):
                if c is not None and c["class"] == cls:
                    out.setdefault(c["body"]["bearer_id"], []).append(i)
            return out

        # Step 3: the publisher's declarations (bearer-receipts §6.2 step 4).
        self.pub, self.pub_aside = {}, set()
        for bid, idx in by_instance("transmission").items():
            bodies = [accepted[i]["body"] for i in idx]
            stands = all(b == bodies[0] for b in bodies)
            if stands:
                self.pub[bid] = bodies[0]
            else:
                self.pub_aside.add(bid)
            for i in idx:
                labels[i] = "declared" if stands else "set-aside reason=bearer_contradicted"
        # Step 4: the caches' instances.
        self.caches, self.cache_aside = {}, set()
        for bid, idx in by_instance("cache-transmission").items():
            cs = [accepted[i] for i in idx]
            taken = (bid in self.pub or bid in self.pub_aside) and not drops("taken")
            same = drops("cache_contradiction") or all(
                (x["signer_pubkey"], x["body"]) == (cs[0]["signer_pubkey"], cs[0]["body"]) for x in cs)
            if drops("same_claim"):
                same = same and all(x["issued_at"] == cs[0]["issued_at"] for x in cs)
            if not taken and not same:
                self.cache_aside.add(bid)
            if taken or not same:
                for i in idx:
                    labels[i] = "set-aside reason=bearer_contradicted"
                continue
            self.caches[bid] = {"claims": cs, "body": cs[0]["body"], "server": cs[0]["signer_pubkey"],
                                "sent": positions(cs[0]["body"]["sent"])}
            for i in idx:
                labels[i] = "cache-declared"
        # Step 5: the reconciliations.
        counted, self.bits, self.instance_bits, self.counted = set(), {}, {}, []
        for i, c in enumerate(accepted):
            if c is None or c["class"] != "reconciliation":
                continue
            b, key = c["body"], c["signer_pubkey"]
            bid = b["bearer_id"]
            d, ci = self.pub.get(bid), self.caches.get(bid)
            stated = d if d is not None else None if ci is None else ci["body"]
            if bid in self.pub_aside or bid in self.cache_aside:
                aside = "bearer_contradicted"
            elif stated is None:
                aside = "bearer_not_declared"
            elif stated["bearer"] != b["bearer"]:
                aside = "bearer_contradicted"
            elif d is None and key == ci["server"] and not drops("self_ack"):
                aside = "self_acknowledged"
            elif count_only is not None and key not in count_only:
                aside = "sink_not_confirmed"
            else:
                aside = None
            if aside:
                labels[i] = "set-aside reason=" + aside
                continue
            labels[i] = "joined reason=reconciliation_duplicate_sink" if (key, bid) in counted else "counted"
            counted.add((key, bid))
            got = positions(b["received"])
            self.bits.setdefault(key, set()).update(got)
            self.instance_bits.setdefault(bid, {}).setdefault(key, set()).update(got)
            self.counted.append(i)
        self.labels, self.accepted = labels, accepted

    def outcome(self, case):
        """§7.2 step 6, §7.3 and §7.4, as the corpus states them."""
        g, heads = self.g, case["heads"]
        n = g.count
        sinks = len(self.bits)
        whole = sum(1 for s in self.bits.values() if len(s) == n)
        received_by = [sum(1 for s in self.bits.values() if j in s) for j in range(n)]

        def energy(e, k):
            if e is None:
                return None, None
            j = e["joules_micro"]
            share = j // max(1, k) if drops("energy_ceiling") else ceil_div(j, max(1, k))
            return {"joules_micro": j, "class": e["energy_provenance"]}, share

        bearers = []
        for bid in sorted(self.pub, key=lambda s: s.encode()):
            d = self.pub[bid]
            k = len(self.instance_bits.get(bid, {}))
            e, share = energy(d["energy"], k)
            bearers.append({"bearer_id": bid, "bearer": d["bearer"], "sinks": k, "energy": e,
                            "per_receiver_upper_bound": share})
        caches = []
        for bid in sorted(self.caches, key=lambda s: s.encode()):
            ci = self.caches[bid]
            mine = self.instance_bits.get(bid, {})
            if drops("per_instance"):
                mine = {k: self.bits[k] for k in mine}
            at = (min if drops("cache_earliest") else max)(issued_ms(x["issued_at"])[0] for x in ci["claims"])
            if drops("cache_instant"):
                at = g.start * 1000
            e, share = energy(ci["body"]["energy"], len(mine))
            sent = list(range(n)) if drops("sent_only") else ci["sent"]
            caches.append({"bearer_id": bid, "bearer": ci["body"]["bearer"], "server": ci["server"],
                           "placement": place(ci["server"], at, heads), "sent": sent, "sinks": len(mine),
                           "received_by": [[j, sum(1 for s in mine.values() if j in s)] for j in sent],
                           "energy": e, "per_receiver_upper_bound": share})
        transmitted = bool(self.pub) or (drops("cache_status") and bool(self.caches))
        status = "reconciled" if sinks else "transmitted" if transmitted else "published"
        sink = None
        s = case["sink"]
        if s is not None:
            sid, sb = s["bearer_id"], s["bearer"]
            if sid in self.pub_aside or sid in self.cache_aside:
                sink = "bearer_contradicted"
            elif sid in self.pub:
                sink = "declared" if self.pub[sid]["bearer"] == sb else "bearer_contradicted"
            elif sid in self.caches and not drops("sink_cache"):
                sink = "cache-declared" if self.caches[sid]["body"]["bearer"] == sb else "bearer_contradicted"
            else:
                sink = "bearer_not_declared"
        assertion = None
        a = case["assertion"]
        if a is not None:
            if "transmitted" in a:
                over = [d["bearer"] for d in self.pub.values()]
                if drops("statement_publisher"):
                    over += [ci["body"]["bearer"] for ci in self.caches.values()]
                assertion = "holds" if a["transmitted"] in over else "reject reason=bearer_not_declared"
            elif whole == 0:
                assertion = "reject reason=delivery_not_reconciled"
            elif whole < a["delivered"]:
                assertion = "reject reason=delivery_exceeds_reconciled"
            else:
                assertion = "holds"
        return {"objects": g.verdicts, "claims": self.labels, "status": status, "sinks": sinks, "whole": whole,
                "received_by": received_by, "bearers": bearers, "caches": caches, "sink": sink,
                "assertion": assertion}


def count_only_of(case):
    return None if case["count_only"] is None else set(case["count_only"])


def verdict(case):
    try:
        v = Verdict(case, count_only_of(case))
    except JV.Refuse as r:
        return "refuse reason=" + r.reason
    return v.outcome(case)


# ------------------------------------------------------------------ §8: held by a deadline

def population_valid(region, keys):
    """§8.1: a region code of rung-price §2, and 1 to 2^20 keys in strictly increasing order of
    their bytes."""
    if not drops("region") and not RP.region_code(region):
        return False
    if not isinstance(keys, list) or len(keys) > MAX_KEYS or (not keys and not drops("population_empty")):
        return False
    if not all(CV.is_hex(k, 32) for k in keys):
        return False
    if drops("population_order"):
        return True
    raw = [bytes.fromhex(k) for k in keys]
    return all(a < b or (a == b and drops("population_strict")) for a, b in zip(raw, raw[1:]))


def population_digest(region, keys):
    """§8.1: SHA-256 over the domain, the region's length and the region, n and the keys."""
    r = region.encode("ascii")
    head = POPULATION + (b"" if drops("digest_region") else bytes([len(r)]) + r)
    return hashlib.sha256(head + len(keys).to_bytes(4, "big") + b"".join(bytes.fromhex(k) for k in keys)).hexdigest()


def verified(g, i):
    """§8.2 step 5: the content hash and length of object i held `delivered-verified`, or None."""
    memo = g.__dict__.setdefault("ca_verified", {})
    if i not in memo:
        memo[i] = None
        for ob, env in g.pairs:
            if env is not None and ob["object_id"] == i:
                try:
                    JV.accept(ob, g.receipt, env)
                    memo[i] = (ob["content_hash"].hex(), len(env))
                    break
                except JV.Refuse:
                    pass
    return memo[i]


def deadline_entry(wire, g, i):
    """§8.2 step 6: the entry for object i of a timeliness claim that stands against the group, with
    no claim of its signer held before it, and the claim's link hash; or None."""
    try:
        c = TV.claim_steps(wire)
        CV.check_chain(c, None)
        if c["class"] != "timeliness":
            raise CV.Reject(6, "context_mismatch")
        held = g
        if drops("deadline_envelopes"):
            held = copy.copy(g)
            held.envelopes = {}
        TV.check_timeliness(c, held, {g.link: held})
    except CV.Reject:
        return None
    entries = c["body"]["objects"]
    e = next((e for e in entries if e["leaf_index"] == i), None)
    if e is None and drops("deadline_entry"):
        e = entries[0]
    return None if e is None else (e, CV.link_hash(c))


_ACKS = {}


def read_ack(x):
    """settlement §3.6 for a record given, and its item lists against its entries: (reason or None,
    record, items, the item lists' reason or None)."""
    k = (x["record"], tuple(x["items"]))
    if k not in _ACKS:
        try:
            a = SET.read_ack(bytes.fromhex(x["record"]))
        except SET.Refuse as r:
            _ACKS[k] = (r.reason, None, [], None)
            return _ACKS[k]
        items, why = [], None
        try:
            for j, e in enumerate(a["entries"]):
                if j >= len(x["items"]):
                    raise SET.Refuse("record_missing")
                raw = bytes.fromhex(x["items"][j])
                its = SET.read_items(raw)
                if not SET.recomputes(e, its, raw):
                    raise SET.Refuse("items_mismatch")
                items += its
        except SET.Refuse as r:
            why, items = r.reason, []
        _ACKS[k] = (None, a, items, why)
    return _ACKS[k]


def acknowledgements(acks, g):
    """§8.3: each record's label, in order, with the record and its items where it reads."""
    out = []
    for x in acks:
        why, a, items, items_why = read_ack(x)
        if why is not None:
            out.append(("reject reason=" + why, None, []))
        elif a["mode"] == DECLARATION and not drops("declaration"):
            out.append(("declaration", a, []))
        elif items_why is not None:
            out.append(("reject reason=" + items_why, None, []))
        elif before(ack_ms(a["end"])[1], g.start, "ack_before"):
            out.append(("before-group", a, items))
        else:
            out.append(("read", a, items))
    return out


def covers(spans, length):
    """Whether the union of the byte ranges `spans` covers [0, length)."""
    if drops("covers"):
        return bool(spans)
    at = 0
    for s, e in sorted(spans):
        if at >= length:
            break
        if s > at:
            return False
        at = max(at, e)
    return at >= length


def held_at(v, acks, i, h, length, keys):
    """§8.3's held-at statements of acknowledgements and counted reconciliations for object i (its
    content hash h and length), for the keys `keys` (None for every key): {key: [(instant,
    servers)]}, and the statements set aside before the group, as (key, source)."""
    g = v.g
    out, aside = {}, []

    def wanted(k):
        return keys is None or k in keys or drops("population_only")

    recs = v.counted
    if drops("line_counted"):
        recs = [x for x, c in enumerate(v.accepted) if c is not None and c["class"] == "reconciliation"]
    for x in recs:
        c = v.accepted[x]
        k, b = c["signer_pubkey"], c["body"]
        if not wanted(k) or i not in positions(b["received"]):
            continue
        ms, s = issued_ms(c["issued_at"])
        if before(ms // 1000 if drops("rec_before_ms") else s, g.start, "rec_before"):
            aside.append((k, "reconciliation"))
            continue
        bid, server = b["bearer_id"], None
        if bid in v.pub:
            server = g.signer
        elif bid in v.caches and (i in v.caches[bid]["sent"] or drops("sent_server")):
            server = v.caches[bid]["server"]
        out.setdefault(k, []).append((ms, set() if server is None else {server}))
    readable = [(a, items) for label, a, items in acks if label in ("read", "before-group")
                or (drops("declaration") and label == "declaration")]
    for k in sorted({a["downstream"].hex() for a, _ in readable}):
        if not wanted(k):
            continue
        mine = [(a, items) for a, items in readable
                if a["downstream"].hex() == k and any(c.hex() == h for c, _, _ in items)]
        rest = []
        for a, items in mine:
            if before(ack_ms(a["end"])[1], g.start, "ack_before"):
                aside.append((k, a["digest"].hex()))
            else:
                rest.append((a, items))
        if drops("ack_digest"):
            rest.sort(key=lambda p: p[0]["end"])
        elif not drops("ack_order"):
            rest.sort(key=lambda p: (p[0]["end"], p[0]["digest"]))
        spans = []
        for n, (a, items) in enumerate(rest):
            if drops("union"):
                spans = []
            spans += [(s, e) for c, s, e in items if c.hex() == h]
            if covers(spans, length):
                upto = rest[n:n + 1] if drops("servers_all") else rest[:n + 1]
                out.setdefault(k, []).append((ack_ms(a["end"])[0], {p[0]["upstream"].hex() for p in upto}))
                break
    return out, aside


def tally(case):
    """§8: the tally, or `refuse reason=R` for §8.2."""
    region, keys, i = case["region"], case["population"], case["index"]
    if not population_valid(region, keys):
        return "refuse reason=population_invalid"
    stated, target = case["deadline_ms"], case["miss_target_ppm"]
    if case["deadline_claim"] is None and not (CV.is_int(stated)
                                               and (stated <= LAST_INSTANT or drops("deadline_bound"))):
        return "refuse reason=target_invalid"
    if not (CV.is_int(target) and (target <= MAX_TARGET or drops("target_bound"))):
        return "refuse reason=target_invalid"
    try:
        v = Verdict(case, None if drops("count_only") else set(keys))
    except JV.Refuse as r:
        return "refuse reason=" + r.reason
    g = v.g
    if g.start is None and not drops("group_start"):
        return "refuse reason=context_invalid"
    if not (CV.is_int(i) and (i < g.count or drops("out_of_range"))):
        return "refuse reason=object_out_of_range"
    obj = verified(g, i)
    if obj is None:
        if not drops("verified"):
            return "refuse reason=object_not_verified"
        obj = (g.included.get(i, ""), 0)
    h, length = obj
    deadline, link = stated, None
    if case["deadline_claim"] is not None:
        got = deadline_entry(case["deadline_claim"], g, i)
        if got is None:
            return "refuse reason=deadline_not_stated"
        deadline, link = got[0]["by_ms"], got[1]
    acks = acknowledgements(case["acknowledgements"], g)
    statements, aside = held_at(v, acks, i, h, length, set(keys))
    rlabels, kept = TV.records_step(case["records"], [g])
    standing = [kept[x] for x in sorted(kept) if rlabels[x] == "stands"]
    if not drops("records"):
        for holder, _, _, at in TV.held_from_records(standing, g.link, i):
            if holder in keys:
                statements.setdefault(holder, []).append((at, set()))
    # §8.3: a hit has a held-at statement at or before the deadline; §8.4: its servers.
    counted_keys = sorted(statements) if drops("population_only") else keys
    hits = {}
    for k in counted_keys:
        on = [(t, s) for t, s in statements.get(k, []) if (t < deadline if drops("hit_inclusive") else t <= deadline)]
        if on:
            hits[k] = (set().union(*(s for _, s in on)), any(not s for _, s in on), on)
    n = len(keys)
    misses = n - len(hits)
    miss_ppm = misses * PPM // n if drops("miss_ceil") else ceil_div(misses * PPM, n)
    servers = []
    for s in sorted(set().union(set(), *(S for S, _, _ in hits.values())), key=bytes.fromhex):
        crediting = [(k, S, U, on) for k, (S, U, on) in hits.items() if s in S]
        at = (min if drops("server_earliest") else max)(t for _, _, _, on in crediting for t, ss in on if s in ss)
        if drops("server_instant"):
            at = g.start * 1000
        placement = "publisher" if s == g.signer and not drops("publisher") else place(s, at, case["heads"])
        sole = sum(1 for _, S, U, _ in crediting if S == {s} and (drops("sole_unattributed") or not U))
        servers.append({"key": s, "placement": placement, "hits": len(crediting), "sole": sole})
    met = miss_ppm < target if drops("target_inclusive") else miss_ppm <= target
    return {"outcome": "met" if met else "refuse reason=miss_target_exceeded", "region": region,
            "population_digest": population_digest(region, keys), "deadline_ms": deadline, "deadline_claim": link,
            "n": n, "hits": len(hits), "misses": misses, "miss_ppm": miss_ppm,
            "missing": [k for k in keys if k not in hits],
            "set_aside": [{"key": k, "source": src, "reason": "held_before_group"} for k, src in sorted(aside)],
            "servers": servers, "unattributed": sum(1 for S, _, _ in hits.values() if not S),
            "acknowledgements": [label for label, _, _ in acks], "records": rlabels, "claims": v.labels}


# ------------------------------------------------------------------ §9: timeliness lines

def lines(case):
    """§9: timeliness §6.3's lines for object `index`. Records give held-at statements as
    timeliness §6 reads them, for the primary and its fallback whatever envelopes are held, with a
    line for each holder with a record that stands and has an entry for either group;
    acknowledgements and counted reconciliations give them for each object of the group held
    `delivered-verified`, the primary and a fallback in its group among them, with a line for each
    key with one."""
    try:
        v = Verdict(case, count_only_of(case))
    except JV.Refuse as r:
        return "refuse reason=" + r.reason
    g, i = v.g, case["index"]
    got = deadline_entry(case["claim"], g, i)
    if got is None:
        return "reject reason=deadline_not_stated"
    entry = got[0]
    fb = entry["fallback"]
    acks = acknowledgements(case["acknowledgements"], g)
    rlabels, kept = TV.records_step(case["records"], [g])
    standing = [kept[x] for x in sorted(kept) if rlabels[x] == "stands"]
    holders = TV.line_holders(g.link, entry, standing)
    if drops("line_fallback_holders"):
        holders = [h for h in holders if any(TV.has_entry(r, g.link) for r in standing if r[0].hex() == h)]
    primary = TV.held_from_records(standing, g.link, i)
    if drops("line_records_verified") and verified(g, i) is None:
        primary = []
    fallback = [] if fb is None else TV.held_from_records(standing, fb["group"], fb["leaf_index"])
    for j in range(g.count):
        obj = verified(g, j)
        if obj is None:
            continue
        statements, _ = held_at(v, acks, j, obj[0], obj[1], None)
        for k, st in statements.items():
            if k not in holders and not drops("line_keys"):
                holders.append(k)
            if j == i:
                primary += [(k, g.link, i, t) for t, _ in st]
            if fb is not None and fb["group"] == g.link and fb["leaf_index"] == j and not drops("line_fallback_acks"):
                fallback += [(k, g.link, j, t) for t, _ in st]
    return TV.lines(entry, holders, primary, fallback)


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

def placement_outcome(case):
    return place(case["key"], case["at_ms"], case["heads"])


RUNS = (("claims", claim_outcome), ("placements", placement_outcome), ("verdicts", verdict), ("tallies", tally),
        ("lines", lines))
KINDS = tuple(k for k, _ in RUNS)


def check(doc, quiet=False):
    """The failures of a corpus document: each case's outcome, and every reason of §11."""
    fails = []
    for kind, run in RUNS:
        for c in doc[kind]:
            try:
                got = run(c)
            except CV.Reject as r:
                got = {"error": f"reject step={r.step} reason={r.reason}"}
            except (KeyError, TypeError, ValueError, IndexError, AttributeError, StopIteration, ZeroDivisionError,
                    JV.Refuse, SET.Refuse, SG.Stop, HV.Refused) as e:
                got = {"error": repr(e)}
            if got != c["outcome"]:
                fails.append(c["name"])
                if not quiet:
                    print(f"FAIL {kind} {c['name']}\n  stated {json.dumps(c['outcome'])}\n  got    {json.dumps(got)}")
    stated = []
    for kind in KINDS:
        for c in doc[kind]:
            TV.strings(c["outcome"], stated)
    for code, _ in reasons_table():
        if not any(s == code or re.search(r"reason=" + re.escape(code) + r"(\s|$)", s) for s in stated):
            fails.append("no case gives " + code)
    return fails


def mutations(doc):
    """Copies of the corpus, one defect each, every one of which must fail."""
    def case(d, name, kinds=KINDS):
        return next(c for k in kinds for c in d[k] if c["name"] == name)

    def m_until(d):
        case(d, "retired_at_until")["at_ms"] += 1

    def m_least(d):
        c = case(d, "confirmed_least_origin")
        c["outcome"] = c["outcome"].replace("operator.example", "other.example")

    def m_receipt(d):
        # Object 3's receipt held too: the claim's root over objects 1 and 3 then verifies.
        c = case(d, "objects_not_held")
        c["objects"].append(case(d, "sent_partial")["objects"][3])

    def m_credit(d):
        case(d, "per_instance_credit")["outcome"]["caches"][0]["received_by"][1][1] = 1

    def m_status(d):
        case(d, "cache_declared_only")["outcome"]["status"] = "transmitted"

    def m_head(d):
        case(d, "head_not_held", ("verdicts",))["heads"]["https://operator.example"] = d["heads"]["operator"]

    def m_count_only(d):
        case(d, "confirmed_keys_only")["count_only"] = None

    def m_bound(d):
        case(d, "energy_bound_after")["outcome"]["caches"][0]["per_receiver_upper_bound"] = 3_333_333

    def m_population(d):
        c = case(d, "met_by_bearer_and_peers")
        c["population"] = list(reversed(c["population"]))

    def m_region(d):
        case(d, "region_country")["region"] = "XB"

    def m_deadline(d):
        case(d, "ack_last_us_at_deadline")["deadline_ms"] -= 1

    def m_ppm(d):
        case(d, "rounded_up")["outcome"]["miss_ppm"] = 333_333

    def m_sole(d):
        case(d, "two_peers_complete")["outcome"]["servers"][0]["sole"] = 1

    def m_items(d):
        case(d, "ack_any_class")["acknowledgements"][0]["items"] = []

    def m_claim_link(d):
        case(d, "deadline_from_timeliness")["outcome"]["deadline_claim"] = "00" * 32

    def m_line(d):
        case(d, "timeliness_line_from_ack")["outcome"][0]["held_ms"] += 1

    def m_reason_gone(d):
        d["claims"].remove(case(d, "leaf_count_wrong"))

    return [m_until, m_least, m_receipt, m_credit, m_status, m_head, m_count_only, m_bound, m_population, m_region,
            m_deadline, m_ppm, m_sole, m_items, m_claim_link, m_line, m_reason_gone]


WEAKENINGS = ["rec_second", "rec_round", "ack_last_us", "ack_round", "until_inclusive", "serving", "least_origin",
              "revoked", "not_held", "revoked_first", "until", "ct_subject", "ct_publisher", "ct_leaf_count",
              "ct_held", "ct_root", "context_kind", "unchecked", "taken", "cache_contradiction", "same_claim",
              "self_ack", "energy_ceiling", "per_instance", "cache_instant", "sent_only", "cache_status",
              "sink_cache", "statement_publisher", "region", "population_empty", "population_order",
              "population_strict", "digest_region", "deadline_entry", "declaration", "ack_before", "rec_before",
              "covers", "union", "servers_all", "sent_server", "population_only", "line_counted", "deadline_bound",
              "target_bound", "count_only", "group_start", "out_of_range", "verified", "records", "hit_inclusive",
              "miss_ceil", "server_instant", "publisher", "target_inclusive", "line_keys", "deadline_envelopes",
              "ack_order", "ack_digest", "server_earliest", "cache_earliest", "rec_before_ms", "line_fallback_acks",
              "line_records_verified", "line_fallback_holders"]


def direct_checks():
    """What no vector carries, and what this text states of the texts beside it."""
    fails = []
    # §8.1's bound: 2^20 keys are a population, and one more is not.
    keys = [k.to_bytes(32, "big").hex() for k in range(1, MAX_KEYS + 2)]
    if population_valid("XA", keys):
        fails.append("a population of 2^20 + 1 keys is accepted")
    keys.pop()
    if not population_valid("XA", keys):
        fails.append("a population of 2^20 keys is refused")
    del keys
    # §8.3: an acknowledgement's hit is `end` − 1 ≤ deadline_ms × 1000 in 64 bits, its rounded
    # instant at or before the deadline the same test, at the last instant and the widest `end`.
    if LAST_INSTANT * 1000 >= 1 << 64:
        fails.append("the last instant in microseconds does not fit in 64 bits")
    for end in (LAST_INSTANT * 1000 + 1, LAST_INSTANT * 1000 + 2, (1 << 64) - 1):
        if (ack_ms(end)[0] <= LAST_INSTANT) != (end - 1 <= LAST_INSTANT * 1000):
            fails.append(f"an acknowledgement ending at {end} reads two ways")
    # jwp-receipts §3.1 lists §5.2's codes at step 4 and §5.3's at step 6; its §2.4.4 row admits a
    # receipt subject, as the claim verifier does.
    jwp = TV.text_of(JWP_RECEIPTS)
    for code, where in reasons_table():
        step = {"§5.2": 4, "§5.3": 6}.get(where.split(" ")[0])
        if step is not None and f"| {step} | `{code}` |" not in jwp:
            fails.append(f"jwp-receipts §3.1 does not list {code} at step {step}")
    if "| `cache-transmission` | `claim` | `receipt`" not in jwp or CV.ADMITS.get("cache-transmission") != {"receipt"}:
        fails.append("jwp-receipts §2.4.4 and the claim verifier differ on the class's subject")
    # §4.4: the serving roles after `cache/serve` are those settlement §13 places.
    text = TV.text_of(SETTLEMENT)
    s13 = text[text.index("\n## 13."):]
    for role in SERVING[1:]:
        if f"`{role}`" not in s13:
            fails.append(f"settlement §13 does not place {role}")
    return fails


def main(argv):
    global DROPPED
    if len(argv) != 2:
        print(__doc__)
        return 2
    doc = json.load(open(os.path.join(argv[1], "vectors.json"), encoding="utf-8"))
    extra = sorted(set(os.listdir(argv[1])) - {"vectors.json", "README.md"})
    fails = check(doc)
    if extra:
        fails.append(f"files the corpus does not hold: {extra}")
    fails += direct_checks()
    print(f"{sum(len(doc[k]) for k in KINDS)} cases ({', '.join(f'{len(doc[k])} {k}' for k in KINDS)}): "
          f"{len(fails)} failure(s)")
    caught = 0
    for m in mutations(doc):
        d = copy.deepcopy(doc)
        m(d)
        if check(d, quiet=True):
            caught += 1
        else:
            fails.append(f"mutation {m.__name__} was not caught")
    print(f"{caught} of {len(mutations(doc))} mutated copies fail")
    weak = 0
    for w in WEAKENINGS:
        DROPPED = w
        if check(doc, quiet=True):
            weak += 1
        else:
            fails.append(f"a verifier without its {w} check passes the corpus")
    DROPPED = None
    print(f"{weak} of {len(WEAKENINGS)} weakened verifiers are caught")
    for f in fails:
        print("FAIL", f)
    return 1 if fails else 0


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