ab23c8ab6b
live-events §4.4 and §4.5: a chat and a reaction may state an origin after their last field (1 human, 2 machine, 3 machine-reviewed, with the model's digest for 2 and 3), as a caption does. live-rooms §4.4: the grant attribute `ai` names the model that acts for an AI participant; §9.2 step 6 refuses its chat, reaction or caption that does not state origin 2 or 3 with that model (`ai_unmarked`), and a relay does not forward one. Both corpora gain vectors (additions only); both second implementations read the new fields. Co-Authored-By: Claude Opus 5.5 <[email protected]>
769 lines
27 KiB
Python
769 lines
27 KiB
Python
#!/usr/bin/env python3
|
|
"""Independent verifier of the live-events corpus (extensions/live-events.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 with
|
|
../claim-conformance/_blake3.py, written from its specification. Everything else is here:
|
|
|
|
- the record (§3): its reader and refusals, its digest (SHA-256 without `sig`) and the
|
|
bytes its signature covers;
|
|
- every kind's payload (§4) and the text rules (§4.1);
|
|
- the policy record (§5) and every policy case;
|
|
- verification (§6): one record, each principal's chain in stream order with duplicates,
|
|
gaps and their refusals, both rates, and the outcome of a sequence;
|
|
- the poll tallies (§7): which votes count and which close closes, the feed0 records and
|
|
their state hash, and the check of a stated tally;
|
|
- the bindings (§8): the MSF record, the EXT-X-DATERANGE tag and the version 1 emsg box,
|
|
each rendered byte for byte and read back, and every binding case.
|
|
|
|
For every vector it requires the outcome, reason, index, gaps, digests, tally and
|
|
renderings the corpus states; and of the corpus, that every kind has an accepted event
|
|
and every reason a case.
|
|
|
|
python3 live_events_verify.py <live-conformance/events>
|
|
|
|
Exit status 0 only if every case verifies. Python 3.8 or later; standard library only.
|
|
"""
|
|
import base64
|
|
import datetime
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import struct
|
|
import sys
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance"))
|
|
from _blake3 import blake3 # noqa: E402
|
|
import _ed25519 as ED # noqa: E402
|
|
|
|
MAGIC = b"WEV1"
|
|
POLICY_MAGIC = b"WEP1"
|
|
SIGN_DOMAIN = b"wai:live-event\x01"
|
|
EVENT_TYPE = "urn:wai:event:1"
|
|
HLS_CLASS = "science.transaction.wai.event"
|
|
HLS_TAG = "#EXT-X-DATERANGE:"
|
|
MAX_SAFE = (1 << 53) - 1
|
|
|
|
# §4: the kinds and their names.
|
|
KINDS = {
|
|
0x01: "meta", 0x02: "cue", 0x03: "chat", 0x04: "reaction", 0x05: "poll_open",
|
|
0x06: "vote", 0x07: "poll_close", 0x08: "marker", 0x09: "control", 0x0A: "caption",
|
|
0x0B: "decision", 0x0C: "take_commit", 0x10: "join", 0x11: "leave", 0x12: "publish",
|
|
0x13: "unpublish", 0x14: "grant_exchange", 0x15: "takeover", 0x16: "failover",
|
|
}
|
|
LIFECYCLE = range(0x10, 0x17)
|
|
|
|
# §9, in its order.
|
|
REASONS = [
|
|
"truncated", "malformed", "unknown_version", "unknown_kind", "wrong_room", "bad_key",
|
|
"bad_signature", "kind_not_permitted", "payload_too_large", "malformed_payload",
|
|
"bad_utf8", "text_too_long", "seq_replay", "seq_not_increasing", "chain_broken",
|
|
"time_reversed", "delivery_gap", "rate_exceeded", "tally_mismatch", "policy_malformed",
|
|
"binding_malformed", "binding_mismatch",
|
|
]
|
|
|
|
|
|
class Refuse(Exception):
|
|
def __init__(self, reason):
|
|
super().__init__(reason)
|
|
self.reason = reason
|
|
|
|
|
|
class Reader:
|
|
"""Reads fields in order; running out raises the reason the caller names."""
|
|
|
|
def __init__(self, b, short):
|
|
self.b, self.p, self.short = b, 0, short
|
|
|
|
def take(self, n):
|
|
if self.p + n > len(self.b):
|
|
raise Refuse(self.short)
|
|
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 rest(self):
|
|
s = self.b[self.p:]
|
|
self.p = len(self.b)
|
|
return s
|
|
|
|
def done(self):
|
|
return self.p == len(self.b)
|
|
|
|
|
|
# ---------------------------------------------------------------- §3 the record
|
|
|
|
def read_record(b):
|
|
"""§3.3."""
|
|
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")
|
|
r = Reader(b, "truncated")
|
|
r.take(5)
|
|
ev = {"room": r.take(32), "principal": r.take(32), "seq": r.u64(), "prev": r.take(32), "t": r.u64()}
|
|
flag = r.u8()
|
|
if flag == 0:
|
|
ev["anchor"] = None
|
|
elif flag == 1:
|
|
ev["anchor"] = (r.take(32), r.u64(), r.u64())
|
|
else:
|
|
raise Refuse("malformed")
|
|
ev["kind"] = r.u8()
|
|
n = r.u16()
|
|
ev["payload"] = r.take(n)
|
|
ev["body_len"] = r.p
|
|
ev["sig"] = r.take(64)
|
|
if not r.done():
|
|
raise Refuse("malformed")
|
|
ev["body"] = b[:ev["body_len"]]
|
|
ev["digest"] = hashlib.sha256(ev["body"]).digest()
|
|
return ev
|
|
|
|
|
|
# ---------------------------------------------------------------- §4 payloads
|
|
|
|
LANG = re.compile(r"[A-Za-z]{2,8}(-[A-Za-z0-9]{1,8})*")
|
|
|
|
|
|
def text(r, max_cp):
|
|
"""§4.1."""
|
|
n = r.u16()
|
|
raw = r.take(n)
|
|
try:
|
|
s = raw.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
raise Refuse("bad_utf8")
|
|
if len(s) > max_cp:
|
|
raise Refuse("text_too_long")
|
|
return s
|
|
|
|
|
|
def splice_info_section_ok(b):
|
|
"""§4.3, form 1."""
|
|
return len(b) >= 3 and b[0] == 0xFC and 3 + (((b[1] & 0x0F) << 8) | b[2]) == len(b)
|
|
|
|
|
|
def stated_origin(r):
|
|
"""§4.4, §4.5: the origin a chat or a reaction may state after its last field, as
|
|
(origin, model), or (None, None) when no byte follows."""
|
|
if r.done():
|
|
return None, None
|
|
o = r.u8()
|
|
if o == 1:
|
|
return 1, None
|
|
if o in (2, 3):
|
|
return o, r.take(32)
|
|
raise Refuse("malformed_payload")
|
|
|
|
|
|
def read_payload(kind, b, max_cp):
|
|
"""§4: the payload as a dict, or Refuse."""
|
|
m = "malformed_payload"
|
|
r = Reader(b, m)
|
|
p = {"kind": kind}
|
|
if kind in (0x01, 0x0C):
|
|
p["bytes"] = r.rest()
|
|
elif kind == 0x02:
|
|
p["type"] = r.u8()
|
|
if p["type"] not in (1, 2, 3, 4):
|
|
raise Refuse(m)
|
|
p["cue_id"], p["duration_ms"] = r.u32(), r.u32()
|
|
form = r.u8()
|
|
p["form"] = form
|
|
if form in (1, 2):
|
|
p["scte"] = r.take(r.u16())
|
|
if form == 1 and not splice_info_section_ok(p["scte"]):
|
|
raise Refuse(m)
|
|
if form == 2 and len(p["scte"]) == 0:
|
|
raise Refuse(m)
|
|
elif form != 0:
|
|
raise Refuse(m)
|
|
elif kind == 0x03:
|
|
p["text"] = text(r, max_cp)
|
|
p["origin"], p["model"] = stated_origin(r)
|
|
elif kind == 0x04:
|
|
p["type"] = r.u8()
|
|
if p["type"] not in (1, 2, 3, 4, 5):
|
|
raise Refuse(m)
|
|
p["value"] = r.u16()
|
|
if p["value"] == 0:
|
|
raise Refuse(m)
|
|
flags = r.u8()
|
|
if flags & ~0x03:
|
|
raise Refuse(m)
|
|
p["target"] = r.take(32) if flags & 1 else None
|
|
p["remove"] = bool(flags & 2)
|
|
p["origin"], p["model"] = stated_origin(r)
|
|
elif kind == 0x05:
|
|
p["question"] = text(r, max_cp)
|
|
n = r.u8()
|
|
if n < 2:
|
|
raise Refuse(m)
|
|
p["options"] = [text(r, max_cp) for _ in range(n)]
|
|
elif kind == 0x06:
|
|
p["poll"], p["option"] = r.take(32), r.u8()
|
|
elif kind == 0x07:
|
|
p["poll"] = r.take(32)
|
|
n = r.u8()
|
|
if n < 2:
|
|
raise Refuse(m)
|
|
p["counts"] = [r.u32() for _ in range(n)]
|
|
elif kind == 0x08:
|
|
p["taxonomy"] = r.u32()
|
|
p["polarity"] = r.u8()
|
|
if p["polarity"] > 2:
|
|
raise Refuse(m)
|
|
p["text"] = text(r, max_cp)
|
|
has = r.u8()
|
|
if has not in (0, 1):
|
|
raise Refuse(m)
|
|
p["attachment"] = r.take(32) if has else None
|
|
elif kind == 0x09:
|
|
p["target"] = r.take(32)
|
|
c = r.u8()
|
|
p["command"] = c
|
|
if c == 1:
|
|
p["count_in_ms"] = r.u32()
|
|
elif c == 2:
|
|
pass
|
|
elif c == 3:
|
|
p["q16"] = r.u32()
|
|
if p["q16"] == 0:
|
|
raise Refuse(m)
|
|
elif c == 4:
|
|
p["on"] = r.u8()
|
|
if p["on"] not in (0, 1):
|
|
raise Refuse(m)
|
|
elif c == 5:
|
|
p["layer"] = r.u8()
|
|
else:
|
|
raise Refuse(m)
|
|
elif kind == 0x0A:
|
|
p["duration_ms"] = r.u32()
|
|
p["origin"] = r.u8()
|
|
if p["origin"] == 1:
|
|
p["model"] = None
|
|
elif p["origin"] in (2, 3):
|
|
p["model"] = r.take(32)
|
|
else:
|
|
raise Refuse(m)
|
|
lb = r.take(r.u8())
|
|
try:
|
|
lang = lb.decode("ascii")
|
|
except UnicodeDecodeError:
|
|
raise Refuse(m)
|
|
if not (2 <= len(lang) <= 35 and LANG.fullmatch(lang)):
|
|
raise Refuse(m)
|
|
p["lang"] = lang
|
|
p["text"] = text(r, max_cp)
|
|
elif kind == 0x0B:
|
|
p["verdict"] = r.u8()
|
|
if p["verdict"] not in (1, 2, 3):
|
|
raise Refuse(m)
|
|
p["target"], p["reason"] = r.take(32), r.u16()
|
|
elif kind in LIFECYCLE:
|
|
p["subject"], p["detail"] = r.take(32), r.rest()
|
|
else:
|
|
raise Refuse("unknown_kind")
|
|
if not r.done():
|
|
raise Refuse(m)
|
|
return p
|
|
|
|
|
|
# ---------------------------------------------------------------- §5 the policy
|
|
|
|
def read_policy(b):
|
|
if len(b) < 4:
|
|
raise Refuse("truncated")
|
|
if b[:4] != POLICY_MAGIC:
|
|
raise Refuse("policy_malformed")
|
|
if len(b) < 5:
|
|
raise Refuse("truncated")
|
|
if b[4] != 1:
|
|
raise Refuse("unknown_version")
|
|
if len(b) < 65:
|
|
raise Refuse("truncated")
|
|
if len(b) > 65:
|
|
raise Refuse("policy_malformed")
|
|
kinds = b[5:37]
|
|
max_payload, max_cp, max_events, window, max_reactions, rwindow = struct.unpack("<HHIQIQ", b[37:65])
|
|
for k in range(256):
|
|
if kinds[k // 8] >> (k % 8) & 1 and k not in KINDS:
|
|
raise Refuse("policy_malformed")
|
|
if 0 in (max_payload, max_cp, max_events, window, max_reactions, rwindow):
|
|
raise Refuse("policy_malformed")
|
|
return {"kinds": kinds, "max_payload": max_payload, "max_text_cp": max_cp, "max_events": max_events,
|
|
"window_us": window, "max_reactions": max_reactions, "reaction_window_us": rwindow}
|
|
|
|
|
|
def permits(policy, k):
|
|
return policy["kinds"][k // 8] >> (k % 8) & 1 == 1
|
|
|
|
|
|
# ---------------------------------------------------------------- §6 verification
|
|
|
|
class Verifier:
|
|
def __init__(self, room, default_policy, own):
|
|
self.room, self.default, self.own = room, default_policy, own
|
|
self.chains = {} # principal -> {"held": {seq: digest}, "head": (seq, digest, t) | None, "times": [], "reactions": []}
|
|
self.accepted = [] # (index, ev, payload)
|
|
self.gaps = []
|
|
self.offered = 0
|
|
|
|
def offer(self, b):
|
|
index = self.offered
|
|
self.offered += 1
|
|
ev = read_record(b) # §6.1 step 1
|
|
if ev["kind"] not in KINDS: # step 2
|
|
raise Refuse("unknown_kind")
|
|
if ev["room"] != self.room: # step 3
|
|
raise Refuse("wrong_room")
|
|
bad = ED.verify(ev["principal"], SIGN_DOMAIN + ev["body"], ev["sig"]) # step 4
|
|
if bad is not None:
|
|
raise Refuse("bad_key" if bad == "bad_key" else "bad_signature")
|
|
policy = self.own.get(ev["principal"], self.default)
|
|
if not permits(policy, ev["kind"]): # step 5
|
|
raise Refuse("kind_not_permitted")
|
|
if len(ev["payload"]) > policy["max_payload"]: # step 6
|
|
raise Refuse("payload_too_large")
|
|
p = read_payload(ev["kind"], ev["payload"], policy["max_text_cp"]) # step 7
|
|
c = self.chains.setdefault(ev["principal"], {"held": {}, "head": None, "times": [], "reactions": []})
|
|
seq, d, t = ev["seq"], ev["digest"], ev["t"]
|
|
if c["held"].get(seq) == d: # §6.2 step 1
|
|
return "duplicate"
|
|
gap = None
|
|
if c["head"] is not None: # step 2
|
|
hseq, hd, ht = c["head"]
|
|
if seq <= hseq:
|
|
raise Refuse("seq_replay" if seq in c["held"] else "seq_not_increasing")
|
|
if seq == hseq + 1:
|
|
if ev["prev"] != hd:
|
|
raise Refuse("chain_broken")
|
|
else:
|
|
gap = (hseq + 1, seq - 1)
|
|
if t < ht:
|
|
raise Refuse("time_reversed")
|
|
elif seq == 0: # step 3
|
|
if ev["prev"] != bytes(32):
|
|
raise Refuse("chain_broken")
|
|
else:
|
|
gap = (0, seq - 1)
|
|
within = lambda times, w: 1 + sum(1 for f in times if t - f < w) # noqa: E731
|
|
if within(c["times"], policy["window_us"]) > policy["max_events"]: # §6.3
|
|
raise Refuse("rate_exceeded")
|
|
standalone = ev["kind"] == 0x04 and p["target"] is None
|
|
if standalone and within(c["reactions"], policy["reaction_window_us"]) > policy["max_reactions"]:
|
|
raise Refuse("rate_exceeded")
|
|
c["held"][seq] = d
|
|
c["head"] = (seq, d, t)
|
|
c["times"].append(t)
|
|
if standalone:
|
|
c["reactions"].append(t)
|
|
if gap:
|
|
self.gaps.append((ev["principal"], gap[0], gap[1]))
|
|
self.accepted.append((index, ev, p))
|
|
return "accepted"
|
|
|
|
|
|
def run(verifier, records):
|
|
"""§6.4: ("refuse", reason, index) | ("gap", gaps) | ("accept",)."""
|
|
for i, b in enumerate(records):
|
|
try:
|
|
verifier.offer(b)
|
|
except Refuse as e:
|
|
return ("refuse", e.reason, i)
|
|
if verifier.gaps:
|
|
return ("gap", list(verifier.gaps))
|
|
bad = stated_tally_mismatch(verifier.accepted)
|
|
if bad is not None:
|
|
return ("refuse", "tally_mismatch", bad)
|
|
return ("accept",)
|
|
|
|
|
|
# ---------------------------------------------------------------- §7 polls
|
|
|
|
def poll_state(accepted):
|
|
polls = {}
|
|
for _, ev, p in accepted:
|
|
if p["kind"] == 0x05:
|
|
polls[ev["digest"]] = {"author": ev["principal"], "seq": ev["seq"], "n": len(p["options"]), "t": ev["t"]}
|
|
closes = {}
|
|
for index, ev, p in accepted:
|
|
if p["kind"] == 0x07 and p["poll"] in polls and polls[p["poll"]]["author"] == ev["principal"]:
|
|
cur = closes.get(p["poll"])
|
|
if cur is None or ev["seq"] < cur["seq"]:
|
|
closes[p["poll"]] = {"seq": ev["seq"], "t": ev["t"], "index": index, "counts": p["counts"]}
|
|
votes = {}
|
|
for _, ev, p in sorted((a for a in accepted if a[2]["kind"] == 0x06), key=lambda a: (a[1]["principal"], a[1]["seq"])):
|
|
poll = polls.get(p["poll"])
|
|
if poll is None or p["option"] >= poll["n"] or ev["t"] < poll["t"]:
|
|
continue
|
|
close = closes.get(p["poll"])
|
|
if close is not None and ev["t"] >= close["t"]:
|
|
continue
|
|
votes[(p["poll"], ev["principal"])] = p["option"]
|
|
return polls, votes, closes
|
|
|
|
|
|
def counts_of(poll_digest, poll, votes):
|
|
counts = [0] * poll["n"]
|
|
for (d, _), option in votes.items():
|
|
if d == poll_digest:
|
|
counts[option] += 1
|
|
return counts
|
|
|
|
|
|
def node_id(author, seq):
|
|
return blake3(b"wai:world-node\x01" + author + struct.pack("<Q", seq))[:16]
|
|
|
|
|
|
def tally(accepted):
|
|
"""The feed0 state the polls replay to, and each poll's cells."""
|
|
polls, votes, closes = poll_state(accepted)
|
|
nodes = {}
|
|
out = []
|
|
for d in sorted(polls):
|
|
poll = polls[d]
|
|
state = 0 if d in closes else 1
|
|
counts = counts_of(d, poll, votes)
|
|
cells = [state] + counts
|
|
body = struct.pack("<H", len(cells)) + b"".join(b"\x01" + struct.pack("<q", c) for c in cells)
|
|
nodes[node_id(poll["author"], poll["seq"])] = body
|
|
out.append({"poll": d.hex(), "open": state == 1, "counts": counts})
|
|
canon = struct.pack("<I", len(nodes)) + b"".join(k + struct.pack("<I", len(v)) + v for k, v in sorted(nodes.items()))
|
|
return {"state_hash": blake3(b"wai:world-state\x01" + canon).hex(), "polls": out}
|
|
|
|
|
|
def stated_tally_mismatch(accepted):
|
|
polls, votes, closes = poll_state(accepted)
|
|
bad = [c["index"] for d, c in closes.items() if c["counts"] != counts_of(d, polls[d], votes)]
|
|
return min(bad) if bad else None
|
|
|
|
|
|
# ---------------------------------------------------------------- §8 bindings
|
|
|
|
def b64url(b):
|
|
return base64.urlsafe_b64encode(b).decode().rstrip("=")
|
|
|
|
|
|
def b64url_read(s):
|
|
"""§8.1: canonical base64url without padding, or None."""
|
|
if not re.fullmatch(r"[A-Za-z0-9_-]*", s) or len(s) % 4 == 1:
|
|
return None
|
|
try:
|
|
b = base64.urlsafe_b64decode(s + "=" * (-len(s) % 4))
|
|
except ValueError:
|
|
return None
|
|
return b if b64url(b) == s else None
|
|
|
|
|
|
def bound_record(b):
|
|
"""The record a binding carries: one §3.3 reads with a kind §4 defines, with a cue's or
|
|
caption's payload read for the fields a binding derives."""
|
|
try:
|
|
ev = read_record(b)
|
|
if ev["kind"] not in KINDS:
|
|
raise Refuse("unknown_kind")
|
|
ev["p"] = read_payload(ev["kind"], ev["payload"], 0xFFFF) if ev["kind"] in (0x02, 0x0A) else None
|
|
except Refuse:
|
|
raise Refuse("binding_malformed")
|
|
return ev
|
|
|
|
|
|
def wallclock_ms(ev, t0):
|
|
return (t0 + ev["t"]) // 1000
|
|
|
|
|
|
def iso(ms):
|
|
try:
|
|
d = datetime.datetime(1970, 1, 1) + datetime.timedelta(milliseconds=ms)
|
|
except OverflowError:
|
|
raise Refuse("binding_malformed")
|
|
return d.strftime("%Y-%m-%dT%H:%M:%S") + ".%03dZ" % (ms % 1000)
|
|
|
|
|
|
def seconds(ms):
|
|
return "%d.%03d" % (ms // 1000, ms % 1000)
|
|
|
|
|
|
def msf_render(b, t0):
|
|
ev = bound_record(b)
|
|
a = ev["anchor"]
|
|
if a is not None and a[1] <= MAX_SAFE and a[2] <= MAX_SAFE:
|
|
idx = '"l":[%d,%d]' % (a[1], a[2])
|
|
else:
|
|
ms = wallclock_ms(ev, t0)
|
|
if ms > MAX_SAFE:
|
|
raise Refuse("binding_malformed")
|
|
idx = '"t":%d' % ms
|
|
return '{%s,"data":{"wev1":"%s"}}' % (idx, b64url(b))
|
|
|
|
|
|
def is_int(v):
|
|
return isinstance(v, int) and not isinstance(v, bool)
|
|
|
|
|
|
def msf_read(s, t0):
|
|
try:
|
|
o = json.loads(s)
|
|
except ValueError:
|
|
raise Refuse("binding_malformed")
|
|
if not isinstance(o, dict) or not isinstance(o.get("data"), dict) or not isinstance(o["data"].get("wev1"), str):
|
|
raise Refuse("binding_malformed")
|
|
b = b64url_read(o["data"]["wev1"])
|
|
if b is None:
|
|
raise Refuse("binding_malformed")
|
|
ev = bound_record(b)
|
|
if sum(k in o for k in ("t", "l", "m")) != 1:
|
|
raise Refuse("binding_malformed")
|
|
want = json.loads(msf_render(b, t0))
|
|
if "l" in want:
|
|
got = o.get("l")
|
|
ok = isinstance(got, list) and len(got) == 2 and all(is_int(x) for x in got) and got == want["l"]
|
|
else:
|
|
ok = is_int(o.get("t")) and o["t"] == want["t"]
|
|
if not ok:
|
|
raise Refuse("binding_mismatch")
|
|
del ev
|
|
return b
|
|
|
|
|
|
def daterange_attrs(b, t0):
|
|
"""§8.3's table: (name, value as written), in order."""
|
|
ev = bound_record(b)
|
|
a = [("ID", '"%s"' % ev["digest"].hex()), ("CLASS", '"%s"' % HLS_CLASS),
|
|
("START-DATE", '"%s"' % iso(wallclock_ms(ev, t0)))]
|
|
p = ev["p"]
|
|
if p is not None and p["kind"] == 0x02:
|
|
if p["duration_ms"] > 0:
|
|
a.append(("PLANNED-DURATION" if p["type"] in (1, 4) else "DURATION", seconds(p["duration_ms"])))
|
|
if p["form"] == 1:
|
|
name = {1: "SCTE35-OUT", 4: "SCTE35-OUT", 2: "SCTE35-IN"}.get(p["type"], "SCTE35-CMD")
|
|
a.append((name, "0x" + p["scte"].hex().upper()))
|
|
elif p is not None and p["duration_ms"] > 0:
|
|
a.append(("DURATION", seconds(p["duration_ms"])))
|
|
a.append(("X-WAI-EVENT", '"%s"' % b64url(b)))
|
|
return a
|
|
|
|
|
|
DATERANGE_NAMES = ("ID", "CLASS", "START-DATE", "DURATION", "PLANNED-DURATION", "SCTE35-CMD", "SCTE35-OUT", "SCTE35-IN", "X-WAI-EVENT")
|
|
|
|
|
|
def daterange_render(b, t0):
|
|
return HLS_TAG + ",".join("%s=%s" % nv for nv in daterange_attrs(b, t0))
|
|
|
|
|
|
def attribute_list(s):
|
|
"""RFC 8216 §4.2, as §8.3 reads it; None when it does not read."""
|
|
out, i = {}, 0
|
|
while i < len(s):
|
|
m = re.match(r"[A-Z0-9-]+", s[i:])
|
|
if not m or i + m.end() >= len(s) or s[i + m.end()] != "=":
|
|
return None
|
|
name = m.group(0)
|
|
i += m.end() + 1
|
|
start = i
|
|
if s[i] == '"':
|
|
j = s.find('"', i + 1)
|
|
if j < 0:
|
|
return None
|
|
i = j + 1
|
|
else:
|
|
while i < len(s) and s[i] != ",":
|
|
i += 1
|
|
if i == start or name in out:
|
|
return None
|
|
out[name] = s[start:i]
|
|
if i < len(s):
|
|
if s[i] != "," or i + 1 == len(s):
|
|
return None
|
|
i += 1
|
|
return out
|
|
|
|
|
|
def daterange_read(line, t0):
|
|
if not line.startswith(HLS_TAG):
|
|
raise Refuse("binding_malformed")
|
|
got = attribute_list(line[len(HLS_TAG):])
|
|
if got is None or "X-WAI-EVENT" not in got:
|
|
raise Refuse("binding_malformed")
|
|
v = got["X-WAI-EVENT"]
|
|
if len(v) < 2 or v[0] != '"' or v[-1] != '"':
|
|
raise Refuse("binding_malformed")
|
|
b = b64url_read(v[1:-1])
|
|
if b is None:
|
|
raise Refuse("binding_malformed")
|
|
want = dict(daterange_attrs(b, t0))
|
|
for name in DATERANGE_NAMES:
|
|
if got.get(name) != want.get(name):
|
|
raise Refuse("binding_mismatch")
|
|
return b
|
|
|
|
|
|
def emsg_render(b, pt0):
|
|
ev = bound_record(b)
|
|
p = ev["p"]
|
|
if p is not None:
|
|
duration = p["duration_ms"] if p["duration_ms"] else 0xFFFFFFFF
|
|
else:
|
|
duration = 0
|
|
pt = pt0 + ev["t"] // 1000
|
|
if pt >= 1 << 64:
|
|
raise Refuse("binding_malformed")
|
|
body = struct.pack(">BBBBIQI", 1, 0, 0, 0, 1000, pt, duration) + ev["digest"][:4]
|
|
body += EVENT_TYPE.encode() + b"\x00" + KINDS[ev["kind"]].encode() + b"\x00" + b
|
|
return struct.pack(">I", 8 + len(body)) + b"emsg" + body
|
|
|
|
|
|
def emsg_read(box, pt0):
|
|
if len(box) < 32 or struct.unpack(">I", box[:4])[0] != len(box) or box[4:8] != b"emsg" or box[8] != 1:
|
|
raise Refuse("binding_malformed")
|
|
rest = box[32:]
|
|
z1 = rest.find(b"\x00")
|
|
z2 = rest.find(b"\x00", z1 + 1) if z1 >= 0 else -1
|
|
if z2 < 0:
|
|
raise Refuse("binding_malformed")
|
|
b = rest[z2 + 1:]
|
|
bound_record(b)
|
|
if emsg_render(b, pt0) != box:
|
|
raise Refuse("binding_mismatch")
|
|
return b
|
|
|
|
|
|
def read_binding(kind, value, t0, pt0):
|
|
if kind == "daterange":
|
|
return daterange_read(value, t0)
|
|
if kind == "msf":
|
|
return msf_read(value, t0)
|
|
if kind == "emsg":
|
|
try:
|
|
box = bytes.fromhex(value)
|
|
except ValueError:
|
|
raise Refuse("binding_malformed")
|
|
return emsg_read(box, pt0)
|
|
raise Refuse("binding_malformed")
|
|
|
|
|
|
# ---------------------------------------------------------------- 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)
|
|
t0, pt0 = doc["t0_unix_us"], doc["pt0_ms"]
|
|
policies = {}
|
|
for name, h in doc["policies"].items():
|
|
try:
|
|
policies[name] = read_policy(bytes.fromhex(h))
|
|
except Refuse as e:
|
|
fails.append("policy %s: %s" % (name, e.reason))
|
|
rooms = {n: bytes.fromhex(h) for n, h in doc["rooms"].items()}
|
|
kinds_accepted, reasons_seen = set(), set()
|
|
n_vec = 0
|
|
for v in doc["vectors"]:
|
|
n_vec += 1
|
|
name = v["name"]
|
|
own = {bytes.fromhex(k): policies[p] for k, p in v["policies"].items()}
|
|
ver = Verifier(rooms[v["room"]], policies[v["default_policy"]], own)
|
|
records = [bytes.fromhex(e) for e in v["events"]]
|
|
out = run(ver, records)
|
|
problems = []
|
|
if out[0] != v["outcome"]:
|
|
problems.append("outcome %s, the corpus states %s" % (out, v["outcome"]))
|
|
elif out[0] == "refuse":
|
|
if (out[1], out[2]) != (v["reason"], v["index"]):
|
|
problems.append("refused %s at %d, the corpus states %s at %s" % (out[1], out[2], v["reason"], v["index"]))
|
|
elif out[0] == "gap":
|
|
gaps = [{"principal": p.hex(), "first": a, "last": z} for p, a, z in out[1]]
|
|
if gaps != v["gaps"] or v.get("reason") != "delivery_gap":
|
|
problems.append("gaps %s differ" % gaps)
|
|
if out[0] in ("accept", "gap"):
|
|
if [hashlib.sha256(read_record(r)["body"]).hexdigest() for r in records] != v["digests"]:
|
|
problems.append("digests differ")
|
|
if out[0] == "accept":
|
|
if tally(ver.accepted) != v["tally"]:
|
|
problems.append("tally %s differs" % tally(ver.accepted))
|
|
if len(v["bindings"]) != len(records):
|
|
problems.append("not one binding entry per event")
|
|
for i, (r, bnd) in enumerate(zip(records, v["bindings"])):
|
|
rendered = {"msf": msf_render(r, t0), "daterange": daterange_render(r, t0), "emsg": emsg_render(r, pt0).hex()}
|
|
if rendered != bnd:
|
|
problems.append("event %d: a rendering differs: %s" % (i, [k for k in rendered if rendered[k] != bnd.get(k)]))
|
|
for k in ("msf", "daterange", "emsg"):
|
|
try:
|
|
if read_binding(k, bnd[k], t0, pt0) != r:
|
|
problems.append("event %d: %s reads back to other bytes" % (i, k))
|
|
except Refuse as e:
|
|
problems.append("event %d: %s refused %s" % (i, k, e.reason))
|
|
for _, ev, _ in ver.accepted:
|
|
kinds_accepted.add(ev["kind"])
|
|
if v.get("reason"):
|
|
reasons_seen.add(v["reason"])
|
|
print(("PASS" if not problems else "FAIL") + " vector %s: %s%s" % (name, out[0], "" if out[0] == "accept" else " " + str(out[1] if out[0] == "refuse" else "delivery_gap")))
|
|
fails += ["%s: %s" % (name, p) for p in problems]
|
|
for c in doc["policy_cases"]:
|
|
try:
|
|
read_policy(bytes.fromhex(c["policy"]))
|
|
got = ("accept", None)
|
|
except Refuse as e:
|
|
got = ("refuse", e.reason)
|
|
want = (c["outcome"], c.get("reason"))
|
|
if got != want:
|
|
fails.append("policy case %s: %s, the corpus states %s" % (c["name"], got, want))
|
|
if c.get("reason"):
|
|
reasons_seen.add(c["reason"])
|
|
print("%s policy cases" % ("PASS" if not any(f.startswith("policy case") for f in fails) else "FAIL"))
|
|
for c in doc["binding_cases"]:
|
|
try:
|
|
got = ("accept", read_binding(c["binding"], c["input"], t0, pt0).hex())
|
|
except Refuse as e:
|
|
got = ("refuse", e.reason)
|
|
want = (c["outcome"], c["record"] if c["outcome"] == "accept" else c.get("reason"))
|
|
if got != want:
|
|
fails.append("binding case %s: %s, the corpus states %s" % (c["name"], got, want))
|
|
if c.get("reason"):
|
|
reasons_seen.add(c["reason"])
|
|
print("%s binding cases" % ("PASS" if not any(f.startswith("binding case") for f in fails) else "FAIL"))
|
|
missing_kinds = sorted(KINDS[k] for k in set(KINDS) - kinds_accepted)
|
|
missing_reasons = [r for r in REASONS if r not in reasons_seen]
|
|
if missing_kinds:
|
|
fails.append("no accepted event of kind %s" % ", ".join(missing_kinds))
|
|
if missing_reasons:
|
|
fails.append("no case refused with %s" % ", ".join(missing_reasons))
|
|
print("%s every kind accepted and every reason exercised" % ("PASS" if not (missing_kinds or missing_reasons) else "FAIL"))
|
|
for f in fails:
|
|
print("FAIL " + f)
|
|
print(("PASS" if not fails else "FAIL") + " live-events corpus: %d vectors, every policy and binding case" % n_vec)
|
|
return 1 if fails else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) != 2:
|
|
print("usage: live_events_verify.py <live-conformance/events>", file=sys.stderr)
|
|
sys.exit(2)
|
|
sys.exit(main(sys.argv[1]))
|