#!/usr/bin/env python3
"""Independent verifier of the remote desktop corpus (extensions/remote-desktop.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), written from its text alone. Seals are opened with the HPKE base mode, X25519,
HKDF-SHA256 and AES-128-GCM of the aggregation corpus's second implementation,
telemetry_aggregate_verify.py, written from RFC 9180, RFC 7748, RFC 5869, FIPS 197 and SP
800-38D, itself checked against RFC 9180 A.1.1 here. Everything else is here:

- the grant attributes `remote_host` and `remote_control` (§3.1);
- the screen track's digest, the screen statement, its rules and its chain (§3.2 to §3.4);
- reading every statement and who signs it (§4);
- input statements and their actions (§5.1, §5.2), and the host's rules (§5.3);
- sealed actions (§5.5): the seal's form, its opening under the host's opening key, each
  opening read and each commitment recomputed, at the opened level (§11.4);
- outcomes and their checks, the rules recomputed (§6);
- floor statements and the floor in effect (§7): a floor statement §7.2 refuses is stated with
  its reason and changes nothing;
- latency records, their checks and their legs (§8);
- what a verification states (§11.3).

For every room it requires the outcome, reason and place, and everything else the corpus
states, at the commitment level and, under the corpus's opening key, at the opened level; of
the corpus, that every reason of the four tables of §12 appears in it and that no other file is
there. Then it mutates copies of the corpus, one defect each, and requires every copy to
fail; and it drops each of its own checks in turn and requires the corpus to catch the weakened
verifier.

    python3 remote_desktop_verify.py <live-conformance/remote>
    python3 remote_desktop_verify.py --room <room.json> [--opening-key <hex>]...

With `--room`, it verifies one room given in a vector's form (descriptor, grants, chains) and
prints what §11.3 states of it, as JSON: a session a deployment kept, checked from the text
alone; given opening keys (X25519 private keys, hexadecimal), at the opened level.

Exit status 0 only if every case verifies, every mutated copy fails and every weakened verifier
is caught. 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 telemetry_aggregate_verify as HPKE  # noqa: E402
import _ed25519 as ED  # noqa: E402

Refuse = LEV.Refuse

_verified = {}
_verify = ED.verify


def _memo_verify(pk, msg, sig, skip=()):
    k = (bytes(pk), bytes(msg), bytes(sig), tuple(skip))
    if k not in _verified:
        _verified[k] = _verify(pk, msg, sig, skip)
    return _verified[k]


ED.verify = _memo_verify
EXTENSION = os.path.join(HERE, "..", "extensions", "remote-desktop.md")

MAGICS = {b"WDS1": "screen", b"WDI1": "input", b"WDO1": "outcome", b"WDF1": "floor", b"WDL1": "latency"}
LATENCY_DOMAIN = b"wai:remote-latency\x01"                     # §8.1
SEAL_INFO = b"wai:remote-seal\x01"                             # §5.5
COMMIT_DOMAIN = b"wai:remote-commit\x01"                       # §5.5
SALT, TAG, TEXT_PAD = 16, 16, 64                                # §5.5
ZERO = bytes(32)
NO_RTT = (1 << 32) - 1
META, JOIN, LEAVE, EXCHANGE = 0x01, 0x10, 0x11, 0x14
DIRECT = 2                                                     # live-rooms §3.3, bit 1
SCREEN_PUBLISH = 4                                             # live-rooms §3.3, publish bit 2
TYPES = {1: "move", 2: "button", 3: "wheel", 4: "key", 5: "text", 6: "touch", 7: "clipboard"}
CLASS_OF = {1: 1, 2: 1, 3: 1, 4: 2, 5: 2, 6: 4, 7: 8}           # §5.2: pointer, keys, touch, clipboard
SEALED = 8                                                     # §5.2: a sealed action
SEALABLE = (4, 5, 7)                                           # §5.5: key, text, clipboard
CLASS_NAMES = ((1, "pointer"), (2, "keys"), (4, "touch"), (8, "clipboard"))
REQUEST, GIVE, RELEASE, TAKE = 1, 2, 3, 4

DROP = set()


def on(name):
    return name not in DROP


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 tables_of_extension():
    """§12: the codes of the first table (they refuse a room), the numbered reasons of the second
    (a host states them), the codes of the third (a floor statement refused, which changes
    nothing), and those of the fourth (why a seal is not opened, §5.5)."""
    with open(EXTENSION, encoding="utf-8") as f:
        text = f.read()
    sec = text[text.index("\n## 12. Reasons"):]
    sec = sec[:sec.index("\n## ", 1)]
    tables = re.findall(r"((?:^\|.*\n)+)", sec, re.M)
    if len(tables) != 4:
        return [], {}, [], []
    codes = re.findall(r"^\| `([a-z_0-9]+)` \|", tables[0], re.M)
    numbered = {int(n): c for n, c in re.findall(r"^\| ([0-9]+) \| `([a-z_0-9]+)` \|", tables[1], re.M)}
    floor_codes = re.findall(r"^\| `([a-z_0-9]+)` \|", tables[2], re.M)
    seal_codes = re.findall(r"^\| `([a-z_0-9]+)` \|", tables[3], re.M)
    return codes, numbered, floor_codes, seal_codes


REASONS, HOST_REASONS, FLOOR_REASONS, SEAL_REASONS = tables_of_extension()
ACTION_NAMES = {1: "request", 2: "give", 3: "release", 4: "take"}


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

    def take(self, n):
        if self.p + n > len(self.b):
            raise Refuse("rd_statement_malformed")
        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 i16(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 short(self):
        return self.take(self.u8())


def bad(cond):
    if cond:
        raise Refuse("rd_statement_malformed")


def utf8(b):
    """RFC 3629: Python's strict decoder refuses overlong forms, surrogates and anything above
    U+10FFFF."""
    try:
        return b.decode("utf-8")
    except UnicodeDecodeError:
        raise Refuse("rd_statement_malformed")


def token(b):
    """live-rooms §2: one or more bytes of 0x21 to 0x7E."""
    return len(b) > 0 and all(0x21 <= c <= 0x7E for c in b)


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

def attrs_of(g):
    """§3.1: (hosts, control classes); Refuse when either attribute is malformed."""
    host, control = False, 0
    for k, v in LRV.read_attrs(g["attrs"]) or []:
        if k == "remote_host":
            if v not in ("true", "false"):
                raise Refuse("rd_attr_malformed")
            host = v == "true"
        elif k == "remote_control":
            if not re.fullmatch(r"[1-9][0-9]?", v) or int(v) > 15:
                raise Refuse("rd_attr_malformed")
            control = int(v)
    return host, control


def screen_track(room, host):
    """§3.2: SHA-256 over `<room>-<host>-screen`, a zero byte, and `0`."""
    return sha(("%s-%s-screen" % (room.hex(), host.hex())).encode(), b"\x00", b"0")


def rect(r):
    return {"x": r.u16(), "y": r.u16(), "w": r.u16(), "h": r.u16()}


def read_screen(r):
    """§3.3."""
    s = {"prev": r.take(32), "state": r.u8(), "track": r.take(32), "width": r.u16(), "height": r.u16(), "fps": r.u16(),
         "scope": r.u8(), "region": rect(r)}
    s["redactions"] = [rect(r) for _ in range(r.u8())]
    s.update(pointer=r.u8(), accepts=r.u8(), clip_max=r.u16(), max_age_ms=r.u32(), floor=r.u8())
    s["capability"], s["codec"], s["encoder"] = r.short(), r.short(), r.short()
    s["api"], s["hardware"] = r.u8(), r.u8()
    # §3.3: the statement ends after `hardware`, or after a 32-byte sealing key; read_statement
    # refuses any other byte after its last field.
    s["seal"] = r.take(32) if len(r.b) - r.p == 32 else None
    if on("screen_fields"):
        reg = s["region"]
        bad(s["state"] not in (1, 2) or s["width"] == 0 or s["height"] == 0 or s["fps"] == 0 or s["scope"] not in (1, 2, 3))
        bad((reg["w"] == 0 or reg["h"] == 0) if s["scope"] == 3 else any(reg.values()))
        bad(len(s["redactions"]) > 16)
        for x in s["redactions"]:
            bad(x["w"] == 0 or x["h"] == 0 or x["x"] + x["w"] > s["width"] or x["y"] + x["h"] > s["height"])
        bad(s["pointer"] > 1 or s["accepts"] & ~0x7F or (s["clip_max"] > 0) != bool(s["accepts"] & 0x40))
        bad(s["max_age_ms"] == 0 or s["floor"] not in (1, 2) or s["api"] not in range(1, 6) or s["hardware"] > 2)
        bad(not token(s["capability"]) or not token(s["codec"]))
        e = s["encoder"]
        bad(not token(e) or e.count(b"@") != 1 or e[:1] == b"@" or e[-1:] == b"@")
    return s


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

def read_action(r):
    """§5.2: (type, fields)."""
    start = r.p
    ty, a = read_action_fields(r)
    a["raw"] = r.b[start:r.p]
    return ty, a


def read_action_fields(r):
    ty = r.u8()
    if ty == 1:
        return ty, {"x": r.u16(), "y": r.u16()}
    if ty == 2:
        a = {"button": r.u8(), "down": r.u8(), "x": r.u16(), "y": r.u16()}
        bad(a["button"] > 4 or a["down"] > 1)
        return ty, a
    if ty == 3:
        a = {"dx": r.i16(), "dy": r.i16(), "x": r.u16(), "y": r.u16()}
        bad(a["dx"] == 0 and a["dy"] == 0)
        return ty, a
    if ty == 4:
        a = {"usage": r.u16(), "down": r.u8(), "modifiers": r.u8()}
        bad(not 0x04 <= a["usage"] <= 0xE7 or a["down"] > 1 or a["modifiers"] & ~0x0F)
        return ty, a
    if ty == 5:
        b = r.short()
        bad(not 1 <= len(b) <= 64)
        return ty, {"text": utf8(b)}
    if ty == 6:
        a = {"id": r.u8(), "phase": r.u8(), "x": r.u16(), "y": r.u16()}
        bad(not 1 <= a["phase"] <= 4)
        return ty, a
    if ty == 7:
        b = r.take(r.u16())
        return ty, {"text": utf8(b), "bytes": len(b)}
    if ty == SEALED:
        a = {"of": r.u8(), "length": r.u16(), "commit": r.take(32)}
        if on("sealed_of"):
            bad(a["of"] not in SEALABLE or (a["of"] != 7 and a["length"] != 0))
        return ty, a
    raise Refuse("rd_statement_malformed")


def opening_len(ty, a):
    """§5.5: the length of a sealed action's opening: the salt, then 4 for a key, 1 + 64 for a
    text, a clipboard text's `length`."""
    if ty != SEALED:
        return 0
    return SALT + {4: 4, 5: 1 + TEXT_PAD}.get(a["of"], a["length"])


def acts_as(ty, a):
    """§5.2: a sealed action is of the type it seals."""
    return a["of"] if ty == SEALED else ty


def read_input(r):
    """§5.1, and the seal of §5.5."""
    i = {"host": r.take(32), "screen": r.take(32), "actions": [], "seal": None}
    n = r.u8()
    bad(not 1 <= n <= 32)
    for _ in range(n):
        dt = r.u32()
        if on("dt_order"):
            bad(i["actions"] and dt > i["actions"][-1][0])
        ty, a = read_action(r)
        i["actions"].append((dt, ty, a))
    i["clear"] = r.b[:r.p]
    if any(ty == SEALED for _, ty, _ in i["actions"]):
        enc = r.take(32)
        want = TAG + sum(opening_len(ty, a) for _, ty, a in i["actions"])
        if on("seal_length"):
            bad(len(r.b) - r.p != want)
        i["seal"] = (enc, r.take(len(r.b) - r.p))
    return i


def read_outcome(r):
    """§6.1."""
    o = {"input": r.take(32), "outcome": r.u8(), "reason": r.u8(), "applied_at": r.u64(), "captured_at": r.u64(),
         "content": r.take(32), "rtt_us": r.u32()}
    bad(o["outcome"] not in (1, 2) or o["reason"] > 9)
    return o


def read_floor(r):
    """§7.2."""
    f = {"host": r.take(32), "action": r.u8(), "to": r.take(32)}
    bad(f["action"] not in (1, 2, 3, 4) or (f["to"] != ZERO) != (f["action"] == GIVE))
    return f


def read_latency(r):
    """§8.1."""
    start = r.p
    l = {"room": r.take(32), "viewer": r.take(32), "host": r.take(32), "input": r.take(32), "action": r.u8(),
         "outcome": r.take(32), "track": r.take(32), "group": r.u64(), "object": r.u64(), "content": r.take(32),
         "input_at": r.u64(), "presented_at": r.u64(), "source": r.u8(), "latency_us": r.u32(), "rtt_us": r.u32()}
    l["body"] = r.b[:r.p]
    l["sig"] = r.take(64)
    bad(l["source"] not in (1, 2, 3))
    del start
    return l


def read_statement(payload):
    """§4.1, §4.2: None for a payload not this extension's; else (kind, fields)."""
    kind = MAGICS.get(payload[:4])
    if kind is None:
        return None
    r = Rd(payload)
    r.take(4)
    bad(r.u8() != 1)
    f = {"screen": read_screen, "input": read_input, "outcome": read_outcome, "floor": read_floor, "latency": read_latency}[kind](r)
    bad(r.p != len(payload))
    return kind, f


def coords(ty, a):
    return (a["x"], a["y"]) if ty in (1, 2, 3, 6) else None


def rules(inp, screen_digest, screen, control, holds_floor, me):
    """§5.3 steps 1 to 6 at the input's place: the first reason's number, or 0."""
    if on("rule_screen") and (inp["host"] != me or screen is None or inp["screen"] != screen_digest):
        return 1
    if screen is None:
        return 1
    classes = 0
    for _, ty, a in inp["actions"]:
        classes |= CLASS_OF[acts_as(ty, a)]
    if on("rule_power") and classes & ~control:
        return 2
    if on("rule_floor") and screen["floor"] == 2 and not holds_floor:
        return 3
    if on("rule_paused") and screen["state"] == 2:
        return 4
    if on("rule_accepts") and any(not screen["accepts"] >> (acts_as(ty, a) - 1) & 1 for _, ty, a in inp["actions"]):
        return 5
    for _, ty, a in inp["actions"]:
        c = coords(ty, a)
        clip = a["bytes"] if ty == 7 else a["length"] if ty == SEALED and a["of"] == 7 else 0
        if on("rule_bounds") and ((c and (c[0] >= screen["width"] or c[1] >= screen["height"])) or clip > screen["clip_max"]):
            return 6
    return 0


def unsealed(inp, screen):
    """§5.3 step 8: a key, text or clipboard action in the clear to a screen with a sealing key."""
    return on("rule_unsealed") and screen is not None and screen["seal"] is not None and any(ty in SEALABLE for _, ty, _ in inp["actions"])


# ---------------------------------------------------------------- §5.5

_opened = {}


def open_seal(inp, room, author, screen, keys):
    """§5.5, rule 9 of §5.3, against the screen statement the input names: ("opened", the opened
    statement), (a reason of §12's fourth table, None), or ("unknown", None) when no opening key
    for its sealing key is given (the commitment level gives none)."""
    if screen is None or screen["seal"] is None:
        return ("seal_no_key", None) if on("seal_no_key") else ("unknown", None)
    sk = next((k for k in keys if HPKE.x25519(k, HPKE.BASE) == screen["seal"]), None)
    if sk is None:
        return "unknown", None
    enc, ct = inp["seal"]
    memo = (sk, enc, ct, room, author, inp["clear"])
    if memo in _opened and not DROP:
        return _opened[memo]
    out = _open(inp, room, author, sk, enc, ct)
    if not DROP:
        _opened[memo] = out
    return out


def _open(inp, room, author, sk, enc, ct):
    info = SEAL_INFO + room + inp["host"]
    aad = author + inp["clear"]
    try:
        pt = HPKE.hpke_open(enc, sk, info, aad, ct)
    except ValueError:
        if on("seal_open"):
            return "seal_not_opened", None
        return "opened", inp["clear"]
    at, actions, commits = 0, [], []
    for k, (dt, ty, a) in enumerate(inp["actions"]):
        if ty != SEALED:
            actions.append(a["raw"])
            continue
        n = opening_len(ty, a)
        opening = pt[at:at + n]
        at += n
        f = opening[SALT:]
        raw = read_opening(a["of"], f, len(opening) == n)
        if raw is None:
            if on("seal_read"):
                return "seal_unreadable", None
            raw = bytes([a["of"]])
        actions.append(raw)
        commits.append((k, dt, a, opening))
    if on("seal_read") and at != len(pt):
        return "seal_unreadable", None
    for k, dt, a, opening in commits:
        c = sha(COMMIT_DOMAIN, room, author, inp["host"], inp["screen"], bytes([k]), struct.pack("<I", dt), bytes([a["of"]]), opening)
        if on("seal_commit") and c != a["commit"]:
            return "seal_commitment", None
    opened = inp["clear"][:70] + b"".join(struct.pack("<I", dt) + raw for (dt, _, _), raw in zip(inp["actions"], actions))
    return "opened", opened


def read_opening(of, f, whole):
    """§5.5: the action an opening's fields (after the salt) read as, in its clear form (§5.2), or
    None when they do not read."""
    if not whole:
        return None
    if of == 4:
        usage, down, mods = struct.unpack("<HBB", f)
        if not 0x04 <= usage <= 0xE7 or down > 1 or mods & ~0x0F:
            return None
        return bytes([4]) + f
    if of == 5:
        n = f[0]
        if not 1 <= n <= 64 or any(f[1 + n:]):
            return None
        try:
            f[1:1 + n].decode("utf-8")
        except UnicodeDecodeError:
            return None
        return bytes([5, n]) + f[1:1 + n]
    try:
        f.decode("utf-8")
    except UnicodeDecodeError:
        return None
    return bytes([7]) + struct.pack("<H", len(f)) + f


def stale(inp, t, screen, n):
    """§5.3 step 7."""
    first = t - inp["actions"][0][0]
    return on("rule_stale") and n - first > screen["max_age_ms"] * 1000


# ---------------------------------------------------------------- §11

def verify(descriptor, grants, chains, keys=None):
    """§11.2: a dict of what the corpus states (§11.3), at the commitment level, or with opening
    keys (X25519 private keys) at the opened level (§11.4)."""
    room = LRV.verify_room(descriptor, grants, None, chains)                              # step 1
    if room["outcome"] == "refuse":
        return {"outcome": "refuse", "reason": room["reason"], "at": ["room"] + room["at"]}
    gap = room["outcome"] == "gap"
    d = LRV.read_descriptor(descriptor)
    room_id = d["room"]
    held_grants, _ = LRV.check_grants(d, grants)
    room_keys = (d["authority"], d["clock"])
    try:
        attrs = {}
        for i, g in enumerate(held_grants):                                              # step 2
            try:
                host, control = attrs_of(g)
            except Refuse as e:
                if on("attrs"):
                    raise Stop(e.reason, ["grant", i])
                host, control = False, 0
            attrs[g["digest"]] = {"host": host and g["publish"] & SCREEN_PUBLISH != 0, "control": control,
                                  "direct": g["powers"] & DIRECT != 0}
        records, seen = [], set()
        for c, i, _ in room["order"]:
            ev = LEV.read_record(chains[c][i])
            if ev["digest"] in seen:
                continue
            seen.add(ev["digest"])
            ev["chain"], ev["index"] = c, i
            records.append(ev)
        in_effect, hosts = {}, {}
        screens, inputs, floor, floor_refused, later = [], [], [], [], []
        named = {}

        def attrs_now(p):
            g = in_effect.get(p)
            return attrs.get(g, {"host": False, "control": 0, "direct": False})

        def lose(p, t):
            for h, st in hosts.items():
                if st["holder"] == p:
                    st["holder"] = None
                    floor.append({"host": h.hex(), "t": t, "holder": None})

        for ev in records:                                                               # step 3
            t, author = ev["t"], ev["principal"]
            if author in room_keys and ev["kind"] in (JOIN, LEAVE, EXCHANGE):
                pl = LEV.read_payload(ev["kind"], ev["payload"], 1 << 20)
                x = LRV.read_detail(ev["kind"], pl["detail"])
                who = pl["subject"]
                if ev["kind"] == JOIN:
                    in_effect[who] = x["grant"]
                elif ev["kind"] == EXCHANGE:
                    in_effect[who] = x["new"]
                    if on("floor_loss") and attrs_now(who)["control"] == 0:
                        lose(who, t)
                else:
                    in_effect.pop(who, None)
                    if on("floor_loss"):
                        lose(who, t)
                        if who in hosts and hosts[who]["holder"] is not None:
                            hosts[who]["holder"] = None
                            floor.append({"host": who.hex(), "t": t, "holder": None})
                continue
            if ev["kind"] != META:
                continue
            at = ["record", ev["chain"], ev["index"]]
            try:
                s = read_statement(ev["payload"])
                if s is None:
                    continue
                kind, f = s
                is_key = author in room_keys
                me = attrs_now(author)
                if kind == "screen":
                    if on("who_screen") and (is_key or not me["host"]):
                        raise Refuse("rd_not_permitted")
                    if on("track") and f["track"] != screen_track(room_id, author):
                        raise Refuse("screen_wrong_track")
                    st = hosts.setdefault(author, {"screen": None, "digest": ZERO, "holder": None})
                    if on("screen_link") and f["prev"] != st["digest"]:
                        raise Refuse("screen_unlinked")
                    st["screen"], st["digest"] = f, ev["digest"]
                    named[(author, ev["digest"])] = f
                    screens.append({"digest": ev["digest"].hex(), "host": author.hex(), "t": t,
                                    "state": "live" if f["state"] == 1 else "paused", "width": f["width"], "height": f["height"],
                                    "fps": f["fps"], "scope": f["scope"], "floor": "open" if f["floor"] == 1 else "floor",
                                    "accepts": [TYPES[k] for k in range(1, 8) if f["accepts"] >> (k - 1) & 1],
                                    "capability": f["capability"].decode(), "codec": f["codec"].decode(), "encoder": f["encoder"].decode(),
                                    "api": f["api"], "hardware": f["hardware"]})
                elif kind == "input":
                    if on("who_input") and is_key:
                        raise Refuse("rd_not_permitted")
                    st = hosts.get(f["host"])
                    scr = st["screen"] if st else None
                    first = rules(f, st["digest"] if st else None, scr, me["control"], bool(st) and st["holder"] == author, f["host"])
                    classes = 0
                    for _, ty, a in f["actions"]:
                        classes |= CLASS_OF[acts_as(ty, a)]
                    # Rule 9 against the screen statement the input names (§5.5).
                    seal = None
                    if f["seal"] is not None:
                        seal = open_seal(f, room_id, author, named.get((f["host"], f["screen"])), keys or [])
                    inputs.append({"digest": ev["digest"], "author": author, "t": t, "input": f, "first": first,
                                   "unsealed": unsealed(f, scr), "seal": seal,
                                   "sealed": sum(1 for _, ty, _ in f["actions"] if ty == SEALED),
                                   "screen": scr, "classes": classes, "fate": {"outcome": "unanswered"}})
                elif kind == "floor":
                    # §7.2: a statement its rules refuse is recorded, stated with its reason, and
                    # changes nothing; the floor is computed without it (§7.3, §11.3).
                    st = hosts.get(f["host"])
                    a = f["action"]
                    why = None
                    if on("floor_host") and (st is None or st["screen"] is None):
                        why = "floor_not_permitted"
                    else:
                        st = st or {"holder": None}
                        steward = is_key or author == f["host"] or me["direct"]
                        if a == REQUEST:
                            if on("floor_request") and (is_key or me["control"] == 0):
                                why = "floor_not_permitted"
                        elif a == GIVE:
                            if on("floor_give") and not (steward or st["holder"] == author):
                                why = "floor_not_permitted"
                            elif on("floor_eligible") and (f["to"] == f["host"] or attrs_now(f["to"])["control"] == 0):
                                why = "floor_not_eligible"
                        elif a == RELEASE:
                            if on("floor_release") and st["holder"] != author:
                                why = "floor_not_holder"
                        elif on("floor_take") and not steward:
                            why = "floor_not_permitted"
                    if why is not None:
                        floor_refused.append({"digest": ev["digest"].hex(), "author": author.hex(), "host": f["host"].hex(), "t": t,
                                              "action": ACTION_NAMES[a], "reason": why})
                    elif a == GIVE:
                        st["holder"] = f["to"]
                        floor.append({"host": f["host"].hex(), "t": t, "holder": f["to"].hex()})
                    elif a == RELEASE:
                        st["holder"] = None
                        floor.append({"host": f["host"].hex(), "t": t, "holder": None})
                    elif a == TAKE and st["holder"] is not None:
                        st["holder"] = None
                        floor.append({"host": f["host"].hex(), "t": t, "holder": None})
                elif kind == "outcome":
                    later.append(("outcome", at, ev, f))
                else:
                    if on("who_latency") and is_key:
                        raise Refuse("rd_not_permitted")
                    if f["room"] != room_id:
                        raise Refuse("wrong_room")
                    if on("latency_key") and f["viewer"] != author:
                        raise Refuse("latency_wrong_key")
                    if on("latency_sig"):
                        why = ED.verify(f["viewer"], LATENCY_DOMAIN + f["body"], f["sig"])
                        if why:
                            raise Refuse(why)
                    later.append(("latency", at, ev, f))
            except Refuse as e:
                raise Stop(e.reason, at)
        by_input = {x["digest"]: x for x in inputs}
        outcomes = {ev["digest"]: (ev, f) for k, _, ev, f in later if k == "outcome"}
        answered, latencies = set(), []
        for k, at, ev, f in later:                                                       # §6.2, §8.2
            try:
                if k == "outcome":
                    x = by_input.get(f["input"])
                    if x is None:
                        if gap:
                            continue
                        raise Refuse("outcome_unknown_input")
                    if on("who_outcome") and ev["principal"] != x["input"]["host"]:
                        raise Refuse("rd_not_permitted")
                    if on("duplicate") and f["input"] in answered:
                        raise Refuse("outcome_duplicate")
                    answered.add(f["input"])
                    an = ev["anchor"]
                    if f["outcome"] == 1:
                        ok = (f["reason"] == 0 and an is not None and an[0] == screen_track(room_id, x["input"]["host"]) and an[2] == 0
                              and f["content"] != ZERO and x["t"] <= f["applied_at"] <= f["captured_at"] <= ev["t"])
                    else:
                        ok = (1 <= f["reason"] <= 9 and an is None and f["applied_at"] == 0 and f["captured_at"] == 0
                              and f["content"] == ZERO and ev["t"] >= x["t"])
                    if on("consistent") and not ok:
                        raise Refuse("outcome_invalid")
                    want = x["first"]
                    if want == 0 and stale(x["input"], x["t"], x["screen"], ev["t"]):
                        want = 7
                    if want == 0 and x["unsealed"]:
                        want = 8
                    # Rule 9: made where the seal's result is known, and otherwise either (§6.2 step 5).
                    allowed = {want}
                    if want == 0 and x["seal"] is not None:
                        allowed = {9} if x["seal"][0] not in ("opened", "unknown") else {0} if x["seal"][0] == "opened" else {0, 9}
                    stated = 0 if f["outcome"] == 1 else f["reason"]
                    if on("recompute") and stated not in allowed:
                        raise Refuse("outcome_mismatch")
                    x["fate"] = ({"outcome": "applied", "group": an[1], "object": an[2]} if f["outcome"] == 1 and an
                                 else {"outcome": "refused", "reason": HOST_REASONS.get(f["reason"], "?")})
                else:
                    x = by_input.get(f["input"])
                    if x is None and gap:
                        latencies.append({"digest": sha(f["body"]).hex(), "viewer": f["viewer"].hex(), "source": f["source"],
                                          "latency_us": f["latency_us"], "legs": "unchecked"})
                        continue
                    if on("latency_input") and (x is None or x["author"] != f["viewer"] or x["input"]["host"] != f["host"]
                                                or f["action"] >= len(x["input"]["actions"])):
                        raise Refuse("latency_unknown_input")
                    o = outcomes.get(f["outcome"])
                    if o is None and gap:
                        latencies.append({"digest": sha(f["body"]).hex(), "viewer": f["viewer"].hex(), "source": f["source"],
                                          "latency_us": f["latency_us"], "legs": "unchecked"})
                        continue
                    dt = x["input"]["actions"][f["action"]][0]
                    if on("latency_frame") and not (o is not None and o[1]["input"] == f["input"] and o[1]["outcome"] == 1
                                                    and o[0]["anchor"] == (f["track"], f["group"], f["object"])
                                                    and o[1]["content"] == f["content"] and f["input_at"] == x["t"] - dt):
                        raise Refuse("latency_mismatch")
                    if on("latency_ahead") and f["presented_at"] > ev["t"]:
                        raise Refuse("latency_ahead")
                    if on("latency_sum") and (f["presented_at"] < f["input_at"] or f["presented_at"] - f["input_at"] != f["latency_us"]):
                        raise Refuse("latency_mismatch")
                    oo = o[1]
                    unc = None if f["rtt_us"] == NO_RTT or oo["rtt_us"] == NO_RTT else (f["rtt_us"] + oo["rtt_us"]) // 2
                    latencies.append({"digest": sha(f["body"]).hex(), "viewer": f["viewer"].hex(), "source": f["source"],
                                      "latency_us": f["latency_us"],
                                      "legs": {"up": oo["applied_at"] - f["input_at"], "host": oo["captured_at"] - oo["applied_at"],
                                               "down": f["presented_at"] - oo["captured_at"], "uncertainty": unc}})
            except Refuse as e:
                raise Stop(e.reason, at)
    except Stop as s:
        return {"outcome": "refuse", "reason": s.reason, "at": s.at}
    totals = {}
    for x in inputs:
        if gap and x["fate"]["outcome"] == "unanswered":
            x["fate"] = {"outcome": "unchecked"}
        h = x["input"]["host"].hex()
        tt = totals.setdefault(h, {"host": h, "applied": 0, "refused": {HOST_REASONS[n]: 0 for n in sorted(HOST_REASONS)},
                                   "unanswered": 0, "unchecked": 0})
        fo = x["fate"]["outcome"]
        if fo == "refused":
            tt["refused"][x["fate"]["reason"]] += 1
        else:
            tt[fo] += 1
    # §11.3: rules 1 to 7 always, rules 8 and 9 where they refused.
    for tt in totals.values():
        tt["refused"] = {k: v for k, v in tt["refused"].items() if v or k not in (HOST_REASONS.get(8), HOST_REASONS.get(9))}
    out = {"outcome": "gap" if gap else "accept", "room": room_id.hex(),
           "gaps": room.get("gaps", []) if gap else [],
           "screens": screens,
           "inputs": [input_fact(x, keys is not None) for x in inputs],
           "floor": floor, "latencies": latencies, "totals": [totals[h] for h in sorted(totals)]}
    if floor_refused:
        out["floor_refused"] = floor_refused
    return out


def input_fact(x, opened_level):
    """§11.3: an input as a verification states it; at the opened level, a sealed input's seal."""
    o = {"digest": x["digest"].hex(), "author": x["author"].hex(), "host": x["input"]["host"].hex(), "t": x["t"],
         "classes": [n for b, n in CLASS_NAMES if x["classes"] & b]}
    if x["sealed"]:
        o["sealed"] = x["sealed"]
    o["fate"] = x["fate"]
    if opened_level and x["seal"] is not None:
        why, opened = x["seal"]
        o["seal"] = "unopened" if why == "unknown" else why
        if why == "opened":
            o["opened"] = opened.hex()
    return o


def run_vector(v, keys=None):
    return verify(bytes.fromhex(v["descriptor"]), [bytes.fromhex(g) for g in v["grants"]], [[bytes.fromhex(r) for r in c] for c in v["chains"]], keys)


STATED = ("outcome", "reason", "at", "room", "gaps", "screens", "inputs", "floor", "floor_refused", "latencies", "totals")


def given(got, seen):
    if got.get("outcome") == "refuse":
        seen.add(got["reason"])
    for x in got.get("inputs", []):
        if x["fate"]["outcome"] == "refused":
            seen.add(x["fate"]["reason"])
        if x.get("seal", "opened") not in ("opened", "unopened"):
            seen.add(x["seal"])
    for x in got.get("floor_refused", []):
        seen.add(x["reason"])


def check(doc, quiet=False, first=False):
    """Every vector of `doc` against this verifier, at both levels: the failures (with `first`, the
    first alone), and every reason the corpus gives."""
    fails, seen = [], set()
    try:
        keys = [bytes.fromhex(doc["opening_key"])]
    except (KeyError, ValueError, TypeError):
        return ["the corpus states no opening key"], seen
    for v in doc["vectors"]:
        if first and fails:
            return fails, seen
        try:
            got = run_vector(v)
            opened = run_vector(v, keys)
        except Exception as e:  # a copy that does not read is a failure to verify it
            fails.append("%s: does not verify (%r)" % (v["name"], e))
            continue
        badk = [k for k in STATED if (k in got or k in v) and got.get(k) != v.get(k)]
        # The opened level: what the vector states of it, or, where it states nothing of it, what
        # the commitment level gives (§13).
        want = v.get("opened", got)
        bado = ["opened." + k for k in STATED if (k in opened or k in want) and opened.get(k) != (want.get(k) if isinstance(want, dict) else None)]
        given(got, seen)
        given(opened, seen)
        if not quiet:
            what = got["outcome"] + ("" if got["outcome"] != "refuse" else " " + got["reason"] + " at %s" % got["at"])
            if "opened" in v:
                what += "; opened: " + opened["outcome"] + ("" if opened["outcome"] != "refuse" else " " + opened["reason"] + " at %s" % opened["at"])
            print(("PASS" if not badk and not bado else "FAIL") + " room %s: %s" % (v["name"], what))
        for k in badk:
            fails.append("%s: %s is %s, the corpus states %s" % (v["name"], k, json.dumps(got.get(k))[:200], json.dumps(v.get(k))[:200]))
        for k in bado:
            fails.append("%s: %s differs" % (v["name"], k))
    return fails, seen


def mutations(doc):
    """Copies of the corpus, each with one defect a conforming verifier must catch."""
    vn = {v["name"]: k for k, v in enumerate(doc["vectors"])}
    fs, op, gp = vn["floor_session"], vn["open_handover"], vn["floor_session_gap"]

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

    def vs(d):
        return d["vectors"]

    def flip(h, i):
        b = bytearray(bytes.fromhex(h))
        b[i] ^= 1
        return b.hex()

    def flip_record(d, k, chain, rec, i):
        vs(d)[k]["chains"][chain][rec] = flip(vs(d)[k]["chains"][chain][rec], i)

    host = doc["principals"]["host"]
    ss = vn["sealed_session"]
    return [
        mutate("an applied input stated refused", lambda d: vs(d)[fs]["inputs"][0].update(fate={"outcome": "refused", "reason": "input_no_floor"})),
        mutate("a refusal's stated reason", lambda d: vs(d)[fs]["inputs"][3]["fate"].update(reason="input_not_permitted")),
        mutate("an unanswered input stated applied", lambda d: vs(d)[fs]["inputs"][10].update(fate={"outcome": "applied", "group": 9, "object": 0})),
        mutate("a stated frame", lambda d: vs(d)[fs]["inputs"][1]["fate"].update(group=7)),
        mutate("a stated floor holder", lambda d: vs(d)[fs]["floor"][1].update(holder=host)),
        mutate("a floor change unstated", lambda d: vs(d)[fs]["floor"].pop()),
        mutate("a stated leg", lambda d: vs(d)[fs]["latencies"][0]["legs"].update(up=vs(d)[fs]["latencies"][0]["legs"]["up"] + 1)),
        mutate("a stated uncertainty", lambda d: vs(d)[fs]["latencies"][1]["legs"].update(uncertainty=0)),
        mutate("a stated total", lambda d: vs(d)[fs]["totals"][0].update(applied=5)),
        mutate("a stated encoder", lambda d: vs(d)[fs]["screens"][0].update(encoder="another@2")),
        mutate("the handover's second applied input stated refused", lambda d: vs(d)[op]["inputs"][3].update(fate={"outcome": "refused", "reason": "input_not_permitted"})),
        mutate("the gap's unchecked input stated unanswered", lambda d: vs(d)[gp]["inputs"][-3].update(fate={"outcome": "unanswered"})),
        mutate("a byte of a latency record's own signature", lambda d: flip_record(d, fs, 2, 1, -70)),
        mutate("a refusal's reason", lambda d: vs(d)[vn["applies_stale"]].update(reason="outcome_invalid")),
        mutate("a refusal's place", lambda d: vs(d)[vn["screen_unlinked"]].update(at=["record", 1, 5])),
        mutate("a refused floor statement's reason", lambda d: vs(d)[vn["floor_late_release"]]["floor_refused"][0].update(reason="floor_not_permitted")),
        mutate("a refused floor statement unstated", lambda d: vs(d)[vn["floor_given_by_viewer"]].pop("floor_refused")),
        mutate("a floor statement stated refused that the rules accept", lambda d: vs(d)[fs].update(floor_refused=vs(d)[vn["floor_give_by_non_holder"]]["floor_refused"])),
        mutate("a refused give stated as passing the floor", lambda d: vs(d)[vn["floor_give_by_non_holder"]]["floor"].insert(2, {"host": host, "t": vs(d)[vn["floor_give_by_non_holder"]]["floor_refused"][0]["t"], "holder": doc["principals"]["alice"]})),
        mutate("a sealed input's count of sealed actions", lambda d: vs(d)[ss]["inputs"][0].update(sealed=4)),
        mutate("a text in the clear to a sealing key stated applied", lambda d: vs(d)[ss]["inputs"][2].update(fate={"outcome": "applied", "group": 9, "object": 0})),
        mutate("a refusal by rule 9 stated as rule 2's", lambda d: vs(d)[ss]["inputs"][3]["fate"].update(reason="input_not_permitted")),
        mutate("a host's count of refusals by rule 9", lambda d: vs(d)[ss]["totals"][0]["refused"].update(input_bad_seal=4)),
        mutate("a host's count of refusals by rule 8 left out", lambda d: vs(d)[ss]["totals"][0]["refused"].pop("input_unsealed")),
        mutate("an opened seal stated not opened", lambda d: vs(d)[ss]["opened"]["inputs"][0].update(seal="seal_not_opened")),
        mutate("a seal to the wrong key stated a commitment that does not match", lambda d: vs(d)[ss]["opened"]["inputs"][3].update(seal="seal_commitment")),
        mutate("a byte of an opened statement", lambda d: vs(d)[ss]["opened"]["inputs"][0].update(opened=flip(vs(d)[ss]["opened"]["inputs"][0]["opened"], -1))),
        mutate("a seal the opened level refuses stated accepted there", lambda d: vs(d)[vn["sealed_applies_wrong_key"]].update(opened={k: v for k, v in vs(d)[vn["sealed_applies_wrong_key"]].items() if k in STATED})),
        mutate("the opened level stated of a vector with no seal", lambda d: vs(d)[fs].update(opened={"outcome": "refuse", "reason": "outcome_mismatch", "at": ["record", 1, 3]})),
        mutate("another opening key", lambda d: d.update(opening_key="11" * 32)),
    ]


CHECKS = ("attrs", "screen_fields", "dt_order", "who_screen", "track", "screen_link", "who_input", "floor_host", "floor_request",
          "floor_give", "floor_eligible", "floor_release", "floor_take", "floor_loss", "who_latency", "latency_key", "latency_sig",
          "who_outcome", "duplicate", "consistent", "recompute", "rule_screen", "rule_power", "rule_floor", "rule_paused",
          "rule_accepts", "rule_bounds", "rule_stale", "latency_input", "latency_frame", "latency_ahead", "latency_sum",
          "sealed_of", "seal_length", "rule_unsealed", "seal_no_key", "seal_open", "seal_read", "seal_commit")


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 []
    st = HPKE.crypto_self_test()
    print(("PASS" if not st else "FAIL") + " X25519 (RFC 7748 §5.2) and HPKE base mode (RFC 9180 A.1.1) " + ",".join(st))
    fails += ["HPKE self-test"] if st else []
    ok = len(REASONS) == 13 and sorted(HOST_REASONS) == list(range(1, 10)) and len(FLOOR_REASONS) == 3 and len(SEAL_REASONS) == 4
    print(("PASS" if ok else "FAIL") + " §12 read from the extension: %d reasons, %d host refusals, %d floor refusals, %d seal reasons" % (len(REASONS), len(HOST_REASONS), len(FLOOR_REASONS), len(SEAL_REASONS)))
    fails += [] if ok else ["§12 does not read"]
    names = sorted(n for n in os.listdir(corpus_dir) if n != "README.md")
    ok = names == ["vectors.json"]
    print(("PASS" if ok else "FAIL") + " the corpus holds vectors.json and README.md alone")
    fails += [] if ok else ["other files: %s" % names]
    with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as f:
        doc = json.load(f)
    f2, seen = check(doc)
    fails += f2
    missing = [r for r in REASONS + [HOST_REASONS[n] for n in sorted(HOST_REASONS)] + FLOOR_REASONS + SEAL_REASONS if r not in seen]
    print(("PASS" if not missing else "FAIL") + " every reason of the four tables of §12 appears in the corpus")
    fails += ["the corpus never gives %s" % ", ".join(missing)] if missing else []
    for label, d in mutations(doc):
        caught, _ = check(d, quiet=True, first=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:
        DROP.add(name)
        try:
            caught, _ = check(doc, quiet=True, first=True)
        finally:
            DROP.discard(name)
        print(("PASS" if caught else "FAIL") + " the corpus catches the check dropped: %s%s" % (name, " (%s)" % caught[0][:90] if caught else ""))
        if not caught:
            fails.append("the corpus does not catch the check dropped: %s" % name)
    print("remote_desktop_verify: %s" % ("PASS" if not fails else "FAIL (%d)" % len(fails)))
    for f in fails:
        print("  " + f)
    return 0 if not fails else 1


def room_main(path, keys):
    with open(path, encoding="utf-8") as f:
        v = json.load(f)
    print(json.dumps(run_vector(v, [bytes.fromhex(k) for k in keys] if keys else None)))
    return 0


if __name__ == "__main__":
    if len(sys.argv) >= 3 and sys.argv[1] == "--room" and len(sys.argv) % 2 == 1 and all(a == "--opening-key" for a in sys.argv[3::2]):
        sys.exit(room_main(sys.argv[2], sys.argv[4::2]))
    if len(sys.argv) != 2:
        print("usage: remote_desktop_verify.py <live-conformance/remote> | --room <room.json> [--opening-key <hex>]...", file=sys.stderr)
        sys.exit(2)
    sys.exit(main(sys.argv[1]))