open-standards/wai/live-conformance/native_sink_verify.py
Transaction Science 643f86f8d7 wai: the Swift package of the native sink, and a log that keeps its order
wai/wai-sink-swift: the native sink for the platforms whose native player is reached
from Swift. WaiSink wraps the C interface (every call JSON in, JSON out); WaiRelay is the
loopback server the player is given in place of the origin: it fetches each object,
gives the sink a playlist with its receipt and the chain's parents, and answers the
player only with bytes the sink presents (502 otherwise), every URI of a playlist
rewritten to it; WaiPlayerSession drives the platform's player through it, observes it
every 200 ms, logs its first frame, stalls and end, ends a session nobody can see, and
posts the bundle to the owner's collector; WaiViewAttention reads the platform's signals
of who can see the player. scripts/build-xcframework.sh builds WaiFFI.xcframework from
wai-rs. Tests: the corpus through the wrapper, the relay over the reference floor, and
the player end to end over a presentation made with the platform's own encoder.

A native host's player and its requests run on threads of their own, so an object can
reach the sink with a time before the log's last entry. delivery-telemetry §7.1 has a
log's times never decrease and nothing after end or abandon: the sink logs such an entry
at the last entry's time, and takes no entry after end or abandon, while an object
received then is still verified and acknowledged (native-sink §5.2). The corpus gains
log-order, and the second implementation checks both rules.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
2026-10-25 22:55:08 -04:00

1027 lines
45 KiB
Python
Raw Permalink Blame History

This file contains invisible Unicode characters

This file contains invisible Unicode characters that are indistinguishable to humans but may be processed differently by a computer. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
"""Independent verifier of the native sink corpus (extensions/native-sink.md).
Written from the extension's text. It shares no code with wai-rs, which generates the corpus, and
never runs it. It is a second native sink: for every case it opens a session from the case's
configuration, derives the session key from the case's seed (RFC 8032 §5.1.5), and gives each call
the answer it reaches by the extension's rules, from the standard library and these second
implementations of the texts the extension cites:
- a playlist receipt and the chain of a live playlist's receipts (http-delivery §8.6, §11.5), and
RFC 3986 resolution, with ../http-conformance/http_verify.py;
- the bundle (stream-audit §9, §10) with stream_audit_verify.py, whose readers of acknowledgements,
claims and the session record are settlement_verify.py's, ad_presentation_verify.py's and its
own; every signature with ../claim-conformance/_ed25519.py, WAI's strict rule.
Everything else is here: the configuration (§3.1), reading a playlist (§4.1), a playlist's verdict
(§4.2), an object's (§4.3), the observations, spans and claim spans (§5.1, stream-audit §5.2,
ad-presentation §10), the log (§5.2), the termination rule (§6.1), the energy entries (§7) and the
refused calls (§3.3, §8). For the end of a session it requires the bundle the corpus states to be
the one its own session holds: every acknowledgement's fields and item lists, every claim's fields,
the session record's fields, the log and the notes byte for byte, and the collector's outcome
(stream-audit §10.3) the case states. It does not sign: Ed25519 is deterministic, and every byte a
signature covers is checked.
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 -I native_sink_verify.py <live-conformance/native-sink>
Exit status 0 only if every case verifies, every mutated copy fails and every weakened verifier
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, HERE)
sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance"))
sys.path.insert(0, os.path.join(HERE, "..", "http-conformance"))
import _ed25519 as ED # noqa: E402
import _blake3 as B3 # noqa: E402
import http_verify as HV # noqa: E402
import settlement_verify as SV # noqa: E402
import ad_presentation_verify as APV # noqa: E402
import stream_audit_verify as SAV # noqa: E402
EXTENSION = os.path.join(HERE, "..", "extensions", "native-sink.md")
CHANNEL_DOMAIN = b"wai:audit-channel\x01" # stream-audit §3.1
KEY_SESSION_DOMAIN = b"wai:ad-presentation-session\x01" # stream-audit §3.2
ORIGIN_DOMAIN = b"wai:settle-origin\x01" # settlement §14.2
MARKER_DOMAIN = b"wai:audit-marker\x01" # stream-audit §6.1
CREATIVE_DOMAIN = b"wai:ad-creative\x01" # ad-presentation §4
ITEMS_DOMAIN = b"wai:settle-items\x01" # settlement §3.2
OBJECT_DOMAIN = b"moq-jwp:object\x01" # jwp-receipts §2.1
LIVE, ON_DEMAND = 1, 2
ENDED = {"ended": 1, "user": 2, "hidden": 3, "idle": 4, "error": 5}
EVENTS = ["start", "present", "stall-start", "stall-end", "end", "abandon"]
ZERO = bytes(32)
NO_STARTUP = 0xFFFFFFFF
NOT_EXPOSED = 0xFFFFFFFF
U64 = 2 ** 64 - 1
DROP = set() # the checks dropped, for the weakened-verifier pass
class Refused(Exception):
def __init__(self, code):
super().__init__(code)
self.code = code
class Disagree(Exception):
pass
def sha(*parts):
return hashlib.sha256(b"".join(parts)).digest()
def unhex(s):
return bytes.fromhex(s)
def is_uint(v):
return isinstance(v, int) and not isinstance(v, bool) and 0 <= v <= U64
def public_key(seed):
"""RFC 8032 §5.1.5: the public key of a 32-byte seed."""
h = hashlib.sha512(seed).digest()
a = int.from_bytes(h[:32], "little")
a &= (1 << 254) - 8
a |= 1 << 254
q, p = (0, 1, 1, 0), ED.B
while a:
if a & 1:
q = ED._add(q, p)
p = ED._add(p, p)
a >>= 1
return ED.encode(q)
# ------------------------------------------------------------------ §3.1 the configuration
def channel_id(name):
if not isinstance(name, str) or not 1 <= len(name) <= 255 or not all(0x21 <= ord(c) <= 0x7E for c in name):
return None
return sha(CHANNEL_DOMAIN, name.encode())
def read_config(v):
members = ["channel", "owner", "presentation", "window_us", "ad_classes", "content_addressed", "require", "unseen_grace_ms", "idle_ms"]
bad = Refused("native_config_invalid")
if not isinstance(v, dict) or ("config-members" not in DROP and any(k not in members for k in v)):
raise bad
if channel_id(v.get("channel")) is None:
raise bad
owner = v.get("owner")
if not isinstance(owner, str) or not re.fullmatch(r"[0-9a-f]{64}", owner):
raise bad
pres = v.get("presentation")
if pres not in (None, "live", "on_demand"):
raise bad
def num(k, d):
x = v.get(k)
if x is None:
return d
if not is_uint(x):
raise bad
return x
cfg = {"channel": v["channel"], "owner": unhex(owner), "presentation": {None: None, "live": LIVE, "on_demand": ON_DEMAND}[pres],
"window_us": num("window_us", 10_000_000), "unseen_grace_ms": num("unseen_grace_ms", 10_000), "idle_ms": num("idle_ms", 0)}
if cfg["window_us"] < 1_000_000 and "window-floor" not in DROP:
raise bad
ac = v.get("ad_classes")
if ac is None:
ac = []
if not isinstance(ac, list) or not all(isinstance(x, str) for x in ac):
raise bad
cfg["ad_classes"] = ac
ca = v.get("content_addressed")
if ca is None:
ca = False
if not isinstance(ca, bool):
raise bad
cfg["content_addressed"] = ca
req = v.get("require")
if req is None:
req = "nothing"
if req not in ("nothing", "signed"):
raise bad
cfg["require"] = req
return cfg
# ------------------------------------------------------------------ §4.1 reading a playlist
def attributes(s):
out = {}
i, n = 0, len(s)
while i < n:
j = i
while j < n and (s[j].isupper() and s[j].isascii() or s[j].isdigit() and s[j].isascii() or s[j] == "-"):
j += 1
if j == i or j >= n or s[j] != "=":
i = j if j > i else i + 1
continue
name = s[i:j]
j += 1
if j < n and s[j] == '"':
e = s.find('"', j + 1)
if e >= 0:
out[name] = s[j + 1:e]
i = e + 1
continue
e = s.find(",", j)
e = n if e < 0 else e
out[name] = s[j:e]
i = e
return out
def days_from_civil(y, m, d):
y -= m <= 2
era = y // 400
yoe = y - era * 400
doy = (153 * ((m + 9) % 12) + 2) // 5 + d - 1
doe = yoe * 365 + yoe // 4 - yoe // 100 + doy
return era * 146097 + doe - 719468
def date_ms(s):
"""RFC 3339 §5.6 date-time, the fraction truncated to the millisecond."""
m = re.fullmatch(r"(\d{4})-(\d{2})-(\d{2})[Tt](\d{2}):(\d{2}):(\d{2})(\.(\d+))?([Zz]|([+-])(\d{2}):(\d{2}))", s)
if not m:
return None
y, mo, d, h, mi, se = (int(m.group(k)) for k in range(1, 7))
leap = y % 4 == 0 and (y % 100 != 0 or y % 400 == 0)
dim = [31, 29 if leap else 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
if not 1 <= mo <= 12 or not 1 <= d <= dim[mo - 1] or h > 23 or mi > 59 or se > 59:
return None
ms = int((m.group(8) or "0")[:3].ljust(3, "0"))
off = 0
if m.group(10):
oh, om = int(m.group(11)), int(m.group(12))
if oh > 23 or om > 59:
return None
off = (oh * 60 + om) * 60_000 * (1 if m.group(10) == "+" else -1)
return ((days_from_civil(y, mo, d) * 24 + h) * 60 + mi) * 60_000 + se * 1000 + ms - off
def seconds_ms(s):
"""A decimal number of seconds read exactly, rounded half up to the millisecond."""
m = re.match(r"\s*(\d*)(?:\.(\d*))?", s)
whole, frac = m.group(1), m.group(2)
if not whole and frac is None:
return None
if not whole:
whole = "0"
if frac is not None and frac == "" and m.group(1) == "":
return None
f = (frac or "").ljust(4, "0")
ms = int(whole) * 1000 + int(f[:3])
return ms + (1 if int(f[3]) >= 5 else 0)
def resolve(base, ref):
return HV.rfc3986_resolve(base, ref)
def read_playlist(text, base, ad_classes):
p = {"segments": [], "map": None, "keys": [], "ads": [], "on_demand": False, "multivariant": False}
duration, program = None, None
for raw in text.split("\n"):
line = raw.rstrip("\r \t")
if line.startswith("#EXTINF:"):
d = seconds_ms(line[8:])
duration = max(1, d or 0)
elif line.startswith("#EXT-X-PROGRAM-DATE-TIME:"):
program = date_ms(line[25:].strip())
elif line.startswith("#EXT-X-ENDLIST"):
p["on_demand"] = True
elif line.startswith("#EXT-X-STREAM-INF:") or line.startswith("#EXT-X-I-FRAME-STREAM-INF:"):
p["multivariant"] = True
elif line.startswith("#EXT-X-MAP:"):
u = attributes(line[11:]).get("URI")
if u is not None:
p["map"] = resolve(base, u)
elif line.startswith("#EXT-X-KEY:") or line.startswith("#EXT-X-SESSION-KEY:"):
u = attributes(line[line.index(":") + 1:]).get("URI")
if u is not None:
p["keys"].append(resolve(base, u))
elif line.startswith("#EXT-X-DATERANGE:"):
a = attributes(line[17:])
is_ad = "SCTE35-OUT" in a or "SCTE35-CMD" in a or a.get("CLASS") in ad_classes
dv = a.get("DURATION", a.get("PLANNED-DURATION"))
d = seconds_ms(dv) if dv is not None else None
st = date_ms(a["START-DATE"]) if "START-DATE" in a else None
if is_ad and d is not None and st is not None:
p["ads"].append({"marker": line, "start": st, "duration": d})
elif line and not line.startswith("#"):
if not p["multivariant"]:
p["segments"].append({"url": resolve(base, line), "duration": duration or 1, "program": program})
if program is not None and duration is not None:
program += duration
duration = None
if p["multivariant"]:
p["segments"], p["map"] = [], None
return p
def origin_of(url):
m = re.match(r"([^:/?#]+)://([^/?#]*)", url)
if not m:
return None
scheme, auth = m.group(1).lower(), m.group(2)
auth = auth.rsplit("@", 1)[-1]
if not auth:
return None
if auth.startswith("["):
host, _, rest = auth[1:].partition("]")
host, port = "[" + host.lower() + "]", rest[1:] if rest.startswith(":") else None
elif ":" in auth:
host, port = auth.rsplit(":", 1)
host = host.lower()
else:
host, port = auth.lower(), None
default = {"http": "80", "https": "443"}.get(scheme, "")
return f"{scheme}://{host}:{port}" if port and port != default else f"{scheme}://{host}"
def stated_cid(line):
"""http-delivery §3.1, §5.2: the cid a URI states, its fragment's `cid`, else a last path
segment `<cid>.wai`."""
base, _, frag = line.partition("#")
for kv in frag.split("&") if frag else []:
k, eq, v = kv.partition("=")
if eq and k == "cid" and re.fullmatch(r"[0-9a-f]{64}", v):
return unhex(v)
name = last_segment(base)
if name and name.endswith(".wai") and re.fullmatch(r"[0-9a-f]{64}", name[:-4]):
return unhex(name[:-4])
return None
def last_segment(u):
u = re.split(r"[?#]", u, maxsplit=1)[0]
s = u.rsplit("/", 1)[-1]
return s or None
def stated_sha256(url):
name = last_segment(url)
if name is None:
return None
m = re.fullmatch(r"([0-9a-f]{64})\.([A-Za-z0-9]+)", name)
if not m or m.group(2) == "wai":
return None
return unhex(m.group(1))
def content_hash(b):
return B3.blake3(OBJECT_DOMAIN + b)
# ------------------------------------------------------------------ §5 spans, ad spans, the log
class ContentSpans:
def __init__(self, start):
self.start, self.done, self.cur = start, [], None
def ms(self, at):
return max(0, (at - self.start) // 1000)
def close(self, at):
if self.cur is not None:
end = self.ms(at)
if end > self.cur["s"]:
self.done.append({"start": self.cur["s"], "length": end - self.cur["s"], "visible": self.cur["v"], "flags": (1 if self.cur["a"] else 0) | (2 if self.cur["ad"] else 0)})
self.cur = None
def observe(self, at, playing, vis, audible, ad):
if not playing:
return self.close(at)
cls = vis // 100
if self.cur is not None and (cls != self.cur["c"] or audible != self.cur["a"] or ad != self.cur["ad"]):
self.close(at)
if self.cur is not None:
self.cur["v"] = min(self.cur["v"], vis)
return
self.cur = {"s": self.ms(at), "v": vis, "c": cls, "a": audible, "ad": ad}
class AdSpans:
def __init__(self):
self.done, self.cur, self.max_end, self.resume_next = [], None, 0, False
def close(self):
c = self.cur
if c is not None and c["e"] > c["s"]:
self.done.append({"start": c["s"], "length": c["e"] - c["s"], "visible": c["v"], "flags": (1 if c["a"] else 0) | (2 if c["r"] else 0)})
self.max_end = max(self.max_end, c["e"])
self.cur = None
def observe(self, pos, playing, vis, audible, seeked):
pos = max(0, pos)
if not playing or seeked:
self.close()
if self.done:
self.resume_next = True
if not playing:
return
cls = vis // 100
c = self.cur
if c is not None:
if pos < c["e"]:
return
if cls != c["c"] or audible != c["a"]:
c["e"] = pos
self.close()
else:
c["e"], c["v"] = pos, min(c["v"], vis)
return
if pos < self.max_end:
return
continues = not self.resume_next and bool(self.done) and pos == self.max_end
self.cur = {"s": pos, "e": pos, "v": vis, "c": cls, "a": audible, "r": bool(self.done) and not continues}
self.resume_next = False
OBJECT_NULLS = ["ot", "url", "content_hash", "capability", "rendition", "layer", "layers_total", "layer_cap", "counted", "transferred", "d_ms", "dl_ms",
"ttfb_ms", "ttlb_ms", "rc", "path", "receipt"]
def observation(sid, sn, at, event, st, content_hash=None, rc=None):
obj = None
if event == "response":
obj = {k: None for k in OBJECT_NULLS}
obj["content_hash"], obj["rc"] = content_hash, rc
present = {"unit": 0, "layers": 1, "first": True, "late_ms": None, "decode_us": None} if event == "present" else None
o = {"v": 1, "sid": sid, "sn": sn, "ts_ms": at // 1000, "event": event, "cid": None, "st": st, "sf": "h", "object": obj, "present": present}
for k in ["switch", "layer_change", "capable", "cohort", "metrics", "claim"]:
o[k] = None
return o
def cmcd_report(o):
"""delivery-telemetry §3, §4: the Event-mode report of an observation."""
if o["event"] == "present" and not o["present"]["first"]:
return None
sf = lambda s: '"' + s.replace("\\", "\\\\").replace('"', '\\"') + '"' # noqa: E731
pre = "science.transaction.wai-"
m = {"v": "2", "sid": sf(o["sid"]), "sn": str(o["sn"]), "st": o["st"], "sf": o["sf"], "ts": str(o["ts_ms"])}
if o["object"] is not None:
if o["object"]["content_hash"] is not None:
m[pre + "ch"] = sf(o["object"]["content_hash"])
if o["object"]["rc"] is not None:
m["rc"] = str(o["object"]["rc"])
e, sta = {"response": ("rr", None), "start": ("ps", "s"), "present": ("ps", "p"), "stall-end": ("ps", "p"), "stall-start": ("ps", "r"),
"abandon": ("ps", "q"), "end": ("ps", "e")}[o["event"]]
m["e"] = e
if sta:
m["sta"] = sta
if o["present"] is not None:
m[pre + "u"], m[pre + "pl"] = sf(str(o["present"]["unit"])), sf(str(o["present"]["layers"]))
return ",".join(f"{k}={m[k]}" for k in sorted(m))
def class_tuple(e):
"""An entry's class as energy-measurement §2.7 encodes it: its tag, then its evidence."""
lab, u = e["energy_provenance"], e.get("energy_uncertainty") or {}
if lab == "ModelBased":
return (2,)
if lab == "Estimator":
return (3,)
if lab == "OnChipCounter":
return (4, u.get("relative_ppm"), u.get("window_us"), {"Unknown": 0, "Off": 1, "On": 2}.get(e.get("energy_counter_filtering")))
return (5, u.get("relative_ppm"), u.get("window_us"), e.get("energy_calibration_ref"))
# ------------------------------------------------------------------ the sink
class Sink:
def __init__(self, cfg, seed, start):
self.cfg, self.start = cfg, start
self.player = public_key(seed)
self.channel = channel_id(cfg["channel"])
self.session = sha(KEY_SESSION_DOMAIN, self.channel, self.player)[:16]
self.sid = self.session.hex()
self.presentation = cfg["presentation"]
self.received, self.by_url, self.notes, self.log, self.cmcd_taken = [], {}, [], [], 0
self.spans = ContentSpans(start)
self.segments, self.ads, self.keys = {}, [], set()
self.roles = {3: (0, None), 4: (0, None)}
self.attachments, self.last = [], None
self.holds, self.listed, self.given = {}, {}, {}
self.unseen_since, self.last_interaction, self.ended = None, None, None
if "log-start" not in DROP:
self.log_event("start", start)
def st(self):
return "l" if (self.presentation or LIVE) == LIVE else "v"
def log_event(self, name, at):
self.log_push(name, at)
def log_push(self, event, at, content_hash=None, rc=None):
"""§5.2: nothing after `end` or `abandon`; times never decrease."""
if self.has(["end", "abandon"]) and "log-closed" not in DROP:
return
last = self.log[-1]["ts_ms"] if self.log else 0
if "log-order" not in DROP:
at = max(at, last * 1000)
self.log.append(observation(self.sid, len(self.log), at, event, self.st(), content_hash, rc))
def has(self, names):
return any(o["event"] in names for o in self.log)
def live(self):
if self.ended is not None and "after-end" not in DROP:
raise Refused("native_session_ended")
# §5.2
def event(self, name, at):
self.live()
if name not in EVENTS:
raise Refused("native_input_invalid")
if name == "present" and self.has(["present"]):
return
self.log_event(name, at)
def hold(self, k):
return self.holds.setdefault(k, {"chain": {"head": None, "bindings": HV.new_bindings(k), "bound": HV.WALK_BOUND}, "parents": {}})
# §4.2
def playlist(self, url, b, receipt):
self.live()
v = self.take_playlist(url, b, receipt)
if v["need_parent"] is None:
self.given[(resolve(url, url), B3.blake3(b))] = (v["listing"], v["refused"])
v["receipt_name"] = "mr/" + B3.blake3(b).hex() + ".json"
v["present"] = v["refused"] is None and v["need_parent"] is None
return {k: v[k] for k in ["kind", "listing", "receipt", "receipt_name", "step", "need_parent", "refused", "present", "segments", "ads"]}
def take_playlist(self, url, b, receipt):
v = {"kind": "media", "listing": "listed", "receipt": None, "step": None, "need_parent": None, "refused": None, "segments": 0, "ads": 0}
try:
text = b.decode("utf-8")
except UnicodeDecodeError:
text = None
if text is None or len(b) > 16 * 1024 * 1024 or not text.lstrip("").startswith("#EXTM3U"):
v["refused"] = "playlist_malformed"
return v
k = resolve(url, url)
p = read_playlist(text, k, self.cfg["ad_classes"])
if p["multivariant"]:
v["kind"] = "multivariant"
return v
signed = False
if receipt is not None and "receipts" not in DROP:
try:
r = HV.check_playlist(b, receipt, None)
except HV.Refusal as e:
v["refused"] = e.code
return v
h = self.hold(k)
needed = []
def fetch(x):
if x in h["parents"]:
return h["parents"][x]
needed.append(x)
return None
if "chain" not in DROP:
try:
step, _ = HV.advance(h["chain"], r, fetch)
except HV.Refusal as e:
if e.code == "playlist_receipt_parent_missing" and needed:
v["need_parent"] = "mr/" + needed[0].hex() + ".json"
return v
v["refused"] = e.code
return v
v["step"] = step
v["receipt"] = HV.receipt_hash(r).hex()
signed = True
elif self.cfg["require"] == "signed" and "require-signed" not in DROP:
v["refused"] = "playlist_unsigned"
return v
v["listing"] = "signed" if signed else "listed"
if self.presentation is None:
self.presentation = ON_DEMAND if p["on_demand"] else LIVE
for x in p["keys"]:
self.keys.add(resolve(x, x))
cids = {}
for raw in text.split("\n"):
line = raw.rstrip("\r \t")
if line and not line.startswith("#"):
c = stated_cid(line)
if c is not None:
cids[resolve(k, line)] = c
def list_(u, media):
e = self.listed.setdefault(u, {"signed": signed, "cid": cids.get(u), "media": media})
e["signed"] = e["signed"] or signed
e["media"] = e["media"] or media
if e["cid"] is None:
e["cid"] = cids.get(u)
if p["map"] is not None:
list_(p["map"], True)
for s in p["segments"]:
list_(s["url"], True)
self.segments[s["url"]] = s
for a in p["ads"]:
if not any(x["ad"]["marker"] == a["marker"] for x in self.ads):
self.ads.append({"ad": a, "rec": AdSpans()})
v["segments"], v["ads"] = len(p["segments"]), len(p["ads"])
return v
def parent(self, url, name, b):
self.live()
m = re.fullmatch(r"mr/([0-9a-f]{64})\.json", name)
self.hold(resolve(url, url))["parents"][unhex(m.group(1))] = b
# §4.3
def object(self, meta, b):
self.live()
members = ["url", "t_us", "media", "status", "cmsd_static", "cmsd_dynamic"]
if not isinstance(meta, dict) or any(k not in members for k in meta) or not isinstance(meta.get("url"), str) or not is_uint(meta.get("t_us")):
raise Refused("native_input_invalid")
for k in ["cmsd_static", "cmsd_dynamic"]:
if meta.get(k) is not None and not isinstance(meta[k], str):
raise Refused("native_input_invalid")
if meta.get("media") is not None and not isinstance(meta["media"], bool):
raise Refused("native_input_invalid")
status = 200 if meta.get("status") is None else meta["status"]
if not is_uint(status):
raise Refused("native_input_invalid")
url, at = meta["url"], meta["t_us"]
k = resolve(url, url)
s256 = sha(b)
listed = self.listed.get(k)
given = None if "playlist-body" in DROP else self.given.get((k, B3.blake3(b)))
media = meta["media"] if meta.get("media") is not None else bool(listed and listed["media"])
listing = given[0] if given else ("signed" if listed and listed["signed"] else "listed" if listed else "unlisted")
v = {"sha256": s256.hex(), "cid": None, "listing": listing, "bound": "none", "media": media, "key": False, "acknowledged": True,
"refused": given[1] if given else None}
if k in self.keys and "keys" not in DROP:
v["key"], v["acknowledged"] = True, False
v["present"] = True
return v
stated = (listed or {}).get("cid") or stated_cid(url)
if stated is not None and "cid-name" not in DROP:
v["bound"] = "cid"
h = content_hash(b)
v["cid"] = h.hex()
if h != stated and v["refused"] is None:
v["refused"] = "object_digest_mismatch"
elif stated_sha256(k) is not None and (self.cfg["content_addressed"] or listing == "signed") and "sha256-name" not in DROP:
v["bound"] = "sha256"
if stated_sha256(k) != s256 and v["refused"] is None:
v["refused"] = "object_digest_mismatch"
if v["refused"] is None and self.cfg["require"] == "signed" and given is None and listing != "signed" and "require-signed" not in DROP:
v["refused"] = "object_unsigned"
origin = origin_of(k)
if origin is None:
raise Refused("native_input_invalid")
self.received.append({"at": at, "origin": origin, "media": media, "len": len(b), "hash": s256})
self.log_push("response", at, s256.hex(), status)
if meta.get("cmsd_static") is not None or meta.get("cmsd_dynamic") is not None:
self.notes.append({"content_hash": s256.hex(), "cmsd_static": meta.get("cmsd_static"), "cmsd_dynamic": meta.get("cmsd_dynamic")})
self.by_url[k] = s256
v["present"] = v["refused"] is None
return v
# §5.1, §6.1
def ad_at(self, program):
if program is None:
return None
for i, x in enumerate(self.ads):
if x["ad"]["start"] <= program < x["ad"]["start"] + x["ad"]["duration"]:
return i
return None
def ad_segments(self, ad):
segs = [s for s in self.segments.values() if s["program"] is not None and s["program"] < ad["start"] + ad["duration"] and s["program"] + s["duration"] > ad["start"]]
return sorted(segs, key=lambda s: s["program"])
def observe(self, o):
self.live()
members = ["t_us", "playing", "visible_bp", "audible", "program_ms", "seeked", "interaction"]
if not isinstance(o, dict) or any(k not in members for k in o) or not is_uint(o.get("t_us")):
raise Refused("native_input_invalid")
vis = o.get("visible_bp")
if isinstance(vis, bool) or not isinstance(vis, (int, float)):
raise Refused("native_input_invalid")
flags = {}
for k in ["playing", "audible", "seeked", "interaction"]:
x = o.get(k)
if x is not None and not isinstance(x, bool):
raise Refused("native_input_invalid")
flags[k] = bool(x)
prog = o.get("program_ms")
if prog is not None and (isinstance(prog, bool) or not isinstance(prog, int)):
raise Refused("native_input_invalid")
at, playing, audible, seeked = o["t_us"], flags["playing"], flags["audible"], flags["seeked"]
vis = int(min(10000, max(0, (vis + 0.5) // 1)))
if playing and not self.has(["present"]) and "log-present" not in DROP:
self.log_event("present", at)
ad = self.ad_at(prog)
if "spans" not in DROP:
self.spans.observe(at, playing, vis, audible, ad is not None)
last = self.last
for i, x in enumerate(self.ads):
segs = self.ad_segments(x["ad"])
base = segs[0]["program"] if segs else x["ad"]["start"]
rec = x["rec"]
if i == ad:
if rec.cur is None and not rec.done and playing and not seeked and last and last[0] and last[1] is not None and last[1] < base:
rec.observe(0, True, vis, audible, False)
rec.observe(prog - base, playing, vis, audible, seeked)
elif rec.cur is not None:
if playing and not seeked and prog is not None:
d = sum(s["duration"] for s in segs) if segs else x["ad"]["duration"]
rec.observe(min(prog - base, d), True, rec.cur["v"], rec.cur["a"], False)
rec.observe(0, False, 0, False, False)
self.last = (playing, prog)
if flags["interaction"]:
self.last_interaction = at
if playing and vis == 0:
if self.unseen_since is None:
self.unseen_since = at
elif vis > 0:
self.unseen_since = None
g = self.cfg["unseen_grace_ms"]
if g > 0 and self.unseen_since is not None and at - self.unseen_since >= g * 1000 and "rule-hidden" not in DROP:
return {"end": "hidden"}
idle = self.cfg["idle_ms"]
since = self.last_interaction if self.last_interaction is not None else self.start
if idle > 0 and at - since >= idle * 1000 and "rule-idle" not in DROP:
return {"end": "idle"}
return {"end": None}
# §7
def energy(self, e):
self.live()
role = e.get("role")
if not is_uint(role) or not 1 <= role <= 4:
raise Refused("native_input_invalid")
if e.get("absent") is True:
self.roles.pop(role, None)
return
j = e.get("joules_micro")
label = e.get("energy_provenance")
if not is_uint(j) or label not in (None, "OnChipCounter", "CalibratedInstrument", "ModelBased", "Estimator"):
raise Refused("native_input_invalid")
if "energy-cohere" not in DROP and (label is None) != (j == 0):
raise Refused("native_input_invalid")
self.roles[role] = (j, None if label is None else e)
def attach(self, name, b):
self.live()
if not name:
raise Refused("native_input_invalid")
self.attachments.append((name, b))
def take_cmcd(self):
self.live()
fresh = self.log[self.cmcd_taken:]
self.cmcd_taken = len(self.log)
return "\n".join(r for r in (cmcd_report(o) for o in fresh) if r is not None)
# stream-audit §4.2: what each acknowledgement holds
def windows(self, end):
w = self.cfg["window_us"]
media_class = 1 if (self.presentation or LIVE) == LIVE else 2
wins, seen, rep = {}, set(), [0, 0]
for r in self.received:
kk = (r["at"] - self.start) // w
if (r["origin"], kk, r["hash"]) in seen:
rep[0] += 1
rep[1] += r["len"]
continue
seen.add((r["origin"], kk, r["hash"]))
c = media_class if r["media"] else 4
wins.setdefault((r["origin"], kk), {}).setdefault(c, []).append(r)
out = []
for (origin, kk) in sorted(wins):
ws = self.start + kk * w
out.append({"origin": origin, "start": ws, "end": min(ws + w, end), "classes": sorted(wins[(origin, kk)].items())})
return out, rep
def end(self, name, at, bundle_text):
"""The bundle the corpus states is checked against what this session holds."""
self.live()
if name not in ENDED or at <= self.start:
raise Refused("native_input_invalid")
ended = ENDED[name]
if not self.has(["end", "abandon"]) and "log-end" not in DROP:
self.log_event("end" if ended == 1 else "abandon", at)
self.ended = ended
spans = [] if "spans" in DROP else (self.spans.close(at) or self.spans.done)
try:
b = json.loads(bundle_text)
except ValueError:
raise Disagree("the bundle is not JSON")
if set(b) != {"v", "session", "acks", "claims", "log", "notes", "attachments"}:
raise Disagree("the bundle's members")
log = json.dumps([self.log], separators=(",", ":"), ensure_ascii=False)
notes = json.dumps(self.notes, separators=(",", ":"), ensure_ascii=False) if self.notes else None
if b["log"] != log:
raise Disagree("the log")
if b["notes"] != notes:
raise Disagree("the notes")
# The acknowledgements: one per origin and window, chained, in order.
wins, rep = self.windows(at)
if len(b["acks"]) != len(wins):
raise Disagree("the number of acknowledgements")
prev, ack_digests = {}, []
for a, w in zip(b["acks"], wins):
raw = unhex(a["record"])
r = SV.read_ack(raw)
want_items = [b"".join(x["hash"] + struct.pack("<QQ", 0, x["len"]) for x in items) for _, items in w["classes"]]
if [unhex(x) for x in a["items"]] != want_items:
raise Disagree("an acknowledgement's items")
if (r["mode"], r["role"], r["upstream"], r["downstream"], r["start"], r["end"], r["prev"]) != (1, 2, sha(ORIGIN_DOMAIN, w["origin"].encode()), self.player, w["start"], w["end"], prev.get(w["origin"], ZERO)):
raise Disagree("an acknowledgement's fields")
ents = [(e["class"], e["grant"], e["proof"], e["objects"], e["bytes"], e["items"]) for e in r["entries"]]
want = [(c, ZERO, ZERO, len(items), sum(x["len"] for x in items), sha(ITEMS_DOMAIN, il)) for (c, items), il in zip(w["classes"], want_items)]
if ents != want:
raise Disagree("an acknowledgement's entries")
d = sha(raw[:-64])
prev[w["origin"]] = d
ack_digests.append(d)
# The claims.
want_claims = []
for x in self.ads:
x["rec"].close()
sp = x["rec"].done
segs = self.ad_segments(x["ad"])
if not sp or not segs or any(s["url"] not in self.by_url for s in segs):
continue
creative = [(self.by_url[s["url"]], min(s["duration"], 0xFFFFFFFF)) for s in segs]
cd = sha(CREATIVE_DOMAIN, struct.pack("<I", len(creative)), *(h + struct.pack("<I", d) for h, d in creative))
touched, t0 = [], 0
for h, d in creative:
if any(s["start"] < t0 + d and s["start"] + s["length"] > t0 for s in sp):
touched.append(h)
t0 += d
want_claims.append({"marker": x["ad"]["marker"], "creative": creative, "cd": cd, "objects": touched, "spans": sp})
if len(b["claims"]) != len(want_claims):
raise Disagree("the number of claims")
claim_digests = []
for c, w in zip(b["claims"], want_claims):
r = APV.read_claim(unhex(c["claim"]))
got = (r["room"], r["cue"], r["session"], r["slot"], r["creative_hash"], r["player"], r["geometry"], r["frames"], r["objects"], r["spans"], r["attestation"])
if got != (self.channel, sha(MARKER_DOMAIN, w["marker"].encode()), self.session, 0, w["cd"], self.player, 3, NOT_EXPOSED, w["objects"], w["spans"], b""):
raise Disagree("a claim's fields")
if c["marker"] != w["marker"] or [(unhex(o["content_hash"]), o["duration_ms"]) for o in c["creative"]] != w["creative"]:
raise Disagree("a claim's creative")
claim_digests.append(sha(r["body"]))
# The attachments.
if [(a["name"], unhex(a["record"])) for a in b["attachments"]] != self.attachments:
raise Disagree("the attachments")
# The record.
s = SAV.read_session(unhex(b["session"]))
origins = sorted({r["origin"] for r in self.received})
got_energy = [(e["role"], e["joules"], e["class"]) for e in s["energy"]]
want_energy = [(role, j, None if e is None else class_tuple(e)) for role, (j, e) in sorted(self.roles.items())]
if got_energy != want_energy:
raise Disagree("the record's energy")
fields = (s["channel"], s["owner"], s["player"], s["session"], s["presentation"], s["start"], s["end"], s["ended"],
[o.decode() for o in s["origins"]], s["acks"], s["claims"], s["spans"], s["repeat_objects"], s["repeat_bytes"],
s["log"], s["notes"], s["attachments"])
want = (self.channel, self.cfg["owner"], self.player, self.session, self.presentation or LIVE, self.start, at, ended,
origins, sorted(ack_digests), sorted(claim_digests), spans, rep[0], rep[1],
sha(log.encode()), sha(notes.encode()) if notes else ZERO, sorted(sha(x) for _, x in self.attachments))
for i, (g, wv) in enumerate(zip(fields, want)):
if g != wv:
raise Disagree(f"the record's field {i}")
return {"bundle": bundle_text}
# ------------------------------------------------------------------ replaying the corpus
def run_call(sink, objects, d, answer):
body = lambda k: None if d.get(k) is None else unhex(objects[d[k]]) # noqa: E731
try:
c = d["call"]
if c == "playlist":
return sink.playlist(d["url"], body("body"), body("receipt"))
if c == "parent":
sink.parent(d["url"], d["name"], body("body"))
return {"ok": True}
if c == "object":
return sink.object(d["meta"], body("body"))
if c == "event":
sink.event(d["name"], d["t_us"])
return {"ok": True}
if c == "observe":
return sink.observe(d["obs"])
if c == "energy":
sink.energy(d["entry"])
return {"ok": True}
if c == "attach":
sink.attach(d["name"], body("body"))
return {"ok": True}
if c == "cmcd":
return {"cmcd": sink.take_cmcd()}
if c == "end":
if not isinstance(answer, dict) or "bundle" not in answer:
sink.live()
raise Disagree("an end the corpus states no bundle for")
return sink.end(d["ended"], d["t_us"], answer["bundle"])
raise Disagree(f"an unknown call {c}")
except Refused as e:
return {"error": e.code}
def check(doc, quiet=False):
fails = []
objects = doc["objects"]
for c in doc["configs"]:
try:
read_config(c["config"])
got = None
except Refused as e:
got = e.code
if got != c["refused"]:
fails.append(f"config {c['note']}: {got}")
for case in doc["cases"]:
name = case["name"]
try:
cfg = read_config(case["config"])
sink = Sink(cfg, unhex(case["seed"]), case["start_us"])
bundle = None
for i, call in enumerate(case["calls"]):
got = run_call(sink, objects, call["do"], call["answer"])
if got != call["answer"]:
raise Disagree(f"call {i}: {json.dumps(got)[:300]} where the corpus states {json.dumps(call['answer'])[:300]}")
if bundle is None and isinstance(got, dict) and "bundle" in got:
bundle = got["bundle"]
if bundle is None:
raise Disagree("no bundle")
out = SAV.verify_bundle(bundle.encode(), unhex(doc["owner"]), doc["channels"])
if out != case["outcome"]:
raise Disagree(f"the collector's outcome {json.dumps(out)[:300]}")
if not quiet:
print(f"PASS {name}")
except Exception as e: # a verifier reports every failure as a failure, never a crash
fails.append(f"{name}: {e!r}"[:500])
return fails
def reasons_of_extension():
text = open(EXTENSION, encoding="utf-8").read()
sec = text.split("## 8. Reasons", 1)[1].split("\n## ", 1)[0]
return {m.group(1) for m in re.finditer(r"^\| `([a-z_]+)` \|", sec, re.M)}
def reasons_given(doc):
out = {c["refused"] for c in doc["configs"]}
for case in doc["cases"]:
for call in case["calls"]:
a = call["answer"]
for k in ("refused", "error"):
if isinstance(a, dict) and isinstance(a.get(k), str):
out.add(a[k])
return out
def flip_hex(h, at):
i = at % len(h)
return h[:i] + ("0" if h[i] != "0" else "1") + h[i + 1:]
def find(doc, name):
return next(c for c in doc["cases"] if c["name"] == name)
def mutations(doc):
"""Copies of the corpus with one defect each; every one must fail."""
def m(fn):
d = copy.deepcopy(doc)
fn(d)
return d
out = []
out.append(("a fragment's bytes changed", m(lambda d: d["objects"].__setitem__("f2", flip_hex(d["objects"]["f2"], 40)))))
out.append(("a receipt's bytes changed", m(lambda d: d["objects"].__setitem__("vod-receipt", d["objects"]["vod-receipt"].replace("7075626c6963", "70726976617a", 1)))))
out.append(("a verdict says present for a refused fragment", m(lambda d: next(c for c in find(d, "tampered-fragment")["calls"] if c["answer"].get("refused"))["answer"].__setitem__("present", True))))
out.append(("a refusal's reason renamed", m(lambda d: next(c for c in find(d, "unsigned-refused")["calls"] if c["answer"].get("refused"))["answer"].__setitem__("refused", "playlist_receipt_invalid"))))
out.append(("content addressing turned off", m(lambda d: find(d, "content-addressed")["config"].__setitem__("content_addressed", False))))
out.append(("a session's seed changed", m(lambda d: find(d, "signed-vod").__setitem__("seed", flip_hex(find(d, "signed-vod")["seed"], 3)))))
out.append(("two calls swapped", m(lambda d: find(d, "wai-objects")["calls"].insert(1, find(d, "wai-objects")["calls"].pop(0)))))
out.append(("the hidden rule's grace lengthened", m(lambda d: find(d, "hidden")["config"].__setitem__("unseen_grace_ms", 5000))))
out.append(("an observation's end dropped", m(lambda d: next(c for c in find(d, "idle")["calls"] if c["answer"].get("end"))["answer"].__setitem__("end", None))))
def bundle_edit(d, case, fn):
a = next(c for c in find(d, case)["calls"] if "bundle" in c["answer"])["answer"]
b = json.loads(a["bundle"])
fn(b)
a["bundle"] = json.dumps(b, separators=(",", ":"))
out.append(("a bundle's record changed", m(lambda d: bundle_edit(d, "signed-vod", lambda b: b.__setitem__("session", flip_hex(b["session"], 300))))))
out.append(("a bundle's item list changed", m(lambda d: bundle_edit(d, "page-parity", lambda b: b["acks"][0]["items"].__setitem__(0, flip_hex(b["acks"][0]["items"][0], 5))))))
out.append(("a bundle's log changed", m(lambda d: bundle_edit(d, "hidden", lambda b: b.__setitem__("log", b["log"].replace('"abandon"', '"end"'))))))
out.append(("an outcome's time presented changed", m(lambda d: find(d, "signed-vod")["outcome"].__setitem__("presented_ms", 1))))
out.append(("a key acknowledged", m(lambda d: next(c for c in find(d, "third-party-ad")["calls"] if c["answer"].get("key"))["answer"].__setitem__("acknowledged", True))))
out.append(("an energy entry with no class beside a figure accepted", m(lambda d: next(c for c in find(d, "energy")["calls"] if c["answer"].get("error"))["answer"].update({"ok": True, "error": None}))))
out.append(("a parent named under another hash", m(lambda d: next(c for c in find(d, "live-chain")["calls"] if c["do"]["call"] == "parent")["do"].__setitem__("name", "mr/" + "0" * 64 + ".json"))))
out.append(("a configuration's refusal dropped", m(lambda d: d["configs"][0].__setitem__("refused", None))))
return out
DROPPABLE = ["config-members", "window-floor", "receipts", "chain", "require-signed", "playlist-body", "keys", "cid-name", "sha256-name",
"rule-hidden", "rule-idle", "energy-cohere", "after-end", "spans", "log-start", "log-present", "log-end", "log-order", "log-closed"]
def main(corpus_dir):
doc = json.load(open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8"))
status = 0
files = sorted(os.listdir(corpus_dir))
if files != ["README.md", "vectors.json"]:
print(f"FAIL the corpus holds other files: {files}")
status = 1
fails = check(doc)
for f in fails:
print(f"FAIL {f}")
status |= bool(fails)
want, given = reasons_of_extension(), reasons_given(doc)
if want != given:
print(f"FAIL §8 lists {sorted(want - given)} that no vector reaches, and the corpus reaches {sorted(given - want)} that §8 does not list")
status = 1
else:
print(f"PASS every reason of §8 refuses a vector ({len(want)})")
caught = 0
for name, d in mutations(doc):
if check(d, quiet=True):
caught += 1
else:
print(f"FAIL a mutated corpus passed: {name}")
status = 1
print(f"{'PASS' if caught == len(mutations(doc)) else 'FAIL'} every mutated copy of the corpus fails ({caught})")
weak = 0
for k in DROPPABLE:
DROP.clear()
DROP.add(k)
if check(doc, quiet=True):
weak += 1
else:
print(f"FAIL the corpus does not catch the verifier with its check `{k}` dropped")
status = 1
DROP.clear()
print(f"{'PASS' if weak == len(DROPPABLE) else 'FAIL'} every weakened verifier fails on the corpus ({weak})")
return status
if __name__ == "__main__":
if len(sys.argv) != 2:
print("usage: native_sink_verify.py <live-conformance/native-sink>", file=sys.stderr)
sys.exit(2)
sys.exit(main(sys.argv[1]))