d07295fdc3
live_messaging_verify.py reads descriptors with live_rooms_verify.py, Key IDs, keys, AES-GCM and plaintexts with e2ee_verify.py, and implements the rest: every record, the receipts-track object of jwp-receipts' binary form, a conversation entry by entry, what a principal is owed, and calls. It agrees with wai-rs on every vector, fails every mutated copy of the corpus, and catches each of its own checks dropped. The corpus gains receipt-level and plaintext-level controls, a conversation under AES_256_GCM_SHA512_128, and history requests that state whether the requester is admitted (§9.3). Co-Authored-By: Claude Opus 5.5 <[email protected]>
889 lines
39 KiB
Python
889 lines
39 KiB
Python
#!/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]))
|