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

#!/usr/bin/env python3
"""Independent verifier of the live-metering corpus (extensions/live-metering.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. Each room is verified by the live-rooms corpus's own
second implementation, live_rooms_verify.py (live-rooms §9); each take by
capture_take_verify.py (capture-binding §7.2, §9.2); each composition receipt by
live_compose_verify.py (live-compose §8.3). Everything else is here, from the text:

- the per-hop session record (§3): read in version 1 or 2 by its own layout, its class, its
  rules (version 2's DVR figures held to its entries), the size bound of its version, its
  ledger and the ledger's root;
- the room-session receipt (§4): read in version 1 or 2 by its own layout (version 2 states the
  bytes the hops sent the session's sessions, and prices them per 10^9 bytes), its rules, who
  may sign it;
- the room's replay (§5.3): who publishes, the grants in effect, the media tracks
  published;
- the units (§4.4), the joules per hop and per class (§4.5), each grant's consumption and
  its verdicts (§4.6), and the QoE cell (§4.7) from the logs' session metrics
  (delivery-telemetry §7), all recomputed with Python's integers.

For every vector it requires the outcome, reason and place, and the digest, units,
records (each one's digest, size, bound and version, and a version 2 record's DVR figures),
totals, coverage, consumption entries, root, grants past a ceiling and QoE cell the corpus
states; of the corpus, that every reason of §9 refuses a vector and that §9
lists exactly those reasons. Then it mutates copies of the corpus, one defect each, and
requires every copy to fail.

    python3 live_metering_verify.py <live-conformance/metering>

Exit status 0 only if every vector verifies and every mutated copy fails. Python 3.8 or
later; standard library only.
"""
import copy
import hashlib
import json
import os
import re
import struct
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 capture_take_verify as CT  # noqa: E402
import live_compose_verify as LCV  # noqa: E402
import _ed25519 as ED  # noqa: E402

Refuse = LEV.Refuse

HOP_MAGIC = b"WMH1"
HOP_VERSIONS = (1, 2)
SESSION_MAGIC = b"WMR1"
SESSION_VERSIONS = (1, 2)
DVR_FIELDS = ("dvr_objects_out", "dvr_bytes_out", "dvr_objects_stored", "dvr_bytes_stored", "dvr_byte_s", "dvr_dropped")
HOP_DOMAIN = b"wai:metering-hop\x01"
SESSION_DOMAIN = b"wai:metering-session\x01"
LEDGER_DOMAIN = b"wai:metering-ledger\x01"
CONSUMPTION_DOMAIN = b"wai:metering-consumption\x01"
NODE_DOMAIN = b"wai:out-node\x01"
NO_CEILING = 2 ** 64 - 1
NO_FUNDS = 2 ** 64 - 1
CLASSES = 9
GAP_US = 1000000
QOE_MIN = 50
FLOOR = 288676
LABELS = {2: "ModelBased", 3: "Estimator", 4: "OnChipCounter", 5: "CalibratedInstrument"}
HOP_AUTHORITY = {1: "live/relay", 2: "live/compositor", 3: "live/ingest"}
METER_AUTHORITY = "live/meter"
ZERO = bytes(32)
JOULES_VERDICT = ["none", "within", "past", "partial", "unmetered"]
FUNDS_VERDICT = ["none", "within", "past", "no_tariff"]
LIFECYCLE = range(0x10, 0x17)

REASONS = [
    "session_invalid", "signer_not_authorised", "signer_revoked", "record_missing", "record_too_large",
    "energy_invalid", "record_invalid", "hop_not_authorised", "hop_revoked", "span_outside_session",
    "track_unknown", "ledger_invalid", "ledger_mismatch", "unit_mismatch", "classes_summed",
    "ceiling_mismatch", "qoe_invalid", "qoe_not_suppressed",
]


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


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


def bound(m, version=1):
    """§3.7: B(M) = 552 + 180 × (M + 2) for version 1, B₂(M) = 600 + 180 × (M + 2) for version 2."""
    return (600 if version == 2 else 552) + 180 * (m + 2)


def bound_of(b, m):
    """§3.7: the bound of the version the fifth byte names, read before anything else; B(M)
    for a fifth byte other than 2, or none."""
    return bound(m, 2 if len(b) >= 5 and b[4] == 2 else 1)


def given_digest(b):
    """§5.1 step 4: SHA-256 over all but the last 64 bytes."""
    return sha(b[:max(len(b) - 64, 0)])


def node(a, b):
    return sha(NODE_DOMAIN, a, b)


def mth(leaves):
    """RFC 6962 §2.1, with the node of staged-measurement Appendix C; zero for none."""
    if not leaves:
        return ZERO
    if len(leaves) == 1:
        return leaves[0]
    k = 1
    while k * 2 < len(leaves):
        k *= 2
    return node(mth(leaves[:k]), mth(leaves[k:]))


class R:
    """A cursor; running out is `truncated`."""

    def __init__(self, b, p=0):
        self.b, self.p = b, p

    def take(self, n):
        if self.p + n > len(self.b):
            raise Refuse("truncated")
        s = self.b[self.p:self.p + n]
        self.p += n
        return s

    def u8(self):
        return self.take(1)[0]

    def u16(self):
        return struct.unpack("<H", self.take(2))[0]

    def u32(self):
        return struct.unpack("<I", self.take(4))[0]

    def u64(self):
        return struct.unpack("<Q", self.take(8))[0]

    def be32(self):
        return struct.unpack(">I", self.take(4))[0]

    def be64(self):
        return struct.unpack(">Q", self.take(8))[0]

    def done(self):
        return self.p == len(self.b)


def read_head(b, magic, versions):
    """live-events §3.3's first two steps: the version, one of `versions`."""
    if len(b) < 4:
        raise Refuse("truncated")
    if b[:4] != magic:
        raise Refuse("malformed")
    if len(b) < 5:
        raise Refuse("truncated")
    if b[4] not in versions:
        raise Refuse("unknown_version")
    return b[4]


INVALID = "invalid"


def read_class(r):
    """energy-measurement §2.7's tag and evidence: None (tag 0), a tuple, or INVALID."""
    tag = r.u8()
    if tag == 0:
        return None
    if tag in (2, 3):
        return (tag,)
    if tag == 4:
        ppm, window, f = r.be32(), r.be64(), r.u8()
        return (4, ppm, window, f) if ppm > 0 and window > 0 and f in (0, 1, 2) else INVALID
    if tag == 5:
        ppm, window, n = r.be32(), r.be64(), r.be32()
        if n > 255:
            raise Refuse("malformed")
        ref = r.take(n)
        try:
            text = ref.decode("utf-8")
        except UnicodeDecodeError:
            return INVALID
        return (5, ppm, window, text) if ppm > 0 and window > 0 and text.strip() != "" else INVALID
    return INVALID


def energy_ok(joules, cls):
    """§3.6's `energy_invalid` rules."""
    if cls == INVALID:
        return False
    if cls is None:
        return joules == 0
    if joules == 0:
        return False
    return cls[0] not in (4, 5) or joules * cls[1] >= FLOOR


def check_sig(key, domain, body, sig):
    bad = ED.verify(key, domain + body, sig)
    if bad:
        raise Refuse(bad)


# ---------------------------------------------------------------- §3 the per-hop record

def read_hop(b, room):
    """§3.6: the record, of version 1 or 2, its class, its rules, its room, its signature."""
    version = read_head(b, HOP_MAGIC, HOP_VERSIONS)
    r = R(b, 5)
    x = {"version": version, "room": r.take(32), "hop": r.take(32), "role": r.u8(), "start": r.u64(), "end": r.u64(),
         "joules": r.u64()}
    x["class"] = read_class(r)
    x["tracks"] = []
    for _ in range(r.u16()):
        e = {"kind": r.u8(), "track": r.take(32), "dir": r.u8()}
        for k in ("objects", "bytes", "timeout", "refused", "duplicate", "closed", "receipted"):
            e[k] = r.u64()
        x["tracks"].append(e)
    x["sessions"] = r.u32()
    x["sub"] = [r.u64() for _ in range(CLASSES)]
    x["events"], x["grants"], x["ledger"] = r.u64(), r.u32(), r.take(32)
    # §3.1: version 2 has `dvr`, six u64, after `ledger` and before `sig` (§3.9).
    x["dvr"] = {k: r.u64() for k in DVR_FIELDS} if version == 2 else None
    sig = r.take(64)
    if not r.done():
        raise Refuse("malformed")
    if not energy_ok(x["joules"], x["class"]):
        raise Refuse("energy_invalid")
    order = [(e["kind"], e["track"], e["dir"]) for e in x["tracks"]]

    def entry_ok(e):
        if e["kind"] == 1:
            kind_ok = e["track"] != ZERO
        elif e["kind"] in (2, 3):
            kind_ok = e["track"] == ZERO
        else:
            return False
        if e["dir"] == 1:
            return kind_ok and e["timeout"] == 0 and e["closed"] == 0 and e["receipted"] == 0 \
                and e["refused"] + e["duplicate"] <= e["objects"]
        if e["dir"] == 2:
            return kind_ok and e["refused"] == 0 and e["duplicate"] == 0 and e["receipted"] <= e["timeout"] + e["closed"]
        return False

    def dvr_ok(d):
        """§3.6, version 2: what it sent from its store within its out entries, what it
        stored within its in entries."""
        if d is None:
            return True
        out = [e for e in x["tracks"] if e["dir"] == 2]
        inn = [e for e in x["tracks"] if e["dir"] == 1]
        return (d["dvr_objects_out"] <= sum(e["objects"] for e in out) and d["dvr_bytes_out"] <= sum(e["bytes"] for e in out)
                and d["dvr_objects_stored"] <= sum(e["objects"] for e in inn)
                and d["dvr_bytes_stored"] <= sum(e["bytes"] for e in inn))
    rules = (x["role"] in (1, 2, 3) and x["start"] < x["end"] and all(a < z for a, z in zip(order, order[1:]))
             and all(entry_ok(e) for e in x["tracks"])
             and (x["sessions"] != 0 or (not any(x["sub"]) and x["events"] == 0))
             and ((x["grants"] == 0) == (x["ledger"] == ZERO))
             and dvr_ok(x["dvr"]))
    if not rules:
        raise Refuse("record_invalid")
    if x["room"] != room:
        raise Refuse("wrong_room")
    check_sig(x["hop"], HOP_DOMAIN, b[:-64], sig)
    x["digest"] = sha(b[:-64])
    return x


def read_ledger(b):
    """§3.6: the ledger's entries, or Refuse("ledger_invalid")."""
    if len(b) % 143:
        raise Refuse("ledger_invalid")
    out = []
    for k in range(0, len(b), 143):
        c = b[k:k + 143]
        e = {"raw": c, "grant": c[:32], "jc": int.from_bytes(c[32:40], "little"), "fc": int.from_bytes(c[40:48], "little"),
             "currency": c[48:51], "sessions": int.from_bytes(c[51:55], "little"),
             "in": int.from_bytes(c[55:63], "little"), "out": int.from_bytes(c[63:71], "little"),
             "sub": [int.from_bytes(c[71 + 8 * i:79 + 8 * i], "little") for i in range(CLASSES)]}
        currency_ok = e["currency"] == b"\x00\x00\x00" if e["fc"] == NO_CEILING else all(0x41 <= x <= 0x5A for x in e["currency"])
        zero_ok = e["grant"] != ZERO or (e["jc"] == NO_CEILING and e["fc"] == NO_CEILING)
        if not currency_ok or not zero_ok or (out and out[-1]["grant"] >= e["grant"]):
            raise Refuse("ledger_invalid")
        out.append(e)
    return out


def ledger_root(entries):
    return mth([sha(LEDGER_DOMAIN, e["raw"]) for e in entries])


# ---------------------------------------------------------------- §4 the receipt

def read_receipt(b, room):
    """§4.2: the receipt, `classes_summed` for a total of tag 0, its rules, its room, its signature."""
    version = read_head(b, SESSION_MAGIC, SESSION_VERSIONS)
    r = R(b, 5)
    x = {"version": version, "room": r.take(32), "signer": r.take(32), "start": r.u64(), "end": r.u64()}
    x["hops"], classes_ok = [], True
    for _ in range(r.u8()):
        h = {"record": r.take(32), "hop": r.take(32), "role": r.u8(), "joules": r.u64()}
        h["class"] = read_class(r)
        classes_ok = classes_ok and energy_ok(h["joules"], h["class"])
        x["hops"].append(h)
    x["takes"] = [r.take(32) for _ in range(r.u8())]
    x["compose_count"], x["compose_digest"] = r.u32(), r.take(32)
    x["qoe_count"], x["qoe_digest"] = r.u32(), r.take(32)
    # §4.1: version 2 states a sixteenth unit, the bytes out (§4.4), and a fourteenth price.
    x["units"] = units_of([r.u64() for _ in range(16 if version == 2 else 15)])
    x["unmetered"] = r.u8()
    x["totals"] = [(r.u8(), r.u32(), r.u64()) for _ in range(r.u8())]
    x["currency"] = r.take(3)
    x["prices"] = [r.u32() for _ in range(14 if version == 2 else 13)]
    x["grants"], x["root"] = r.u32(), r.take(32)
    x["past"] = [(r.take(32), r.u8()) for _ in range(r.u32())]
    cell = [r.u8()] + [r.u32() for _ in range(3)] + [r.u64() for _ in range(3)] + [r.u32(), r.u64()]
    x["cell"] = cell_dict(cell)
    sig = r.take(64)
    if not r.done():
        raise Refuse("malformed")
    if any(t[0] == 0 for t in x["totals"]):
        raise Refuse("classes_summed")
    tariff_ok = all(p == 0 for p in x["prices"]) if x["currency"] == b"\x00\x00\x00" else all(0x41 <= c <= 0x5A for c in x["currency"])
    recs, takes, tags, past = [h["record"] for h in x["hops"]], x["takes"], [t[0] for t in x["totals"]], [p[0] for p in x["past"]]
    rules = (x["start"] < x["end"] and all(a < z for a, z in zip(recs, recs[1:])) and all(h["role"] in (1, 2, 3) for h in x["hops"])
             and classes_ok and all(a < z for a, z in zip(takes, takes[1:]))
             and ((x["compose_count"] == 0) == (x["compose_digest"] == ZERO))
             and ((x["qoe_count"] == 0) == (x["qoe_digest"] == ZERO))
             and all(a < z for a, z in zip(tags, tags[1:])) and all(2 <= t[0] <= 5 and t[1] > 0 for t in x["totals"])
             and tariff_ok and all(a < z for a, z in zip(past, past[1:])) and all(f in (1, 2, 3) for _, f in x["past"])
             and (cell[0] == 0 or (cell[0] == 1 and not any(cell[1:]))))
    if not rules:
        raise Refuse("session_invalid")
    if x["room"] != room:
        raise Refuse("wrong_room")
    check_sig(x["signer"], SESSION_DOMAIN, b[:-64], sig)
    x["digest"] = sha(b[:-64])
    return x


def units_of(v):
    u = {"publisher_us": v[0], "subscriber_us": v[1:10], "sessions": v[10], "composition_us": v[11],
         "events_sent": v[12], "events_delivered": v[13], "recorded_us": v[14]}
    if len(v) == 16:
        u["bytes_out"] = v[15]
    return u


def cell_dict(c):
    return {"suppressed": c[0] == 1, "started": c[1], "start_failed": c[2], "exit_before_start": c[3], "stalls": c[4],
            "stall_ms": c[5], "engaged_ms": c[6], "vst_count": c[7], "vst_ms": c[8]}


# ---------------------------------------------------------------- §5.3 the room

EMPTY = {"present": {}, "own": {}, "slots": None}


class Ctx:
    """A room that verified: its descriptor, grants, key log, accepted records and replay."""

    def __init__(self, room):
        descriptor = bytes.fromhex(room["descriptor"])
        grants = [bytes.fromhex(g) for g in room["grants"]]
        head = room["key_log_head"]
        self.chains = [[bytes.fromhex(x) for x in c] for c in room["chains"]]
        out = LRV.verify_room(descriptor, grants, head, self.chains)
        if out["outcome"] == "refuse":
            raise Stop(out["reason"], ["room"] + out["at"])
        self.d = LRV.read_descriptor(descriptor)
        self.id = self.d["room"]
        if out["outcome"] == "gap" and any(LRV.is_room_key(self.d, bytes.fromhex(g["principal"])) for g in out["gaps"]):
            raise Stop("lifecycle_incomplete", ["room"])
        self.gaps = out.get("gaps", [])
        held, _ = LRV.check_grants(self.d, grants)
        self.grants = {g["digest"]: g for g in held}
        self.log = json.loads(head) if self.d["origin"] != b"" else None
        self.view = LCV.view_room(room)
        self.seen, digests, instants = [], set(), set()
        for c, i, g in out["order"]:
            ev = LEV.read_record(self.chains[c][i])
            if ev["digest"] in digests:
                continue
            digests.add(ev["digest"])
            self.seen.append({"chain": c, "record": i, "ev": ev, "grant": bytes.fromhex(g) if g else None})
            if LRV.is_room_key(self.d, ev["principal"]) and ev["kind"] in LIFECYCLE:
                instants.add(ev["t"])
        empty = {"present": {}, "own": {}, "slots": [{"live": None, "standby": None} for _ in self.d["slots"]]}
        self.segments = [[0, empty]]
        for x in sorted(instants):
            st = self.view.at(x)
            if x == 0:
                self.segments[0][1] = st
            else:
                self.segments.append([x, st])

    def over(self, s, e):
        out = []
        for k, (a, st) in enumerate(self.segments):
            z = self.segments[k + 1][0] if k + 1 < len(self.segments) else 2 ** 64
            lo, hi = max(a, s), min(z, e)
            if lo < hi:
                out.append((st, lo, hi))
        return out

    def at(self, x):
        return next(st for a, st in reversed(self.segments) if a <= x)

    def published(self, st):
        out = set()
        for (p, m), (_, layers) in st["own"].items():
            out |= {LCV.track_digest(self.id, p.hex(), m, str(n)) for n in range(layers)}
        for slot, s in zip(self.d["slots"], st["slots"]):
            if s["live"]:
                out |= {LCV.track_digest(self.id, "_" + slot["name"].decode(), slot["media"], str(n)) for n in range(s["live"][2])}
        return out

    def ceilings(self, g):
        if g == ZERO:
            return (NO_CEILING, NO_CEILING, b"\x00\x00\x00")
        x = self.grants.get(g)
        return None if x is None else (x["joules"], x["funds"], x["currency"])

    def entry(self, key):
        if self.log is None:
            return None
        return next((e for e in self.log["keys"] if e["key"] == key.hex()), None)


def publishers(st):
    out = {p for p, _ in st["own"]}
    for s in st["slots"]:
        out |= {h[0] for h in (s["live"], s["standby"]) if h}
    return out


def hop_authorised(ctx, x):
    """§5.2."""
    if ctx.d["origin"] == b"":
        if not LRV.is_room_key(ctx.d, x["hop"]):
            raise Refuse("hop_not_authorised")
        return
    e = ctx.entry(x["hop"])
    if e is None:
        raise Refuse("hop_not_authorised")
    need = [HOP_AUTHORITY[x["role"]]] + (["energy/" + LABELS[x["class"][0]]] if x["class"] else [])
    if not all(a in e["may_sign"] for a in need):
        raise Refuse("hop_not_authorised")
    if e["status"] == "revoked":
        raise Refuse("hop_revoked")


def signer_authorised(ctx, rc):
    """§4.3."""
    d = ctx.d
    if d["origin"] == b"":
        if rc["signer"] != d["authority"]:
            raise Refuse("signer_not_authorised")
        return
    e = ctx.entry(rc["signer"])
    if e is None or not (rc["signer"] == d["authority"] or METER_AUTHORITY in e["may_sign"]):
        raise Refuse("signer_not_authorised")
    if not all("energy/" + LABELS[t[0]] in e["may_sign"] for t in rc["totals"]):
        raise Refuse("signer_not_authorised")
    if e["status"] == "revoked":
        raise Refuse("signer_revoked")


# ---------------------------------------------------------------- delivery-telemetry §2, §7

EVENTS = {"request", "response", "start", "present", "switch", "layer-change", "stall-start", "stall-end", "frame-drop",
          "capable", "abandon", "end"}
MEMBERS = {"v", "sid", "sn", "ts_ms", "event", "cid", "st", "sf", "object", "present", "switch", "layer_change", "capable",
           "cohort", "metrics", "claim"}
WITH = {"object": ("request", "response"), "present": ("present",), "switch": ("switch",),
        "layer_change": ("layer-change",), "capable": ("capable",)}


def is_int(v):
    return isinstance(v, int) and not isinstance(v, bool) and 0 <= v <= 2 ** 53 - 1


def read_observation(o):
    """delivery-telemetry §2.1 and §2.4, as far as the session metrics read an observation."""
    if not isinstance(o, dict) or set(o) != MEMBERS or o["v"] != 1 or isinstance(o["v"], bool):
        raise Refuse("qoe_invalid")
    if not isinstance(o["sid"], str) or not re.fullmatch(r"[0-9a-f]{32}", o["sid"]):
        raise Refuse("qoe_invalid")
    if not is_int(o["sn"]) or not (o["ts_ms"] is None or is_int(o["ts_ms"])) or o["event"] not in EVENTS:
        raise Refuse("qoe_invalid")
    if o["st"] not in ("v", "l", "ll") or o["sf"] not in ("o", "h", "d"):
        raise Refuse("qoe_invalid")
    for k, events in WITH.items():
        if (o[k] is not None) != (o["event"] in events):
            raise Refuse("qoe_invalid")
    p = o["present"]
    if p is not None:
        if not isinstance(p, dict) or set(p) != {"unit", "layers", "first", "late_ms", "decode_us"}:
            raise Refuse("qoe_invalid")
        if not is_int(p["unit"]) or not is_int(p["layers"]) or p["layers"] == 0 or not isinstance(p["first"], bool):
            raise Refuse("qoe_invalid")
    if o["claim"] is not None and o["event"] != "end":
        raise Refuse("qoe_invalid")
    event_mode = o["event"] != "request" and (o["event"] != "present" or p["first"])
    if event_mode and o["ts_ms"] is None:
        raise Refuse("qoe_invalid")
    return o


def session_metrics(log):
    """delivery-telemetry §7: one log's metrics, or Refuse("qoe_invalid") for one that is not a log (§7.1)."""
    m = {"started": False, "vst": None, "exit_before_start": False, "start_failed": False, "stalls": 0, "stall_ms": 0, "engaged_ms": 0}
    start = present = stall = None
    last, ended = 0, False
    for i, o in enumerate(log):
        if o["sid"] != log[0]["sid"] or (i > 0 and o["sn"] <= log[i - 1]["sn"]) or ended:
            raise Refuse("qoe_invalid")
        t = o["ts_ms"]
        if t is not None:
            if t < last:
                raise Refuse("qoe_invalid")
            last = t
        e = o["event"]
        if e == "start" and start is None:
            m["started"], start = True, t
        elif e == "present":
            if o["present"]["first"]:
                if present is not None:
                    raise Refuse("qoe_invalid")
                present = t
                m["vst"] = t - min(start, t) if start is not None and t is not None else None
        elif e == "stall-start" and present is not None and stall is None:
            stall = t
        elif e == "stall-end":
            if stall is not None and t is not None:
                m["stalls"] += 1
                m["stall_ms"] += t - stall
            stall = None
        elif e in ("abandon", "end"):
            ended = True
            if stall is not None:
                m["stalls"] += 1
                m["stall_ms"] += t - stall
                stall = None
            if present is not None:
                m["engaged_ms"] = t - present
            elif e == "abandon":
                m["exit_before_start"] = True
            else:
                m["start_failed"] = m["started"]
    return m


def read_logs(b):
    """§6.1: each log's metrics."""
    try:
        doc = json.loads(b.decode("utf-8"))
    except (UnicodeDecodeError, ValueError):
        raise Refuse("qoe_invalid")
    if not isinstance(doc, list):
        raise Refuse("qoe_invalid")
    out = []
    for log in doc:
        if not isinstance(log, list) or not log:
            raise Refuse("qoe_invalid")
        out.append(session_metrics([read_observation(o) for o in log]))
    return out


def cell_of(ms):
    """§4.7."""
    if len(ms) < QOE_MIN:
        return cell_dict([1] + [0] * 8)
    return cell_dict([0, sum(m["started"] for m in ms), sum(m["start_failed"] for m in ms), sum(m["exit_before_start"] for m in ms),
                      sum(m["stalls"] for m in ms), sum(m["stall_ms"] for m in ms), sum(m["engaged_ms"] for m in ms),
                      sum(m["vst"] is not None for m in ms), sum(m["vst"] for m in ms if m["vst"] is not None)])


# ---------------------------------------------------------------- §5 verification

def u64s(*v):
    return b"".join(struct.pack("<Q", x) for x in v)


def evaluate(ctx, s, e, hops, takes, compositions, descriptors, logs, prices, currency, version=1):
    """§5.1 steps 4 to 7 and the computations of §4: what the session gives, or Stop."""
    room = ctx.id
    published = set()
    for st, _, _ in ctx.over(s, e):
        published |= ctx.published(st)
    m = len(published)
    records = []
    for i, (given, entry) in enumerate(hops):
        at = ["hop", i]
        try:
            if given is None:
                raise Refuse("record_missing")
            rb, lb = given
            b_ = bound_of(rb, m)
            if len(rb) > b_:
                raise Refuse("record_too_large")
            x = read_hop(rb, room)
            if entry is not None and (entry["hop"], entry["role"], entry["joules"], entry["class"]) != (x["hop"], x["role"], x["joules"], x["class"]):
                raise Refuse("unit_mismatch")
            hop_authorised(ctx, x)
            if x["start"] < s or x["end"] > e:
                raise Refuse("span_outside_session")
            if any(t["kind"] == 1 and t["track"] not in published for t in x["tracks"]):
                raise Refuse("track_unknown")
            led = read_ledger(lb)
            if len(led) != x["grants"] or ledger_root(led) != x["ledger"] \
                    or any(ctx.ceilings(l["grant"]) != (l["jc"], l["fc"], l["currency"]) for l in led):
                raise Refuse("ledger_mismatch")
            if any(sum(l["sub"][c] for l in led) != x["sub"][c] for c in range(CLASSES)):
                raise Refuse("unit_mismatch")
        except Refuse as err:
            raise Stop(err.reason, at)
        x["ledger_entries"], x["size"], x["bound"] = led, len(rb), b_
        records.append(x)
    take_digests, recorded = [], []
    for i, given in enumerate(takes):
        try:
            if given is None:
                raise Refuse("record_missing")
            mf = CT.read_manifest(given, room)
            _, _, _, commits = CT.find(ctx.d, ctx.grants, ctx.seen, ctx.gaps, mf)
        except Refuse as err:
            raise Stop(err.reason, ["take", i])
        except CT.Stop as err:
            raise Stop(err.reason, ["take", i])
        take_digests.append(sha(given[:-64]))
        us = 0
        if commits:
            o, size = commits[0][1]["offset"], commits[-1][1]["size"]
            for c in mf["chunks"][:min(size, len(mf["chunks"]))]:
                us += max(min(c["end"] + o, e) - max(c["start"] + o, s), 0)
        recorded.append((us, commits[0][0]["grant"] if commits else None))
    digests = [given_digest(b) for b in compositions]
    outputs = {}
    for i, b in enumerate(compositions):
        try:
            rc = LCV.read_receipt(b)
            dbytes = next((d for d in descriptors if given_digest(d) == rc["descriptor"]), None)
            if dbytes is None:
                raise Refuse("record_missing")
            out = LCV.verify_receipt(ctx.view, dbytes, b)
            if out["outcome"] == "refuse":
                raise Refuse(out["reason"])
            if rc["t"] < s or rc["t"] >= e:
                raise Refuse("span_outside_session")
        except Refuse as err:
            raise Stop(err.reason, ["compose", i])
        outputs.setdefault((rc["compositor"], rc["media"], rc["output"]["track"]), []).append((rc["t"], rc["compositor"]))
    metrics = []
    if logs is not None:
        try:
            metrics = read_logs(logs)
        except Refuse as err:
            raise Stop(err.reason, ["qoe"])
    # §4.4 the units and §4.6 the consumption.
    units = {"publisher_us": 0, "subscriber_us": [0] * CLASSES, "sessions": 0, "composition_us": 0, "events_sent": 0,
             "events_delivered": 0, "recorded_us": 0}
    if version == 2:
        units["bytes_out"] = 0                  # §4.4: the ledgers' bytes out, version 2
    per = {}

    def grant(g):
        if g not in per:
            jc, fc, cur = ctx.ceilings(g) or (NO_CEILING, NO_CEILING, b"\x00\x00\x00")
            per[g] = {"grant": g, "jc": jc, "fc": fc, "currency": cur, "pub": 0, "sub": [0] * CLASSES, "comp": 0, "rec": 0,
                      "ev": 0, "out": 0, "attr": [0, 0, 0, 0], "unmetered": False, "metered": False}
        return per[g]
    for st, a, z in ctx.over(s, e):
        for p in publishers(st):
            units["publisher_us"] += z - a
            if p in st["present"]:
                grant(st["present"][p])["pub"] += z - a
        for g in st["present"].values():
            grant(g)
    for x in records:
        for c in range(CLASSES):
            units["subscriber_us"][c] += x["sub"][c]
        units["sessions"] += x["sessions"]
        units["events_delivered"] += x["events"]
        moved = sum(l["in"] + l["out"] for l in x["ledger_entries"])
        for l in x["ledger_entries"]:
            g = grant(l["grant"])
            for c in range(CLASSES):
                g["sub"][c] += l["sub"][c]
            if version == 2:
                units["bytes_out"] += l["out"]
                g["out"] += l["out"]
            b = l["in"] + l["out"]
            if b == 0:
                continue
            if x["class"] is None:
                g["unmetered"] = True
            else:
                g["attr"][x["class"][0] - 2] += x["joules"] * b // moved
                g["metered"] = True
    for runs in outputs.values():
        runs.sort()
        for (t1, comp), (t2, _) in zip(runs, runs[1:]):
            if t2 - t1 <= GAP_US:
                units["composition_us"] += t2 - t1
                g = ctx.at(t1)["present"].get(comp)
                if g is not None:
                    grant(g)["comp"] += t2 - t1
    for x in ctx.seen:
        if s <= x["ev"]["t"] < e:
            units["events_sent"] += 1
            if x["grant"] is not None:
                grant(x["grant"])["ev"] += 1
    for us, g in recorded:
        units["recorded_us"] += us
        if g is not None:
            grant(g)["rec"] += us
    totals = {}
    for x in records:
        if x["class"] is not None:
            f, j = totals.get(x["class"][0], (0, 0))
            totals[x["class"][0]] = (f + 1, j + x["joules"])
    consumption = []
    for g in sorted(per):
        c = per[g]
        attributed = sum(c["attr"])
        if c["jc"] == NO_CEILING:
            jv = 0
        elif attributed > c["jc"]:
            jv = 2
        elif not c["metered"]:
            jv = 4
        elif c["unmetered"]:
            jv = 3
        else:
            jv = 1
        priced = currency != b"\x00\x00\x00" and currency == c["currency"]
        funds = NO_FUNDS
        if priced:
            n = c["pub"] * prices[0] + sum(c["sub"][k] * prices[1 + k] for k in range(CLASSES)) + c["comp"] * prices[10] \
                + c["rec"] * prices[11] + c["ev"] * prices[12] * 10 ** 6
            if version == 2:            # §4.6: N in 10^-12 of a minor unit, bytes at a price per 10^9 bytes
                funds = min(-(-(n * 1000 + c["out"] * prices[13]) // 10 ** 15), NO_FUNDS - 1)
            else:
                funds = min(-(-n // 10 ** 12), NO_FUNDS - 1)
        if c["fc"] == NO_CEILING:
            fv = 0
        elif not priced:
            fv = 3
        elif funds > c["fc"]:
            fv = 2
        else:
            fv = 1
        out = u64s(c["out"]) if version == 2 else b""
        raw = (g + u64s(c["jc"], c["fc"]) + c["currency"] + u64s(c["pub"], *c["sub"], c["comp"], c["rec"], c["ev"]) + out
               + u64s(*c["attr"]) + bytes([1 if c["unmetered"] else 0]) + u64s(funds) + bytes([jv, fv]))
        assert len(raw) == (206 if version == 2 else 198)
        c.update(funds=funds, jv=jv, fv=fv, leaf=sha(CONSUMPTION_DOMAIN, raw))
        consumption.append(c)
    return {"records": records, "m": m, "units": units, "totals": totals, "takes": take_digests, "compose_digests": digests,
            "metrics": metrics, "consumption": consumption}


def list_digest(digests):
    return sha(b"".join(digests)) if digests else ZERO


def check_totals(rc, comp):
    """§5.1 step 9."""
    unmetered = sum(1 for x in comp["records"] if x["class"] is None)
    if rc["unmetered"] != unmetered:
        raise Refuse("unit_mismatch")
    tt = comp["totals"]
    for tag, figures, joules in rc["totals"]:
        if tag not in tt:
            raise Refuse("unit_mismatch")
        if joules != tt[tag][1]:
            others = [c for c in tt if c != tag]
            summed = any(tt[tag][1] + sum(tt[o][1] for k, o in enumerate(others) if mask >> k & 1) == joules
                         for mask in range(1, 1 << len(others)))
            raise Refuse("classes_summed" if summed else "unit_mismatch")
        if figures != tt[tag][0]:
            raise Refuse("unit_mismatch")
    if len(rc["totals"]) != len(tt):
        raise Refuse("unit_mismatch")


def record_facts(x):
    """§5.4: a hop record's digest, size, bound and version, and a version 2 record's DVR figures."""
    out = {"digest": x["digest"].hex(), "size": x["size"], "bound": x["bound"], "version": x["version"]}
    if x["dvr"] is not None:
        out.update(x["dvr"])
    return out


def verify_vector(v, doc, cache):
    """§5: the outcome the vector should state."""
    try:
        ctx = cache.ctx(doc["rooms"][v["room"]])
    except Stop as st:
        return {"outcome": "refuse", "reason": st.reason, "at": st.at}
    given_hops = [(bytes.fromhex(h["record"]), bytes.fromhex(h["ledger"])) for h in v["hops"]]
    descriptors = [bytes.fromhex(x) for x in v["descriptors"]]
    compositions = [bytes.fromhex(x) for x in v["compositions"]]
    takes = [bytes.fromhex(x) for x in v["takes"]]
    logs = v["logs"].encode("utf-8") if v["logs"] is not None else None
    try:
        rc = read_receipt(bytes.fromhex(v["receipt"]), ctx.id)
        signer_authorised(ctx, rc)
    except Refuse as err:
        return {"outcome": "refuse", "reason": err.reason, "at": ["receipt"]}
    hops = [(next(((r, l) for r, l in given_hops if given_digest(r) == h["record"]), None), h) for h in rc["hops"]]
    named_takes = [next((t for t in takes if given_digest(t) == d), None) for d in rc["takes"]]
    comps = sorted(compositions, key=given_digest)
    s, e = rc["start"], rc["end"]
    try:
        if len(comps) != rc["compose_count"] or list_digest([given_digest(b) for b in comps]) != rc["compose_digest"]:
            evaluate(ctx, s, e, hops, named_takes, [], descriptors, None, rc["prices"], rc["currency"], rc["version"])
            raise Stop("record_missing", ["compose"])
        named_logs = None
        if rc["qoe_count"] != 0:
            if logs is None or sha(logs) != rc["qoe_digest"]:
                evaluate(ctx, s, e, hops, named_takes, comps, descriptors, None, rc["prices"], rc["currency"], rc["version"])
                raise Stop("record_missing", ["qoe"])
            named_logs = logs
        comp = evaluate(ctx, s, e, hops, named_takes, comps, descriptors, named_logs, rc["prices"], rc["currency"], rc["version"])
        if len(comp["metrics"]) != rc["qoe_count"]:
            raise Stop("unit_mismatch", ["qoe"])
        if comp["units"] != rc["units"]:
            raise Stop("unit_mismatch", ["units"])
        try:
            check_totals(rc, comp)
        except Refuse as err:
            raise Stop(err.reason, ["energy"])
        cell = cell_of(comp["metrics"])
        if len(comp["metrics"]) < QOE_MIN and not rc["cell"]["suppressed"]:
            raise Stop("qoe_not_suppressed", ["qoe"])
        if cell != rc["cell"]:
            raise Stop("unit_mismatch", ["qoe"])
        root = mth([c["leaf"] for c in comp["consumption"]])
        past = [(c["grant"], (c["jv"] == 2) | ((c["fv"] == 2) << 1)) for c in comp["consumption"] if c["jv"] == 2 or c["fv"] == 2]
        if len(comp["consumption"]) != rc["grants"] or root != rc["root"] or past != rc["past"]:
            raise Stop("ceiling_mismatch", ["ceilings"])
    except Stop as st:
        return {"outcome": "refuse", "reason": st.reason, "at": st.at}
    whole = not any(x["class"] is None for x in comp["records"]) and len(comp["records"]) > 0

    def ceiling(x):
        return None if x == NO_CEILING else x
    return {
        "outcome": "accept",
        "digest": rc["digest"].hex(),
        "media_tracks": comp["m"],
        "records": [record_facts(x) for x in comp["records"]],
        "units": comp["units"],
        "unmetered": rc["unmetered"],
        "totals": [{"class": LABELS[t], "figures": f, "joules": j} for t, (f, j) in sorted(comp["totals"].items())],
        "coverage": "whole" if whole else "partial",
        "consumption": [{
            "grant": c["grant"].hex(), "joules_ceiling": ceiling(c["jc"]), "funds_ceiling": ceiling(c["fc"]),
            "currency": None if c["currency"] == b"\x00\x00\x00" else c["currency"].decode(),
            "publisher_us": c["pub"], "subscriber_us": c["sub"], "composition_us": c["comp"], "recorded_us": c["rec"],
            "events_sent": c["ev"], "attributed": {LABELS[t]: c["attr"][t - 2] for t in (2, 3, 4, 5)},
            "unmetered": c["unmetered"], "funds": None if c["funds"] == NO_FUNDS else c["funds"],
            "joules_verdict": JOULES_VERDICT[c["jv"]], "funds_verdict": FUNDS_VERDICT[c["fv"]], "leaf": c["leaf"].hex(),
            **({"bytes_out": c["out"]} if rc["version"] == 2 else {}),
        } for c in comp["consumption"]],
        "consumption_root": root.hex(),
        "past": [{"grant": g.hex(), "flags": f} for g, f in past],
        "qoe_cell": cell,
    }


class Cache:
    """Each room of the corpus verified once, by its content."""

    def __init__(self):
        self.rooms = {}

    def ctx(self, room):
        k = json.dumps(room, sort_keys=True)
        if k not in self.rooms:
            try:
                self.rooms[k] = Ctx(room)
            except Stop as st:
                self.rooms[k] = st
        got = self.rooms[k]
        if isinstance(got, Stop):
            raise got
        return got


OUTCOME_KEYS = ("outcome", "reason", "at", "digest", "media_tracks", "records", "units", "unmetered", "totals", "coverage",
                "consumption", "consumption_root", "past", "qoe_cell")


def check(doc, cache, quiet=False, only=None):
    fails = []
    for v in doc["vectors"]:
        if only is not None and v["name"] not in only:
            continue
        try:
            got = verify_vector(v, doc, cache)
        except Exception as e:  # a crash on a mutated copy is a failure to read it
            got, problems = None, ["the vector does not verify: %s: %s" % (type(e).__name__, e)]
        else:
            problems = [k for k in OUTCOME_KEYS if v.get(k) != got.get(k)]
        if not quiet:
            what = "?" if got is None else got["outcome"] + (" " + got["reason"] if got["outcome"] == "refuse" else "")
            print(("PASS" if not problems else "FAIL") + " vector %s: %s" % (v["name"], what))
        for k in problems:
            fails.append("vector %s: %s is %s, the corpus states %s" % (
                v["name"], k, json.dumps(got.get(k))[:200] if got else k, json.dumps(v.get(k))[:200]))
    return fails


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


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

    def vec(d, name):
        return d["vectors"][idx[name]]

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

    def relay_hop(d, name):
        relay = doc["principals"]["relay"]
        return next(h for h in vec(d, name)["hops"] if h["record"][74:138] == relay)

    def drop_join(d):
        clock = doc["principals"]["relay"]
        chain = next(c for c in d["rooms"]["lesson"]["chains"] if LEV.read_record(bytes.fromhex(c[0]))["principal"].hex() == clock)
        chain.pop(4)

    def cons(d, name, k):
        return vec(d, name)["consumption"][k]

    def dvr_record(d):
        return next(r for r in vec(d, "dvr_relay")["records"] if r["version"] == 2)

    def as_version_1(d):
        r = dvr_record(d)
        for k in DVR_FIELDS:
            del r[k]
        r.update(version=1, size=r["size"] - 48, bound=r["bound"] - 48)

    def dvr_byte(d, k):
        h = relay_hop(d, "dvr_relay")
        n = len(h["record"]) // 2
        h.update(record=flip(h["record"], n - 64 - 48 + k))

    lesson_units = ["lesson", "signed_by_authority"]
    bp = "bytes_out_priced"
    priced = next(k for k, c in enumerate(doc["vectors"][idx[bp]]["consumption"]) if c.get("bytes_out", 0) > 0 and c["funds"] is not None)
    return [
        mutate("bytes_out_priced: bytes out stated one less", [bp], lambda d: vec(d, bp)["units"].update(bytes_out=vec(d, bp)["units"]["bytes_out"] - 1)),
        mutate("bytes_out_priced: bytes out left out of the units", [bp], lambda d: vec(d, bp)["units"].pop("bytes_out")),
        mutate("bytes_out_priced: a grant's bytes out stated one more", [bp], lambda d: cons(d, bp, priced).update(bytes_out=cons(d, bp, priced)["bytes_out"] + 1)),
        mutate("bytes_out_priced: a grant's funds without its bytes", [bp], lambda d: cons(d, bp, priced).update(funds=cons(d, bp, priced)["funds"] - 1)),
        mutate("lesson: bytes out stated by a receipt of version 1", ["lesson"], lambda d: vec(d, "lesson")["units"].update(bytes_out=0)),
        mutate("receipt_unknown_version: stated malformed", ["receipt_unknown_version"], lambda d: vec(d, "receipt_unknown_version").update(reason="malformed")),
        mutate("lesson: a bit of the receipt's signature", ["lesson"], lambda d: vec(d, "lesson").update(receipt=flip(vec(d, "lesson")["receipt"], len(vec(d, "lesson")["receipt"]) // 2 - 1))),
        mutate("lesson: publisher-seconds stated one more", ["lesson"], lambda d: vec(d, "lesson")["units"].update(publisher_us=vec(d, "lesson")["units"]["publisher_us"] + 1)),
        mutate("lesson: recorded-seconds stated one less", ["lesson"], lambda d: vec(d, "lesson")["units"].update(recorded_us=vec(d, "lesson")["units"]["recorded_us"] - 1)),
        mutate("lesson: a byte of the relay's record", ["lesson"], lambda d: relay_hop(d, "lesson").update(record=flip(relay_hop(d, "lesson")["record"], 200))),
        mutate("lesson: a byte of the relay's ledger", ["lesson"], lambda d: relay_hop(d, "lesson").update(ledger=flip(relay_hop(d, "lesson")["ledger"], 60))),
        mutate("lesson: a composition receipt not given", ["lesson"], lambda d: vec(d, "lesson")["compositions"].pop()),
        mutate("lesson: a take not given", ["lesson"], lambda d: vec(d, "lesson")["takes"].pop()),
        mutate("lesson: a log not given", ["lesson"], lambda d: vec(d, "lesson").update(logs=None)),
        mutate("lesson: a stated verdict", ["lesson"], lambda d: cons(d, "lesson", 2).update(joules_verdict="past")),
        mutate("lesson: a stated attributed share", ["lesson"], lambda d: cons(d, "lesson", 3)["attributed"].update(OnChipCounter=1)),
        mutate("lesson: the stated coverage", ["lesson"], lambda d: vec(d, "lesson").update(coverage="partial")),
        mutate("lesson: a record's stated bound", ["lesson"], lambda d: vec(d, "lesson")["records"][0].update(bound=1)),
        mutate("mixed_classes: two totals stated as one class", ["mixed_classes"], lambda d: vec(d, "mixed_classes")["totals"][0].update(**{"class": "OnChipCounter"})),
        mutate("ceilings_past: a grant past stated within", ["ceilings_past"], lambda d: vec(d, "ceilings_past").update(past=vec(d, "ceilings_past")["past"][:1])),
        mutate("large_audience: the QoE cell stated suppressed", ["large_audience"], lambda d: vec(d, "large_audience")["qoe_cell"].update(suppressed=True)),
        mutate("large_audience: an observation's sn", ["large_audience"], lambda d: vec(d, "large_audience").update(logs=vec(d, "large_audience")["logs"].replace('"sn":1,', '"sn":0,', 1))),
        mutate("relay_recorded: a byte of the relay's own record", ["relay_recorded"], lambda d: vec(d, "relay_recorded")["hops"][0].update(record=flip(vec(d, "relay_recorded")["hops"][0]["record"], 120))),
        mutate("relay_recorded: subscriber-seconds stated other", ["relay_recorded"], lambda d: vec(d, "relay_recorded")["units"]["subscriber_us"].__setitem__(0, 1)),
        mutate("track_unknown: stated record_invalid", ["track_unknown"], lambda d: vec(d, "track_unknown").update(reason="record_invalid")),
        mutate("classes_summed: stated unit_mismatch", ["classes_summed"], lambda d: vec(d, "classes_summed").update(reason="unit_mismatch")),
        mutate("record_too_large: stated at another hop", ["record_too_large"], lambda d: vec(d, "record_too_large").update(at=["hop", 0])),
        mutate("lesson: a record stated as version 2", ["lesson"], lambda d: vec(d, "lesson")["records"][0].update(version=2)),
        mutate("dvr_relay: its record stated as version 1, with no DVR figures", ["dvr_relay"], as_version_1),
        mutate("dvr_relay: the stated byte-seconds one more", ["dvr_relay"], lambda d: dvr_record(d).update(dvr_byte_s=dvr_record(d)["dvr_byte_s"] + 1)),
        mutate("dvr_relay: the stated drops left out", ["dvr_relay"], lambda d: dvr_record(d).pop("dvr_dropped")),
        mutate("dvr_relay: the stated bound version 1's", ["dvr_relay"], lambda d: dvr_record(d).update(bound=dvr_record(d)["bound"] - 48)),
        mutate("dvr_relay: a byte of the relay's DVR figures", ["dvr_relay"], lambda d: dvr_byte(d, 16)),
        mutate("dvr_relay: the relay's record without its DVR figures", ["dvr_relay"],
               lambda d: relay_hop(d, "dvr_relay").update(record=relay_hop(d, "dvr_relay")["record"][:-2 * (64 + 48)] + relay_hop(d, "dvr_relay")["record"][-128:])),
        mutate("dvr_out_exceeds: stated unknown_version", ["dvr_out_exceeds"], lambda d: vec(d, "dvr_out_exceeds").update(reason="unknown_version")),
        mutate("dvr_stored_exceeds: stated accepted", ["dvr_stored_exceeds"], lambda d: vec(d, "dvr_stored_exceeds").update(outcome="accept")),
        mutate("record_unknown_version: stated record_invalid", ["record_unknown_version"], lambda d: vec(d, "record_unknown_version").update(reason="record_invalid")),
        mutate("dvr_record_too_large: stated record_invalid", ["dvr_record_too_large"], lambda d: vec(d, "dvr_record_too_large").update(reason="record_invalid")),
        mutate("lesson room: a lifecycle record not delivered", lesson_units, drop_join),
        mutate("lesson room: a byte of a grant", lesson_units, lambda d: d["rooms"]["lesson"]["grants"].__setitem__(4, flip(d["rooms"]["lesson"]["grants"][4], 230))),
    ]


def reasons_listed():
    """The codes §9's table lists, in order."""
    with open(os.path.join(HERE, "..", "extensions", "live-metering.md"), encoding="utf-8") as f:
        text = f.read()
    at = text.index("\n## 9. Reasons")
    end = text.find("\n## ", at + 1)
    section = text[at:end if end != -1 else len(text)]
    return [m.group(1) for m in re.finditer(r"^\| `([a-z_]+)` \|", section, re.M)]


def main(corpus_dir):
    fails = []
    st = ED._self_test()
    print(("PASS" if not st else "FAIL") + " Ed25519 strict rule self-test " + ",".join(st))
    if st:
        fails.append("ed25519 self-test")
    with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as f:
        doc = json.load(f)
    cache = Cache()
    fails += check(doc, cache)
    seen = {v.get("reason") for v in doc["vectors"]}
    missing = [r for r in REASONS if r not in seen]
    if missing:
        fails.append("no vector is refused %s" % ", ".join(missing))
    print("%s every reason of §9 refuses a vector" % ("PASS" if not missing else "FAIL"))
    listed = reasons_listed()
    if listed != REASONS:
        fails.append("§9 lists %s" % listed)
    print("%s §9 lists exactly the reasons this verifier knows" % ("PASS" if listed == REASONS else "FAIL"))
    for label, names, mutated in mutations(doc):
        caught = check(mutated, Cache() if label.startswith("lesson room") else cache, quiet=True, only=set(names))
        print(("PASS" if caught else "FAIL") + " mutated copy fails: %s%s" % (label, " (%s)" % caught[0][:120] if caught else ""))
        if not caught:
            fails.append("a mutated copy verifies: %s" % label)
    for f in fails:
        print("FAIL " + f)
    print(("PASS" if not fails else "FAIL") + " live-metering corpus: %d vectors" % len(doc["vectors"]))
    return 1 if fails else 0


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