b6b9c17456
extensions/ad-presentation.md defines the WPC1 claim, the session a session key names, the creative digest, the video and cross-media viewability rules as their texts state them, the quartile, the RFC 9577 token bound to one impression and verified as RFC 9578 §6.4 has it, the presentation fields of a WAP1 policy of version 2, and each claim's verdict; live-ads.md takes the version and the spend that counts. ad_presentation_verify.py is written from the text, verifies RFC 9578's own vector, gives every claim the corpus's verdict, and catches every mutated copy and every check dropped from itself. The corpus gains a signed claim that breaks a span rule. Co-Authored-By: Claude Opus 5.5 <[email protected]>
796 lines
36 KiB
Python
796 lines
36 KiB
Python
#!/usr/bin/env python3
|
|
"""Independent verifier of the live-ads corpus (extensions/live-ads.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, and
|
|
computes BLAKE3 with ../claim-conformance/_blake3.py, written from its specification. Each room
|
|
is verified by the live-rooms corpus's own second implementation, live_rooms_verify.py
|
|
(live-rooms §9), written from its text alone. Everything else is here:
|
|
|
|
- the ad policy (§3): its layout, rules and signature, and the decider grant;
|
|
- the break annotation and the schedule (§4), with its refusals;
|
|
- the HLS interstitial, closing and preload tags (§5) and the DASH events (§6), byte for byte;
|
|
- the decision and the asset lists (§7), byte for byte;
|
|
- the break form of the wai.video.ssai receipt (§8): its JSON carriage, hashes, signing
|
|
payload, identity and rules;
|
|
- verification, from the room to the spend (§9).
|
|
|
|
A policy of version 2 is read here with its presentation fields (§3.1, ad-presentation §8). A
|
|
room whose policy requires presentation claims is verified with ad_presentation_verify.py,
|
|
which gives accept() the judge of those claims (ad-presentation §9).
|
|
|
|
For every room it requires the outcome, reason and place, and everything else the corpus
|
|
states; every HLS tag file, byte for byte; of the corpus, that every reason of §10 appears in
|
|
it and that no other file is there. Then it mutates copies of the corpus, one defect each, and
|
|
requires every copy to fail.
|
|
|
|
python3 live_ads_verify.py <live-conformance/ads>
|
|
|
|
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
|
|
import _blake3 as B3 # noqa: E402
|
|
|
|
Refuse = LEV.Refuse
|
|
EXTENSION = os.path.join(HERE, "..", "extensions", "live-ads.md")
|
|
|
|
POLICY_MAGIC, ANNOTATION_MAGIC = b"WAP1", b"WAB1"
|
|
POLICY_DOMAIN = b"wai:live-ads-policy\x01" # §3.2
|
|
BREAK_DOMAIN = b"wai:video-ssai-break-receipt\x01" # §8.4
|
|
ID_DOMAIN = b"wai:video-ssai-receipt-id\x01"
|
|
POD_DOMAIN, IMP_DOMAIN = b"wai:video-ssai-pod\x01", b"wai:video-ssai-imp\x01" # §8.3
|
|
LEAF, NODE = b"moq-jwp:merkle-leaf\x01", b"moq-jwp:merkle-node\x01" # jwp-receipts §2.2
|
|
INTERSTITIAL = "com.apple.hls.interstitial" # allow-name: the interstitial date-range class, a protocol identifier
|
|
PRELOAD = "com.apple.hls.preload" # allow-name: the preload date-range class, a protocol identifier
|
|
REPLACE = "urn:mpeg:dash:event:alternativeMPD:replace:2025" # §6.1
|
|
INSERT = "urn:mpeg:dash:event:alternativeMPD:insert:2025"
|
|
OUT, IN, CHAPTER, BREAK = 1, 2, 3, 4
|
|
META, CUE = 0x01, 0x02
|
|
MID, POST = 1, 2
|
|
HLS, DASH = 1, 2
|
|
NO_SKIP = 65535
|
|
NO_CEILING = 2 ** 64 - 1
|
|
U64 = 2 ** 64 - 1
|
|
FLOOR = 288676 # §8.5 rule 5
|
|
# Unicode White_Space, the code points a calibration reference must hold more than (§8.5).
|
|
WHITE_SPACE = set(map(chr, [0x09, 0x0A, 0x0B, 0x0C, 0x0D, 0x20, 0x85, 0xA0, 0x1680] + list(range(0x2000, 0x200B))
|
|
+ [0x2028, 0x2029, 0x202F, 0x205F, 0x3000]))
|
|
|
|
|
|
def sha(*parts):
|
|
h = hashlib.sha256()
|
|
for p in parts:
|
|
h.update(p)
|
|
return h.digest()
|
|
|
|
|
|
# ---------------------------------------------------------------- the extension's own table
|
|
|
|
def reasons():
|
|
"""§10's codes, from the text."""
|
|
with open(EXTENSION, encoding="utf-8") as f:
|
|
text = f.read()
|
|
at = text.index("\n## 10. Reasons")
|
|
end = text.index("\n## ", at + 1)
|
|
return [line.split("`")[1] for line in text[at:end].splitlines() if line.startswith("| `")]
|
|
|
|
|
|
REASONS = reasons()
|
|
|
|
|
|
# ---------------------------------------------------------------- §3 the ad policy
|
|
|
|
def is_base(b):
|
|
"""§3.1."""
|
|
if not b.startswith(b"https://"):
|
|
return False
|
|
rest = b[8:]
|
|
return (1 <= len(rest) <= 492 and rest[-1:] == b"/"
|
|
and all(0x21 <= c <= 0x7E and c not in b"\"#&'<>?\\" for c in rest))
|
|
|
|
|
|
def read_policy(b):
|
|
"""§3.3: the policy, or Refuse."""
|
|
if len(b) < 4:
|
|
raise Refuse("truncated")
|
|
if b[:4] != POLICY_MAGIC:
|
|
raise Refuse("malformed")
|
|
if len(b) < 5:
|
|
raise Refuse("truncated")
|
|
if b[4] not in (1, 2):
|
|
raise Refuse("unknown_version")
|
|
r = LEV.Reader(b, "truncated")
|
|
r.take(5)
|
|
p = {"room": r.take(32), "decider_grant": r.take(32), "slots": r.u8(), "open_avail_ms": r.u32(),
|
|
"preload_ms": r.u32(), "restrict": r.u8(), "skip_after_s": r.u16(), "snap": r.u8()}
|
|
p["asset_base"] = r.take(r.u16())
|
|
p["creative_base"] = r.take(r.u16())
|
|
p["presentation"] = None
|
|
if b[4] == 2: # ad-presentation §8
|
|
p["presentation"] = {"require": r.u8(), "rule": r.u8(), "token_type": r.u16(), "issuer": r.take(r.u16())}
|
|
p["body"] = b[:r.p]
|
|
p["sig"] = r.take(64)
|
|
if not r.done():
|
|
raise Refuse("malformed")
|
|
if (not 1 <= p["slots"] <= 16 or p["restrict"] > 3 or p["snap"] > 3
|
|
or not is_base(p["asset_base"]) or not is_base(p["creative_base"])
|
|
or (p["presentation"] is not None and not presentation_ok(p["presentation"]))):
|
|
raise Refuse("ad_policy_malformed")
|
|
p["asset_base"], p["creative_base"] = p["asset_base"].decode(), p["creative_base"].decode()
|
|
return p
|
|
|
|
|
|
def presentation_ok(q):
|
|
"""ad-presentation §8's rules for a policy of version 2."""
|
|
req, claim, attest = q["require"], q["require"] & 1, q["require"] & 2
|
|
if req > 7 or (not claim and req != 0) or q["rule"] > 2 or bool(claim) != (q["rule"] != 0):
|
|
return False
|
|
if attest:
|
|
return q["token_type"] in (1, 2) and len(q["issuer"]) > 0
|
|
return q["token_type"] == 0 and q["issuer"] == b""
|
|
|
|
|
|
def requires_claims(p):
|
|
return p["presentation"] is not None and p["presentation"]["require"] & 1 == 1
|
|
|
|
|
|
# ---------------------------------------------------------------- §4 breaks
|
|
|
|
def read_annotation(b):
|
|
"""§4.2."""
|
|
if len(b) != 38 or b[:4] != ANNOTATION_MAGIC or b[4] != 1 or b[37] not in (MID, POST):
|
|
raise Refuse("annotation_malformed")
|
|
return {"cue": b[5:37], "placement": b[37]}
|
|
|
|
|
|
def cue_of(ev):
|
|
return LEV.read_payload(CUE, ev["payload"], 65535)
|
|
|
|
|
|
def schedule(held, p):
|
|
"""§4.3 over the held events in the merged order: (breaks, chapters, written) or Stop."""
|
|
breaks, chapters, written = [], [], []
|
|
open_, last_end, closed = None, None, False
|
|
for k, (c, i, ev) in enumerate(held):
|
|
where = ["record", c, i]
|
|
if ev["kind"] == CUE:
|
|
q = cue_of(ev)
|
|
if closed:
|
|
raise Stop("cue_after_close", where)
|
|
t = ev["t"]
|
|
overlaps = open_ is not None or (last_end is not None and t < last_end)
|
|
if q["type"] == OUT:
|
|
if overlaps:
|
|
raise Stop("break_overlap", where)
|
|
avail = q["duration_ms"] or p["open_avail_ms"]
|
|
if avail == 0:
|
|
raise Stop("break_duration_unknown", where)
|
|
open_ = len(breaks)
|
|
written.append(("open", len(breaks)))
|
|
breaks.append({"digest": ev["digest"], "author": ev["principal"], "cue_id": q["cue_id"], "t": t,
|
|
"type": OUT, "placement": MID, "planned_ms": q["duration_ms"], "avail_ms": avail,
|
|
"duration_ms": None, "closed_by": None, "end": None})
|
|
elif q["type"] == IN:
|
|
b = breaks[open_] if open_ is not None else None
|
|
if b is None or b["cue_id"] != q["cue_id"] or b["author"] != ev["principal"]:
|
|
raise Stop("cue_in_unmatched", where)
|
|
duration = (t - b["t"]) // 1000
|
|
if duration == 0:
|
|
raise Stop("break_zero_duration", where)
|
|
if q["duration_ms"] != 0 and q["duration_ms"] != duration:
|
|
raise Stop("cue_duration_mismatch", where)
|
|
b.update(duration_ms=duration, closed_by=ev["digest"], end=t)
|
|
written.append(("close", open_))
|
|
open_, last_end = None, t
|
|
elif q["type"] == CHAPTER:
|
|
chapters.append({"digest": ev["digest"], "t": t, "duration_ms": q["duration_ms"]})
|
|
else:
|
|
if q["duration_ms"] == 0:
|
|
raise Stop("break_zero_duration", where)
|
|
if overlaps:
|
|
raise Stop("break_overlap", where)
|
|
last_end = t + q["duration_ms"] * 1000
|
|
written.append(("open", len(breaks)))
|
|
breaks.append({"digest": ev["digest"], "author": ev["principal"], "cue_id": q["cue_id"], "t": t,
|
|
"type": BREAK, "placement": MID, "planned_ms": q["duration_ms"],
|
|
"avail_ms": q["duration_ms"], "duration_ms": q["duration_ms"], "closed_by": None,
|
|
"end": last_end})
|
|
elif ev["kind"] == META and ev["payload"][:4] == ANNOTATION_MAGIC:
|
|
try:
|
|
a = read_annotation(ev["payload"])
|
|
except Refuse as e:
|
|
raise Stop(e.reason, where)
|
|
before = next((x for _, _, x in reversed(held[:k])
|
|
if x["principal"] == ev["principal"] and x["seq"] == ev["seq"] - 1), None)
|
|
if (before is None or before["digest"] != a["cue"] or before["kind"] != CUE or before["t"] != ev["t"]
|
|
or cue_of(before)["type"] != BREAK):
|
|
raise Stop("annotation_misplaced", where)
|
|
if a["placement"] == POST:
|
|
for b in breaks:
|
|
if b["digest"] == a["cue"]:
|
|
b["placement"] = POST
|
|
closed = True
|
|
return breaks, chapters, written
|
|
|
|
|
|
# ---------------------------------------------------------------- §4.4, §5, §6, §7
|
|
|
|
def break_id(d):
|
|
return "wai-ad-" + d.hex()
|
|
|
|
|
|
def event_id(d):
|
|
return int.from_bytes(d[:8], "big") & ((1 << 53) - 1)
|
|
|
|
|
|
def secs(ms):
|
|
return "%d.%03d" % (ms // 1000, ms % 1000)
|
|
|
|
|
|
def hls_uri(p, d):
|
|
i = break_id(d)
|
|
return "%s%s.json?_HLS_interstitial_id=%s" % (p["asset_base"], i, i)
|
|
|
|
|
|
def dash_uri(p, d):
|
|
return "%s%s.mpd" % (p["asset_base"], break_id(d))
|
|
|
|
|
|
def date(t0, t, less=0):
|
|
return LEV.iso((t0 + t) // 1000 - less)
|
|
|
|
|
|
def bits(v, names):
|
|
return ",".join(n for k, n in enumerate(names) if v & (1 << k))
|
|
|
|
|
|
def interstitial(p, t0, b):
|
|
"""§5.1."""
|
|
post = b["placement"] == POST
|
|
a = ['ID="%s"' % break_id(b["digest"]), 'CLASS="%s"' % INTERSTITIAL, 'START-DATE="%s"' % date(t0, b["t"]),
|
|
'CUE="%s"' % ("POST,ONCE" if post else "ONCE")]
|
|
known = b["planned_ms"] if b["type"] == BREAK else None
|
|
if known is not None:
|
|
a.append("DURATION=" + secs(known))
|
|
elif b["planned_ms"]:
|
|
a.append("PLANNED-DURATION=" + secs(b["planned_ms"]))
|
|
a.append('X-ASSET-LIST="%s"' % hls_uri(p, b["digest"]))
|
|
if known is not None:
|
|
if not post:
|
|
a.append("X-RESUME-OFFSET=" + secs(known))
|
|
a.append("X-PLAYOUT-LIMIT=" + secs(known))
|
|
if p["snap"]:
|
|
a.append('X-SNAP="%s"' % bits(p["snap"], ["OUT", "IN"]))
|
|
if p["restrict"]:
|
|
a.append('X-RESTRICT="%s"' % bits(p["restrict"], ["SKIP", "JUMP"]))
|
|
if p["skip_after_s"] != NO_SKIP:
|
|
a.append("X-SKIP-CONTROL-OFFSET=%d" % p["skip_after_s"])
|
|
return "#EXT-X-DATERANGE:" + ",".join(a)
|
|
|
|
|
|
def closing(b):
|
|
"""§5.2."""
|
|
s = secs(b["duration_ms"])
|
|
return '#EXT-X-DATERANGE:ID="%s",CLASS="%s",DURATION=%s,X-RESUME-OFFSET=%s,X-PLAYOUT-LIMIT=%s' % (
|
|
break_id(b["digest"]), INTERSTITIAL, s, s, s)
|
|
|
|
|
|
def preload(p, t0, b):
|
|
"""§5.3."""
|
|
i = break_id(b["digest"])
|
|
return ('#EXT-X-DATERANGE:ID="%s-preload",CLASS="%s",START-DATE="%s",DURATION=%s,X-URI="%s",X-TARGET-ID="%s",'
|
|
'X-TARGET-CLASS="%s"' % (i, PRELOAD, date(t0, b["t"], p["preload_ms"]), secs(p["preload_ms"]),
|
|
hls_uri(p, b["digest"]), i, INTERSTITIAL))
|
|
|
|
|
|
def hls_tags(p, t0, breaks, written):
|
|
"""§5.4: in the order a packager writes them."""
|
|
out = []
|
|
for what, i in written:
|
|
if what == "open":
|
|
if p["preload_ms"]:
|
|
out.append(preload(p, t0, breaks[i]))
|
|
out.append(interstitial(p, t0, breaks[i]))
|
|
else:
|
|
out.append(closing(breaks[i]))
|
|
return out
|
|
|
|
|
|
def dash_event(p, pt0, b, closed):
|
|
"""§6.2."""
|
|
known = b["duration_ms"] if closed else None
|
|
post = b["placement"] == POST
|
|
e = '<Event presentationTime="%d"' % (pt0 + b["t"] // 1000)
|
|
if known is not None:
|
|
e += ' duration="%d"' % known
|
|
e += ' id="%d">' % event_id(b["digest"])
|
|
child = "InsertPresentation" if post else "ReplacePresentation"
|
|
e += '<%s uri="%s"' % (child, dash_uri(p, b["digest"]))
|
|
if p["preload_ms"]:
|
|
e += ' earliestResolutionTimeOffset="%d"' % p["preload_ms"]
|
|
if known is not None:
|
|
e += ' maxDuration="%d"' % known
|
|
e += ' executeOnce="true"'
|
|
skip = p["skip_after_s"] if p["skip_after_s"] != NO_SKIP else -(-b["avail_ms"] // 1000)
|
|
e += ' skipAfter="PT%dS"' % skip
|
|
if known is not None and not post:
|
|
e += ' returnOffset="%d"' % known
|
|
return e + "/></Event>"
|
|
|
|
|
|
def dash_streams(p, pt0, breaks):
|
|
"""§6.3."""
|
|
out = ""
|
|
for scheme, post in ((REPLACE, False), (INSERT, True)):
|
|
evs = [dash_event(p, pt0, b, True) for b in breaks if (b["placement"] == POST) == post]
|
|
if evs:
|
|
out += '<EventStream schemeIdUri="%s" timescale="1000">%s</EventStream>' % (scheme, "".join(evs))
|
|
return out
|
|
|
|
|
|
def decide(pod, slots, avail, funds):
|
|
"""§7.1: the fill, each (slot, ad_id, creative_hash, duration_ms, price_micros)."""
|
|
fill = []
|
|
for c in sorted(pod, key=lambda c: (-c["price_micros"], c["ad_id"].encode())):
|
|
if len(fill) >= slots:
|
|
break
|
|
if c["duration_ms"] <= avail and c["price_micros"] <= funds:
|
|
fill.append((len(fill), c["ad_id"], c["creative_hash"], c["duration_ms"], c["price_micros"]))
|
|
avail -= c["duration_ms"]
|
|
funds -= c["price_micros"]
|
|
return fill
|
|
|
|
|
|
def asset_list(p, fmt, fill):
|
|
"""§7.2, §7.3: the exact bytes."""
|
|
if fmt == HLS:
|
|
assets = ['{"URI":"%s%s.m3u8","DURATION":%s}' % (p["creative_base"], h.hex(), secs(d)) for _, _, h, d, _ in fill]
|
|
return ('{"ASSETS":[%s]}' % ",".join(assets)).encode()
|
|
if not fill:
|
|
return b""
|
|
periods = "".join('<Period id="%d" duration="PT%sS"><ImportedMPD>%s%s.mpd</ImportedMPD></Period>'
|
|
% (s, secs(d), p["creative_base"], h.hex()) for s, _, h, d, _ in fill)
|
|
return ('<?xml version="1.0" encoding="UTF-8"?>\n<MPD xmlns="urn:mpeg:dash:schema:mpd:2011" '
|
|
'profiles="urn:mpeg:dash:profile:list:2024" type="list" minBufferTime="PT1S">%s</MPD>' % periods).encode()
|
|
|
|
|
|
# ---------------------------------------------------------------- §8 the break form
|
|
|
|
MEMBERS = ["form", "room", "cue", "session", "start_offset_ms", "format", "asset_list_sha256", "avail_ms", "slots",
|
|
"funds_remaining_micros", "pod_hash", "impressions", "merkle_root", "budget_spent_micros",
|
|
"work_decide_ops", "joules_micro", "energy_provenance", "grant", "signer_pubkey", "signer_id", "sig"]
|
|
EVIDENCE = {"OnChipCounter": ["energy_uncertainty", "energy_counter_filtering"],
|
|
"CalibratedInstrument": ["energy_uncertainty", "energy_calibration_ref"],
|
|
"ModelBased": [], "Estimator": []}
|
|
CLASS_TAG = {"ModelBased": 2, "Estimator": 3, "OnChipCounter": 4, "CalibratedInstrument": 5}
|
|
FILTERING = {"Unknown": 0, "Off": 1, "On": 2}
|
|
|
|
|
|
def u(v, top=U64):
|
|
if not isinstance(v, int) or isinstance(v, bool) or not 0 <= v <= top:
|
|
raise Refuse("receipt_malformed")
|
|
return v
|
|
|
|
|
|
def hexn(v, n):
|
|
if not isinstance(v, str) or len(v) != 2 * n or any(c not in "0123456789abcdef" for c in v):
|
|
raise Refuse("receipt_malformed")
|
|
return bytes.fromhex(v)
|
|
|
|
|
|
def text(v):
|
|
if not isinstance(v, str):
|
|
raise Refuse("receipt_malformed")
|
|
return v
|
|
|
|
|
|
def read_receipt(o):
|
|
"""§8.2: the receipt, or Refuse receipt_malformed."""
|
|
if not isinstance(o, dict):
|
|
raise Refuse("receipt_malformed")
|
|
label = o.get("energy_provenance")
|
|
if label is not None and label not in EVIDENCE:
|
|
raise Refuse("receipt_malformed")
|
|
want = set(MEMBERS) | set(EVIDENCE.get(label, []))
|
|
if set(o) != want or u(o["form"]) != 2:
|
|
raise Refuse("receipt_malformed")
|
|
r = {"room": hexn(o["room"], 32), "cue": hexn(o["cue"], 32), "session": hexn(o["session"], 16),
|
|
"start_offset_ms": u(o["start_offset_ms"]), "format": u(o["format"], 255),
|
|
"asset_list_sha256": hexn(o["asset_list_sha256"], 32), "avail_ms": u(o["avail_ms"]), "slots": u(o["slots"]),
|
|
"funds_remaining_micros": u(o["funds_remaining_micros"]), "pod_hash": hexn(o["pod_hash"], 32),
|
|
"merkle_root": hexn(o["merkle_root"], 32), "budget_spent_micros": u(o["budget_spent_micros"]),
|
|
"work_decide_ops": u(o["work_decide_ops"]), "joules_micro": u(o["joules_micro"]),
|
|
"signer_pubkey": hexn(o["signer_pubkey"], 32), "signer_id": text(o["signer_id"]), "sig": hexn(o["sig"], 64)}
|
|
if not isinstance(o["impressions"], list):
|
|
raise Refuse("receipt_malformed")
|
|
r["impressions"] = []
|
|
for i in o["impressions"]:
|
|
if not isinstance(i, dict) or set(i) != {"slot", "ad_id", "creative_hash", "duration_ms", "price_micros",
|
|
"quartiles_reached"}:
|
|
raise Refuse("receipt_malformed")
|
|
r["impressions"].append({"slot": u(i["slot"]), "ad_id": text(i["ad_id"]),
|
|
"creative_hash": hexn(i["creative_hash"], 32), "duration_ms": u(i["duration_ms"]),
|
|
"price_micros": u(i["price_micros"]), "quartiles_reached": u(i["quartiles_reached"], 255)})
|
|
g = o["grant"]
|
|
if not isinstance(g, dict) or set(g) != {"grant_hash", "capability", "joule_ceiling_micro", "funds_ceiling"}:
|
|
raise Refuse("receipt_malformed")
|
|
r["grant"] = {"grant_hash": hexn(g["grant_hash"], 32), "capability": text(g["capability"]),
|
|
"joule_ceiling_micro": u(g["joule_ceiling_micro"]),
|
|
"funds_ceiling": None if g["funds_ceiling"] is None else u(g["funds_ceiling"])}
|
|
r["class"] = None
|
|
if label is not None:
|
|
c = {"label": label}
|
|
if label in ("OnChipCounter", "CalibratedInstrument"):
|
|
unc = o["energy_uncertainty"]
|
|
if not isinstance(unc, dict) or set(unc) != {"relative_ppm", "window_us"}:
|
|
raise Refuse("receipt_malformed")
|
|
c["ppm"], c["window"] = u(unc["relative_ppm"], 2 ** 32 - 1), u(unc["window_us"])
|
|
if label == "OnChipCounter":
|
|
if o["energy_counter_filtering"] not in FILTERING:
|
|
raise Refuse("receipt_malformed")
|
|
c["filtering"] = FILTERING[o["energy_counter_filtering"]]
|
|
if label == "CalibratedInstrument":
|
|
c["ref"] = text(o["energy_calibration_ref"])
|
|
r["class"] = c
|
|
return r
|
|
|
|
|
|
def b3(*parts):
|
|
return B3.blake3(b"".join(parts))
|
|
|
|
|
|
def be64(v):
|
|
return struct.pack(">Q", v)
|
|
|
|
|
|
def pod_hash(pod):
|
|
"""§8.3."""
|
|
out = [POD_DOMAIN, be64(len(pod))]
|
|
for c in pod:
|
|
a = c["ad_id"].encode()
|
|
out += [struct.pack(">I", len(a)), a, c["creative_hash"], be64(c["duration_ms"]), be64(c["price_micros"])]
|
|
return b3(*out)
|
|
|
|
|
|
def leaf(i):
|
|
a = i["ad_id"].encode()
|
|
return b3(IMP_DOMAIN, be64(i["slot"]), struct.pack(">I", len(a)), a, i["creative_hash"], be64(i["duration_ms"]),
|
|
be64(i["price_micros"]), bytes([i["quartiles_reached"]]))
|
|
|
|
|
|
def merkle(hashes):
|
|
"""jwp-receipts §2.2, each leaf of its first form."""
|
|
if not hashes:
|
|
return bytes(32)
|
|
level = [b3(LEAF, h) for h in hashes]
|
|
while len(level) > 1:
|
|
if len(level) % 2:
|
|
level.append(level[-1])
|
|
level = [b3(NODE, level[k], level[k + 1]) for k in range(0, len(level), 2)]
|
|
return level[0]
|
|
|
|
|
|
def class_bytes(c):
|
|
"""energy-measurement §2.7's signed encoding."""
|
|
if c is None:
|
|
return b"\x00"
|
|
out = bytes([CLASS_TAG[c["label"]]])
|
|
if c["label"] == "OnChipCounter":
|
|
out += struct.pack(">IQ", c["ppm"], c["window"]) + bytes([c["filtering"]])
|
|
elif c["label"] == "CalibratedInstrument":
|
|
ref = c["ref"].encode()
|
|
out += struct.pack(">IQ", c["ppm"], c["window"]) + struct.pack(">I", len(ref)) + ref
|
|
return out
|
|
|
|
|
|
def payload(r):
|
|
"""§8.4."""
|
|
g = r["grant"]
|
|
cap, sid = g["capability"].encode(), r["signer_id"].encode()
|
|
out = [BREAK_DOMAIN, r["room"], r["cue"], r["session"], be64(r["start_offset_ms"]), bytes([r["format"]]),
|
|
r["asset_list_sha256"], be64(r["avail_ms"]), be64(r["slots"]), be64(r["funds_remaining_micros"]),
|
|
r["pod_hash"], r["merkle_root"], be64(r["budget_spent_micros"]), be64(r["work_decide_ops"]),
|
|
be64(r["joules_micro"]), class_bytes(r["class"]), g["grant_hash"], struct.pack(">I", len(cap)), cap,
|
|
be64(g["joule_ceiling_micro"])]
|
|
out.append(b"\x00" if g["funds_ceiling"] is None else b"\x01" + be64(g["funds_ceiling"]))
|
|
out.append(be64(len(r["impressions"])))
|
|
out += [leaf(i) for i in r["impressions"]]
|
|
out += [r["signer_pubkey"], struct.pack(">I", len(sid)), sid]
|
|
return b"".join(out)
|
|
|
|
|
|
def own_rules(r):
|
|
"""§8.5: None, or the refusal."""
|
|
if r["format"] not in (HLS, DASH):
|
|
return "receipt_invalid"
|
|
bad = ED.verify(r["signer_pubkey"], payload(r), r["sig"])
|
|
if bad:
|
|
return bad
|
|
imps = r["impressions"]
|
|
spend = sum(i["price_micros"] for i in imps if i["quartiles_reached"] >= 1)
|
|
if merkle([leaf(i) for i in imps]) != r["merkle_root"] or spend != r["budget_spent_micros"]:
|
|
return "receipt_invalid"
|
|
f = r["grant"]["funds_ceiling"]
|
|
if f is not None and spend > f * 10 ** 6:
|
|
return "receipt_invalid"
|
|
if any(i["slot"] != k or i["quartiles_reached"] > 5 for k, i in enumerate(imps)) or r["work_decide_ops"] < len(imps):
|
|
return "receipt_invalid"
|
|
c, j = r["class"], r["joules_micro"]
|
|
if c is None:
|
|
return None if j == 0 else "receipt_invalid"
|
|
if j == 0:
|
|
return "receipt_invalid"
|
|
if c["label"] in ("OnChipCounter", "CalibratedInstrument"):
|
|
if c["ppm"] == 0 or c["window"] == 0 or j * c["ppm"] < FLOOR:
|
|
return "receipt_invalid"
|
|
if c["label"] == "CalibratedInstrument" and all(ch in WHITE_SPACE for ch in c["ref"]):
|
|
return "receipt_invalid"
|
|
return None
|
|
|
|
|
|
def identity(r):
|
|
return b3(ID_DOMAIN, payload(r), r["sig"])
|
|
|
|
|
|
# ---------------------------------------------------------------- §9 verification
|
|
|
|
class Stop(Exception):
|
|
def __init__(self, reason, at):
|
|
super().__init__(reason)
|
|
self.reason, self.at = reason, at
|
|
|
|
|
|
def ads(v, judge=None):
|
|
"""§9: a dict of what the corpus states for vector v, and the run's HLS tags. judge, when
|
|
given, makes the judge of a policy's presentation claims (ad-presentation §9)."""
|
|
descriptor = bytes.fromhex(v["descriptor"])
|
|
grants = [bytes.fromhex(g) for g in v["grants"]]
|
|
chains = [[bytes.fromhex(h) for h in c] for c in v["chains"]]
|
|
room = LRV.verify_room(descriptor, grants, v["key_log"], chains) # step 1
|
|
if room["outcome"] == "refuse":
|
|
return {"outcome": "refuse", "reason": room["reason"], "at": ["room"] + room["at"]}, None
|
|
if room["outcome"] == "gap":
|
|
return {"outcome": "gap", "gaps": room["gaps"]}, None
|
|
try:
|
|
return accept(v, descriptor, grants, chains, room, judge)
|
|
except Stop as e:
|
|
return {"outcome": "refuse", "reason": e.reason, "at": e.at}, None
|
|
|
|
|
|
def accept(v, descriptor, grants, chains, room, judge=None):
|
|
d = LRV.read_descriptor(descriptor)
|
|
held_grants, _ = LRV.check_grants(d, grants)
|
|
try: # step 2
|
|
p = read_policy(bytes.fromhex(v["policy"]))
|
|
if p["room"] != d["room"]:
|
|
raise Refuse("wrong_room")
|
|
bad = ED.verify(d["authority"], POLICY_DOMAIN + p["body"], p["sig"])
|
|
if bad:
|
|
raise Refuse(bad)
|
|
decider = next((g for g in held_grants if g["digest"] == p["decider_grant"]), None)
|
|
if decider is None:
|
|
raise Refuse("decider_grant_unknown")
|
|
if LRV.revoked(decider, LRV.check_key_log(d, v["key_log"])):
|
|
raise Refuse("grant_revoked")
|
|
except Refuse as e:
|
|
raise Stop(e.reason, ["policy"])
|
|
held, seen = [], set() # step 3
|
|
for c, i, _ in room["order"]:
|
|
ev = LEV.read_record(chains[c][i])
|
|
if ev["digest"] not in seen:
|
|
seen.add(ev["digest"])
|
|
held.append((c, i, ev))
|
|
breaks, chapters, written = schedule(held, p)
|
|
t0, pt0 = d["t0"], v["pt0_ms"] # step 4
|
|
tags = hls_tags(p, t0, breaks, written)
|
|
funds = None if decider["funds"] == NO_CEILING else decider["funds"]
|
|
claims = None
|
|
if requires_claims(p): # ad-presentation §9.2 steps 1, 2
|
|
if judge is None:
|
|
raise ValueError("a policy that requires presentation claims: verify it with ad_presentation_verify.py")
|
|
claims = judge(v, d, p)
|
|
stated, sessions, total = [], set(), 0 # step 5
|
|
for k, (o, pod_json) in enumerate(zip(v["receipts"], v["pods"] + [None] * len(v["receipts"]))):
|
|
at = ["receipt", k]
|
|
try:
|
|
r = read_receipt(o)
|
|
except Refuse as e:
|
|
raise Stop(e.reason, at)
|
|
if r["room"] != d["room"]:
|
|
raise Stop("wrong_room", at)
|
|
bad = own_rules(r)
|
|
if bad:
|
|
raise Stop(bad, at)
|
|
if r["signer_pubkey"] != decider["principal"]:
|
|
raise Stop("receipt_not_decider", at)
|
|
g = r["grant"]
|
|
if (g["grant_hash"] != decider["digest"] or g["capability"] != "wai.live.room"
|
|
or g["joule_ceiling_micro"] != decider["joules"] or g["funds_ceiling"] != funds):
|
|
raise Stop("receipt_grant_mismatch", at)
|
|
b = next((x for x in breaks if x["digest"] == r["cue"]), None)
|
|
if b is None:
|
|
raise Stop("receipt_break_unknown", at)
|
|
if not LRV.good_at(decider, d, b["t"]):
|
|
raise Stop("outside_grant_window", at)
|
|
pod = [{"ad_id": c["ad_id"], "creative_hash": bytes.fromhex(c["creative_hash"]), "duration_ms": c["duration_ms"],
|
|
"price_micros": c["price_micros"]} for c in (pod_json or [])]
|
|
if r["pod_hash"] != pod_hash(pod):
|
|
raise Stop("pod_mismatch", at)
|
|
fill = decide(pod, p["slots"], max(b["avail_ms"] - r["start_offset_ms"], 0), r["funds_remaining_micros"])
|
|
got = [(i["slot"], i["ad_id"], i["creative_hash"], i["duration_ms"], i["price_micros"]) for i in r["impressions"]]
|
|
if r["avail_ms"] != b["avail_ms"] or r["slots"] != p["slots"] or got != fill:
|
|
raise Stop("decision_mismatch", at)
|
|
listed = asset_list(p, r["format"], fill)
|
|
if r["asset_list_sha256"] != sha(listed):
|
|
raise Stop("asset_list_mismatch", at)
|
|
if (r["cue"], r["session"]) in sessions:
|
|
raise Stop("receipt_duplicate", at)
|
|
sessions.add((r["cue"], r["session"]))
|
|
counted, extra = r["budget_spent_micros"], {} # step 12
|
|
if claims is not None:
|
|
counted, extra = claims.receipt(k, r)
|
|
total += counted
|
|
if funds is not None and (r["funds_remaining_micros"] > funds * 10 ** 6 or total > funds * 10 ** 6):
|
|
raise Stop("funds_exceeded", at)
|
|
fact = {"identity": identity(r).hex(), "asset_list": listed.decode(), "spend_micros": r["budget_spent_micros"]}
|
|
fact.update(extra)
|
|
stated.append(fact)
|
|
out = {
|
|
"outcome": "accept",
|
|
"room": d["room"].hex(),
|
|
"breaks": [{"digest": b["digest"].hex(), "id": break_id(b["digest"]), "event_id": event_id(b["digest"]),
|
|
"author": b["author"].hex(), "cue_id": b["cue_id"], "t": b["t"],
|
|
"type": "out" if b["type"] == OUT else "break", "placement": "post" if b["placement"] == POST else "mid",
|
|
"planned_ms": b["planned_ms"], "avail_ms": b["avail_ms"], "duration_ms": b["duration_ms"],
|
|
"closed_by": b["closed_by"].hex() if b["closed_by"] else None, "end": b["end"],
|
|
"hls_list_uri": hls_uri(p, b["digest"]), "dash_list_uri": dash_uri(p, b["digest"])} for b in breaks],
|
|
"chapters": [{"digest": c["digest"].hex(), "t": c["t"], "duration_ms": c["duration_ms"]} for c in chapters],
|
|
"hls": v["name"] + ".m3u8",
|
|
"dash": dash_streams(p, pt0, breaks),
|
|
"dash_open": [{"break": k, "event": dash_event(p, pt0, b, False)} for k, b in enumerate(breaks) if b["type"] == OUT],
|
|
"receipt_facts": stated,
|
|
"spend_micros": min(total, U64),
|
|
}
|
|
if claims is not None:
|
|
out["claim_facts"] = claims.finish()
|
|
return out, "".join(t + "\n" for t in tags).encode()
|
|
|
|
|
|
# ---------------------------------------------------------------- the corpus
|
|
|
|
def check(doc, files, quiet=False):
|
|
"""Every room against its statement, and every tag file: the failures, and the files written."""
|
|
fails, written = [], {}
|
|
for v in doc.get("vectors", []):
|
|
try:
|
|
got, tags = ads(v)
|
|
except Exception as e: # a crash on a mutated copy is a failure to read it
|
|
got, tags = None, None
|
|
bad = ["the vector does not verify: %s: %s" % (type(e).__name__, e)]
|
|
else:
|
|
bad = [k for k in got if v.get(k) != got[k]]
|
|
bad += [k for k in ("outcome", "reason", "at", "gaps", "breaks", "hls", "dash", "receipt_facts")
|
|
if k in v and k not in got]
|
|
if tags is not None:
|
|
name = v["name"] + ".m3u8"
|
|
written[name] = tags
|
|
if files.get(name) != tags:
|
|
bad.append("its tag file %s" % name)
|
|
if not quiet:
|
|
what = "?" if got is None else got["outcome"] + (" " + got["reason"] if got["outcome"] == "refuse" else "")
|
|
print(("PASS" if not bad else "FAIL") + " room %s: %s" % (v.get("name"), what))
|
|
fails += ["room %s: %s differs from the corpus" % (v.get("name"), k) for k in bad]
|
|
return fails, written
|
|
|
|
|
|
def flip(h, at):
|
|
b = bytearray(bytes.fromhex(h))
|
|
b[at] ^= 0x01
|
|
return b.hex()
|
|
|
|
|
|
def mutations(doc, files):
|
|
"""Copies of the corpus, each with one defect a conforming verifier must catch."""
|
|
vn = {v["name"]: k for k, v in enumerate(doc["vectors"])}
|
|
live, mid, post, chap, dup = vn["live_break"], vn["mid_roll"], vn["post_roll"], vn["chapters"], vn["receipt_duplicate"]
|
|
|
|
def mutate(label, f):
|
|
d, fs = copy.deepcopy(doc), dict(files)
|
|
f(d["vectors"], fs)
|
|
return label, d, fs
|
|
|
|
def drop_last(vs, chain_of):
|
|
"""The last record of the chain holding a producer's cues."""
|
|
ch = vs[chain_of]["chains"]
|
|
longest = max(range(len(ch)), key=lambda c: len(ch[c]))
|
|
ch[longest].pop()
|
|
|
|
def swap_pod(vs, fs):
|
|
pod = vs[mid]["pods"][0]
|
|
pod[1], pod[3] = pod[3], pod[1]
|
|
|
|
def edit_file(name, a, b):
|
|
return lambda vs, fs: fs.update({name: fs[name].replace(a, b, 1)})
|
|
|
|
return [
|
|
mutate("a byte of the live break's policy signature", lambda vs, fs: vs[live].update(policy=flip(vs[live]["policy"], -2))),
|
|
mutate("the live break's in withheld from the end of its chain", lambda vs, fs: drop_last(vs, live)),
|
|
mutate("the post-roll's annotation withheld", lambda vs, fs: drop_last(vs, post)),
|
|
mutate("a stated asset list with one ad's duration changed",
|
|
lambda vs, fs: vs[live]["receipt_facts"][0].update(asset_list=vs[live]["receipt_facts"][0]["asset_list"].replace("15.000", "14.000", 1))),
|
|
mutate("the DASH events stated with executeOnce false",
|
|
lambda vs, fs: vs[live].update(dash=vs[live]["dash"].replace('executeOnce="true"', 'executeOnce="false"'))),
|
|
mutate("the open out's DASH event stated with its duration",
|
|
lambda vs, fs: vs[live]["dash_open"][0].update(event=vs[live]["dash"].split(">", 1)[1].rsplit("</EventStream>", 1)[0])),
|
|
mutate("a byte of the mid-roll's tag file", edit_file("mid_roll.m3u8", b"X-RESUME-OFFSET=30.000", b"X-RESUME-OFFSET=31.000")),
|
|
mutate("the post-roll's tag stated without POST", edit_file("post_roll.m3u8", b'CUE="POST,ONCE"', b'CUE="ONCE"')),
|
|
mutate("the mid-roll pod given in another order", swap_pod),
|
|
mutate("the mid-roll's presentation time of t0 changed", lambda vs, fs: vs[mid].update(pt0_ms=vs[mid]["pt0_ms"] + 1)),
|
|
mutate("a live break receipt's quartile raised",
|
|
lambda vs, fs: vs[live]["receipts"][0]["impressions"][1].update(quartiles_reached=4)),
|
|
mutate("a live break receipt's session changed",
|
|
lambda vs, fs: vs[live]["receipts"][1].update(session="00" * 16)),
|
|
mutate("the stated spend", lambda vs, fs: vs[live].update(spend_micros=vs[live]["spend_micros"] + 1)),
|
|
mutate("a chapter's stated length", lambda vs, fs: vs[chap]["chapters"][0].update(duration_ms=59999)),
|
|
mutate("the duplicated receipt given once", lambda vs, fs: vs[dup]["receipts"].pop()),
|
|
mutate("the funds control's second receipt given once",
|
|
lambda vs, fs: vs[vn["funds_exceeded"]]["receipts"].pop()),
|
|
mutate("a tag file removed", lambda vs, fs: fs.pop("chapters.m3u8")),
|
|
]
|
|
|
|
|
|
def main(corpus_dir):
|
|
fails = []
|
|
st = ED._self_test()
|
|
print(("PASS" if not st else "FAIL") + " Ed25519 strict rule self-test " + ",".join(st))
|
|
fails += ["ed25519 self-test"] if st else []
|
|
ok = len(REASONS) == 20
|
|
print(("PASS" if ok else "FAIL") + " §10 read from the extension: %d reasons" % len(REASONS))
|
|
fails += [] if ok else ["§10 does not read"]
|
|
with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as f:
|
|
doc = json.load(f)
|
|
files = {}
|
|
for name in sorted(os.listdir(corpus_dir)):
|
|
if name.endswith(".m3u8"):
|
|
with open(os.path.join(corpus_dir, name), "rb") as f:
|
|
files[name] = f.read()
|
|
got, written = check(doc, files)
|
|
fails += got
|
|
extra = sorted(set(os.listdir(corpus_dir)) - set(written) - {"vectors.json", "README.md"})
|
|
print(("PASS" if not extra else "FAIL") + " no file but vectors.json, README.md and the tag files the rooms write")
|
|
fails += ["an extra file: %s" % n for n in extra]
|
|
seen = {v.get("reason") for v in doc["vectors"]}
|
|
missing = [r for r in REASONS if r not in seen]
|
|
print(("PASS" if not missing else "FAIL") + " every reason of §10 appears in the corpus")
|
|
fails += ["no case gives %s" % r for r in missing]
|
|
for label, d, fs in mutations(doc, files):
|
|
caught, _ = check(d, fs, quiet=True)
|
|
print(("PASS" if caught else "FAIL") + " mutated copy fails: %s%s" % (label, " (%s)" % caught[0][:100] 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") + " live-ads corpus: %d rooms, %d tag files" % (len(doc["vectors"]), len(files)))
|
|
return 1 if fails else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) != 2:
|
|
print("usage: live_ads_verify.py <live-conformance/ads>", file=sys.stderr)
|
|
sys.exit(2)
|
|
sys.exit(main(sys.argv[1]))
|