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

#!/usr/bin/env python3
"""Independent verifier of the AI usage corpus (extensions/ai-usage.md).

Written from the extension's text. It shares no code with wai-rs, which generates the corpus,
and never runs it. It checks Ed25519 signatures with ../claim-conformance/_ed25519.py, WAI's
strict rule (SPEC §7 *Signatures*) written from RFC 8032 with the standard library, and
computes BLAKE3 with ../claim-conformance/_blake3.py, written from its specification. A room's
descriptor and grants are read by the live-rooms corpus's own second implementation,
live_rooms_verify.py (live-rooms §3, §4), written from its text alone. Everything else is here:

- RFC 9651's Dictionary parser, as §3.3 reads it, and the vocabularies' preferences (§3);
- the canonical form (RFC 8785 over the subset jwp-receipts §2.4.3 allows) and the digest;
- the envelope's binding (§4.1): the content hash, the group receipt's legacy payload and its
  signature, the object's inclusion proof (jwp-receipts §2.1-§2.3), the manifest's member;
- the room usage record (§4.2) and a grant's terms (§4.3);
- the HTTP fields (§4.4) and the catalog field (§4.5);
- the precedence rule (§5);
- the consumption claim (§6): jwp-receipts §3.1 steps 1 to 5 for its class, then §6.3;
- the gate (§7).

For every case it requires the outcome the corpus states; of the corpus, that every reason of
§9 appears in it and that no other file is there. 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 ai_usage_verify.py <live-conformance/usage>

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

import copy
import hashlib
import json
import os
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"))
import live_events_verify as LEV  # noqa: E402
import live_rooms_verify as LRV  # noqa: E402
import _ed25519 as ED  # noqa: E402
import _blake3 as B3  # noqa: E402

EXTENSION = os.path.join(HERE, "..", "extensions", "ai-usage.md")

AIPREF_REV = "draft-ietf-aipref-vocab-08"   # §3.2
CAWG_REV = "1.1"                            # §3.4
TDMREP_REV = "CG-FINAL-tdmrep-20240510"     # §3.5
LABELS = ("train-ai", "ai-use", "search")   # draft-ietf-aipref-vocab-08 §6.1
CAWG_KEYS = ("cawg.data_mining", "cawg.ai_inference", "cawg.ai_generative_training", "cawg.ai_training")
TDM_TERM = "tdm-reservation"
VOCABULARIES = ("aipref", "cawg", "tdmrep")
ROOM_MAGIC = b"WAU1"
ROOM_DOMAIN = b"wai:ai-usage-room\x01"      # §4.2
OBJECT_DOMAIN = b"moq-jwp:object\x01"       # jwp-receipts §2.1
LEAF, NODE = b"moq-jwp:merkle-leaf\x01", b"moq-jwp:merkle-node\x01"
GROUP_SIGN = b"moq-jwp:group-sign\x01"      # jwp-receipts §2.3
CLAIM_SIGN, CLAIM_LINK = b"moq-jwp:claim\x01", b"moq-jwp:claim-link\x01"
MAX_INT = 2 ** 53 - 1

# A check named here is left out: the verifier weakened on purpose, to show the corpus catches it.
SKIP = set()


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


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


def b3(b):
    return B3.blake3(b)


def reasons_of_extension():
    """§9's codes, read from the text."""
    with open(EXTENSION, encoding="utf-8") as f:
        text = f.read()
    at = text.index("\n## 9. Reasons")
    end = text.index("\n## ", at + 1)
    return [line.split("`")[1] for line in text[at:end].splitlines() if line.startswith("| `")]


REASONS = reasons_of_extension()


# ---------------------------------------------------------------- JSON and the canonical form

class Number:
    """A JSON number that is not an unsigned integer: never a value a statement or claim holds."""


def strict_json(text):
    """JSON in which no object names a member twice (§3.6); a number that is not a plain
    unsigned integer is kept as a Number marker. ValueError when it does not read."""
    def pairs(ps):
        d = {}
        for k, v in ps:
            if k in d:
                raise ValueError("repeated member " + k)
            d[k] = v
        return d

    def no_constant(c):
        raise ValueError(c)

    v = json.loads(text, object_pairs_hook=pairs, parse_constant=no_constant,
                   parse_float=lambda s: Number(), parse_int=lambda s: Number() if s.startswith("-") else int(s))

    def walk(x, level):
        # jwp-receipts §3.1 step 1: no escape that leaves a surrogate unpaired, at most 32 levels.
        if isinstance(x, str):
            if any("\ud800" <= c <= "\udfff" for c in x):
                raise ValueError("an unpaired surrogate")
        elif isinstance(x, (list, dict)):
            if level + 1 > 32:
                raise ValueError("nested too deeply")
            for k in (x if isinstance(x, dict) else []):
                walk(k, level)
            for y in (x.values() if isinstance(x, dict) else x):
                walk(y, level + 1)
    walk(v, 0)
    return v


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


def jcs(v):
    """RFC 8785 over strings, unsigned integers, null, booleans, arrays and objects."""
    if v is None:
        return "null"
    if v is True or v is False:
        return "true" if v else "false"
    if isinstance(v, int) and not isinstance(v, Number):
        if not 0 <= v <= MAX_INT:
            raise ValueError("an integer outside the subset")
        return str(v)
    if isinstance(v, str):
        return json.dumps(v, ensure_ascii=False)
    if isinstance(v, list):
        return "[" + ",".join(jcs(x) for x in v) + "]"
    if isinstance(v, dict):
        keys = sorted(v, key=lambda k: k.encode("utf-16-be"))
        return "{" + ",".join(json.dumps(k, ensure_ascii=False) + ":" + jcs(v[k]) for k in keys) + "}"
    raise ValueError("a value outside the subset")


# ---------------------------------------------------------------- §3.3 RFC 9651 Dictionary

class SFFail(Exception):
    pass


LCALPHA = set("abcdefghijklmnopqrstuvwxyz")
DIGIT = set("0123456789")
ALPHA = LCALPHA | set("ABCDEFGHIJKLMNOPQRSTUVWXYZ")
TCHAR = ALPHA | DIGIT | set("!#$%&'*+-.^_`|~")
B64 = ALPHA | DIGIT | set("+/")


class SF:
    def __init__(self, s):
        self.s, self.i = s, 0

    def peek(self):
        return self.s[self.i] if self.i < len(self.s) else None

    def take(self):
        c = self.peek()
        if c is None:
            raise SFFail()
        self.i += 1
        return c

    def need(self, c):
        if self.peek() != c:
            raise SFFail()
        self.i += 1

    def spaces(self):
        while self.peek() == " ":
            self.i += 1

    def ows(self):
        while self.peek() in (" ", "\t"):
            self.i += 1

    def key(self):                                    # RFC 9651 §4.2.3.3
        c = self.peek()
        if c is None or not (c in LCALPHA or c == "*"):
            raise SFFail()
        out = ""
        while self.peek() is not None and (self.peek() in LCALPHA or self.peek() in DIGIT or self.peek() in "_-.*"):
            out += self.take()
        return out

    def parameters(self):                             # §4.2.3.2
        while self.peek() == ";":
            self.i += 1
            self.spaces()
            self.key()
            if self.peek() == "=":
                self.i += 1
                self.bare()

    def number(self):                                 # §4.2.4; True for an Integer
        kind, digits = "integer", ""
        if self.peek() == "-":
            self.i += 1
        if self.peek() is None or self.peek() not in DIGIT:
            raise SFFail()
        while self.peek() is not None:
            c = self.take()
            if c in DIGIT:
                digits += c
            elif kind == "integer" and c == ".":
                if len(digits) > 12:
                    raise SFFail()
                digits += c
                kind = "decimal"
            else:
                self.i -= 1
                break
            if kind == "integer" and len(digits) > 15:
                raise SFFail()
            if kind == "decimal" and len(digits) > 16:
                raise SFFail()
        if kind == "decimal":
            if digits.endswith(".") or len(digits.split(".")[1]) > 3:
                raise SFFail()
        return kind == "integer"

    def string(self):                                 # §4.2.5
        self.need('"')
        while self.peek() is not None:
            c = self.take()
            if c == "\\":
                n = self.take()
                if n not in ('"', "\\"):
                    raise SFFail()
            elif c == '"':
                return
            elif ord(c) < 0x20 or ord(c) >= 0x7F:
                raise SFFail()
        raise SFFail()

    def token(self):                                  # §4.2.6
        c = self.peek()
        if c is None or not (c in ALPHA or c == "*"):
            raise SFFail()
        out = ""
        while self.peek() is not None and (self.peek() in TCHAR or self.peek() in ":/"):
            out += self.take()
        return out

    def byte_sequence(self):                          # §4.2.7, with §3.3's reading of its content
        self.need(":")
        end = self.s.find(":", self.i)
        if end < 0:
            raise SFFail()
        content, self.i = self.s[self.i:end], end + 1
        body = content.rstrip("=")
        pad = len(content) - len(body)
        if pad > 2 or any(c not in B64 for c in body) or len(body) % 4 == 1:
            raise SFFail()

    def boolean(self):                                # §4.2.8
        self.need("?")
        if self.peek() not in ("0", "1"):
            raise SFFail()
        self.i += 1

    def date(self):                                   # §4.2.9
        self.need("@")
        if not self.number():
            raise SFFail()

    def display(self):                                # §4.2.10
        self.need("%")
        self.need('"')
        out = b""
        while self.peek() is not None:
            c = self.take()
            if ord(c) < 0x20 or ord(c) >= 0x7F:
                raise SFFail()
            if c == "%":
                h = self.s[self.i:self.i + 2]
                if len(h) != 2 or any(x not in "0123456789abcdef" for x in h):
                    raise SFFail()
                out += bytes.fromhex(h)
                self.i += 2
            elif c == '"':
                try:
                    out.decode("utf-8")
                except UnicodeDecodeError:
                    raise SFFail()
                return
            else:
                out += c.encode()
        raise SFFail()

    def bare(self):                                   # §4.2.3.1; ("token", text) or ("other",)
        c = self.peek()
        if c is None:
            raise SFFail()
        if c == "-" or c in DIGIT:
            self.number()
            return ("other",)
        if c == '"':
            self.string()
            return ("other",)
        if c in ALPHA or c == "*":
            return ("token", self.token())
        if c == ":":
            self.byte_sequence()
            return ("other",)
        if c == "?":
            self.boolean()
            return ("other",)
        if c == "@":
            self.date()
            return ("other",)
        if c == "%":
            self.display()
            return ("other",)
        raise SFFail()

    def item(self):                                   # §4.2.3
        v = self.bare()
        self.parameters()
        return v

    def inner_list(self):                             # §4.2.1.2
        self.need("(")
        while self.peek() is not None:
            self.spaces()
            if self.peek() == ")":
                self.i += 1
                self.parameters()
                return ("other",)
            self.item()
            if self.peek() not in (" ", ")"):
                raise SFFail()
        raise SFFail()

    def dictionary(self):                             # §4.2.2
        d = {}
        while self.peek() is not None:
            k = self.key()
            if self.peek() == "=":
                self.i += 1
                d[k] = self.inner_list() if self.peek() == "(" else self.item()
            else:
                self.parameters()
                d[k] = ("other",)
            self.ows()
            if self.peek() is None:
                return d
            if self.take() != ",":
                raise SFFail()
            self.ows()
            if self.peek() is None:
                raise SFFail()
        return d


def parse_dictionary(text):
    """§4.2 for a Dictionary: the members, or None when it fails to parse."""
    if any(ord(c) > 0x7F for c in text):
        return None
    p = SF(text)
    try:
        p.spaces()
        d = p.dictionary()
        p.spaces()
    except SFFail:
        return None
    return d if p.peek() is None else None


def aipref_prefs(value):
    """draft-ietf-aipref-vocab-08 §6.5: each label's preference, or None when it does not parse."""
    d = parse_dictionary(value)
    if d is None:
        return None
    out = {}
    for label in LABELS:
        m = d.get(label)
        out[label] = ("allow" if m == ("token", "y") else "disallow" if m == ("token", "n") else "unknown")
    return out


# ---------------------------------------------------------------- §3 the statement

def cawg_key(k):
    """§3.4's key form."""
    if not 1 <= len(k.encode()) <= 255:
        return False
    return all(part and all(c in ALPHA or c in DIGIT or c in "_-" for c in part) for part in k.split("."))


def is_policy(s):
    b = s.encode()
    return (8 <= len(b) <= 2048 and all(0x21 <= c <= 0x7E for c in b)
            and (s.startswith("https://") or s.startswith("http://")))


def read_statement(v):
    """§3.6: the statement as {vocabulary: {term: preference}} with its digest, or Refuse."""
    bad = Refuse("usage_malformed")
    if not isinstance(v, dict) or not v or any(k not in VOCABULARIES for k in v):
        raise bad
    prefs, revisions = {}, []
    if "aipref" in v:
        a = v["aipref"]
        if not isinstance(a, dict) or set(a) != {"revision", "value"}:
            raise bad
        if not isinstance(a["revision"], str) or not isinstance(a["value"], str):
            raise bad
        val = a["value"]
        if "aipref-length" not in SKIP and not 1 <= len(val) <= 1024:
            raise bad
        p = aipref_prefs(val) if "aipref-parse" not in SKIP else (aipref_prefs(val) or {x: "unknown" for x in LABELS})
        if p is None or any(ord(c) > 0x7F for c in val):
            raise bad
        prefs["aipref"] = p
        revisions.append((a["revision"], AIPREF_REV))
    if "cawg" in v:
        c = v["cawg"]
        if not isinstance(c, dict) or set(c) != {"revision", "entries"} or not isinstance(c["revision"], str):
            raise bad
        e = c["entries"]
        if not isinstance(e, dict) or not 1 <= len(e) <= 32:
            raise bad
        p = {}
        for k, x in e.items():
            if not cawg_key(k):
                raise bad
            if "cawg-reserved" not in SKIP and k.startswith("cawg.") and k not in CAWG_KEYS:
                raise bad
            if not isinstance(x, dict) or "use" not in x or any(m not in ("use", "constraint_info") for m in x):
                raise bad
            if x["use"] not in ("allowed", "notAllowed", "constrained"):
                raise bad
            if "constraint_info" in x and not (isinstance(x["constraint_info"], str)
                                               and 1 <= len(x["constraint_info"].encode()) <= 1024):
                raise bad
            p[k] = "allow" if x["use"] == "allowed" else "disallow"
        prefs["cawg"] = p
        revisions.append((c["revision"], CAWG_REV))
    if "tdmrep" in v:
        t = v["tdmrep"]
        if not isinstance(t, dict) or set(t) != {"revision", "tdm-reservation", "tdm-policy"}:
            raise bad
        if not isinstance(t["revision"], str):
            raise bad
        r = t["tdm-reservation"]
        if "tdm-values" not in SKIP and not (is_uint(r) and r in (0, 1)):
            raise bad
        pol = t["tdm-policy"]
        if not (pol is None or (isinstance(pol, str) and is_policy(pol))):
            raise bad
        prefs["tdmrep"] = {TDM_TERM: "disallow" if r == 1 else "allow"}
        revisions.append((t["revision"], TDMREP_REV))
    if "revisions" not in SKIP and any(got != want for got, want in revisions):
        raise Refuse("usage_revision_unknown")
    canonical = jcs(v).encode()
    if len(canonical) > 4096:
        raise bad
    return {"prefs": prefs, "digest": sha(canonical), "canonical": canonical, "policy": v.get("tdmrep", {}).get("tdm-policy")}


def statement_text(text):
    try:
        v = strict_json(text)
    except ValueError:
        raise Refuse("usage_malformed")
    return read_statement(v)


def statement_canonical(b):
    """§4.2, §4.3: the bytes read and are the canonical form of what they hold."""
    try:
        text = b.decode("utf-8")
    except UnicodeDecodeError:
        raise Refuse("usage_malformed")
    s = statement_text(text)
    if "canonical-bytes" not in SKIP and s["canonical"] != b:
        raise Refuse("usage_malformed")
    return s


def pref(s, vocabulary, term):
    """The preference a statement gives a term, or None when it does not carry the vocabulary."""
    if vocabulary not in s["prefs"]:
        return None
    return s["prefs"][vocabulary].get(term, "unknown")


def statement_report(s):
    return {"digest": s["digest"].hex(), "preferences": {v: dict(p) for v, p in s["prefs"].items()}}


# ---------------------------------------------------------------- §4.1 the envelope

def node(a, b):
    return b3(NODE + a + b)


def group_payload(g):
    """jwp-receipts §2.3, the legacy payload."""
    b = GROUP_SIGN + g["track_alias"].to_bytes(8, "big") + g["group_id"].to_bytes(8, "big")
    b += bytes.fromhex(g["root_hash"]) + g["leaf_count"].to_bytes(8, "big") + g["joules_total"].to_bytes(8, "big")
    for k in ("group_start", "group_end"):
        e = g[k].encode()
        b += len(e).to_bytes(4, "big") + e
    sid = g["signer_id"].encode()
    return b + bytes.fromhex(g["signer_pubkey"]) + len(sid).to_bytes(4, "big") + sid


def receipts_hold(g, o):
    """jwp-receipts §3 steps 1-5, the content hash left to §4.1 step 2: the legacy group
    verifies, and the object's unlabelled leaf is included at its position."""
    if "sign_version" in g or "leaf_version" in o:
        return False   # a labelled form: no case of this corpus uses one
    if "group-signature" not in SKIP and ED.verify(bytes.fromhex(g["signer_pubkey"]), group_payload(g), bytes.fromhex(g["sig"])):
        return False
    n, p = g["leaf_count"], o["merkle_proof"]
    i = p["leaf_index"]
    depth = 0 if n <= 1 else (n - 1).bit_length()
    if (o["track_alias"], o["group_id"]) != (g["track_alias"], g["group_id"]) or o["object_id"] != i or not i < n \
            or len(p["siblings"]) != depth or p["direction_bits"] != i:
        return False
    cur = b3(LEAF + bytes.fromhex(o["content_hash"]))
    for k, sib in enumerate(p["siblings"]):
        s = bytes.fromhex(sib)
        cur = node(s, cur) if (i >> k) & 1 else node(cur, s)
    return cur == bytes.fromhex(g["root_hash"])


def manifest_of(env):
    """SPEC §2, §7.1: the manifest of a WAI1 or WAI2 envelope, read with no member named twice."""
    if len(env) < 8 or env[:4] not in (b"WAI1", b"WAI2"):
        return None
    n = int.from_bytes(env[4:8], "little")
    if 8 + n > len(env):
        return None
    try:
        return strict_json(env[8:8 + n].decode("utf-8"))
    except (ValueError, UnicodeDecodeError):
        return None


def hold(env, o, g):
    """§4.1: the object held, or Refuse."""
    if not receipts_hold(g, o):
        raise Refuse("receipt_invalid")
    h = b3(OBJECT_DOMAIN + env)
    if "content-hash" not in SKIP and h != bytes.fromhex(o["content_hash"]):
        raise Refuse("usage_unbound")
    m = manifest_of(env)
    if not isinstance(m, dict):
        raise Refuse("usage_malformed")
    s = read_statement(m["ai_usage"]) if "ai_usage" in m else None
    live = m.get("live")
    lt = live.get("t") if isinstance(live, dict) else None
    t = lt if isinstance(lt, int) and not isinstance(lt, bool) and 0 <= lt < 2 ** 64 else None
    return {"hash": h, "statement": s, "t": t}


# ---------------------------------------------------------------- §4.2 the room, §4.3 terms

def read_record(b):
    """§4.2's reading."""
    if len(b) < 4:
        raise Refuse("truncated")
    if b[:4] != ROOM_MAGIC:
        raise Refuse("malformed")
    if len(b) < 5:
        raise Refuse("truncated")
    if b[4] != 1:
        raise Refuse("unknown_version")
    if len(b) < 5 + 32 + 32 + 8 + 2:
        raise Refuse("truncated")
    room, prev = b[5:37], b[37:69]
    effective = int.from_bytes(b[69:77], "little")
    n = int.from_bytes(b[77:79], "little")
    if len(b) < 79 + n + 64:
        raise Refuse("truncated")
    if len(b) > 79 + n + 64:
        raise Refuse("malformed")
    st = statement_canonical(b[79:79 + n])
    return {"room": room, "prev": prev, "effective": effective, "statement": st, "body": b[:79 + n],
            "sig": b[79 + n:], "digest": sha(b[:79 + n])}


def check_records(d, records):
    """§4.2 over a room's records: the list, or (index, reason)."""
    out = []
    for i, b in enumerate(records):
        try:
            r = read_record(b)
        except Refuse as e:
            return None, (i, e.reason)
        if "record-room" not in SKIP and r["room"] != d["room"]:
            return None, (i, "wrong_room")
        bad = ED.verify(d["authority"], ROOM_DOMAIN + r["body"], r["sig"])
        if bad and "record-signature" not in SKIP:
            return None, (i, bad)
        if out:
            linked = r["prev"] == out[-1]["digest"] and r["effective"] > out[-1]["effective"]
        else:
            linked = r["prev"] == bytes(32) and r["effective"] == 0
        if not linked and "record-chain" not in SKIP:
            return None, (i, "room_usage_unlinked")
        out.append(r)
    return out, None


def terms_of(g):
    """§4.3: None for no terms, the statement, or the string "grant_usage_malformed"."""
    attrs = dict(LRV.read_attrs(g["attrs"]) or [])
    if "ai_usage" not in attrs:
        return None
    try:
        return statement_canonical(attrs["ai_usage"].encode())
    except Refuse:
        return "grant_usage_malformed"


def check_room(case):
    d = LRV.read_descriptor(bytes.fromhex(case["descriptor"]))
    grants, gref = LRV.check_grants(d, [bytes.fromhex(x) for x in case["grants"]])
    records, rref = check_records(d, [bytes.fromhex(x) for x in case["records"]])
    return {"d": d, "grants": grants or [], "grants_refused": gref, "records": records, "records_refused": rref}


def records_at(room, t):
    """§4.2: the record in effect at t, every record where t is None; none when refused."""
    rs = room["records"] or []
    if t is None:
        return rs
    if "record-in-effect" in SKIP:
        return rs[:1]
    at = [r for r in rs if r["effective"] <= t]
    return at[-1:]


def room_outcome(case):
    r = check_room(case)
    terms = []
    for g in r["grants"]:
        t = terms_of(g)
        terms.append({"grant": g["digest"].hex(), "principal": g["principal"].hex(),
                      "terms": None if t is None else t if isinstance(t, str) else t["digest"].hex()})
    ref = lambda e: {"index": e[0], "reason": e[1]}  # noqa: E731
    out = {"grants": ref(r["grants_refused"]) if r["grants_refused"] else "accept",
           "records": ref(r["records_refused"]) if r["records_refused"] else "accept", "terms": terms}
    if r["records"] is not None:
        S = 1000000
        out["in_effect"] = [{"t": t, "statements": [x["statement"]["digest"].hex() for x in records_at(r, t)]}
                            for t in (0, 59 * S, 60 * S, 3600 * S, None)]
    return out


# ---------------------------------------------------------------- §4.4 fields, §4.5 catalog

def check_fields(signed, fields):
    """§4.4: the unsigned vocabularies, or Refuse."""
    lines = lambda n: [v for k, v in fields if k.lower() == n.lower()]  # noqa: E731
    trim = lambda s: s.strip(" \t")  # noqa: E731
    unsigned = []
    usage = lines("Content-Usage")
    if usage:
        if signed is None or "aipref" not in signed["prefs"]:
            unsigned.append("aipref")
        elif "field-meaning" not in SKIP and aipref_prefs(", ".join(usage)) != signed["prefs"]["aipref"]:
            raise Refuse("header_conflicts")
    res, pol = lines("tdm-reservation"), lines("tdm-policy")
    if res or pol:
        if signed is None or "tdmrep" not in signed["prefs"]:
            unsigned.append("tdmrep")
        else:
            want = "1" if signed["prefs"]["tdmrep"][TDM_TERM] == "disallow" else "0"
            if len(res) > 1 or len(pol) > 1:
                raise Refuse("header_conflicts")
            if res and "field-tdm" not in SKIP and trim(res[0]) != want:
                raise Refuse("header_conflicts")
            if pol and trim(pol[0]) != signed["policy"]:
                raise Refuse("header_conflicts")
    return unsigned


def check_catalog(text, objects):
    """§4.5."""
    try:
        e = strict_json(text)
    except ValueError:
        raise Refuse("catalog_conflicts")
    if not isinstance(e, dict) or "wai_ai_usage" not in e:
        return
    try:
        s = read_statement(e["wai_ai_usage"])
    except Refuse:
        raise Refuse("catalog_conflicts")
    if "catalog-compare" not in SKIP and any(o["statement"] and o["statement"]["digest"] != s["digest"] for o in objects):
        raise Refuse("catalog_conflicts")


# ---------------------------------------------------------------- §5 precedence

def governing(o, room):
    """§5.1."""
    out = [o["statement"]] if o["statement"] else []
    if room is not None and o["room"]:
        out += [r["statement"] for r in records_at(room, o["t"])]
    return out


def preference(vocabulary, term, statements, terms):
    """§5.2 for one use; None outside the vocabulary."""
    if term is None:
        return None
    if terms is not None and "terms" not in SKIP:
        p = pref(terms, vocabulary, term)
        if p in ("allow", "disallow"):
            return p
    ps = [pref(s, vocabulary, term) for s in statements]
    if "disallow" in ps and "most-restrictive" not in SKIP:
        return "disallow"
    if "allow" in ps:
        return "allow"
    return "unknown"


def combine(prefs):
    """§5.2's last paragraph; None for a disallow."""
    if "disallow" in prefs:
        return None
    if "allow" in prefs:
        return "allowed"
    if all(p is None for p in prefs):
        return "outside"
    return "unknown"


# ---------------------------------------------------------------- §6 the claim

ENVELOPE = ("kind", "class", "subject_type", "subject", "seq", "issued_at", "body", "signer_pubkey", "signer_id",
            "parent_receipt_hash", "sig")
BODY = ("objects", "usage", "seen", "grant")
B58 = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz"


def did_key(pk):
    """jwp-receipts §7.1: did:key:z and base58btc of 0xed 0x01 and the key."""
    b = b"\xed\x01" + pk
    n = int.from_bytes(b, "big")
    out = ""
    while n:
        n, r = divmod(n, 58)
        out = B58[r] + out
    return "did:key:z" + "1" * (len(b) - len(b.lstrip(b"\x00"))) + out


def is_hex(v, n):
    return isinstance(v, str) and len(v) == 2 * n and all(c in "0123456789abcdef" for c in v)


def is_date_time(s):
    """jwp-receipts §2.4.3's date-time."""
    import re
    m = re.fullmatch(r"([0-9]{4})-([0-9]{2})-([0-9]{2})[Tt]([0-9]{2}):([0-9]{2}):([0-9]{2})(\.[0-9]+)?([Zz]|[+-]([0-9]{2}):([0-9]{2}))", s)
    if not m:
        return False
    y, mo, d, h, mi, se = (int(m.group(i)) for i in range(1, 7))
    leap = y % 4 == 0 and (y % 100 != 0 or y % 400 == 0)
    days = [31, 29 if leap else 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
    if not (1 <= mo <= 12 and 1 <= d <= days[mo - 1] and h <= 23 and mi <= 59 and se <= 60):
        return False
    return m.group(9) is None or (int(m.group(9)) <= 23 and int(m.group(10)) <= 59)


def read_claim(wire):
    """jwp-receipts §3.1 steps 1-4 for a consumption claim: the claim, or Refuse(reason, step)."""
    def at(step, reason):
        return Refuse(reason, step)
    try:
        c = strict_json(wire)
    except ValueError:
        raise at(1, "malformed")
    if not isinstance(c, dict):
        raise at(1, "malformed")
    if "kind" not in c:
        raise at(1, "missing_required_field")
    if c["kind"] != "claim":
        raise at(1, "wrong_kind")
    if any(k not in c for k in ENVELOPE):
        raise at(1, "missing_required_field")
    if any(k not in ENVELOPE for k in c):
        raise at(1, "unknown_member")
    if not isinstance(c["class"], str):
        raise at(1, "field_type")
    if c["class"] == "delivery":
        raise at(1, "class_reserved")
    if c["class"] != "consumption":
        raise at(1, "unknown_class")
    ok = (c["subject_type"] in ("object", "receipt") and is_hex(c["subject"], 32) and is_uint(c["seq"])
          and isinstance(c["issued_at"], str) and is_date_time(c["issued_at"]) and is_hex(c["signer_pubkey"], 32)
          and isinstance(c["signer_id"], str) and (c["parent_receipt_hash"] is None or is_hex(c["parent_receipt_hash"], 32))
          and is_hex(c["sig"], 64))
    if not ok:
        raise at(1, "field_type")
    if c["subject_type"] != "object":
        raise at(1, "subject_type_not_admitted")
    # Step 2: §6.1.
    b = c["body"]
    if not isinstance(b, dict):
        raise at(2, "field_type")
    if any(k not in b for k in BODY):
        raise at(2, "missing_required_field")
    if any(k not in BODY for k in b):
        raise at(2, "unknown_member")
    if not (isinstance(b["objects"], list) and 1 <= len(b["objects"]) <= 1024 and all(is_hex(h, 32) for h in b["objects"])):
        raise at(2, "field_type")
    if not (isinstance(b["usage"], list) and 1 <= len(b["usage"]) <= 3):
        raise at(2, "field_type")
    for e in b["usage"]:
        if not isinstance(e, dict):
            raise at(2, "field_type")
        if "vocabulary" not in e or "term" not in e:
            raise at(2, "missing_required_field")
        if len(e) != 2:
            raise at(2, "unknown_member")
        if e["vocabulary"] not in VOCABULARIES:
            raise at(2, "field_type")
        if not (e["term"] is None or (isinstance(e["term"], str) and 1 <= len(e["term"].encode()) <= 255)):
            raise at(2, "field_type")
    if not (isinstance(b["seen"], list) and len(b["seen"]) <= 64 and all(is_hex(h, 32) for h in b["seen"])):
        raise at(2, "field_type")
    if not (b["grant"] is None or is_hex(b["grant"], 32)):
        raise at(2, "field_type")
    # Step 3.
    unsigned = {k: v for k, v in c.items() if k != "sig"}
    why = ED.verify(bytes.fromhex(c["signer_pubkey"]), CLAIM_SIGN + jcs(unsigned).encode(), bytes.fromhex(c["sig"]))
    if why == "bad_key":
        raise at(3, "bad_signer_key")
    if why and "claim-signature" not in SKIP:
        raise at(3, "signature_verify_failed")
    # Step 4: §6.2, in its order.
    asc = lambda xs: all(x < y for x, y in zip(xs, xs[1:]))  # noqa: E731
    if "claim-order" not in SKIP and not asc(b["objects"]):
        raise at(4, "consumption_objects_unordered")
    if c["subject"] != b["objects"][0]:
        raise at(4, "consumption_subject_mismatch")
    if not asc([e["vocabulary"] for e in b["usage"]]):
        raise at(4, "consumption_usage_unordered")
    for e in b["usage"]:
        t, v = e["term"], e["vocabulary"]
        if t is None:
            continue
        known = (t in LABELS if v == "aipref" else t == TDM_TERM if v == "tdmrep"
                 else t in CAWG_KEYS or (cawg_key(t) and not t.startswith("cawg.")))
        if not known and "claim-terms" not in SKIP:
            raise at(4, "consumption_term_unknown")
    if not asc(b["seen"]):
        raise at(4, "consumption_seen_unordered")
    if "claim-did-key" not in SKIP and c["signer_id"] != did_key(bytes.fromhex(c["signer_pubkey"])):
        raise at(4, "consumption_signer_not_principal")
    return c


def link_hash(c):
    return b3(CLAIM_LINK + jcs(c).encode())


def verify_claim(wire, objects, room, previous_wire):
    """§6.3 after jwp-receipts §3.1 steps 1-5: ("accept", outcomes) or Refuse."""
    c = read_claim(wire)
    # Step 5: the previous claim is held only if it passes steps 1-4.
    prev = None
    if previous_wire is not None:
        try:
            prev = read_claim(previous_wire)
        except Refuse:
            prev = None
    parent = c["parent_receipt_hash"]
    if prev is None and parent is not None and "chain" not in SKIP:
        raise Refuse("parent_not_held", 5)
    if prev is not None:
        if parent is None or prev["signer_pubkey"] != c["signer_pubkey"] or bytes.fromhex(parent) != link_hash(prev) \
                or c["seq"] <= prev["seq"]:
            raise Refuse("chain_break", 5)
    b = c["body"]
    held = []
    for h in b["objects"]:
        if h not in objects:
            raise Refuse("consumption_object_unknown", 6)
        held.append(objects[h])
    terms = None
    if b["grant"] is not None and "grant" not in SKIP:
        g = None if room is None else next((x for x in room["grants"] if x["digest"].hex() == b["grant"]), None)
        good = g is not None and g["principal"].hex() == c["signer_pubkey"] and all(
            o["room"] and (o["t"] is None or LRV.good_at(g, room["d"], o["t"])) for o in held)
        if not good:
            raise Refuse("grant_mismatch", 6)
        t = terms_of(g)
        terms = t if isinstance(t, dict) else None
    rooms = lambda o: room if o["room"] else None  # noqa: E731
    seen = sorted({s["digest"].hex() for o in held for s in governing(o, rooms(o))})
    if "seen" not in SKIP and seen != b["seen"]:
        raise Refuse("preference_mismatch", 6)
    carried = {v for o in held for s in governing(o, rooms(o)) for v in s["prefs"]}
    if "answered" not in SKIP and not carried <= {e["vocabulary"] for e in b["usage"]}:
        raise Refuse("usage_unanswered", 6)
    outcomes = []
    for h, o in zip(b["objects"], held):
        g = governing(o, rooms(o))
        r = combine([preference(e["vocabulary"], e["term"], g, terms) for e in b["usage"]])
        if r is None:
            raise Refuse("usage_disallowed", 6)
        outcomes.append({"object": h, "outcome": r})
    return c, outcomes


# ---------------------------------------------------------------- §7 the gate

def gate(case, keys, results):
    """§7.2: None to serve, or the reason."""
    agent = keys[case["agent"]]
    allowed, disallowing = [], []
    for name in case["claims"]:
        c, r = results[name]
        if c is None or c["signer_pubkey"] != agent:
            continue
        if isinstance(r, list):
            allowed.append((c, r))
        elif r == "usage_disallowed":
            disallowing.append(c)
    w = case["policy"]["window"]
    if w == 0:
        checked = list(case["request"])
    else:
        served = case["served"]
        checked = [h for grp in served[:max(len(served) - w, 0)] for h in grp]
    for h in checked:
        names = lambda c: h in c["body"]["objects"]  # noqa: E731
        if not any(names(c) for c, _ in allowed) and not any(names(c) for c in disallowing):
            return "claim_required"
        if any(names(c) for c in disallowing):
            return "usage_disallowed"
        if case["policy"]["require_allowed"] and "require-allowed" not in SKIP and any(
                o["object"] == h and o["outcome"] != "allowed" for _, r in allowed for o in r):
            return "usage_not_granted"
    return None


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

def held_objects(doc):
    """The held objects of each publication, by content hash."""
    groups = {p["name"]: p for p in doc["publications"]}
    out = {}
    for case in doc["objects"]:
        p = groups[case["publication"]]
        try:
            o = hold(bytes.fromhex(case["envelope"]), case["receipt"], p["group"])
        except Refuse:
            continue
        o["room"] = p["room"] is not None
        out.setdefault(case["publication"], {})[o["hash"].hex()] = o
    return out


def object_named(doc, held, publication, name):
    case = next(o for o in doc["objects"] if o["publication"] == publication and o["name"] == name)
    return held[publication][b3(OBJECT_DOMAIN + bytes.fromhex(case["envelope"])).hex()]


def outcomes(doc):
    """Every case's outcome, by section, from the document's inputs."""
    out = {}
    res = []
    for case in doc["statements"]:
        try:
            r = statement_report(statement_text(case["json"]))
            r["outcome"] = "accept"
        except Refuse as e:
            r = {"outcome": "refuse", "reason": e.reason}
        res.append(r)
    out["statements"] = res
    groups = {p["name"]: p for p in doc["publications"]}
    res = []
    for case in doc["objects"]:
        try:
            o = hold(bytes.fromhex(case["envelope"]), case["receipt"], groups[case["publication"]]["group"])
            res.append({"outcome": "held", "content_hash": o["hash"].hex(),
                        "statement": o["statement"]["digest"].hex() if o["statement"] else None, "t": o["t"]})
        except Refuse as e:
            res.append({"outcome": "refuse", "reason": e.reason})
    out["objects"] = res
    out["rooms"] = [room_outcome(case) for case in doc["rooms"]]
    held = held_objects(doc)
    res = []
    for case in doc["responses"]:
        o = object_named(doc, held, case["publication"], case["object"])
        try:
            res.append({"outcome": "accept", "unsigned": check_fields(o["statement"], [tuple(f) for f in case["fields"]])})
        except Refuse as e:
            res.append({"outcome": "refuse", "reason": e.reason})
    out["responses"] = res
    res = []
    for case in doc["catalogs"]:
        os_ = [object_named(doc, held, case["publication"], n) for n in case["objects"]]
        try:
            check_catalog(case["entry"], os_)
            res.append({"outcome": "accept"})
        except Refuse as e:
            res.append({"outcome": "refuse", "reason": e.reason})
    out["catalogs"] = res
    rooms = {case["name"]: case for case in doc["rooms"]}
    wires, results, res = {}, {}, []
    for case in doc["claims"]:
        ctx = case["context"]
        objects = {}
        for p in ctx["publications"]:
            objects.update(held.get(p, {}))
        room = check_room(rooms[ctx["room"]]) if ctx["room"] else None
        prev = wires[ctx["previous"]] if ctx["previous"] else None
        try:
            c, outs = verify_claim(case["wire"], objects, room, prev)
            res.append({"outcome": "accept", "objects": outs})
            results[case["name"]] = (c, outs)
        except Refuse as e:
            res.append({"outcome": "refuse", "step": e.step, "reason": e.reason})
            claim = None
            if e.step == 6:
                claim = read_claim(case["wire"])
            results[case["name"]] = (claim, e.reason)
        wires[case["name"]] = case["wire"]
    out["claims"] = res
    keys = {k: v for k, v in doc["keys"].items()}
    res = []
    for case in doc["gates"]:
        r = gate(case, keys, results)
        res.append({"outcome": "serve"} if r is None else {"outcome": "refuse", "reason": r})
    out["gates"] = res
    return out


SECTIONS = ("statements", "objects", "rooms", "responses", "catalogs", "claims", "gates")


def check(doc, quiet=False):
    """The failures: every case whose stated outcome is not the one this verifier reaches."""
    fails = []
    try:
        got = outcomes(doc)
    except (Refuse, KeyError, ValueError, TypeError, StopIteration, AttributeError, LEV.Refuse) as e:
        return ["the corpus does not read: %r" % e]
    for s in SECTIONS:
        if len(got[s]) != len(doc[s]):
            fails.append("%s: %d cases, %d outcomes" % (s, len(doc[s]), len(got[s])))
        for case, out in zip(doc[s], got[s]):
            if case.get("outcome") != out:
                fails.append("%s/%s: stated %s, got %s" % (s, case["name"], json.dumps(case.get("outcome"))[:120],
                                                           json.dumps(out)[:120]))
    if not quiet:
        for s in SECTIONS:
            bad = [f for f in fails if f.startswith(s + "/")]
            print(("PASS" if not bad else "FAIL") + " %s: %d cases" % (s, len(doc[s])))
    return fails


def reasons_given(doc):
    seen = set()
    for s in SECTIONS:
        for case in doc[s]:
            o = case["outcome"]
            if s == "rooms":
                for k in ("records", "grants"):
                    if isinstance(o[k], dict):
                        seen.add(o[k]["reason"])
                seen.update(t["terms"] for t in o["terms"] if isinstance(t["terms"], str) and len(t["terms"]) != 64)
            elif "reason" in o:
                seen.add(o["reason"])
    return seen


def flip_hex(h, at):
    """A hex string with one digit changed."""
    c = "0" if h[at] != "0" else "1"
    return h[:at] + c + h[at + 1:]


def mutations(doc):
    """Copies of the corpus, each with one defect a conforming verifier catches."""
    def named(d, s, name):
        return next(c for c in d[s] if c["name"] == name)

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

    def set_(s, name, path, value):
        def f(d):
            x = named(d, s, name)
            for k in path[:-1]:
                x = x[k]
            x[path[-1]] = value
        return f

    def edit(s, name, key, old, new):
        def f(d):
            x = named(d, s, name)
            assert old in x[key], (s, name, old)
            x[key] = x[key].replace(old, new, 1)
        return f

    yield m("a statement's stated preference", set_("statements", "aipref_only", ["outcome", "preferences", "aipref", "train-ai"], "allow"))
    yield m("a statement's value", edit("statements", "aipref_only", "json", "train-ai=n", "train-ai=y"))
    yield m("an unknown revision stated accepted", set_("statements", "aipref_revision_unknown", ["outcome"], {"outcome": "accept"}))
    yield m("an object's payload byte", lambda d: named(d, "objects", "feature").update(
        envelope=flip_hex(named(d, "objects", "feature")["envelope"], len(named(d, "objects", "feature")["envelope"]) - 2)))
    yield m("an object's stated statement", set_("objects", "trailer", ["outcome", "statement"], None))
    yield m("a group receipt's signature", lambda d: d["publications"][0]["group"].update(
        sig=flip_hex(d["publications"][0]["group"]["sig"], 10)))
    yield m("a room record's signature", lambda d: named(d, "rooms", "lecture")["records"].__setitem__(
        1, flip_hex(named(d, "rooms", "lecture")["records"][1], len(named(d, "rooms", "lecture")["records"][1]) - 4)))
    yield m("a room's stated records in effect", lambda d: named(d, "rooms", "lecture").__setitem__(
        "outcome", dict(named(d, "rooms", "lecture")["outcome"], in_effect=[])))
    yield m("a response's field", lambda d: named(d, "responses", "respelled")["fields"].__setitem__(0, ["Content-Usage", "train-ai=y"]))
    yield m("a catalog's statement", edit("catalogs", "agrees", "entry", '"tdm-reservation": 1', '"tdm-reservation": 0'))
    def flip_sig(d):
        x = named(d, "claims", "inference_allowed")
        at = x["wire"].index('"sig":"') + len('"sig":"') + 5
        x["wire"] = flip_hex(x["wire"], at)
    yield m("a claim's signature", flip_sig)
    yield m("a claim's stated object outcome", set_("claims", "unknown_use", ["outcome", "objects", 0, "outcome"], "allowed"))
    yield m("a claim's context room", set_("claims", "room_terms_lift", ["context", "room"], None))
    yield m("a claim's previous", set_("claims", "unknown_use", ["context", "previous"], None))
    yield m("a gate's window", set_("gates", "claim_behind_past_window", ["policy", "window"], 2))
    yield m("a gate's policy", set_("gates", "unknown_not_granted", ["policy", "require_allowed"], False))
    yield m("a gate's claims", lambda d: named(d, "gates", "claim_ahead_disallowed").update(claims=["inference_allowed"]))


CHECKS = {
    "aipref-length": "§3.2: the length of an aipref value",
    "aipref-parse": "§3.3: a value that does not parse is refused",
    "cawg-reserved": "§3.4: the reserved cawg. prefix",
    "tdm-values": "§3.5: tdm-reservation is 0 or 1",
    "revisions": "§3.6 step 2: the revisions",
    "canonical-bytes": "§4.2, §4.3: the bytes are the canonical form",
    "group-signature": "§4.1 step 1: the group receipt's signature",
    "content-hash": "§4.1 step 2: the envelope hashes to its receipt",
    "record-room": "§4.2: a record names the room",
    "record-signature": "§4.2: the authority signs a record",
    "record-chain": "§4.2: records are chained",
    "record-in-effect": "§4.2: the record in effect at an instant",
    "field-meaning": "§4.4: Content-Usage compared by meaning",
    "field-tdm": "§4.4: tdm-reservation compared",
    "catalog-compare": "§4.5: the catalog compared with the objects",
    "terms": "§5.2 rule 1: a grant's terms",
    "most-restrictive": "§5.2 rule 2: the most restrictive",
    "claim-signature": "jwp-receipts §3.1 step 3: the claim's signature",
    "claim-order": "§6.2 rule 1: objects in order",
    "claim-terms": "§6.2 rule 4: the vocabularies' own terms",
    "claim-did-key": "§6.2 rule 6: the signer's did:key",
    "chain": "jwp-receipts §3.1 step 5: the parent is held",
    "grant": "§6.3 step 2: the grant",
    "seen": "§6.3 step 3: the statements seen",
    "answered": "§6.3 step 4: every vocabulary answered",
    "require-allowed": "§7.2 step 3: require allowed",
}


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 []
    print(("PASS" if len(REASONS) == 21 else "FAIL") + " §9 read from the extension: %d reasons" % len(REASONS))
    fails += [] if len(REASONS) == 21 else ["§9 does not read"]
    with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as f:
        doc = json.load(f)
    fails += check(doc)
    extra = sorted(set(os.listdir(corpus_dir)) - {"vectors.json", "README.md"})
    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]
    missing = [r for r in REASONS if r not in reasons_given(doc)]
    print(("PASS" if not missing else "FAIL") + " every reason of §9 appears in the corpus")
    fails += ["no case gives %s" % r for r in missing]
    for label, d in mutations(doc):
        caught = check(d, quiet=True)
        print(("PASS" if caught else "FAIL") + " mutated copy fails: %s%s" % (label, " (%s)" % caught[0][:90] if caught else ""))
        if not caught:
            fails.append("a mutated copy verifies: %s" % label)
    for name, what in CHECKS.items():
        SKIP.clear()
        SKIP.add(name)
        caught = check(doc, quiet=True)
        SKIP.clear()
        print(("PASS" if caught else "FAIL") + " a verifier without %s fails the corpus%s" % (what, " (%s)" % caught[0][:70] if caught else ""))
        if not caught:
            fails.append("the corpus does not catch a verifier without %s" % what)
    for f in fails:
        print("FAIL " + f)
    print(("PASS" if not fails else "FAIL") + " AI usage corpus: %s" % ", ".join("%d %s" % (len(doc[s]), s) for s in SECTIONS))
    return 1 if fails else 0


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