open-standards/wai/live-conformance/stitched_presentation_verify.py

#!/usr/bin/env python3
"""Independent verifier of the stitched-presentation corpus (extensions/stitched-presentation.md).

Written from the extension's text. It shares no code with wai-rs, which generates the corpus,
and never runs it. Ed25519 signatures are checked with ../claim-conformance/_ed25519.py, WAI's
strict rule (SPEC §7 *Signatures*), and BLAKE3 with ../claim-conformance/_blake3.py. From the
corpora's other second implementations it takes, by reference as the extension does: the spans'
quartile and run and RSASSA-PSS from ad_presentation_verify.py (ad-presentation §6, §7); a
form-1 receipt's leaves, Merkle root and class encoding from live_ads_verify.py (live-ads §8.3,
energy-measurement §2.7), and the consent signal's rules from it (live-ads §8.6, §4.7 here);
and an ad frame from live_compose_verify.py (live-compose §5.9).
Everything else is here, from the standard library:

- markers, their bodies and digests (§3.1), and reading them from a Media Playlist's date range
  tags, an asset list and an MPD over the XML subset of §3.5, with a reader of its own;
- the pod-input record, its rules and signature (§4), a record of version 2's consent signal
  in its layout (§4.7), the VAST check and a form-1 receipt's JSON carriage, payload, identity
  and rules (§4.5, §7.2);
- the claim, its session, its rules and signature, and its attestation challenge (§5);
- each record's and claim's verdict, each ad's presentations and quartiles, and the spend by
  currency (§7).

For every vector it requires every fact the corpus states; of the corpus, that every reason of
§9 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 -I stitched_presentation_verify.py <live-conformance/stitched>

Exit status 0 only if every vector verifies, every mutated copy fails and every weakened
verifier is caught. 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 _blake3 as B3  # noqa: E402
import _ed25519 as ED  # noqa: E402
import ad_presentation_verify as APV  # noqa: E402
import live_ads_verify as LAV  # noqa: E402
import live_compose_verify as LCV  # noqa: E402

EXTENSION = os.path.join(HERE, "..", "extensions", "stitched-presentation.md")

POD_MAGIC, CLAIM_MAGIC = b"WSI1", b"WSC1"
POD_DOMAIN = b"wai:stitched-pod-input\x01"                                         # §4.4
CLAIM_DOMAIN = b"wai:stitched-presentation-claim\x01"                             # §5.3
SESSION_DOMAIN = b"wai:stitched-presentation-session\x01"                         # §5.2
ATTEST_DOMAIN = b"wai:stitched-presentation-attest\x01"                           # §5.5
MARKER_DOMAIN = b"wai:stitched-marker\x01"                                        # §3.1
FORM1_DOMAIN = b"wai:video-ssai-receipt\x01"                                      # §4.5
FORM1_ID_DOMAIN = b"wai:video-ssai-receipt-id\x01"
INTERSTITIAL = "com.apple.hls.interstitial"  # allow-name: the interstitial date-range class, a protocol identifier
DATERANGE, DATERANGE_I, EVENT = 1, 2, 3                                           # §3.1
VAST, VMAP, FORM1 = 1, 2, 3                                                       # §4.1
SERVING_ID, AD_ID, FORM1_ID = 1, 2, 3
PRICING, WINNING_BID, STATED = 1, 2, 3                                            # §4.2
CLAIM, ATTEST = 1, 2                                                              # ad-presentation §8
MAX_PRICE = 10 ** 12
U64 = 2 ** 64 - 1

# The checks a weakened verifier may drop (main(), the last part).
SKIP = set()


def sha(*parts):
    h = hashlib.sha256()
    for p in parts:
        h.update(p)
    return h.digest()


def reasons():
    """§9's table of the codes this extension adds, from the text."""
    with open(EXTENSION, encoding="utf-8") as f:
        text = f.read()
    at = text.index("\n## 9. Reasons")
    end = text.index("\nA verification also", at)
    return [line.split("`")[1] for line in text[at:end].splitlines() if line.startswith("| `")]


REASONS = reasons()
OTHER = ["bad_signature", "claim_key_unbound", "claim_no_decision", "claim_duplicate", "claim_asset_mismatch",
         "claim_not_viewable", "attestation_absent", "attestation_invalid", "claim_absent"]


class Bad(Exception):
    """A verdict other than accepted or presented."""


# ---------------------------------------------------------------- §3.1 markers

def lng(b):
    return struct.pack("<H", len(b)) + b


def marker_body(m):
    if m["carriage"] in (DATERANGE, DATERANGE_I):
        out = lng(m["id"]) + lng(m["start"])
        if m["carriage"] == DATERANGE_I:
            out += struct.pack("<H", m["index"])
        return out
    return lng(m["period"]) + lng(m["scheme"]) + lng(m["value"]) + lng(m["id"])


def marker_digest(carriage, body):
    return sha(MARKER_DOMAIN, bytes([carriage]), body)


def read_body(carriage, b):
    """§3.1: the marker a body of a carriage holds, or None."""
    p = 0

    def field():
        nonlocal p
        if p + 2 > len(b):
            raise ValueError
        n = struct.unpack("<H", b[p:p + 2])[0]
        if p + 2 + n > len(b):
            raise ValueError
        p += 2 + n
        return b[p - n:p]
    try:
        if carriage == DATERANGE:
            m = {"carriage": 1, "id": field(), "start": field()}
        elif carriage == DATERANGE_I:
            m = {"carriage": 2, "id": field(), "start": field()}
            if p + 2 > len(b):
                return None
            m["index"] = struct.unpack("<H", b[p:p + 2])[0]
            p += 2
        elif carriage == EVENT:
            m = {"carriage": 3, "period": field(), "scheme": field(), "value": field(), "id": field()}
        else:
            return None
    except ValueError:
        return None
    return m if p == len(b) else None


# ---------------------------------------------------------------- §3.2, §3.3 HLS

def attribute_list(s):
    """§3.2: [(name, value, quoted)], or None."""
    out, rest = [], s
    while True:
        if "=" not in rest:
            return None
        name, after = rest.split("=", 1)
        if not name or any(not (c.isascii() and (c.isupper() or c.isdigit() or c == "-")) for c in name):
            return None
        if any(n == name for n, _, _ in out):
            return None
        if after.startswith('"'):
            close = after.find('"', 1)
            if close < 0:
                return None
            v = after[1:close]
            if "\r" in v or "\n" in v:
                return None
            out.append((name, v, True))
            after = after[close + 1:]
        else:
            end = after.find(",")
            end = len(after) if end < 0 else end
            v = after[:end]
            if not v or any(c == '"' or c in " \t\r\n\x0b\x0c" for c in v):
                return None
            out.append((name, v, False))
            after = after[end:]
        if not after:
            return out
        if not after.startswith(",") or after == ",":
            return None
        rest = after[1:]


def daterange_tags(b):
    """§3.2: each date range tag that reads, as {name: (value, quoted)}; None when not a playlist."""
    try:
        s = b.decode("utf-8")
    except UnicodeDecodeError:
        return None
    lines = [ln[:-1] if ln.endswith("\r") else ln for ln in s.split("\n")]
    if lines[0] != "#EXTM3U":
        return None
    out = []
    for ln in lines:
        if not ln.startswith("#EXT-X-DATERANGE:"):
            continue
        a = attribute_list(ln[len("#EXT-X-DATERANGE:"):])
        if a is None:
            continue
        d = {n: (v, q) for n, v, q in a}
        if d.get("ID", ("", False))[1] and d.get("START-DATE", ("", False))[1]:
            out.append(d)
    return out


def hls_ads(b, attr):
    tags = daterange_tags(b)
    if tags is None:
        return None
    return [({"carriage": 1, "id": d["ID"][0].encode(), "start": d["START-DATE"][0].encode()}, d[attr][0].encode())
            for d in tags if attr in d]


def interstitials(b):
    tags = daterange_tags(b)
    if tags is None:
        return None
    return [(d["ID"][0].encode(), d["START-DATE"][0].encode()) for d in tags if d.get("CLASS") == (INTERSTITIAL, True)]


def asset_uris(b):
    try:
        v = json.loads(b.decode("utf-8"))
    except (UnicodeDecodeError, ValueError):
        return None
    if not isinstance(v, dict) or not isinstance(v.get("ASSETS"), list):
        return None
    out = []
    for a in v["ASSETS"]:
        if not isinstance(a, dict) or not isinstance(a.get("URI"), str):
            return None
        out.append(a["URI"])
    return out


# ---------------------------------------------------------------- §3.5 the XML subset

WS = " \t\r\n"


def unescape(t):
    out, i = [], 0
    while True:
        j = t.find("&", i)
        if j < 0:
            out.append(t[i:])
            return "".join(out)
        out.append(t[i:j])
        k = t.find(";", j)
        if k < 0:
            raise ValueError("reference")
        r = t[j + 1:k]
        named = {"lt": "<", "gt": ">", "amp": "&", "quot": '"', "apos": "'"}
        if r in named:
            out.append(named[r])
        elif r.startswith("#x") and len(r) > 2:
            out.append(chr(int(r[2:], 16)))
        elif r.startswith("#") and len(r) > 1 and r[1:].isdigit():
            out.append(chr(int(r[1:])))
        else:
            raise ValueError("reference")
        i = k + 1


def local(n):
    return n.rsplit(":", 1)[-1]


class XmlReader:
    def __init__(self, s):
        self.s, self.p = s, 0

    def rest(self):
        return self.s[self.p:]

    def ws(self):
        while self.p < len(self.s) and self.s[self.p] in WS:
            self.p += 1

    def eat(self, t):
        if self.s.startswith(t, self.p):
            self.p += len(t)
            return True
        return False

    def until(self, t):
        j = self.s.find(t, self.p)
        if j < 0:
            raise ValueError("unterminated")
        out = self.s[self.p:j]
        self.p = j + len(t)
        return out

    def misc(self):
        while True:
            self.ws()
            if self.eat("<!--"):
                self.until("-->")
            elif self.s.startswith("<!", self.p):
                raise ValueError("outside the subset")
            elif self.eat("<?"):
                self.until("?>")
            else:
                return

    def name(self):
        j = self.p
        while j < len(self.s) and self.s[j] not in WS + "/>=":
            j += 1
        n = self.s[self.p:j]
        if not n or not (n[0].isalpha() or n[0] in "_:") or any(not (c.isalnum() or c in "_-.:") for c in n):
            raise ValueError("name")
        self.p = j
        return n

    def element(self, depth):
        if depth > 64 or not self.eat("<"):
            raise ValueError("element")
        q = self.name()
        attrs = {}
        while True:
            before = self.p
            self.ws()
            if self.eat("/>"):
                return {"name": local(q), "attrs": attrs, "children": []}
            if self.eat(">"):
                break
            if self.p == before:
                raise ValueError("attribute")
            an = self.name()
            self.ws()
            if not self.eat("="):
                raise ValueError("attribute")
            self.ws()
            if self.eat('"'):
                raw = self.until('"')
            elif self.eat("'"):
                raw = self.until("'")
            else:
                raise ValueError("attribute")
            if "<" in raw or local(an) in attrs:
                raise ValueError("attribute")
            attrs[local(an)] = unescape(raw)
        kids = []
        while True:
            if self.eat("</"):
                c = self.name()
                self.ws()
                if c != q or not self.eat(">"):
                    raise ValueError("end tag")
                return {"name": local(q), "attrs": attrs, "children": kids}
            if self.eat("<![CDATA["):
                kids.append(self.until("]]>"))
            elif self.eat("<!--"):
                self.until("-->")
            elif self.s.startswith("<!", self.p):
                raise ValueError("outside the subset")
            elif self.eat("<?"):
                self.until("?>")
            elif self.s.startswith("<", self.p):
                kids.append(self.element(depth + 1))
            else:
                j = self.s.find("<", self.p)
                if j < 0:
                    raise ValueError("unterminated")
                kids.append(unescape(self.s[self.p:j]))
                self.p = j


def read_xml(b):
    """§3.5: the root element, or None."""
    try:
        s = b.decode("utf-8")
        if s.startswith(""):
            return None
        x = XmlReader(s)
        x.misc()
        root = x.element(1)
        x.misc()
        return root if x.p == len(s) else None
    except (UnicodeDecodeError, ValueError, OverflowError):
        return None


def kids(e, name):
    return [c for c in e["children"] if isinstance(c, dict) and c["name"] == name]


def text(e):
    return "".join(c for c in e["children"] if isinstance(c, str)).strip(WS)


def mpd_ads(b, attr):
    root = read_xml(b)
    if root is None or root["name"] != "MPD":
        return None
    out = []
    for period in kids(root, "Period"):
        for es in kids(period, "EventStream"):
            if "schemeIdUri" not in es["attrs"]:
                continue
            for ev in kids(es, "Event"):
                if "id" in ev["attrs"] and attr in ev["attrs"]:
                    m = {"carriage": 3, "period": period["attrs"].get("id", "").encode(),
                         "scheme": es["attrs"]["schemeIdUri"].encode(), "value": es["attrs"].get("value", "").encode(),
                         "id": ev["attrs"]["id"].encode()}
                    out.append((m, ev["attrs"][attr].encode()))
    return out


# ---------------------------------------------------------------- §4 the record

def is_text(b):
    if not 1 <= len(b) <= 255:
        return False
    try:
        b.decode("utf-8")
        return True
    except UnicodeDecodeError:
        return False


class Cursor:
    def __init__(self, b, why):
        self.b, self.p, self.why = b, 0, why

    def take(self, n):
        if self.p + n > len(self.b):
            raise Bad(self.why)
        self.p += n
        return self.b[self.p - n:self.p]

    def u(self, fmt):
        return struct.unpack("<" + fmt, self.take(struct.calcsize("<" + fmt)))[0]

    def short(self):
        return self.take(self.u("B"))

    def long(self):
        return self.take(self.u("H"))


def frame_ok(f):
    """§4.3."""
    if f["kind"] == 0:
        return all(f[k] == 0 for k in ("width", "height", "corner", "window_w", "window_h", "gap", "aspect"))
    if f["kind"] not in (3, 4):
        return False
    w, h = f["width"], f["height"]
    if not (16 <= w <= 1920 and 16 <= h <= 1920 and w % 2 == 0 and h % 2 == 0 and w * h <= 2073600):
        return False
    if f["kind"] == 3 and (f["gap"] or f["aspect"]):
        return False
    if f["kind"] == 4 and f["aspect"] > 3:
        return False
    return LCV.frame_fields_ok(w, h, f["kind"], f["corner"], f["window_w"], f["window_h"], f["gap"])


def frame_region(f):
    """§6: the ad region's rectangles, live-compose §5.9's."""
    return LCV.ad_frame(f["width"], f["height"], f["kind"], f["corner"], f["window_w"], f["window_h"], f["gap"],
                        f["aspect"])[2]


def share_of(f):
    """§6: the region's share of the picture."""
    if f["kind"] == 0:
        return 10000
    _, _, region = LCV.ad_frame(f["width"], f["height"], f["kind"], f["corner"], f["window_w"], f["window_h"], f["gap"],
                                f["aspect"])
    return LCV.share_bp(region, f["width"], f["height"])


def terms_ok(q):
    """ad-presentation §8, and bit 0 set (§4.4 rule 7)."""
    r, rule, tt, iss = q["require"], q["rule"], q["token_type"], q["issuer"]
    claim = bool(r & CLAIM)
    if r > 7 or (not claim and r) or rule > 2 or claim != (rule != 0):
        return False
    if r & ATTEST:
        if tt not in (1, 2) or not iss:
            return False
    elif tt != 0 or iss:
        return False
    return claim


def xml_name(b):
    return 1 <= len(b) <= 255 and (chr(b[0]).isalpha() and b[0] < 128 or b[0] == 0x5F) and \
        all(chr(c).isalnum() and c < 128 or c in b"_-." for c in b)


def read_record(b):
    """§4.4: the record, or Bad(pod_input_malformed)."""
    why = "pod_input_malformed"
    c = Cursor(b, why)
    if c.take(4) != POD_MAGIC:
        raise Bad(why)
    version = c.u("B")
    if version not in (1, 2):
        raise Bad(why)
    r = {"carriage": c.u("B"), "session": c.take(16), "avail_id": c.short(), "avail_ms": c.u("I"),
         "format": c.u("B"), "response": c.take(32), "id_kind": c.u("B"), "id_attr": c.short(),
         "currency": c.take(3), "exponent": c.u("B")}
    r["frame"] = {"kind": c.u("B"), "width": c.u("H"), "height": c.u("H"), "corner": c.u("B"), "window_w": c.u("H"),
                  "window_h": c.u("H"), "gap": c.u("H"), "aspect": c.u("B")}
    r["terms"] = {"require": c.u("B"), "rule": c.u("B"), "token_type": c.u("H"), "issuer": c.long()}
    r["ads"] = []
    for _ in range(c.u("B")):
        r["ads"].append({"slot": c.u("B"), "marker": c.take(32), "ad_id": c.short(), "serving_id": c.short(),
                         "creative_id": c.short(), "registry": c.short(), "creative_hash": c.take(32),
                         "duration_ms": c.u("I"), "price_micros": c.u("Q"), "price_source": c.u("B"),
                         "evidence": c.take(32)})
    r["consent"] = None
    if version == 2:                                                              # §4.7
        gpp = c.long()
        k = c.u("B")
        sid = [c.u("H") for _ in range(k)]
        try:
            r["consent"] = {"gpp": gpp.decode("utf-8"), "gpp_sid": sid, "child_directed": c.u("B")}
        except UnicodeDecodeError:
            raise Bad(why)
    r["signer"] = c.take(32)
    r["body"] = b[:c.p]
    r["sig"] = c.take(64)
    if c.p != len(b):
        raise Bad(why)
    if "record_rules" in SKIP:
        return r
    ok = r["carriage"] in (1, 2, 3) and is_text(r["avail_id"]) and r["avail_ms"] >= 1
    ok = ok and ((r["format"] in (VAST, VMAP) and r["id_kind"] in (SERVING_ID, AD_ID))
                 or (r["format"] == FORM1 and r["id_kind"] == FORM1_ID))
    a = r["id_attr"]
    if r["carriage"] == DATERANGE:
        ok = ok and 1 <= len(a) <= 255 and all(ch in b"ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-" for ch in a)
    elif r["carriage"] == EVENT:
        ok = ok and xml_name(a)
    else:
        ok = ok and a == b""
    ok = ok and all(65 <= ch <= 90 for ch in r["currency"]) and r["exponent"] <= 4
    ok = ok and frame_ok(r["frame"]) and terms_ok(r["terms"]) and 1 <= len(r["ads"]) <= 16
    for k, ad in enumerate(r["ads"]):
        src = ad["price_source"]
        ok = ok and ad["slot"] == k and is_text(ad["ad_id"]) and is_text(ad["serving_id"])
        ok = ok and (ad["creative_id"] == b"") == (ad["registry"] == b"")
        ok = ok and (ad["creative_id"] == b"" or (is_text(ad["creative_id"]) and is_text(ad["registry"])))
        ok = ok and ad["duration_ms"] >= 1 and ad["price_micros"] <= MAX_PRICE and src in (1, 2, 3)
        ok = ok and not (src == PRICING and r["format"] != VAST) and not (src == WINNING_BID and r["format"] == FORM1)
        ok = ok and not (r["format"] == FORM1 and src != STATED)
        ok = ok and (ad["evidence"] != bytes(32)) == (src == WINNING_BID)
        ok = ok and all(x["marker"] != ad["marker"] for x in r["ads"][:k])
    cs = r["consent"]                                                             # rule 10
    if cs is not None and "consent" not in SKIP:
        ok = ok and LAV.consent_ok(cs) and not (r["session"] == bytes(16) and cs["gpp"])
    if not ok:
        raise Bad(why)
    return r


# ---------------------------------------------------------------- §4.5 a form-1 receipt

FORM1_MEMBERS = {"form", "avail_id", "manifest_hash", "pod_hash", "impressions", "merkle_root", "budget_spent_micros",
                 "work_decide_ops", "joules_micro", "energy_provenance", "grant", "parent_receipt_hash",
                 "signer_pubkey", "signer_id", "sig"}


def read_form1(o):
    """§4.5's JSON carriage, or None. Its integers, hexadecimal and class read as live-ads §8.2's."""
    try:
        if not isinstance(o, dict):
            return None
        label = o.get("energy_provenance")
        if label is not None and label not in LAV.EVIDENCE:
            return None
        if set(o) != FORM1_MEMBERS | set(LAV.EVIDENCE.get(label, [])) or LAV.u(o["form"]) != 1:
            return None
        # A form-2 shaped copy holds the shared members, so live-ads' reader reads its class,
        # grant and impressions; the form-1 members are read here.
        shaped = {k: o[k] for k in o if k not in ("form", "avail_id", "manifest_hash", "parent_receipt_hash")}
        shaped.update({"form": 2, "room": "00" * 32, "cue": "00" * 32, "session": "00" * 16, "start_offset_ms": 0,
                       "format": 1, "asset_list_sha256": "00" * 32, "avail_ms": 0, "slots": 0,
                       "funds_remaining_micros": 0})
        r = LAV.read_receipt(shaped)
        r["avail_id"] = LAV.text(o["avail_id"])
        r["manifest_hash"] = LAV.hexn(o["manifest_hash"], 32)
        p = o["parent_receipt_hash"]
        r["parent"] = None if p is None else LAV.hexn(p, 32)
        return r
    except LAV.Refuse:
        return None


def form1_payload(r):
    g = r["grant"]
    a, cap, sid = r["avail_id"].encode(), g["capability"].encode(), r["signer_id"].encode()
    out = [FORM1_DOMAIN, struct.pack(">I", len(a)), a, r["manifest_hash"], r["pod_hash"], r["merkle_root"],
           LAV.be64(r["budget_spent_micros"]), LAV.be64(r["work_decide_ops"]), LAV.be64(r["joules_micro"]),
           LAV.class_bytes(r["class"]), g["grant_hash"], struct.pack(">I", len(cap)), cap,
           LAV.be64(g["joule_ceiling_micro"])]
    out.append(b"\x00" if g["funds_ceiling"] is None else b"\x01" + LAV.be64(g["funds_ceiling"]))
    out.append(b"\x00" if r["parent"] is None else b"\x01" + r["parent"])
    out.append(LAV.be64(len(r["impressions"])))
    out += [LAV.leaf(i) for i in r["impressions"]]
    out += [r["signer_pubkey"], struct.pack(">I", len(sid)), sid]
    return b"".join(out)


def form1_identity(r):
    return B3.blake3(FORM1_ID_DOMAIN + form1_payload(r) + r["sig"])


def form1_verifies(r):
    """§4.5: a form-1 receipt verifies."""
    if ED.verify(r["signer_pubkey"], form1_payload(r), r["sig"]):
        return False
    imps = r["impressions"]
    spend = sum(i["price_micros"] for i in imps if i["quartiles_reached"] >= 1)
    if LAV.merkle([LAV.leaf(i) for i in imps]) != r["merkle_root"] or spend != r["budget_spent_micros"]:
        return False
    f = r["grant"]["funds_ceiling"]
    if f is not None and spend > f:
        return False
    if r["work_decide_ops"] < len(imps) or any(i["slot"] != k or i["quartiles_reached"] > 5 for k, i in enumerate(imps)):
        return False
    c, j = r["class"], r["joules_micro"]
    if c is None:
        return j == 0
    if j == 0:
        return False
    if c["label"] in ("OnChipCounter", "CalibratedInstrument") and (c["ppm"] == 0 or c["window"] == 0 or j * c["ppm"] < LAV.FLOOR):
        return False
    return not (c["label"] == "CalibratedInstrument" and all(ch in LAV.WHITE_SPACE for ch in c["ref"]))


# ---------------------------------------------------------------- §7.2 the VAST check

def decimal_scaled(s, scale):
    if "." in s:
        i, f = s.split(".", 1)
        if not f:
            return None
    else:
        i, f = s, ""
    if not i or not (i.isascii() and i.isdigit()) or (f and not (f.isascii() and f.isdigit())):
        return None
    if len(f) > scale:
        if f[scale:].strip("0"):
            return None
        f = f[:scale]
    v = int(i + f + "0" * (scale - len(f)))
    return v if v <= U64 else None


def vast_matches(rec, b):
    root = read_xml(b)
    if root is None or root["name"] != "VAST":
        return False
    for ad in rec["ads"]:
        want = ad["serving_id"].decode()

        def ident(a):
            if rec["id_kind"] == SERVING_ID:
                il = kids(a, "InLine")
                sv = kids(il[0], "AdServingId") if il else []
                return text(sv[0]) if sv else None
            return a["attrs"].get("id")
        hits = [a for a in kids(root, "Ad") if ident(a) == want]
        if len(hits) != 1 or not kids(hits[0], "InLine"):
            return False
        inline = kids(hits[0], "InLine")[0]
        if ad["creative_id"] and "vast_creative" not in SKIP:
            cid, reg = ad["creative_id"].decode(), ad["registry"].decode()
            if not any(text(u) == cid and u["attrs"].get("idRegistry") == reg
                       for cs in kids(inline, "Creatives") for c in kids(cs, "Creative") for u in kids(c, "UniversalAdId")):
                return False
        if ad["price_source"] == PRICING and "vast_price" not in SKIP:
            ps = kids(inline, "Pricing")
            if not ps:
                return False
            p = ps[0]
            if p["attrs"].get("model") != "CPM" or p["attrs"].get("currency") != rec["currency"].decode():
                return False
            if decimal_scaled(text(p), rec["exponent"] + 3) != ad["price_micros"]:
                return False
    return True


def form1_matches(rec, r):
    if not form1_verifies(r) or len(r["impressions"]) != len(rec["ads"]):
        return False
    return all(i["slot"] == a["slot"] and i["ad_id"].encode() == a["serving_id"] and i["creative_hash"] == a["creative_hash"]
               and i["duration_ms"] == a["duration_ms"] and i["price_micros"] == a["price_micros"]
               for i, a in zip(r["impressions"], rec["ads"]))


# ---------------------------------------------------------------- §5 the claim

def session_of(key):
    return sha(SESSION_DOMAIN, key)[:16]


def read_claim(b):
    """§5.4: the claim, or Bad(stitched_claim_malformed)."""
    why = "stitched_claim_malformed"
    c = Cursor(b, why)
    if c.take(4) != CLAIM_MAGIC or c.take(1) != b"\x01":
        raise Bad(why)
    x = {"carriage": c.u("B"), "manifest": c.take(32), "session": c.take(16), "creative_hash": c.take(32),
         "player": c.take(32), "geometry": c.u("B"), "share_bp": c.u("H"), "frames": c.u("I"), "marker": c.long(),
         "ad_id": c.short()}
    n = c.u("H")
    x["objects"] = [c.take(32) for _ in range(n)]
    m = c.u("H")
    x["spans"] = []
    for _ in range(m):
        start, length, vis, flags = struct.unpack("<IIHB", c.take(11))
        x["spans"].append({"start": start, "length": length, "visible": vis, "flags": flags})
    x["attestation"] = c.long()
    x["body"] = b[:c.p]
    x["sig"] = c.take(64)
    if c.p != len(b):
        raise Bad(why)
    x["m"] = read_body(x["carriage"], x["marker"])
    if "claim_rules" in SKIP:
        return x
    if x["m"] is None or x["geometry"] not in (1, 2, 3) or not 1 <= x["share_bp"] <= 10000 or not is_text(x["ad_id"]):
        raise Bad(why)
    if (n == 0) != (m == 0):
        raise Bad(why)
    prev = None
    for s in x["spans"]:
        if s["length"] == 0 or s["visible"] > 10000 or s["flags"] > 3:
            raise Bad(why)
        if prev is None and s["flags"] & APV.RESUMED:
            raise Bad(why)
        if prev is not None and (s["start"] < prev or (s["start"] > prev and not s["flags"] & APV.RESUMED)):
            raise Bad(why)
        prev = s["start"] + s["length"]
    return x


def redemption_context(c):
    return sha(ATTEST_DOMAIN, bytes([c["carriage"]]), marker_digest(c["carriage"], c["marker"]), c["session"],
               c["creative_hash"], c["player"], bytes([len(c["ad_id"])]), c["ad_id"])


def attestation(c, q, issuers):
    if not c["attestation"]:
        return "attestation_absent"
    t = APV.read_token(c["attestation"])
    if t is None or t["type"] != q["token_type"]:
        return "attestation_invalid"
    if t["digest"] != sha(APV.token_challenge(q["token_type"], q["issuer"], redemption_context(c))):
        return "attestation_invalid"
    for i in issuers:
        if i["name"] == q["issuer"] and i["token_type"] == t["type"]:
            for k in i["keys"]:
                if sha(k) == t["key_id"] and t["type"] == 2 and APV.pss_verify(k, t["input"], t["auth"]):
                    return None
    return "attestation_invalid"


def marker_in_manifest(c, rec, man, manifests):
    """§7.3 step 6."""
    m, attr = c["m"], rec["id_attr"].decode()
    if m is None:
        return False
    if m["carriage"] == DATERANGE:
        return any(f == m and a == c["ad_id"] for f, a in (hls_ads(man, attr) or []))
    if m["carriage"] == EVENT:
        return any(f == m and a == c["ad_id"] for f, a in (mpd_ads(man, attr) or []))
    uris = asset_uris(man)
    if uris is None or m["index"] >= len(uris) or uris[m["index"]].encode() != c["ad_id"]:
        return False
    return any((m["id"], m["start"]) in (interstitials(p) or []) for p in manifests)


# ---------------------------------------------------------------- §7 verification

def creative_digest(objects):
    return APV.creative_digest(objects)


def verify(inp, through_attestation=True):
    """§7: what a verification states, as the corpus writes it. With `through_attestation` false,
    §7.3 step 11 is not run (impression-aggregates §6.2 step 4 judges claims through step 10)."""
    accepted, rfacts, stated = [], [], []
    for b in inp["records"]:
        f = {"digest": None, "verdict": "accepted"}
        q_stated = None
        try:
            r = read_record(b)
            f["digest"] = sha(r["body"]).hex()
            bad = ED.verify(r["signer"], POD_DOMAIN + r["body"], r["sig"])
            if bad:
                raise Bad(bad)
            if r["signer"] not in inp["signers"] and "signer" not in SKIP:
                raise Bad("pod_signer_unknown")
            if "pod_duplicate" not in SKIP and any(p is not None and p["session"] == r["session"] and
                                                   any(a["marker"] == x["marker"] for a in p["ads"] for x in r["ads"])
                                                   for p in accepted):
                raise Bad("pod_duplicate")
            checked, stated_spend = False, None
            if r["format"] == VAST:
                v = next((x for x in inp["responses"] if sha(x) == r["response"]), None)
                if v is not None:
                    if not vast_matches(r, v) and "response" not in SKIP:
                        raise Bad("pod_response_mismatch")
                    checked = True
            elif r["format"] == FORM1:
                held = [x for x in (read_form1(o) for o in inp["receipts"]) if x is not None]
                x = next((h for h in held if form1_identity(h) == r["response"]), None)
                if x is not None:
                    if not form1_matches(r, x) and "response" not in SKIP:
                        raise Bad("pod_response_mismatch")
                    checked = True
                    q_stated = [i["quartiles_reached"] for i in x["impressions"]]
                    stated_spend = x["budget_spent_micros"]
            region = None if r["frame"]["kind"] == 0 else [list(x) for x in frame_region(r["frame"])]
            f.update(response_checked=checked, stated_spend_micros=stated_spend, share_bp=share_of(r["frame"]),
                     region=region)
            if r["consent"] is not None:                                          # §7.5
                f["consent"] = dict(r["consent"])
            accepted.append(r)
        except Bad as e:
            f = {"digest": f["digest"], "verdict": str(e)}
            q_stated = None
            accepted.append(None)
        rfacts.append(f)
        stated.append(q_stated)

    cfacts, own, passed = [], {}, set()
    by_hash = {sha(m): m for m in inp["manifests"]}
    for j, b in enumerate(inp["claims"]):
        f = {"digest": None, "verdict": "presented", "record": None, "slot": None, "quartile": None, "run_ms": None,
             "viewable_ms": None}
        try:
            c = read_claim(b)
            f["digest"] = sha(c["body"]).hex()
            bad = ED.verify(c["player"], CLAIM_DOMAIN + c["body"], c["sig"])
            if bad:
                raise Bad(bad)
            if session_of(c["player"]) != c["session"] and "key_bound" not in SKIP:
                raise Bad("claim_key_unbound")
            md = marker_digest(c["carriage"], c["marker"])

            def names(p):
                return p["carriage"] == c["carriage"] and any(a["marker"] == md and a["ad_id"] == c["ad_id"] for a in p["ads"])
            i = next((k for k, p in enumerate(accepted) if p is not None and names(p) and
                      (p["session"] == bytes(16) or p["session"] == c["session"] or "session" in SKIP)), None)
            if i is None:
                raise Bad("claim_session_unbound" if any(p is not None and names(p) for p in accepted) else "claim_no_decision")
            rec = accepted[i]
            ad = next(a for a in rec["ads"] if a["marker"] == md and a["ad_id"] == c["ad_id"])
            f["record"], f["slot"] = i, ad["slot"]
            man = by_hash.get(c["manifest"])
            if man is None and "manifest" not in SKIP:
                raise Bad("claim_manifest_unknown")
            if man is not None and "marker" not in SKIP and not marker_in_manifest(c, rec, man, inp["manifests"]):
                raise Bad("claim_marker_absent")
            key = (i, ad["slot"], c["session"])
            if key in passed and "claim_duplicate" not in SKIP:
                raise Bad("claim_duplicate")
            passed.add(key)
            own.setdefault((i, ad["slot"]), []).append(j)
            q = APV.quartile(c["spans"], ad["duration_ms"])
            run, viewable = APV.runs(c, rec["terms"])
            f.update(quartile=q, run_ms=run, viewable_ms=viewable)
            cr = next((x for x in inp["creatives"] if creative_digest(x) == ad["creative_hash"]), None)
            asset_ok = c["creative_hash"] == ad["creative_hash"] and cr is not None and \
                sum(o["duration_ms"] for o in cr) == ad["duration_ms"] and \
                all(s["start"] + s["length"] <= ad["duration_ms"] for s in c["spans"]) and \
                APV.touched(cr, c["spans"]) == c["objects"]
            if not asset_ok and "asset" not in SKIP:
                raise Bad("claim_asset_mismatch")
            if c["share_bp"] != share_of(rec["frame"]) and "frame" not in SKIP:
                raise Bad("claim_frame_mismatch")
            if run < 2000 and "viewable" not in SKIP:
                raise Bad("claim_not_viewable")
            if through_attestation and rec["terms"]["require"] & ATTEST and "attest" not in SKIP:
                why = attestation(c, rec["terms"], inp["issuers"])
                if why:
                    raise Bad(why)
        except Bad as e:
            f["verdict"] = str(e)
        cfacts.append(f)

    spend = {}
    for i, rec in enumerate(accepted):
        if rec is None:
            continue
        ads, total = [], 0
        for a in rec["ads"]:
            mine = own.get((i, a["slot"]), [])
            pres = [j for j in mine if cfacts[j]["verdict"] == "presented"]
            qs = [cfacts[j]["quartile"] for j in pres]
            if pres:
                verdict = "presented"
            else:
                verdict = cfacts[mine[0]]["verdict"] if mine else "claim_absent"
            s = a["price_micros"] * len(pres)
            total += s
            ads.append({"slot": a["slot"], "own_claims": mine, "presentations": len(pres), "verdict": verdict,
                        "quartiles": [sum(1 for x in qs if x >= k) for k in range(1, 6)], "presented_spend_micros": s,
                        "stated_quartile": None if stated[i] is None else stated[i][a["slot"]]})
        rfacts[i].update(ads=ads, presented_spend_micros=total)
        cur = rec["currency"].decode()
        spend[cur] = spend.get(cur, 0) + total
    return {"record_facts": rfacts, "claim_facts": cfacts, "spend": spend}


# ---------------------------------------------------------------- the corpus

def held(doc):
    creatives = [[{"content_hash": bytes.fromhex(o["content_hash"]), "duration_ms": o["duration_ms"]} for o in c["objects"]]
                 for c in doc["creatives"]]
    issuers = [{"name": i["name"].encode(), "token_type": i["token_type"], "keys": [bytes.fromhex(k) for k in i["keys"]]}
               for i in doc["issuers"]]
    return creatives, issuers, [bytes.fromhex(k) for k in doc["signers"]]


def inputs(doc, v, files):
    cr, iss, sg = held(doc)
    get = lambda k: [files[n] for n in v[k]]  # noqa: E731
    return {"records": [bytes.fromhex(x) for x in v["records"]], "claims": [bytes.fromhex(x) for x in v["claims"]],
            "manifests": get("manifests"), "responses": get("responses"),
            "receipts": [json.loads(x.decode("utf-8")) for x in get("receipts")],
            "creatives": cr, "issuers": iss, "signers": sg}


def check(doc, files, quiet=False):
    """Every vector's facts; the reasons; the files. A list of failures."""
    fails = []
    names = {"vectors.json"}
    seen = set()
    for v in doc["vectors"]:
        for k in ("manifests", "responses", "receipts", "bound"):
            names.update(v[k])
        try:
            got = verify(inputs(doc, v, files))
        except (KeyError, ValueError, TypeError, IndexError) as e:
            fails.append("%s: its inputs do not read (%s)" % (v["name"], e))
            continue
        for k, want in got.items():
            if v.get(k) != want:
                fails.append("%s: %s differs" % (v["name"], k))
        for r in got["record_facts"]:
            seen.add(r["verdict"])
            seen.update(a["verdict"] for a in r.get("ads", []))
        seen.update(c["verdict"] for c in got["claim_facts"])
    for r in REASONS + OTHER:
        if r not in seen:
            fails.append("no record, claim or ad gives %s" % r)
    if set(files) - {"README.md"} != names:
        fails.append("the files are not exactly the corpus's: %s" % sorted(set(files) ^ names - {"README.md"}))
    if not quiet:
        for f in fails:
            print("FAIL", f)
    return fails


def load(d):
    files = {}
    for n in sorted(os.listdir(d)):
        with open(os.path.join(d, n), "rb") as f:
            files[n] = f.read()
    return json.loads(files["vectors.json"].decode("utf-8")), files


def flip(h, at):
    b = bytearray(bytes.fromhex(h))
    b[at % len(b)] ^= 0x01
    return b.hex()


def vec(d, name):
    return next(v for v in d["vectors"] if v["name"] == name)


def mutations(doc, files):
    """Copies of the corpus, one defect each: (what, doc, files)."""
    out = []

    def mut(what, f=None, g=None):
        d, fs = copy.deepcopy(doc), dict(files)
        if f:
            f(d)
        if g:
            g(fs)
        out.append((what, d, fs))

    def edit(name, old, new):
        def g(fs):
            fs[name] = fs[name].replace(old, new, 1)
        return g

    def set_claim(name, k, h):
        return lambda d: vec(d, name)["claims"].__setitem__(k, h(vec(d, name)["claims"][k]))

    def set_record(name, k, h):
        return lambda d: vec(d, name)["records"].__setitem__(k, h(vec(d, name)["records"][k]))

    mut("hls_stitched: a record's signature flipped", set_record("hls_stitched", 0, lambda h: flip(h, len(h) // 2 - 3)))
    mut("hls_stitched: a claim's span visible fraction changed", set_claim("hls_stitched", 1, lambda h: flip(h, 300)))
    mut("hls_stitched: viewer 1's playlist carries another ad identity",
        g=edit("hls_stitched.viewer1.m3u8", b'X-AD-ID="srv-v1-0"', b'X-AD-ID="srv-v1-9"'))
    mut("hls_stitched: viewer 1's VAST prices ad-a at 25.01",
        g=edit("hls_stitched.viewer1.vast.xml", b">25.00<", b">25.01<"))
    mut("hls_stitched: viewer 2's VAST names another universal identifier",
        g=edit("hls_stitched.viewer2.vast.xml", b">UID-C<", b">UID-X<"))
    mut("hls_stitched: the stated spend", lambda d: vec(d, "hls_stitched")["spend"].__setitem__("USD", 7400000))
    mut("hls_stitched: the prefetched ad stated presented",
        lambda d: vec(d, "hls_stitched")["record_facts"][1]["ads"][1].update(verdict="presented"))
    mut("hls_stitched: viewer 1's playlist missing", lambda d: vec(d, "hls_stitched")["manifests"].pop(0))
    mut("hls_stitched: an unknown signer trusted", lambda d: d["signers"].append(vec(d, "hls_stitched")["records"][2][-192:-128]))
    mut("hls_interstitial: an asset's URI changed", g=edit("hls_interstitial.assets.json", b'"URI":"https://ads.example/c/6', b'"URI":"https://ads.example/c/7'))
    mut("hls_interstitial: the interstitial class changed", g=edit("hls_interstitial.m3u8", INTERSTITIAL.encode(), b"urn:example:other"))
    mut("hls_interstitial: a token's byte flipped", set_claim("hls_interstitial", 0, lambda h: flip(h, len(h) // 2 - 70)))
    mut("hls_interstitial: the issuer's key changed", lambda d: d["issuers"][0]["keys"].__setitem__(0, flip(d["issuers"][0]["keys"][0], 100)))
    mut("squeezeback: the MPD's event identity changed", g=edit("squeezeback.mpd", b'messageData="srv-sq-1"', b'messageData="srv-sq-2"'))
    mut("squeezeback: the region's side band moved", lambda d: vec(d, "squeezeback")["record_facts"][0]["region"][1].__setitem__(0, 1440))
    mut("squeezeback: the quartile counts", lambda d: vec(d, "squeezeback")["record_facts"][0]["ads"][0].update(quartiles=[3, 3, 3, 3, 2]))
    mut("squeezeback: a creative's duration", lambda d: d["creatives"][5]["objects"][0].update(duration_ms=5001))
    mut("double_box: the stated currency's spend", lambda d: vec(d, "double_box")["spend"].update({"EUR": 1800001}))
    mut("form1_prefetch: the receipt's stated quartile", g=edit("form1_prefetch.receipt.json", b'"quartiles_reached": 1', b'"quartiles_reached": 2'))
    mut("form1_prefetch: the stated spend beside the presented", lambda d: vec(d, "form1_prefetch")["record_facts"][0].update(stated_spend_micros=3000000))
    mut("consent: viewer 51's record with a byte of its GPP string changed",
        set_record("consent", 0, lambda h: h.replace(b"1YNN".hex(), b"1YYN".hex(), 1)))
    mut("consent: viewer 52's stated child-directed flag",
        lambda d: vec(d, "consent")["record_facts"][1]["consent"].update(child_directed=0))
    mut("consent: viewer 51's stated sections", lambda d: vec(d, "consent")["record_facts"][0]["consent"].update(gpp_sid=[]))
    mut("a file the corpus does not name", g=lambda fs: fs.__setitem__("extra.m3u8", b"#EXTM3U\n"))
    return out


CHECKS = ["record_rules", "signer", "pod_duplicate", "response", "vast_price", "vast_creative", "claim_rules",
          "key_bound", "session", "manifest", "marker", "claim_duplicate", "asset", "frame", "viewable", "attest",
          "consent"]


def main(corpus_dir):
    doc, files = load(corpus_dir)
    ok = True
    fails = check(doc, files)
    print("%s every vector gets the facts the corpus states, every reason of §9 is given, and no other file is there"
          % ("FAIL" if fails else "PASS"))
    ok &= not fails
    caught = 0
    for what, d, fs in mutations(doc, files):
        try:
            bad = check(d, fs, quiet=True)
        except (KeyError, ValueError, TypeError, IndexError, StopIteration):
            bad = ["does not read"]
        if bad:
            caught += 1
        else:
            print("FAIL the mutation was not caught:", what)
            ok = False
    print("%s every mutated copy of the corpus fails (%d)" % ("PASS" if ok else "FAIL", caught))
    weak_ok = True
    for k in CHECKS:
        SKIP.clear()
        SKIP.add(k)
        try:
            bad = check(doc, files, quiet=True)
        except (KeyError, ValueError, TypeError, IndexError):
            bad = ["does not run"]
        if not bad:
            print("FAIL the corpus does not catch a verifier without its check", k)
            weak_ok = False
    SKIP.clear()
    print("%s the corpus catches the verifier without each of its checks" % ("PASS" if weak_ok else "FAIL"))
    return 0 if ok and weak_ok else 1


if __name__ == "__main__":
    if len(sys.argv) != 2:
        print("usage: stitched_presentation_verify.py <live-conformance/stitched>", file=sys.stderr)
        sys.exit(2)
    sys.exit(main(sys.argv[1]))