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

Written from the extension's text and the texts it cites, alone. It shares no code with the
reference implementation in Rust, which generates the corpus, and never runs it, nor the
generator. It reads with the second implementations of the corpora this extension builds on:

- room descriptors with live_rooms_verify.py (live-rooms §3.4, §3.2): their reader, their
  rules, their signature and the room id;
- confidentiality records, Key IDs, key derivation, AES-GCM and plaintexts with
  e2ee_verify.py (confidentiality §3.3, §4, §5.7, §6 step 1), which writes AES and GCM from
  their specifications for test vectors only;

and checks Ed25519 signatures with ../claim-conformance/_ed25519.py, WAI's strict rule (SPEC §7
*Signatures*) written from RFC 8032, and BLAKE3 with ../claim-conformance/_blake3.py; SHA-256 is
the standard library's. Everything else is here:

- every record of the extension, read as §2 reads a record: the conversation statement (§3),
  the epoch statement (§7.3), the message record (§5), the acknowledgement (§8), the call
  descriptor (§10.1), the invitation (§10.2) and the answer (§10.3);
- a message's receipts-track object (§4.3, §11.3 step 6): the binary form of jwp-receipts
  §2.7, the labelled signing payload of §2.3, the link hash of §3 step 3, the content hash of
  §2.1, the leaf, the proof's position and the root of §2.2, and the group total of §3 for a
  group of one object; the chain kept as §11.3 step 6 pins it, the head alone. Of the
  acquisition classes it reads tag 0, the unmetered marker, which the extension requires; a
  receipt carrying another class is refused as `class_not_supported_here`, a verifier-local
  outcome no conforming corpus vector states;
- a conversation, entry by entry (§11): its descriptor, record and statement, its epoch
  statements, each message (steps 1 to 13: membership, the Key ID, the epoch, the owner, the
  group, the receipts, opening, the record, its position, its time, its kind and body, its
  target and what it does) and each acknowledgement, and the view §11.5 states;
- what a principal is owed of a log (§9.3), given whether it is admitted;
- a call (§10.1 to §10.3) and every invitation's state.

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 every one.

    python3 -I live_messaging_verify.py <live-conformance/messaging>

Exit status 0 only if every vector agrees. 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_rooms_verify as LRV  # noqa: E402
import e2ee_verify as E2  # noqa: E402
from _blake3 import blake3  # noqa: E402
import _ed25519 as ED  # noqa: E402

EXT = os.path.join(HERE, "..", "extensions", "live-messaging.md")


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


# The checks this verifier can drop, one at a time, to show that the corpus catches each.
DROP = set()


def need(name, ok, code):
    """A check: refuse with `code` when it fails, unless it is dropped."""
    if not ok and name not in DROP:
        raise Refuse(code)


# §2, §3.1, §5.1, §7.3, §8, §10
MAGIC = {"statement": b"WMC1", "epoch": b"WME1", "message": b"WMG1", "ack": b"WMA1",
         "call": b"WCT1", "invitation": b"WCI1", "answer": b"WCA1"}
DOMAIN = {"statement": b"wai:live-conversation\x01", "epoch": b"wai:live-epoch\x01",
          "message": b"wai:live-message\x01", "ack": b"wai:live-message-ack\x01",
          "call": b"wai:live-call\x01", "invitation": b"wai:live-call-invite\x01",
          "answer": b"wai:live-call-answer\x01"}
assert all(len(DOMAIN[k]) == n for k, n in [("statement", 22), ("epoch", 15), ("message", 17), ("ack", 21),
                                             ("call", 14), ("invitation", 21), ("answer", 21)])
TEXT, REPLY, REACTION, EDIT, RETRACTION = 1, 2, 3, 4, 5
KIND_NAME = {TEXT: "text", REPLY: "reply", REACTION: "reaction", EDIT: "edit", RETRACTION: "retraction"}
DELIVERED, READ = 1, 2
ACCEPT, DECLINE, CANCEL, MISSED = 1, 2, 3, 4
STATE = {ACCEPT: "accepted", DECLINE: "declined", CANCEL: "cancelled", MISSED: "missed"}
RECEIPT_ALIAS = 5          # §4.3
PRIORITY = 64              # §4.2
MAX_TEXT, MAX_REACTION = 16384, 64
MIN_RING, MAX_RING, MAX_INVITEES = 1000, 300000, 32
ZERO = bytes(32)

# jwp-receipts §2.1, §2.2, §2.3, §3 step 3
OBJ_DOMAIN = b"moq-jwp:object\x01"
LEAF1 = b"moq-jwp:merkle-leaf\x01"
LEAF2 = b"moq-jwp:merkle-leaf\x02"
NODE = b"moq-jwp:merkle-node\x01"
SIGN1 = b"moq-jwp:group-sign\x01"
SIGN2 = b"moq-jwp:group-sign\x02"
LINK = b"moq-jwp:receipt-link\x01"


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


def sig_code(r):
    return r  # "bad_key" or "bad_signature", as SPEC §7 names them


# ============================================================================ §2 records

class Rd:
    def __init__(self, b, magic):
        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")
        self.b, self.p = b, 5

    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 long(self):
        return self.take(self.u16())

    def sig(self):
        body = self.b[:self.p]
        s = self.take(64)
        if self.p != len(self.b):
            raise Refuse("malformed")
        return body, s


def read(kind, b):
    """The record's fields, the bytes its digest and signature cover, and its signature."""
    r = Rd(b, MAGIC[kind])
    if kind == "statement":
        f = {"room": r.take(32), "max_text": r.u16(), "max_reaction": r.u8(), "retention_days": r.u16()}
    elif kind == "epoch":
        f = {"room": r.take(32), "generation": r.u32(), "epoch": r.u64(), "opened_by": r.take(32), "prev": r.take(32)}
        n = r.u32()
        f["roster"] = []
        for _ in range(n):
            f["roster"].append((r.u32(), r.take(32)))
    elif kind == "message":
        f = {"room": r.take(32), "author": r.take(32), "group": r.u64(), "t": r.u64(), "kind": r.u8(),
             "target": r.take(32), "body": r.long()}
    elif kind == "ack":
        f = {"room": r.take(32), "member": r.take(32), "author": r.take(32), "through": r.u64(),
             "link": r.take(32), "kind": r.u8(), "t": r.u64()}
    elif kind == "call":
        f = {"room": r.take(32), "caller": r.take(32), "media": r.u8(), "ring_ms": r.u32()}
        f["invitees"] = [r.take(32) for _ in range(r.u8())]
    elif kind == "invitation":
        f = {"room": r.take(32), "call": r.take(32), "caller": r.take(32), "callee": r.take(32), "t": r.u64()}
    else:
        f = {"room": r.take(32), "invitation": r.take(32), "signer": r.take(32), "kind": r.u8(), "t": r.u64()}
    body, s = r.sig()
    f["body_bytes"], f["sig"] = body, s
    f["digest"] = sha256(body)
    return f


def verify_sig(kind, key, f, check="signature"):
    bad = ED.verify(key, DOMAIN[kind] + f["body_bytes"], f["sig"])
    need(check, bad is None, bad or "bad_signature")


def record_outcome(kind, b, room, key):
    try:
        f = read(kind, b)
        need("record_room", f["room"] == room, "wrong_room")
        verify_sig(kind, key, f)
        return "ok"
    except Refuse as e:
        return e.code


# ===================================================================== §4.3 receipts

class Bin:
    """Big-endian fields of jwp-receipts §2.7; running out, or anything left, is `malformed`."""

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

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

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

    def u32(self):
        return int.from_bytes(self.take(4), "big")

    def u64(self):
        return int.from_bytes(self.take(8), "big")

    def string(self):
        s = self.take(self.u32())
        try:
            s.decode("utf-8")
        except UnicodeDecodeError:
            raise Refuse("malformed")
        return s


def read_class(r):
    tag = r.u8()
    if tag != 0:
        # energy-measurement §2.7's other classes carry evidence this verifier does not read.
        raise Refuse("class_not_supported_here")
    return tag


def read_receipts(b):
    """jwp-receipts §2.7.3 and §2.7.4: (group, [objects])."""
    r = Bin(b)
    if r.take(4) != b"WJR1":
        raise Refuse("malformed")
    start = r.p
    domain = r.take(19)
    if domain not in (SIGN1, SIGN2):
        raise Refuse("malformed")
    g = {"labelled": domain == SIGN2, "track_alias": r.u64(), "group_id": r.u64(), "root": r.take(32),
         "leaf_count": r.u64(), "joules_total": r.u64(), "group_start": r.string(), "group_end": r.string(),
         "signer": r.take(32), "signer_id": r.string()}
    if g["labelled"]:
        g["class"] = read_class(r)
        g["coverage"] = r.u8()
        if g["coverage"] not in (0, 1):
            raise Refuse("malformed")
        flag, parent = r.u8(), r.take(32)
        payload_end = r.p
        g["sig"] = r.take(64)
        canonical = b[start:r.p]
        payload = b[start:payload_end]
    else:
        payload_end = r.p
        g["sig"] = r.take(64)
        canonical = b[start:r.p]
        payload = b[start:payload_end]
        flag, parent = r.u8(), r.take(32)
    if flag not in (0, 1) or (flag == 0 and parent != ZERO):
        raise Refuse("malformed")
    g["parent"] = parent if flag == 1 else None
    g["payload"], g["link"] = payload, blake3(LINK + canonical)
    count = r.u32()
    objs = []
    for _ in range(count):
        o = {"track_alias": g["track_alias"], "group_id": g["group_id"], "object_id": r.u64(),
             "content_hash": r.take(32), "joules": r.u64(), "origin": r.string()}
        n = r.u8()
        o["siblings"] = [r.take(32) for _ in range(n)]
        o["direction_bits"], o["leaf_index"] = r.u64(), r.u64()
        o["leaf_version"] = r.u8()
        if o["leaf_version"] not in (1, 2):
            raise Refuse("malformed")
        o["class"] = read_class(r) if o["leaf_version"] == 2 else None
        objs.append(o)
    if r.p != len(b):
        raise Refuse("malformed")
    return g, objs


def leaf_of(o):
    if o["leaf_version"] == 1:
        return blake3(LEAF1 + o["content_hash"])
    return blake3(LEAF2 + o["content_hash"] + o["joules"].to_bytes(8, "big") + bytes([o["class"]])
                  + len(o["origin"]).to_bytes(4, "big") + o["origin"])


def walk(leaf, siblings, bits):
    v = leaf
    for i, s in enumerate(siblings):
        v = blake3(NODE + s + v) if (bits >> i) & 1 else blake3(NODE + v + s)
    return v


def proof_len(n):
    k = 0
    while (1 << k) < n:
        k += 1
    return k


def check_receipts(author, group, b, sealed, head):
    """§11.3 step 6: the group receipt when it passes."""
    g, objs = read_receipts(b)
    need("one_object", len(objs) == 1, "malformed")
    o = objs[0]
    need("labelled", g["labelled"], "receipt_unlabelled")
    need("receipt_track", g["signer"] == author and g["track_alias"] == RECEIPT_ALIAS and g["group_id"] == group
         and o["object_id"] == 0, "moqt_mismatch")
    # jwp-receipts §3 steps 1 and 2: the key, the signature, the group's own class.
    bad = ED.verify(g["signer"], g["payload"], g["sig"])
    need("receipt_key", bad != "bad_key", "bad_signer_key")
    need("receipt_signature", bad is None, "signature_verify_failed")
    need("group_label", g["joules_total"] == 0 and g["coverage"] == 0, "group_label_invalid")
    # Step 3, the head alone (§11.3 step 6).
    if head is None:
        need("chain", g["parent"] is None, "parent_not_held")
    else:
        need("chain", g["parent"] is not None, "chain_break")
        need("chain", g["parent"] == head, "parent_not_held")
    # Step 4, the content hash over the sealed payload.
    need("content_hash", blake3(OBJ_DOMAIN + sealed) == o["content_hash"], "content_hash_mismatch")
    # Step 5: the label, the position, the inclusion.
    need("object_label", o["leaf_version"] == 1 or o["joules"] == 0, "object_label_invalid")
    need("position", o["leaf_index"] < g["leaf_count"] and len(o["siblings"]) == proof_len(g["leaf_count"])
         and o["direction_bits"] == o["leaf_index"] and o["object_id"] == o["leaf_index"], "proof_position_mismatch")
    need("inclusion", walk(leaf_of(o), o["siblings"], o["direction_bits"]) == g["root"], "inclusion_proof_failed")
    # The group total: one object for its one leaf.
    need("total", g["leaf_count"] != 1 or o["joules"] == g["joules_total"], "group_total_mismatch")
    return g


def link_of(b):
    try:
        return read_receipts(b)[0]["link"]
    except Refuse:
        return None


# ===================================================================== §4.2 sealing

def ns_of(room, author):
    return [room.hex().encode(), author.hex().encode(), b"messages"]


def open_message(suite, secret, room, author, group, props, sealed):
    """§11.3 step 7: the original payload."""
    kid, sip = E2.key_id_of(props)
    base = E2.track_base_key(secret, ns_of(room, author))
    key, salt = E2.derive_key_salt(suite, kid, base, E2.ser_ftn(ns_of(room, author), b"0"))
    nonce = E2.xor(salt, E2.ctr_of(group, 0))
    pt = E2.open_(suite, key, nonce, E2.aad_of(group, 0, PRIORITY, sip), sealed)
    need("aead", pt is not None, "bad_tag")
    payload, digest = E2.plaintext_read(pt)
    if digest is not None:
        need("inner_digest", digest == blake3(payload), "inner_digest_mismatch")
    return payload


# ===================================================================== §11 a conversation

class Conversation:
    def __init__(self, descriptor, record, statement):
        try:
            d = LRV.read_descriptor(descriptor)
        except LRV.Refuse as e:
            raise At(e.reason, "descriptor")
        self.room, self.authority, self.clock = d["room"], d["authority"], d["clock"]
        if record is None:
            raise At("not_encrypted", "record")
        try:
            self.suite = E2.check_record(record, self.room, self.authority, d["mode"])
        except E2.Refuse as e:
            raise At(e.code, "record")
        try:
            s = read("statement", statement)
            need("statement_rules", 1 <= s["max_text"] <= MAX_TEXT and 1 <= s["max_reaction"] <= MAX_REACTION,
                 "conversation_invalid")
            need("statement_room", s["room"] == self.room, "wrong_room")
            verify_sig("statement", self.authority, s)
        except Refuse as e:
            raise At(e.code, "statement")
        self.max_text, self.max_reaction = s["max_text"], s["max_reaction"]
        self.epochs = []          # (position, fields)
        self.tracks = {}          # author: {"next", "head", "last_t", "groups": {group: link}}
        self.messages = []        # views, in order
        self.by_link = {}         # link: index into messages, accepted only
        self.acks = []
        self.acked = {}           # author: member: [delivered, read]

    def latest(self):
        return self.epochs[-1][1] if self.epochs else None

    # §7.3
    def offer_epoch(self, pos, b):
        s = read("epoch", b)
        need("epoch_room", s["room"] == self.room, "wrong_room")
        verify_sig("epoch", self.clock, s, "epoch_signature")
        leaves = [l for l, _ in s["roster"]]
        keys = [k for _, k in s["roster"]]
        need("epoch_rules", s["epoch"] < 1 << 40 and len(leaves) >= 1 and all(a < b for a, b in zip(leaves, leaves[1:]))
             and all(l < 1 << 24 for l in leaves) and len(set(keys)) == len(keys), "epoch_invalid")
        prev = self.latest()["digest"] if self.epochs else ZERO
        need("epoch_link", s["prev"] == prev, "epoch_unlinked")
        p = self.latest()
        if p is None:
            ok = s["epoch"] == 0
        else:
            ok = (s["generation"] == p["generation"] and s["epoch"] == p["epoch"] + 1) or \
                 (s["generation"] > p["generation"] and s["epoch"] == 0)
        need("epoch_order", ok, "epoch_out_of_order")
        s["members"] = set(keys)
        self.epochs.append((pos, s))

    # §11.3 steps 1 to 6
    def steps_1_to_6(self, m):
        e = self.latest()
        need("epoch_known", e is not None, "epoch_unknown")
        need("member", m["author"] in e["members"], "sender_not_member")
        kid, _ = E2.key_id_of(m["properties"])
        epoch, leaf = kid >> 24, kid & 0xFFFFFF
        need("stale", epoch == e["epoch"], "epoch_stale")
        at = dict(e["roster"]).get(leaf)
        need("leaf", at is not None, "key_id_unknown")
        need("owner", at == m["author"], "sender_not_owner")
        t = self.tracks.get(m["author"])
        need("gap", m["group"] == (t["next"] if t else 0), "message_gap")
        return check_receipts(m["author"], m["group"], m["receipts"], m["payload"], t["head"] if t else None)

    def offer_message(self, pos, m, secret):
        view = {"position": pos, "author": m["author"], "group": m["group"], "link": link_of(m["receipts"]),
                "outcome": None}
        try:
            g = self.steps_1_to_6(m)
        except (Refuse, E2.Refuse) as e:
            view["outcome"] = getattr(e, "code")
            self.messages.append(view)
            return
        view["link"] = g["link"]
        t = self.tracks.setdefault(m["author"], {"next": 0, "head": None, "last_t": None, "groups": {}})
        t["next"], t["head"] = m["group"] + 1, g["link"]
        t["groups"][m["group"]] = g["link"]
        if secret is None:
            view["outcome"] = "sealed"
            self.messages.append(view)
            return
        try:
            payload = open_message(self.suite, secret, self.room, m["author"], m["group"], m["properties"], m["payload"])
            rec = self.content_steps(m, t, payload)
        except (Refuse, E2.Refuse) as e:
            view["outcome"] = getattr(e, "code")
            self.messages.append(view)
            return
        view.update(outcome="accepted", kind=rec["kind"], t=rec["t"],
                    target=None if rec["kind"] == TEXT else rec["target"], body=rec["body"],
                    text=rec["body"] if rec["kind"] in (TEXT, REPLY) else None, edits=0, reactions=[],
                    replies=[], retracted_by=None)
        i = len(self.messages)
        self.messages.append(view)
        self.by_link[g["link"]] = i
        self.apply(view)

    # §11.3 steps 8 to 12
    def content_steps(self, m, t, payload):
        rec = read("message", payload)
        verify_sig("message", rec["author"], rec, "message_signature")
        need("position_bound", rec["room"] == self.room and rec["author"] == m["author"] and rec["group"] == m["group"],
             "message_mismatch")
        need("time", t["last_t"] is None or rec["t"] >= t["last_t"], "time_reversed")
        t["last_t"] = rec["t"]
        k, body = rec["kind"], rec["body"]
        need("kind", k in KIND_NAME, "unknown_kind")
        try:
            body.decode("utf-8")
            utf8 = True
        except UnicodeDecodeError:
            utf8 = False
        need("utf8", utf8, "bad_utf8")
        if k == RETRACTION:
            need("body", len(body) == 0, "body_invalid")
        else:
            bound = self.max_reaction if k == REACTION else self.max_text
            need("length", len(body) <= bound, "text_too_long")
            need("body", len(body) > 0, "body_invalid")
        need("ref_form", (k == TEXT) == (rec["target"] == ZERO), "ref_invalid")
        if k != TEXT:
            ti = self.by_link.get(rec["target"])
            need("ref_known", ti is not None, "ref_unknown")
            if ti is None:
                return rec
            tv = self.messages[ti]
            need("ref_live", tv["retracted_by"] is None, "ref_retracted")
            tk = tv["kind"]
            if k == REPLY:
                need("thread", tk == TEXT, "thread_not_root")
            elif k in (REACTION, EDIT):
                need("target_kind", tk in (TEXT, REPLY), "target_kind")
            else:
                need("target_kind", tk in (TEXT, REPLY, REACTION), "target_kind")
            if k == REACTION:
                need("repeat", not any(w == rec["author"] and b == body for w, b, _ in tv["reactions"]),
                     "reaction_repeated")
            if k == EDIT:
                need("edit_author", tv["author"] == rec["author"], "edit_not_author")
            if k == RETRACTION:
                need("retract_author", tv["author"] == rec["author"], "retract_not_author")
        return rec

    # §11.3 step 13
    def apply(self, v):
        if v["kind"] == TEXT or v["target"] not in self.by_link:
            return
        tv = self.messages[self.by_link[v["target"]]]
        if v["kind"] == REPLY:
            tv["replies"].append(v["link"])
        elif v["kind"] == REACTION:
            tv["reactions"].append((v["author"], v["body"], v["link"]))
        elif v["kind"] == EDIT:
            tv["text"], tv["edits"] = v["body"], tv["edits"] + 1
        elif v["kind"] == RETRACTION:
            tv["retracted_by"], tv["text"] = v["link"], None
            if tv["kind"] == REACTION and tv["target"] in self.by_link:
                rt = self.messages[self.by_link[tv["target"]]]
                rt["reactions"] = [x for x in rt["reactions"] if x[2] != tv["link"]]

    # §8
    def offer_ack(self, pos, b):
        try:
            a = read("ack", b)
            need("ack_room", a["room"] == self.room, "wrong_room")
            verify_sig("ack", a["member"], a, "ack_signature")
            need("ack_kind", a["kind"] in (DELIVERED, READ), "ack_invalid")
            e = self.latest()
            need("ack_member", e is not None and a["member"] in e["members"], "ack_not_member")
            held = self.tracks.get(a["author"], {"groups": {}})["groups"].get(a["through"])
            need("ack_known", held is not None and held == a["link"], "ack_unknown")
            last = self.acked.get(a["author"], {}).get(a["member"], [None, None])[0 if a["kind"] == DELIVERED else 1]
            need("ack_order", last is None or a["through"] >= last, "ack_regressed")
        except Refuse as e:
            self.acks.append({"position": pos, "outcome": e.code})
            return
        cell = self.acked.setdefault(a["author"], {}).setdefault(a["member"], [None, None])
        if a["kind"] == DELIVERED:
            cell[0] = a["through"]
        else:
            cell[1] = a["through"]
            cell[0] = a["through"] if cell[0] is None else max(cell[0], a["through"])
        self.acks.append({"position": pos, "outcome": "accepted"})

    # §11.5
    def outcome(self):
        msgs = []
        for v in self.messages:
            o = {"position": v["position"], "author": v["author"].hex(), "group": v["group"],
                 "link": v["link"].hex() if v["link"] else None, "outcome": v["outcome"]}
            if v["outcome"] == "accepted":
                o["kind"] = KIND_NAME[v["kind"]]
                o["t"] = v["t"]
                o["target"] = v["target"].hex() if v["target"] else None
                o["retracted_by"] = v["retracted_by"].hex() if v["retracted_by"] else None
                if v["kind"] in (TEXT, REPLY):
                    o["status"] = "retracted" if v["retracted_by"] else ("edited" if v["edits"] else "shown")
                    o["text"] = v["text"].decode("utf-8") if v["text"] is not None else None
                    o["edits"] = v["edits"]
                    o["reactions"] = [[w.hex(), b.decode("utf-8")] for w, b, _ in v["reactions"]]
                if v["kind"] == TEXT:
                    o["replies"] = [l.hex() for l in v["replies"]]
            msgs.append(o)
        acked = {}
        for author in sorted(self.acked):
            acked[author.hex()] = {m.hex(): {"delivered": c[0], "read": c[1]} for m, c in sorted(self.acked[author].items())}
        return {"accept": True,
                "epochs": [{"position": p, "generation": s["generation"], "epoch": s["epoch"],
                            "roster": [[l, k.hex()] for l, k in s["roster"]]} for p, s in self.epochs],
                "messages": msgs, "acks": self.acks, "acknowledged": acked}


class At(Exception):
    def __init__(self, code, where):
        super().__init__(code)
        self.code, self.where = code, where


def verify_conversation(inp):
    try:
        c = Conversation(bytes.fromhex(inp["descriptor"]), None if inp["record"] is None else bytes.fromhex(inp["record"]),
                         bytes.fromhex(inp["statement"]))
    except At as e:
        return {"accept": False, "reason": e.code, "at": e.where}
    secrets = {(s["generation"], s["epoch"]): bytes.fromhex(s["secret"]) for s in inp["secrets"]}
    for e in inp["entries"]:
        pos = e["position"]
        if "epoch" in e:
            try:
                c.offer_epoch(pos, bytes.fromhex(e["epoch"]))
            except Refuse as r:
                return {"accept": False, "reason": r.code, "at": {"entry": pos}}
        elif "ack" in e:
            c.offer_ack(pos, bytes.fromhex(e["ack"]))
        else:
            m = {k: (bytes.fromhex(v) if isinstance(v, str) else v) for k, v in e["message"].items()}
            last = c.latest()
            secret = secrets.get((last["generation"], last["epoch"])) if last else None
            c.offer_message(pos, m, secret)
    return c.outcome()


# ===================================================================== §9.3 owed

def owed(entries, key, frm, admitted):
    end = entries[-1]["position"] + 1 if entries else 0
    kinds = ["membership"] * end
    rosters = {}
    for e in entries:
        if "epoch" in e:
            kinds[e["position"]] = "epoch"
            rosters[e["position"]] = {k for _, k in read("epoch", bytes.fromhex(e["epoch"]))["roster"]}
        else:
            kinds[e["position"]] = "content"
    roster, out = None, []
    for i, k in enumerate(kinds):
        if k == "epoch":
            roster = rosters[i]
        elif k == "content" and i >= frm and roster is not None and key in roster:
            out.append(i)
    if not out and not admitted and "member_rule" not in DROP:
        return {"refused": "not_a_member"}
    return {"owed": out}


# ===================================================================== §10 calls

def verify_call(v):
    try:
        d = LRV.read_descriptor(bytes.fromhex(v["descriptor"]))
    except LRV.Refuse as e:
        return {"accept": False, "reason": e.reason, "at": "descriptor"}
    room = d["room"]
    try:
        c = read("call", bytes.fromhex(v["call"]))
        inv = c["invitees"]
        need("call_rules", c["media"] in (1, 2, 3) and MIN_RING <= c["ring_ms"] <= MAX_RING and 1 <= len(inv) <= MAX_INVITEES
             and len(set(inv)) == len(inv) and c["caller"] not in inv, "call_invalid")
        need("call_room", c["room"] == room, "wrong_room")
        verify_sig("call", d["authority"], c, "call_signature")
    except Refuse as e:
        return {"accept": False, "reason": e.code, "at": "call"}
    rung = []
    for i, h in enumerate(v["invitations"]):
        try:
            x = read("invitation", bytes.fromhex(h))
            need("invitation_bound", x["room"] == room and x["call"] == c["digest"] and x["caller"] == c["caller"],
                 "invitation_mismatch")
            verify_sig("invitation", x["caller"], x, "invitation_signature")
            need("invited", x["callee"] in inv, "callee_not_invited")
            need("once", all(r["callee"] != x["callee"] for r in rung), "already_invited")
        except Refuse as e:
            return {"accept": False, "reason": e.code, "at": {"invitation": i}}
        rung.append({"callee": x["callee"], "t": x["t"], "deadline": x["t"] + c["ring_ms"] * 1000, "digest": x["digest"],
                     "caller": x["caller"], "state": "ringing", "at": None})
    for i, h in enumerate(v["answers"]):
        try:
            a = read("answer", bytes.fromhex(h))
            need("answer_room", a["room"] == room, "wrong_room")
            r = next((r for r in rung if r["digest"] == a["invitation"]), None)
            need("answer_known", r is not None, "answer_unknown")
            need("answer_kind", a["kind"] in STATE, "answer_invalid")
            signer = {ACCEPT: r["callee"], DECLINE: r["callee"], CANCEL: r["caller"], MISSED: d["clock"]}[a["kind"]]
            need("answer_signer", a["signer"] == signer, "answer_wrong_signer")
            verify_sig("answer", a["signer"], a, "answer_signature")
            need("answer_once", r["state"] == "ringing", "already_answered")
            need("answer_after", a["t"] >= r["t"], "answer_early")
            if a["kind"] == MISSED:
                need("missed_time", a["t"] >= r["deadline"], "missed_early")
            else:
                need("answer_time", a["t"] < r["deadline"], "answer_late")
        except Refuse as e:
            return {"accept": False, "reason": e.code, "at": {"answer": i}}
        r["state"], r["at"] = STATE[a["kind"]], a["t"]
    return {"accept": True, "invitations": [{"callee": r["callee"].hex(), "t": r["t"], "deadline": r["deadline"],
                                             "state": r["state"], "at": r["at"]} for r in rung]}


# ===================================================================== the corpus

def reasons_of_extension():
    """§12's table, in its order."""
    with open(EXT, encoding="utf-8") as f:
        text = f.read()
    at = text.index("\n## 12. Reasons")
    end = text.index("\n## ", at + 1)
    return re.findall(r"^\| `([a-z_]+)` \|", text[at:end], re.M)


def check(doc, quiet=False, first=False):
    fails, seen = [], set()

    def fail(msg):
        fails.append(msg)
        if not quiet:
            print("FAIL " + msg)

    logs = {}
    for v in doc["conversations"]:
        try:
            got = verify_conversation(v["input"])
        except Exception as e:  # a corpus that does not read, or a check dropped, fails here
            got = {"error": "%s: %s" % (type(e).__name__, e)}
        if got != v["expect"]:
            fail("conversation %s: the outcome differs from the corpus's" % v["name"])
            if first:
                return fails, seen
        else:
            if not got["accept"]:
                seen.add(got["reason"])
            else:
                seen.update(m["outcome"] for m in got["messages"])
                seen.update(a["outcome"] for a in got["acks"])
        logs[v["name"]] = v["input"]["entries"]
    for h in doc["history"]:
        try:
            got = owed(logs[h["conversation"]], bytes.fromhex(h["key"]), h["from"], h["admitted"])
        except Exception as e:
            got = {"error": "%s: %s" % (type(e).__name__, e)}
        if got != h["expect"]:
            fail("history of %s from %d: %s where the corpus states %s" % (h["requester"], h["from"], got, h["expect"]))
            if first:
                return fails, seen
        seen.update([got.get("refused")] if "refused" in got else [])
    for v in doc["calls"]:
        try:
            got = verify_call(v)
        except Exception as e:
            got = {"accept": None, "error": "%s: %s" % (type(e).__name__, e)}
        if got != v["expect"]:
            fail("call %s: %s where the corpus states %s" % (v["name"], got, v["expect"]))
            if first:
                return fails, seen
        if got.get("accept") is False:
            seen.add(got["reason"])
    for r in doc["records"]:
        got = record_outcome(r["record"], bytes.fromhex(r["bytes"]), bytes.fromhex(r["room"]), bytes.fromhex(r["key"]))
        if got != r["expect"]:
            fail("record %s: %s where the corpus states %s" % (r["about"], got, r["expect"]))
            if first:
                return fails, seen
    if not quiet:
        print("%s %d conversations, %d history requests, %d calls, %d records" % (
            "PASS" if not fails else "FAIL", len(doc["conversations"]), len(doc["history"]), len(doc["calls"]),
            len(doc["records"])))
    return fails, seen


def mutations(doc):
    """Copies of the corpus, each with one defect a conforming verifier must catch."""
    cv = {v["name"]: k for k, v in enumerate(doc["conversations"])}
    calls = {v["name"]: k for k, v in enumerate(doc["calls"])}

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

    def entries(d, name):
        return d["conversations"][cv[name]]["input"]["entries"]

    def expect(d, name):
        return d["conversations"][cv[name]]["expect"]

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

    def first_message(d, name, who=None):
        for e in entries(d, name):
            if "message" in e and (who is None or e["message"]["author"] == who):
                return e["message"]
        raise KeyError(name)

    def view_where(d, name, f):
        return next(m for m in expect(d, name)["messages"] if f(m))

    def index_where(d, name, f):
        return next(i for i, m in enumerate(expect(d, name)["messages"]) if f(m))

    def edited(m):
        return m.get("status") == "edited" and m.get("kind") == "text"

    def retracted(m):
        return m.get("status") == "retracted"

    alice, bob = doc["principals"]["alice"], doc["principals"]["bob"]
    return [
        mutate("a byte of a sealed payload", lambda d: first_message(d, "conversation", alice).update(
            payload=flip(first_message(d, "conversation", alice)["payload"], 3))),
        mutate("a byte of a receipts object's signature", lambda d: first_message(d, "conversation", bob).update(
            receipts=flip(first_message(d, "conversation", bob)["receipts"], -100))),
        mutate("a message's group", lambda d: first_message(d, "conversation", bob).update(group=7)),
        mutate("a message's author", lambda d: first_message(d, "conversation", bob).update(author=alice)),
        mutate("an epoch statement's byte", lambda d: entries(d, "removal")[2].update(epoch=flip(entries(d, "removal")[2]["epoch"], 50))),
        mutate("an epoch secret withheld", lambda d: d["conversations"][cv["conversation"]]["input"]["secrets"].pop()),
        mutate("an edited text stated as first written", lambda d: view_where(d, "conversation", edited).update(text="lunch at noon?")),
        mutate("a retracted text stated shown", lambda d: view_where(d, "conversation", retracted).update(status="shown")),
        mutate("a retracted text left out of the view", lambda d: expect(d, "conversation")["messages"].pop(index_where(d, "conversation", retracted))),
        mutate("a stated reaction dropped", lambda d: view_where(d, "conversation", edited)["reactions"].clear()),
        mutate("a refusal's reason", lambda d: view_where(d, "conversation", lambda m: m["outcome"] == "edit_not_author").update(outcome="retract_not_author")),
        mutate("an acknowledgement's outcome", lambda d: expect(d, "conversation")["acks"][0].update(outcome="ack_unknown")),
        mutate("a stated read position", lambda d: next(iter(next(iter(expect(d, "conversation")["acknowledged"].values())).values())).update(read=99)),
        mutate("a history answer", lambda d: d["history"][1]["expect"].update(owed=[5, 6, 7, 10])),
        mutate("a removed member stated owed", lambda d: d["history"][2].update(expect={"owed": []})),
        mutate("a call's state", lambda d: d["calls"][calls["group_call"]]["expect"]["invitations"][2].update(state="ringing")),
        mutate("an answer's byte", lambda d: d["calls"][calls["one_to_one"]].update(answers=[flip(d["calls"][calls["one_to_one"]]["answers"][0], 120)])),
        mutate("an invitation's byte", lambda d: d["calls"][calls["one_to_one"]].update(invitations=[flip(d["calls"][calls["one_to_one"]]["invitations"][0], 40)])),
        mutate("a record's expectation", lambda d: d["records"][0].update(expect="truncated")),
    ]


CHECKS = ("member", "stale", "leaf", "owner", "gap", "labelled", "receipt_track", "receipt_signature", "chain",
          "content_hash", "aead", "message_signature", "position_bound", "time", "kind", "utf8", "length", "body",
          "ref_form", "ref_known", "ref_live", "thread", "target_kind", "repeat", "edit_author", "retract_author",
          "epoch_signature", "epoch_rules", "epoch_link", "epoch_order", "statement_rules", "statement_room",
          "ack_kind", "ack_member", "ack_known", "ack_order", "member_rule", "call_rules", "invitation_bound",
          "invitation_signature", "invited", "once", "answer_known", "answer_kind", "answer_signer", "answer_signature",
          "answer_once", "answer_after", "missed_time", "answer_time", "record_room", "signature", "one_object",
          "group_label", "position", "inclusion", "inner_digest", "epoch_room", "epoch_known", "ack_room",
          "ack_signature", "call_room", "call_signature", "answer_room")


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 []
    reasons = reasons_of_extension()
    ok = len(reasons) == 36 and len(set(reasons)) == len(reasons)
    print(("PASS" if ok else "FAIL") + " §12 read from the extension: %d reasons" % len(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 if r not in seen]
    print(("PASS" if not missing else "FAIL") + " every reason of §12 is given by a vector")
    fails += ["no vector 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("live_messaging_verify: %s" % ("PASS" if not fails else "FAIL (%d)" % len(fails)))
    for f in fails:
        print("  " + f)
    return 0 if not fails else 1


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