617 lines
26 KiB
Python
617 lines
26 KiB
Python
#!/usr/bin/env python3
|
||
"""Independent verifier of the stream-termination corpus (extensions/stream-termination.md,
|
||
and the display match of extensions/capability-constraints.md §4).
|
||
|
||
Written from the two extensions' text. It shares no code with wai-rs, which generates the
|
||
corpus, and never runs it. It checks Ed25519 signatures with
|
||
../claim-conformance/_ed25519.py, WAI's strict rule (SPEC §7 *Signatures*) written from
|
||
RFC 8032 with the standard library, and SHA-256 with hashlib. Everything else is here:
|
||
|
||
- the display match (capability-constraints §4.2): a rendition's exceedances, the least
|
||
rendition, the admissible set and the match a receipt records;
|
||
- the WSL1 records (§3): their layout, reading and rules (§3.9) and their signature (§3.8);
|
||
- a log's verification (§6.1), the state at an instant (§6.2) and the facts an accepted log
|
||
states (§6.3): the totals and unmetered counts by role and class, the absent roles, the
|
||
panel counts, the unseen time and the unwatched segments.
|
||
|
||
For every match it requires the exceedances, match, admissible set and least rendition the
|
||
corpus states; for every log the outcome, and the reason and index of a refusal or every fact
|
||
of an acceptance. Of the corpus, that every reason of §7 refuses a log and that §7 lists
|
||
exactly those reasons, and that capability-constraints §4.2's table names the four bits this
|
||
verifier sets. Then it mutates copies of the corpus, one defect each, and requires every copy
|
||
to fail.
|
||
|
||
python3 stream_termination_verify.py <live-conformance/termination>
|
||
python3 stream_termination_verify.py --log <log.json> one log a sink signed
|
||
|
||
Exit status 0 only if every check holds and every mutated copy fails. 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, os.path.join(HERE, "..", "claim-conformance"))
|
||
import _ed25519 as ED # noqa: E402
|
||
|
||
MAGIC = b"WSL1"
|
||
DOMAIN = b"wai:stream-log\x01"
|
||
OPEN, SEGMENT, OBSERVATION, DISPLAY, TERMINATE = 1, 2, 3, 4, 5
|
||
VIDEO, AUDIO = 1, 2
|
||
USER_START, AUTOPLAY_START = 1, 2
|
||
SETTING_OFF = 2
|
||
DISALLOWED = 3
|
||
VISIBILITY, OCCLUSION, DISPLAY_POWER, INTERACTION = 1, 2, 3, 4
|
||
SEEN, UNSEEN = 1, 2
|
||
HIDDEN, OCCLUDED, DISPLAY_OFF, IDLE, AUTOPLAY_UNATTENDED, USER, ENDED = 1, 2, 3, 4, 5, 6, 7
|
||
WHY = {1: "hidden", 2: "occluded", 3: "display_off", 4: "idle", 5: "autoplay_unattended", 6: "user", 7: "ended"}
|
||
ROLE = {1: "origin", 2: "relay", 3: "sink_decode", 4: "display"}
|
||
LABEL = {2: "ModelBased", 3: "Estimator", 4: "OnChipCounter", 5: "CalibratedInstrument"}
|
||
NO_RENDITION = 255
|
||
FLOOR = 288676
|
||
RESOLUTION, REFRESH, BIT_DEPTH, DYNAMIC_RANGE, NONE_WITHIN = 1, 2, 4, 8, 16
|
||
HDR_TRANSFERS = (16, 18)
|
||
|
||
REASONS = [
|
||
"reason_unknown", "display_invalid", "record_invalid", "after_termination", "order_invalid",
|
||
"sink_changed", "autoplay_against_setting", "media_order_invalid", "rendition_unknown",
|
||
"match_mismatch", "rendition_above_panel", "reason_not_applicable", "reason_unsupported",
|
||
"totals_mismatch",
|
||
]
|
||
BORROWED = ["truncated", "malformed", "unknown_version", "bad_key", "bad_signature", "energy_invalid"]
|
||
|
||
|
||
class Stop(Exception):
|
||
def __init__(self, reason):
|
||
super().__init__(reason)
|
||
self.reason = reason
|
||
|
||
|
||
# ------------------------------------------------- capability-constraints §4.2: the match
|
||
|
||
def exceedances(r, d):
|
||
"""Bits 0 to 3 of §4.2's table: r = (w, h, rate, depth, transfer), d a declaration dict."""
|
||
w, h, rate, depth, transfer = r
|
||
x = 0
|
||
if max(w, h) > d["long"] or min(w, h) > d["short"]:
|
||
x |= RESOLUTION
|
||
if rate > d["refresh_milli"]:
|
||
x |= REFRESH
|
||
if depth > d["bit_depth"]:
|
||
x |= BIT_DEPTH
|
||
if transfer in HDR_TRANSFERS and not d["hdr"]:
|
||
x |= DYNAMIC_RANGE
|
||
return x
|
||
|
||
|
||
def least(ladder):
|
||
"""The first in the order of (w × h, r, d, 1 for PQ or HLG), then the first listed."""
|
||
if not ladder:
|
||
return None
|
||
key = lambda i: (ladder[i][0] * ladder[i][1], ladder[i][2], ladder[i][3], 1 if ladder[i][4] in HDR_TRANSFERS else 0, i)
|
||
return min(range(len(ladder)), key=key)
|
||
|
||
|
||
def admissible(ladder, d):
|
||
within = [i for i, r in enumerate(ladder) if exceedances(r, d) == 0]
|
||
return within if within else ([least(ladder)] if ladder else [])
|
||
|
||
|
||
def match(ladder, i, d):
|
||
none = all(exceedances(r, d) != 0 for r in ladder)
|
||
return exceedances(ladder[i], d) | (NONE_WITHIN if none else 0)
|
||
|
||
|
||
def rendition_of(j):
|
||
return (j["width"], j["height"], j["frame_rate_milli"], j["bit_depth"], j.get("transfer", 1))
|
||
|
||
|
||
# ------------------------------------------------------------------ §3: the records
|
||
|
||
class Reader:
|
||
def __init__(self, b):
|
||
self.b, self.p = b, 0
|
||
|
||
def take(self, n):
|
||
if self.p + n > len(self.b):
|
||
raise Stop("truncated")
|
||
s = self.b[self.p:self.p + n]
|
||
self.p += n
|
||
return s
|
||
|
||
def u8(self):
|
||
return self.take(1)[0]
|
||
|
||
def u32(self):
|
||
return struct.unpack("<I", self.take(4))[0]
|
||
|
||
def u64(self):
|
||
return struct.unpack("<Q", self.take(8))[0]
|
||
|
||
|
||
def read_declaration(r):
|
||
long_, short, refresh = r.u32(), r.u32(), r.u32()
|
||
depth, hdr = r.u8(), r.u8()
|
||
long_mm, short_mm = r.u32(), r.u32()
|
||
return {"long": long_, "short": short, "refresh_milli": refresh, "bit_depth": depth, "hdr": hdr,
|
||
"long_mm": long_mm, "short_mm": short_mm}
|
||
|
||
|
||
def read_class(r):
|
||
"""energy-measurement §2.7's tag and evidence; a tag other than 4 and 5 has none."""
|
||
tag = r.u8()
|
||
if tag == 4:
|
||
ppm, window, filtering = struct.unpack(">IQB", r.take(13))
|
||
return (4, ppm, window, filtering)
|
||
if tag == 5:
|
||
ppm, window, n = struct.unpack(">IQI", r.take(16))
|
||
if n > 255:
|
||
raise Stop("malformed")
|
||
return (5, ppm, window, r.take(n))
|
||
return (tag,)
|
||
|
||
|
||
def energy_ok(joules, cls):
|
||
"""live-metering §3.6's energy_invalid rules, as §3.9 step 2 applies them."""
|
||
tag = cls[0]
|
||
if tag not in (0, 2, 3, 4, 5):
|
||
return False
|
||
if (tag == 0) != (joules == 0):
|
||
return False
|
||
if tag == 4:
|
||
_, ppm, window, filtering = cls
|
||
return ppm > 0 and window > 0 and filtering in (0, 1, 2) and joules * ppm >= FLOOR
|
||
if tag == 5:
|
||
_, ppm, window, ref = cls
|
||
try:
|
||
text = ref.decode("utf-8")
|
||
except UnicodeDecodeError:
|
||
return False
|
||
return ppm > 0 and window > 0 and text.strip() != "" and joules * ppm >= FLOOR
|
||
return True
|
||
|
||
|
||
def declaration_ok(d):
|
||
return 1 <= d["short"] <= d["long"] and d["refresh_milli"] >= 1 and 1 <= d["bit_depth"] <= 16 and d["hdr"] in (0, 1)
|
||
|
||
|
||
def read_record(b):
|
||
"""§3.9: the record, its rules and its signature; raises Stop with the reason."""
|
||
if len(b) < 4:
|
||
raise Stop("truncated")
|
||
if b[:4] != MAGIC:
|
||
raise Stop("malformed")
|
||
if len(b) < 5:
|
||
raise Stop("truncated")
|
||
if b[4] != 1:
|
||
raise Stop("unknown_version")
|
||
r = Reader(b)
|
||
r.take(5)
|
||
kind = r.u8()
|
||
if kind not in (1, 2, 3, 4, 5):
|
||
raise Stop("malformed")
|
||
rec = {"kind": kind, "sink": r.take(32), "seq": r.u32(), "prev": r.take(32), "t": r.u64()}
|
||
if kind == OPEN:
|
||
rec["stream"] = r.take(32)
|
||
rec["media"], rec["start"], rec["autoplay"], rec["ua_policy"] = r.take(4)
|
||
rec["unseen_grace_ms"], rec["idle_ms"], rec["autoplay_grace_ms"] = r.u32(), r.u32(), r.u32()
|
||
rec["display_bytes"] = b[r.p:r.p + 22]
|
||
rec["display"] = read_declaration(r)
|
||
rec["ladder"] = [(r.u32(), r.u32(), r.u32(), r.u8(), r.u8()) for _ in range(r.u8())]
|
||
elif kind == SEGMENT:
|
||
rec["segment"] = r.take(32)
|
||
rec["objects"], rec["bytes"] = r.u32(), r.u64()
|
||
rec["media_start"], rec["media_end"] = r.u64(), r.u64()
|
||
rec["rendition"], rec["match"] = r.u8(), r.u8()
|
||
hops = []
|
||
for _ in range(r.u8()):
|
||
role, joules = r.u8(), r.u64()
|
||
hops.append((role, joules, read_class(r)))
|
||
rec["hops"] = hops
|
||
elif kind == OBSERVATION:
|
||
rec["signal"], rec["value"] = r.u8(), r.u8()
|
||
elif kind == DISPLAY:
|
||
rec["display"] = read_declaration(r)
|
||
else:
|
||
rec["reason"], rec["media"], rec["segments"] = r.u8(), r.u64(), r.u32()
|
||
rec["totals"] = [(r.u8(), r.u8(), r.u32(), r.u64()) for _ in range(r.u8())]
|
||
rec["unmetered"] = [(r.u8(), r.u32()) for _ in range(r.u8())]
|
||
rec["sig"] = r.take(64)
|
||
if r.p != len(b):
|
||
raise Stop("malformed")
|
||
# Step 1.
|
||
if kind == TERMINATE and rec["reason"] not in WHY:
|
||
raise Stop("reason_unknown")
|
||
# Step 2.
|
||
if kind == SEGMENT and not all(energy_ok(j, c) for _, j, c in rec["hops"]):
|
||
raise Stop("energy_invalid")
|
||
# Step 3.
|
||
if kind == OPEN:
|
||
if rec["media"] == AUDIO:
|
||
if rec["display_bytes"] != bytes(22):
|
||
raise Stop("display_invalid")
|
||
elif not declaration_ok(rec["display"]):
|
||
raise Stop("display_invalid")
|
||
if kind == DISPLAY and not declaration_ok(rec["display"]):
|
||
raise Stop("display_invalid")
|
||
# Step 4.
|
||
if not rules_hold(rec):
|
||
raise Stop("record_invalid")
|
||
bad = verify_sig(rec["sink"], DOMAIN + b[:-64], rec["sig"])
|
||
if bad:
|
||
raise Stop(bad)
|
||
return rec
|
||
|
||
|
||
_SIGS = {}
|
||
|
||
|
||
def verify_sig(key, msg, sig):
|
||
"""ED.verify, remembered: the mutated copies of the corpus share most of their records,
|
||
and the rule is a function of its three inputs."""
|
||
k = (key, hashlib.sha256(msg).digest(), sig)
|
||
if k not in _SIGS:
|
||
_SIGS[k] = ED.verify(key, msg, sig)
|
||
return _SIGS[k]
|
||
|
||
|
||
def ascending(keys):
|
||
return all(a < b for a, b in zip(keys, keys[1:]))
|
||
|
||
|
||
def rules_hold(rec):
|
||
k = rec["kind"]
|
||
if k == OPEN:
|
||
if not (rec["media"] in (1, 2) and rec["start"] in (1, 2) and rec["autoplay"] in (1, 2) and rec["ua_policy"] <= 3):
|
||
return False
|
||
if rec["media"] == VIDEO:
|
||
return 1 <= len(rec["ladder"]) <= 32 and all(w >= 1 and h >= 1 and rate >= 1 and 1 <= d <= 16 for w, h, rate, d, _ in rec["ladder"])
|
||
return rec["ladder"] == []
|
||
if k == SEGMENT:
|
||
roles = [h[0] for h in rec["hops"]]
|
||
return (rec["objects"] >= 1 and rec["media_start"] < rec["media_end"] and rec["match"] < 32
|
||
and len(roles) <= 4 and all(1 <= x <= 4 for x in roles) and ascending(roles))
|
||
if k == OBSERVATION:
|
||
if rec["signal"] in (VISIBILITY, OCCLUSION, DISPLAY_POWER):
|
||
return rec["value"] in (1, 2)
|
||
return rec["signal"] == INTERACTION and rec["value"] == 1
|
||
if k == TERMINATE:
|
||
t = rec["totals"]
|
||
u = rec["unmetered"]
|
||
return (all(1 <= ro <= 4 and 2 <= tg <= 5 and f >= 1 for ro, tg, f, _ in t) and ascending([(x[0], x[1]) for x in t])
|
||
and all(1 <= ro <= 4 and c >= 1 for ro, c in u) and ascending([x[0] for x in u]))
|
||
return True
|
||
|
||
|
||
# ------------------------------------------------------------------ §6: verification
|
||
|
||
def sums_of(segments):
|
||
totals, unmetered = {}, {}
|
||
for s in segments:
|
||
for role, joules, cls in s["hops"]:
|
||
if joules == 0:
|
||
unmetered[role] = unmetered.get(role, 0) + 1
|
||
else:
|
||
f, j = totals.get((role, cls[0]), (0, 0))
|
||
totals[(role, cls[0])] = (f + 1, j + joules)
|
||
return totals, unmetered
|
||
|
||
|
||
def sums_json(totals, unmetered):
|
||
return ([{"role": ROLE[r], "class": LABEL[g], "figures": f, "joules": j} for (r, g), (f, j) in sorted(totals.items())],
|
||
[{"role": ROLE[r], "count": c} for r, c in sorted(unmetered.items())])
|
||
|
||
|
||
def digest(b):
|
||
return hashlib.sha256(b[:-64]).digest()
|
||
|
||
|
||
def verify_log(blobs):
|
||
"""§6.1: ("refuse", reason, index) or ("accept", facts)."""
|
||
if not blobs:
|
||
return ("refuse", "order_invalid", 0)
|
||
o = None
|
||
prev, prev_t = bytes(32), 0
|
||
ended = None
|
||
obs = {VISIBILITY: [], OCCLUSION: [], DISPLAY_POWER: []}
|
||
last_interaction = None
|
||
unseen = False
|
||
unseen_since = None
|
||
unseen_us = 0
|
||
decl = None
|
||
segments, unwatched = [], []
|
||
within = none_within = 0
|
||
last_t = 0
|
||
|
||
def value(s):
|
||
return obs[s][-1][0] if obs[s] else SEEN
|
||
|
||
def since(s):
|
||
"""The t of the earliest observation of the trailing run of one value, or T0."""
|
||
if not obs[s]:
|
||
return o["t"]
|
||
v = obs[s][-1][0]
|
||
t = obs[s][-1][1]
|
||
for vv, tt in reversed(obs[s]):
|
||
if vv != v:
|
||
break
|
||
t = tt
|
||
return t
|
||
|
||
def is_unseen():
|
||
return any(value(s) == UNSEEN for s in obs)
|
||
|
||
for i, b in enumerate(blobs):
|
||
try:
|
||
rec = read_record(b)
|
||
except Stop as e:
|
||
return ("refuse", e.reason, i)
|
||
if ended is not None:
|
||
return ("refuse", "after_termination", i)
|
||
first = i == 0
|
||
if (rec["kind"] == OPEN) != first or rec["seq"] != i or rec["prev"] != prev or (not first and rec["t"] < prev_t):
|
||
return ("refuse", "order_invalid", i)
|
||
if not first and rec["sink"] != o["sink"]:
|
||
return ("refuse", "sink_changed", i)
|
||
prev, prev_t, last_t = digest(b), rec["t"], rec["t"]
|
||
k = rec["kind"]
|
||
if k == OPEN:
|
||
if rec["start"] == AUTOPLAY_START and (rec["autoplay"] == SETTING_OFF or rec["ua_policy"] == DISALLOWED):
|
||
return ("refuse", "autoplay_against_setting", i)
|
||
o = rec
|
||
decl = rec["display"]
|
||
elif k == DISPLAY:
|
||
if o["media"] == AUDIO:
|
||
return ("refuse", "display_invalid", i)
|
||
decl = rec["display"]
|
||
elif k == OBSERVATION:
|
||
if rec["signal"] == INTERACTION:
|
||
last_interaction = rec["t"]
|
||
else:
|
||
obs[rec["signal"]].append((rec["value"], rec["t"]))
|
||
now = is_unseen()
|
||
if now and not unseen:
|
||
unseen_since = rec["t"]
|
||
elif unseen and not now:
|
||
unseen_us += rec["t"] - unseen_since
|
||
unseen_since = None
|
||
unseen = now
|
||
elif k == SEGMENT:
|
||
if segments and rec["media_start"] < segments[-1]["media_end"]:
|
||
return ("refuse", "media_order_invalid", i)
|
||
if o["media"] == VIDEO:
|
||
ladder = o["ladder"]
|
||
if rec["rendition"] >= len(ladder):
|
||
return ("refuse", "rendition_unknown", i)
|
||
if rec["match"] != match(ladder, rec["rendition"], decl):
|
||
return ("refuse", "match_mismatch", i)
|
||
if rec["rendition"] not in admissible(ladder, decl):
|
||
return ("refuse", "rendition_above_panel", i)
|
||
within += rec["match"] == 0
|
||
none_within += 1 if rec["match"] & NONE_WITHIN else 0
|
||
if unseen and rec["t"] - unseen_since >= o["unseen_grace_ms"] * 1000:
|
||
unwatched.append(rec)
|
||
else:
|
||
if rec["rendition"] != NO_RENDITION:
|
||
return ("refuse", "rendition_unknown", i)
|
||
if rec["match"] != 0:
|
||
return ("refuse", "match_mismatch", i)
|
||
segments.append(rec)
|
||
else:
|
||
if segments and not (segments[-1]["media_start"] <= rec["media"] <= segments[-1]["media_end"]):
|
||
return ("refuse", "media_order_invalid", i)
|
||
if o["media"] == AUDIO and rec["reason"] in (HIDDEN, OCCLUDED, DISPLAY_OFF):
|
||
return ("refuse", "reason_not_applicable", i)
|
||
if not supports(o, rec["reason"], rec["t"], value, since, last_interaction):
|
||
return ("refuse", "reason_unsupported", i)
|
||
totals, unmetered = sums_of(segments)
|
||
stated_t = {(ro, tg): (f, j) for ro, tg, f, j in rec["totals"]}
|
||
stated_u = dict(rec["unmetered"])
|
||
if (rec["segments"] != len(segments) or stated_t != totals or stated_u != unmetered
|
||
or len(stated_t) != len(rec["totals"]) or len(stated_u) != len(rec["unmetered"])):
|
||
return ("refuse", "totals_mismatch", i)
|
||
ended = rec
|
||
video = o["media"] == VIDEO
|
||
totals, unmetered = sums_of(segments)
|
||
tj, uj = sums_json(totals, unmetered)
|
||
present = {h[0] for s in segments for h in s["hops"]}
|
||
facts = {
|
||
"outcome": "accept",
|
||
"record_count": len(blobs),
|
||
"segments": len(segments),
|
||
"media": "video" if video else "audio",
|
||
"terminated": WHY[ended["reason"]] if ended else None,
|
||
"media_end": ended["media"] if ended else None,
|
||
"totals": tj,
|
||
"unmetered": uj,
|
||
"absent": [ROLE[r] for r in (1, 2, 3, 4) if r not in present],
|
||
"panel": {"within": within, "none_within": none_within} if video else None,
|
||
"unseen_us": (unseen_us + (last_t - unseen_since if unseen else 0)) if video else None,
|
||
"unwatched": None,
|
||
}
|
||
if video:
|
||
wt, wu = sums_json(*sums_of(unwatched))
|
||
facts["unwatched"] = {"segments": len(unwatched), "totals": wt, "unmetered": wu}
|
||
return ("accept", facts)
|
||
|
||
|
||
def supports(o, reason, t, value, since, last_interaction):
|
||
"""§6.2's table at a terminate record made at t."""
|
||
grace = o["unseen_grace_ms"] * 1000
|
||
if reason == HIDDEN:
|
||
return value(VISIBILITY) == UNSEEN and t - since(VISIBILITY) >= grace
|
||
if reason == OCCLUDED:
|
||
return value(OCCLUSION) == UNSEEN and t - since(OCCLUSION) >= grace
|
||
if reason == DISPLAY_OFF:
|
||
return value(DISPLAY_POWER) == UNSEEN and t - since(DISPLAY_POWER) >= grace
|
||
if reason == IDLE:
|
||
last = last_interaction if last_interaction is not None else o["t"]
|
||
return o["idle_ms"] > 0 and t - last >= o["idle_ms"] * 1000
|
||
if reason == AUTOPLAY_UNATTENDED:
|
||
return (o["start"] == AUTOPLAY_START and o["autoplay_grace_ms"] > 0 and last_interaction is None
|
||
and t - o["t"] >= o["autoplay_grace_ms"] * 1000)
|
||
return True
|
||
|
||
|
||
# ------------------------------------------------------------------ the corpus
|
||
|
||
def check(doc, quiet=False):
|
||
fails = []
|
||
for m in doc["matches"]:
|
||
d = dict(m["declaration"])
|
||
d["hdr"] = 1 if d["hdr"] is True else 0 if d["hdr"] is False else d["hdr"]
|
||
ladder = [rendition_of(x) for x in m["ladder"]]
|
||
i = m["rendition"]
|
||
got = {"exceedances": exceedances(ladder[i], d), "match": match(ladder, i, d), "admissible": admissible(ladder, d), "least": least(ladder)}
|
||
for k, v in got.items():
|
||
if m.get(k) != v:
|
||
fails.append("match %s: %s is %r here, %r in the corpus" % (m["name"], k, v, m.get(k)))
|
||
for g in doc["logs"]:
|
||
try:
|
||
blobs = [bytes.fromhex(x) for x in g["records"]]
|
||
except (ValueError, TypeError):
|
||
fails.append("log %s: records do not read as hex" % g["name"])
|
||
continue
|
||
res = verify_log(blobs)
|
||
if res[0] == "refuse":
|
||
want = {"outcome": "refuse", "reason": res[1], "at": res[2]}
|
||
else:
|
||
want = res[1]
|
||
for k, v in want.items():
|
||
if g.get(k) != v:
|
||
fails.append("log %s: %s is %r here, %r in the corpus" % (g["name"], k, v, g.get(k)))
|
||
if res[0] == "refuse" and any(k in g for k in ("totals", "segments", "panel")):
|
||
fails.append("log %s: a refused log states facts" % g["name"])
|
||
if not quiet:
|
||
print("%s every match and log gets the outcome the corpus states (%d matches, %d logs)"
|
||
% ("PASS" if not fails else "FAIL", len(doc["matches"]), len(doc["logs"])))
|
||
return fails
|
||
|
||
|
||
def section(path, start):
|
||
with open(path, encoding="utf-8") as f:
|
||
text = f.read()
|
||
at = text.index(start)
|
||
end = text.find("\n## ", at + 1)
|
||
nxt = text.find("\n### ", at + 1)
|
||
return text[at:end if end != -1 else len(text)], text[at:nxt if nxt != -1 else len(text)]
|
||
|
||
|
||
def reasons_listed():
|
||
sec, _ = section(os.path.join(HERE, "..", "extensions", "stream-termination.md"), "\n## 7. Reasons")
|
||
return [m.group(1) for m in re.finditer(r"^\| `([a-z_]+)` \|", sec, re.M)]
|
||
|
||
|
||
def match_bits_listed():
|
||
_, sec = section(os.path.join(HERE, "..", "extensions", "capability-constraints.md"), "\n### 4.2 The match")
|
||
return [(int(m.group(1)), m.group(2)) for m in re.finditer(r"^\| (\d) \| `([a-z_]+)` \|", sec, re.M)]
|
||
|
||
|
||
def flip(h, at):
|
||
b = bytearray(bytes.fromhex(h))
|
||
b[at] ^= 1
|
||
return b.hex()
|
||
|
||
|
||
def mutations(doc):
|
||
"""Copies of the corpus, each with one defect, by name."""
|
||
out = []
|
||
|
||
def mutate(name, f):
|
||
d = copy.deepcopy(doc)
|
||
f(d)
|
||
out.append((name, d))
|
||
|
||
def log(d, n):
|
||
return next(g for g in d["logs"] if g["name"] == n)
|
||
|
||
def mt(d, n):
|
||
return next(m for m in d["matches"] if m["name"] == n)
|
||
|
||
mutate("a hidden log's terminate record, one byte", lambda d: log(d, "hidden")["records"].__setitem__(-1, flip(log(d, "hidden")["records"][-1], 100)))
|
||
mutate("a hidden log's segment, its signature", lambda d: log(d, "hidden")["records"].__setitem__(1, flip(log(d, "hidden")["records"][1], -1)))
|
||
mutate("a hidden log with its observation left out", lambda d: log(d, "hidden")["records"].pop(3))
|
||
mutate("a hidden log's unwatched count", lambda d: log(d, "hidden")["unwatched"].update(segments=0))
|
||
mutate("a hidden log's unseen time", lambda d: log(d, "hidden").update(unseen_us=10800001))
|
||
mutate("a hidden log's absent roles", lambda d: log(d, "hidden").update(absent=[]))
|
||
mutate("a hidden log's reason", lambda d: log(d, "hidden").update(terminated="user"))
|
||
mutate("the segment after the termination dropped", lambda d: log(d, "segment_after_termination")["records"].pop())
|
||
mutate("a refused log's index", lambda d: log(d, "segment_after_termination").update(at=6))
|
||
mutate("a refused log's reason", lambda d: log(d, "above_panel").update(reason="match_mismatch"))
|
||
mutate("two classes of one hop summed in the facts", lambda d: log(d, "user_stop").update(totals=[t for t in log(d, "user_stop")["totals"] if t["class"] != "CalibratedInstrument"]))
|
||
mutate("a total's joules", lambda d: log(d, "user_stop")["totals"][0].update(joules=log(d, "user_stop")["totals"][0]["joules"] + 1))
|
||
mutate("an unmetered count", lambda d: log(d, "user_stop")["unmetered"][0].update(count=3))
|
||
mutate("the panel counts", lambda d: log(d, "none_within").update(panel={"within": 2, "none_within": 0}))
|
||
mutate("an audio log given unwatched facts", lambda d: log(d, "audio_hidden").update(unwatched={"segments": 0, "totals": [], "unmetered": []}))
|
||
mutate("a match's exceedances", lambda d: mt(d, "every_exceedance").update(exceedances=7))
|
||
mutate("a match's none_within bit", lambda d: mt(d, "none_within_least").update(match=1))
|
||
mutate("a match's admissible set", lambda d: mt(d, "none_within_other").update(admissible=[0, 1]))
|
||
mutate("the least of a tie", lambda d: mt(d, "least_tie").update(least=1))
|
||
mutate("the least by rate", lambda d: mt(d, "least_by_rate").update(least=0))
|
||
mutate("a portrait panel's sides", lambda d: mt(d, "portrait_panel")["declaration"].update(long=1080, short=1080))
|
||
mutate("an HDR panel declared SDR", lambda d: mt(d, "pq_on_hdr")["declaration"].update(hdr=False))
|
||
mutate("the only log refused sink_changed", lambda d: d["logs"].remove(log(d, "sink_changed")))
|
||
mutate("an accepted log's records reordered", lambda d: log(d, "occluded")["records"].reverse())
|
||
mutate("an open log given a termination", lambda d: log(d, "open").update(terminated="ended"))
|
||
return out
|
||
|
||
|
||
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)
|
||
fails += check(doc)
|
||
seen = {g.get("reason") for g in doc["logs"]}
|
||
missing = [r for r in REASONS + BORROWED if r not in seen]
|
||
if missing:
|
||
fails.append("no log is refused %s" % ", ".join(missing))
|
||
print("%s every reason of §7, and every borrowed code, refuses a log" % ("PASS" if not missing else "FAIL"))
|
||
listed = reasons_listed()
|
||
if listed != REASONS:
|
||
fails.append("§7 lists %s" % listed)
|
||
print("%s §7 lists exactly the reasons this verifier knows" % ("PASS" if listed == REASONS else "FAIL"))
|
||
bits = match_bits_listed()
|
||
want_bits = [(0, "resolution"), (1, "refresh"), (2, "bit_depth"), (3, "dynamic_range")]
|
||
if bits != want_bits:
|
||
fails.append("capability-constraints §4.2 lists %s" % bits)
|
||
print("%s capability-constraints §4.2 names the four bits this verifier sets" % ("PASS" if bits == want_bits else "FAIL"))
|
||
for label, mutated in mutations(doc):
|
||
caught = check(mutated, quiet=True)
|
||
if not caught:
|
||
seen2 = {g.get("reason") for g in mutated["logs"]}
|
||
caught = ["reason %s refuses no log" % r for r in REASONS + BORROWED if r not in seen2]
|
||
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)
|
||
print(("PASS" if not fails else "FAIL") + " stream-termination corpus: %d matches, %d logs" % (len(doc["matches"]), len(doc["logs"])))
|
||
return 1 if fails else 0
|
||
|
||
|
||
def verify_one(path):
|
||
"""`--log <file>`: one log, a JSON array of its records in hexadecimal (as a page's
|
||
`StreamLog.records` gives them); prints what §6.3 states of it, or the refusal."""
|
||
with open(path, encoding="utf-8") as f:
|
||
blobs = [bytes.fromhex(x) for x in json.load(f)]
|
||
res = verify_log(blobs)
|
||
print(json.dumps({"outcome": "refuse", "reason": res[1], "at": res[2]} if res[0] == "refuse" else res[1], sort_keys=True))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
if len(sys.argv) == 3 and sys.argv[1] == "--log":
|
||
sys.exit(verify_one(sys.argv[2]))
|
||
if len(sys.argv) != 2:
|
||
print("usage: stream_termination_verify.py <live-conformance/termination> | --log <log.json>", file=sys.stderr)
|
||
sys.exit(2)
|
||
sys.exit(main(sys.argv[1]))
|