2ccc58bd25
It verifies each room with live_rooms_verify.py, recomputes the controlled position, samples, percentiles and verdicts from §5 to §9, requires every mutated copy of the corpus to fail and catches each of its own checks dropped. Standard library only, run isolated. Co-Authored-By: Claude Opus 5.5 <[email protected]>
794 lines
35 KiB
Python
794 lines
35 KiB
Python
#!/usr/bin/env python3
|
||
"""Independent verifier of the co-watch corpus (extensions/co-watch.md).
|
||
|
||
Written from the extension's text. It shares no code with wai-rs, which generates the corpus,
|
||
and never runs it. It checks Ed25519 signatures with ../claim-conformance/_ed25519.py, WAI's
|
||
strict rule (SPEC §7 *Signatures*) written from RFC 8032 with the standard library. Each room
|
||
is verified by the live-rooms corpus's own second implementation, live_rooms_verify.py
|
||
(live-rooms §9), written from its text alone. Everything else is here:
|
||
|
||
- the co-watch policy (§3): its layout, rules, signature and chain, and the object-sequence
|
||
source digest (§3.5);
|
||
- the statements (§4): reading, who signs, the control attribute (§4.4);
|
||
- controls (§5): who may control, the lead and the horizon, the range, and the order of
|
||
application with the overridden ones;
|
||
- the controlled position (§6);
|
||
- presentation statements (§7);
|
||
- windows, due slots, samples, settling, percentiles, verdicts, the skew statement and
|
||
completeness (§8);
|
||
- the catch-up rule (§9).
|
||
|
||
For every room it requires the outcome, reason and place, and everything else the corpus
|
||
states (§11.3); for every catch-up case, its correction; of the corpus, that every reason of
|
||
§12 appears in it and that no other file is there. Then it mutates copies of the corpus, one
|
||
defect each, and requires every copy to fail; and it drops each of its own checks in turn and
|
||
requires the corpus to catch the weakened verifier.
|
||
|
||
python3 co_watch_verify.py <live-conformance/cowatch>
|
||
python3 co_watch_verify.py --room <room.json>
|
||
|
||
With `--room`, it verifies one room given in a vector's form (descriptor, grants, chains,
|
||
policies, sources) and prints what §11.3 states of it, as JSON: a room a deployment kept,
|
||
checked from the text alone.
|
||
|
||
Exit status 0 only if every case verifies, every mutated copy fails and every weakened
|
||
verifier is caught. Python 3.8 or later; standard library only.
|
||
"""
|
||
|
||
import copy
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import re
|
||
import struct
|
||
import sys
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
sys.path.insert(0, HERE)
|
||
sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance"))
|
||
import live_events_verify as LEV # noqa: E402
|
||
import live_rooms_verify as LRV # noqa: E402
|
||
import _ed25519 as ED # noqa: E402
|
||
|
||
Refuse = LEV.Refuse
|
||
|
||
# Ed25519 is a pure function of its inputs: each signature is verified once, and every later
|
||
# pass over the corpus (its mutated copies, the weakened verifiers) reads the answer back.
|
||
_verified = {}
|
||
_verify = ED.verify
|
||
|
||
|
||
def _memo_verify(pk, msg, sig, skip=()):
|
||
k = (bytes(pk), bytes(msg), bytes(sig), tuple(skip))
|
||
if k not in _verified:
|
||
_verified[k] = _verify(pk, msg, sig, skip)
|
||
return _verified[k]
|
||
|
||
|
||
ED.verify = _memo_verify
|
||
EXTENSION = os.path.join(HERE, "..", "extensions", "co-watch.md")
|
||
|
||
POLICY_MAGIC, CONTROL_MAGIC, PRESENTATION_MAGIC, SKEW_MAGIC = b"WXP1", b"WXC1", b"WXS1", b"WXK1"
|
||
POLICY_DOMAIN = b"wai:co-watch-policy\x01" # §3.2
|
||
PRESENTATION_DOMAIN = b"wai:co-watch-presentation\x01" # §7.1
|
||
SOURCE_DOMAIN = b"wai:co-watch-source\x01" # §3.5
|
||
ZERO = bytes(32)
|
||
NORMAL = 1000000
|
||
META, JOIN, LEAVE, EXCHANGE = 0x01, 0x10, 0x11, 0x14
|
||
DIRECT = 2 # live-rooms §3.3, bit 1
|
||
PLAY, PAUSE, SEEK, RATE = 1, 2, 3, 4
|
||
ACTIONS = {1: "play", 2: "pause", 3: "seek", 4: "rate"}
|
||
STATES = {1: "playing", 2: "paused", 3: "stalled", 4: "ended"}
|
||
VERDICTS = {1: "in_sync", 2: "out_of_sync", 3: "unknown"}
|
||
IN_SYNC, OUT_OF_SYNC, UNKNOWN = 1, 2, 3
|
||
U32 = (1 << 32) - 1
|
||
|
||
# Each check this verifier makes, by name; the self-test drops one at a time (§13).
|
||
DROP = set()
|
||
|
||
|
||
def on(name):
|
||
return name not in DROP
|
||
|
||
|
||
class Stop(Exception):
|
||
def __init__(self, reason, at):
|
||
super().__init__(reason)
|
||
self.reason, self.at = reason, at
|
||
|
||
|
||
def sha(*parts):
|
||
h = hashlib.sha256()
|
||
for p in parts:
|
||
h.update(p)
|
||
return h.digest()
|
||
|
||
|
||
def reasons_of_extension():
|
||
"""§12's table: the codes this extension adds, in order."""
|
||
with open(EXTENSION, encoding="utf-8") as f:
|
||
text = f.read()
|
||
sec = text[text.index("\n## 12. Reasons"):]
|
||
sec = sec[:sec.index("\n## ", 1)]
|
||
return re.findall(r"^\| `([a-z_0-9]+)` \|", sec, re.M)
|
||
|
||
|
||
REASONS = reasons_of_extension()
|
||
|
||
|
||
class Rd:
|
||
def __init__(self, b, p=0):
|
||
self.b, self.p = b, p
|
||
|
||
def take(self, n, why):
|
||
if self.p + n > len(self.b):
|
||
raise Refuse(why)
|
||
s = self.b[self.p:self.p + n]
|
||
self.p += n
|
||
return s
|
||
|
||
def u8(self, why):
|
||
return self.take(1, why)[0]
|
||
|
||
def u16(self, why):
|
||
return struct.unpack("<H", self.take(2, why))[0]
|
||
|
||
def u32(self, why):
|
||
return struct.unpack("<I", self.take(4, why))[0]
|
||
|
||
def u64(self, why):
|
||
return struct.unpack("<Q", self.take(8, why))[0]
|
||
|
||
|
||
# ---------------------------------------------------------------- §3 the policy
|
||
|
||
def read_policy(b):
|
||
"""§3.3: steps 1 to 3, then the rules of §3.1's table."""
|
||
if len(b) < 4:
|
||
raise Refuse("truncated")
|
||
if b[:4] != POLICY_MAGIC:
|
||
raise Refuse("malformed")
|
||
if len(b) < 5:
|
||
raise Refuse("truncated")
|
||
if b[4] != 1:
|
||
raise Refuse("unknown_version")
|
||
r, t = Rd(b, 5), "truncated"
|
||
p = {"room": r.take(32, t), "prev": r.take(32, t), "effective": r.u64(t), "source_type": r.u8(t),
|
||
"source": r.take(32, t), "duration_us": r.u64(t)}
|
||
for k in ("lead_ms", "horizon_ms", "min_rate", "max_rate", "period_ms", "window_ms", "settle_ms", "bound_us"):
|
||
p[k] = r.u32(t)
|
||
p["quantile"] = r.u8(t)
|
||
for k in ("tolerance_us", "nudge", "seek_us"):
|
||
p[k] = r.u32(t)
|
||
p["body"] = b[:r.p]
|
||
p["sig"] = r.take(64, t)
|
||
if r.p != len(b):
|
||
raise Refuse("malformed")
|
||
if on("policy_rules"):
|
||
ok = (p["source_type"] in (1, 2, 3) and p["duration_us"] > 0 and p["lead_ms"] > 0 and p["period_ms"] > 0
|
||
and p["bound_us"] > 0 and p["nudge"] > 0 and p["horizon_ms"] >= p["lead_ms"]
|
||
and 0 < p["min_rate"] <= NORMAL <= p["max_rate"]
|
||
and p["window_ms"] % p["period_ms"] == 0 and 1 <= p["window_ms"] // p["period_ms"] <= 1000
|
||
and 1 <= p["quantile"] <= 100 and p["tolerance_us"] <= p["bound_us"] and p["seek_us"] > p["bound_us"])
|
||
if not ok:
|
||
raise Refuse("cw_policy_malformed")
|
||
p["digest"] = sha(p["body"])
|
||
return p
|
||
|
||
|
||
def source_digest(objects):
|
||
"""§3.5: an object sequence's digest."""
|
||
out = SOURCE_DOMAIN + struct.pack("<I", len(objects))
|
||
for o in objects:
|
||
out += bytes.fromhex(o["content_hash"]) + struct.pack("<I", o["duration_ms"])
|
||
return sha(out)
|
||
|
||
|
||
def check_policies(d, blobs, sources):
|
||
"""§3.3, §3.4, then §3.5 for the sources given: the policies, or Stop at ("policy", i)."""
|
||
held = []
|
||
for i, b in enumerate(blobs):
|
||
try:
|
||
p = read_policy(b)
|
||
if p["room"] != d["room"]:
|
||
raise Refuse("wrong_room")
|
||
bad = ED.verify(d["authority"], POLICY_DOMAIN + p["body"], p["sig"])
|
||
if bad:
|
||
raise Refuse(bad)
|
||
if on("policy_link"):
|
||
if not held:
|
||
linked = p["prev"] == ZERO and p["effective"] == 0
|
||
else:
|
||
q = held[-1]
|
||
same = (p["source_type"], p["source"]) == (q["source_type"], q["source"])
|
||
linked = p["prev"] == q["digest"] and p["effective"] > q["effective"] and (not same or p["duration_us"] == q["duration_us"])
|
||
if not linked:
|
||
raise Refuse("cw_policy_unlinked")
|
||
objs = sources[i] if i < len(sources) else None
|
||
if objs is not None and on("sources"):
|
||
total = sum(o["duration_ms"] for o in objs) * 1000
|
||
if p["source_type"] != 2 or source_digest(objs) != p["source"] or total != p["duration_us"]:
|
||
raise Refuse("cw_source_mismatch")
|
||
except Refuse as e:
|
||
raise Stop(e.reason, ["policy", i])
|
||
held.append(p)
|
||
return held
|
||
|
||
|
||
def policy_at(ps, t):
|
||
"""§3.4: the index of the last policy whose effective is at or below t, or None."""
|
||
k = None
|
||
for i, p in enumerate(ps):
|
||
if p["effective"] <= t:
|
||
k = i
|
||
return k
|
||
|
||
|
||
def starts_of(ps):
|
||
"""§3.4, §6.4: (instant, policy index) of each source's start."""
|
||
out = []
|
||
for i, p in enumerate(ps):
|
||
if i == 0 or (p["source_type"], p["source"]) != (ps[i - 1]["source_type"], ps[i - 1]["source"]):
|
||
out.append((p["effective"], i))
|
||
return out
|
||
|
||
|
||
# ---------------------------------------------------------------- §4 statements
|
||
|
||
def control_bits(g):
|
||
"""§4.4: the actions `watch_control` allows, None without it; Refuse when malformed."""
|
||
for k, v in LRV.read_attrs(g["attrs"]) or []:
|
||
if k == "watch_control":
|
||
if not re.fullmatch(r"[1-9][0-9]?", v) or int(v) > 15:
|
||
raise Refuse("cw_attr_malformed")
|
||
return int(v)
|
||
return None
|
||
|
||
|
||
def may_control(g, action):
|
||
"""§5.2."""
|
||
if g["powers"] & DIRECT:
|
||
return True
|
||
try:
|
||
bits = control_bits(g)
|
||
except Refuse:
|
||
return False
|
||
return bits is not None and bits >> (action - 1) & 1 == 1
|
||
|
||
|
||
def read_statement(payload):
|
||
"""§4.1, §4.2: None for a payload not this extension's; else (kind, fields)."""
|
||
if len(payload) < 4 or payload[:4] not in (CONTROL_MAGIC, PRESENTATION_MAGIC, SKEW_MAGIC):
|
||
return None
|
||
bad = "cw_statement_malformed"
|
||
if len(payload) < 5 or payload[4] != 1:
|
||
raise Refuse(bad)
|
||
r = Rd(payload, 5)
|
||
magic = payload[:4]
|
||
if magic == CONTROL_MAGIC: # §5.1
|
||
s = ("control", {"action": r.u8(bad), "at": r.u64(bad), "arg": r.u64(bad)})
|
||
a, arg = s[1]["action"], s[1]["arg"]
|
||
if a in (PLAY, PAUSE):
|
||
ok = arg == 0
|
||
elif a == SEEK:
|
||
ok = True
|
||
elif a == RATE:
|
||
ok = 1 <= arg <= U32
|
||
else:
|
||
ok = False
|
||
if not ok:
|
||
raise Refuse(bad)
|
||
elif magic == PRESENTATION_MAGIC: # §7.1
|
||
f = {"room": r.take(32, bad), "player": r.take(32, bad), "source": r.take(32, bad), "at": r.u64(bad),
|
||
"position": r.u64(bad), "state": r.u8(bad), "rate": r.u32(bad)}
|
||
f["body"] = payload[:r.p]
|
||
f["sig"] = r.take(64, bad)
|
||
if f["state"] not in STATES or f["rate"] == 0:
|
||
raise Refuse(bad)
|
||
s = ("presentation", f)
|
||
else: # §8.4
|
||
f = {"from": r.u64(bad), "to": r.u64(bad), "bound_us": r.u32(bad), "quantile": r.u8(bad)}
|
||
nv = r.u8(bad)
|
||
if not 1 <= nv <= 8:
|
||
raise Refuse(bad)
|
||
f["viewers"] = []
|
||
for _ in range(nv):
|
||
e = {"principal": r.take(32, bad), "held": r.u64(bad), "due": r.u16(bad), "samples": r.u16(bad),
|
||
"counted": r.u16(bad), "p50": r.u32(bad), "pq": r.u32(bad), "max": r.u32(bad), "verdict": r.u8(bad)}
|
||
if e["verdict"] not in VERDICTS:
|
||
raise Refuse(bad)
|
||
f["viewers"].append(e)
|
||
keys = [e["principal"] for e in f["viewers"]]
|
||
if any(keys[i] >= keys[i + 1] for i in range(len(keys) - 1)):
|
||
raise Refuse(bad)
|
||
want = [(a, b) for a in range(nv) for b in range(a + 1, nv)]
|
||
if r.u8(bad) != len(want):
|
||
raise Refuse(bad)
|
||
f["pairs"] = []
|
||
for a, b in want:
|
||
e = {"a": r.u8(bad), "b": r.u8(bad), "n": r.u16(bad), "p50": r.u32(bad), "pq": r.u32(bad),
|
||
"max": r.u32(bad), "verdict": r.u8(bad)}
|
||
if (e["a"], e["b"]) != (a, b) or e["verdict"] not in VERDICTS:
|
||
raise Refuse(bad)
|
||
f["pairs"].append(e)
|
||
s = ("skew", f)
|
||
if r.p != len(payload):
|
||
raise Refuse(bad)
|
||
return s
|
||
|
||
|
||
# ---------------------------------------------------------------- §5.4, §6 the controlled position
|
||
|
||
def position(st, x, duration):
|
||
"""§6.2."""
|
||
if not st["playing"]:
|
||
return st["pos"]
|
||
return min(duration, st["pos"] + (x - st["anchor"]) * st["rate"] // 1000000)
|
||
|
||
|
||
class Timeline:
|
||
"""§5.4 and §6 over the accepted controls (each with its event's t and digest)."""
|
||
|
||
def __init__(self, policies, controls):
|
||
self.policies = policies
|
||
self.starts = starts_of(policies)
|
||
self.controls = sorted(controls, key=lambda c: (c["at"], c["t"], c["digest"]))
|
||
self.overridden = set()
|
||
aspect = {PLAY: 0, PAUSE: 0, SEEK: 1, RATE: 2}
|
||
for i, c in enumerate(self.controls):
|
||
for d in self.controls[i + 1:]:
|
||
if d["at"] != c["at"]:
|
||
break
|
||
if aspect[d["action"]] == aspect[c["action"]]:
|
||
self.overridden.add(c["digest"])
|
||
events = [(s, 0, k, None) for k, (s, _) in enumerate(self.starts)]
|
||
events += [(c["at"], 1, k, c) for k, c in enumerate(self.controls)]
|
||
events.sort(key=lambda e: (e[0], e[1], e[2]))
|
||
self.points, st = [], None
|
||
for x, what, _, c in events:
|
||
if what == 0:
|
||
st = {"playing": False, "rate": NORMAL, "anchor": x, "pos": 0}
|
||
else:
|
||
p = position(st, x, self.duration_at(x))
|
||
st = dict(st, anchor=x, pos=p)
|
||
if c["action"] == PLAY:
|
||
st["playing"] = True
|
||
elif c["action"] == PAUSE:
|
||
st["playing"] = False
|
||
elif c["action"] == SEEK:
|
||
st["pos"] = c["arg"]
|
||
else:
|
||
st["rate"] = c["arg"]
|
||
self.points.append((x, st))
|
||
|
||
def duration_at(self, x):
|
||
k = policy_at(self.policies, x)
|
||
return 0 if k is None else self.policies[k]["duration_us"]
|
||
|
||
def state_at(self, x):
|
||
st = None
|
||
for at, s in self.points:
|
||
if at <= x:
|
||
st = s
|
||
return st
|
||
|
||
def position_at(self, x):
|
||
st = self.state_at(x)
|
||
return 0 if st is None else position(st, x, self.duration_at(x))
|
||
|
||
def settle_points(self):
|
||
return sorted({c["at"] for c in self.controls} | {s for s, _ in self.starts})
|
||
|
||
|
||
# ---------------------------------------------------------------- §9 catch-up
|
||
|
||
def catch_up(p, playing, rate, controlled, duration, presented):
|
||
"""§9."""
|
||
e = presented - controlled
|
||
if not playing or controlled >= duration:
|
||
return {"action": "seek", "value": controlled} if abs(e) > p["tolerance_us"] else {"action": "hold"}
|
||
if abs(e) <= p["tolerance_us"]:
|
||
return {"action": "rate", "value": rate}
|
||
if abs(e) > p["seek_us"]:
|
||
return {"action": "seek", "value": controlled}
|
||
r = rate - p["nudge"] if e > 0 else rate + p["nudge"]
|
||
return {"action": "rate", "value": max(p["min_rate"], min(p["max_rate"], r))}
|
||
|
||
|
||
# ---------------------------------------------------------------- §8 skew
|
||
|
||
def window_policy(ps, frm, to):
|
||
"""§8.1: the policy of the window frm–to, or None when it is not one."""
|
||
if frm >= to:
|
||
return None
|
||
k = policy_at(ps, frm)
|
||
if k is None or policy_at(ps, to - 1) != k:
|
||
return None
|
||
per = ps[k]["period_ms"] * 1000
|
||
n = (to - frm) // per
|
||
return ps[k] if (to - frm) % per == 0 and 1 <= n <= 1000 else None
|
||
|
||
|
||
def percentile(vals, q):
|
||
"""§8.5: nearest rank; 0 for none."""
|
||
if not vals:
|
||
return 0
|
||
v = sorted(vals)
|
||
rank = -(-q * len(v) // 100)
|
||
return v[max(rank, 1) - 1]
|
||
|
||
|
||
class Facts:
|
||
"""Presence (§8.2), grants, samples (§8.3) and the timeline."""
|
||
|
||
def __init__(self, timeline, stretches, grant_changes, samples):
|
||
self.tl, self.stretches, self.grants, self.samples = timeline, stretches, grant_changes, samples
|
||
|
||
def subscribes_at(self, w, x):
|
||
last = None
|
||
for g in self.grants:
|
||
if g["principal"] == w and g["t"] <= x:
|
||
last = g
|
||
return last is not None and last["subscribe"]
|
||
|
||
def due(self, p, frm, to, w):
|
||
per = p["period_ms"] * 1000
|
||
out = []
|
||
for j in range((to - frm) // per):
|
||
s = frm + j * per
|
||
inside = any(st["principal"] == w and st["join"] <= s and (st["leave"] is None or st["leave"] >= s + per) for st in self.stretches)
|
||
out.append(inside and self.subscribes_at(w, s))
|
||
return out
|
||
|
||
def viewers(self, frm, to):
|
||
p = window_policy(self.tl.policies, frm, to)
|
||
if p is None:
|
||
return []
|
||
who = sorted({st["principal"] for st in self.stretches})
|
||
return [w for w in who if any(self.due(p, frm, to, w))]
|
||
|
||
def settling(self, p, w, at):
|
||
span = p["settle_ms"] * 1000
|
||
if not on("settle"):
|
||
return False
|
||
near = [x for x in self.tl.settle_points()] + [st["join"] for st in self.stretches if st["principal"] == w]
|
||
return any(x <= at < x + span for x in near)
|
||
|
||
def row(self, p, frm, to, w, held):
|
||
due = self.due(p, frm, to, w)
|
||
per = p["period_ms"] * 1000
|
||
sample = [None] * len(due)
|
||
for s in self.samples:
|
||
if s["principal"] != w or s["seq"] >= held or not frm <= s["at"] < to:
|
||
continue
|
||
j = (s["at"] - frm) // per
|
||
if not due[j] or sample[j] is not None:
|
||
continue
|
||
e = s["position"] - self.tl.position_at(s["at"])
|
||
sample[j] = (e, not self.settling(p, w, s["at"]))
|
||
return due, sample
|
||
|
||
def entries(self, frm, to, listed):
|
||
p = window_policy(self.tl.policies, frm, to)
|
||
rows = [self.row(p, frm, to, w, held) for w, held in listed]
|
||
bound = p["bound_us"]
|
||
|
||
def judge(unknown, pq):
|
||
return UNKNOWN if unknown else (IN_SYNC if pq <= bound else OUT_OF_SYNC)
|
||
|
||
viewers = []
|
||
for (w, held), (due, sample) in zip(listed, rows):
|
||
nd = sum(1 for d in due if d)
|
||
ns = sum(1 for s in sample if s is not None)
|
||
errs = [abs(s[0]) for s in sample if s is not None and s[1]]
|
||
p50, pq = percentile(errs, 50), percentile(errs, p["quantile"])
|
||
viewers.append({"principal": w, "held": held, "due": nd, "samples": ns, "counted": len(errs),
|
||
"p50": min(p50, U32), "pq": min(pq, U32), "max": min(max(errs, default=0), U32),
|
||
"verdict": judge(ns < nd or not errs, pq)})
|
||
pairs = []
|
||
for a in range(len(listed)):
|
||
for b in range(a + 1, len(listed)):
|
||
d = [abs(x[0] - y[0]) for x, y in zip(rows[a][1], rows[b][1]) if x is not None and y is not None and x[1] and y[1]]
|
||
p50, pq = percentile(d, 50), percentile(d, p["quantile"])
|
||
unknown = viewers[a]["verdict"] == UNKNOWN or viewers[b]["verdict"] == UNKNOWN or not d
|
||
pairs.append({"a": a, "b": b, "n": len(d), "p50": min(p50, U32), "pq": min(pq, U32),
|
||
"max": min(max(d, default=0), U32), "verdict": judge(unknown, pq)})
|
||
return viewers, pairs
|
||
|
||
|
||
# ---------------------------------------------------------------- §11 verification
|
||
|
||
def viewer_json(v):
|
||
return {"principal": v["principal"].hex(), "held": v["held"], "due": v["due"], "samples": v["samples"], "counted": v["counted"],
|
||
"p50": v["p50"], "pq": v["pq"], "max": v["max"], "verdict": VERDICTS[v["verdict"]]}
|
||
|
||
|
||
def pair_json(p):
|
||
return {"a": p["a"], "b": p["b"], "n": p["n"], "p50": p["p50"], "pq": p["pq"], "max": p["max"], "verdict": VERDICTS[p["verdict"]]}
|
||
|
||
|
||
def verify(descriptor, grants, chains, policy_blobs, sources):
|
||
"""§11.2: a dict of what the corpus states (§11.3)."""
|
||
room = LRV.verify_room(descriptor, grants, None, chains) # step 1
|
||
if room["outcome"] == "refuse":
|
||
return {"outcome": "refuse", "reason": room["reason"], "at": ["room"] + room["at"]}
|
||
gap = room["outcome"] == "gap"
|
||
d = LRV.read_descriptor(descriptor)
|
||
held_grants, _ = LRV.check_grants(d, grants)
|
||
by_digest = {g["digest"]: g for g in held_grants}
|
||
room_keys = (d["authority"], d["clock"])
|
||
try:
|
||
policies = check_policies(d, policy_blobs, sources) # step 2
|
||
for i, g in enumerate(held_grants): # step 3
|
||
if on("attr"):
|
||
try:
|
||
control_bits(g)
|
||
except Refuse as e:
|
||
raise Stop(e.reason, ["grant", i])
|
||
records, seen = [], set()
|
||
for c, i, under in room["order"]:
|
||
ev = LEV.read_record(chains[c][i])
|
||
if ev["digest"] in seen:
|
||
continue
|
||
seen.add(ev["digest"])
|
||
ev["chain"], ev["index"], ev["under"] = c, i, bytes.fromhex(under) if under else None
|
||
records.append(ev)
|
||
stretches, grant_changes = [], [] # §8.2
|
||
for ev in records:
|
||
if ev["principal"] not in room_keys or ev["kind"] not in (JOIN, LEAVE, EXCHANGE):
|
||
continue
|
||
pl = LEV.read_payload(ev["kind"], ev["payload"], 1 << 20)
|
||
x = LRV.read_detail(ev["kind"], pl["detail"])
|
||
who = pl["subject"]
|
||
if ev["kind"] == JOIN:
|
||
stretches.append({"principal": who, "join": ev["t"], "leave": None})
|
||
grant_changes.append({"principal": who, "t": ev["t"], "subscribe": by_digest[x["grant"]]["subscribe"] == 1})
|
||
elif ev["kind"] == LEAVE:
|
||
for st in reversed(stretches):
|
||
if st["principal"] == who and st["leave"] is None:
|
||
st["leave"] = ev["t"]
|
||
break
|
||
else:
|
||
grant_changes.append({"principal": who, "t": ev["t"], "subscribe": by_digest[x["new"]]["subscribe"] == 1})
|
||
controls, presentations, pending = [], [], []
|
||
for ev in records: # step 4
|
||
if ev["kind"] != META:
|
||
continue
|
||
at = ["record", ev["chain"], ev["index"]]
|
||
try:
|
||
s = read_statement(ev["payload"])
|
||
if s is None:
|
||
continue
|
||
kind, f = s
|
||
t = ev["t"]
|
||
k = policy_at(policies, t)
|
||
if k is None:
|
||
raise Refuse("cw_no_policy")
|
||
p = policies[k]
|
||
is_key = ev["principal"] in room_keys
|
||
if kind == "control":
|
||
if on("control_power") and not is_key and not may_control(by_digest[ev["under"]], f["action"]):
|
||
raise Refuse("control_not_permitted")
|
||
if on("lead") and f["at"] < t + p["lead_ms"] * 1000:
|
||
raise Refuse("control_not_ahead")
|
||
if on("horizon") and f["at"] > t + p["horizon_ms"] * 1000:
|
||
raise Refuse("control_beyond_horizon")
|
||
q = policies[policy_at(policies, f["at"])]
|
||
if on("range") and ((f["action"] == SEEK and f["arg"] > q["duration_us"])
|
||
or (f["action"] == RATE and not q["min_rate"] <= f["arg"] <= q["max_rate"])):
|
||
raise Refuse("control_out_of_range")
|
||
controls.append(dict(f, t=t, digest=ev["digest"], author=ev["principal"]))
|
||
elif kind == "presentation":
|
||
if on("who_presents") and is_key:
|
||
raise Refuse("cw_not_permitted")
|
||
if f["room"] != d["room"]:
|
||
raise Refuse("wrong_room")
|
||
if on("wrong_key") and f["player"] != ev["principal"]:
|
||
raise Refuse("presentation_wrong_key")
|
||
bad = ED.verify(f["player"], PRESENTATION_DOMAIN + f["body"], f["sig"]) if on("pres_sig") else None
|
||
if bad:
|
||
raise Refuse(bad)
|
||
if on("ahead") and f["at"] > t:
|
||
raise Refuse("presentation_ahead")
|
||
if on("source") and f["source"] != policies[policy_at(policies, f["at"])]["source"]:
|
||
raise Refuse("presentation_wrong_source")
|
||
presentations.append(dict(f, digest=ev["digest"], seq=ev["seq"]))
|
||
else:
|
||
if on("who_skews") and not is_key:
|
||
raise Refuse("cw_not_permitted")
|
||
wp = window_policy(policies, f["from"], f["to"])
|
||
if wp is None or f["to"] > t:
|
||
raise Refuse("cw_statement_malformed")
|
||
if on("skew_bound") and (f["bound_us"], f["quantile"]) != (wp["bound_us"], wp["quantile"]):
|
||
raise Refuse("skew_mismatch")
|
||
pending.append((at, ev["digest"], f))
|
||
except Refuse as e:
|
||
raise Stop(e.reason, at)
|
||
tl = Timeline(policies, controls)
|
||
facts = Facts(tl, stretches, grant_changes,
|
||
[{"principal": p["player"], "seq": p["seq"], "at": p["at"], "position": p["position"]} for p in presentations])
|
||
skews = []
|
||
counts = {}
|
||
for ev in records:
|
||
counts[ev["principal"]] = counts.get(ev["principal"], 0) + 1
|
||
for at, digest, f in pending: # §8.4 steps 2 and 3
|
||
if gap:
|
||
skews.append({"digest": digest.hex(), "from": f["from"], "to": f["to"], "viewers": "unchecked", "pairs": "unchecked"})
|
||
continue
|
||
vs = set(facts.viewers(f["from"], f["to"]))
|
||
if on("skew_viewers") and any(v["principal"] not in vs or v["held"] > counts.get(v["principal"], 0) for v in f["viewers"]):
|
||
raise Stop("skew_mismatch", at)
|
||
viewers, pairs = facts.entries(f["from"], f["to"], [(v["principal"], v["held"]) for v in f["viewers"]])
|
||
if on("skew_recompute") and (viewers != f["viewers"] or pairs != f["pairs"]):
|
||
raise Stop("skew_mismatch", at)
|
||
skews.append({"digest": digest.hex(), "from": f["from"], "to": f["to"],
|
||
"viewers": [viewer_json(v) for v in viewers], "pairs": [pair_json(p) for p in pairs]})
|
||
if not gap and on("complete"): # step 5, §8.6
|
||
windows = {}
|
||
for at, _, f in pending:
|
||
w = windows.setdefault((f["from"], f["to"]), [at, set()])
|
||
w[1].update(v["principal"] for v in f["viewers"])
|
||
for (frm, to), (at, listed) in windows.items():
|
||
if any(w not in listed for w in facts.viewers(frm, to)):
|
||
raise Stop("skew_incomplete", at)
|
||
except Stop as s:
|
||
return {"outcome": "refuse", "reason": s.reason, "at": s.at}
|
||
out = {"outcome": "gap" if gap else "accept", "room": d["room"].hex(), "gaps": room.get("gaps", []),
|
||
"policy_digests": [p["digest"].hex() for p in policies],
|
||
"starts": [{"at": s, "policy": i} for s, i in tl.starts],
|
||
"controls": [{"digest": c["digest"].hex(), "author": c["author"].hex(), "t": c["t"], "action": ACTIONS[c["action"]],
|
||
"at": c["at"], "arg": c["arg"], "applied": c["digest"] not in tl.overridden} for c in controls],
|
||
"timeline": [{"at": x, "playing": s["playing"], "rate": s["rate"], "position": position(s, x, tl.duration_at(x))} for x, s in tl.points],
|
||
"presentations": [], "skews": skews}
|
||
for p in presentations:
|
||
controlled = tl.position_at(p["at"])
|
||
pol = policies[policy_at(policies, p["at"])]
|
||
out["presentations"].append({"digest": p["digest"].hex(), "player": p["player"].hex(), "at": p["at"], "position": p["position"],
|
||
"state": STATES[p["state"]], "controlled": controlled, "error": p["position"] - controlled,
|
||
"settling": facts.settling(pol, p["player"], p["at"])})
|
||
return out
|
||
|
||
|
||
def run_vector(v):
|
||
sources = []
|
||
for s in v["sources"]:
|
||
sources.append(None if s is None else s)
|
||
return verify(bytes.fromhex(v["descriptor"]), [bytes.fromhex(g) for g in v["grants"]],
|
||
[[bytes.fromhex(r) for r in c] for c in v["chains"]], [bytes.fromhex(p) for p in v["policies"]], sources)
|
||
|
||
|
||
STATED = ("outcome", "reason", "at", "room", "gaps", "policy_digests", "starts", "controls", "timeline", "presentations", "skews")
|
||
|
||
|
||
def check(doc, quiet=False, first=False):
|
||
"""Every vector and catch-up case of `doc` against this verifier: the failures (with
|
||
`first`, the first alone)."""
|
||
fails, seen = [], set()
|
||
for v in doc["vectors"]:
|
||
if first and fails:
|
||
return fails, seen
|
||
try:
|
||
got = run_vector(v)
|
||
except Exception as e: # a copy that does not read is a failure to verify it
|
||
fails.append("%s: does not verify (%r)" % (v["name"], e))
|
||
continue
|
||
bad = [k for k in STATED if (k in got or k in v) and got.get(k) != v.get(k)]
|
||
if got.get("reason"):
|
||
seen.add(got["reason"])
|
||
if not quiet:
|
||
what = got["outcome"] + ("" if "reason" not in got else " " + got["reason"] + " at %s" % got["at"])
|
||
print(("PASS" if not bad else "FAIL") + " room %s: %s" % (v["name"], what))
|
||
for k in bad:
|
||
fails.append("%s: %s is %s, the corpus states %s" % (v["name"], k, json.dumps(got.get(k))[:200], json.dumps(v.get(k))[:200]))
|
||
cu = doc["catch_up"]
|
||
for i, c in enumerate(cu["cases"]):
|
||
got = catch_up(cu["params"], c["playing"], c["rate"], c["controlled"], c["duration"], c["presented"])
|
||
ok = got == c["correction"]
|
||
if not quiet:
|
||
print(("PASS" if ok else "FAIL") + " catch-up %d: %s -> %s" % (i, c["about"], json.dumps(got)))
|
||
if not ok:
|
||
fails.append("catch-up case %d: %s, the corpus states %s" % (i, got, c["correction"]))
|
||
return fails, seen
|
||
|
||
|
||
def mutations(doc):
|
||
"""Copies of the corpus, each with one defect a conforming verifier must catch."""
|
||
vn = {v["name"]: k for k, v in enumerate(doc["vectors"])}
|
||
party, drift, nxt, gapv = vn["watch_party"], vn["drift"], vn["next_episode"], vn["gap"]
|
||
|
||
def mutate(label, f):
|
||
d = copy.deepcopy(doc)
|
||
f(d)
|
||
return label, d
|
||
|
||
def vs(d):
|
||
return d["vectors"]
|
||
|
||
def flip(h, i):
|
||
b = bytearray(bytes.fromhex(h))
|
||
b[i] ^= 1
|
||
return b.hex()
|
||
|
||
def drop_chain(d, k, principal):
|
||
chains = vs(d)[k]["chains"]
|
||
vs(d)[k]["chains"] = [c for c in chains if LEV.read_record(bytes.fromhex(c[0]))["principal"].hex() != principal]
|
||
|
||
late = doc["principals"]["late"]
|
||
return [
|
||
mutate("a stated 95th percentile", lambda d: vs(d)[party]["skews"][2]["viewers"][0].update(pq=vs(d)[party]["skews"][2]["viewers"][0]["pq"] + 1)),
|
||
mutate("a stated verdict", lambda d: vs(d)[drift]["skews"][0]["viewers"][2].update(verdict="in_sync")),
|
||
mutate("a stated pair", lambda d: vs(d)[drift]["skews"][2]["pairs"][0].update(n=0)),
|
||
mutate("a control stated applied", lambda d: vs(d)[party]["controls"][6].update(applied=True)),
|
||
mutate("a stated error", lambda d: vs(d)[party]["presentations"][9].update(error=vs(d)[party]["presentations"][9]["error"] + 1)),
|
||
mutate("a statement stated settling", lambda d: vs(d)[party]["presentations"][7].update(settling=not vs(d)[party]["presentations"][7]["settling"])),
|
||
mutate("a stated position of the timeline", lambda d: vs(d)[party]["timeline"][3].update(position=1)),
|
||
mutate("the second source's start unstated", lambda d: vs(d)[nxt]["starts"].pop()),
|
||
mutate("a byte of a presentation statement's own signature", lambda d: vs(d)[party]["chains"][2].__setitem__(4, flip(vs(d)[party]["chains"][2][4], -70))),
|
||
mutate("the late joiner's chain withheld", lambda d: drop_chain(d, party, late)),
|
||
mutate("an object of the second source one millisecond longer", lambda d: vs(d)[nxt]["sources"][1][0].update(duration_ms=vs(d)[nxt]["sources"][1][0]["duration_ms"] + 1)),
|
||
mutate("the gap's skew statements stated checked", lambda d: vs(d)[gapv]["skews"][0].update(viewers=[])),
|
||
mutate("a refusal's reason", lambda d: vs(d)[vn["control_short_lead"]].update(reason="control_beyond_horizon")),
|
||
mutate("a refusal's place", lambda d: vs(d)[vn["skew_leaves_a_viewer_out"]].update(at=["record", 0, 4])),
|
||
mutate("a catch-up correction", lambda d: d["catch_up"]["cases"][3]["correction"].update(value=1040000)),
|
||
mutate("the catch-up nudge", lambda d: d["catch_up"]["params"].update(nudge=40000)),
|
||
]
|
||
|
||
|
||
CHECKS = ("policy_rules", "policy_link", "sources", "attr", "control_power", "lead", "horizon", "range", "who_presents", "wrong_key",
|
||
"pres_sig", "ahead", "source", "who_skews", "skew_bound", "skew_viewers", "skew_recompute", "complete", "settle")
|
||
|
||
|
||
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 []
|
||
ok = len(REASONS) == 16
|
||
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 refuses a vector")
|
||
fails += ["no vector refused with %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("co_watch_verify: %s" % ("PASS" if not fails else "FAIL (%d)" % len(fails)))
|
||
for f in fails:
|
||
print(" " + f)
|
||
return 0 if not fails else 1
|
||
|
||
|
||
def room_main(path):
|
||
with open(path, encoding="utf-8") as f:
|
||
v = json.load(f)
|
||
v.setdefault("sources", [])
|
||
print(json.dumps(run_vector(v)))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
if len(sys.argv) == 3 and sys.argv[1] == "--room":
|
||
sys.exit(room_main(sys.argv[2]))
|
||
if len(sys.argv) != 2:
|
||
print("usage: co_watch_verify.py <live-conformance/cowatch> | --room <room.json>", file=sys.stderr)
|
||
sys.exit(2)
|
||
sys.exit(main(sys.argv[1]))
|