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

#!/usr/bin/env python3
"""takedown_verify.py — the second implementation of extensions/takedown.md (wai.media.takedown).

Written from the extension's text, sharing no code with wai-rs. It reads every record of §5
under §2's encodings, computes the derivation closure of §6 from the derivation records (read
with the deriving-edge corpus's second implementation, edge-conformance/edge_verify.py), the
origin claims (read with the synthesis corpus's, live-conformance/synthesis_verify.py), the
delivery receipts of the groups it reaches (read with edge_verify.py too), the
acknowledgements that show who served or received an object (read with the settlement corpus's,
live-conformance/settlement_verify.py) and the cache-transmission claims that show what a node sent
from what it held (read through jwp-receipts §3.1 steps 1 to 4 with the claim corpus's,
claim-conformance/claim_verify.py), holds the takedown's
stated set to it, finds the holders and the purges that cover them (§7), computes the time to
removal against the clock the registry gives (§4, §7.3), and recomputes the counts (§9). Every
signature is checked with WAI's strict Ed25519 rule (claim-conformance/_ed25519.py).

Then it requires every mutated copy of the corpus to fail, and, with each of its own checks
dropped in turn, requires the corpus to catch the weakened verifier.

    python3 -I takedown_verify.py <live-conformance/takedown>

The standard library and the `cryptography` package, which the claim verifier imports. Exit
status 0 when everything holds, 1 otherwise.
"""
import calendar
import copy
import json
import os
import re
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance"))
sys.path.insert(0, os.path.join(HERE, "..", "edge-conformance"))
import _ed25519 as ED  # noqa: E402
from _blake3 import blake3  # noqa: E402
import edge_verify as EV  # noqa: E402
import synthesis_verify as SYV  # noqa: E402
import settlement_verify as SEV  # noqa: E402
import claim_verify as CV  # noqa: E402

EXTENSION = os.path.join(HERE, "..", "extensions", "takedown.md")
PUBLISHED = os.path.join(HERE, "..", "registry", "takedown-clocks.json")

SIGN = b"wai:takedown\x01"        # §2
LINK = b"wai:takedown-link\x01"   # §2
MAX_INT = (1 << 53) - 1
ROLES = ("origin", "store", "relay", "cache", "edge")

CHECKS = (
    "registry-shape", "signature", "registry-link", "policy-clocks", "takedown-policy", "takedown-signer",
    "notice-held", "notice-targets", "clock-in-policy", "clock-consistent", "records", "claims", "acknowledgements",
    "cache-transmissions",
    "purge-takedown", "rule-group", "rule-derivation", "rule-claim", "group-expanded",
    "exact-supported", "match-not-exact", "exact-complete", "match-held", "match-anchored",
    "store-holders", "deriver-holders", "claimant-holders", "publisher-holders", "served-holders", "sent-holders",
    "holder-only", "group-coverage", "deadline", "purges-complete", "window",
    "counts-as-of", "counts-rows",
)
DROPPED = set()


def on(check):
    assert check in CHECKS, check
    return check not in DROPPED


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


class Bad(Exception):
    """A member that breaks its rule: the record is record_malformed."""


def need(cond):
    if not cond:
        raise Bad()


# ---------------------------------------------------------------- §2 encodings

def is_hex(v, n):
    return isinstance(v, str) and re.fullmatch("[0-9a-f]{%d}" % (2 * n), v) is not None


def is_int(v):
    return isinstance(v, int) and not isinstance(v, bool) and 0 <= v <= MAX_INT


def text(v, lo, hi):
    need(isinstance(v, str) and lo <= len(v) <= hi)
    return v


def days_in(y, m):
    if m == 2:
        return 29 if y % 4 == 0 and (y % 100 != 0 or y % 400 == 0) else 28
    return 30 if m in (4, 6, 9, 11) else 31


def date(s):
    m = isinstance(s, str) and re.fullmatch(r"([0-9]{4})-([0-9]{2})-([0-9]{2})", s)
    need(bool(m))
    y, mo, d = (int(m.group(i)) for i in (1, 2, 3))
    need(1 <= mo <= 12 and 1 <= d <= days_in(y, mo))
    return y, mo, d


def time_value(s):
    """§2: `YYYY-MM-DDTHH:MM:SSZ`, a year of 1970 or later, a second of at most 59: its value."""
    m = isinstance(s, str) and re.fullmatch(r"([0-9]{4}-[0-9]{2}-[0-9]{2})T([0-9]{2}):([0-9]{2}):([0-9]{2})Z", s)
    need(bool(m))
    y, mo, d = date(m.group(1))
    hh, mi, ss = int(m.group(2)), int(m.group(3)), int(m.group(4))
    need(y >= 1970 and hh <= 23 and mi <= 59 and ss <= 59)
    return calendar.timegm((y, mo, d, hh, mi, ss, 0, 0, 0))


def clock_id(s):
    need(isinstance(s, str) and 1 <= len(s.encode()) <= 64 and re.fullmatch(r"[a-z0-9]+(-[a-z0-9]+)*", s) is not None)
    return s


_SIGS = {}


def ed_verify(pk, msg, sig):
    """WAI's strict Ed25519 rule (_ed25519.verify), remembered: the mutations and the dropped
    checks verify the corpus many times over, and a verdict is a function of its three inputs."""
    k = (pk, msg, sig)
    if k not in _SIGS:
        _SIGS[k] = ED.verify(pk, msg, sig)
    return _SIGS[k]


def link_of(v):
    return blake3(LINK + EV.jcs(v).encode("utf-8"))


def members(v, names):
    need(isinstance(v, dict) and set(v) == set(names))
    return v


# ---------------------------------------------------------------- §3 subjects

def subject(v):
    need(isinstance(v, dict))
    t = v.get("type")
    if t == "object":
        members(v, ["type", "content_hash"])
        need(is_hex(v["content_hash"], 32))
    elif t == "group":
        members(v, ["type", "receipt"])
        need(is_hex(v["receipt"], 32))
    elif t == "track":
        members(v, ["type", "track", "from_group"])
        need(is_hex(v["track"], 32) and is_int(v["from_group"]))
    elif t == "take":
        members(v, ["type", "take"])
        need(is_hex(v["take"], 32))
    else:
        raise Bad()
    return v


def key(s):
    """A subject's JCS form: lists of subjects are in its byte order (§3)."""
    return EV.jcs(s)


def subject_list(v, lo, hi):
    need(isinstance(v, list) and lo <= len(v) <= hi)
    out = [subject(x) for x in v]
    ks = [key(x).encode() for x in out]
    need(all(a < b for a, b in zip(ks, ks[1:])))
    return out


def obj(h):
    return {"type": "object", "content_hash": h}


# ---------------------------------------------------------------- records

def strict(text_):
    v = EV.strict(text_)
    need(EV.in_subset(v))
    return v


def read_registry(t):
    try:
        v = strict(t)
        members(v, ["kind", "clocks"])
        need(v["kind"] == "wai-takedown-clocks")
        cs = v["clocks"]
        need(isinstance(cs, list) and 1 <= len(cs) <= 64)
        clocks = {}
        ids = []
        for c in cs:
            members(c, ["id", "type", "duration_s", "starts", "instrument", "section", "applies_from"])
            clock_id(c["id"])
            need(c["type"] in ("deadline", "reply_window", "unfixed"))
            d = c["duration_s"]
            need(d is None or (is_int(d) and d >= 1))
            need((c["type"] == "unfixed") == (d is None))
            need(c["starts"] in ("receipt", "statement"))
            text(c["instrument"], 1, 500)
            text(c["section"], 1, 200)
            date(c["applies_from"])
            clocks[c["id"]] = c
            ids.append(c["id"].encode())
        need(all(a < b for a, b in zip(ids, ids[1:])))
    except (Bad, ValueError, KeyError, TypeError, AttributeError):
        if on("registry-shape"):
            raise Refuse("record_malformed", ["registry"])
        raise Refuse("record_malformed", ["registry", "unread"]) from None
    return {"link": link_of(v), "clocks": clocks}


def read_signed(t, kind, names, fields, at):
    """§2: the shape, the kind's own members, then the signature."""
    try:
        v = strict(t)
        members(v, names)
        need(v["kind"] == kind and is_hex(v["signer_pubkey"], 32) and is_hex(v["sig"], 64))
        r = fields(v)
    except (Bad, ValueError, KeyError, TypeError, AttributeError):
        raise Refuse("record_malformed", at)
    bad = ed_verify(bytes.fromhex(v["signer_pubkey"]), SIGN + EV.jcs({k: x for k, x in v.items() if k != "sig"}).encode("utf-8"),
                    bytes.fromhex(v["sig"]))
    if bad and on("signature"):
        raise Refuse(bad, at)
    r.update(link=link_of(v), signer=v["signer_pubkey"])
    return r


def policy_fields(v):
    u = v["use"]
    need(isinstance(u, list) and len(u) <= 64)
    for x in u:
        clock_id(x)
    need(all(a.encode() < b.encode() for a, b in zip(u, u[1:])))
    st = v["stores"]
    need(isinstance(st, list) and 1 <= len(st) <= 256)
    keys = []
    for s in st:
        members(s, ["key", "role"])
        need(is_hex(s["key"], 32) and s["role"] in ROLES)
        keys.append(s["key"])
    need(all(a < b for a, b in zip(keys, keys[1:])))
    need(is_hex(v["clocks"], 32))
    time_value(v["issued_at"])
    return {"clocks": v["clocks"], "use": list(u), "stores": keys}


def notice_fields(v):
    need(v["type"] in ("request", "order") and v["ground"] in ("illegal", "terms"))
    text(v["ground_ref"], 0, 500)
    text(v["statement"], 1, 5000)
    text(v["contact"], 0, 500)
    return {"type": v["type"], "targets": subject_list(v["targets"], 1, 16), "issued_at": time_value(v["issued_at"])}


def takedown_fields(v):
    m = v["matched"]
    need(isinstance(m, list) and len(m) <= 256)
    for e in m:
        members(e, ["match", "candidate"])
        need(is_hex(e["match"], 32) and is_hex(e["candidate"], 32))
    need(all(a["match"] < b["match"] for a, b in zip(m, m[1:])))
    need(v["verdict"] in ("remove", "disable"))
    a = members(v["automated"], ["detection", "decision"])
    need(a["detection"] in ("yes", "no") and a["decision"] in ("not", "partially", "fully"))
    text(v["explanation"], 1, 2000)
    text(v["facts"], 1, 5000)
    need(is_hex(v["policy"], 32) and (v["notice"] is None or is_hex(v["notice"], 32)))
    if v["clock"] is not None:
        clock_id(v["clock"])
    cs = None if v["clock_start"] is None else time_value(v["clock_start"])
    ra = None if v["reply_at"] is None else time_value(v["reply_at"])
    need(v["ground"] in ("illegal", "terms") and is_int(v["category"]) and 1 <= v["category"] <= 16)
    return {"policy": v["policy"], "notice": v["notice"], "clock": v["clock"], "clock_start": cs, "reply_at": ra,
            "targets": subject_list(v["targets"], 1, 16), "exact": subject_list(v["exact"], 1, 4096), "matched": m,
            "ground": v["ground"], "category": v["category"], "automated": a["decision"] != "not",
            "decided_at": time_value(v["decided_at"])}


def match_fields(v):
    need(is_hex(v["source"], 32) and is_hex(v["candidate"], 32) and v["source"] != v["candidate"])
    SYV.term(v["method"])
    need(is_int(v["distance"]) and is_int(v["threshold"]) and v["distance"] <= v["threshold"])
    time_value(v["issued_at"])
    return {"source": v["source"], "candidate": v["candidate"]}


def purge_fields(v):
    need(is_hex(v["takedown"], 32))
    return {"takedown": v["takedown"], "subjects": subject_list(v["subjects"], 1, 4096), "purged_at": time_value(v["purged_at"])}


ROW = ["clock", "ground", "category", "source", "takedowns", "met", "missed", "pending", "unfixed", "automated", "median_s"]


def row_order(r):
    return (r["clock"] is not None, (r["clock"] or "").encode(), r["ground"], r["category"], r["source"])


def counts_fields(v):
    td = v["takedowns"]
    need(isinstance(td, list) and len(td) <= 4096 and all(is_hex(x, 32) for x in td))
    need(all(a < b for a, b in zip(td, td[1:])))
    rows = v["rows"]
    need(isinstance(rows, list))
    for r in rows:
        members(r, ROW)
        if r["clock"] is not None:
            clock_id(r["clock"])
        need(r["ground"] in ("illegal", "terms") and is_int(r["category"]) and 1 <= r["category"] <= 16)
        need(r["source"] in ("order", "request", "own"))
        need(all(is_int(r[k]) for k in ROW[4:10]) and (r["median_s"] is None or is_int(r["median_s"])))
        need(r["takedowns"] >= 1 and r["met"] + r["missed"] + r["pending"] + r["unfixed"] == r["takedowns"])
        need(r["automated"] <= r["takedowns"])
    need(all(row_order(a) < row_order(b) for a, b in zip(rows, rows[1:])))
    f, t = time_value(v["from"]), time_value(v["to"])
    need(f < t and is_hex(v["policy"], 32))
    time_value(v["issued_at"])
    return {"policy": v["policy"], "from": f, "to": t, "as_of": time_value(v["as_of"]), "takedowns": td, "rows": rows}


# ---------------------------------------------------------------- §6 group expansion

def expand(receipts, g):
    """§6.3: (publisher, objects) or None."""
    gr = None
    for r in receipts:
        if isinstance(r, dict) and r.get("kind") == "group":
            x = EV.read_group(r)
            if x is not None and EV.link(x).hex() == g:
                gr = x
                break
    if gr is None or not EV.group_verifies(gr) or gr["leaf_count"] > 4096:
        return None
    objs = [EV.read_object(r) for r in receipts if isinstance(r, dict) and r.get("kind") == "object"]
    out = []
    for p in range(gr["leaf_count"]):
        found = next((o for o in objs if o is not None and o["track_alias"] == gr["track_alias"] and o["group_id"] == gr["group_id"]
                      and o["object_id"] == p and EV.included(o, gr)), None)
        if found is None:
            return None
        out.append(found["content_hash"])
    return gr["signer_pubkey"], out


# ---------------------------------------------------------------- §8 one case

def evaluate(case, as_of):
    reg = read_registry(case["registry"])
    pol = read_signed(case["policy"], "wai-takedown-policy", ["kind", "clocks", "use", "stores", "issued_at", "signer_pubkey", "sig"],
                      policy_fields, ["policy"])
    if pol["clocks"] != reg["link"].hex() and on("registry-link"):
        raise Refuse("registry_mismatch", ["policy"])
    if any(u not in reg["clocks"] for u in pol["use"]) and on("policy-clocks"):
        raise Refuse("clock_unknown", ["policy"])
    notices = [read_signed(t, "wai-takedown-notice", ["kind", "type", "targets", "ground", "ground_ref", "statement", "contact",
                                                      "issued_at", "signer_pubkey", "sig"], notice_fields, ["notice", i])
               for i, t in enumerate(case.get("notices", []))]
    at = ["takedown"]
    td = read_signed(case["takedown"], "wai-takedown",
                     ["kind", "policy", "notice", "clock", "clock_start", "reply_at", "targets", "verdict", "exact", "matched", "ground",
                      "category", "automated", "explanation", "facts", "decided_at", "signer_pubkey", "sig"], takedown_fields, at)
    if td["policy"] != pol["link"].hex() and on("takedown-policy"):
        raise Refuse("policy_mismatch", at)
    if td["signer"] != pol["signer"] and on("takedown-signer"):
        raise Refuse("not_authority", at)
    notice = None
    if td["notice"] is not None:
        notice = next((n for n in notices if n["link"].hex() == td["notice"]), None)
        if notice is None and on("notice-held"):
            raise Refuse("notice_unknown", at)
        if notice is not None:
            nk = {key(x) for x in notice["targets"]}
            if (not all(key(x) in nk for x in td["targets"]) or notice["issued_at"] > td["decided_at"]) and on("notice-targets"):
                raise Refuse("notice_mismatch", at)
    clock = None
    if td["clock"] is not None:
        if td["clock"] not in pol["use"] and on("clock-in-policy"):
            raise Refuse("clock_unknown", at)
        clock = reg["clocks"].get(td["clock"])
    cs, ra = td["clock_start"], td["reply_at"]
    ok = (td["clock"] is None) == (cs is None)
    ok = ok and (cs is None or cs <= td["decided_at"])
    if ok and clock is not None and cs is not None and clock["starts"] == "receipt":
        ok = notice is not None and notice["issued_at"] <= cs
    if ok and ra is not None:
        ok = clock is not None and clock["type"] == "reply_window" and cs is not None and ra >= cs
    if not ok and on("clock-consistent"):
        raise Refuse("clock_invalid", at)

    records = []
    for i, t in enumerate(case.get("records", [])):
        try:
            r = EV.read_record(t, i)
            good = ed_verify(bytes.fromhex(r["signer_pubkey"]), EV.signing_input(r["raw"]), bytes.fromhex(r["sig"])) is None
        except EV.Refuse:
            r, good = None, False
        if not good and on("records"):
            raise Refuse("evidence_invalid", ["record", i])
        if r is not None:
            records.append(r)
    claims = []
    for i, t in enumerate(case.get("claims", [])):
        try:
            c = SYV.read(t, "wai-origin", SYV.ORIGIN, SYV.origin_fields)
        except SYV.Refuse:
            if on("claims"):
                raise Refuse("evidence_invalid", ["claim", i])
            continue
        claims.append(c)
    acks = []
    for i, a in enumerate(case.get("acknowledgements", [])):
        try:
            rec = SEV.read_ack(bytes.fromhex(a["record"]))
            lists = [bytes.fromhex(x) for x in a["items"]]
            need(len(lists) == len(rec["entries"]))
            hashes = []
            for e, raw in zip(rec["entries"], lists):
                items = SEV.read_items(raw)
                need(SEV.recomputes(e, items, raw))
                hashes += [h.hex() for h, _, _ in items]
            hop = rec["downstream"].hex() if rec["mode"] == SEV.ACK and rec["role"] == SEV.HOP else None
            acks.append((rec["upstream"].hex(), hop, hashes))
        except (SEV.Refuse, Bad, ValueError, KeyError, TypeError):
            if on("acknowledgements"):
                raise Refuse("evidence_invalid", ["acknowledgement", i])
    # The cache-transmission claims (cache-authority §5): each of that class and through
    # jwp-receipts §3.1 steps 1 to 4; what each shows, (signer, group, positions sent).
    sent = []
    for i, t in enumerate(case.get("cache_transmissions", [])):
        try:
            c = CV.verify_claim(t)
            need(c["class"] == "cache-transmission")
        except (CV.Reject, Bad):
            if on("cache-transmissions"):
                raise Refuse("evidence_invalid", ["cache_transmission", i])
            continue
        bits = bytes.fromhex(c["body"]["sent"])
        sent.append((c["signer_pubkey"], c["subject"], [p for p in range(8 * len(bits)) if bits[p // 8] & (0x80 >> (p % 8))]))
    matches = [read_signed(t, "wai-takedown-match", ["kind", "source", "candidate", "method", "distance", "threshold", "issued_at",
                                                     "signer_pubkey", "sig"], match_fields, ["match", i])
               for i, t in enumerate(case.get("matches", []))]
    purges = []
    for i, t in enumerate(case.get("purges", [])):
        p = read_signed(t, "wai-takedown-purge", ["kind", "takedown", "subjects", "purged_at", "signer_pubkey", "sig"], purge_fields,
                        ["purge", i])
        if (p["takedown"] != td["link"].hex() or p["purged_at"] > as_of) and on("purge-takedown"):
            raise Refuse("purge_mismatch", ["purge", i])
        purges.append(p)

    # §6.1: the closure, by its rules until none adds a subject.
    closure = {key(x): x for x in td["targets"]}
    groups = {}
    via_records, via_claims = set(), set()

    def claim_subject(c):
        # Samples a party presents and does not publish (synthesis §3.1) are no item a holder
        # serves or keeps for others: such a claim adds nothing (§4.2 rule 3).
        s = c["value"]["subject"]
        return None if s.get("type") == "samples" else {k: s[k] for k in s}

    while True:
        before = len(closure)
        for s in list(closure.values()):
            if s["type"] == "group" and s["receipt"] not in groups:
                e = expand(case.get("receipts", []), s["receipt"])
                groups[s["receipt"]] = e
                if e is not None and on("rule-group"):
                    for h in e[1]:
                        closure[key(obj(h))] = obj(h)
        objects = {s["content_hash"] for s in closure.values() if s["type"] == "object"}
        for i, r in enumerate(records):
            if any(x["content_hash"] in objects for x in r["inputs"]) and on("rule-derivation"):
                via_records.add(i)
                closure[key(obj(r["output"]["content_hash"]))] = obj(r["output"]["content_hash"])
        for i, c in enumerate(claims):
            if any(x["content_hash"].hex() in objects for x in c["inputs"]) and on("rule-claim"):
                via_claims.add(i)
                s = claim_subject(c)
                if s is not None:
                    closure[key(s)] = s
        unexpanded = any(s["type"] == "group" and s["receipt"] not in groups for s in closure.values())
        if len(closure) == before and not unexpanded:
            break
    for g in sorted(groups):
        if groups[g] is None and on("group-expanded"):
            raise Refuse("group_unexpanded", ["group", g])
    objects = {s["content_hash"] for s in closure.values() if s["type"] == "object"}

    # §6.4, §6.5: the stated set and the matches.
    candidates = {m["candidate"] for m in matches}
    for k, s in enumerate(td["exact"]):
        if key(s) not in closure:
            if s["type"] == "object" and s["content_hash"] in candidates and on("match-not-exact"):
                raise Refuse("match_as_exact", ["exact", k])
            if on("exact-supported"):
                raise Refuse("closure_unsupported", ["exact", k])
    stated = {key(s) for s in td["exact"]}
    for k, ks in enumerate(sorted(closure, key=lambda x: x.encode())):
        if ks not in stated and on("exact-complete"):
            raise Refuse("closure_incomplete", ["closure", k])
    for k, e in enumerate(td["matched"]):
        m = next((m for m in matches if m["link"].hex() == e["match"] and m["candidate"] == e["candidate"]), None)
        if m is None:
            if on("match-held"):
                raise Refuse("match_unknown", ["matched", k])
            continue
        if m["source"] not in objects and on("match-anchored"):
            raise Refuse("match_unanchored", ["matched", k])

    # §7.1: the required pairs.
    required, roles = {}, {}

    def needs(k, s, role):
        roles.setdefault(k, set()).add(role)
        required[(k, key(s))] = s

    if on("store-holders"):
        for k in pol["stores"]:
            for s in closure.values():
                needs(k, s, "store")
    if on("deriver-holders"):
        for i in sorted(via_records):
            r = records[i]
            for x in r["inputs"]:
                if x["content_hash"] in objects:
                    needs(r["signer_pubkey"], obj(x["content_hash"]), "deriver")
            needs(r["signer_pubkey"], obj(r["output"]["content_hash"]), "deriver")
    if on("claimant-holders"):
        for i in sorted(via_claims):
            c = claims[i]
            for x in c["inputs"]:
                if x["content_hash"].hex() in objects:
                    needs(c["signer"].hex(), obj(x["content_hash"].hex()), "claimant")
            if claim_subject(c) is not None:
                needs(c["signer"].hex(), claim_subject(c), "claimant")
    if on("publisher-holders"):
        for g, e in groups.items():
            if key({"type": "group", "receipt": g}) in closure:
                needs(e[0], {"type": "group", "receipt": g}, "publisher")
    if on("served-holders"):
        for up, hop, hashes in acks:
            for h in hashes:
                if h in objects:
                    needs(up, obj(h), "served")
                    if hop is not None:
                        needs(hop, obj(h), "received")
    if on("sent-holders"):
        # A node's own word that it sent objects of a group, read by position through the group's
        # object receipts (§6.3); a group not among the receipts names no object.
        for signer, g, pos in sent:
            e = expand(case.get("receipts", []), g)
            if e is None:
                continue
            for p in pos:
                if p < len(e[1]) and e[1][p] in objects:
                    needs(signer, obj(e[1][p]), "served")
    matched_required = {}
    cands = {e["candidate"] for e in td["matched"]}
    for k in pol["stores"]:
        for c in cands:
            matched_required[(k, key(obj(c)))] = obj(c)
    for up, hop, hashes in acks:
        for h in hashes:
            if h in cands:
                for k in [up] + ([hop] if hop is not None else []):
                    matched_required[(k, key(obj(h)))] = obj(h)

    # §7.2: what a purge covers.
    def covers(p, s):
        named = {key(x) for x in p["subjects"]}
        if key(s) in named:
            return True
        if not on("group-coverage"):
            return False
        if s["type"] == "object":
            return any(x["type"] == "group" and groups.get(x["receipt"]) and s["content_hash"] in groups[x["receipt"]][1]
                       for x in p["subjects"])
        if s["type"] == "group" and groups.get(s["receipt"]):
            return all(key(obj(h)) in named for h in groups[s["receipt"]][1])
        return False

    def pair_time(k, s):
        ts = [p["purged_at"] for p in purges if (p["signer"] == k or not on("holder-only")) and covers(p, s)]
        return min(ts) if ts else None

    missing, times = [], []
    for (k, ks) in sorted(required, key=lambda x: (x[0], x[1].encode())):
        tt = pair_time(k, required[(k, ks)])
        if tt is None:
            missing.append({"holder": k, "subject": required[(k, ks)]})
        else:
            times.append(tt)
    m_missing, m_times = [], []
    for (k, ks) in sorted(matched_required, key=lambda x: (x[0], x[1].encode())):
        tt = pair_time(k, matched_required[(k, ks)])
        if tt is None:
            m_missing.append({"holder": k, "subject": matched_required[(k, ks)]})
        else:
            m_times.append(tt)
    uncounted = []
    for i, p in enumerate(purges):
        for s in p["subjects"]:
            if (p["signer"], key(s)) not in required and (p["signer"], key(s)) not in matched_required:
                uncounted.append({"purge": i, "subject": s})

    # §7.3: the time to removal and the clock.
    complete = not missing
    first = min(times) if times else None
    last = max(times) if times else None
    start = cs if cs is not None else td["decided_at"]
    ttr = max(0, last - start) if complete and last is not None else None
    deadline = window_end = None
    reason = None
    if clock is not None and clock["type"] == "deadline":
        deadline = start + clock["duration_s"]
        if complete and last is not None and (last <= deadline or not on("deadline")):
            verdict = "met"
        elif complete:
            verdict, reason = "missed", "deadline_missed"
        elif as_of > deadline and on("purges-complete"):
            verdict, reason = "missed", "purge_missing"
        else:
            verdict = "pending"
    elif clock is not None and clock["type"] == "reply_window":
        window_end = start + clock["duration_s"] if ra is None else min(ra, start + clock["duration_s"])
        if first is not None and first < window_end and on("window"):
            verdict, reason = "missed", "window_not_observed"
        else:
            verdict = "met" if complete else "pending"
    else:
        verdict = "unfixed" if complete else "pending"

    facts = {
        "closure": [closure[k] for k in sorted(closure, key=lambda x: x.encode())],
        "holders": [{"key": k, "roles": sorted(roles[k]), "required": sum(1 for (kk, _) in required if kk == k)} for k in sorted(roles)],
        "missing": missing,
        "uncounted": uncounted,
        "clock": td["clock"],
        "start": start,
        "deadline": deadline,
        "window_end": window_end,
        "first_purge": first,
        "last_purge": last,
        "time_to_removal_s": ttr,
        "verdict": verdict,
        "matched": {"candidates": sorted({e["candidate"] for e in td["matched"]}), "missing": m_missing,
                    "last_purge": max(m_times) if m_times else None},
    }
    if reason is not None:
        out = {"outcome": "missed", "reason": reason, "facts": facts}
    elif verdict == "pending":
        out = {"outcome": "pending", "facts": facts}
    else:
        out = {"outcome": "accept", "facts": facts}
    source = "own" if notice is None else notice["type"]
    return {"policy": pol, "takedown": td, "source": source, "outcome": out}


def verify_case(case):
    try:
        return evaluate(case, case["as_of"])["outcome"]
    except Refuse as e:
        return {"outcome": "refuse", "reason": e.reason, "at": e.at}


# ---------------------------------------------------------------- §9 counts

def rows_of(done):
    by = {}
    for e in done:
        t = e["takedown"]
        k = (t["clock"] is not None, (t["clock"] or "").encode(), t["ground"], t["category"], e["source"])
        r, ttr = by.setdefault(k, ({"clock": t["clock"], "ground": t["ground"], "category": t["category"], "source": e["source"],
                                    "takedowns": 0, "met": 0, "missed": 0, "pending": 0, "unfixed": 0, "automated": 0,
                                    "median_s": None}, []))
        r["takedowns"] += 1
        r["automated"] += 1 if t["automated"] else 0
        f = e["outcome"]["facts"]
        r[f["verdict"]] += 1
        if f["time_to_removal_s"] is not None:
            ttr.append(f["time_to_removal_s"])
    out = []
    for k in sorted(by):
        r, ttr = by[k]
        ttr.sort()
        r["median_s"] = ttr[(len(ttr) - 1) // 2] if ttr else None
        out.append(r)
    return out


def verify_counts(counts_text, cases):
    c = read_signed(counts_text, "wai-takedown-counts", ["kind", "policy", "from", "to", "as_of", "takedowns", "rows", "issued_at",
                                                         "signer_pubkey", "sig"], counts_fields, ["counts"])
    done = []
    for i, case in enumerate(cases):
        try:
            done.append(evaluate(case, c["as_of"] if on("counts-as-of") else case["as_of"]))
        except Refuse as e:
            raise Refuse(e.reason, ["case", i] + e.at)
    for e in done:
        if e["policy"]["link"].hex() != c["policy"]:
            raise Refuse("policy_mismatch", ["counts"])
        if e["policy"]["signer"] != c["signer"]:
            raise Refuse("not_authority", ["counts"])
    links = [e["takedown"]["link"].hex() for e in done]
    if set(links) != set(c["takedowns"]) or len(set(links)) != len(links) \
            or any(not c["from"] <= e["takedown"]["decided_at"] < c["to"] for e in done):
        raise Refuse("counts_mismatch", ["counts"])
    rows = rows_of(done)
    if rows != c["rows"] and on("counts-rows"):
        raise Refuse("counts_mismatch", ["counts"])
    return rows


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

def case_of(world, over):
    c = dict(world)
    c.update(over)
    c["as_of"] = time_value(c["as_of"])
    return c


def section(text_, head):
    at = text_.index("\n" + head)
    nxt = re.search(r"\n#{2} ", text_[at + 1:])
    return text_[at: at + 1 + nxt.start()] if nxt else text_[at:]


with open(EXTENSION, encoding="utf-8") as _f:
    _TEXT = _f.read()
REASONS = [line.split("`")[1] for line in section(_TEXT, "## 11. Reasons").splitlines() if line.startswith("| `")]


def check(doc, quiet=True):
    fails = []
    world = doc["world"]
    by = {}
    for v in doc["cases"]:
        try:
            case = case_of(world, v["case"])
            by[v["name"]] = case
            got = verify_case(case)
        except Exception as e:  # noqa: BLE001 — a crash on a mutated or weakened run is a failure
            got = {"outcome": "crash: %r" % (e,)}
        if got != v["expect"]:
            fails.append("%s: got %s" % (v["name"], json.dumps(got)[:240]))
            if not quiet:
                print("FAIL case %s: got %s, the corpus states %s" % (v["name"], json.dumps(got)[:300], json.dumps(v["expect"])[:300]))
        elif not quiet:
            print("PASS case %s: %s" % (v["name"], got["outcome"] + (" %s" % got["reason"] if "reason" in got else "")
                                        + (" at %s" % got["at"] if "at" in got else "")))
    for v in doc["counts"]:
        try:
            rows = verify_counts(v["counts"], [by[n] for n in v["cases"]])
            got = {"outcome": "accept", "rows": rows}
        except Refuse as e:
            got = {"outcome": "refuse", "reason": e.reason, "at": e.at}
        except Exception as e:  # noqa: BLE001
            got = {"outcome": "crash: %r" % (e,)}
        if got != v["expect"]:
            fails.append("%s: got %s" % (v["name"], json.dumps(got)[:240]))
            if not quiet:
                print("FAIL counts %s: got %s" % (v["name"], json.dumps(got)[:300]))
        elif not quiet:
            print("PASS counts %s: %s" % (v["name"], got["outcome"] + (" %s" % got["reason"] if "reason" in got else "")))
    return fails


def flip_hex_digit(s, at):
    return s[:at] + ("1" if s[at] != "1" else "2") + s[at + 1:]


def mutations(doc):
    """Copies of the corpus, each with one defect a conforming verifier must catch."""
    by = {v["name"]: i for i, v in enumerate(doc["cases"])}

    def mutate(label, f):
        d = copy.deepcopy(doc)
        f(d)
        return label, d

    def case(d, name):
        return d["cases"][by[name]]

    def retext(texts, i, f):
        v = json.loads(texts[i])
        f(v)
        texts[i] = EV.jcs(v)

    def facts(d, name):
        return case(d, name)["expect"]["facts"]

    return [
        mutate("a digit of the met request's takedown signature",
               lambda d: case(d, "request-48h-met")["case"].update(takedown=EV.jcs(dict(json.loads(case(d, "request-48h-met")["case"]["takedown"]), sig=flip_hex_digit(json.loads(case(d, "request-48h-met")["case"]["takedown"])["sig"], 7))))),
        mutate("the met request's last purge withheld", lambda d: case(d, "request-48h-met")["case"]["purges"].pop()),
        mutate("the late purge of the missed order moved an hour earlier",
               lambda d: retext(case(d, "order-3h-missed")["case"]["purges"], 5, lambda v: v.update(purged_at="2026-10-21T10:57:40Z"))),
        mutate("the met request's time to removal stated otherwise", lambda d: facts(d, "request-48h-met").update(time_to_removal_s=42899)),
        mutate("the met request's closure stated without its live track", lambda d: facts(d, "request-48h-met")["closure"].pop()),
        mutate("the stranger's purge stated as counted", lambda d: facts(d, "stranger-purge-not-counted").update(uncounted=[])),
        mutate("the matched candidate stated as purged by nobody", lambda d: facts(d, "request-48h-met")["matched"].update(last_purge=None)),
        mutate("a registry duration changed", lambda d: d["world"].update(registry=d["world"]["registry"].replace("172800", "172799"))),
        mutate("the generated group's receipts withheld from the world",
               lambda d: d["world"].update(receipts=[r for r in d["world"]["receipts"] if r.get("group_id") != 5])),
        mutate("the thumbnail's derivation record withheld", lambda d: d["world"]["records"].pop(1)),
        mutate("the upscale's claim withheld", lambda d: d["world"]["claims"].pop(1)),
        mutate("the cache's acknowledgement of the thumbnail withheld", lambda d: d["world"]["acknowledgements"].pop(0)),
        mutate("a byte of an acknowledgement's signature",
               lambda d: d["world"]["acknowledgements"][1].update(record=flip_hex_digit(d["world"]["acknowledgements"][1]["record"], len(d["world"]["acknowledgements"][1]["record"]) - 9))),
        mutate("the cache's claim that it sent the source withheld",
               lambda d: case(d, "cache-transmitted-not-purged")["case"]["cache_transmissions"].pop()),
        mutate("the unfixed request stated met", lambda d: facts(d, "request-unfixed").update(verdict="met")),
        mutate("the reply window's end moved", lambda d: facts(d, "own-reply-early").update(window_end=facts(d, "own-reply-early")["window_end"] + 1)),
        mutate("a refused case's place moved", lambda d: case(d, "derivative-left-out")["expect"].update(at=["closure", 5])),
        mutate("a counts row's median stated otherwise", lambda d: d["counts"][0]["expect"]["rows"][0].update(median_s=1199)),
        mutate("the counts verified without the pending order", lambda d: d["counts"][0]["cases"].remove("order-3h-pending")),
    ]


def main(corpus_dir):
    fails = []
    st = ED._self_test()
    print(("PASS" if not st else "FAIL") + " Ed25519 strict rule self-test " + ",".join(st))
    fails += ["ed25519 self-test"] if st else []
    ok = len(REASONS) == 22 and len(set(REASONS)) == len(REASONS)
    print(("PASS" if ok else "FAIL") + " §11 reads from the extension: %d reasons" % len(REASONS))
    fails += [] if ok else ["the extension's §11 does not read"]
    with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as f:
        doc = json.load(f)
    with open(PUBLISHED, encoding="utf-8") as f:
        published = f.read()
    same = doc["world"]["registry"] == published
    print(("PASS" if same else "FAIL") + " the world's registry is registry/takedown-clocks.json as published")
    fails += [] if same else ["the world's registry is not the published one"]
    fails += check(doc, quiet=False)
    extra = sorted(set(os.listdir(corpus_dir)) - {"vectors.json", "README.md", "__pycache__"})
    print(("PASS" if not extra else "FAIL") + " no file but vectors.json and README.md")
    fails += ["an extra file: %s" % n for n in extra]
    seen = {v["expect"].get("reason") for v in doc["cases"]} | {v["expect"].get("reason") for v in doc["counts"]}
    missing = [r for r in REASONS if r not in seen]
    print(("PASS" if not missing else "FAIL") + " every reason of §11 appears in the corpus")
    fails += ["no case gives %s" % r for r in missing]
    for label, d in mutations(doc):
        caught = check(d)
        print(("PASS" if caught else "FAIL") + " mutated copy fails: %s%s" % (label, " (%s)" % caught[0][:90] if caught else ""))
        if not caught:
            fails.append("a mutated copy verifies: %s" % label)
    for name in CHECKS:
        DROPPED.clear()
        DROPPED.add(name)
        caught = check(doc)
        DROPPED.clear()
        print(("PASS" if caught else "FAIL") + " the corpus catches the verifier without its check %s%s"
              % (name, " (%s)" % caught[0][:90] if caught else ""))
        if not caught:
            fails.append("the check %s can be dropped unnoticed" % name)
    for f in fails:
        print("FAIL " + f)
    print(("PASS" if not fails else "FAIL") + " takedown corpus: %d cases, %d counts" % (len(doc["cases"]), len(doc["counts"])))
    return 1 if fails else 0


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