886 lines
35 KiB
Python
886 lines
35 KiB
Python
#!/usr/bin/env python3
|
|
"""Independent verifier of the live-composition corpus (extensions/live-compose.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), whose lifecycle replay this
|
|
verifier steps through itself, record by record, to take the room at each instant (§4.1).
|
|
Everything else is here, from the text:
|
|
|
|
- the descriptor (§3): read, its canvas, its field rules, its signature and who may sign;
|
|
- the participant set (§4): sources, attributes, order, and the set's digest;
|
|
- the layout function (§5): the grid and its candidates, the featured tile, picture in
|
|
picture, the fill modes; and the mixer (§6), over the corpus's PCM;
|
|
- the receipt (§8): read, its inputs, and its verification.
|
|
|
|
For every case it requires the outcome, reason and place, and the order, tiles, voices,
|
|
samples, digests, participants and sources the corpus states; of the corpus, that every
|
|
reason of §10 refuses a case. Then it mutates copies of the corpus, one defect each, and
|
|
requires every copy to fail.
|
|
|
|
python3 live_compose_verify.py <live-conformance/compose>
|
|
|
|
Exit status 0 only if every case verifies and every mutated copy fails. Python 3.8 or
|
|
later; standard library only.
|
|
"""
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import os
|
|
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
|
|
|
|
DESCRIPTOR_DOMAIN = b"wai:live-compose\x01"
|
|
RECEIPT_DOMAIN = b"wai:live-compose-receipt\x01"
|
|
SET_DOMAIN = b"wai:live-compose-set\x01"
|
|
ZERO = bytes(32)
|
|
AUDIO, VIDEO, SCREEN = 1, 2, 3
|
|
MEDIA_NAME = {1: "audio", 2: "video", 3: "screen"}
|
|
COMPOSE_POWER = 4
|
|
JOIN = 0x10
|
|
GRID, PIP = 1, 2
|
|
AUTO, VIDEO_ASPECT, SQUARE, PORTRAIT = 0, 1, 2, 3
|
|
DEFAULT, FILL, COVER, CONTAIN = 0, 1, 2, 3
|
|
STATIC, DYNAMIC = 1, 2
|
|
SINGLE, SIMULCAST, SVC, STAGED = 0, 1, 2, 3
|
|
|
|
# §10, in its order.
|
|
REASONS = [
|
|
"canvas_out_of_range", "compose_invalid", "compose_not_permitted", "not_yet_effective",
|
|
"lifecycle_incomplete", "attr_malformed", "mix_input_malformed", "receipt_invalid",
|
|
"descriptor_mismatch", "compositor_not_permitted", "output_not_published", "state_mismatch",
|
|
"inputs_mismatch",
|
|
]
|
|
|
|
PARAM_KEYS = ["width", "height", "fps", "layout", "gap", "aspect", "fill", "featured_attr", "order_attr",
|
|
"omit_stopped", "pip_attr", "corner", "pip_width", "pip_height", "offset", "behaviour",
|
|
"channels", "sample_rate", "gain_attr", "pan_attr", "max_tiles", "max_voices"]
|
|
|
|
|
|
def sha(*parts):
|
|
h = hashlib.sha256()
|
|
for p in parts:
|
|
h.update(p)
|
|
return h.digest()
|
|
|
|
|
|
class Reader:
|
|
"""Fields in order; running out is `truncated`."""
|
|
|
|
def __init__(self, b, p=0):
|
|
self.b, self.p = b, p
|
|
|
|
def take(self, n):
|
|
if self.p + n > len(self.b):
|
|
raise Refuse("truncated")
|
|
s = self.b[self.p:self.p + n]
|
|
self.p += n
|
|
return s
|
|
|
|
def u8(self):
|
|
return self.take(1)[0]
|
|
|
|
def u16(self):
|
|
return struct.unpack("<H", self.take(2))[0]
|
|
|
|
def u32(self):
|
|
return struct.unpack("<I", self.take(4))[0]
|
|
|
|
def u64(self):
|
|
return struct.unpack("<Q", self.take(8))[0]
|
|
|
|
def short(self):
|
|
return self.take(self.u8())
|
|
|
|
|
|
def read_head(b, magic):
|
|
"""§3.3 step 1."""
|
|
if len(b) < 4:
|
|
raise Refuse("truncated")
|
|
if b[:4] != magic:
|
|
raise Refuse("malformed")
|
|
if len(b) < 5:
|
|
raise Refuse("truncated")
|
|
if b[4] != 1:
|
|
raise Refuse("unknown_version")
|
|
|
|
|
|
def is_key(k):
|
|
"""live-rooms §2: a slot name, 1 to 32 of a-z, 0-9 and _."""
|
|
return 1 <= len(k) <= 32 and all(c in b"abcdefghijklmnopqrstuvwxyz0123456789_" for c in k)
|
|
|
|
|
|
# ---------------------------------------------------------------- §3 the descriptor
|
|
|
|
def canvas_ok(p):
|
|
"""§3.4."""
|
|
def edge(e):
|
|
return 16 <= e <= 1920 and e % 2 == 0
|
|
return edge(p["width"]) and edge(p["height"]) and p["width"] * p["height"] <= 2073600 and 1 <= p["fps"] <= 60
|
|
|
|
|
|
def least_root(n):
|
|
k = 0
|
|
while k * k < n:
|
|
k += 1
|
|
return k
|
|
|
|
|
|
def fields_ok(p):
|
|
"""§3.3's rules after the canvas's."""
|
|
w, h, g = p["width"], p["height"], p["gap"]
|
|
if p["layout"] not in (GRID, PIP) or p["aspect"] > 3 or p["fill"] > 3:
|
|
return False
|
|
for k in ("featured_attr", "order_attr", "pip_attr", "gain_attr", "pan_attr"):
|
|
if p[k] != "" and not is_key(p[k].encode()):
|
|
return False
|
|
if p["layout"] == GRID:
|
|
if p["pip_attr"] != "" or p["corner"] or p["pip_width"] or p["pip_height"] or p["offset"] or p["behaviour"]:
|
|
return False
|
|
else:
|
|
o = p["offset"]
|
|
if not (1 <= p["corner"] <= 4 and p["behaviour"] in (STATIC, DYNAMIC) and o < w and o < h
|
|
and p["pip_width"] + o <= w and p["pip_height"] + o <= h):
|
|
return False
|
|
if p["channels"] not in (1, 2) or not 8000 <= p["sample_rate"] <= 192000:
|
|
return False
|
|
if not (1 <= p["max_tiles"] <= 64 and 1 <= p["max_voices"] <= 64):
|
|
return False
|
|
k = least_root(p["max_tiles"])
|
|
need = k + (k - 1) * g
|
|
return g < h and w >= need and h - g - (h - g) * 3 // 4 >= need
|
|
|
|
|
|
def check_params(p):
|
|
if not canvas_ok(p):
|
|
raise Refuse("canvas_out_of_range")
|
|
if not fields_ok(p):
|
|
raise Refuse("compose_invalid")
|
|
|
|
|
|
def read_descriptor(b):
|
|
"""§3.3: the descriptor, or Refuse."""
|
|
read_head(b, b"WCD1")
|
|
r = Reader(b, 5)
|
|
c = {"room": r.take(32), "signer": r.take(32), "effective": r.u64(), "compositor": r.take(32)}
|
|
p = {"width": r.u16(), "height": r.u16(), "fps": r.u8()}
|
|
c["background"], c["layer"] = r.take(3), r.u8()
|
|
p["layout"], p["gap"], p["aspect"], p["fill"] = r.u8(), r.u16(), r.u8(), r.u8()
|
|
raw = {"featured_attr": r.short(), "order_attr": r.short()}
|
|
omit = r.u8()
|
|
raw["pip_attr"] = r.short()
|
|
p["corner"], p["pip_width"], p["pip_height"], p["offset"], p["behaviour"] = r.u8(), r.u16(), r.u16(), r.u16(), r.u8()
|
|
p["channels"], p["sample_rate"] = r.u8(), r.u32()
|
|
raw["gain_attr"], raw["pan_attr"] = r.short(), r.short()
|
|
p["max_tiles"], p["max_voices"] = r.u8(), r.u8()
|
|
body = b[:r.p]
|
|
c["sig"] = r.take(64)
|
|
if r.p != len(b):
|
|
raise Refuse("malformed")
|
|
keys_ok = all(v == b"" or is_key(v) for v in raw.values())
|
|
for k, v in raw.items():
|
|
p[k] = v.decode("latin-1")
|
|
p["omit_stopped"] = omit == 1
|
|
if not canvas_ok(p):
|
|
raise Refuse("canvas_out_of_range")
|
|
if not keys_ok or omit > 1 or not fields_ok(p):
|
|
raise Refuse("compose_invalid")
|
|
bad = ED.verify(c["signer"], DESCRIPTOR_DOMAIN + body, c["sig"])
|
|
if bad:
|
|
raise Refuse(bad)
|
|
c["params"] = p
|
|
c["digest"] = sha(body)
|
|
return c
|
|
|
|
|
|
def fields_of(c):
|
|
out = {"room": c["room"].hex(), "signer": c["signer"].hex(), "effective": c["effective"],
|
|
"compositor": c["compositor"].hex(), "background": c["background"].hex(), "layer": c["layer"]}
|
|
for k in PARAM_KEYS:
|
|
out[k] = c["params"][k]
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------- §4.3 attributes
|
|
|
|
def decimal(v):
|
|
"""A decimal integer of §4.3, or None."""
|
|
neg = v.startswith("-")
|
|
digits = v[1:] if neg else v
|
|
if not 1 <= len(digits) <= 10 or any(ch not in "0123456789" for ch in digits):
|
|
return None
|
|
if digits[0] == "0" and (len(digits) > 1 or neg):
|
|
return None
|
|
n = int(digits)
|
|
return -n if neg else n
|
|
|
|
|
|
def read_attributes(p, attrs):
|
|
"""§4.3 for one participant: (featured, order, pip, gain, pan), or Refuse."""
|
|
def look(key):
|
|
if key == "":
|
|
return None
|
|
return next((v for k, v in attrs if k == key), None)
|
|
|
|
def boolean(key):
|
|
v = look(key)
|
|
if v is None or v == "false":
|
|
return False
|
|
if v == "true":
|
|
return True
|
|
raise Refuse("attr_malformed")
|
|
|
|
def ranged(key, lo, hi, absent):
|
|
v = look(key)
|
|
if v is None:
|
|
return absent
|
|
n = decimal(v)
|
|
if n is None or not lo <= n <= hi:
|
|
raise Refuse("attr_malformed")
|
|
return n
|
|
|
|
featured = boolean(p["featured_attr"])
|
|
order = ranged(p["order_attr"], -2 ** 31, 2 ** 31 - 1, None)
|
|
pip = boolean(p["pip_attr"])
|
|
gain = ranged(p["gain_attr"], 0, 16384, 4096)
|
|
pan = ranged(p["pan_attr"], -4096, 4096, 0)
|
|
return featured, order, pip, gain, pan
|
|
|
|
|
|
# ---------------------------------------------------------------- §5 the layout
|
|
|
|
def fit(bw, bh, a, b):
|
|
"""§5.1."""
|
|
if bw * b <= bh * a:
|
|
return bw, max(1, bw * b // a)
|
|
return max(1, bh * a // b), bh
|
|
|
|
|
|
def centred(outer, w, h):
|
|
x, y, ow, oh = outer
|
|
return (x + (ow - w) // 2, y + (oh - h) // 2, w, h)
|
|
|
|
|
|
def reference(aspect):
|
|
return {SQUARE: (1, 1), PORTRAIT: (3, 4)}.get(aspect, (16, 9))
|
|
|
|
|
|
def grid(region, n, gap, aspect):
|
|
"""§5.3: n tile rectangles."""
|
|
X, Y, W, H = region
|
|
a, b = reference(aspect)
|
|
chosen = None
|
|
for c in range(1, n + 1):
|
|
r = -(-n // c)
|
|
if (c - 1) * gap >= W or (r - 1) * gap >= H:
|
|
continue
|
|
cw, ch = (W - (c - 1) * gap) // c, (H - (r - 1) * gap) // r
|
|
if cw < 1 or ch < 1:
|
|
continue
|
|
tw, th = fit(cw, ch, a, b)
|
|
if chosen is None or tw * th >= chosen[0]:
|
|
chosen = (tw * th, c, r, cw, ch)
|
|
_, c, r, cw, ch = chosen
|
|
y0 = Y + (H - (r * ch + (r - 1) * gap)) // 2
|
|
out = []
|
|
for k in range(n):
|
|
i, j = k // c, k % c
|
|
m = n - (r - 1) * c if i == r - 1 else c
|
|
x0 = X + (W - (m * cw + (m - 1) * gap)) // 2
|
|
cell = (x0 + j * (cw + gap), y0 + i * (ch + gap), cw, ch)
|
|
out.append(cell if aspect == AUTO else centred(cell, *fit(cw, ch, a, b)))
|
|
return out
|
|
|
|
|
|
def show(tile, size, mode):
|
|
"""§5.6: (destination, source) of a source of `size` in `tile`."""
|
|
tx, ty, tw, th = tile
|
|
sw, sh = size
|
|
wider = sw * th >= sh * tw
|
|
whole = (0, 0, sw, sh)
|
|
if mode == FILL:
|
|
return tile, whole
|
|
if mode == CONTAIN:
|
|
if wider:
|
|
dh = max(1, tw * sh // sw)
|
|
return (tx, ty + (th - dh) // 2, tw, dh), whole
|
|
dw = max(1, th * sw // sh)
|
|
return (tx + (tw - dw) // 2, ty, dw, th), whole
|
|
if wider:
|
|
cw = max(1, sh * tw // th)
|
|
return tile, ((sw - cw) // 2, 0, cw, sh)
|
|
ch = max(1, sw * th // tw)
|
|
return tile, (0, (sh - ch) // 2, sw, ch)
|
|
|
|
|
|
def layout(p, entrants):
|
|
"""§5 and §6's voices over entrants (principal, join, video, audio, attrs): a dict."""
|
|
check_params(p)
|
|
members = []
|
|
for e in entrants:
|
|
featured, order, pip, gain, pan = read_attributes(p, e["attrs"])
|
|
members.append(dict(e, featured=featured, order=order, pip=pip, gain=gain, pan=pan))
|
|
# §4.4
|
|
members.sort(key=lambda m: (m["order"] is None, m["order"] or 0, m["join"], m["principal"]))
|
|
voices = [{"principal": m["principal"].hex(), "gain": m["gain"], "pan": m["pan"]}
|
|
for m in members if m["audio"]][:p["max_voices"]]
|
|
# §5.2
|
|
shown = [m for m in members if not (p["omit_stopped"] and m["video"] is None)]
|
|
pip_m = next((m for m in shown if m["pip"]), None) if p["layout"] == PIP else None
|
|
rest = [m for m in shown if m is not pip_m]
|
|
f = next((i for i, m in enumerate(rest) if m["featured"]), None)
|
|
if f is not None:
|
|
rest.insert(0, rest.pop(f))
|
|
rest = rest[:p["max_tiles"] - (1 if pip_m is not None else 0)]
|
|
featured = f is not None and bool(rest)
|
|
window = pip_m
|
|
if pip_m is not None and not rest and p["behaviour"] == DYNAMIC:
|
|
rest, featured, window = [pip_m], pip_m["featured"], None
|
|
tiles = grid_tiles(p, rest, featured)
|
|
if window is not None:
|
|
tiles.append(pip_tile(p, window))
|
|
return {"order": [m["principal"].hex() for m in members], "tiles": tiles, "voices": voices}
|
|
|
|
|
|
def tile_of(p, m, rect, place, z):
|
|
mode = p["fill"] if p["fill"] != DEFAULT else (CONTAIN if place == "featured" else COVER)
|
|
out = {"principal": m["principal"].hex(), "place": place, "z": z, "tile": list(rect), "dst": None, "src": None}
|
|
if m["video"] is not None:
|
|
dst, src = show(rect, m["video"], mode)
|
|
out["dst"], out["src"] = list(dst), list(src)
|
|
return out
|
|
|
|
|
|
def grid_tiles(p, ms, featured):
|
|
"""§5.3, §5.4."""
|
|
w, h, g = p["width"], p["height"], p["gap"]
|
|
if not ms:
|
|
return []
|
|
if not featured:
|
|
return [tile_of(p, m, r, "grid", 0) for m, r in zip(ms, grid((0, 0, w, h), len(ms), g, p["aspect"]))]
|
|
if len(ms) == 1:
|
|
return [tile_of(p, ms[0], (0, 0, w, h), "featured", 0)]
|
|
mh = (h - g) * 3 // 4
|
|
out = [tile_of(p, ms[0], (0, 0, w, mh), "featured", 0)]
|
|
band = (0, mh + g, w, h - mh - g)
|
|
out += [tile_of(p, m, r, "grid", 0) for m, r in zip(ms[1:], grid(band, len(ms) - 1, g, p["aspect"]))]
|
|
return out
|
|
|
|
|
|
def pip_tile(p, m):
|
|
"""§5.5."""
|
|
w, h, o = p["width"], p["height"], p["offset"]
|
|
if p["pip_width"] > 0 and p["pip_height"] > 0:
|
|
pw, ph = p["pip_width"], p["pip_height"]
|
|
else:
|
|
a, b = m["video"] if m["video"] is not None else (w, h)
|
|
if p["pip_width"] > 0:
|
|
bw = p["pip_width"]
|
|
elif p["pip_height"] > 0:
|
|
bw = w - o
|
|
else:
|
|
bw = min(w // 4, w - o)
|
|
bh = p["pip_height"] if p["pip_height"] > 0 else h - o
|
|
pw, ph = fit(bw, bh, a, b)
|
|
x = {1: o, 2: w - o - pw, 3: o, 4: w - o - pw}[p["corner"]]
|
|
y = {1: o, 2: o, 3: h - o - ph, 4: h - o - ph}[p["corner"]]
|
|
return tile_of(p, m, (x, y, pw, ph), "pip", 1)
|
|
|
|
|
|
# ---------------------------------------------------------------- §6 the mix
|
|
|
|
def mix(channels, frames, inputs):
|
|
"""§6: the mixed samples, or Refuse."""
|
|
if channels not in (1, 2):
|
|
raise Refuse("mix_input_malformed")
|
|
for v in inputs:
|
|
if v["channels"] not in (1, 2) or not 0 <= v["gain"] <= 16384 or not -4096 <= v["pan"] <= 4096 \
|
|
or len(v["pcm"]) != frames * v["channels"]:
|
|
raise Refuse("mix_input_malformed")
|
|
sums = [0] * (frames * channels)
|
|
for v in inputs:
|
|
x, g, ch = v["pcm"], v["gain"], v["channels"]
|
|
if channels == 1:
|
|
for f in range(frames):
|
|
sums[f] += (x[f] * 4096 if ch == 1 else (x[2 * f] + x[2 * f + 1]) * 2048) * g
|
|
else:
|
|
al, ar = 4096 - max(v["pan"], 0), 4096 + min(v["pan"], 0)
|
|
for f in range(frames):
|
|
left, right = (x[f], x[f]) if ch == 1 else (x[2 * f], x[2 * f + 1])
|
|
sums[2 * f] += left * g * al
|
|
sums[2 * f + 1] += right * g * ar
|
|
return [min(32767, max(-32768, (s + (1 << 23)) >> 24)) for s in sums]
|
|
|
|
|
|
def pcm_of(h):
|
|
b = bytes.fromhex(h)
|
|
if len(b) % 2:
|
|
raise ValueError("odd PCM")
|
|
return list(struct.unpack("<%dh" % (len(b) // 2), b))
|
|
|
|
|
|
def pcm_hex(samples):
|
|
return struct.pack("<%dh" % len(samples), *samples).hex()
|
|
|
|
|
|
# ---------------------------------------------------------------- §4 the room at an instant
|
|
|
|
def render_field(f):
|
|
"""MOQT §8.8."""
|
|
return "".join(chr(c) if (chr(c).isascii() and chr(c).isalnum()) or c == 0x5F else ".%02x" % c for c in f.encode())
|
|
|
|
|
|
def track_digest(room_id, owner, media, name):
|
|
"""live-events §3.1's anchor track: the rendered namespace, a zero byte, the name."""
|
|
ns = "-".join(render_field(f) for f in (room_id.hex(), owner, MEDIA_NAME[media]))
|
|
return sha(ns.encode(), b"\x00", name.encode())
|
|
|
|
|
|
class RoomView:
|
|
"""A room that verified (§4.1), its replay stepped record by record."""
|
|
|
|
def __init__(self, d, held, steps, final):
|
|
self.d, self.held, self.steps, self.final = d, held, steps, final
|
|
|
|
def grant(self, digest):
|
|
return next(g for g in self.held if g["digest"] == digest)
|
|
|
|
def at(self, t):
|
|
"""The room at t: the state before the first record of the merged order past t."""
|
|
return next((s for rt, s in self.steps if rt > t), self.final)
|
|
|
|
|
|
def snapshot(room, joins):
|
|
return {"present": dict(room.present), "own": dict(room.own), "slots": copy.deepcopy(room.slots), "joins": dict(joins)}
|
|
|
|
|
|
def view_room(room):
|
|
"""live-rooms §9 over the room, stepped: a RoomView, or (reason, place)."""
|
|
descriptor = bytes.fromhex(room["descriptor"])
|
|
grants = [bytes.fromhex(g) for g in room["grants"]]
|
|
head = room["key_log_head"]
|
|
chains = [[bytes.fromhex(r) for r in c] for c in room["chains"]]
|
|
try:
|
|
d = LRV.read_descriptor(descriptor)
|
|
except Refuse as e:
|
|
return e.reason, ["room", "descriptor"]
|
|
try:
|
|
revoked = LRV.check_key_log(d, head)
|
|
except Refuse as e:
|
|
return e.reason, ["room", "key_log"]
|
|
held, bad = LRV.check_grants(d, grants)
|
|
if bad:
|
|
return bad[1], ["room", "grant", bad[0]]
|
|
state = LRV.Room(d, held, revoked)
|
|
auth = LRV.authority_policy()
|
|
ver = LEV.Verifier(d["room"], auth, {})
|
|
seen, joins, steps, at_index = {}, {}, [], []
|
|
numbered = 0
|
|
for c, i in LRV.merged_order(d, chains):
|
|
b = chains[c][i]
|
|
at_index.append((c, i))
|
|
try:
|
|
ev = LEV.read_record(b)
|
|
duplicate = ev["digest"] in seen
|
|
except Refuse:
|
|
ev, duplicate = None, False
|
|
if ev is not None:
|
|
steps.append((ev["t"], snapshot(state, joins)))
|
|
try:
|
|
LRV.step(d, state, ver, seen, auth, b)
|
|
except Refuse as e:
|
|
return e.reason, ["room", "record", c, i]
|
|
if not duplicate and LRV.is_room_key(d, ev["principal"]) and ev["kind"] == JOIN:
|
|
joins[ver.accepted[-1][2]["subject"]] = numbered
|
|
numbered += 1
|
|
if ver.gaps:
|
|
if any(LRV.is_room_key(d, p) for p, _, _ in ver.gaps):
|
|
return "lifecycle_incomplete", ["room"]
|
|
else:
|
|
mismatch = LEV.stated_tally_mismatch(ver.accepted)
|
|
if mismatch is not None:
|
|
c, i = at_index[mismatch]
|
|
return "tally_mismatch", ["room", "record", c, i]
|
|
return RoomView(d, held, steps, snapshot(state, joins))
|
|
|
|
|
|
def source(view, c, profile, layers, slot, media):
|
|
"""§4.2: a source and its composed layer."""
|
|
prof = view.d["profiles"][profile]
|
|
layer = min(layers - 1, c["layer"])
|
|
w, h = prof["layers"][layer][0], prof["layers"][layer][1]
|
|
return {"slot": slot, "media": media, "profile": profile, "layers": layers, "layer": layer,
|
|
"layering": prof["layering"], "width": w, "height": h}
|
|
|
|
|
|
def tracks(view, s, principal):
|
|
"""§8.2: the tracks a source is read from."""
|
|
owner = principal.hex() if s["slot"] == "" else "_" + s["slot"]
|
|
if s["layering"] == SIMULCAST:
|
|
names = [s["layer"]]
|
|
elif s["layering"] == STAGED:
|
|
names = list(range(s["layer"] + 1))
|
|
else:
|
|
names = [0]
|
|
return [track_digest(view.d["room"], owner, s["media"], str(n)) for n in names]
|
|
|
|
|
|
def participants(view, c, t):
|
|
"""§4.2 at t, in ascending order of key."""
|
|
st = view.at(t)
|
|
out = []
|
|
slots = list(zip(view.d["slots"], st["slots"]))
|
|
|
|
def own(p, m):
|
|
h = st["own"].get((p, m))
|
|
return None if h is None else source(view, c, h[0], h[1], "", m)
|
|
|
|
def live_in(p, kinds):
|
|
for s, ss in slots:
|
|
if ss["live"] and ss["live"][0] == p and s["media"] in kinds:
|
|
return source(view, c, ss["live"][1], ss["live"][2], s["name"].decode(), s["media"])
|
|
return None
|
|
|
|
for p in sorted(st["present"]):
|
|
if p == c["compositor"]:
|
|
continue
|
|
if not (any(q == p for q, _ in st["own"]) or any(ss["live"] and ss["live"][0] == p for _, ss in slots)):
|
|
continue
|
|
video = own(p, VIDEO) or own(p, SCREEN) or live_in(p, (VIDEO, SCREEN))
|
|
audio = own(p, AUDIO) or live_in(p, (AUDIO,))
|
|
g = view.grant(st["present"][p])
|
|
out.append({"principal": p, "grant": g["digest"], "join": st["joins"][p],
|
|
"attrs": LRV.read_attrs(g["attrs"]), "video": video, "audio": audio})
|
|
return out
|
|
|
|
|
|
def set_digest(view, t, ps):
|
|
"""§4.5."""
|
|
def src(s):
|
|
if s is None:
|
|
return b"\x00"
|
|
slot = s["slot"].encode()
|
|
return bytes([s["media"], len(slot)]) + slot + bytes([s["profile"], s["layers"], s["layer"]]) \
|
|
+ struct.pack("<HH", s["width"], s["height"])
|
|
out = SET_DOMAIN + view.d["room"] + struct.pack("<QI", t, len(ps))
|
|
for p in ps:
|
|
out += p["principal"] + p["grant"] + struct.pack("<I", p["join"]) + src(p["video"]) + src(p["audio"])
|
|
return sha(out)
|
|
|
|
|
|
def frame(view, c, t):
|
|
"""§4 and §5 at t: the frame as the corpus writes it, with what §8.2 reads."""
|
|
if t < c["effective"]:
|
|
raise Refuse("not_yet_effective")
|
|
ps = participants(view, c, t)
|
|
entrants = [{"principal": p["principal"], "join": p["join"],
|
|
"video": None if p["video"] is None else (p["video"]["width"], p["video"]["height"]),
|
|
"audio": p["audio"] is not None, "attrs": p["attrs"]} for p in ps]
|
|
lay = layout(c["params"], entrants)
|
|
|
|
def src_json(s, principal):
|
|
if s is None:
|
|
return None
|
|
return dict(s, tracks=[x.hex() for x in tracks(view, s, principal)])
|
|
out = {"t": t, "participants": [{"principal": p["principal"].hex(), "grant": p["grant"].hex(), "join": p["join"],
|
|
"attrs": [[k, v] for k, v in p["attrs"]],
|
|
"video": src_json(p["video"], p["principal"]),
|
|
"audio": src_json(p["audio"], p["principal"])} for p in ps],
|
|
"set_digest": set_digest(view, t, ps).hex()}
|
|
out.update(lay)
|
|
by = {p["principal"].hex(): p for p in ps}
|
|
inputs = {VIDEO: [], AUDIO: []}
|
|
for tl in lay["tiles"]:
|
|
if tl["dst"] is not None:
|
|
p = by[tl["principal"]]
|
|
inputs[VIDEO] += [(p["principal"], x) for x in tracks(view, p["video"], p["principal"])]
|
|
for v in lay["voices"]:
|
|
p = by[v["principal"]]
|
|
inputs[AUDIO] += [(p["principal"], x) for x in tracks(view, p["audio"], p["principal"])]
|
|
return out, inputs
|
|
|
|
|
|
def check_descriptor(view, b):
|
|
"""§3.3, §3.5."""
|
|
c = read_descriptor(b)
|
|
if c["room"] != view.d["room"]:
|
|
raise Refuse("wrong_room")
|
|
if c["signer"] != view.d["authority"]:
|
|
st = view.at(c["effective"])
|
|
g = st["present"].get(c["signer"])
|
|
if g is None or view.grant(g)["powers"] & COMPOSE_POWER == 0:
|
|
raise Refuse("compose_not_permitted")
|
|
return c
|
|
|
|
|
|
# ---------------------------------------------------------------- §8 the receipt
|
|
|
|
def read_object(r):
|
|
return {"track": r.take(32), "group": r.u64(), "object": r.u64(), "digest": r.take(32)}
|
|
|
|
|
|
def read_receipt(b):
|
|
"""§8.1: the receipt, or Refuse."""
|
|
read_head(b, b"WCR1")
|
|
r = Reader(b, 5)
|
|
rc = {"room": r.take(32), "compositor": r.take(32), "descriptor": r.take(32), "t": r.u64(), "state": r.take(32),
|
|
"media": r.u8(), "output": read_object(r)}
|
|
rc["inputs"] = [(r.take(32), read_object(r)) for _ in range(r.u8())]
|
|
body = b[:r.p]
|
|
rc["sig"] = r.take(64)
|
|
if r.p != len(b):
|
|
raise Refuse("malformed")
|
|
pairs = [(p, o["track"]) for p, o in rc["inputs"]]
|
|
if rc["media"] not in (AUDIO, VIDEO) or len(set(pairs)) != len(pairs):
|
|
raise Refuse("receipt_invalid")
|
|
bad = ED.verify(rc["compositor"], RECEIPT_DOMAIN + body, rc["sig"])
|
|
if bad:
|
|
raise Refuse(bad)
|
|
rc["digest"] = sha(body)
|
|
return rc
|
|
|
|
|
|
def verify_receipt(view, dbytes, rbytes):
|
|
"""§8.3: what the corpus states for a receipt case, the room already viewed."""
|
|
try:
|
|
c = check_descriptor(view, dbytes)
|
|
except Refuse as e:
|
|
return {"outcome": "refuse", "reason": e.reason, "at": ["descriptor"]}
|
|
|
|
def no(reason):
|
|
return {"outcome": "refuse", "reason": reason, "at": ["receipt"]}
|
|
try:
|
|
rc = read_receipt(rbytes)
|
|
except Refuse as e:
|
|
return no(e.reason)
|
|
if rc["room"] != view.d["room"]:
|
|
return no("wrong_room")
|
|
if rc["descriptor"] != c["digest"]:
|
|
return no("descriptor_mismatch")
|
|
if rc["t"] < c["effective"]:
|
|
return no("not_yet_effective")
|
|
st = view.at(rc["t"])
|
|
g = st["present"].get(rc["compositor"])
|
|
if c["compositor"] == ZERO or rc["compositor"] != c["compositor"] or g is None \
|
|
or view.grant(g)["powers"] & COMPOSE_POWER == 0:
|
|
return no("compositor_not_permitted")
|
|
held = st["own"].get((rc["compositor"], rc["media"]))
|
|
if held is None or rc["output"]["track"] not in [
|
|
track_digest(view.d["room"], rc["compositor"].hex(), rc["media"], str(n)) for n in range(held[1])]:
|
|
return no("output_not_published")
|
|
try:
|
|
f, inputs = frame(view, c, rc["t"])
|
|
except Refuse as e:
|
|
return {"outcome": "refuse", "reason": e.reason, "at": ["set", rc["t"]]}
|
|
if rc["state"].hex() != f["set_digest"]:
|
|
return no("state_mismatch")
|
|
if [(p, o["track"]) for p, o in rc["inputs"]] != inputs[rc["media"]]:
|
|
return no("inputs_mismatch")
|
|
return {"outcome": "accept", "digest": rc["digest"].hex(), "set_digest": f["set_digest"]}
|
|
|
|
|
|
# ---------------------------------------------------------------- the corpus
|
|
|
|
def refuse(reason, at=None):
|
|
out = {"outcome": "refuse", "reason": reason}
|
|
if at is not None:
|
|
out["at"] = at
|
|
return out
|
|
|
|
|
|
def run_layout(v):
|
|
p = v["params"]
|
|
entrants = [{"principal": bytes.fromhex(e["principal"]), "join": e["join"],
|
|
"video": None if e["video"] is None else tuple(e["video"]), "audio": e["audio"],
|
|
"attrs": [tuple(kv) for kv in e["attrs"]]} for e in v["entrants"]]
|
|
try:
|
|
return dict(outcome="layout", **layout(p, entrants))
|
|
except Refuse as e:
|
|
return refuse(e.reason)
|
|
|
|
|
|
def run_mix(v):
|
|
inputs = [{"channels": x["channels"], "gain": x["gain"], "pan": x["pan"], "pcm": pcm_of(x["pcm"])} for x in v["inputs"]]
|
|
try:
|
|
return {"outcome": "mix", "pcm": pcm_hex(mix(v["channels"], v["frames"], inputs))}
|
|
except Refuse as e:
|
|
return refuse(e.reason)
|
|
|
|
|
|
def run_descriptor(v):
|
|
try:
|
|
c = read_descriptor(bytes.fromhex(v["bytes"]))
|
|
except Refuse as e:
|
|
return refuse(e.reason)
|
|
return {"outcome": "read", "digest": c["digest"].hex(), "fields": fields_of(c)}
|
|
|
|
|
|
class Rooms:
|
|
"""Each room of the corpus viewed once, by its content."""
|
|
|
|
def __init__(self):
|
|
self.cache = {}
|
|
|
|
def get(self, room):
|
|
k = json.dumps(room, sort_keys=True)
|
|
if k not in self.cache:
|
|
self.cache[k] = view_room(room)
|
|
return self.cache[k]
|
|
|
|
|
|
def run_composition(v, rooms, doc):
|
|
view = rooms.get(doc["rooms"][v["room"]])
|
|
if not isinstance(view, RoomView):
|
|
return refuse(view[0], view[1])
|
|
try:
|
|
c = check_descriptor(view, bytes.fromhex(v["descriptor"]))
|
|
except Refuse as e:
|
|
return refuse(e.reason, ["descriptor"])
|
|
frames = []
|
|
for t in v["instants"]:
|
|
try:
|
|
frames.append(frame(view, c, t)[0])
|
|
except Refuse as e:
|
|
return refuse(e.reason, ["set", t])
|
|
return {"outcome": "compose", "digest": c["digest"].hex(), "frames": frames}
|
|
|
|
|
|
def run_receipt(v, rooms, doc):
|
|
view = rooms.get(doc["rooms"][v["room"]])
|
|
if not isinstance(view, RoomView):
|
|
return refuse(view[0], view[1])
|
|
return verify_receipt(view, bytes.fromhex(v["descriptor"]), bytes.fromhex(v["receipt"]))
|
|
|
|
|
|
SINGULAR = {"layouts": "layout", "mixes": "mix", "descriptors": "descriptor", "compositions": "composition",
|
|
"receipts": "receipt"}
|
|
|
|
SECTIONS = [("layouts", lambda v, rooms, doc: run_layout(v)), ("mixes", lambda v, rooms, doc: run_mix(v)),
|
|
("descriptors", lambda v, rooms, doc: run_descriptor(v)), ("compositions", run_composition),
|
|
("receipts", run_receipt)]
|
|
|
|
|
|
def check(doc, rooms, quiet=False):
|
|
"""Every case against its statement: the failures."""
|
|
fails = []
|
|
for section, run in SECTIONS:
|
|
for v in doc[section]:
|
|
try:
|
|
got = run(v, rooms, doc)
|
|
except Exception as e: # a crash on a mutated copy is a failure to read it
|
|
got, problems = None, ["the case does not verify: %s: %s" % (type(e).__name__, e)]
|
|
else:
|
|
problems = [k for k in got if v.get(k) != got[k]]
|
|
problems += [k for k in ("outcome", "reason", "at") if k in v and k not in got]
|
|
if not quiet:
|
|
what = "?" if got is None else got["outcome"] + (" " + got["reason"] if got["outcome"] == "refuse" else "")
|
|
print(("PASS" if not problems else "FAIL") + " %s %s: %s" % (SINGULAR[section], v["name"], what))
|
|
for k in problems:
|
|
fails.append("%s %s: %s is %s, the corpus states %s" % (
|
|
SINGULAR[section], v["name"], k, json.dumps(got.get(k))[:200] if got else k, json.dumps(v.get(k))[:200]))
|
|
return fails
|
|
|
|
|
|
def flip(h, at, bit=0x01):
|
|
b = bytearray(bytes.fromhex(h))
|
|
b[at] ^= bit
|
|
return b.hex()
|
|
|
|
|
|
def mutations(doc):
|
|
"""Copies of the corpus, each with one defect a conforming verifier must catch."""
|
|
idx = {s: {v["name"]: i for i, v in enumerate(doc[s])} for s, _ in SECTIONS}
|
|
|
|
def case(d, section, name):
|
|
return d[section][idx[section][name]]
|
|
|
|
def mutate(label, f):
|
|
d = copy.deepcopy(doc)
|
|
f(d)
|
|
return label, d
|
|
|
|
def frame_at(d, name, t):
|
|
return next(f for f in case(d, "compositions", name)["frames"] if f["t"] == t)
|
|
|
|
def relay_drop(d):
|
|
relay = doc["principals"]["relay"]
|
|
chain = next(c for c in d["rooms"]["studio"]["chains"] if LEV.read_record(bytes.fromhex(c[0]))["principal"].hex() == relay)
|
|
chain.pop(5)
|
|
|
|
return [
|
|
mutate("grid_5: a tile one pixel right", lambda d: case(d, "layouts", "grid_5")["tiles"][4]["tile"].__setitem__(0, case(d, "layouts", "grid_5")["tiles"][4]["tile"][0] + 1)),
|
|
mutate("featured_4: two tiles swapped", lambda d: case(d, "layouts", "featured_4")["tiles"].reverse()),
|
|
mutate("pip_bottom_right_static: the window under the grid", lambda d: case(d, "layouts", "pip_bottom_right_static")["tiles"][-1].update(z=0)),
|
|
mutate("fill_cover: a source rectangle", lambda d: case(d, "layouts", "fill_cover")["tiles"][0]["src"].__setitem__(1, 1)),
|
|
mutate("fill_contain: a destination rectangle", lambda d: case(d, "layouts", "fill_contain")["tiles"][0]["dst"].__setitem__(3, 1)),
|
|
mutate("order_key_tie_break: two of the order swapped", lambda d: case(d, "layouts", "order_key_tie_break")["order"].reverse()),
|
|
mutate("order_attribute: an entrant's order", lambda d: case(d, "layouts", "order_attribute")["entrants"][2]["attrs"][0].__setitem__(1, "9")),
|
|
mutate("stopped_omitted: omit_stopped off", lambda d: case(d, "layouts", "stopped_omitted")["params"].update(omit_stopped=False)),
|
|
mutate("pip_alone_dynamic: stated static", lambda d: case(d, "layouts", "pip_alone_dynamic")["params"].update(behaviour=STATIC)),
|
|
mutate("attr_order_leading_zero: stated compose_invalid", lambda d: case(d, "layouts", "attr_order_leading_zero").update(reason="compose_invalid")),
|
|
mutate("voices: a gain", lambda d: case(d, "layouts", "voices")["voices"][0].update(gain=4097)),
|
|
mutate("sum_saturates_high: an output sample", lambda d: case(d, "mixes", "sum_saturates_high").update(pcm=flip(case(d, "mixes", "sum_saturates_high")["pcm"], 0))),
|
|
mutate("half_gain_ties: an input sample", lambda d: case(d, "mixes", "half_gain_ties")["inputs"][0].update(pcm=flip(case(d, "mixes", "half_gain_ties")["inputs"][0]["pcm"], 0))),
|
|
mutate("pan_hard_left: panned right", lambda d: case(d, "mixes", "pan_hard_left")["inputs"][0].update(pan=4096)),
|
|
mutate("descriptor grid: a byte of its body", lambda d: case(d, "descriptors", "grid").update(bytes=flip(case(d, "descriptors", "grid")["bytes"], 40))),
|
|
mutate("descriptor grid: its stated digest", lambda d: case(d, "descriptors", "grid").update(digest="00" * 32)),
|
|
mutate("descriptor canvas_too_wide: stated compose_invalid", lambda d: case(d, "descriptors", "canvas_too_wide").update(reason="compose_invalid")),
|
|
mutate("grid_session: a tile at 5 s", lambda d: frame_at(d, "grid_session", 5000000)["tiles"][1]["tile"].__setitem__(2, 1)),
|
|
mutate("grid_session: the set digest at 7 s", lambda d: frame_at(d, "grid_session", 7000000).update(set_digest="00" * 32)),
|
|
mutate("grid_session: a participant's join order", lambda d: frame_at(d, "grid_session", 4000000)["participants"][0].update(join=9)),
|
|
mutate("grid_session: a staged source's tracks reordered", lambda d: next(
|
|
p for p in frame_at(d, "grid_session", 4000000)["participants"] if p["video"] and len(p["video"]["tracks"]) > 1)["video"]["tracks"].reverse()),
|
|
mutate("grid_session: an instant moved", lambda d: case(d, "compositions", "grid_session")["instants"].__setitem__(4, 4500000)),
|
|
mutate("pip_session: a voice's pan", lambda d: frame_at(d, "pip_session", 6000000)["voices"][0].update(pan=1)),
|
|
mutate("studio: a lifecycle record not delivered", relay_drop),
|
|
mutate("studio: a byte of a grant", lambda d: d["rooms"]["studio"]["grants"].__setitem__(1, flip(d["rooms"]["studio"]["grants"][1], 200))),
|
|
mutate("receipt video: a byte of it", lambda d: case(d, "receipts", "video").update(receipt=flip(case(d, "receipts", "video")["receipt"], 150))),
|
|
mutate("receipt video: its stated digest", lambda d: case(d, "receipts", "video").update(digest="00" * 32)),
|
|
mutate("receipt inputs_missing: stated accepted", lambda d: case(d, "receipts", "inputs_missing").update(outcome="accept")),
|
|
mutate("receipt state_stale: stated inputs_mismatch", lambda d: case(d, "receipts", "state_stale").update(reason="inputs_mismatch")),
|
|
]
|
|
|
|
|
|
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)
|
|
rooms = Rooms()
|
|
fails += check(doc, rooms)
|
|
seen = {v.get("reason") for s, _ in SECTIONS for v in doc[s]}
|
|
missing = [r for r in REASONS if r not in seen]
|
|
if missing:
|
|
fails.append("no case is refused %s" % ", ".join(missing))
|
|
print("%s every reason of §10 refuses a case" % ("PASS" if not missing else "FAIL"))
|
|
for label, mutated in mutations(doc):
|
|
caught = check(mutated, rooms, quiet=True)
|
|
print(("PASS" if caught else "FAIL") + " mutated copy fails: %s%s" % (label, " (%s)" % caught[0][:120] if caught else ""))
|
|
if not caught:
|
|
fails.append("a mutated copy verifies: %s" % label)
|
|
for f in fails:
|
|
print("FAIL " + f)
|
|
total = sum(len(doc[s]) for s, _ in SECTIONS)
|
|
print(("PASS" if not fails else "FAIL") + " live-composition corpus: %d cases" % total)
|
|
return 1 if fails else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) != 2:
|
|
print("usage: live_compose_verify.py <live-conformance/compose>", file=sys.stderr)
|
|
sys.exit(2)
|
|
sys.exit(main(sys.argv[1]))
|