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

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

Written from the extension's text. It shares no code with wai-rs, which generates the
corpus, and never runs it. It checks Ed25519 signatures with
../claim-conformance/_ed25519.py, WAI's strict rule (SPEC §7 *Signatures*) written from
RFC 8032 with the standard library, and computes BLAKE3 (for a key log's commitments)
with ../claim-conformance/_blake3.py. Every event is verified by the live-events corpus's
own second implementation, live_events_verify.py (live-events §6), which is likewise
written from its text alone. Everything else is here:

- the room descriptor (§3): its reader, its field rules, its digest and signature, and the
  registered publish profiles (§8), each written from §8.2's table and compared with the
  record the corpus states;
- the room grant (§4): its reader and the eight checks of §4.3 in their order, its
  attributes (§4.4), and its window;
- revocation (§5): the key-log head held for the descriptor's origin (jwp-receipts §7.2
  rules 1 to 3: its members, its canonical form and its root's signature), the
  authority's entry, and the revoked grants and viewers;
- track naming (§6) and the lifecycle detail (§7.2), with slots, takeover and failover;
- room verification (§9): the merged order, each record's steps, the room state, the
  outcome, and the `moqt` scopes of each grant (§10.2), with the catalog and receipts
  tracks of §6.1;
- proof of possession (§10.5): the proof record, its base64url header form, and the seven
  checks of §10.5.4 in their order;
- seeking back (§10.6): a request that starts in a group below the track's live edge is
  served only under a grant whose powers have bit 4, seek.

For every vector it requires the outcome, reason and place, and for one that does not
refuse the room id, the gaps or the grant digests, merged order, state, tally and scopes
the corpus states, with each grant's seek power; for every proof and every request of
`seek` its outcome and reason; and of the corpus, that every reason of §11 refuses a
vector, a proof or a request.

    python3 live_rooms_verify.py <live-conformance/rooms>

Exit status 0 only if every vector verifies. Python 3.8 or later; standard library only.
"""
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
from _blake3 import blake3  # noqa: E402
import _ed25519 as ED  # noqa: E402

Refuse = LEV.Refuse

DESCRIPTOR_MAGIC = b"WRD1"
GRANT_MAGIC = b"WRG1"
DESCRIPTOR_DOMAIN = b"wai:live-room\x01"
GRANT_DOMAIN = b"wai:live-room-grant\x01"
VIEWER_DOMAIN = b"wai:live-viewer\x01"
AUTHORITY = "live/room"
MAX_GRANT_S = 14 * 86400
NO_CEILING = 2 ** 64 - 1
MEDIA = {1: "audio", 2: "video", 3: "screen"}
MODERATE, DIRECT, COMPOSE, RECORD, SEEK = 1, 2, 4, 8, 16
POWERS = MODERATE | DIRECT | COMPOSE | RECORD | SEEK  # §3.4, §4.3 step 5: none above bit 4
CONTROL, DECISION = 0x09, 0x0B
JOIN, LEAVE, PUBLISH, UNPUBLISH, EXCHANGE, TAKEOVER, FAILOVER = range(0x10, 0x17)

# §11, in its order.
REASONS = [
    "descriptor_invalid", "key_log_missing", "key_log_not_held", "authority_not_listed",
    "authority_revoked", "grant_invalid", "grant_too_long", "caps_exceed_role",
    "grant_exceeds_room", "exchange_unlinked", "exchange_mismatch", "lifecycle_malformed",
    "grant_unknown", "grant_wrong_principal", "grant_revoked", "grant_reused",
    "already_present", "not_present", "participant_limit", "outside_grant_window",
    "power_not_granted", "ai_unmarked", "publish_not_permitted", "profile_not_granted", "already_published",
    "not_published", "slot_unknown", "slot_full", "publisher_limit", "exchange_not_current",
    "exchange_off_tick", "takeover_not_higher", "failover_not_standby", "failover_early",
    "pop_missing", "pop_malformed", "pop_wrong_grant", "pop_wrong_key", "pop_bad_signature",
    "pop_replayed", "pop_wrong_challenge", "pop_wrong_context", "seek_not_granted",
]


class Reader:
    """Reads fields in order; running out is `truncated` (§3.4)."""

    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 name(self):
        return self.take(self.u8())

    def string(self):
        return self.take(self.u16())


def read_head(b, magic):
    """§3.4 step 1."""
    if len(b) < 4:
        raise Refuse("truncated")
    if b[:4] != magic:
        raise Refuse("malformed")
    if len(b) < 5:
        raise Refuse("truncated")
    if b[4] != 1:
        raise Refuse("unknown_version")


NAME = re.compile(rb"[a-z0-9_-]{1,32}")
SLOT_NAME = re.compile(rb"[a-z0-9_]{1,32}")
MODE = re.compile(rb"[A-Z0-9_]{1,16}")


def is_name(b):
    return NAME.fullmatch(b) is not None


def is_slot_name(b):
    return SLOT_NAME.fullmatch(b) is not None


def is_token(b, most):
    return 1 <= len(b) <= most and all(0x21 <= c <= 0x7E for c in b)


def bitset(kinds):
    out = bytearray(32)
    for k in kinds:
        out[k // 8] |= 1 << (k % 8)
    return bytes(out)


def has(bm, k):
    return bm[k // 8] >> (k % 8) & 1 == 1


def within(a, b):
    """Every bit of bitmap or byte `a` is set in `b`."""
    if isinstance(a, int):
        return a & ~b == 0
    return all(x & ~y == 0 for x, y in zip(a, b))


def lifecycle_permitted(policy):
    return any(LEV.permits(policy, k) for k in range(0x10, 0x20))


def powers_cover(kinds, powers):
    """§3.3: decision needs moderate; control needs direct or record."""
    return (not has(kinds, DECISION) or powers & MODERATE) and (not has(kinds, CONTROL) or powers & (DIRECT | RECORD))


# ---------------------------------------------------------------- §8 profiles

def profile_bytes(name, media, codecs, layering, mode, layers, sample_rate, channels, processing):
    """§8.1's record."""
    o = bytes([len(name)]) + name.encode() + bytes([media, len(codecs)])
    for c in codecs:
        o += bytes([len(c)]) + c.encode()
    o += bytes([layering, len(mode)]) + mode.encode() + bytes([len(layers)])
    for w, h, fps, lo, hi in layers:
        o += struct.pack("<HHHII", w, h, fps, lo, hi)
    return o + struct.pack("<IBB", sample_rate, channels, processing)


# §8.2's table: processing bit 0 echo cancellation, 1 noise suppression, 2 gain control, 3 DTX.
REGISTERED = {
    "speech": profile_bytes("speech", 1, ["opus"], 0, "", [(0, 0, 0, 16, 64)], 48000, 1, 0b1111),
    "music": profile_bytes("music", 1, ["opus"], 0, "", [(0, 0, 0, 128, 510)], 48000, 2, 0),
    "screen": profile_bytes("screen", 3, ["av01.0.08M.08", "avc1.640028"], 2, "L1T3", [(1920, 1080, 15, 300, 2500)], 0, 0, 0),
    "camera-720": profile_bytes("camera-720", 2, ["av01.0.05M.08", "avc1.64001F"], 1, "",
                                [(640, 360, 30, 150, 500), (1280, 720, 30, 500, 1500)], 0, 0, 0),
    "camera-1080": profile_bytes("camera-1080", 2, ["av01.0.08M.08", "avc1.640028"], 1, "",
                                 [(640, 360, 30, 150, 500), (1280, 720, 30, 500, 1500), (1920, 1080, 30, 1500, 4000)], 0, 0, 0),
}


def read_profile(r):
    start = r.p
    p = {"name": r.name(), "media": r.u8()}
    p["codecs"] = [r.name() for _ in range(r.u8())]
    p["layering"], p["mode"] = r.u8(), r.name()
    p["layers"] = [struct.unpack("<HHHII", r.take(14)) for _ in range(r.u8())]
    p["sample_rate"], p["channels"], p["processing"] = r.u32(), r.u8(), r.u8()
    p["bytes"] = r.b[start:r.p]
    return p


def profile_ok(p):
    """§8.1."""
    if not is_name(p["name"]) or p["media"] not in MEDIA or not 1 <= len(p["codecs"]) <= 4:
        return False
    if not all(is_token(c, 64) for c in p["codecs"]) or p["layering"] > 3:
        return False
    if (p["layering"] == 2) != (len(p["mode"]) > 0) or (p["mode"] and not MODE.fullmatch(p["mode"])):
        return False
    n = len(p["layers"])
    if not 1 <= n <= 8 or (p["layering"] == 0 and n != 1):
        return False
    audio = p["media"] == 1
    for w, h, fps, lo, hi in p["layers"]:
        if hi == 0 or lo > hi:
            return False
        if audio and (w, h, fps) != (0, 0, 0):
            return False
        if not audio and not (w and h and fps):
            return False
    if audio:
        return p["sample_rate"] > 0 and 1 <= p["channels"] <= 8 and p["processing"] & ~0x0F == 0
    return p["sample_rate"] == 0 and p["channels"] == 0 and p["processing"] == 0


# ---------------------------------------------------------------- §3 the descriptor

def read_descriptor(b):
    """§3.4, then §3.2: the descriptor, or Refuse."""
    read_head(b, DESCRIPTOR_MAGIC)
    r = Reader(b, 5)
    d = {"authority": r.take(32), "clock": r.take(32), "t0": r.u64(), "tick": r.u64(), "origin": r.string(),
         "max_present": r.u32(), "max_publishers": r.u16(), "reval": r.u32(), "ceiling": r.take(65)}
    d["roles"] = []
    for _ in range(r.u8()):
        d["roles"].append({"name": r.name(), "publish": r.u8(), "subscribe": r.u8(), "max_layer": r.u8(),
                           "powers": r.u8(), "events": r.take(32)})
    d["slots"] = []
    for _ in range(r.u8()):
        d["slots"].append({"name": r.name(), "media": r.u8(), "redundancy": r.u8(), "failover_ms": r.u32()})
    d["profiles"] = [read_profile(r) for _ in range(r.u8())]
    d["review"] = (r.take(32), r.u32(), r.u8())
    d["mode"], d["retention"], d["commit_interval_ms"], d["jitter_ms"] = r.u8(), r.u16(), r.u32(), r.u32()
    d["regions"] = [r.take(2) for _ in range(r.u8())]
    body = b[:r.p]
    d["sig"] = r.take(64)
    if r.p != len(b):
        raise Refuse("malformed")
    if not descriptor_ok(d):
        raise Refuse("descriptor_invalid")
    bad = ED.verify(d["authority"], DESCRIPTOR_DOMAIN + body, d["sig"])
    if bad:
        raise Refuse(bad)
    d["room"] = hashlib.sha256(body).digest()
    try:
        d["ceiling_policy"] = LEV.read_policy(d["ceiling"])
    except Refuse:
        raise Refuse("descriptor_invalid")
    return d


def descriptor_ok(d):
    """§3.4's field rules."""
    if d["tick"] == 0 or not (d["origin"] == b"" or is_token(d["origin"], 255)):
        return False
    if d["max_present"] == 0 or d["max_publishers"] == 0:
        return False
    try:
        ceiling = LEV.read_policy(d["ceiling"])
    except Refuse:
        return False
    if lifecycle_permitted(ceiling) or not 1 <= len(d["roles"]) <= 16:
        return False
    names = [x["name"] for x in d["roles"]]
    if len(set(names)) != len(names):
        return False
    for x in d["roles"]:
        if not is_name(x["name"]) or x["subscribe"] > 1 or x["publish"] & ~0b111 or x["powers"] & ~POWERS:
            return False
        if x["subscribe"] == 0 and x["max_layer"] != 0:
            return False
        if not within(x["events"], ceiling["kinds"]) or not powers_cover(x["events"], x["powers"]):
            return False
    if len(d["slots"]) > 16 or len({s["name"] for s in d["slots"]}) != len(d["slots"]):
        return False
    for s in d["slots"]:
        if not is_slot_name(s["name"]) or s["media"] not in MEDIA:
            return False
        if (s["redundancy"], s["failover_ms"] > 0) not in ((1, False), (2, True)):
            return False
    if len(d["profiles"]) > 16 or len({p["name"] for p in d["profiles"]}) != len(d["profiles"]):
        return False
    for p in d["profiles"]:
        if not profile_ok(p):
            return False
        reg = REGISTERED.get(p["name"].decode())
        if reg is not None and reg != p["bytes"]:
            return False
    kinds, deadline, fallback = d["review"]
    if any(kinds):
        if not within(kinds, ceiling["kinds"]) or deadline == 0 or fallback not in (1, 2):
            return False
    elif deadline != 0 or fallback != 0:
        return False
    if d["mode"] > 2 or len(d["regions"]) > 16:
        return False
    if d["mode"] == 0:
        if d["commit_interval_ms"] != 0 or d["jitter_ms"] != 0:
            return False
    elif d["commit_interval_ms"] == 0 or d["jitter_ms"] > d["commit_interval_ms"]:
        return False
    if not all(re.fullmatch(rb"[A-Z]{2}", x) for x in d["regions"]):
        return False
    return all(a < b for a, b in zip(d["regions"], d["regions"][1:]))


def is_room_key(d, k):
    return k == d["authority"] or k == d["clock"]


# ---------------------------------------------------------------- §4 grants

def read_attrs(b):
    """§4.4: the entries, or None when they break a rule."""
    if len(b) > 1024:
        return None
    r = Reader(b)
    out, last = [], None
    try:
        while r.p < len(b):
            k, v = r.name(), r.string()
            if not is_slot_name(k) or (last is not None and last >= k):
                return None
            out.append((k.decode(), v.decode("utf-8")))
            last = k
    except (Refuse, UnicodeDecodeError):
        return None
    return out


def read_grant(b):
    """§4.3 step 1: the grant's fields as read."""
    read_head(b, GRANT_MAGIC)
    r = Reader(b, 5)
    g = {"room": r.take(32), "principal": r.take(32), "prev": r.take(32), "effective": r.u64(), "nbf": r.u64(),
         "exp": r.u64(), "role": r.u8(), "publish": r.u8(), "subscribe": r.u8(), "max_layer": r.u8(),
         "powers": r.u8(), "precedence": r.u8(), "once": r.take(16), "viewer": r.take(16), "policy_bytes": r.take(65),
         "joules": r.u64(), "funds": r.u64(), "currency": r.take(3)}
    g["profiles"] = list(r.take(r.u8()))
    g["origins"] = [r.name() for _ in range(r.u8())]
    g["attrs"] = r.string()
    g["body"] = b[:r.p]
    g["sig"] = r.take(64)
    if r.p != len(b):
        raise Refuse("malformed")
    g["digest"] = hashlib.sha256(g["body"]).digest()
    return g


def grant_fields_ok(g, d, held):
    """§4.3 step 5."""
    if g["exp"] <= g["nbf"]:
        return False
    if g["prev"] == bytes(32):
        if g["effective"] != 0:
            return False
    elif g["effective"] % d["tick"] != 0:
        return False
    if g["role"] >= len(d["roles"]) or g["subscribe"] > 1:
        return False
    if g["publish"] & ~0b111 or g["powers"] & ~POWERS or (g["subscribe"] == 0 and g["max_layer"] != 0):
        return False
    try:
        policy = LEV.read_policy(g["policy_bytes"])
    except Refuse:
        return False
    if lifecycle_permitted(policy) or not powers_cover(policy["kinds"], g["powers"]):
        return False
    funded = g["funds"] != NO_CEILING
    if funded != (g["currency"] != bytes(3)) or (funded and not re.fullmatch(rb"[A-Z]{3}", g["currency"])):
        return False
    media = set()
    for i in g["profiles"]:
        if i >= len(d["profiles"]):
            return False
        m = d["profiles"][i]["media"]
        if m in media or not publishes(g, m):
            return False
        media.add(m)
    if len(g["origins"]) > 8 or not all(is_token(o, 255) for o in g["origins"]):
        return False
    attrs = read_attrs(g["attrs"])
    if attrs is None:
        return False
    if any(k == "ai" and not re.fullmatch(r"[0-9a-f]{64}", v) for k, v in attrs):
        return False
    return all(h["digest"] != g["digest"] for h in held)


def ai_model(g):
    """§4.4: the model an AI participant's grant names, as bytes, or None."""
    for k, v in read_attrs(g["attrs"]) or []:
        if k == "ai":
            return bytes.fromhex(v)
    return None


def publishes(g, m):
    return m in MEDIA and g["publish"] >> (m - 1) & 1 == 1


def check_grants(d, grants):
    """§4.3 over every grant, in order: the grants, or (index, reason)."""
    held = []
    for i, b in enumerate(grants):
        try:
            g = read_grant(b)
        except Refuse as e:
            return None, (i, e.reason)
        bad = ED.verify(d["authority"], GRANT_DOMAIN + g["body"], g["sig"])
        if bad:
            return None, (i, bad)
        if g["prev"] != bytes(32):
            old = next((h for h in held if h["digest"] == g["prev"]), None)
            if old is None:
                return None, (i, "exchange_unlinked")
            if old["room"] != g["room"] or old["principal"] != g["principal"]:
                return None, (i, "exchange_mismatch")
        if g["room"] != d["room"]:
            return None, (i, "wrong_room")
        if not grant_fields_ok(g, d, held):
            return None, (i, "grant_invalid")
        g["policy"] = LEV.read_policy(g["policy_bytes"])
        if g["exp"] - g["nbf"] > MAX_GRANT_S:
            return None, (i, "grant_too_long")
        role = d["roles"][g["role"]]
        if (not within(g["publish"], role["publish"]) or g["subscribe"] > role["subscribe"]
                or g["max_layer"] > role["max_layer"] or not within(g["powers"], role["powers"])
                or not within(g["policy"]["kinds"], role["events"])):
            return None, (i, "caps_exceed_role")
        p, c = g["policy"], d["ceiling_policy"]
        if (p["max_payload"] > c["max_payload"] or p["max_text_cp"] > c["max_text_cp"]
                or p["max_events"] > c["max_events"] or p["window_us"] < c["window_us"]
                or p["max_reactions"] > c["max_reactions"] or p["reaction_window_us"] < c["reaction_window_us"]):
            return None, (i, "grant_exceeds_room")
        held.append(g)
    return held, None


def good_at(g, d, t):
    """§4.3: nbf ≤ ⌊(t0 + t) / 10^6⌋ < exp."""
    return g["nbf"] <= (d["t0"] + t) // 1000000 < g["exp"]


def viewer_tag(g):
    return None if g["viewer"] == bytes(16) else hashlib.sha256(VIEWER_DOMAIN + g["viewer"]).digest()


# ---------------------------------------------------------------- §5 the key log

REVISION_MEMBERS = {"kind", "origin", "seq", "prev", "issued_at", "root", "next_root", "keys", "revoked_modules",
                    "grant_max_s", "status_max_age_s", "sig"}
ENTRY_MEMBERS = {"id", "key", "may_sign", "status", "until", "compromised_since"}
HEX32 = re.compile(r"[0-9a-f]{64}")
LOG_TIME = re.compile(r"[0-9]{4}-(0[1-9]|1[0-2])-(0[1-9]|[12][0-9]|3[01])T([01][0-9]|2[0-3]):[0-5][0-9]:[0-5][0-9]Z")
KEY_ID = re.compile(r"[A-Za-z0-9._-]{1,64}")
AUTH_SEG = re.compile(r"[A-Za-z0-9._-]+")


def no_duplicates(pairs):
    keys = [k for k, _ in pairs]
    if len(set(keys)) != len(keys):
        raise ValueError("a member named twice")
    return dict(pairs)


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


def commitment(k):
    return blake3(b"wai:key-set-next\x01" + k)


def jcs(v):
    """RFC 8785 over what a key-log revision holds: strings of ASCII, integers, null, arrays."""
    return json.dumps(v, sort_keys=True, separators=(",", ":"), ensure_ascii=False)


def entry_ok(e):
    if not isinstance(e, dict) or set(e) != ENTRY_MEMBERS:
        return False
    if not isinstance(e["id"], str) or not KEY_ID.fullmatch(e["id"]):
        return False
    if not isinstance(e["key"], str) or not HEX32.fullmatch(e["key"]) or not ED.strict_key(bytes.fromhex(e["key"])):
        return False
    ms = e["may_sign"]
    if not isinstance(ms, list) or not ms or len(set(map(str, ms))) != len(ms):
        return False
    for a in ms:
        if not isinstance(a, str) or not 1 <= len(a) <= 128 or not all(AUTH_SEG.fullmatch(s) for s in a.split("/")):
            return False
    for t in ("until", "compromised_since"):
        if e[t] is not None and not (isinstance(e[t], str) and LOG_TIME.fullmatch(e[t])):
            return False
    shape = {"active": e["until"] is None and e["compromised_since"] is None,
             "retired": e["until"] is not None and e["compromised_since"] is None,
             "revoked": e["compromised_since"] is not None}
    return shape.get(e["status"], False)


def revision_ok(v):
    """jwp-receipts §7.2 rule 1, as far as this verifier reads it."""
    if not isinstance(v, dict) or set(v) != REVISION_MEMBERS or v["kind"] != "wai-key-set":
        return False
    if not isinstance(v["origin"], str) or not is_int(v["seq"]):
        return False
    if (v["seq"] == 0) != (v["prev"] is None) or (v["prev"] is not None and not (isinstance(v["prev"], str) and HEX32.fullmatch(v["prev"]))):
        return False
    if not isinstance(v["issued_at"], str) or not LOG_TIME.fullmatch(v["issued_at"]):
        return False
    for k in ("root", "next_root"):
        if not isinstance(v[k], str) or not HEX32.fullmatch(v[k]):
            return False
    root, nxt = bytes.fromhex(v["root"]), bytes.fromhex(v["next_root"])
    if nxt == commitment(root):
        return False
    if not isinstance(v["keys"], list) or not all(entry_ok(e) for e in v["keys"]):
        return False
    ids, keys = [e["id"] for e in v["keys"]], [e["key"] for e in v["keys"]]
    if len(set(ids)) != len(ids) or len(set(keys)) != len(keys):
        return False
    if any(bytes.fromhex(k) == root or commitment(bytes.fromhex(k)) == nxt for k in keys):
        return False
    rm = v["revoked_modules"]
    if not isinstance(rm, list) or not all(isinstance(x, str) and HEX32.fullmatch(x) for x in rm) or len(set(rm)) != len(rm):
        return False
    if not (is_int(v["grant_max_s"]) and 1 <= v["grant_max_s"] <= 86400):
        return False
    if not (is_int(v["status_max_age_s"]) and 1 <= v["status_max_age_s"] <= 3600):
        return False
    return isinstance(v["sig"], str) and re.fullmatch(r"[0-9a-f]{128}", v["sig"]) is not None


def check_key_log(d, head):
    """§5: the revoked digests, or Refuse."""
    if d["origin"] == b"":
        return set()
    if head is None:
        raise Refuse("key_log_missing")
    try:
        v = json.loads(head, object_pairs_hook=no_duplicates)
    except ValueError:
        raise Refuse("key_log_not_held")
    if not revision_ok(v) or v["origin"] != d["origin"].decode():
        raise Refuse("key_log_not_held")
    unsigned = {k: x for k, x in v.items() if k != "sig"}
    if ED.verify(bytes.fromhex(v["root"]), b"wai:key-set\x01" + jcs(unsigned).encode(), bytes.fromhex(v["sig"])):
        raise Refuse("key_log_not_held")
    entry = next((e for e in v["keys"] if e["key"] == d["authority"].hex() and AUTHORITY in e["may_sign"]), None)
    if entry is None:
        raise Refuse("authority_not_listed")
    if entry["status"] == "revoked":
        raise Refuse("authority_revoked")
    return {bytes.fromhex(x) for x in v["revoked_modules"]}


def revoked(g, revoked_set):
    return g["digest"] in revoked_set or viewer_tag(g) in revoked_set


# ---------------------------------------------------------------- §7 lifecycle detail

def read_detail(kind, b):
    """§7.2: the detail as a dict, or Refuse("lifecycle_malformed")."""
    r = Reader(b)
    try:
        def slot(required):
            s = r.name()
            if (not s and required) or (s and not is_slot_name(s)):
                raise Refuse("lifecycle_malformed")
            return s.decode()

        def media():
            m = r.u8()
            if m not in MEDIA:
                raise Refuse("lifecycle_malformed")
            return m

        def layers():
            n = r.u8()
            if n == 0:
                raise Refuse("lifecycle_malformed")
            return n

        if kind == JOIN:
            x = {"grant": r.take(32)}
        elif kind == LEAVE:
            x = {"reason": r.u8()}
            if x["reason"] not in (1, 2, 3, 4):
                raise Refuse("lifecycle_malformed")
        elif kind == PUBLISH:
            x = {"slot": slot(False), "media": media(), "profile": r.u8()}
            x["layers"] = layers()
        elif kind == UNPUBLISH:
            x = {"slot": slot(False), "media": media()}
        elif kind == EXCHANGE:
            x = {"old": r.take(32), "new": r.take(32)}
        elif kind == TAKEOVER:
            x = {"slot": slot(True), "media": media(), "profile": r.u8()}
            x["layers"] = layers()
            x["previous"] = r.take(32)
        elif kind == FAILOVER:
            x = {"slot": slot(True), "media": media(), "failed": r.take(32), "last_seen": r.u64()}
        else:
            raise Refuse("lifecycle_malformed")
    except Refuse:
        raise Refuse("lifecycle_malformed")
    if r.p != len(b):
        raise Refuse("lifecycle_malformed")
    return x


class Room:
    """The room state of §9.3 and the checks of §7.2 that move it."""

    def __init__(self, d, grants, revoked_set):
        self.d, self.grants, self.revoked = d, grants, revoked_set
        self.present = {}      # principal -> grant digest
        self.own = {}          # (principal, media) -> (profile, layers)
        self.slots = [{"live": None, "standby": None} for _ in d["slots"]]  # each (principal, profile, layers)
        self.timeline = []
        self.used_once = set()

    def grant(self, digest):
        return next((g for g in self.grants if g["digest"] == digest), None)

    def current(self, p):
        return self.grant(self.present[p]) if p in self.present else None

    def slot_index(self, name, m):
        return next((i for i, s in enumerate(self.d["slots"]) if s["name"].decode() == name and s["media"] == m), None)

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

    def may_publish(self, g, m, profile, layers):
        if not publishes(g, m):
            raise Refuse("publish_not_permitted")
        ps = self.d["profiles"]
        if not (profile < len(ps) and profile in g["profiles"] and ps[profile]["media"] == m and layers <= len(ps[profile]["layers"])):
            raise Refuse("profile_not_granted")

    def limit(self):
        if len(self.publishers()) > self.d["max_publishers"]:
            raise Refuse("publisher_limit")

    def drop(self, p):
        for s in self.slots:
            if s["standby"] and s["standby"][0] == p:
                s["standby"] = None
            if s["live"] and s["live"][0] == p:
                s["live"], s["standby"] = s["standby"], None

    def publishing(self, p):
        out = [(m, v[0]) for (q, m), v in sorted(self.own.items()) if q == p]
        for i, s in enumerate(self.slots):
            for h in (s["live"], s["standby"]):
                if h and h[0] == p:
                    out.append((self.d["slots"][i]["media"], h[1]))
        return out

    def need_present(self, p):
        g = self.current(p)
        if g is None:
            raise Refuse("not_present")
        return g

    def slot_of(self, name, m):
        i = self.slot_index(name, m)
        if i is None:
            raise Refuse("slot_unknown")
        return i

    def apply(self, kind, subject, x, t):
        if kind == JOIN:
            g = self.grant(x["grant"])
            if g is None:
                raise Refuse("grant_unknown")
            if g["principal"] != subject:
                raise Refuse("grant_wrong_principal")
            if subject in self.present:
                raise Refuse("already_present")
            if revoked(g, self.revoked):
                raise Refuse("grant_revoked")
            if g["once"] != bytes(16) and g["once"] in self.used_once:
                raise Refuse("grant_reused")
            if not good_at(g, self.d, t):
                raise Refuse("outside_grant_window")
            if len(self.present) + 1 > self.d["max_present"]:
                raise Refuse("participant_limit")
            if g["once"] != bytes(16):
                self.used_once.add(g["once"])
            self.present[subject] = g["digest"]
            self.timeline.append((subject, g["digest"], t))
        elif kind == LEAVE:
            self.need_present(subject)
            del self.present[subject]
            self.own = {k: v for k, v in self.own.items() if k[0] != subject}
            self.drop(subject)
        elif kind == PUBLISH:
            g = self.need_present(subject)
            i = None if x["slot"] == "" else self.slot_of(x["slot"], x["media"])
            self.may_publish(g, x["media"], x["profile"], x["layers"])
            held = (subject, x["profile"], x["layers"])
            if i is None:
                if (subject, x["media"]) in self.own:
                    raise Refuse("already_published")
                self.own[(subject, x["media"])] = (x["profile"], x["layers"])
            else:
                s = self.slots[i]
                if any(h and h[0] == subject for h in (s["live"], s["standby"])):
                    raise Refuse("already_published")
                if s["live"] and (self.d["slots"][i]["redundancy"] < 2 or s["standby"]):
                    raise Refuse("slot_full")
                if s["live"] is None:
                    s["live"] = held
                else:
                    s["standby"] = held
            self.limit()
        elif kind == UNPUBLISH:
            self.need_present(subject)
            if x["slot"] == "":
                if (subject, x["media"]) not in self.own:
                    raise Refuse("not_published")
                del self.own[(subject, x["media"])]
            else:
                s = self.slots[self.slot_of(x["slot"], x["media"])]
                if not any(h and h[0] == subject for h in (s["live"], s["standby"])):
                    raise Refuse("not_published")
                if s["standby"] and s["standby"][0] == subject:
                    s["standby"] = None
                else:
                    s["live"], s["standby"] = s["standby"], None
        elif kind == EXCHANGE:
            cur = self.need_present(subject)
            if cur["digest"] != x["old"]:
                raise Refuse("exchange_not_current")
            g = self.grant(x["new"])
            if g is None:
                raise Refuse("grant_unknown")
            if g["prev"] != x["old"]:
                raise Refuse("exchange_unlinked")
            if t != g["effective"]:
                raise Refuse("exchange_off_tick")
            if revoked(g, self.revoked):
                raise Refuse("grant_revoked")
            if not good_at(g, self.d, t):
                raise Refuse("outside_grant_window")
            for m, profile in self.publishing(subject):
                if not publishes(g, m):
                    raise Refuse("publish_not_permitted")
                if profile not in g["profiles"]:
                    raise Refuse("profile_not_granted")
            self.present[subject] = g["digest"]
            self.timeline.append((subject, g["digest"], t))
        elif kind == TAKEOVER:
            g = self.need_present(subject)
            i = self.slot_of(x["slot"], x["media"])
            s = self.slots[i]
            if not s["live"] or s["live"][0] != x["previous"]:
                raise Refuse("not_published")
            self.may_publish(g, x["media"], x["profile"], x["layers"])
            theirs = self.current(x["previous"])
            if subject == x["previous"] or g["precedence"] <= (theirs["precedence"] if theirs else 0):
                raise Refuse("takeover_not_higher")
            if s["standby"] and s["standby"][0] == subject:
                s["standby"] = None
            s["live"] = (subject, x["profile"], x["layers"])
            self.limit()
        elif kind == FAILOVER:
            self.need_present(subject)
            i = self.slot_of(x["slot"], x["media"])
            s = self.slots[i]
            if not s["live"] or s["live"][0] != x["failed"]:
                raise Refuse("not_published")
            if not s["standby"] or s["standby"][0] != subject:
                raise Refuse("failover_not_standby")
            if x["last_seen"] > t or t - x["last_seen"] < self.d["slots"][i]["failover_ms"] * 1000:
                raise Refuse("failover_early")
            s["live"], s["standby"] = s["standby"], None


# ---------------------------------------------------------------- §6 names, §10.2 scopes

def render(fields, name):
    """MOQT §8.8: a-z, A-Z, 0-9 and _ as themselves; every other byte as . and two hex digits."""
    def one(f):
        return "".join(chr(c) if (chr(c).isascii() and chr(c).isalnum()) or c == 0x5F else ".%02x" % c for c in f.encode())
    return "-".join(one(f) for f in fields) + "--" + one(name)


def scopes(d, g):
    """§10.2, in its order."""
    room, me = d["room"].hex(), g["principal"].hex()
    out = [[[0]]]
    for m in (1, 2, 3):
        if publishes(g, m):
            out.append([[2, 6], [room, me, MEDIA[m], None]])
            for s in d["slots"]:
                if s["media"] == m:
                    out.append([[6], [room, "_" + s["name"].decode(), MEDIA[m], None]])
                    out.append([[6], [room, "_" + s["name"].decode(), None], "catalog"])
    if any(publishes(g, m) for m in (1, 2, 3)):
        out.append([[2, 6], [room, me, None], "catalog"])
    if any(g["policy"]["kinds"]):
        out.append([[2, 6], [room, me, "events", None], "0"])
    if g["subscribe"]:
        out.append([[3], [room]])
        out.append([[4, 7, 8], [room], "catalog"])
        for layer in range(g["max_layer"] + 1):
            out.append([[4, 7, 8], [room], str(layer)])
            out.append([[4, 7, 8], [room], str(layer) + "_receipts"])
    return out


# ---------------------------------------------------------------- §10.5 proof of possession

PROOF_MAGIC = b"WRP1"
PROOF_DOMAIN = b"wai:live-room-proof\x01"
B64 = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_"


def from_base64url(t):
    """RFC 4648 §5 without padding, strictly: the alphabet, a length an encoding has, and
    zero bits past the last byte; None otherwise."""
    if any(c not in B64 for c in t) or len(t) % 4 == 1:
        return None
    out = bytearray()
    for i in range(0, len(t), 4):
        c = t[i:i + 4]
        n = 0
        for j, ch in enumerate(c):
            n |= B64.index(ch) << (18 - 6 * j)
        k = len(c) - 1
        out += n.to_bytes(3, "big")[:k]
        if k < 3 and n & ((1 << (8 * (3 - k))) - 1):
            return None
    return bytes(out)


def read_proof(b):
    """§10.5.4 step 2."""
    r = Reader(b)
    try:
        if r.take(4) != PROOF_MAGIC or r.u8() != 1:
            raise Refuse("pop_malformed")
        p = {"grant": r.take(32), "key": r.take(32), "challenge": r.take(32)}
        ctx = r.take(r.u16())
        p["body"] = b[:r.p]
        p["sig"] = r.take(64)
    except Refuse:
        raise Refuse("pop_malformed")
    if r.p != len(b) or any(c < 0x20 or c > 0x7E for c in ctx):
        raise Refuse("pop_malformed")
    p["context"] = ctx.decode("ascii")
    return p


def check_proof(grant, proof, challenge, consumed, context):
    """§10.5.4, in its order, for a grant record already checked by §4.3."""
    g = read_grant(grant)
    if proof is None:
        raise Refuse("pop_missing")
    p = read_proof(proof)
    if p["grant"] != g["digest"]:
        raise Refuse("pop_wrong_grant")
    if p["key"] != g["principal"]:
        raise Refuse("pop_wrong_key")
    if ED.verify(p["key"], PROOF_DOMAIN + p["body"], p["sig"]) is not None:
        raise Refuse("pop_bad_signature")
    if p["challenge"] != challenge:
        raise Refuse("pop_replayed" if p["challenge"] in consumed else "pop_wrong_challenge")
    if p["context"] != context:
        raise Refuse("pop_wrong_context")


def proof_outcome(v):
    """A proof vector's outcome: the grant and proof from the MOQT options or the HTTP
    headers (§10.5.2, §10.5.3), then §10.5.4."""
    if v["binding"] == "moqt":
        grant = bytes.fromhex(v["token"])
        if len(v["proofs"]) > 1:
            raise ValueError("one proof per vector")
        proof = bytes.fromhex(v["proofs"][0]) if v["proofs"] else None
    else:
        auth = v["headers"]["authorization"]
        if not auth.startswith("Bearer "):
            raise ValueError("a Bearer token")
        grant = from_base64url(auth[len("Bearer "):])
        raw = v["headers"].get("wai-room-proof")
        proof = None if raw is None else (from_base64url(raw) or b"")
    consumed = {bytes.fromhex(c) for c in v["consumed"]}
    try:
        check_proof(grant, proof, bytes.fromhex(v["challenge"]), consumed, v["verb"] + " " + v["target"])
    except Refuse as r:
        return {"outcome": "refuse", "reason": r.reason}
    return {"outcome": "accept"}


# ---------------------------------------------------------------- §10.6 seeking back

def seeks_back(live_edge, start):
    """§10.6: a request seeks back when its start Location (g, o) has g below the live
    edge L, the group of the track's Largest Object; one from (L, o) does not."""
    return start[0] < live_edge


def seek_outcome(doc, v):
    """A request of `seek`: the grant it names, by vector and index, from that room's
    grants as §4.3 checks them, then §10.6's rule."""
    room = next(x for x in doc["vectors"] if x["name"] == v["vector"])
    d = read_descriptor(bytes.fromhex(room["descriptor"]))
    held, bad = check_grants(d, [bytes.fromhex(g) for g in room["grants"]])
    if bad:
        raise ValueError("the room's grants do not check: %s" % (bad,))
    g = held[v["grant"]]
    if v["request"] not in ("fetch", "fill"):
        raise ValueError("a request is a fetch or a fill")
    start = v["start"]
    if not (isinstance(start, list) and len(start) == 2 and all(is_int(x) for x in start) and is_int(v["live_edge"])):
        raise ValueError("a live edge and a start Location of integers")
    if seeks_back(v["live_edge"], start) and not g["powers"] & SEEK:
        return {"outcome": "refuse", "reason": "seek_not_granted"}
    return {"outcome": "serve"}


# ---------------------------------------------------------------- §9 room verification

def authority_policy():
    """live-events §5: the default bounds, every kind permitted."""
    b = b"WEP1\x01" + bitset(LEV.KINDS) + struct.pack("<HHIQIQ", 1024, 500, 50, 1000000, 1, 1000000)
    return LEV.read_policy(b)


def merged_order(d, chains):
    """§9.2."""
    def key(c, i):
        try:
            ev = LEV.read_record(chains[c][i])
        except Refuse:
            return (0, 0, c)
        return (ev["t"], 0 if is_room_key(d, ev["principal"]) else 1, c)
    nxt, out = [0] * len(chains), []
    while True:
        live = [c for c in range(len(chains)) if nxt[c] < len(chains[c])]
        if not live:
            return out
        c = min(live, key=lambda c: key(c, nxt[c]))
        out.append((c, nxt[c]))
        nxt[c] += 1


def verify_room(descriptor, grants, head, chains):
    """§9: a dict of what the corpus states for the vector."""
    try:
        d = read_descriptor(descriptor)
    except Refuse as e:
        return {"outcome": "refuse", "reason": e.reason, "at": ["descriptor"]}
    try:
        revoked_set = check_key_log(d, head)
    except Refuse as e:
        return {"outcome": "refuse", "reason": e.reason, "at": ["key_log"]}
    held, bad = check_grants(d, grants)
    if bad:
        return {"outcome": "refuse", "reason": bad[1], "at": ["grant", bad[0]]}
    room = Room(d, held, revoked_set)
    auth = authority_policy()
    ver = LEV.Verifier(d["room"], auth, {})
    seen = {}
    order, at_index = [], []
    for c, i in merged_order(d, chains):
        b = chains[c][i]
        at_index.append((c, i))
        try:
            under = step(d, room, ver, seen, auth, b)
        except Refuse as e:
            return {"outcome": "refuse", "reason": e.reason, "at": ["record", c, i]}
        order.append([c, i, under.hex() if under else None])
    if ver.gaps:
        return {"outcome": "gap", "reason": "delivery_gap", "room": d["room"].hex(),
                "gaps": [{"principal": p.hex(), "first": a, "last": z} for p, a, z in ver.gaps], "order": order}
    mismatch = LEV.stated_tally_mismatch(ver.accepted)
    if mismatch is not None:
        c, i = at_index[mismatch]
        return {"outcome": "refuse", "reason": "tally_mismatch", "at": ["record", c, i]}
    return {"outcome": "accept", "room": d["room"].hex(), "grant_digests": [g["digest"].hex() for g in held],
            "order": order, "state": state_of(d, room), "tally": LEV.tally(ver.accepted),
            "moqt": [{"grant": g["digest"].hex(), "reval": d["reval"], "seek": g["powers"] & SEEK != 0,
                      "scopes": scopes(d, g)} for g in held]}


def step(d, room, ver, seen, auth, b):
    """§9.2 over one record: the digest of the grant it was verified under, or None."""
    ev = LEV.read_record(b)
    if ev["kind"] not in LEV.KINDS:
        raise Refuse("unknown_kind")
    if ev["room"] != d["room"]:
        raise Refuse("wrong_room")
    bad = ED.verify(ev["principal"], LEV.SIGN_DOMAIN + ev["body"], ev["sig"])
    if bad:
        raise Refuse(bad)
    p = ev["principal"]
    if ev["digest"] in seen:
        policy, under = seen[ev["digest"]]
        ver.own[p] = policy
        ver.offer(b)
        return under
    if is_room_key(d, p):
        ver.own[p] = auth
        ver.offer(b)
        seen[ev["digest"]] = (auth, None)
        payload = ver.accepted[-1][2]
        if 0x10 <= ev["kind"] <= 0x16:
            room.apply(ev["kind"], payload["subject"], read_detail(ev["kind"], payload["detail"]), ev["t"])
        return None
    g = room.current(p)
    if g is None:
        raise Refuse("not_present")
    if not good_at(g, d, ev["t"]):
        raise Refuse("outside_grant_window")
    ver.own[p] = g["policy"]
    ver.offer(b)
    seen[ev["digest"]] = (g["policy"], g["digest"])
    payload = ver.accepted[-1][2]
    if payload["kind"] == CONTROL:
        need = RECORD if payload["command"] in (1, 2) else DIRECT
        if g["powers"] & need == 0:
            raise Refuse("power_not_granted")
    model = ai_model(g)
    if model is not None and payload["kind"] in (0x03, 0x04, 0x0A):
        if payload["origin"] not in (2, 3) or payload["model"] != model:
            raise Refuse("ai_unmarked")
    return g["digest"]


def state_of(d, room):
    """§9.3, as the corpus writes it."""
    def precedence(p):
        g = room.current(p)
        return g["precedence"] if g else None
    tracks = []
    for (p, m), (_, layers) in room.own.items():
        tracks += [render([d["room"].hex(), p.hex(), MEDIA[m]], str(n)) for n in range(layers)]
    for s, st in zip(d["slots"], room.slots):
        if st["live"]:
            tracks += [render([d["room"].hex(), "_" + s["name"].decode(), MEDIA[s["media"]]], str(n)) for n in range(st["live"][2])]
    return {
        "present": [{"principal": p.hex(), "grant": g.hex()} for p, g in sorted(room.present.items())],
        "published": [{"principal": p.hex(), "media": MEDIA[m], "profile": d["profiles"][v[0]]["name"].decode(), "layers": v[1]}
                      for (p, m), v in sorted(room.own.items())],
        "slots": [{"slot": s["name"].decode(), "media": MEDIA[s["media"]],
                   "live": st["live"][0].hex() if st["live"] else None,
                   "standby": st["standby"][0].hex() if st["standby"] else None,
                   "precedence": precedence(st["live"][0]) if st["live"] else None}
                  for s, st in zip(d["slots"], room.slots)],
        "tracks": sorted(tracks),
        "timeline": [{"principal": p.hex(), "grant": g.hex(), "t": t} for p, g, t in room.timeline],
    }


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

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)
    stated = {p["name"]: p["record"] for p in doc["registered_profiles"]}
    mine = {n: b.hex() for n, b in REGISTERED.items()}
    if stated != mine:
        fails.append("the registered profiles differ from §8.2's table: %s" % sorted(n for n in set(stated) | set(mine) if stated.get(n) != mine.get(n)))
    print("%s registered profiles (§8.2)" % ("PASS" if stated == mine else "FAIL"))
    reasons_seen, outcomes = set(), {}
    for v in doc["vectors"]:
        name = v["name"]
        got = verify_room(bytes.fromhex(v["descriptor"]), [bytes.fromhex(g) for g in v["grants"]], v["key_log_head"],
                          [[bytes.fromhex(r) for r in c] for c in v["chains"]])
        problems = [k for k in got if v.get(k) != got[k]]
        problems += [k for k in ("outcome", "reason", "at") if k in v and k not in got]
        if v.get("reason"):
            reasons_seen.add(v["reason"])
        outcomes[name] = got["outcome"]
        what = got["outcome"] + ("" if got["outcome"] == "accept" else " " + got["reason"]) + (" at %s" % got["at"] if "at" in got else "")
        print(("PASS" if not problems else "FAIL") + " vector %s: %s" % (name, what))
        for k in problems:
            fails.append("%s: %s is %s, the corpus states %s" % (name, k, json.dumps(got.get(k))[:300], json.dumps(v.get(k))[:300]))
    proofs = doc.get("proofs", [])
    if not proofs:
        fails.append("the corpus holds no proof vector")
    for v in proofs:
        got = proof_outcome(v)
        want = {k: v[k] for k in ("outcome", "reason") if k in v}
        if want.get("reason"):
            reasons_seen.add(want["reason"])
        ok = got == want
        print(("PASS" if ok else "FAIL") + " proof %s: %s" % (v["name"], got["outcome"] + (" " + got["reason"] if "reason" in got else "")))
        if not ok:
            fails.append("%s: %s, the corpus states %s" % (v["name"], got, want))
    seek = doc.get("seek", [])
    if not seek:
        fails.append("the corpus holds no seek request")
    for v in seek:
        try:
            got = seek_outcome(doc, v)
        except (ValueError, StopIteration, IndexError, KeyError, TypeError, Refuse) as e:
            got = {"outcome": "unreadable: %r" % (e,)}
        want = {k: v[k] for k in ("outcome", "reason") if k in v}
        if want.get("reason"):
            reasons_seen.add(want["reason"])
        ok = got == want
        print(("PASS" if ok else "FAIL") + " seek %s: %s" % (v["name"], got["outcome"] + (" " + got["reason"] if "reason" in got else "")))
        if not ok:
            fails.append("%s: %s, the corpus states %s" % (v["name"], got, want))
    if {v.get("outcome") for v in seek} != {"serve", "refuse"}:
        fails.append("seek does not hold both a request served and one refused")
    missing = [r for r in REASONS if r not in reasons_seen]
    if missing:
        fails.append("no vector refused with %s" % ", ".join(missing))
    rooms = ["lesson", "broadcast", "takeover", "redundant_failover", "promote_viewer", "dvr", "ai_participant"]
    absent = [r for r in rooms if outcomes.get(r) != "accept"]
    if absent:
        fails.append("rooms not accepted: %s" % ", ".join(absent))
    print("%s every reason of §11 refuses a vector, a proof or a request, and every room of §12 is accepted"
          % ("PASS" if not (missing or absent) else "FAIL"))
    for f in fails:
        print("FAIL " + f)
    print(("PASS" if not fails else "FAIL") + " live-rooms corpus: %d vectors, %d proofs, %d seek requests"
          % (len(doc["vectors"]), len(proofs), len(seek)))
    return 1 if fails else 0


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