1eabee8f24
The registry row and its export; the verifiers' mutation self-tests; a step for both verifiers in wai-conformance, and joule-code-conformance's path filters for the corpora and texts its tests read. Co-Authored-By: Claude Opus 5.5 <[email protected]>
647 lines
26 KiB
Python
647 lines
26 KiB
Python
#!/usr/bin/env python3
|
|
"""Second implementation of WAI's cue injection (extensions/cue-injection.md), written from
|
|
that text, from ANSI/SCTE 104 2023 and ANSI/SCTE 35-1 2023r2, and from live-events and
|
|
live-ads where it cites them. It shares no code with the reference (joule-moq-ingest).
|
|
|
|
For the corpus live-conformance/injection/vectors.json it:
|
|
|
|
- translates every message (§4 to §6) and requires the processing instant, the warning, and
|
|
each request's opID, splice instant, section, cues and withdrawals the corpus states; a
|
|
translation that names a standard section requires that section byte for byte;
|
|
- requires every refusal's result code, extension and reason (§4.1, §5, §9);
|
|
- runs the injector's rules (§7.3) over every scenario, requires each message's result,
|
|
what was dropped and withdrawn, and the chain: each event verifies under the principal
|
|
(WAI's strict Ed25519), links to the one before, and is the record and cue §7.2 makes;
|
|
then verifies the chain as §8 does and derives its breaks and chapters by live-ads §4.3;
|
|
- verifies every negative chain and requires the event and reason it is refused at.
|
|
|
|
python3 -I injection_verify.py [--self-test] <live-conformance/injection>
|
|
|
|
With --self-test it then requires every mutated copy of the corpus (a section byte, a cue's
|
|
instant or duration, a refusal's reason, an event dropped or its signature changed, a break's
|
|
end, a count, where a negative chain is refused) to fail.
|
|
|
|
Exit status 0 only if every case agrees. Python 3.8 or later; standard library only.
|
|
"""
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import struct
|
|
import sys
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance"))
|
|
import _ed25519 as ED # noqa: E402
|
|
|
|
EVENT_DOMAIN = b"wai:live-event\x01"
|
|
CUEI = b"CUEI"
|
|
EPOCH_1980 = 315_964_800
|
|
MIN_PREROLL = 4000
|
|
|
|
# §4.3: the requests this revision implements.
|
|
NORMAL = {0x0101: "splice", 0x0102: "splice_null", 0x0104: "time_signal"}
|
|
SUPPLEMENTAL = {0x0108, 0x0109, 0x010A, 0x010B, 0x010F, 0x0110}
|
|
UNIMPLEMENTED = {0x0100, 0x0103, 0x0105, 0x0106, 0x0107, 0x010C, 0x010D, 0x010E, 0x0111, 0x0112, 0x0113, 0x0300, 0x0301}
|
|
# §6.2.
|
|
BREAK_STARTS = {0x22, 0x30, 0x32, 0x34, 0x36}
|
|
BREAK_ENDS = {0x23, 0x31, 0x33, 0x35, 0x37}
|
|
CHAPTER_START = 0x20
|
|
SUB_SEGMENT_TYPES = {0x30, 0x32, 0x34, 0x36, 0x38, 0x3A, 0x44, 0x46}
|
|
|
|
|
|
class Refused(Exception):
|
|
def __init__(self, result, reason, ext=0xFFFF):
|
|
super().__init__(reason)
|
|
self.result, self.reason, self.ext = result, reason, ext
|
|
|
|
|
|
def crc32_mpeg2(b):
|
|
"""ISO/IEC 13818-1 Annex A: polynomial 0x04C11DB7, preset all ones, no reflection."""
|
|
c = 0xFFFFFFFF
|
|
for x in b:
|
|
c ^= x << 24
|
|
for _ in range(8):
|
|
c = ((c << 1) ^ 0x04C11DB7) & 0xFFFFFFFF if c & 0x80000000 else (c << 1) & 0xFFFFFFFF
|
|
return c
|
|
|
|
|
|
def pts(s_us):
|
|
"""§5.2: the room clock at 90 kHz, modulo 2^33."""
|
|
return (s_us * 9 // 100) % (1 << 33)
|
|
|
|
|
|
def splice_time(p):
|
|
"""SCTE 35-1 Table 14: time_specified_flag 1, six reserved bits (1), pts_time."""
|
|
return bytes([0xFE | (p >> 32)]) + (p & 0xFFFFFFFF).to_bytes(4, "big")
|
|
|
|
|
|
def break_duration(auto, ticks):
|
|
"""SCTE 35-1 Table 15."""
|
|
return bytes([(0x80 if auto else 0) | 0x7E | (ticks >> 32)]) + (ticks & 0xFFFFFFFF).to_bytes(4, "big")
|
|
|
|
|
|
def section(tier, cmd_type, cmd, descriptors):
|
|
"""§5.1 and SCTE 35-1 Table 5."""
|
|
loop = b"".join(descriptors)
|
|
length = 1 + 5 + 1 + 3 + 1 + len(cmd) + 2 + len(loop) + 4
|
|
if length > 4093:
|
|
raise Refused(115, "section_too_long")
|
|
o = bytearray([0xFC, 0x30 | (length >> 8), length & 0xFF, 0])
|
|
o += bytes(5) # encrypted_packet, encryption_algorithm, pts_adjustment
|
|
o.append(0xFF) # cw_index
|
|
o += bytes([(tier >> 4) & 0xFF, ((tier & 0xF) << 4) | (len(cmd) >> 8), len(cmd) & 0xFF, cmd_type])
|
|
o += cmd + len(loop).to_bytes(2, "big") + loop
|
|
o += crc32_mpeg2(bytes(o)).to_bytes(4, "big")
|
|
return bytes(o)
|
|
|
|
|
|
def descriptor(tag, body):
|
|
return bytes([tag, len(body)]) + body
|
|
|
|
|
|
class Reader:
|
|
def __init__(self, b):
|
|
self.b, self.p = b, 0
|
|
|
|
def take(self, n):
|
|
if self.p + n > len(self.b):
|
|
raise Refused(114, "message_size")
|
|
s = self.b[self.p:self.p + n]
|
|
self.p += n
|
|
return s
|
|
|
|
def u(self, n):
|
|
return int.from_bytes(self.take(n), "big")
|
|
|
|
def left(self):
|
|
return len(self.b) - self.p
|
|
|
|
|
|
def flag(x):
|
|
if x not in (0, 1):
|
|
raise Refused(115, "field_range")
|
|
return x == 1
|
|
|
|
|
|
def segmentation(d, prm):
|
|
"""§5.6, SCTE 104 Table 9-29 to SCTE 35-1 Table 20."""
|
|
r = Reader(d)
|
|
eid = r.u(4)
|
|
cancel = flag(r.u(1))
|
|
duration = r.u(2)
|
|
upid_type = r.u(1)
|
|
upid = r.take(r.u(1))
|
|
type_id, seg_num, seg_exp = r.u(1), r.u(1), r.u(1)
|
|
rest = r.left()
|
|
ext, flags, sub = 0, None, None
|
|
if rest == 0:
|
|
pass
|
|
elif rest == 1:
|
|
ext = r.u(1)
|
|
elif rest in (6, 9):
|
|
ext = r.u(1)
|
|
flags = [r.u(1) for _ in range(5)]
|
|
if rest == 9:
|
|
sub = [r.u(1) for _ in range(3)]
|
|
else:
|
|
raise Refused(114, "message_size")
|
|
body = CUEI + eid.to_bytes(4, "big")
|
|
if cancel:
|
|
return {"event_id": eid, "cancel": True, "type_id": type_id, "ticks": None, "descriptor": descriptor(0x02, body + b"\xff")}
|
|
body += b"\x7f"
|
|
if flags is None:
|
|
dnr, web, nrb, arch, dev = True, False, False, False, 0
|
|
else:
|
|
if flags[4] > 3:
|
|
raise Refused(115, "field_range")
|
|
dnr, web, nrb, arch = (flag(x) for x in flags[:4])
|
|
dev = flags[4]
|
|
ticks = None
|
|
if duration > 0:
|
|
ticks = duration * 90000 + (ext * 90000 * prm["fps_den"]) // prm["fps_num"]
|
|
b = 0x80 | (0x40 if ticks is not None else 0) | (0x20 if dnr else 0)
|
|
b |= 0x1F if dnr else (web << 4) | (nrb << 3) | (arch << 2) | dev
|
|
body += bytes([b])
|
|
if ticks is not None:
|
|
body += ticks.to_bytes(5, "big")
|
|
body += bytes([upid_type, len(upid)]) + upid + bytes([type_id, seg_num, seg_exp])
|
|
if sub is not None and flag(sub[0]):
|
|
if type_id not in SUB_SEGMENT_TYPES:
|
|
raise Refused(115, "field_range")
|
|
body += bytes(sub[1:])
|
|
if len(body) > 254:
|
|
raise Refused(115, "descriptor_too_long")
|
|
return {"event_id": eid, "cancel": False, "type_id": type_id, "ticks": ticks, "descriptor": descriptor(0x02, body)}
|
|
|
|
|
|
def supplements(sups, prm):
|
|
descs, tier, segs = [], None, []
|
|
for opid, d in sups:
|
|
if opid == 0x010A:
|
|
if not d or len(d) != 1 + 4 * d[0]:
|
|
raise Refused(114, "message_size")
|
|
for i in range(d[0]):
|
|
descs.append(descriptor(0x00, CUEI + d[1 + 4 * i:5 + 4 * i]))
|
|
elif opid == 0x0109:
|
|
if len(d) < 2 or len(d) != 2 + d[1]:
|
|
raise Refused(114, "message_size")
|
|
if d[1] > 7:
|
|
raise Refused(115, "field_range")
|
|
descs.append(descriptor(0x01, CUEI + bytes([d[0], (d[1] << 5) | 0x1F]) + d[2:]))
|
|
elif opid == 0x0110:
|
|
if len(d) != 12:
|
|
raise Refused(114, "message_size")
|
|
descs.append(descriptor(0x03, CUEI + d))
|
|
elif opid == 0x0108:
|
|
if not d:
|
|
raise Refused(114, "message_size")
|
|
q = 1
|
|
for _ in range(d[0]):
|
|
if q + 2 > len(d) or q + 2 + d[q + 1] > len(d):
|
|
raise Refused(114, "message_size")
|
|
img = d[q:q + 2 + d[q + 1]]
|
|
if img[0] <= 0x04 and img[2:6] == CUEI:
|
|
raise Refused(115, "descriptor_image_standard")
|
|
descs.append(img)
|
|
q += len(img)
|
|
if q != len(d):
|
|
raise Refused(114, "message_size")
|
|
elif opid == 0x010F:
|
|
if len(d) != 2:
|
|
raise Refused(114, "message_size")
|
|
t = int.from_bytes(d, "big")
|
|
if t >> 12 or tier is not None:
|
|
raise Refused(115, "field_range")
|
|
tier = t
|
|
elif opid == 0x010B:
|
|
s = segmentation(d, prm)
|
|
descs.append(s["descriptor"])
|
|
segs.append(s)
|
|
return descs, (0xFFF if tier is None else tier), segs
|
|
|
|
|
|
def cue_payload(c):
|
|
o = bytes([c["kind"]]) + struct.pack("<II", c["cue_id"], c["duration_ms"])
|
|
if c["section"] is None:
|
|
return o + b"\x00"
|
|
return o + b"\x01" + struct.pack("<H", len(c["section"])) + c["section"]
|
|
|
|
|
|
def request(name, data, sups, P, prm):
|
|
descs, tier, segs = supplements(sups, prm)
|
|
warning = None
|
|
if name == "splice":
|
|
if len(data) not in (14, 15):
|
|
raise Refused(114, "message_size")
|
|
ty = data[0]
|
|
eid = int.from_bytes(data[1:5], "big")
|
|
upid = data[5:7]
|
|
preroll = int.from_bytes(data[7:9], "big")
|
|
tenths = int.from_bytes(data[9:11], "big")
|
|
avail_num, avails_exp, auto = data[11], data[12], data[13] != 0
|
|
if ty == 5:
|
|
return {"opid": 0x0101, "splice_us": P, "section": None, "cues": [], "withdraws": [["splice", eid]]}, None
|
|
if ty < 1 or ty > 4:
|
|
raise Refused(121, "splice_type")
|
|
start = ty <= 2
|
|
delay = preroll if ty in (1, 3) else 0
|
|
if ty == 1 and 0 < delay < MIN_PREROLL:
|
|
warning = 122
|
|
S = P + delay * 1000
|
|
cmd = bytearray(eid.to_bytes(4, "big") + b"\x7f")
|
|
has_dur = start and tenths > 0
|
|
cmd.append((0x80 if start else 0) | 0x40 | (0x20 if has_dur else 0) | (0x10 if delay == 0 else 0) | 0x08 | 0x07)
|
|
if delay > 0:
|
|
cmd += splice_time(pts(S))
|
|
if has_dur:
|
|
cmd += break_duration(auto, tenths * 9000)
|
|
cmd += upid + bytes([avail_num, avails_exp])
|
|
sec = section(tier, 0x05, bytes(cmd), descs)
|
|
dur = tenths * 100 if start else 0
|
|
cues = [{"t": S, "kind": 1 if start else 2, "cue_id": eid, "duration_ms": dur, "derived": False, "section": sec}]
|
|
withdraws = []
|
|
if start and auto and tenths > 0:
|
|
cues.append({"t": S + tenths * 100_000, "kind": 2, "cue_id": eid, "duration_ms": dur, "derived": True, "section": None})
|
|
if not start:
|
|
withdraws.append(["auto_return", eid])
|
|
return {"opid": 0x0101, "splice_us": S, "section": sec, "cues": cues, "withdraws": withdraws}, warning
|
|
if name == "time_signal":
|
|
if len(data) != 2:
|
|
raise Refused(114, "message_size")
|
|
S = P + int.from_bytes(data, "big") * 1000
|
|
sec = section(tier, 0x06, splice_time(pts(S)), descs)
|
|
ins, outs, chapters, withdraws = [], [], [], []
|
|
for s in segs:
|
|
if s["cancel"]:
|
|
withdraws.append(["segmentation", s["event_id"]])
|
|
continue
|
|
dur = (s["ticks"] // 90) if s["ticks"] is not None else 0
|
|
c = {"t": S, "cue_id": s["event_id"], "derived": False, "section": sec}
|
|
if s["type_id"] in BREAK_ENDS:
|
|
ins.append(dict(c, kind=2, duration_ms=0))
|
|
elif s["type_id"] in BREAK_STARTS:
|
|
outs.append(dict(c, kind=1, duration_ms=dur))
|
|
elif s["type_id"] == CHAPTER_START:
|
|
chapters.append(dict(c, kind=3, duration_ms=dur))
|
|
return {"opid": 0x0104, "splice_us": S, "section": sec, "cues": ins + outs + chapters, "withdraws": withdraws}, None
|
|
if data:
|
|
raise Refused(114, "message_size")
|
|
return {"opid": 0x0102, "splice_us": P, "section": section(tier, 0x00, b"", descs), "cues": [], "withdraws": []}, None
|
|
|
|
|
|
def translate(msg, received, prm):
|
|
"""§4 to §6. Returns the translation, or raises Refused."""
|
|
r = Reader(msg)
|
|
if r.u(2) != 0xFFFF:
|
|
raise Refused(115, "not_multiple")
|
|
if r.u(2) != len(msg):
|
|
raise Refused(114, "message_size")
|
|
pv, as_index, number, dpi, scte_pv = r.u(1), r.u(1), r.u(1), r.u(2), r.u(1)
|
|
if pv != 0 or scte_pv != 0:
|
|
raise Refused(127, "version_unsupported")
|
|
tt = r.u(1)
|
|
if tt == 0:
|
|
P = received
|
|
elif tt == 1:
|
|
secs, us = r.u(4), r.u(2) << 8
|
|
unix_us = (secs + EPOCH_1980 - prm["leap_s"]) * 1_000_000 + us
|
|
if unix_us < prm["t0_us"]:
|
|
raise Refused(115, "timestamp_before_room")
|
|
P = max(unix_us - prm["t0_us"], received)
|
|
else:
|
|
raise Refused(123, "time_type_unsupported")
|
|
n = r.u(1)
|
|
if n == 0:
|
|
raise Refused(115, "no_operation")
|
|
ops = []
|
|
for _ in range(n):
|
|
opid = r.u(2)
|
|
ops.append((opid, r.take(r.u(2))))
|
|
if r.left():
|
|
raise Refused(114, "message_size")
|
|
groups = []
|
|
for opid, d in ops:
|
|
if opid in NORMAL:
|
|
groups.append([NORMAL[opid], d, []])
|
|
elif opid in SUPPLEMENTAL:
|
|
if not groups:
|
|
raise Refused(115, "supplemental_first")
|
|
groups[-1][2].append((opid, d))
|
|
elif opid in UNIMPLEMENTED:
|
|
raise Refused(124, "operation_unsupported", opid)
|
|
else:
|
|
raise Refused(125, "opid_unknown", opid)
|
|
reqs, warning = [], None
|
|
for name, d, sups in groups:
|
|
q, w = request(name, d, sups, P, prm)
|
|
warning = warning or w
|
|
reqs.append(q)
|
|
if sum(len(q["cues"]) for q in reqs) > 255:
|
|
raise Refused(115, "too_many_cues")
|
|
return {"number": number, "as_index": as_index, "dpi": dpi, "processed_us": P, "warning": warning, "requests": reqs}
|
|
|
|
|
|
def cues_of(tr):
|
|
return [c for q in tr["requests"] for c in q["cues"]]
|
|
|
|
|
|
def injection_record(received, prm, index, msg):
|
|
"""§7.1."""
|
|
return b"WIJ1\x01" + struct.pack("<QBIIBH", received, prm["leap_s"], prm["fps_num"], prm["fps_den"], index, len(msg)) + msg
|
|
|
|
|
|
# ── §7.3: the injector's rules ──────────────────────────────────────────────────
|
|
|
|
class Injector:
|
|
def __init__(self, prm):
|
|
self.prm = prm
|
|
self.held = [] # (cue, message, received, index, number, from_time_signal)
|
|
self.open = None
|
|
self.last_end = 0
|
|
self.outstanding = {}
|
|
self.counts = {}
|
|
self.signed = [] # (t, kind, payload)
|
|
|
|
def count(self, k, n=1):
|
|
self.counts[k] = self.counts.get(k, 0) + n
|
|
|
|
def accept(self, msg, received):
|
|
tr = translate(msg, received, self.prm)
|
|
if tr["number"] in self.outstanding:
|
|
self.count("repeated_message")
|
|
return 100
|
|
result = tr["warning"] or 100
|
|
for q in tr["requests"]:
|
|
for kind, eid in q["withdraws"]:
|
|
before = len(self.held)
|
|
if kind == "splice":
|
|
self.held = [h for h in self.held if not (h[0]["cue_id"] == eid and not h[5])]
|
|
elif kind == "segmentation":
|
|
self.held = [h for h in self.held if not (h[0]["cue_id"] == eid and h[5])]
|
|
else:
|
|
self.held = [h for h in self.held if not (h[0]["cue_id"] == eid and h[0]["derived"])]
|
|
gone = before - len(self.held)
|
|
if gone == 0 and kind != "auto_return":
|
|
self.count("cancel_too_late")
|
|
result = 121
|
|
if gone:
|
|
self.count("withdrawn", gone)
|
|
i = 0
|
|
for q in tr["requests"]:
|
|
for c in q["cues"]:
|
|
self.held.append((c, msg, received, i, tr["number"], q["opid"] == 0x0104))
|
|
i += 1
|
|
if i:
|
|
self.outstanding[tr["number"]] = True
|
|
return result
|
|
|
|
def due(self, now):
|
|
due = sorted([h for h in self.held if h[0]["t"] <= now], key=lambda h: (h[0]["t"], h[2], h[3]))
|
|
self.held = [h for h in self.held if h[0]["t"] > now]
|
|
for c, msg, received, index, number, _ in due:
|
|
k = c["kind"]
|
|
if k == 1:
|
|
if self.open is not None or c["t"] < self.last_end:
|
|
self.count("break_overlap")
|
|
continue
|
|
self.open = (c["cue_id"], c["t"])
|
|
elif k == 2:
|
|
if self.open is None or self.open[0] != c["cue_id"]:
|
|
self.count("cue_in_unmatched")
|
|
continue
|
|
if (c["t"] - self.open[1]) // 1000 == 0:
|
|
self.count("break_zero_duration")
|
|
continue
|
|
self.last_end = c["t"]
|
|
self.open = None
|
|
self.signed.append((c["t"], 0x01, injection_record(received, self.prm, index, msg)))
|
|
self.signed.append((c["t"], 0x02, cue_payload(c)))
|
|
alive = {h[4] for h in self.held}
|
|
for m in list(self.outstanding):
|
|
if m not in alive:
|
|
del self.outstanding[m]
|
|
|
|
|
|
# ── Events (live-events §3) and §8 ───────────────────────────────────────────────
|
|
|
|
def read_event(e):
|
|
"""An event without an anchor: (room, principal, seq, prev, t, kind, payload, body, sig)."""
|
|
if len(e) < 121 + 64 or e[:5] != b"WEV1\x01" or e[117] != 0:
|
|
return None
|
|
n = struct.unpack_from("<H", e, 119)[0]
|
|
if len(e) != 121 + n + 64:
|
|
return None
|
|
return {"room": e[5:37], "principal": e[37:69], "seq": struct.unpack_from("<Q", e, 69)[0], "prev": e[77:109],
|
|
"t": struct.unpack_from("<Q", e, 109)[0], "kind": e[118], "payload": e[121:121 + n], "body": e[:121 + n], "sig": e[121 + n:]}
|
|
|
|
|
|
def verify_chain(events, principal, prm):
|
|
"""§8 over an injector's chain: None, or (index, reason)."""
|
|
prev = bytes(32)
|
|
evs = []
|
|
for i, raw in enumerate(events):
|
|
e = read_event(raw)
|
|
if e is None or e["principal"] != principal:
|
|
return (i, "malformed")
|
|
if ED.verify(principal, EVENT_DOMAIN + e["body"], e["sig"]) is not None:
|
|
return (i, "bad_signature")
|
|
if e["seq"] != i or e["prev"] != prev:
|
|
return (i, "chain_broken")
|
|
prev = hashlib.sha256(e["body"]).digest()
|
|
evs.append(e)
|
|
i = 0
|
|
while i < len(evs):
|
|
e = evs[i]
|
|
if e["kind"] == 0x01 and e["payload"][:4] == b"WIJ1":
|
|
p = e["payload"]
|
|
if len(p) < 25 or p[4] != 1:
|
|
return (i, "injection_malformed")
|
|
received, leap, fn, fd, index, mlen = struct.unpack_from("<QBIIBH", p, 5)
|
|
if fn == 0 or fd == 0 or len(p) != 25 + mlen:
|
|
return (i, "injection_malformed")
|
|
try:
|
|
tr = translate(p[25:], received, {"t0_us": prm["t0_us"], "leap_s": leap, "fps_num": fn, "fps_den": fd})
|
|
except Refused:
|
|
return (i, "injection_untranslatable")
|
|
if i + 1 >= len(evs) or evs[i + 1]["kind"] != 0x02:
|
|
return (i, "injection_unanswered")
|
|
cs = cues_of(tr)
|
|
nxt = evs[i + 1]
|
|
if index >= len(cs) or nxt["payload"] != cue_payload(cs[index]) or nxt["t"] != cs[index]["t"] or e["t"] != cs[index]["t"]:
|
|
return (i + 1, "injection_mismatch")
|
|
i += 2
|
|
continue
|
|
if e["kind"] == 0x02:
|
|
return (i, "injection_missing")
|
|
i += 1
|
|
return None
|
|
|
|
|
|
def breaks_of(events):
|
|
"""live-ads §4.3 over one chain's cues: (breaks, chapters)."""
|
|
open_, breaks, chapters = None, [], []
|
|
for raw in events:
|
|
e = read_event(raw)
|
|
if e["kind"] != 0x02:
|
|
continue
|
|
ty, cid, dur = e["payload"][0], struct.unpack_from("<I", e["payload"], 1)[0], struct.unpack_from("<I", e["payload"], 5)[0]
|
|
if ty == 1:
|
|
assert open_ is None, "an out while a break is open"
|
|
open_ = (cid, e["t"])
|
|
elif ty == 2:
|
|
assert open_ is not None and open_[0] == cid, "an in that closes nothing"
|
|
d = (e["t"] - open_[1]) // 1000
|
|
assert d > 0 and (dur == 0 or dur == d), "cue_duration_mismatch"
|
|
breaks.append({"cue_id": cid, "out_t": open_[1], "in_t": e["t"], "duration_ms": d})
|
|
open_ = None
|
|
elif ty == 3:
|
|
chapters.append({"cue_id": cid, "t": e["t"], "duration_ms": dur})
|
|
return {"breaks": breaks, "chapters": chapters}
|
|
|
|
|
|
# ── The corpus ───────────────────────────────────────────────────────────────────
|
|
|
|
def hx(b):
|
|
return None if b is None else b.hex()
|
|
|
|
|
|
def check(v):
|
|
"""Every check of the corpus `v`: (failures, checks)."""
|
|
prm = v["params"]
|
|
principal = bytes.fromhex(v["principal"])
|
|
fails, n = [], 0
|
|
|
|
def expect(ok, what):
|
|
nonlocal n
|
|
n += 1
|
|
if not ok:
|
|
fails.append(what)
|
|
|
|
for t in v["translations"]:
|
|
try:
|
|
tr = translate(bytes.fromhex(t["message"]), t["received_us"], prm)
|
|
except Refused as e:
|
|
expect(False, f"{t['name']}: refused {e.reason}")
|
|
continue
|
|
expect(tr["processed_us"] == t["processed_us"] and tr["warning"] == t["warning"], f"{t['name']}: processing")
|
|
expect(len(tr["requests"]) == len(t["requests"]), f"{t['name']}: requests")
|
|
for mine, want in zip(tr["requests"], t["requests"]):
|
|
expect(mine["opid"] == want["opid"] and mine["splice_us"] == want["splice_us"], f"{t['name']}: request")
|
|
expect(hx(mine["section"]) == want["section"], f"{t['name']}: section")
|
|
if mine["section"] is not None:
|
|
expect(crc32_mpeg2(mine["section"]) == 0, f"{t['name']}: CRC over the section")
|
|
expect(mine["withdraws"] == want["withdraws"], f"{t['name']}: withdrawals")
|
|
cs = [{"t": c["t"], "kind": c["kind"], "cue_id": c["cue_id"], "duration_ms": c["duration_ms"], "derived": c["derived"], "section": hx(c["section"]), "payload": cue_payload(c).hex()} for c in mine["cues"]]
|
|
expect(cs == want["cues"], f"{t['name']}: cues")
|
|
if "standard_section" in t:
|
|
expect(hx(tr["requests"][0]["section"]) == t["standard_section"], f"{t['name']}: the standard's sample, byte for byte")
|
|
for r in v["refusals"]:
|
|
try:
|
|
translate(bytes.fromhex(r["message"]), r["received_us"], prm)
|
|
expect(False, f"{r['name']}: accepted")
|
|
except Refused as e:
|
|
expect((e.result, e.ext, e.reason) == (r["result"], r["ext"], r["reason"]), f"{r['name']}: {e.result} {e.ext} {e.reason}")
|
|
for s in v["scenarios"]:
|
|
inj = Injector(prm)
|
|
try:
|
|
for m in s["messages"]:
|
|
inj.due(m["received_us"])
|
|
res = inj.accept(bytes.fromhex(m["message"]), m["received_us"])
|
|
expect(res == m["result"], f"{s['name']}: a message's result {res}")
|
|
inj.due(s["until_us"])
|
|
except Refused as e:
|
|
expect(False, f"{s['name']}: a message refused {e.reason}")
|
|
continue
|
|
expect(inj.counts == s["counts"], f"{s['name']}: counts {inj.counts}")
|
|
evs = [bytes.fromhex(e) for e in s["events"]]
|
|
expect(len(evs) == len(inj.signed), f"{s['name']}: {len(inj.signed)} events, the corpus {len(evs)}")
|
|
for raw, (t, kind, payload) in zip(evs, inj.signed):
|
|
e = read_event(raw)
|
|
expect(e is not None and (e["t"], e["kind"], e["payload"]) == (t, kind, payload) and e["room"] == bytes.fromhex(v["room"]), f"{s['name']}: an event")
|
|
expect(verify_chain(evs, principal, prm) is None, f"{s['name']}: the chain verifies")
|
|
try:
|
|
b = breaks_of(evs)
|
|
except (AssertionError, TypeError):
|
|
b = None
|
|
expect(b == s["breaks"], f"{s['name']}: breaks {b}")
|
|
for c in v["negative_chains"]:
|
|
got = verify_chain([bytes.fromhex(e) for e in c["events"]], principal, prm)
|
|
expect(got == (c["refused_at"], c["reason"]), f"{c['name']}: {got}")
|
|
return fails, n
|
|
|
|
|
|
def flip(h, i):
|
|
b = bytearray(bytes.fromhex(h))
|
|
b[i % len(b)] ^= 0x01
|
|
return b.hex()
|
|
|
|
|
|
def mutations(v):
|
|
"""Copies of the corpus, each with one stated value changed: each must fail."""
|
|
import copy
|
|
for i, t in enumerate(v["translations"]):
|
|
for q, r in enumerate(t["requests"]):
|
|
if r["section"]:
|
|
m = copy.deepcopy(v)
|
|
m["translations"][i]["requests"][q]["section"] = flip(r["section"], 7 + q)
|
|
yield f"translation {t['name']}: a section byte", m
|
|
for c, cue in enumerate(r["cues"]):
|
|
m = copy.deepcopy(v)
|
|
m["translations"][i]["requests"][q]["cues"][c]["t"] += 1
|
|
yield f"translation {t['name']}: a cue's instant", m
|
|
m = copy.deepcopy(v)
|
|
m["translations"][i]["requests"][q]["cues"][c]["duration_ms"] += 1
|
|
yield f"translation {t['name']}: a cue's duration", m
|
|
if "standard_section" in t:
|
|
m = copy.deepcopy(v)
|
|
m["translations"][i]["standard_section"] = flip(t["standard_section"], -1)
|
|
yield "the standard's sample's CRC", m
|
|
for i, r in enumerate(v["refusals"]):
|
|
m = copy.deepcopy(v)
|
|
m["refusals"][i]["reason"] = "message_size" if r["reason"] != "message_size" else "field_range"
|
|
yield f"refusal {r['name']}: its reason", m
|
|
for i, s in enumerate(v["scenarios"]):
|
|
if s["events"]:
|
|
m = copy.deepcopy(v)
|
|
m["scenarios"][i]["events"] = s["events"][:-2]
|
|
yield f"scenario {s['name']}: its last cue dropped", m
|
|
m = copy.deepcopy(v)
|
|
m["scenarios"][i]["events"][-1] = flip(s["events"][-1], -1)
|
|
yield f"scenario {s['name']}: a signature", m
|
|
if s["breaks"]["breaks"]:
|
|
m = copy.deepcopy(v)
|
|
m["scenarios"][i]["breaks"]["breaks"][0]["in_t"] += 1000
|
|
yield f"scenario {s['name']}: a break's end", m
|
|
m = copy.deepcopy(v)
|
|
m["scenarios"][i]["counts"]["withdrawn"] = 99
|
|
yield f"scenario {s['name']}: a count", m
|
|
for i, c in enumerate(v["negative_chains"]):
|
|
m = copy.deepcopy(v)
|
|
m["negative_chains"][i]["refused_at"] += 1
|
|
yield f"negative {c['name']}: where it is refused", m
|
|
|
|
|
|
def main(args):
|
|
self_test = "--self-test" in args
|
|
args = [a for a in args if a != "--self-test"]
|
|
d = args[0] if args else os.path.join(HERE, "injection")
|
|
v = json.load(open(os.path.join(d, "vectors.json")))
|
|
fails, n = check(v)
|
|
for f in fails:
|
|
print("FAIL", f)
|
|
print(f"injection: {n - len(fails)} of {n} checks agree "
|
|
f"({len(v['translations'])} translations, {len(v['refusals'])} refusals, {len(v['scenarios'])} scenarios, {len(v['negative_chains'])} negative chains)")
|
|
if fails:
|
|
return 1
|
|
if self_test:
|
|
survived = [what for what, m in mutations(v) if not check(m)[0]]
|
|
total = sum(1 for _ in mutations(v))
|
|
for w in survived:
|
|
print("SURVIVED", w)
|
|
print(f"injection self-test: {total - len(survived)} of {total} mutated corpora refused")
|
|
return 1 if survived else 0
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv[1:]))
|