open-standards/wai/edge-conformance/edge_verify.py

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

Written from the extension's text and the texts it cites. It shares no code with wai-rs,
which generates the corpus. From the other corpora's second implementations, each written
from its own specification, it takes BLAKE3 and WAI's strict Ed25519 rule
(claim-conformance/_blake3.py and _ed25519.py), the WIV1 keyframe-0 decode
(int-payload-conformance/int_payload_verify.py), the explode of a WAI2 component
(staged-conformance/staged_verify.py) and the reading of a key-log revision
(live-conformance/live_rooms_verify.py). It implements the rest itself:

- the record's reading under §5.3's rules in their order, its canonical form (RFC 8785, on
  the values a record carries), its signature and its hash (§5.2);
- the JWP group and object receipts the records name (jwp-receipts §2.2-§2.4.2): their
  payloads, link hashes, labels and inclusion proofs;
- reading an input as samples (§4.1): packaging §5.7's still reader, with the chunk rule;
- the operations of §4.2, the pinned manifest of §4.3 and packaging §5.7's still writer, so
  that every replicate hop is re-executed and compared byte for byte;
- verification of a chain (§7.2-§7.4) with the authority of §8, and the facts it states.

It reads §9's reasons and §4's derivations from the extension, requires every vector's
outcome, reason, place and facts, requires every reason of §9 to appear, mutates copies of
the corpus and requires every copy to fail, and drops each of its own checks in turn and
requires the corpus to catch the weakened verifier.

    python3 -I edge_verify.py <edge-conformance dir>

Exit status 0 only if all of that holds. Standard library only.
"""
import copy
import hashlib
import json
import os
import re
import struct
import sys
import unicodedata
import zlib

HERE = os.path.dirname(os.path.abspath(__file__))
WAI = os.path.normpath(os.path.join(HERE, ".."))
for d in ("claim-conformance", "int-payload-conformance", "staged-conformance", "live-conformance"):
    sys.path.insert(0, os.path.join(WAI, d))
from _blake3 import blake3  # noqa: E402
import _ed25519 as ED  # noqa: E402
import int_payload_verify as IPV  # noqa: E402
import staged_verify as SV  # noqa: E402
import live_rooms_verify as LRV  # noqa: E402

EXTENSION = os.path.join(WAI, "extensions", "deriving-edge.md")
MAX_INT = 2 ** 53 - 1
PNG, JPEG, INT_MOTION = "wai.image.png", "wai.image.jpeg", "wai.video.int_motion"
EXPLODE = "wai-edge:explode"
AUTHORITY = "edge/derive"
MAX_INPUTS, MAX_HOPS, MAX_SAMPLE_BYTES = 16, 8, 1 << 26
RECORD_MEMBERS = {"kind", "derivation", "params", "encoder", "check", "inputs", "output", "energy", "provenance",
                  "synthesis", "issued_at", "signer_pubkey", "signer_id", "sig"}
INPUT_MEMBERS = {"content_hash", "capability", "receipt", "object_id", "frame", "samples", "derivation", "synthesis"}
ITEM_MEMBERS = {"input", "kind", "digest", "compartment"}
MAX_PROVENANCE = 1024
LABELS = {None: 0, "assist": 1, "generate": 2}  # §6.2, in rising order
OUTPUT_MEMBERS = {"content_hash", "capability", "receipt", "object_id", "width", "height", "samples"}
ENERGY_MEMBERS = {"joules_micro", "energy_provenance", "energy_uncertainty", "energy_counter_filtering",
                  "energy_calibration_ref"}
CLASS_ORDER = ["CalibratedInstrument", "OnChipCounter", "HwShunt", "ModelBased", "Estimator"]
RESOLUTION_FLOOR = 288676  # energy-measurement §2.3: joules_micro x relative_ppm, at least


# ---------------------------------------------------------------- the extension's tables

def section(text, heading):
    at = text.index(heading)
    end = text.find("\n## ", at + 1)
    return text[at:] if end < 0 else text[at:end]


def first_cells(s):
    return [line.split("`")[1] for line in s.splitlines() if line.startswith("| `")]


with open(EXTENSION, encoding="utf-8") as _f:
    _TEXT = _f.read()
REASONS = first_cells(section(_TEXT, "\n## 9. Reasons"))
_S4 = section(_TEXT, "\n## 4. The derivations")
DERIVATIONS = first_cells(_S4[:_S4.index("\n### ")])


# ---------------------------------------------------------------- refusals and checks

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


# Every check this verifier makes, by name. The self-test drops each in turn.
CHECKS = (
    "shape", "registered", "params", "writer", "input-count", "input-form", "check", "energy", "key", "signature",
    "output-held", "output-receipts", "output-signer", "output-energy", "output-hash", "output-manifest",
    "output-samples", "input-held", "input-receipts", "source", "depth", "hop-held", "hop-match", "input-hash",
    "input-reads", "input-samples", "provenance-held", "provenance-hop", "synthesis-hop", "synthesis-record",
    "evaluable", "replicate", "attested-samples", "inputs-held", "head-held",
    "authorised", "not-revoked", "authority-required", "attested",
)
DROPPED = set()


def need(check, ok, reason, at):
    assert check in CHECKS, check
    if not ok and check not in DROPPED:
        raise Refuse(reason, at)


# ---------------------------------------------------------------- JSON and its canonical form

class Outside:
    """A number that is not an unsigned integer written plainly: outside the record's subset."""


OUTSIDE = Outside()


def strict(text):
    """jwp-receipts §2.4.3's reading: RFC 8259, no member named twice, no unpaired surrogate,
    nesting to 32 levels; a number outside the subset kept as OUTSIDE. ValueError otherwise."""
    def pairs(ps):
        d = {}
        for k, v in ps:
            if k in d:
                raise ValueError("a member named twice")
            d[k] = v
        return d

    def integer(s):
        return OUTSIDE if s.startswith("-") or len(s) > 20 else int(s)

    def constant(c):
        raise ValueError(c)

    def tree(v, level):
        if isinstance(v, str):
            if any("\ud800" <= ch <= "\udfff" for ch in v):
                raise ValueError("an unpaired surrogate")
        elif isinstance(v, (list, dict)):
            if level + 1 > 32:
                raise ValueError("nested too deeply")
            for k in (v if isinstance(v, dict) else []):
                tree(k, level)
            for x in (v.values() if isinstance(v, dict) else v):
                tree(x, level + 1)

    try:
        v = json.loads(text, object_pairs_hook=pairs, parse_float=lambda s: OUTSIDE, parse_int=integer,
                       parse_constant=constant)
    except RecursionError:
        raise ValueError("nested too deeply")
    tree(v, 0)
    return v


def jcs(v):
    """RFC 8785 on strings, unsigned integers to 2^53 - 1, booleans, null, arrays, 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):
        if not 0 <= v <= MAX_INT:
            raise ValueError("a number 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")


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


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


def is_term(s):
    return (isinstance(s, str) and s != "" and len(s.encode()) <= 2048
            and not any(c.isspace() or unicodedata.category(c) == "Cc" for c in s))


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 is_date_time(s):
    """RFC 3339 §5.6 with §5.7's days, ASCII digits, a second of 60 allowed."""
    m = isinstance(s, str) and re.fullmatch(r"([0-9]{4})-([0-9]{2})-([0-9]{2})[Tt]([0-9]{2}):([0-9]{2}):([0-9]{2})"
                                            r"(\.[0-9]+)?([Zz]|[+-]([0-9]{2}):([0-9]{2}))", s)
    if not m:
        return False
    y, mo, d, hh, mi, ss = (int(m.group(i)) for i in range(1, 7))
    if not (1 <= mo <= 12 and 1 <= d <= days_in(y, mo) and hh <= 23 and mi <= 59 and ss <= 60):
        return False
    return m.group(9) is None or (int(m.group(9)) <= 23 and int(m.group(10)) <= 59)


def in_subset(v):
    if v is None or isinstance(v, str):
        return True
    if isinstance(v, bool) or isinstance(v, Outside):
        return False
    if isinstance(v, int):
        return is_int(v)
    if isinstance(v, list):
        return all(in_subset(x) for x in v)
    if isinstance(v, dict):
        return all(in_subset(x) for x in v.values())
    return False


_SIGS = {}


def ed_verify(pk, msg, sig):
    """WAI's strict Ed25519 rule (_ed25519.verify), remembered: the self-test re-verifies 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 content_hash(env):
    return blake3(b"moq-jwp:object\x01" + env)


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


# ---------------------------------------------------------------- energy classes (energy-measurement §2.7)

TAG = {"HwShunt": 1, "ModelBased": 2, "Estimator": 3, "OnChipCounter": 4, "CalibratedInstrument": 5}
FILTERING = {"Unknown": 0, "Off": 1, "On": 2}


def read_class(m, filtering_written):
    """The class a set of `energy_*` members names, or None for none; ValueError where they do
    not make one. With `filtering_written`, an on-chip counter's filtering state must be written
    (a record writes it, as a claim does); a receipt's label may leave it out, as Unknown."""
    label = m.get("energy_provenance")
    u, f, r = m.get("energy_uncertainty"), m.get("energy_counter_filtering"), m.get("energy_calibration_ref")
    if label is None:
        if u is not None or f is not None or r is not None:
            raise ValueError("evidence without a class")
        return None
    if label not in TAG:
        raise ValueError("an unknown class")
    if label in ("HwShunt", "ModelBased", "Estimator"):
        if u is not None or f is not None or r is not None:
            raise ValueError("evidence beside a class that carries none")
        return {"label": label}
    if not isinstance(u, dict) or set(u) != {"relative_ppm", "window_us"} or not is_int(u["relative_ppm"]) \
            or not is_int(u["window_us"]) or not 1 <= u["relative_ppm"] <= 2 ** 32 - 1 or u["window_us"] == 0:
        raise ValueError("no declared uncertainty")
    if label == "OnChipCounter":
        if r is not None or (f is None and filtering_written):
            raise ValueError("an on-chip counter's evidence")
        f = "Unknown" if f is None else f
        if f not in FILTERING:
            raise ValueError("a filtering state")
        return {"label": label, "ppm": u["relative_ppm"], "window": u["window_us"], "filtering": f}
    if f is not None or not isinstance(r, str) or r.strip() == "":
        raise ValueError("a calibrated instrument's evidence")
    return {"label": label, "ppm": u["relative_ppm"], "window": u["window_us"], "ref": r}


def encode_class(c):
    if c is None:
        return b"\x00"
    b = bytes([TAG[c["label"]]])
    if c["label"] == "OnChipCounter":
        b += struct.pack(">IQB", c["ppm"], c["window"], FILTERING[c["filtering"]])
    elif c["label"] == "CalibratedInstrument":
        ref = c["ref"].encode()
        b += struct.pack(">IQI", c["ppm"], c["window"], len(ref)) + ref
    return b


def label_ok(figure, c):
    """A class beside a figure (energy-measurement §2.8): never HwShunt, never beside a zero
    figure, at the resolution floor where it declares an uncertainty."""
    return c["label"] != "HwShunt" and figure != 0 and ("ppm" not in c or figure * c["ppm"] >= RESOLUTION_FLOOR)


# ---------------------------------------------------------------- JWP receipts (jwp-receipts §2.2-§2.4.2)

GROUP = {"kind", "track_alias", "group_id", "root_hash", "leaf_count", "joules_total", "group_start", "group_end",
         "signer_pubkey", "signer_id", "sig", "parent_receipt_hash"}
GROUP_LABEL = {"sign_version", "energy_provenance", "energy_uncertainty", "energy_counter_filtering",
               "energy_calibration_ref", "energy_coverage"}
OBJECT = {"kind", "track_alias", "group_id", "object_id", "content_hash", "joules_micro", "origin", "merkle_proof"}
OBJECT_LABEL = {"leaf_version", "energy_provenance", "energy_uncertainty", "energy_counter_filtering",
                "energy_calibration_ref"}


def u64(v):
    return isinstance(v, int) and not isinstance(v, bool) and 0 <= v < 2 ** 64


def read_group(g):
    """§2.4.1's group receipt, or None where it does not read (the verifier does not hold it)."""
    if not isinstance(g, dict) or g.get("kind") != "group" or not GROUP <= set(g) or not set(g) <= GROUP | GROUP_LABEL:
        return None
    if not (all(u64(g[k]) for k in ("track_alias", "group_id", "leaf_count", "joules_total"))
            and all(isinstance(g[k], str) for k in ("group_start", "group_end", "signer_id"))
            and is_hex(g["root_hash"], 32) and is_hex(g["signer_pubkey"], 32) and is_hex(g["sig"], 64)
            and (g["parent_receipt_hash"] is None or is_hex(g["parent_receipt_hash"], 32))):
        return None
    out = dict(g, labelled=False)
    if g.get("sign_version") is None:
        return None if any(g.get(k) is not None for k in GROUP_LABEL) else out
    if g["sign_version"] != 2 or isinstance(g["sign_version"], bool):
        return None
    try:
        c = read_class(g, False)
    except ValueError:
        return None
    cov = g.get("energy_coverage")
    if c is None and cov is not None or c is not None and cov not in ("Whole", "Partial"):
        return None
    out.update(labelled=True, cls=c, coverage=cov or "Whole")
    return out


def group_payload(g):
    b = (b"moq-jwp:group-sign\x02" if g["labelled"] else b"moq-jwp:group-sign\x01")
    b += struct.pack(">QQ", g["track_alias"], g["group_id"]) + bytes.fromhex(g["root_hash"])
    b += struct.pack(">QQ", g["leaf_count"], g["joules_total"])
    for s in (g["group_start"], g["group_end"]):
        b += struct.pack(">I", len(s.encode())) + s.encode()
    b += bytes.fromhex(g["signer_pubkey"]) + struct.pack(">I", len(g["signer_id"].encode())) + g["signer_id"].encode()
    if g["labelled"]:
        b += encode_class(g["cls"]) + (b"\x01" if g["cls"] is not None and g["coverage"] == "Partial" else b"\x00")
        p = g["parent_receipt_hash"]
        b += b"\x01" + bytes.fromhex(p) if p is not None else b"\x00" + bytes(32)
    return b


def link(g):
    return blake3(b"moq-jwp:receipt-link\x01" + group_payload(g) + bytes.fromhex(g["sig"]))


def group_verifies(g):
    if ed_verify(bytes.fromhex(g["signer_pubkey"]), group_payload(g), bytes.fromhex(g["sig"])) is not None:
        return False
    if g["labelled"]:
        c, t = g["cls"], g["joules_total"]
        return (t == 0 and g["coverage"] == "Whole") if c is None else label_ok(t, c)
    return True


def read_object(o):
    if not isinstance(o, dict) or o.get("kind") != "object" or not OBJECT <= set(o) or not set(o) <= OBJECT | OBJECT_LABEL:
        return None
    p = o["merkle_proof"]
    if not (all(u64(o[k]) for k in ("track_alias", "group_id", "object_id", "joules_micro"))
            and is_hex(o["content_hash"], 32) and isinstance(o["origin"], str) and isinstance(p, dict)
            and set(p) == {"siblings", "direction_bits", "leaf_index"} and u64(p["direction_bits"])
            and u64(p["leaf_index"]) and isinstance(p["siblings"], list) and all(is_hex(s, 32) for s in p["siblings"])):
        return None
    r = dict(o, labelled=False, cls=None, unreadable=False)
    lv = o.get("leaf_version")
    if lv is None:
        r["unreadable"] = any(o.get(k) is not None for k in OBJECT_LABEL)
    elif lv == 2 and not isinstance(lv, bool):
        try:
            r.update(labelled=True, cls=read_class(o, False))
        except ValueError:
            r["unreadable"] = True
    else:
        r["unreadable"] = True
    return r


def leaf(o):
    h = bytes.fromhex(o["content_hash"])
    if not o["labelled"]:
        return blake3(b"moq-jwp:merkle-leaf\x01" + h)
    origin = o["origin"].encode()
    return blake3(b"moq-jwp:merkle-leaf\x02" + h + struct.pack(">Q", o["joules_micro"]) + encode_class(o["cls"])
                  + struct.pack(">I", len(origin)) + origin)


def included(o, g):
    """§3 step 5: the label, the position, the walk to the group's root."""
    if o["unreadable"]:
        return False
    if o["labelled"] and ((o["cls"] is None and o["joules_micro"] != 0) or (o["cls"] is not None and not label_ok(o["joules_micro"], o["cls"]))):
        return False
    n, p = g["leaf_count"], o["merkle_proof"]
    depth = 0 if n <= 1 else (n - 1).bit_length()
    if (o["track_alias"] != g["track_alias"] or o["group_id"] != g["group_id"] or o["object_id"] != p["leaf_index"]
            or not p["leaf_index"] < n or len(p["siblings"]) != depth or p["direction_bits"] != p["leaf_index"]):
        return False
    cur = leaf(o)
    for i, s in enumerate(p["siblings"]):
        s = bytes.fromhex(s)
        cur = blake3(b"moq-jwp:merkle-node\x01" + (s + cur if (p["direction_bits"] >> i) & 1 else cur + s))
    return cur == bytes.fromhex(g["root_hash"])


# ---------------------------------------------------------------- PNG (packaging §5.7)

SIGNATURE = b"\x89PNG\r\n\x1a\n"
LEN_BASE = [3, 4, 5, 6, 7, 8, 9, 10, 11, 13, 15, 17, 19, 23, 27, 31, 35, 43, 51, 59, 67, 83, 99, 115, 131, 163, 195, 227, 258]
LEN_EXTRA = [0] * 8 + [1] * 4 + [2] * 4 + [3] * 4 + [4] * 4 + [5] * 4 + [0]
DIST_BASE = [1, 2, 3, 4, 5, 7, 9, 13, 17, 25, 33, 49, 65, 97, 129, 193, 257, 385, 513, 769, 1025, 1537, 2049, 3073,
             4097, 6145, 8193, 12289, 16385, 24577]
DIST_EXTRA = [0, 0, 0, 0] + [i // 2 for i in range(2, 28)]


class Bits:
    """Deflate's packing: values least significant bit first, Huffman codes most significant first."""
    def __init__(self):
        self.out, self.acc, self.n = bytearray(), 0, 0

    def put(self, v, n):
        self.acc |= v << self.n
        self.n += n
        while self.n >= 8:
            self.out.append(self.acc & 255)
            self.acc >>= 8
            self.n -= 8

    def code(self, c, n):
        self.put(int(format(c, "0%db" % n)[::-1], 2) if n else 0, n)

    def lit(self, s):
        if s <= 143:
            self.code(0x30 + s, 8)
        elif s <= 255:
            self.code(0x190 + s - 144, 9)
        elif s <= 279:
            self.code(s - 256, 7)
        else:
            self.code(0xC0 + s - 280, 8)


def deflate_fixed(d):
    """One final block of the fixed codes; at each position with three bytes left, the latest
    earlier position within 32768 bytes that starts with the same three bytes is the match, as
    far as the runs agree (258 at most); otherwise a literal. Every position a symbol covers with
    three bytes left becomes its three bytes' latest position."""
    b = Bits()
    b.put(1, 1)
    b.put(1, 2)
    latest, n, i = {}, len(d), 0
    while i < n:
        j = latest.get(d[i:i + 3]) if i + 3 <= n else None
        if j is not None and i - j <= 32768:
            m = 0
            while m < min(258, n - i) and d[j + m] == d[i + m]:
                m += 1
            li = max(k for k, x in enumerate(LEN_BASE) if x <= m)
            b.lit(257 + li)
            b.put(m - LEN_BASE[li], LEN_EXTRA[li])
            di = max(k for k, x in enumerate(DIST_BASE) if x <= i - j)
            b.code(di, 5)
            b.put(i - j - DIST_BASE[di], DIST_EXTRA[di])
        else:
            b.lit(d[i])
            m = 1
        for q in range(i, i + m):
            if q + 3 <= n:
                latest[d[q:q + 3]] = q
        i += m
    b.lit(256)
    return bytes(b.out) + (bytes([b.acc]) if b.n else b"")


def chunk(kind, data):
    return struct.pack(">I", len(data)) + kind + data + struct.pack(">I", zlib.crc32(kind + data))


def png_still(s):
    """packaging §5.7's writer: IHDR of 8-bit RGB, one IDAT of the rows filtered Sub under the
    deflate above, IEND."""
    w, h, rgb = s
    raw = bytearray()
    for y in range(h):
        row = rgb[3 * w * y:3 * w * (y + 1)]
        raw.append(1)
        raw += bytes((row[i] - (row[i - 3] if i >= 3 else 0)) & 255 for i in range(3 * w))
    z = b"\x78\x01" + deflate_fixed(bytes(raw)) + struct.pack(">I", zlib.adler32(bytes(raw)))
    return SIGNATURE + chunk(b"IHDR", struct.pack(">IIBBBBB", w, h, 8, 2, 0, 0, 0)) + chunk(b"IDAT", z) + chunk(b"IEND", b"")


def png_read(b):
    """packaging §5.7's reader of an 8-bit RGB PNG without interlace: (w, h, rgb, the chunk
    types in order); ValueError where it does not read."""
    if b[:8] != SIGNATURE:
        raise ValueError("not a PNG")
    at, ihdr, z, ended, plte, idat, kinds = 8, None, b"", False, False, 0, []
    while at < len(b):
        if ended or at + 12 > len(b):
            raise ValueError("bytes after IEND, or a chunk past the end")
        n = struct.unpack_from(">I", b, at)[0]
        kind, data = b[at + 4:at + 8], b[at + 8:at + 8 + n]
        if at + 12 + n > len(b) or not all(65 <= c <= 90 or 97 <= c <= 122 for c in kind):
            raise ValueError("a chunk")
        if zlib.crc32(kind + data) != struct.unpack_from(">I", b, at + 8 + n)[0]:
            raise ValueError("a CRC")
        kinds.append(kind)
        if idat == 1 and kind != b"IDAT":
            idat = 2
        if kind == b"IHDR":
            if ihdr is not None or at != 8 or n != 13:
                raise ValueError("IHDR")
            w, h = struct.unpack(">II", data[:8])
            if data[8:] != bytes([8, 2, 0, 0, 0]) or w == 0 or h == 0 or 3 * w * h > 1 << 26:
                raise ValueError("not 8-bit RGB")
            ihdr = (w, h)
        elif ihdr is None:
            raise ValueError("a chunk before IHDR")
        elif kind == b"PLTE":
            if plte or idat or n == 0 or n % 3 or n > 768:
                raise ValueError("PLTE")
            plte = True
        elif kind == b"IDAT":
            if idat == 2:
                raise ValueError("IDATs not consecutive")
            idat, z = 1, z + data
        elif kind == b"IEND":
            if n:
                raise ValueError("IEND with data")
            ended = True
        elif not kind[0] & 0x20:
            raise ValueError("an unknown critical chunk")
        at += 12 + n
    if ihdr is None or not ended or not z or z[0] >> 4 > 7:
        raise ValueError("no IHDR, no IEND, or no stream")
    w, h = ihdr
    d = zlib.decompressobj()
    try:
        raw = d.decompress(z, (3 * w + 1) * h + 1)
    except zlib.error as e:
        raise ValueError(str(e))
    if not d.eof or d.unused_data or d.unconsumed_tail or len(raw) != (3 * w + 1) * h:
        raise ValueError("the rows")
    stride, out = 3 * w, bytearray(3 * w * h)
    for y in range(h):
        f, row = raw[y * (stride + 1)], raw[y * (stride + 1) + 1:(y + 1) * (stride + 1)]
        if f > 4:
            raise ValueError("a filter")
        for i in range(stride):
            a = out[y * stride + i - 3] if i >= 3 else 0
            up = out[(y - 1) * stride + i] if y else 0
            c = out[(y - 1) * stride + i - 3] if y and i >= 3 else 0
            if f == 4:
                p = a + up - c
                pr = a if abs(p - a) <= abs(p - up) and abs(p - a) <= abs(p - c) else (up if abs(p - up) <= abs(p - c) else c)
            else:
                pr = (0, a, up, (a + up) // 2)[f]
            out[y * stride + i] = (row[i] + pr) & 255
    return w, h, bytes(out), kinds


# ---------------------------------------------------------------- envelopes and inputs (§4.1, §4.4)

def split_wai1(env):
    """A WAI1 envelope's manifest and payload, exactly one envelope; None otherwise."""
    if len(env) < 12 or env[:4] != b"WAI1":
        return None
    ml = struct.unpack_from("<I", env, 4)[0]
    if 8 + ml + 4 > len(env):
        return None
    pl = struct.unpack_from("<I", env, 8 + ml)[0]
    if 12 + ml + pl != len(env):
        return None
    return env[8:8 + ml], env[12 + ml:]


def capability_of(env):
    parts = split_wai1(env)
    if parts is None:
        return None
    try:
        return json.loads(parts[0])["model_requirement"]["capability"]
    except (ValueError, KeyError, TypeError):
        return None


def read_samples(cap, env, frame):
    """§4.1: (w, h, rgb), or None where the input does not read as stated."""
    parts = split_wai1(env)
    if parts is None or capability_of(env) != cap:
        return None
    payload = parts[1]
    if cap == PNG:
        try:
            w, h, rgb, kinds = png_read(payload)
        except ValueError:
            return None
        if frame != 0 or any(k not in (b"IHDR", b"IDAT", b"IEND", b"caBX") for k in kinds):
            return None
        return w, h, rgb
    if cap == INT_MOTION:
        try:
            v = IPV.parse_wiv1(payload, want_keyframe=0)
            frames = IPV.decode_wiv1_k0(v)
        except IPV.Refused:
            return None
        return (v["w"], v["h"], frames[frame]) if frame < len(frames) else None
    return None


def png_chunks(p):
    """A PNG's chunks as (type, data), walked by their lengths alone; None where the signature
    is not PNG's or a chunk runs past the end."""
    if p[:8] != SIGNATURE:
        return None
    at, out = 8, []
    while at < len(p):
        if at + 8 > len(p):
            return None
        n = struct.unpack_from(">I", p, at)[0]
        if at + 12 + n > len(p):
            return None
        out.append((p[at + 4:at + 8], p[at + 8:at + 8 + n]))
        at += 12 + n
    return out


def embedded_stores(cap, env):
    """§6.1: the C2PA Manifest Stores embedded in an input: for wai.image.png, the data of each
    caBX chunk (C2PA 2.4 §A.3.2)."""
    parts = split_wai1(env)
    if parts is None or cap != PNG:
        return []
    chunks = png_chunks(parts[1])
    return [] if chunks is None else [d for k, d in chunks if k == b"caBX"]


def claim_about(text):
    """§6.1: (subject, link hash) of a claim about an object, or None."""
    try:
        v = strict(text)
        if not isinstance(v, dict) or v.get("kind") != "claim" or v.get("subject_type") != "object" \
                or not is_hex(v.get("subject"), 32):
            return None
        return v["subject"], blake3(b"moq-jwp:claim-link\x01" + jcs(v).encode()).hex()
    except (ValueError, TypeError):
        return None


def explode(env, component, cap):
    """§4.4: the exploded envelope of entry `component`, or None."""
    try:
        _, entries, _ = SV.read_wai2(env)
        out = SV.explode_each(env)
    except Exception:  # anything that is not a WAI2 the staged-delivery reader reads
        return None
    if component >= len(entries) or entries[component].get("capability") != cap:
        return None
    return out[component]


# ---------------------------------------------------------------- the operations (§4.2)

def px(s, x, y):
    w, _, rgb = s
    i = 3 * (y * w + x)
    return rgb[i:i + 3]


def size_ok(w, h):
    return w >= 1 and h >= 1 and w * h * 3 <= MAX_SAMPLE_BYTES


def evaluate(d, p, ins):
    """The output samples (w, h, rgb), or None where the inputs cannot satisfy the parameters.
    Scaling is not pinned and is never evaluated here."""
    def build(w, h, f):
        return w, h, b"".join(f(i, j) for j in range(h) for i in range(w))
    if d == "wai.derive.identity":
        return ins[0]
    if d in ("wai.derive.crop", "wai.derive.roi_selection"):
        s = ins[0]
        if p["x"] + p["w"] > s[0] or p["y"] + p["h"] > s[1]:
            return None
        return build(p["w"], p["h"], lambda i, j: px(s, p["x"] + i, p["y"] + j))
    if d == "wai.derive.rotation":
        s, q = ins[0], p["quarter_turns"]
        W, H = s[0], s[1]
        if q == 1:
            return build(H, W, lambda i, j: px(s, j, H - 1 - i))
        if q == 2:
            return build(W, H, lambda i, j: px(s, W - 1 - i, H - 1 - j))
        return build(H, W, lambda i, j: px(s, W - 1 - j, i))
    if d == "wai.derive.mirror":
        s = ins[0]
        if p["flip"] == "left-right":
            return build(s[0], s[1], lambda i, j: px(s, s[0] - 1 - i, j))
        return build(s[0], s[1], lambda i, j: px(s, i, s[1] - 1 - j))
    if d == "wai.derive.grid_composition":
        cw, ch = ins[0][0], ins[0][1]
        if any((s[0], s[1]) != (cw, ch) for s in ins) or len(ins) != p["columns"] * p["rows"]:
            return None
        if not size_ok(p["columns"] * cw, p["rows"] * ch):
            return None
        return build(p["columns"] * cw, p["rows"] * ch, lambda i, j: px(ins[(j // ch) * p["columns"] + i // cw], i % cw, j % ch))
    if d == "wai.derive.overlay_composition":
        W, H = p["width"], p["height"]
        canvas = bytearray(bytes.fromhex(p["fill"]) * (W * H))
        for s, at in zip(ins, p["at"]):
            for j in range(s[1]):
                for i in range(s[0]):
                    x, y = at["x"] + i, at["y"] + j
                    if x < W and y < H:
                        canvas[3 * (y * W + x):3 * (y * W + x) + 3] = px(s, i, j)
        return W, H, bytes(canvas)
    raise AssertionError("not an exact operation: " + d)


def exact(d):
    """§4.2's closing paragraph: every operation but scaling is exact; the explode evaluates none."""
    return d != "wai.derive.scaling"


def check_of(d, out_cap, encoder):
    """§4.5."""
    if d == EXPLODE:
        return "replicate"
    return "replicate" if exact(d) and out_cap == PNG and encoder is None else "attest"


# ---------------------------------------------------------------- the record (§5)

def record_hash(raw):
    return blake3(b"wai:derivation-link\x01" + jcs(raw).encode())


def held_hash(text):
    try:
        return record_hash(strict(text))
    except (ValueError, TypeError):
        return None


def params_ok(d, p, n):
    """§5.3 rule 3's parameter rules."""
    def ints(*ks):
        return set(p) == set(ks) and all(is_int(p[k]) for k in ks)
    if d == "wai.derive.identity":
        return p == {}
    if d in ("wai.derive.crop", "wai.derive.roi_selection"):
        return ints("x", "y", "w", "h") and size_ok(p["w"], p["h"])
    if d == "wai.derive.rotation":
        return ints("quarter_turns") and p["quarter_turns"] in (1, 2, 3)
    if d == "wai.derive.mirror":
        return set(p) == {"flip"} and p["flip"] in ("left-right", "top-bottom")
    if d == "wai.derive.grid_composition":
        return ints("columns", "rows") and p["columns"] >= 1 and p["rows"] >= 1
    if d == "wai.derive.overlay_composition":
        return (set(p) == {"width", "height", "fill", "at"} and is_int(p["width"]) and is_int(p["height"])
                and size_ok(p["width"], p["height"]) and is_hex(p["fill"], 3) and isinstance(p["at"], list)
                and len(p["at"]) == n and all(isinstance(a, dict) and set(a) == {"x", "y"} and is_int(a["x"]) and is_int(a["y"]) for a in p["at"]))
    if d == "wai.derive.scaling":
        return ints("width", "height") and size_ok(p["width"], p["height"])
    if d == EXPLODE:
        return ints("component")
    return False


def read_record(text, r):
    """§5.3, rule by rule: the record as a dict, or Refuse at ["record", r]."""
    at = ["record", r]
    try:
        v = strict(text)
    except (ValueError, TypeError):
        raise Refuse("record_malformed", at)

    def shaped():
        if not isinstance(v, dict) or set(v) != RECORD_MEMBERS or v["kind"] != "wai-derivation":
            return False
        if not is_term(v["derivation"]) or not isinstance(v["params"], dict) or not in_subset(v["params"]):
            return False
        e = v["encoder"]
        if e is not None and not (isinstance(e, dict) and set(e) == {"id", "settings"} and is_term(e["id"])
                                  and isinstance(e["settings"], dict) and in_subset(e["settings"])):
            return False
        if v["check"] not in ("replicate", "attest"):
            return False
        ins = v["inputs"]
        if not isinstance(ins, list) or not 1 <= len(ins) <= MAX_INPUTS:
            return False
        for i in ins:
            if not (isinstance(i, dict) and set(i) == INPUT_MEMBERS and is_hex(i["content_hash"], 32)
                    and is_term(i["capability"]) and is_hex(i["receipt"], 32) and is_int(i["object_id"])
                    and (i["frame"] is None or is_int(i["frame"])) and (i["samples"] is None or is_hex(i["samples"], 32))
                    and (i["derivation"] is None or is_hex(i["derivation"], 32))):
                return False
        o = v["output"]
        if not (isinstance(o, dict) and set(o) == OUTPUT_MEMBERS and is_hex(o["content_hash"], 32)
                and is_term(o["capability"]) and is_hex(o["receipt"], 32) and is_int(o["object_id"])
                and all(o[k] is None or is_int(o[k]) for k in ("width", "height"))
                and (o["samples"] is None or is_hex(o["samples"], 32))):
            return False
        en = v["energy"]
        if en is not None:
            u = en.get("energy_uncertainty") if isinstance(en, dict) else None
            if not (isinstance(en, dict) and set(en) == ENERGY_MEMBERS and is_int(en["joules_micro"])
                    and isinstance(en["energy_provenance"], str)
                    and (u is None or (isinstance(u, dict) and set(u) == {"relative_ppm", "window_us"}
                                       and is_int(u["relative_ppm"]) and is_int(u["window_us"])))
                    and all(en[k] is None or isinstance(en[k], str) for k in ("energy_counter_filtering", "energy_calibration_ref"))):
                return False
        for i in ins:
            if i["synthesis"] not in LABELS:
                return False
        if v["synthesis"] not in LABELS:
            return False
        items = v["provenance"]
        if not isinstance(items, list) or len(items) > MAX_PROVENANCE:
            return False
        keys = []
        for it in items:
            if not (isinstance(it, dict) and set(it) == ITEM_MEMBERS and is_int(it["input"]) and it["input"] < len(ins)
                    and it["kind"] in ("claim", "c2pa") and is_hex(it["digest"], 32)
                    and it["compartment"] in ("system", "personal")):
                return False
            keys.append((it["input"], it["kind"], it["digest"]))
        if any(a >= b for a, b in zip(keys, keys[1:])):
            return False
        return (is_date_time(v["issued_at"]) and is_hex(v["signer_pubkey"], 32) and isinstance(v["signer_id"], str)
                and is_hex(v["sig"], 64))
    need("shape", shaped(), "record_malformed", at)
    d = v["derivation"]
    need("registered", d in DERIVATIONS, "derivation_unknown", at)
    ins, out, enc = v["inputs"], v["output"], v["encoder"]
    evaluates = d != EXPLODE
    need("params", params_ok(d, v["params"], len(ins)), "params_invalid", at)
    if evaluates:
        need("writer", out["capability"] in (PNG, JPEG) and not (out["capability"] == JPEG and enc is None), "params_invalid", at)
    else:
        need("writer", enc is None and out["capability"] == ins[0]["capability"], "params_invalid", at)
    p = v["params"]
    if d == "wai.derive.grid_composition":
        fits = len(ins) == p["columns"] * p["rows"]
    elif d == "wai.derive.overlay_composition":
        fits = 1 <= len(ins) <= MAX_INPUTS
    else:
        fits = len(ins) == 1
    need("input-count", fits, "inputs_invalid", at)
    if evaluates:
        form = all(i["frame"] is not None and i["samples"] is not None and i["capability"] in (PNG, INT_MOTION)
                   and (i["capability"] != PNG or i["frame"] == 0) for i in ins)
        form = form and all(out[k] is not None for k in ("width", "height", "samples"))
    else:
        form = all(i["frame"] is None and i["samples"] is None for i in ins)
        form = form and all(out[k] is None for k in ("width", "height", "samples"))
    need("input-form", form, "inputs_invalid", at)
    need("check", v["check"] == check_of(d, out["capability"], enc), "check_mismatch", at)
    cls = None
    if v["energy"] is not None:
        try:
            cls = read_class(v["energy"], True)
            ok = cls is not None and label_ok(v["energy"]["joules_micro"], cls)
        except ValueError:
            ok = False
        need("energy", ok, "energy_invalid", at)
    need("synthesis-record", all(LABELS[i["synthesis"]] <= LABELS[v["synthesis"]] for i in ins), "synthesis_dropped", at)
    return dict(v, raw=v, cls=cls)


def signing_input(raw):
    return b"wai:derivation\x01" + jcs({k: x for k, x in raw.items() if k != "sig"}).encode()


def settings_digest(rec):
    settings = {"derivation": rec["derivation"], "encoder": rec["encoder"], "frame": rec["inputs"][0]["frame"],
                "params": rec["params"]}
    return blake3(b"wai:render-settings\x01" + jcs(settings).encode())


def output_manifest(rec, w, h):
    """§4.3, exactly."""
    cap = rec["output"]["capability"]
    s = ('{"wai":"1.0","media":"image","intent":"replicate","model_requirement":{"capability":"%s"},'
         '"conditioning":{"kind":"%s"}' % (cap, cap.rsplit(".", 1)[-1]))
    if len(rec["inputs"]) == 1:
        i = rec["inputs"][0]
        renderer = rec["derivation"] + "@1" if rec["encoder"] is None else rec["encoder"]["id"]
        s += (',"derived_from":{"motif":"%s","motif_capability":"%s","renderer":"%s","settings":"%s"}'
              % (i["content_hash"], i["capability"], renderer, settings_digest(rec).hex()))
    return s + ',"target":{"frames":1,"h":%d,"w":%d}}' % (h, w)


def wai1(manifest, payload):
    return b"WAI1" + struct.pack("<I", len(manifest)) + manifest + struct.pack("<I", len(payload)) + payload


# ---------------------------------------------------------------- verification (§7, §8)

def origin_of(signer_id):
    """jwp-receipts §7.1: the origin an origin-bound signer_id names, or None."""
    m = re.fullmatch(r"did:web:([^#]+)#([A-Za-z0-9._-]{1,64})", signer_id)
    if not m:
        return None
    host, _, port = m.group(1).partition("%3A")
    labels = host.split(".")
    if (len(host) > 253 or len(labels) < 2 or labels[-1].isdigit()
            or not all(re.fullmatch(r"[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?", lab) for lab in labels)):
        return None
    if "%3A" in m.group(1) and not (re.fullmatch(r"[1-9][0-9]{0,4}", port) and int(port) <= 65535 and port != "443"):
        return None
    return "https://" + host + (":" + port if port else "")


def head_entry(text, origin, key):
    """§8 over a head: "not-held", "not-listed", "revoked" or "ok"."""
    try:
        v = json.loads(text, object_pairs_hook=LRV.no_duplicates)
    except ValueError:
        return "not-held"
    if not LRV.revision_ok(v) or v["origin"] != origin:
        return "not-held"
    unsigned = {k: x for k, x in v.items() if k != "sig"}
    if ed_verify(bytes.fromhex(v["root"]), b"wai:key-set\x01" + LRV.jcs(unsigned).encode(), bytes.fromhex(v["sig"])):
        return "not-held"
    e = next((e for e in v["keys"] if e["key"] == key and AUTHORITY in e["may_sign"]), None)
    if e is None:
        return "not-listed"
    return "revoked" if e["status"] == "revoked" else "ok"


class Verifier:
    def __init__(self, vec):
        self.vec = vec
        self.records = vec["records"]
        self.by_hash = {}
        for i, t in enumerate(self.records):
            h = held_hash(t)
            if h is not None:
                self.by_hash.setdefault(h, i)
        self.groups = [(link(g), g) for g in (read_group(r) for r in vec["receipts"]) if g is not None]
        self.objects = [o for o in (read_object(r) for r in vec["receipts"]) if o is not None]
        self.held = {bytes.fromhex(h): bytes.fromhex(b) for h, b in vec["objects"].items()}
        self.sources = set(vec["sources"])
        self.heads = vec["heads"]
        self.claims = vec["claims"]
        self.c2pa = {h: [bytes.fromhex(b) for b in st] for h, st in vec["c2pa"].items()}
        pol = vec["policy"]
        self.refuse_attested = pol.get("attested") == "refuse"
        self.require_recompute = pol.get("recompute") == "required"
        self.require_authority = pol.get("authority") == "required"
        self.done, self.hops, self.sources_seen = {}, [], []

    def receipts(self, receipt, object_id, content):
        """("not-held" | "invalid" | "ok", group)."""
        g = next((g for lk, g in self.groups if lk.hex() == receipt), None)
        if g is None:
            return "not-held", None
        cands = [o for o in self.objects if o["track_alias"] == g["track_alias"] and o["group_id"] == g["group_id"]
                 and o["object_id"] == object_id]
        if not cands:
            return "not-held", None
        if not group_verifies(g):
            return "invalid", g
        if not any(o["content_hash"] == content and included(o, g) for o in cands):
            return "invalid", g
        return "ok", g

    def hop(self, r, depth):
        if r in self.done:
            return self.done[r]
        if r >= len(self.records):
            raise Refuse("hop_not_held", ["record", r])
        rec = read_record(self.records[r], r)
        why = ed_verify(bytes.fromhex(rec["signer_pubkey"]), signing_input(rec["raw"]), bytes.fromhex(rec["sig"]))
        need("key", why != "bad_key", "bad_key", ["record", r])
        need("signature", why is None, "bad_signature", ["record", r])
        me = len(self.hops)
        en = rec["energy"]
        self.hops.append({"record": record_hash(rec["raw"]).hex(), "signer": rec["signer_pubkey"],
                          "derivation": rec["derivation"], "check": rec["check"], "checked": "attested",
                          "energy": None if en is None else {"joules_micro": en["joules_micro"], "class": en["energy_provenance"]},
                          "authority": "none", "synthesis": rec["synthesis"]})
        self.output(r, rec)
        origins, reads = set(), []
        for i, inp in enumerate(rec["inputs"]):
            below = None
            at = ["record", r, "input", i]
            state, g = self.receipts(inp["receipt"], inp["object_id"], inp["content_hash"])
            need("input-held", state != "not-held", "input_receipt_not_held", at)
            need("input-receipts", state == "ok", "input_receipt_invalid", at)
            if inp["derivation"] is None:
                need("source", g["signer_pubkey"] in self.sources, "chain_skips_hop", at)
                o = origin_of(g["signer_id"])
                if o is not None:
                    origins.add(o)
                if all(s["object"] != inp["content_hash"] for s in self.sources_seen):
                    self.sources_seen.append({"object": inp["content_hash"], "signer": g["signer_pubkey"],
                                              "signer_id": g["signer_id"]})
            else:
                need("depth", depth + 1 <= MAX_HOPS, "chain_too_long", at)
                j = self.by_hash.get(bytes.fromhex(inp["derivation"]))
                need("hop-held", j is not None, "hop_not_held", at)
                below = read_record(self.records[j], j)
                bo = below["output"]
                need("hop-match", all(bo[k] == inp[k] for k in ("content_hash", "capability", "receipt", "object_id")),
                     "hop_mismatch", at)
                origins |= self.hop(j, depth + 1)
            env = self.held.get(bytes.fromhex(inp["content_hash"]))
            reads.append(None if env is None else self.read_input(rec, i, env, at))
            # Step 4.5: the provenance it had, and its label.
            listed = {(it["kind"], it["digest"]) for it in rec["provenance"] if it["input"] == i}
            had = [("claim", c[1]) for c in (claim_about(t) for t in self.claims) if c is not None and c[0] == inp["content_hash"]]
            had += [("c2pa", sha256(b).hex()) for b in self.c2pa.get(inp["content_hash"], [])]
            if env is not None:
                had += [("c2pa", sha256(b).hex()) for b in embedded_stores(inp["capability"], env)]
            need("provenance-held", all(x in listed for x in had), "provenance_stripped", at)
            if below is not None:
                mine = {(it["kind"], it["digest"], it["compartment"]) for it in rec["provenance"] if it["input"] == i}
                need("provenance-hop", all((it["kind"], it["digest"], it["compartment"]) in mine for it in below["provenance"]),
                     "provenance_stripped", at)
                need("synthesis-hop", LABELS[inp["synthesis"]] >= LABELS[below["synthesis"]], "synthesis_dropped", at)
        checked = self.recompute(r, rec, reads)
        authority = self.authority(r, rec, origins)
        need("attested", not (rec["check"] == "attest" and self.refuse_attested), "attested_refused", ["record", r, "policy"])
        self.hops[me]["checked"], self.hops[me]["authority"] = checked, authority
        self.done[r] = origins
        return origins

    def output(self, r, rec):
        at = ["record", r, "output"]
        out = rec["output"]
        state, g = self.receipts(out["receipt"], out["object_id"], out["content_hash"])
        need("output-held", state != "not-held", "output_receipt_not_held", at)
        need("output-receipts", state == "ok", "output_receipt_invalid", at)
        need("output-signer", g["signer_pubkey"] == rec["signer_pubkey"], "output_receipt_mismatch", at)
        o = next(o for o in self.objects if o["track_alias"] == g["track_alias"] and o["group_id"] == g["group_id"]
                 and o["object_id"] == out["object_id"] and o["content_hash"] == out["content_hash"] and included(o, g))
        figure = 0 if rec["energy"] is None else rec["energy"]["joules_micro"]
        if not o["labelled"]:
            agrees = True
        elif o["cls"] is None:
            agrees = rec["energy"] is None
        else:
            agrees = rec["energy"] is not None and o["cls"] == rec["cls"]
        need("output-energy", o["joules_micro"] == figure and agrees, "energy_disagrees", at)
        env = self.held.get(bytes.fromhex(out["content_hash"]))
        if env is None:
            return
        need("output-hash", content_hash(env).hex() == out["content_hash"], "output_hash_mismatch", at)
        if rec["derivation"] != EXPLODE:
            w, h = out["width"], out["height"]
            parts = split_wai1(env)
            need("output-manifest", parts is not None and parts[0] == output_manifest(rec, w, h).encode(),
                 "output_manifest_mismatch", at)
            if out["capability"] == PNG:
                try:
                    pw, ph, rgb, _ = png_read(parts[1])
                    ok = (pw, ph) == (w, h) and sha256(rgb).hex() == out["samples"]
                except ValueError:
                    ok = False
                need("output-samples", ok, "samples_mismatch", at)
        else:
            need("output-manifest", capability_of(env) == out["capability"], "output_manifest_mismatch", at)

    def read_input(self, rec, i, env, at):
        inp = rec["inputs"][i]
        need("input-hash", content_hash(env).hex() == inp["content_hash"], "input_hash_mismatch", at)
        if rec["derivation"] == EXPLODE:
            out = explode(env, rec["params"]["component"], inp["capability"])
            need("input-reads", out is not None, "input_unreadable", at)
            return out
        s = read_samples(inp["capability"], env, inp["frame"])
        need("input-reads", s is not None, "input_unreadable", at)
        need("input-samples", sha256(s[2]).hex() == inp["samples"], "samples_mismatch", at)
        return s

    def recompute(self, r, rec, reads):
        at = ["record", r, "recompute"]
        if any(x is None for x in reads):
            if rec["check"] == "attest":
                return "attested"
            need("inputs-held", not self.require_recompute, "inputs_not_held", at)
            return "not-recomputed"
        out = rec["output"]
        if rec["derivation"] == EXPLODE:
            need("replicate", content_hash(reads[0]).hex() == out["content_hash"], "replicate_mismatch", at)
            return "recomputed"
        if not exact(rec["derivation"]):
            return "attested"
        s = evaluate(rec["derivation"], rec["params"], reads)
        need("evaluable", s is not None, "params_invalid", at)
        agree = (s[0], s[1]) == (out["width"], out["height"]) and sha256(s[2]).hex() == out["samples"]
        if rec["check"] == "attest":
            need("attested-samples", agree, "samples_mismatch", at)
            return "attested"
        env = wai1(output_manifest(rec, s[0], s[1]).encode(), png_still(s))
        need("replicate", agree and content_hash(env).hex() == out["content_hash"], "replicate_mismatch", at)
        return "recomputed"

    def authority(self, r, rec, origins):
        confirmed = True
        for o in sorted(origins):
            at = ["record", r, "authority", o]
            if o not in self.heads:
                need("authority-required", not self.require_authority, "edge_not_authorised", at)
                confirmed = False
                continue
            state = head_entry(self.heads[o], o, rec["signer_pubkey"])
            need("head-held", state != "not-held", "key_log_not_held", at)
            need("authorised", state != "not-listed", "edge_not_authorised", at)
            need("not-revoked", state != "revoked", "edge_revoked", at)
        if not origins:
            need("authority-required", not self.require_authority, "edge_not_authorised", ["record", r, "authority", None])
            return "none"
        return "confirmed" if confirmed else "unconfirmed"


def verify_vector(vec):
    """§7: ("accept", facts) or ("refuse", reason, at)."""
    v = Verifier(vec)
    try:
        v.hop(vec["subject"], 1)
    except Refuse as e:
        return ("refuse", e.reason, e.at)
    totals, unmetered = {}, 0
    for h in v.hops:
        if h["energy"] is None:
            unmetered += 1
        else:
            totals[h["energy"]["class"]] = totals.get(h["energy"]["class"], 0) + h["energy"]["joules_micro"]
    subject = strict(vec["records"][vec["subject"]])
    facts = {"subject": v.hops[0]["record"], "hops": v.hops, "sources": v.sources_seen,
             "energy": [{"class": c, "joules_micro": totals[c]} for c in CLASS_ORDER if c in totals],
             "unmetered": unmetered, "provenance": subject["provenance"]}
    return ("accept", facts)


def stated(vec):
    if vec["outcome"] == "accept":
        return ("accept", vec["facts"])
    return ("refuse", vec["reason"], vec["at"])


def check(doc, quiet=False):
    """Every vector's verdict against the one it states: the failures."""
    fails = []
    for vec in doc["vectors"]:
        try:
            got = verify_vector(vec)
        except Exception as e:  # a crash on a mutated or weakened run is a failure to verify
            got = ("crash", repr(e))
        if got != stated(vec):
            fails.append("%s: got %s" % (vec["name"], str(got)[:160]))
            if not quiet:
                print("FAIL vector %s: got %s, stated %s" % (vec["name"], str(got)[:200], str(stated(vec))[:200]))
        elif not quiet:
            print("PASS vector %s: %s" % (vec["name"], got[0] if got[0] == "accept" else "%s %s" % (got[1], got[2])))
    return fails


# ---------------------------------------------------------------- mutations

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

    def mutate(label, f):
        d = copy.deepcopy(doc)
        f(d["vectors"])
        return label, d

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

    def edit_record(name, idx, f):
        def go(vs):
            v = vs[by[name]]
            raw = json.loads(v["records"][idx])
            f(raw)
            v["records"][idx] = jcs(raw)
        return go

    def flip_object(name, which):
        def go(vs):
            v = vs[by[name]]
            k = sorted(v["objects"])[which]
            v["objects"][k] = flip_hex(v["objects"][k], len(v["objects"][k]) - 3)
        return go

    def last(name):
        return len(doc["vectors"][by[name]]["records"]) - 1

    return [
        mutate("a byte of a still's record signature", edit_record("still_png", last("still_png"), lambda r: r.update(sig=flip_hex(r["sig"], 5)))),
        mutate("a still's frame stated as another", edit_record("still_png", last("still_png"), lambda r: r["inputs"][0].update(frame=2))),
        mutate("a crop's region moved in its record", edit_record("crop_after_still", last("crop_after_still"), lambda r: r["params"].update(x=0))),
        mutate("a grid's rows and columns swapped", edit_record("grid", last("grid"), lambda r: r["params"].update(columns=1, rows=4))),
        mutate("an overlay's fill changed", edit_record("overlay", last("overlay"), lambda r: r["params"].update(fill="203041"))),
        mutate("the attested JPEG stated replicate", edit_record("still_jpeg", last("still_jpeg"), lambda r: r.update(check="replicate"))),
        mutate("a metered hop's figure raised", edit_record("metered", last("metered"), lambda r: r["energy"].update(joules_micro=501))),
        mutate("an input of the grid held changed", flip_object("grid", 0)),
        mutate("the explode's input held changed", flip_object("explode", 1)),
        mutate("the chain of crop_after_still with its first hop's record dropped",
               lambda vs: vs[by["crop_after_still"]]["records"].pop(0) and vs[by["crop_after_still"]].update(subject=0)),
        mutate("the publisher taken away from the sources", lambda vs: vs[by["rotations"]].update(sources=[])),
        mutate("the confirmed chain stated unconfirmed", lambda vs: vs[by["authority_confirmed"]]["facts"]["hops"][0].update(authority="unconfirmed")),
        mutate("the not-recomputed chain stated recomputed", lambda vs: vs[by["not_recomputed"]]["facts"]["hops"][1].update(checked="recomputed")),
        mutate("the metered totals stated as one", lambda vs: vs[by["metered"]]["facts"].update(energy=[{"class": "ModelBased", "joules_micro": 6000}])),
        mutate("a source stated with another signer_id", lambda vs: vs[by["still_png"]]["facts"]["sources"][0].update(signer_id="did:web:other.example#k")),
        mutate("an attested scaling given the policy that refuses it", lambda vs: vs[by["scaling"]]["policy"].update(attested="refuse")),
        mutate("a refused vector's place moved", lambda vs: vs[by["replicate_mismatch"]].update(at=["record", 0, "output"])),
        mutate("the revoked edge's head taken away", lambda vs: vs[by["edge_revoked"]].update(heads={})),
        mutate("a receipt of the explode withheld", lambda vs: vs[by["explode"]]["receipts"].pop()),
        mutate("the personal store stated system in the facts",
               lambda vs: [it.update(compartment="system") for it in vs[by["provenance_carried"]]["facts"]["provenance"]]),
        mutate("the store beside the still held changed",
               lambda vs: [st.__setitem__(0, st[0][:-2] + "00") for st in vs[by["provenance_carried"]]["c2pa"].values()]),
        mutate("the claim about the still held with its value changed",
               lambda vs: vs[by["provenance_carried"]]["claims"].__setitem__(0, vs[by["provenance_carried"]]["claims"][0].replace('"value":"', '"value":"00'))),
        mutate("the generated chain stated unlabelled",
               lambda vs: vs[by["synthesis_carried"]]["facts"]["hops"][0].update(synthesis=None)),
    ]


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) == 31 and len(set(REASONS)) == len(REASONS) and len(DERIVATIONS) == 9
    print(("PASS" if ok else "FAIL") + " §9 and §4 read from the extension: %d reasons, %d derivations" % (len(REASONS), len(DERIVATIONS)))
    fails += [] if ok else ["the extension's tables do 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", "edge_verify.py", "__pycache__"})
    print(("PASS" if not extra else "FAIL") + " no file but vectors.json, README.md and edge_verify.py")
    fails += ["an extra file: %s" % n for n in extra]
    seen = {v["reason"] for v in doc["vectors"]}
    missing = [r for r in REASONS if r not in seen]
    print(("PASS" if not missing else "FAIL") + " every reason of §9 appears in the corpus")
    fails += ["no vector 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 in CHECKS:
        DROPPED.clear()
        DROPPED.add(name)
        caught = check(doc, quiet=True)
        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") + " deriving-edge corpus")
    return 1 if fails else 0


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