f2acb11dd4
- The falsifications found the check that refuses an acknowledgement the record does not name guarded by no vector in either implementation: the corpus gains ack_not_named (audit_bundle_invalid), and the Python verifier can drop that check too, which the corpus now catches. - The collector's default port moves to 9694: 9690 is the stitched presentation tool's. - stream-audit §6.3 states that this revision claims every marked ad with WPC1, and that a stitched ad is claimed with the stitched presentation record where a deployment holds it, its bundle then carrying the manifests its claims name. Co-Authored-By: Claude Opus 5.5 <[email protected]>
786 lines
32 KiB
Python
786 lines
32 KiB
Python
#!/usr/bin/env python3
|
|
"""Independent verifier of the stream audit corpus (extensions/stream-audit.md).
|
|
|
|
Written from the extension's text. It shares no code with wai-rs, which generates the corpus, and
|
|
never runs it. Acknowledgements and their item lists are read by the settlement corpus's second
|
|
implementation, settlement_verify.py (settlement §3.6, the tree root of jwp-receipts §2.2); claims,
|
|
their quartile and their run by the ad-presentation corpus's, ad_presentation_verify.py
|
|
(ad-presentation §5, §6); the observation log and a figure's acquisition class by the
|
|
live-metering corpus's, live_metering_verify.py (live-metering §3.6, §6.1); and every signature
|
|
with ../claim-conformance/_ed25519.py, WAI's strict rule (SPEC §7 *Signatures*) written from RFC
|
|
8032. Everything else is here, from the standard library:
|
|
|
|
- the channel id, the session and an origin's upstream (§3, settlement §14.2);
|
|
- the session record: reading, rules and signature (§5);
|
|
- the bundle and its verification in §10.2's order, with each refusal's place;
|
|
- each claim's verdict and facts (§10.2 step 9);
|
|
- the channel report, its bytes and its recomputation (§11).
|
|
|
|
For every session and report it requires the outcome the corpus states; of the corpus, that every
|
|
reason of §14 and every code it borrows 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; and it drops each
|
|
of its own checks in turn and requires the corpus to catch the weakened verifier.
|
|
|
|
python3 stream_audit_verify.py <live-conformance/audit>
|
|
|
|
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"))
|
|
import live_events_verify as LEV # noqa: E402
|
|
import live_metering_verify as LMV # noqa: E402
|
|
import settlement_verify as SV # noqa: E402
|
|
import ad_presentation_verify as APV # noqa: E402
|
|
import _ed25519 as ED # noqa: E402
|
|
|
|
EXTENSION = os.path.join(HERE, "..", "extensions", "stream-audit.md")
|
|
|
|
Refuse = LEV.Refuse
|
|
|
|
SESSION_MAGIC = b"WAS1" # §5.1
|
|
SESSION_DOMAIN = b"wai:audit-session\x01" # §5.4
|
|
CHANNEL_DOMAIN = b"wai:audit-channel\x01" # §3.1
|
|
MARKER_DOMAIN = b"wai:audit-marker\x01" # §6.1
|
|
REPORT_DOMAIN = b"wai:audit-report\x01" # §11.3
|
|
ORIGIN_DOMAIN = b"wai:settle-origin\x01" # settlement §14.2
|
|
SESSION_KEY_DOMAIN = b"wai:ad-presentation-session\x01" # §3.2, ad-presentation §3
|
|
NO_STARTUP = 0xFFFFFFFF
|
|
QOE_MIN = 50
|
|
LIVE, ON_DEMAND = 1, 2
|
|
ENDED = ["ended", "user", "hidden", "idle", "error"] # §8.2
|
|
ROLES = ["origin", "delivery", "sink_decode", "display"] # §5.3
|
|
CLASS_NAMES = {1: "live", 2: "fetch", 3: "fill", 4: "ancillary"} # settlement §3.3
|
|
LABELS = {0: "unmetered", 2: "ModelBased", 3: "Estimator", 4: "OnChipCounter", 5: "CalibratedInstrument"}
|
|
AUDIBLE, AD = 1, 2
|
|
ZERO = bytes(32)
|
|
|
|
SKIP = set()
|
|
|
|
|
|
def reasons_of_extension():
|
|
"""§14's codes, read from the text."""
|
|
with open(EXTENSION, encoding="utf-8") as f:
|
|
text = f.read()
|
|
at = text.index("\n## 14. Reasons")
|
|
end = text.index("\n## ", at + 1)
|
|
return [line.split("`")[1] for line in text[at:end].splitlines() if line.startswith("| `")]
|
|
|
|
|
|
REASONS = reasons_of_extension()
|
|
REUSED = ["truncated", "malformed", "unknown_version", "bad_key", "bad_signature", "record_missing", "energy_invalid",
|
|
"qoe_invalid", "ack_invalid", "self_acknowledged", "items_invalid", "items_mismatch", "ack_outside_period",
|
|
"ack_overlap", "ack_chain_broken", "claim_malformed", "claim_duplicate", "claim_asset_mismatch"]
|
|
|
|
|
|
class Stop(Exception):
|
|
"""A refusal and its place (§10.2)."""
|
|
|
|
def __init__(self, reason, at):
|
|
super().__init__(reason)
|
|
self.reason, self.at = reason, at
|
|
|
|
|
|
def sha(*parts):
|
|
h = hashlib.sha256()
|
|
for p in parts:
|
|
h.update(p)
|
|
return h.digest()
|
|
|
|
|
|
def given_digest(b):
|
|
return sha(b[:max(len(b) - 64, 0)])
|
|
|
|
|
|
def channel_id(name):
|
|
"""§3.1, or None for a name that is not 1 to 255 bytes of printable ASCII."""
|
|
b = name.encode("utf-8")
|
|
if not 1 <= len(b) <= 255 or any(c < 0x21 or c > 0x7E for c in b):
|
|
return None
|
|
return sha(CHANNEL_DOMAIN, b)
|
|
|
|
|
|
def session_of(channel, key):
|
|
return sha(SESSION_KEY_DOMAIN, channel, key)[:16]
|
|
|
|
|
|
def upstream_of(origin):
|
|
return sha(ORIGIN_DOMAIN, origin)
|
|
|
|
|
|
def marker_digest(marker):
|
|
return sha(MARKER_DOMAIN, marker)
|
|
|
|
|
|
def unhex(s):
|
|
if not isinstance(s, str) or len(s) % 2 or not re.fullmatch(r"[0-9a-f]*", s):
|
|
return None
|
|
return bytes.fromhex(s)
|
|
|
|
|
|
def check_sig(key, domain, body, sig):
|
|
bad = ED.verify(key, domain + body, sig)
|
|
if bad:
|
|
raise Refuse(bad)
|
|
|
|
|
|
class R:
|
|
"""A cursor for the record and for live-metering's class reader; running out is `truncated`."""
|
|
|
|
def __init__(self, b, p=0):
|
|
self.b, self.p = b, p
|
|
|
|
def take(self, n):
|
|
if self.p + n > len(self.b):
|
|
raise Refuse("truncated")
|
|
s = self.b[self.p:self.p + n]
|
|
self.p += n
|
|
return s
|
|
|
|
def u8(self):
|
|
return self.take(1)[0]
|
|
|
|
def u16(self):
|
|
return struct.unpack("<H", self.take(2))[0]
|
|
|
|
def u32(self):
|
|
return struct.unpack("<I", self.take(4))[0]
|
|
|
|
def u64(self):
|
|
return struct.unpack("<Q", self.take(8))[0]
|
|
|
|
def be32(self):
|
|
return struct.unpack(">I", self.take(4))[0]
|
|
|
|
def be64(self):
|
|
return struct.unpack(">Q", self.take(8))[0]
|
|
|
|
|
|
# ---------------------------------------------------------------- §5 the session record
|
|
|
|
def origin_ok(o):
|
|
"""§5.5."""
|
|
if not 1 <= len(o) <= 2048 or any(c < 0x21 or c > 0x7E for c in o):
|
|
return False
|
|
for scheme in (b"https://", b"http://"):
|
|
if o.startswith(scheme):
|
|
rest = o[len(scheme):]
|
|
return rest != b"" and not any(c in b"/?#@" for c in rest)
|
|
return False
|
|
|
|
|
|
def ascending(v):
|
|
return all(x < y for x, y in zip(v, v[1:]))
|
|
|
|
|
|
def read_session(b):
|
|
"""§5.5: the record, its rules, its energy entries, then its signature."""
|
|
if len(b) < 4:
|
|
raise Refuse("truncated")
|
|
if b[:4] != SESSION_MAGIC:
|
|
raise Refuse("malformed")
|
|
if len(b) < 5:
|
|
raise Refuse("truncated")
|
|
if b[4] != 1:
|
|
raise Refuse("unknown_version")
|
|
r = R(b, 5)
|
|
s = {"channel": r.take(32), "owner": r.take(32), "player": r.take(32), "session": r.take(16), "presentation": r.u8(),
|
|
"start": r.u64(), "end": r.u64(), "ended": r.u8()}
|
|
s["origins"] = [r.take(r.u16()) for _ in range(r.u8())]
|
|
s["acks"] = [r.take(32) for _ in range(r.u16())]
|
|
s["claims"] = [r.take(32) for _ in range(r.u16())]
|
|
s["spans"] = []
|
|
for _ in range(r.u32()):
|
|
start, length, vis, flags = struct.unpack("<IIHB", r.take(11))
|
|
s["spans"].append({"start": start, "length": length, "visible": vis, "flags": flags})
|
|
s["startup_ms"], s["stalls"], s["stall_ms"] = r.u32(), r.u32(), r.u64()
|
|
s["repeat_objects"], s["repeat_bytes"] = r.u32(), r.u64()
|
|
s["log"], s["notes"] = r.take(32), r.take(32)
|
|
s["energy"] = []
|
|
for _ in range(r.u8()):
|
|
role, joules = r.u8(), r.u64()
|
|
cls = LMV.read_class(r)
|
|
s["energy"].append({"role": role, "joules": joules, "class": cls})
|
|
s["attachments"] = [r.take(32) for _ in range(r.u16())]
|
|
sig = r.take(64)
|
|
if r.p != len(b):
|
|
raise Refuse("malformed")
|
|
if "session-rules" not in SKIP and not session_rules(s):
|
|
raise Refuse("audit_session_invalid")
|
|
if "energy" not in SKIP and not all(LMV.energy_ok(e["joules"], e["class"]) for e in s["energy"]):
|
|
raise Refuse("energy_invalid")
|
|
if "session-signature" not in SKIP:
|
|
check_sig(s["player"], SESSION_DOMAIN, b[:-64], sig)
|
|
s["digest"] = sha(b[:-64])
|
|
return s
|
|
|
|
|
|
def session_rules(s):
|
|
limit = (s["end"] - s["start"]) // 1000 if s["end"] > s["start"] else 0
|
|
prev = 0
|
|
for x in s["spans"]:
|
|
if x["length"] < 1 or x["visible"] > 10000 or x["flags"] > 3 or x["start"] < prev or x["start"] + x["length"] > limit:
|
|
return False
|
|
prev = x["start"] + x["length"]
|
|
roles = [e["role"] for e in s["energy"]]
|
|
return (s["presentation"] in (LIVE, ON_DEMAND) and s["start"] < s["end"] and 1 <= s["ended"] <= 5
|
|
and s["origins"] and all(origin_ok(o) for o in s["origins"]) and ascending(s["origins"])
|
|
and s["acks"] and ascending(s["acks"]) and ascending(s["claims"]) and ascending(s["attachments"])
|
|
and (s["startup_ms"] == NO_STARTUP) == (not s["spans"])
|
|
and (s["stalls"] != 0 or s["stall_ms"] == 0)
|
|
and (s["repeat_bytes"] == 0 if s["repeat_objects"] == 0 else s["repeat_bytes"] >= s["repeat_objects"])
|
|
and s["log"] != ZERO and ascending(roles) and all(1 <= x <= 4 for x in roles))
|
|
|
|
|
|
# ---------------------------------------------------------------- §9 the bundle
|
|
|
|
def exact(v, members):
|
|
return isinstance(v, dict) and set(v) == set(members)
|
|
|
|
|
|
def read_bundle(raw):
|
|
"""§9: the bundle, or Stop(audit_bundle_invalid)."""
|
|
def bad():
|
|
return Stop("audit_bundle_invalid", ["bundle"])
|
|
try:
|
|
v = json.loads(raw.decode("utf-8"))
|
|
except (UnicodeDecodeError, ValueError):
|
|
raise bad()
|
|
if not exact(v, ["v", "session", "acks", "claims", "log", "notes", "attachments"]) or v["v"] != 1 or isinstance(v["v"], bool):
|
|
raise bad()
|
|
out = {"session": unhex(v["session"]), "acks": [], "claims": [], "attachments": []}
|
|
if out["session"] is None or not isinstance(v["acks"], list) or not isinstance(v["claims"], list) or not isinstance(v["attachments"], list):
|
|
raise bad()
|
|
for a in v["acks"]:
|
|
if not exact(a, ["record", "items"]) or not isinstance(a["items"], list):
|
|
raise bad()
|
|
rec, items = unhex(a["record"]), [unhex(i) for i in a["items"]]
|
|
if rec is None or any(i is None for i in items):
|
|
raise bad()
|
|
out["acks"].append((rec, items))
|
|
for c in v["claims"]:
|
|
if not exact(c, ["claim", "marker", "creative"]) or not isinstance(c["marker"], str) or not isinstance(c["creative"], list) or not c["creative"]:
|
|
raise bad()
|
|
objects = []
|
|
for o in c["creative"]:
|
|
if not exact(o, ["content_hash", "duration_ms"]):
|
|
raise bad()
|
|
h, d = unhex(o["content_hash"]), o["duration_ms"]
|
|
if h is None or len(h) != 32 or not isinstance(d, int) or isinstance(d, bool) or not 1 <= d <= 0xFFFFFFFF:
|
|
raise bad()
|
|
objects.append({"content_hash": h, "duration_ms": d})
|
|
claim = unhex(c["claim"])
|
|
if claim is None:
|
|
raise bad()
|
|
out["claims"].append({"claim": claim, "marker": c["marker"].encode("utf-8"), "creative": objects})
|
|
if not isinstance(v["log"], str) or not (v["notes"] is None or isinstance(v["notes"], str)):
|
|
raise bad()
|
|
out["log"] = v["log"].encode("utf-8")
|
|
out["notes"] = None if v["notes"] is None else v["notes"].encode("utf-8")
|
|
for a in v["attachments"]:
|
|
if not exact(a, ["name", "record"]) or not isinstance(a["name"], str) or unhex(a["record"]) is None:
|
|
raise bad()
|
|
out["attachments"].append(unhex(a["record"]))
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------- §10 verification
|
|
|
|
def stopped(fn, at):
|
|
try:
|
|
return fn()
|
|
except Refuse as e:
|
|
raise Stop(e.reason, at)
|
|
|
|
|
|
def read_one_log(b):
|
|
"""§10.2 step 6: a logs document of exactly one log: its sid and its metrics."""
|
|
try:
|
|
doc = json.loads(b.decode("utf-8"))
|
|
except (UnicodeDecodeError, ValueError):
|
|
raise Refuse("qoe_invalid")
|
|
if not isinstance(doc, list) or len(doc) != 1:
|
|
raise Refuse("qoe_invalid")
|
|
LMV.read_logs(b)
|
|
return doc[0][0]["sid"], LMV.session_metrics(doc[0])
|
|
|
|
|
|
def read_notes(b, acknowledged):
|
|
"""§7.3: a notes document's count of responses, or None for one that does not read."""
|
|
try:
|
|
doc = json.loads(b.decode("utf-8"))
|
|
except (UnicodeDecodeError, ValueError):
|
|
return None
|
|
if not isinstance(doc, list):
|
|
return None
|
|
for n in doc:
|
|
if not exact(n, ["content_hash", "cmsd_static", "cmsd_dynamic"]):
|
|
return None
|
|
h = n["content_hash"]
|
|
if not isinstance(h, str) or not re.fullmatch(r"[0-9a-f]{64}", h):
|
|
return None
|
|
if "notes-acked" not in SKIP and bytes.fromhex(h) not in acknowledged:
|
|
return None
|
|
if any(not (n[k] is None or isinstance(n[k], str)) for k in ("cmsd_static", "cmsd_dynamic")):
|
|
return None
|
|
return len(doc)
|
|
|
|
|
|
def verify_bundle(raw, owner, channels):
|
|
"""§10.2: the outcome of a bundle, as §10.3 states it."""
|
|
try:
|
|
return accept_json(verify(raw, owner, channels))
|
|
except Stop as e:
|
|
return {"outcome": "refuse", "reason": e.reason, "at": e.at}
|
|
|
|
|
|
def verify(raw, owner, channels):
|
|
# 1. The bundle.
|
|
b = read_bundle(raw)
|
|
# 2. The record.
|
|
s = stopped(lambda: read_session(b["session"]), ["session"])
|
|
# 3. Its binding.
|
|
if "owner" not in SKIP and s["owner"] != owner:
|
|
raise Stop("audit_wrong_owner", ["session"])
|
|
if "channel" not in SKIP and not any(channel_id(c) == s["channel"] for c in channels):
|
|
raise Stop("audit_unknown_channel", ["session"])
|
|
if "session-bound" not in SKIP and session_of(s["channel"], s["player"]) != s["session"]:
|
|
raise Stop("audit_session_unbound", ["session"])
|
|
# 4. Each acknowledgement.
|
|
upstreams = {upstream_of(o): o for o in s["origins"]}
|
|
by_digest = {}
|
|
for k, (rec, _) in enumerate(b["acks"]):
|
|
by_digest.setdefault(given_digest(rec), k)
|
|
media = 1 if s["presentation"] == LIVE else 2
|
|
acks, delivery, acknowledged = [], {}, set()
|
|
for i, d in enumerate(s["acks"]):
|
|
if d not in by_digest:
|
|
raise Stop("record_missing", ["ack", i])
|
|
rec, lists = b["acks"][by_digest[d]]
|
|
a = stopped(lambda: SV.read_ack(rec), ["ack", i])
|
|
if "ack-bound" not in SKIP and (a["mode"] != 1 or a["downstream"] != s["player"]):
|
|
raise Stop("audit_ack_unbound", ["ack", i])
|
|
if a["upstream"] not in upstreams and "origin" not in SKIP:
|
|
raise Stop("audit_origin_unknown", ["ack", i])
|
|
origin = upstreams.get(a["upstream"], b"?")
|
|
if "ack-window" not in SKIP and (a["start"] < s["start"] or a["end"] > s["end"]):
|
|
raise Stop("ack_outside_period", ["ack", i])
|
|
if "ack-classes" not in SKIP and (a["role"] != 2 or any(e["grant"] != ZERO or e["proof"] != ZERO or e["class"] not in (4, media) for e in a["entries"])):
|
|
raise Stop("audit_ack_invalid", ["ack", i])
|
|
if len(lists) > len(a["entries"]):
|
|
raise Stop("audit_bundle_invalid", ["ack", i])
|
|
for j, e in enumerate(a["entries"]):
|
|
at = ["ack", i, "entry", j]
|
|
if j >= len(lists):
|
|
raise Stop("record_missing", at)
|
|
items = stopped(lambda: SV.read_items(lists[j]), at)
|
|
if not SV.recomputes(e, items, lists[j]):
|
|
raise Stop("items_mismatch", at)
|
|
if "items-start" not in SKIP and any(st != 0 for _, st, _ in items):
|
|
raise Stop("audit_ack_invalid", at)
|
|
key = (origin, e["class"])
|
|
o, n = delivery.get(key, (0, 0))
|
|
delivery[key] = (o + e["objects"], n + e["bytes"])
|
|
acknowledged.update(h for h, _, _ in items)
|
|
acks.append((i, a))
|
|
named, once = set(s["acks"]), set()
|
|
for k, (rec, _) in enumerate(b["acks"]):
|
|
d = given_digest(rec)
|
|
if "bundle-extra-ack" not in SKIP and (d not in named or d in once):
|
|
raise Stop("audit_bundle_invalid", ["ack", k])
|
|
once.add(d)
|
|
# 5. The chains.
|
|
chains = {}
|
|
for i, a in acks:
|
|
chains.setdefault(a["upstream"], []).append((i, a))
|
|
breaks = []
|
|
for chain in chains.values():
|
|
chain.sort(key=lambda x: (x[1]["start"], x[0]))
|
|
for n, (i, a) in enumerate(chain):
|
|
if n == 0:
|
|
if "chain-first" not in SKIP and a["prev"] != ZERO:
|
|
breaks.append((i, "ack_chain_broken"))
|
|
continue
|
|
p = chain[n - 1][1]
|
|
if "chain-overlap" not in SKIP and p["end"] > a["start"]:
|
|
breaks.append((i, "ack_overlap"))
|
|
elif "chain-prev" not in SKIP and a["prev"] != p["digest"]:
|
|
breaks.append((i, "ack_chain_broken"))
|
|
if breaks:
|
|
i, r = min(breaks)
|
|
raise Stop(r, ["ack", i])
|
|
# 6. The log.
|
|
if "log-digest" not in SKIP and sha(b["log"]) != s["log"]:
|
|
raise Stop("audit_document_mismatch", ["log"])
|
|
sid, m = stopped(lambda: read_one_log(b["log"]), ["log"])
|
|
startup = None if s["startup_ms"] == NO_STARTUP else s["startup_ms"]
|
|
if "qoe" not in SKIP and (sid != s["session"].hex() or m["vst"] != startup or m["stalls"] != s["stalls"] or m["stall_ms"] != s["stall_ms"]):
|
|
raise Stop("audit_qoe_mismatch", ["log"])
|
|
# 7. The notes.
|
|
if s["notes"] == ZERO and b["notes"] is None:
|
|
notes = 0
|
|
elif s["notes"] != ZERO and b["notes"] is not None and (sha(b["notes"]) == s["notes"] or "notes-digest" in SKIP):
|
|
notes = read_notes(b["notes"], acknowledged)
|
|
if notes is None:
|
|
raise Stop("audit_notes_invalid", ["notes"])
|
|
else:
|
|
raise Stop("audit_document_mismatch", ["notes"])
|
|
# 8. The attachments.
|
|
given = [sha(x) for x in b["attachments"]]
|
|
for i, d in enumerate(s["attachments"]):
|
|
if "attach" not in SKIP and d not in given:
|
|
raise Stop("record_missing", ["attachment", i])
|
|
for j, d in enumerate(given):
|
|
if d not in s["attachments"]:
|
|
raise Stop("audit_document_mismatch", ["attachment", j])
|
|
# 9. The claims.
|
|
by_claim = {}
|
|
for k, c in enumerate(b["claims"]):
|
|
by_claim.setdefault(given_digest(c["claim"]), k)
|
|
order = []
|
|
for i, d in enumerate(s["claims"]):
|
|
if d not in by_claim:
|
|
raise Stop("record_missing", ["claim", i])
|
|
order.append(by_claim[d])
|
|
named, once = set(s["claims"]), set()
|
|
for k, c in enumerate(b["claims"]):
|
|
d = given_digest(c["claim"])
|
|
if "bundle-extra" not in SKIP and (d not in named or d in once):
|
|
raise Stop("audit_bundle_invalid", ["claim", k])
|
|
once.add(d)
|
|
seen = set()
|
|
verdicts = [judge(b["claims"][k], s, acknowledged, seen) for k in order]
|
|
return {"session": s, "delivery": delivery, "claims": verdicts, "notes": notes}
|
|
|
|
|
|
def judge(c, s, acknowledged, seen):
|
|
"""§10.2 step 9: one claim's verdict and facts."""
|
|
try:
|
|
x = APV.read_claim(c["claim"])
|
|
except APV.Bad as e:
|
|
return {"verdict": str(e)}
|
|
bad = ED.verify(x["player"], APV.CLAIM_DOMAIN + x["body"], x["sig"])
|
|
if bad:
|
|
return {"verdict": bad}
|
|
if "claim-bound" not in SKIP and (x["room"] != s["channel"] or x["session"] != s["session"] or x["player"] != s["player"]):
|
|
return {"verdict": "audit_claim_unbound"}
|
|
if "marker" not in SKIP and x["cue"] != marker_digest(c["marker"]):
|
|
return {"verdict": "audit_marker_mismatch"}
|
|
if "duplicate" not in SKIP and (x["cue"], x["slot"]) in seen:
|
|
return {"verdict": "claim_duplicate"}
|
|
seen.add((x["cue"], x["slot"]))
|
|
d = sum(o["duration_ms"] for o in c["creative"])
|
|
if "asset" not in SKIP and (x["creative_hash"] != APV.creative_digest(c["creative"]) or any(sp["start"] + sp["length"] > d for sp in x["spans"])
|
|
or x["objects"] != APV.touched(c["creative"], x["spans"])):
|
|
return {"verdict": "claim_asset_mismatch"}
|
|
if "acked" not in SKIP and any(h not in acknowledged for h in x["objects"]):
|
|
return {"verdict": "audit_ad_not_acknowledged"}
|
|
run, viewable = APV.runs(x, {"rule": APV.VIDEO, "require": APV.CLAIM})
|
|
return {"verdict": "presented", "quartile": APV.quartile(x["spans"], d), "run_ms": run, "viewable_ms": viewable, "viewable": run >= 2000}
|
|
|
|
|
|
def presented(s):
|
|
f = [0, 0, 0, 0]
|
|
for x in s["spans"]:
|
|
f[0] += x["length"]
|
|
f[1] += x["length"] if x["visible"] >= 5000 else 0
|
|
f[2] += x["length"] if x["flags"] & AUDIBLE else 0
|
|
f[3] += x["length"] if x["flags"] & AD else 0
|
|
return f
|
|
|
|
|
|
def tag(cls):
|
|
return 0 if cls is None else cls[0]
|
|
|
|
|
|
def accept_json(v):
|
|
s = v["session"]
|
|
p = presented(s)
|
|
return {
|
|
"outcome": "accept", "digest": s["digest"].hex(), "channel": s["channel"].hex(),
|
|
"presentation": "live" if s["presentation"] == LIVE else "on_demand", "start": s["start"], "end": s["end"],
|
|
"ended": ENDED[s["ended"] - 1], "presented_ms": p[0], "viewable_ms": p[1], "audible_ms": p[2], "ad_ms": p[3],
|
|
"delivery": [{"origin": o.decode(), "class": CLASS_NAMES[c], "objects": n, "bytes": by} for (o, c), (n, by) in sorted(v["delivery"].items())],
|
|
"repeats": {"objects": s["repeat_objects"], "bytes": s["repeat_bytes"]},
|
|
"startup_ms": None if s["startup_ms"] == NO_STARTUP else s["startup_ms"], "stalls": s["stalls"], "stall_ms": s["stall_ms"],
|
|
"energy": [{"role": ROLES[e["role"] - 1], "class": LABELS[tag(e["class"])], "joules_micro": e["joules"]} for e in s["energy"]],
|
|
"claims": v["claims"], "notes": v["notes"], "attachments": len(s["attachments"]),
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------- §11 the report
|
|
|
|
def report(name, owner, frm, to, accepted):
|
|
"""§11.1, §11.2: the report over the accepted sessions of the channel, the owner and the period."""
|
|
channel = channel_id(name)
|
|
sessions = {}
|
|
for a in accepted:
|
|
s = a["session"]
|
|
if s["channel"] == channel and s["owner"] == owner and frm <= s["start"] < to:
|
|
sessions[s["digest"]] = a
|
|
sessions = [sessions[d] for d in sorted(sessions)]
|
|
if "report-dup" not in SKIP and len({a["session"]["session"] for a in sessions}) != len(sessions):
|
|
raise Stop("audit_session_duplicate", ["report"])
|
|
pres, ends = [0, 0, 0, 0], [0] * 5
|
|
claims = shown = viewable = 0
|
|
quartiles, refused = [0] * 5, {}
|
|
with_span = startup = stalls = stall_ms = 0
|
|
delivery, rep, energy = {}, [0, 0], {}
|
|
unmetered, absent = [0] * 4, [0] * 4
|
|
for a in sessions:
|
|
s = a["session"]
|
|
pres = [x + y for x, y in zip(pres, presented(s))]
|
|
ends[s["ended"] - 1] += 1
|
|
for c in a["claims"]:
|
|
claims += 1
|
|
if c["verdict"] == "presented":
|
|
shown += 1
|
|
if 1 <= c["quartile"] <= 5:
|
|
quartiles[c["quartile"] - 1] += 1
|
|
viewable += 1 if c["viewable"] else 0
|
|
else:
|
|
refused[c["verdict"]] = refused.get(c["verdict"], 0) + 1
|
|
if s["spans"]:
|
|
with_span += 1
|
|
startup += s["startup_ms"]
|
|
stalls += s["stalls"]
|
|
stall_ms += s["stall_ms"]
|
|
for k, (n, by) in a["delivery"].items():
|
|
o0, b0 = delivery.get(k, (0, 0))
|
|
delivery[k] = (o0 + n, b0 + by)
|
|
rep[0] += s["repeat_objects"]
|
|
rep[1] += s["repeat_bytes"]
|
|
for r in range(1, 5):
|
|
es = [e for e in s["energy"] if e["role"] == r]
|
|
if not es:
|
|
absent[r - 1] += 1
|
|
elif es[0]["class"] is None and "unmetered-apart" not in SKIP:
|
|
unmetered[r - 1] += 1
|
|
else:
|
|
k = (r, tag(es[0]["class"]))
|
|
f, j = energy.get(k, (0, 0))
|
|
energy[k] = (f + 1, j + es[0]["joules"])
|
|
n = len(sessions)
|
|
sup = n < QOE_MIN and "suppress" not in SKIP
|
|
|
|
def q(v):
|
|
return None if sup else v
|
|
return {
|
|
"v": 1, "channel": channel.hex(), "owner": owner.hex(), "from": frm, "to": to,
|
|
"sessions": [a["session"]["digest"].hex() for a in sessions],
|
|
"presented": {"presented_ms": pres[0], "viewable_ms": pres[1], "audible_ms": pres[2], "ad_ms": pres[3]},
|
|
"ended": dict(zip(ENDED, ends)),
|
|
"ads": {"claims": claims, "presented": shown, "quartiles": quartiles, "viewable": viewable, "refused": refused},
|
|
"rebuffering": {"sessions": n, "suppressed": sup, "presented": q(with_span), "startup_ms": q(startup), "stalls": q(stalls), "stall_ms": q(stall_ms)},
|
|
"delivery": [{"origin": o.decode(), "class": CLASS_NAMES[c], "objects": k, "bytes": by} for (o, c), (k, by) in sorted(delivery.items())],
|
|
"repeats": {"objects": rep[0], "bytes": rep[1]},
|
|
"energy": [{"role": ROLES[r - 1], "class": LABELS[t], "figures": f, "joules_micro": j} for (r, t), (f, j) in sorted(energy.items())],
|
|
"unmetered": dict(zip(ROLES, unmetered)), "absent": dict(zip(ROLES, absent)),
|
|
}
|
|
|
|
|
|
def report_bytes(v):
|
|
"""§11.3: members in bytewise order, no white space, no escape but \\" and \\\\ (the report is ASCII)."""
|
|
return json.dumps(v, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("ascii")
|
|
|
|
|
|
def check_report(served, bundles, name, owner, channels):
|
|
"""§11.4."""
|
|
accepted = []
|
|
for k, raw in enumerate(bundles):
|
|
try:
|
|
accepted.append(verify(raw, owner, channels))
|
|
except Stop as e:
|
|
raise Stop(e.reason, ["bundle", k] + e.at)
|
|
if not exact(served, ["report", "sig"]):
|
|
raise Stop("audit_report_mismatch", ["report"])
|
|
r, sig = served["report"], unhex(served["sig"])
|
|
frm, to, names = r.get("from") if isinstance(r, dict) else None, r.get("to") if isinstance(r, dict) else None, r.get("sessions") if isinstance(r, dict) else None
|
|
if sig is None or len(sig) != 64 or not isinstance(frm, int) or not isinstance(to, int) or not isinstance(names, list):
|
|
raise Stop("audit_report_mismatch", ["report"])
|
|
names = [unhex(x) for x in names]
|
|
if any(x is None or len(x) != 32 for x in names):
|
|
raise Stop("audit_report_mismatch", ["report"])
|
|
recomputed = report(name, owner, frm, to, accepted)
|
|
digests = {a["session"]["digest"] for a in accepted}
|
|
for i, d in enumerate(names):
|
|
if "report-missing" not in SKIP and d not in digests:
|
|
raise Stop("record_missing", ["report", "session", i])
|
|
if "report-bytes" not in SKIP and report_bytes(r) != report_bytes(recomputed):
|
|
raise Stop("audit_report_mismatch", ["report"])
|
|
if "report-signature" not in SKIP and ED.verify(owner, REPORT_DOMAIN + report_bytes(r), sig):
|
|
raise Stop("bad_signature", ["report"])
|
|
|
|
|
|
# ---------------------------------------------------------------- the corpus
|
|
|
|
def bundle_bytes(v):
|
|
if "bundle_text" in v:
|
|
return v["bundle_text"].encode("utf-8")
|
|
return json.dumps(v["bundle"]).encode("utf-8")
|
|
|
|
|
|
def report_outcome(doc, v, owner, channels):
|
|
bundles = []
|
|
for n in v["sessions"]:
|
|
for s in doc["sessions"]:
|
|
if s["name"] == n:
|
|
bundles.append(bundle_bytes(s))
|
|
break
|
|
try:
|
|
check_report(v["served"], bundles, v["channel"], owner, channels)
|
|
return {"outcome": "accept", "digest": sha(report_bytes(v["served"]["report"])).hex()}
|
|
except Stop as e:
|
|
return {"outcome": "refuse", "reason": e.reason, "at": e.at}
|
|
|
|
|
|
def check(doc, quiet=False):
|
|
"""Every vector's outcome: the failures."""
|
|
fails = []
|
|
owner, channels = bytes.fromhex(doc["owner"]), doc["channels"]
|
|
for v in doc["sessions"]:
|
|
got = verify_bundle(bundle_bytes(v), owner, channels)
|
|
if got != v["outcome"]:
|
|
fails.append("session %s: stated %s, got %s" % (v["name"], json.dumps(v["outcome"])[:300], json.dumps(got)[:300]))
|
|
for v in doc["reports"]:
|
|
got = report_outcome(doc, v, owner, channels)
|
|
if got != v["outcome"]:
|
|
fails.append("report %s: stated %s, got %s" % (v["name"], v["outcome"], got))
|
|
if not quiet:
|
|
for f in fails:
|
|
print("FAIL " + f)
|
|
return fails
|
|
|
|
|
|
def reasons_given(doc):
|
|
seen = set()
|
|
for section in ("sessions", "reports"):
|
|
for v in doc[section]:
|
|
if v["outcome"].get("reason"):
|
|
seen.add(v["outcome"]["reason"])
|
|
for c in v["outcome"].get("claims", []):
|
|
seen.add(c["verdict"])
|
|
return seen
|
|
|
|
|
|
def flip_hex(h, at):
|
|
b = bytearray(bytes.fromhex(h))
|
|
b[at % len(b)] ^= 1
|
|
return b.hex()
|
|
|
|
|
|
def mutations(doc):
|
|
"""Copies of the corpus, each with one defect a conforming verifier notices."""
|
|
def find(d, section, name):
|
|
return next(v for v in d[section] if v["name"] == name)
|
|
|
|
def m_sig(d):
|
|
v = find(d, "sessions", "live_two_origins")
|
|
v["bundle"]["session"] = flip_hex(v["bundle"]["session"], -1)
|
|
|
|
def m_figure(d):
|
|
find(d, "sessions", "live_two_origins")["outcome"]["presented_ms"] += 1
|
|
|
|
def m_verdict(d):
|
|
find(d, "sessions", "claim_verdicts")["outcome"]["claims"][0]["quartile"] = 5
|
|
|
|
def m_drop_ack(d):
|
|
find(d, "sessions", "on_demand_hidden")["bundle"]["acks"].pop()
|
|
|
|
def m_item(d):
|
|
v = find(d, "sessions", "live_two_origins")
|
|
v["bundle"]["acks"][0]["items"][0] = flip_hex(v["bundle"]["acks"][0]["items"][0], 5)
|
|
|
|
def m_log(d):
|
|
find(d, "sessions", "nothing_presented")["bundle"]["log"] += " "
|
|
|
|
def m_marker(d):
|
|
find(d, "sessions", "live_two_origins")["bundle"]["claims"][0]["marker"] += "x"
|
|
|
|
def m_reason(d):
|
|
find(d, "sessions", "ack_chain_broken")["outcome"]["reason"] = "ack_overlap"
|
|
|
|
def m_place(d):
|
|
find(d, "sessions", "items_not_whole")["outcome"]["at"] = ["ack", 0]
|
|
|
|
def m_channels(d):
|
|
d["channels"] = [c for c in d["channels"] if c != "film-library"]
|
|
|
|
def m_owner(d):
|
|
d["owner"] = flip_hex(d["owner"], 3)
|
|
|
|
def m_report(d):
|
|
find(d, "reports", "news_east")["served"]["report"]["ads"]["presented"] += 1
|
|
|
|
def m_report_sessions(d):
|
|
find(d, "reports", "fifty")["sessions"].pop()
|
|
|
|
def m_notes(d):
|
|
v = find(d, "sessions", "live_two_origins")
|
|
v["bundle"]["notes"] = v["bundle"]["notes"].replace("rtt=40", "rtt=41")
|
|
|
|
def m_creative(d):
|
|
find(d, "sessions", "live_two_origins")["bundle"]["claims"][1]["creative"][0]["duration_ms"] += 1
|
|
|
|
return [m_sig, m_figure, m_verdict, m_drop_ack, m_item, m_log, m_marker, m_reason, m_place, m_channels, m_owner,
|
|
m_report, m_report_sessions, m_notes, m_creative]
|
|
|
|
|
|
CHECKS = ["session-rules", "energy", "session-signature", "owner", "channel", "session-bound", "ack-bound", "origin",
|
|
"ack-window", "ack-classes", "items-start", "chain-first", "chain-overlap", "chain-prev", "log-digest", "qoe",
|
|
"notes-digest", "notes-acked", "attach", "bundle-extra", "bundle-extra-ack", "claim-bound", "marker", "duplicate", "asset", "acked",
|
|
"report-dup", "report-missing", "report-bytes", "report-signature", "suppress", "unmetered-apart"]
|
|
|
|
|
|
def main(corpus_dir):
|
|
with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as f:
|
|
doc = json.load(f)
|
|
ok = True
|
|
files = sorted(os.listdir(corpus_dir))
|
|
if files != ["README.md", "vectors.json"]:
|
|
print("FAIL the corpus directory holds %s" % files)
|
|
ok = False
|
|
fails = check(doc)
|
|
print("%s %d sessions and %d reports get the outcomes the corpus states" % ("PASS" if not fails else "FAIL", len(doc["sessions"]), len(doc["reports"])))
|
|
ok = ok and not fails
|
|
seen = reasons_given(doc)
|
|
missing = [r for r in REASONS + REUSED if r not in seen]
|
|
print("%s every reason of §14 and every borrowed code appears%s" % ("PASS" if not missing else "FAIL", "" if not missing else ": missing " + ", ".join(missing)))
|
|
ok = ok and not missing
|
|
survived = []
|
|
for m in mutations(doc):
|
|
d = copy.deepcopy(doc)
|
|
m(d)
|
|
if not check(d, quiet=True):
|
|
survived.append(m.__name__)
|
|
n = len(mutations(doc))
|
|
print("%s %d of %d mutated copies of the corpus fail%s" % ("PASS" if not survived else "FAIL", n - len(survived), n, "" if not survived else ": " + ", ".join(survived)))
|
|
ok = ok and not survived
|
|
uncaught = []
|
|
for c in CHECKS:
|
|
SKIP.add(c)
|
|
try:
|
|
if not check(doc, quiet=True):
|
|
uncaught.append(c)
|
|
finally:
|
|
SKIP.discard(c)
|
|
print("%s the corpus catches each of %d checks dropped%s" % ("PASS" if not uncaught else "FAIL", len(CHECKS), "" if not uncaught else ": not caught " + ", ".join(uncaught)))
|
|
ok = ok and not uncaught
|
|
return 0 if ok else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) != 2:
|
|
print("usage: stream_audit_verify.py <live-conformance/audit>")
|
|
sys.exit(2)
|
|
sys.exit(main(sys.argv[1]))
|