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

#!/usr/bin/env python3
"""Independent verifier of the channel lineup corpus (extensions/channel-lineup.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: a channel
receipt's JSON carriage, its leaves and schedule root, its signed bytes, its identity and its
rules (revenue-share §4, the extension's §5) from revenue_share_verify.py; and the XML subset, the
HLS attribute lists and the decimal reading of stitched-presentation §3.2 and §3.5 from
stitched_presentation_verify.py. Everything else is here, from the standard library:

- text values, XMLTV and HLS times, durations, and the document type declaration (§3);
- the import of an XMLTV schedule, Media RSS catalogues, the JSON form and a Media Playlist's
  breaks into a schedule, its assets and its cues (§4);
- the lineup, its rules, digest and signature, and whether it describes a schedule (§6);
- the XMLTV guide and the now/next document (§7), and verification (§8);
- the cue tags of a Media Playlist (§7.4): the breaks and their cues, a playlist's window, its
  alignment and its lines; and a playlist's verification (§8.4).

For every import it requires the outcome the corpus states (the refusal, or a receipt, lineup and
cues it recomputes and whose signatures it verifies); for every verification, the facts; for every
playlist window, the playlist byte for byte or the refusal; and for every playlist verification,
the facts. Of the
corpus, that every reason of §10 and both of live-events' appear, 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 channel_lineup_verify.py <live-conformance/channels>

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 datetime
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 revenue_share_verify as RSV  # noqa: E402
import stitched_presentation_verify as SPV  # noqa: E402

EXTENSION = os.path.join(HERE, "..", "extensions", "channel-lineup.md")

MAGIC, VERSION = b"WLU1", 1                                                       # §6.1
LINEUP_DOMAIN = b"wai:channel-lineup\x01"                                         # §6.3
CUE_ID_DOMAIN = b"wai:channel-cue-id\x01"                                         # §4.8
ID_SYSTEM = "wai-asset-id"                                                        # §7.2
GUIDE_URN = "urn:wai:guide:"
NOW_NEXT_KIND, FEED_KIND = "wai-now-next", "wai-channel-feed"
MAX_DOC, MAX_COUNT, MAX_INSTANT = 16777216, 10000, 2 ** 53 - 1                    # §2, §3.1
U32, U64 = 2 ** 32 - 1, 2 ** 64 - 1
EVERYWHERE, ONLY, EXCEPT = 0, 1, 2                                                # §6.2 geo
DIGITS = "0123456789"
WS = " \t\r\n"

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


class Refuse(Exception):
    def __init__(self, reason):
        super().__init__(reason)
        self.reason = reason


def check(name, ok, reason):
    """A rule: refuse `reason` when it fails, unless this run drops the check `name`."""
    if not ok and name not in SKIP:
        raise Refuse(reason)


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


def reasons():
    """§10's table of the codes this extension adds, and the codes it names after it."""
    with open(EXTENSION, encoding="utf-8") as f:
        text = f.read()
    sec = text[text.index("\n## 10. Reasons"):]
    sec = sec[:sec.index("\n## ", 1)]
    table, after = sec.split("\nA verification also", 1)
    added = [ln.split("`")[1] for ln in table.splitlines() if ln.startswith("| `")]
    reused = [w.strip("`") for w in after.split() if w.startswith("`") and w.rstrip(",.").endswith("`")]
    return added, [r.rstrip("`") for r in reused]


# ---------------------------------------------------------------- §3.2 text values

def is_text(s):
    return isinstance(s, str) and all(c >= " " or c in "\t\n" for c in s) and not any("\ud800" <= c <= "\udfff" for c in s)


def blen(s):
    return len(s.encode("utf-8"))


def short_text(s):
    return is_text(s) and 1 <= blen(s) <= 255


def long_text(s):
    return is_text(s) and blen(s) <= 65535


def field(ok):
    check("fields", ok, "import_field_invalid")


# ---------------------------------------------------------------- §3.4 times

def instant(y, mo, d, h, mi, s, ms, offset_min):
    if y < 1970 or not 1 <= mo <= 12 or h > 23 or mi > 59 or s > 59:
        return None
    try:
        days = (datetime.date(y, mo, d) - datetime.date(1970, 1, 1)).days
    except ValueError:
        return None
    t = (days * 86400 + h * 3600 + mi * 60 + s - offset_min * 60) * 1000 + ms
    return t if t >= 0 else None


def digits(s):
    return len(s) > 0 and all(c in DIGITS for c in s)


def xmltv_time(v):
    """§3.4: an XMLTV time's instant, or Refuse."""
    if len(v) < 14 or not digits(v[:14]):
        raise Refuse("import_time_malformed")
    rest = v[14:]
    if rest == "":
        off = 0
    elif rest.startswith(" "):
        z = rest[1:]
        ok = len(z) == 5 and z[0] in "+-" and digits(z[1:]) and int(z[1:3]) <= 23 and int(z[3:5]) <= 59
        check("timezone", ok, "import_timezone_unknown")
        if not ok:
            off = 0
        else:
            off = (int(z[1:3]) * 60 + int(z[3:5])) * (-1 if z[0] == "-" else 1)
    else:
        raise Refuse("import_time_malformed")
    t = instant(int(v[0:4]), int(v[4:6]), int(v[6:8]), int(v[8:10]), int(v[10:12]), int(v[12:14]), 0, off)
    if t is None:
        raise Refuse("import_time_malformed")
    return t


def xmltv_format(ms):
    d = datetime.datetime(1970, 1, 1) + datetime.timedelta(milliseconds=ms)
    return d.strftime("%Y%m%d%H%M%S") + " +0000"


def hls_date_time(v):
    """§3.4: an HLS date-time's instant, or Refuse."""
    bad = Refuse("import_time_malformed")
    shape = "dddd-dd-ddTdd:dd:dd"
    if len(v) < 19 or any((c not in DIGITS) if p == "d" else (c != p) for p, c in zip(shape, v[:19])):
        raise bad
    p, ms = 19, 0
    if p < len(v) and v[p] == ".":
        q = p + 1
        while q < len(v) and v[q] in DIGITS:
            q += 1
        frac = v[p + 1:q]
        if not frac or frac[3:].strip("0"):
            raise bad
        ms = int((frac[:3] + "000")[:3])
        p = q
    z = v[p:]
    if z == "":
        raise Refuse("import_timezone_unknown")
    if z == "Z":
        off = 0
    elif len(z) == 6 and z[0] in "+-" and z[3] == ":" and digits(z[1:3]) and digits(z[4:6]):
        hh, mm = int(z[1:3]), int(z[4:6])
        if hh > 23 or mm > 59:
            raise bad
        off = (hh * 60 + mm) * (-1 if z[0] == "-" else 1)
    else:
        raise bad
    t = instant(int(v[0:4]), int(v[5:7]), int(v[8:10]), int(v[11:13]), int(v[14:16]), int(v[17:19]), ms, off)
    if t is None or t > MAX_INSTANT:
        raise bad
    return t


def duration(v):
    """§3.4: a duration in milliseconds, or Refuse."""
    d = SPV.decimal_scaled(v, 3)
    if d is None or d > MAX_INSTANT:
        raise Refuse("import_time_malformed")
    return d


# ---------------------------------------------------------------- §3.3 the XML subset

def doctype_end(b, at):
    """The end of a document type declaration of §3.3 at `at`, or None."""
    def skip(p):
        while p < len(b) and b[p] in b" \t\r\n":
            p += 1
        return p

    def literal(p):
        if p >= len(b) or b[p] not in b"\"'":
            return None
        q = b.find(b[p:p + 1], p + 1)
        return None if q < 0 else q + 1

    p = at + len(b"<!DOCTYPE")
    a = skip(p)
    if a == p:
        return None
    p = a
    if p >= len(b) or not (chr(b[p]).isascii() and (chr(b[p]).isalpha() or b[p] == ord("_"))):
        return None
    p += 1
    while p < len(b) and (chr(b[p]).isascii() and (chr(b[p]).isalnum() or chr(b[p]) in "_-.:")):
        p += 1
    a = skip(p)
    if a > p:
        if b.startswith(b"SYSTEM", a):
            q = skip(a + 6)
            if q == a + 6:
                return None
            p = literal(q)
        elif b.startswith(b"PUBLIC", a):
            q = skip(a + 6)
            if q == a + 6:
                return None
            r = literal(q)
            if r is None:
                return None
            s = skip(r)
            if s == r:
                return None
            p = literal(s)
        else:
            p = a
        if p is None:
            return None
    p = skip(p)
    return p + 1 if p < len(b) and b[p] == ord(">") else None


def read_feed_xml(b):
    """§3.3: the root element, or None."""
    b = bytearray(b)
    p = 0
    while True:
        while p < len(b) and b[p] in b" \t\r\n":
            p += 1
        if b.startswith(b"<!--", p):
            q = b.find(b"-->", p + 4)
            if q < 0:
                return None
            p = q + 3
        elif b.startswith(b"<?", p):
            q = b.find(b"?>", p + 2)
            if q < 0:
                return None
            p = q + 2
        elif b.startswith(b"<!DOCTYPE", p) and "doctype" not in SKIP:
            end = doctype_end(bytes(b), p)
            if end is None:
                return None
            b[p:end] = b" " * (end - p)
            break
        else:
            break
    return SPV.read_xml(bytes(b))


def first(e, name):
    k = SPV.kids(e, name)
    return k[0] if k else None


def text_of(e):
    return SPV.text(e) if e is not None else ""


# ---------------------------------------------------------------- §4 import

def check_inputs(inp):
    """§4.1."""
    alone = inp["json"] is None and inp["xmltv"] is None and len(inp["mrss"]) > 0
    if inp["json"] is not None:
        sets = inp["xmltv"] is None and not inp["mrss"]
    else:
        sets = inp["xmltv"] is not None or bool(inp["mrss"])
    owners = [m["owner"] for m in inp["mrss"]]
    ok = (short_text(inp["channel"]) and sets
          and (all(o is not None for o in owners) or all(o is None for o in owners))
          and all(o is None or short_text(o) for o in owners)
          and all(inp[k] is None or short_text(inp[k]) for k in ("owner", "id_system", "xmltv_channel"))
          and short_text(inp["signer_id"])
          and not (alone and inp["start_ms"] is None)
          and (inp["start_ms"] is None or inp["start_ms"] <= MAX_INSTANT))
    check("inputs", ok, "import_inputs_invalid")


def sized(b):
    check("size", len(b) <= MAX_DOC, "import_oversized")


def description(e, desc_name, art, scheme_attr, scheme_default, rating_text):
    """§4.2 step 4, §4.3 step 3."""
    title = text_of(first(e, "title"))
    field(title != "" and long_text(title))
    desc = text_of(first(e, desc_name))
    field(long_text(desc))
    cats = [SPV.text(c) for c in SPV.kids(e, "category")]
    field(len(cats) <= 16 and all(short_text(c) for c in cats))
    uris = [i["attrs"].get(art[1]) for i in SPV.kids(e, art[0])]
    field(len(uris) <= 8 and all(u is not None and u != "" and long_text(u) for u in uris))
    rating = ("", "")
    r = first(e, "rating")
    if r is not None:
        system = r["attrs"].get(scheme_attr, scheme_default)
        field(system == "" or short_text(system))
        value = rating_text(r)
        field(short_text(value))
        rating = (system, value)
    return {"title": title, "description": desc, "categories": cats, "rating": rating, "art": ("uris", uris),
            "geo": (EVERYWHERE, []), "content_uri": "", "duration_ms": 0}


def read_xmltv(b, xch, id_system):
    """§4.2: (channel name or None, [(programme, description)])."""
    sized(b)
    root = read_feed_xml(b)
    if root is None or root["name"] != "tv":
        raise Refuse("import_xml_malformed")
    progs = [p for p in SPV.kids(root, "programme") if p["attrs"].get("channel") == xch]
    check("counts", len(progs) <= MAX_COUNT, "import_oversized")
    name = None
    for c in SPV.kids(root, "channel"):
        if c["attrs"].get("id") == xch:
            name = text_of(first(c, "display-name")) or None
            break
    if name is not None:
        field(short_text(name))
    if not progs:
        raise Refuse("import_empty")
    out = []
    for p in progs:
        if "start" not in p["attrs"]:
            raise Refuse("import_time_malformed")
        start = xmltv_time(p["attrs"]["start"])
        if "stop" not in p["attrs"]:
            check("no_stop", False, "import_no_stop")
            stop = start + 1000
        else:
            stop = xmltv_time(p["attrs"]["stop"])
        if stop <= start:
            raise Refuse("import_time_malformed")
        ids = [SPV.text(e) for e in SPV.kids(p, "episode-num") if e["attrs"].get("system", "onscreen") == id_system]
        check("asset", bool(ids) and short_text(ids[0]), "import_asset_unknown")
        asset = ids[0] if ids else "?"
        d = description(p, "desc", ("icon", "src"), "system", "", lambda r: text_of(first(r, "value")))
        out.append(({"start": start, "stop": stop, "asset": asset}, d))
    return name, out


def restriction(item):
    """§4.3 step 4."""
    r = None
    for x in SPV.kids(item, "restriction"):
        if x["attrs"].get("type") == "country":
            r = x
            break
    if r is None:
        return (EVERYWHERE, [])
    rel = r["attrs"].get("relationship")
    field(rel in ("allow", "deny"))
    allow = rel == "allow"
    toks = [t for t in SPV.text(r).replace("\t", " ").replace("\r", " ").replace("\n", " ").split(" ") if t]
    if toks in (["all"], ["none"]):
        return (EVERYWHERE, []) if (toks == ["all"]) == allow else (ONLY, [])
    ok = bool(toks) and all(len(t) == 2 and t.isascii() and t.isalpha() for t in toks)
    field(ok)
    codes = sorted(set(t.upper() for t in toks))
    return (ONLY if allow else EXCEPT, codes)


def read_mrss(b, owner, seen, count):
    """§4.3: (channel name or None, [item])."""
    sized(b)
    root = read_feed_xml(b)
    ch = first(root, "channel") if root is not None and root["name"] == "rss" else None
    if ch is None:
        raise Refuse("import_xml_malformed")
    items = SPV.kids(ch, "item")
    count[0] += len(items)
    check("counts", count[0] <= MAX_COUNT, "import_oversized")
    name = text_of(first(ch, "title")) or None
    if name is not None:
        field(short_text(name))
    out = []
    for it in items:
        g = first(it, "guid")
        asset = SPV.text(g) if g is not None else ""
        check("asset", short_text(asset), "import_asset_unknown")
        check("duplicate", asset not in seen, "import_asset_duplicate")
        seen.add(asset)
        content = first(it, "content")
        if content is None:
            grp = first(it, "group")
            content = first(grp, "content") if grp is not None else None
        uri = content["attrs"].get("url", "") if content is not None else ""
        field(long_text(uri))
        dur = None
        if content is not None and "duration" in content["attrs"]:
            dur = duration(content["attrs"]["duration"])
            if dur == 0:
                raise Refuse("import_time_malformed")
        d = description(it, "description", ("thumbnail", "url"), "scheme", "urn:simple", SPV.text)
        d["geo"] = restriction(it)
        d["content_uri"], d["duration_ms"] = uri, dur or 0
        out.append({"asset": asset, "desc": d, "duration": dur, "owner": owner})
    return name, out


def strict_json(b):
    """§4.4's reading of JSON: a value, or None (a member named twice, a number that is not an
    integer of the form, an unpaired surrogate, more than 64 levels)."""
    class Bad(Exception):
        pass

    def pairs(ps):
        names = [n for n, _ in ps]
        if len(set(names)) != len(names):
            raise Bad()
        return dict(ps)

    def bad_number(s):
        raise Bad()

    def depth(v, d):
        if d > 64:
            raise Bad()
        if isinstance(v, dict):
            for k, x in v.items():
                if not is_text_or_any(k):
                    raise Bad()
                depth(x, d + 1)
        elif isinstance(v, list):
            for x in v:
                depth(x, d + 1)
        elif isinstance(v, str) and not is_text_or_any(v):
            raise Bad()

    def is_text_or_any(s):
        return not any("\ud800" <= c <= "\udfff" for c in s)

    try:
        s = b.decode("utf-8")
        v = json.loads(s, object_pairs_hook=pairs, parse_float=bad_number, parse_constant=bad_number)
        depth(v, 1)
        return v
    except (UnicodeDecodeError, ValueError, Bad, RecursionError):
        return None


def jint(v):
    return isinstance(v, int) and not isinstance(v, bool) and 0 <= v <= MAX_INSTANT


def exactly(o, names):
    return isinstance(o, dict) and set(o) == set(names) and len(o) == len(names)


def json_structure(v):
    """§4.4's table: True when the form reads."""
    if not exactly(v, ["kind", "channel_name", "programmes", "assets", "breaks"]) or v["kind"] != FEED_KIND:
        return False
    if not isinstance(v["channel_name"], str) or not all(isinstance(v[k], list) for k in ("programmes", "assets", "breaks")):
        return False
    for p in v["programmes"]:
        if not (exactly(p, ["start_ms", "stop_ms", "asset_id"]) and jint(p["start_ms"]) and jint(p["stop_ms"])
                and isinstance(p["asset_id"], str)):
            return False
    for a in v["assets"]:
        if not exactly(a, ["asset_id", "title", "description", "content_uri", "categories", "rating", "artwork", "geo",
                           "rights_holder", "duration_ms"]):
            return False
        if not all(isinstance(a[k], str) for k in ("asset_id", "title", "description", "content_uri")):
            return False
        if not (isinstance(a["categories"], list) and all(isinstance(c, str) for c in a["categories"])):
            return False
        r = a["rating"]
        if r is not None and not (exactly(r, ["system", "value"]) and isinstance(r["system"], str) and isinstance(r["value"], str)):
            return False
        if not isinstance(a["artwork"], list) or not all(exactly(w, ["uri", "sha256"]) and isinstance(w["uri"], str)
                                                         and isinstance(w["sha256"], str) for w in a["artwork"]):
            return False
        g = a["geo"]
        if g is not None and not (isinstance(g, dict) and len(g) == 1 and list(g)[0] in ("allow", "deny")
                                  and isinstance(list(g.values())[0], list)
                                  and all(isinstance(c, str) for c in list(g.values())[0])):
            return False
        if not (a["rights_holder"] is None or isinstance(a["rights_holder"], str)) or not jint(a["duration_ms"]):
            return False
    for k in v["breaks"]:
        if not (exactly(k, ["id", "start_ms", "duration_ms", "scte35"]) and isinstance(k["id"], str) and jint(k["start_ms"])
                and jint(k["duration_ms"]) and (k["scte35"] is None or isinstance(k["scte35"], str))):
            return False
    return True


def hex_any(s):
    if len(s) % 2 or any(c not in "0123456789abcdefABCDEF" for c in s):
        return None
    return bytes.fromhex(s)


def section_reads(b):
    """live-events §4.3's reading of form 1 (§4.5 step 3)."""
    return len(b) >= 3 and b[0] == 0xFC and (((b[1] << 8) | b[2]) & 0x0FFF) + 3 == len(b)


def break_duration(d):
    check("break_duration", 0 < d <= U32, "import_break_duration")
    return d


def scte35(b):
    check("scte35", b is not None and section_reads(b), "import_scte35_malformed")
    return b


def read_json(b, nbreaks):
    """§4.4: (channel name, programmes, {asset id: (description, rights holder)}, breaks)."""
    sized(b)
    v = strict_json(b)
    if v is None or not json_structure(v):
        raise Refuse("import_json_malformed")
    nbreaks[0] += len(v["breaks"])
    check("counts", len(v["programmes"]) <= MAX_COUNT and len(v["assets"]) <= MAX_COUNT and nbreaks[0] <= MAX_COUNT,
          "import_oversized")
    field(short_text(v["channel_name"]))
    progs = []
    for p in v["programmes"]:
        field(short_text(p["asset_id"]))
        if p["stop_ms"] <= p["start_ms"]:
            raise Refuse("import_time_malformed")
        progs.append({"start": p["start_ms"], "stop": p["stop_ms"], "asset": p["asset_id"]})
    assets = {}
    for a in v["assets"]:
        field(short_text(a["asset_id"]))
        check("duplicate", a["asset_id"] not in assets, "import_asset_duplicate")
        field(a["title"] != "" and long_text(a["title"]))
        field(long_text(a["description"]) and long_text(a["content_uri"]))
        field(len(a["categories"]) <= 16 and all(short_text(c) for c in a["categories"]))
        rating = ("", "")
        if a["rating"] is not None:
            s, val = a["rating"]["system"], a["rating"]["value"]
            field((s == "" or short_text(s)) and short_text(val))
            rating = (s, val)
        field(len(a["artwork"]) <= 8 and all(w["uri"] != "" and long_text(w["uri"]) and len(w["sha256"]) == 64
                                             and all(c in "0123456789abcdef" for c in w["sha256"]) for w in a["artwork"]))
        geo = (EVERYWHERE, [])
        if a["geo"] is not None:
            (rel, codes), = a["geo"].items()
            ok = (all(len(c) == 2 and c.isascii() and c.isalpha() and c.isupper() for c in codes)
                  and all(codes[i] < codes[i + 1] for i in range(len(codes) - 1)) and (rel == "allow" or codes))
            field(ok)
            geo = (ONLY if rel == "allow" else EXCEPT, list(codes))
        if a["rights_holder"] is not None:
            field(short_text(a["rights_holder"]))
        d = {"title": a["title"], "description": a["description"], "categories": a["categories"], "rating": rating,
             "art": ("hashed", [(bytes.fromhex(w["sha256"]), w["uri"]) for w in a["artwork"]]), "geo": geo,
             "content_uri": a["content_uri"], "duration_ms": a["duration_ms"]}
        assets[a["asset_id"]] = (d, a["rights_holder"])
    breaks = []
    for k in v["breaks"]:
        field(short_text(k["id"]))
        sec = scte35(hex_any(k["scte35"])) if k["scte35"] is not None else None
        breaks.append({"id": k["id"], "start": k["start_ms"], "duration": break_duration(k["duration_ms"]), "section": sec})
    return v["channel_name"], progs, assets, breaks


def hex_sequence(v):
    """§4.5 step 3: an unquoted hexadecimal-sequence's bytes, or None."""
    value, quoted = v
    if quoted or value[:2] not in ("0x", "0X"):
        return None
    return hex_any(value[2:])


def read_playlist(b, nbreaks):
    """§4.5: the breaks."""
    sized(b)
    bad = Refuse("import_playlist_malformed")
    try:
        s = b.decode("utf-8")
    except UnicodeDecodeError:
        raise bad
    lines = [ln[:-1] if ln.endswith("\r") else ln for ln in s.split("\n")]
    if lines[0] != "#EXTM3U":
        raise bad
    groups = []  # [(id, {name: (value, quoted)})]
    for ln in lines:
        if not ln.startswith("#EXT-X-DATERANGE:"):
            continue
        a = SPV.attribute_list(ln[len("#EXT-X-DATERANGE:"):])
        if a is None:
            raise bad
        d = {n: (v, q) for n, v, q in a}
        if "ID" not in d or not d["ID"][1]:
            raise bad
        gid = d["ID"][0]
        g = next((x for x in groups if x[0] == gid), None)
        if g is None:
            g = (gid, {})
            groups.append(g)
        for n, v in d.items():
            if n in g[1] and g[1][n] != v and "conflict" not in SKIP:
                raise bad
            g[1].setdefault(n, v)
    brks = [g for g in groups if "SCTE35-OUT" in g[1]]
    nbreaks[0] += len(brks)
    check("counts", nbreaks[0] <= MAX_COUNT, "import_oversized")
    out = []
    for gid, a in brks:
        field(short_text(gid))
        sd = a.get("START-DATE")
        if sd is None or not sd[1]:
            raise bad
        start = hls_date_time(sd[0])
        sec = scte35(hex_sequence(a["SCTE35-OUT"]))
        if "SCTE35-IN" in a:
            scte35(hex_sequence(a["SCTE35-IN"]))

        def plain(v):
            if v[1]:
                raise bad
            return duration(v[0])
        if "DURATION" in a:
            d = plain(a["DURATION"])
        elif "END-DATE" in a:
            if not a["END-DATE"][1]:
                raise bad
            end = hls_date_time(a["END-DATE"][0])
            check("break_duration", end >= start, "import_break_duration")
            d = end - start
        elif "PLANNED-DURATION" in a:
            d = plain(a["PLANNED-DURATION"])
        else:
            raise Refuse("import_break_duration")
        out.append({"id": gid, "start": start, "duration": break_duration(d), "section": sec})
    return out


def artwork_name(uri):
    u = uri
    for c in "?#":
        u = u.split(c, 1)[0]
    name = u.rsplit("/", 1)[-1]
    ok = 1 <= len(name) <= 255 and name not in (".", "..") and all(c.isascii() and (c.isalnum() or c in "._-") for c in name)
    return name if ok else None


def cue_id(bid, sec):
    if sec is not None and len(sec) >= 18 and sec[4] & 0x80 == 0 and sec[13] == 0x05:
        return int.from_bytes(sec[14:18], "big")
    return int.from_bytes(sha(CUE_ID_DOMAIN, bid.encode())[:4], "big")


def cue_payload(k):
    """§4.8: live-events §4.3's cue, little-endian."""
    out = bytes([4]) + struct.pack("<II", cue_id(k["id"], k["section"]), k["duration"])
    if k["section"] is None:
        return out + b"\x00"
    return out + b"\x01" + struct.pack("<H", len(k["section"])) + k["section"]


def do_import(inp, files):
    """§4: {"entries", "window", "lineup" (its fields), "cues"}, or Refuse."""
    check_inputs(inp)
    xch = inp["xmltv_channel"] or inp["channel"]
    id_system = inp["id_system"] or ID_SYSTEM
    names, xmltv, items, seen, count, nbreaks, breaks, jf = [], None, [], set(), [0], [0], [], None
    if inp["xmltv"] is not None:
        n, xmltv = read_xmltv(files[inp["xmltv"]], xch, id_system)
        names.append(n)
    for k, m in enumerate(inp["mrss"]):
        n, its = read_mrss(files[m["file"]], m["owner"], seen, count)
        if k == 0:
            names.append(n)
        items += its
    if inp["json"] is not None:
        n, jp, ja, jb = read_json(files[inp["json"]], nbreaks)
        names.append(n)
        jf = (jp, ja)
        breaks += jb
    if inp["playlist"] is not None:
        breaks += read_playlist(files[inp["playlist"]], nbreaks)

    # §4.6 the programmes
    if xmltv is not None:
        progs = [p for p, _ in xmltv]
    elif jf is not None:
        if not jf[0]:
            raise Refuse("import_empty")
        progs = list(jf[0])
    else:
        if not items:
            raise Refuse("import_empty")
        at, progs = inp["start_ms"], []
        for it in items:
            check("duration_unknown", it["duration"] is not None, "import_duration_unknown")
            stop = at + (it["duration"] or 1000)
            if stop > MAX_INSTANT:
                raise Refuse("import_time_malformed")
            progs.append({"start": at, "stop": stop, "asset": it["asset"]})
            at = stop
    progs = sorted(progs, key=lambda p: p["start"])
    for p, q in zip(progs, progs[1:]):
        check("gap", p["stop"] >= q["start"], "import_gap")
        check("overlap", p["stop"] <= q["start"], "import_overlap")
    check("whole_seconds", all(p["start"] % 1000 == 0 for p in progs) and progs[-1]["stop"] % 1000 == 0,
          "import_time_malformed")

    # §4.6 their assets
    item_of = {it["asset"]: it for it in items}
    order = []
    for p in progs:
        if p["asset"] not in order:
            order.append(p["asset"])
    assets = []
    for aid in order:
        if jf is not None:
            check("asset", aid in jf[1], "import_asset_unknown")
            d, holder = jf[1].get(aid, (None, None))
            if d is None:
                continue
        elif aid in item_of:
            d, holder = item_of[aid]["desc"], item_of[aid]["owner"]
        else:
            d, holder = next(x for p, x in xmltv if p["asset"] == aid), None
        holder = holder if holder is not None else inp["owner"]
        check("owner", holder is not None, "import_owner_unknown")
        kind, art = d["art"]
        if kind == "uris":
            hashed = []
            for u in art:
                n = artwork_name(u)
                ok = n is not None and n in inp["artwork_files"]
                check("artwork", ok, "import_artwork_missing")
                body = inp["artwork_files"].get(n, b"") if n else b""
                sized(body)
                hashed.append((hashlib.sha256(body).digest(), u))
            art = hashed
        assets.append({"asset_id": aid, "title": d["title"], "description": d["description"], "categories": d["categories"],
                       "rating_system": d["rating"][0], "rating_value": d["rating"][1], "artwork": art, "geo": d["geo"][0],
                       "countries": d["geo"][1], "rights_holder": holder or "", "content_uri": d["content_uri"],
                       "duration_ms": d["duration_ms"]})

    # §4.7 the schedule
    breaks = sorted(breaks, key=lambda k: k["start"])
    for k, n in zip(breaks, breaks[1:]):
        check("break_overlap", k["start"] + k["duration"] <= n["start"], "import_overlap")
    within = [[] for _ in progs]
    for k in breaks:
        at = next((i for i, p in enumerate(progs) if p["start"] <= k["start"] < p["stop"]), None)
        ok = at is not None and k["start"] + k["duration"] <= progs[at]["stop"]
        check("break_outside", ok, "import_break_outside")
        if at is not None:
            within[at].append(k)
    entries = []
    for p, ks in zip(progs, within):
        at = p["start"]
        for k in ks:
            if k["start"] > at:
                entries.append({"asset_id": p["asset"], "start_ms": at, "duration_ms": k["start"] - at, "kind": "program"})
            entries.append({"asset_id": k["id"], "start_ms": k["start"], "duration_ms": k["duration"], "kind": "ad"})
            at = k["start"] + k["duration"]
        if p["stop"] > at:
            entries.append({"asset_id": p["asset"], "start_ms": at, "duration_ms": p["stop"] - at, "kind": "program"})

    # §4.9
    first_start, last_end = progs[0]["start"], progs[-1]["stop"]
    ws, we = inp["window"] if inp["window"] is not None else (first_start, last_end)
    check("window", ws < we and ws >= first_start and we <= last_end, "import_window_outside")
    name = next((n for n in names if n is not None), inp["channel"])
    lineup = {"channel_id": inp["channel"], "channel_name": name, "schedule_root": schedule_root(entries),
              "schedule_start_ms": first_start, "schedule_end_ms": last_end, "programmes": [p["start"] for p in progs],
              "assets": sorted(assets, key=lambda a: a["asset_id"].encode())}
    cues = [{"break": k["id"], "start_ms": k["start"], "payload": cue_payload(k).hex()} for k in breaks]
    return {"entries": entries, "window": (ws, we), "lineup": lineup, "cues": cues}


# ---------------------------------------------------------------- §5 the receipt

def schedule_root(entries):
    """revenue-share §4.2: the Merkle root of the entries' leaves."""
    return RSV.merkle([RSV.leaf(e) for e in entries])


def aired_of(entries, ws, we):
    return [i for i, e in enumerate(entries) if e["start_ms"] < we and e["start_ms"] + e["duration_ms"] > ws]


def u64(v):
    return isinstance(v, int) and not isinstance(v, bool) and 0 <= v <= U64


def hexn(v, n):
    if not isinstance(v, str) or len(v) != 2 * n or any(c not in "0123456789abcdef" for c in v):
        return None
    return bytes.fromhex(v)


def read_receipt(b):
    """§5: one JSON text of revenue-share §4.1's carriage, read as revenue-share's second
    implementation reads it; the run, or None."""
    try:
        r = RSV.read_run(json.loads(b.decode("utf-8")))
        RSV.run_payload(r)
        return r
    except (UnicodeError, ValueError, RSV.NoRead, KeyError, TypeError):
        return None


def receipt_payload(r):
    return RSV.run_payload(r)


def receipt_identity(r):
    return RSV.run_identity(r)


def receipt_rules(r):
    """§5: revenue-share §4.3's rules 1 to 5 (rule 6, the signature, is §8.2 step 2's)."""
    return "receipt_rules" in SKIP or RSV.run_verifies(r)


# ---------------------------------------------------------------- §6 the lineup

def lshort(s):
    b = s.encode()
    return bytes([len(b)]) + b


def llong(s):
    b = s.encode()
    return struct.pack("<H", len(b)) + b


def lineup_body(l, signer):
    out = [MAGIC, bytes([VERSION]), lshort(l["channel_id"]), lshort(l["channel_name"]), l["schedule_root"],
           struct.pack("<QQI", l["schedule_start_ms"], l["schedule_end_ms"], len(l["programmes"]))]
    out += [struct.pack("<Q", p) for p in l["programmes"]]
    out.append(struct.pack("<I", len(l["assets"])))
    for a in l["assets"]:
        out += [lshort(a["asset_id"]), llong(a["title"]), llong(a["description"]), bytes([len(a["categories"])])]
        out += [lshort(c) for c in a["categories"]]
        out += [lshort(a["rating_system"]), lshort(a["rating_value"]), bytes([len(a["artwork"])])]
        out += [h + llong(u) for h, u in a["artwork"]]
        out += [bytes([a["geo"], len(a["countries"])])] + [c.encode() for c in a["countries"]]
        out += [lshort(a["rights_holder"]), llong(a["content_uri"]), struct.pack("<Q", a["duration_ms"])]
    out.append(signer)
    return b"".join(out)


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

    def take(self, n):
        if self.p + n > len(self.b):
            raise ValueError("short")
        out = self.b[self.p:self.p + n]
        self.p += n
        return out

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

    def short(self):
        return self.take(self.u("<B")).decode("utf-8")

    def long(self):
        return self.take(self.u("<H")).decode("utf-8")


def read_lineup(b):
    """§6.4: the lineup, or None."""
    try:
        c = Cursor(b)
        if c.take(4) != MAGIC or c.u("<B") != VERSION:
            return None
        l = {"channel_id": c.short(), "channel_name": c.short(), "schedule_root": c.take(32),
             "schedule_start_ms": c.u("<Q"), "schedule_end_ms": c.u("<Q")}
        s, e = l["schedule_start_ms"], l["schedule_end_ms"]
        if not (short_text(l["channel_id"]) and short_text(l["channel_name"])):
            return None
        if not (s < e and s % 1000 == 0 and e % 1000 == 0 and e <= MAX_INSTANT):
            return None
        p = c.u("<I")
        if not 1 <= p <= MAX_COUNT:
            return None
        l["programmes"] = [c.u("<Q") for _ in range(p)]
        ps = l["programmes"]
        if ps[0] != s or ps[-1] >= e or any(ps[i] >= ps[i + 1] for i in range(p - 1)) or any(x % 1000 for x in ps):
            return None
        a = c.u("<I")
        if not 1 <= a <= MAX_COUNT:
            return None
        l["assets"] = []
        for _ in range(a):
            x = {"asset_id": c.short(), "title": c.long(), "description": c.long()}
            n = c.u("<B")
            if n > 16:
                return None
            x["categories"] = [c.short() for _ in range(n)]
            x["rating_system"], x["rating_value"] = c.short(), c.short()
            k = c.u("<B")
            if k > 8:
                return None
            x["artwork"] = [(c.take(32), c.long()) for _ in range(k)]
            x["geo"] = c.u("<B")
            g = c.u("<B")
            x["countries"] = [c.take(2).decode("ascii") for _ in range(g)]
            x["rights_holder"], x["content_uri"], x["duration_ms"] = c.short(), c.long(), c.u("<Q")
            cs = x["countries"]
            ok = (short_text(x["asset_id"]) and x["title"] != "" and long_text(x["title"]) and long_text(x["description"])
                  and all(short_text(y) for y in x["categories"])
                  and (x["rating_system"] == "" or short_text(x["rating_system"]))
                  and (x["rating_value"] == "" or short_text(x["rating_value"]))
                  and not (x["rating_value"] == "" and x["rating_system"] != "")
                  and all(u != "" and long_text(u) for _, u in x["artwork"])
                  and x["geo"] <= 2 and not (x["geo"] == EVERYWHERE and cs) and not (x["geo"] == EXCEPT and not cs)
                  and all(len(y) == 2 and all("A" <= ch <= "Z" for ch in y) for y in cs)
                  and all(cs[i] < cs[i + 1] for i in range(len(cs) - 1))
                  and short_text(x["rights_holder"]) and long_text(x["content_uri"]) and x["duration_ms"] <= MAX_INSTANT)
            if not ok or (l["assets"] and l["assets"][-1]["asset_id"].encode() >= x["asset_id"].encode()):
                return None
            l["assets"].append(x)
        l["signer"], l["sig"] = c.take(32), c.take(64)
        return l if c.p == len(b) else None
    except (ValueError, UnicodeDecodeError, struct.error):
        return None


def lineup_digest(l):
    return sha(lineup_body(l, l["signer"]))


def spans(l):
    ps = l["programmes"]
    return [(s, ps[k + 1] if k + 1 < len(ps) else l["schedule_end_ms"]) for k, s in enumerate(ps)]


def programme_assets(l, sched):
    """§6.5 rule 4: each programme's asset, or None."""
    out = []
    for s, e in spans(l):
        ids = [x["asset_id"] for x in sched if x["kind"] == "program" and s <= x["start_ms"] < e]
        if not ids or any(i != ids[0] for i in ids):
            return None
        out.append(ids[0])
    return out


def describes(l, r):
    """§6.5, each rule a check of its own."""
    s = r["schedule"]
    if not s:
        return False
    end = s[-1]["start_ms"] + s[-1]["duration_ms"]
    if "describes_root" not in SKIP and (l["channel_id"] != r["channel_id"] or l["schedule_root"] != r["schedule_root"]
                                         or l["schedule_start_ms"] != s[0]["start_ms"] or l["schedule_end_ms"] != end):
        return False
    starts = {x["start_ms"] for x in s}
    if "describes_starts" not in SKIP and any(p not in starts for p in l["programmes"]):
        return False
    if "describes_assets" not in SKIP and programme_assets(l, s) is None:
        return False
    # Rule 5 over the programmes' assets: every program entry's asset in a programme's span.
    held = {x["asset_id"] for a, b in spans(l) for x in s if x["kind"] == "program" and a <= x["start_ms"] < b}
    return "describes_set" in SKIP or sorted(held, key=str.encode) == [a["asset_id"] for a in l["assets"]]


# ---------------------------------------------------------------- §7 the guide

def all_programmes(r, l):
    ids = programme_assets(l, r["schedule"])
    by = {a["asset_id"]: a for a in l["assets"]}
    return [(s, e, by[i]) for (s, e), i in zip(spans(l), ids)]


def esc_text(s):
    return s.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")


def esc_attr(s):
    return esc_text(s).replace('"', "&quot;").replace("\t", "&#9;").replace("\n", "&#10;")


def xmltv_guide(r, l, frm, to):
    """§7.2."""
    ch = esc_attr(l["channel_id"])
    o = ['<?xml version="1.0" encoding="UTF-8"?>\n',
         '<tv source-data-url="%s%s:%s:%d:%d">\n' % (GUIDE_URN, r["schedule_root"].hex(), lineup_digest(l).hex(), frm, to),
         '  <channel id="%s">\n    <display-name>%s</display-name>\n  </channel>\n' % (ch, esc_text(l["channel_name"]))]
    for s, e, a in all_programmes(r, l):
        if not (s < to and e > frm):
            continue
        o.append('  <programme start="%s" stop="%s" channel="%s">\n' % (xmltv_format(s), xmltv_format(e), ch))
        o.append("    <title>%s</title>\n" % esc_text(a["title"]))
        if a["description"]:
            o.append("    <desc>%s</desc>\n" % esc_text(a["description"]))
        o += ["    <category>%s</category>\n" % esc_text(c) for c in a["categories"]]
        o += ['    <icon src="%s"/>\n' % esc_attr(u) for _, u in a["artwork"]]
        o.append('    <episode-num system="%s">%s</episode-num>\n' % (ID_SYSTEM, esc_text(a["asset_id"])))
        if a["rating_value"]:
            o.append('    <rating system="%s">\n' % esc_attr(a["rating_system"]) if a["rating_system"] else "    <rating>\n")
            o.append("      <value>%s</value>\n    </rating>\n" % esc_text(a["rating_value"]))
        o.append("  </programme>\n")
    o.append("</tv>\n")
    return "".join(o).encode("utf-8")


def now_and_next(progs, at):
    for k, p in enumerate(progs):
        if p[0] <= at < p[1]:
            return p, progs[k + 1] if k + 1 < len(progs) else None
    if progs and at < progs[0][0]:
        return None, progs[0]
    return None, None


def programme_doc(p):
    s, e, a = p
    geo = None
    if a["geo"] == ONLY:
        geo = {"allow": a["countries"]}
    elif a["geo"] == EXCEPT:
        geo = {"deny": a["countries"]}
    return {"start_ms": s, "stop_ms": e, "asset_id": a["asset_id"], "title": a["title"], "description": a["description"],
            "content_uri": a["content_uri"], "categories": a["categories"],
            "rating": {"system": a["rating_system"], "value": a["rating_value"]} if a["rating_value"] else None,
            "artwork": [{"sha256": h.hex(), "uri": u} for h, u in a["artwork"]], "geo": geo,
            "rights_holder": a["rights_holder"], "duration_ms": a["duration_ms"]}


def jcs(v):
    """RFC 8785 over this document's values: sorted members (ASCII names), no space, strings as
    JSON escapes them."""
    return json.dumps(v, sort_keys=True, separators=(",", ":"), ensure_ascii=False)


def now_next_doc(r, l, at):
    """§7.3."""
    now, nxt = now_and_next(all_programmes(r, l), at)
    v = {"kind": NOW_NEXT_KIND, "channel": l["channel_id"], "channel_name": l["channel_name"],
         "schedule_root": r["schedule_root"].hex(), "lineup": lineup_digest(l).hex(), "at_ms": at,
         "now": programme_doc(now) if now else None, "next": programme_doc(nxt) if nxt else None}
    return (jcs(v) + "\n").encode("utf-8")


def plain_uint(s):
    return digits(s) and (s == "0" or s[0] != "0")


def read_guide(b):
    """§8.2 step 8: the guide as read, or None."""
    if b[:1] == b"<":
        root = read_feed_xml(b)
        if root is None or root["name"] != "tv":
            return None
        url = root["attrs"].get("source-data-url", "")
        if not url.startswith(GUIDE_URN):
            return None
        parts = url[len(GUIDE_URN):].split(":")
        if len(parts) != 4 or not all(hexn(x, 32) is not None for x in parts[:2]) or not all(plain_uint(x) for x in parts[2:]):
            return None
        frm, to = int(parts[2]), int(parts[3])
        if frm >= to:
            return None
        progs = []
        for p in SPV.kids(root, "programme"):
            try:
                s, e = xmltv_time(p["attrs"].get("start", "")), xmltv_time(p["attrs"].get("stop", ""))
            except Refuse:
                return None
            ids = [SPV.text(x) for x in SPV.kids(p, "episode-num") if x["attrs"].get("system", "onscreen") == ID_SYSTEM]
            if not ids:
                return None
            progs.append((s, e, ids[0]))
        return {"format": "xmltv", "root": parts[0], "lineup": parts[1], "from": frm, "to": to, "programmes": progs}
    if b[:1] == b"{":
        v = strict_json(b)
        if not exactly(v, ["kind", "channel", "channel_name", "schedule_root", "lineup", "at_ms", "now", "next"]):
            return None
        if not jint(v["at_ms"]) or not isinstance(v["schedule_root"], str) or not isinstance(v["lineup"], str):
            return None
        out = []
        for k in ("now", "next"):
            p = v[k]
            if p is None:
                out.append(None)
            elif isinstance(p, dict) and u64(p.get("start_ms")) and u64(p.get("stop_ms")) and isinstance(p.get("asset_id"), str):
                out.append((p["start_ms"], p["stop_ms"], p["asset_id"]))
            else:
                return None
        return {"format": "now-next", "root": v["schedule_root"], "lineup": v["lineup"], "at": v["at_ms"], "now": out[0],
                "next": out[1]}
    return None


# ---------------------------------------------------------------- §8 verification

def verify(receipt, lineup, guide):
    """§8.2 and §8.3: the facts."""
    f = {"verdict": "verified", "part": None, "receipt": None, "lineup": None, "guide": None, "entry": None}

    def fail(part, reason):
        f["verdict"], f["part"] = reason, part
        return f

    r = read_receipt(receipt)
    if r is None:
        return fail("receipt", "channel_receipt_malformed")
    f["receipt"] = receipt_identity(r).hex()
    bad = ED.verify(r["signer_pubkey"], receipt_payload(r), r["sig"]) if "receipt_sig" not in SKIP else None
    if bad:
        return fail("receipt", bad)
    if not receipt_rules(r):
        return fail("receipt", "channel_receipt_invalid")
    l = read_lineup(lineup)
    if l is None:
        return fail("lineup", "lineup_malformed")
    f["lineup"] = lineup_digest(l).hex()
    bad = ED.verify(l["signer"], LINEUP_DOMAIN + lineup_body(l, l["signer"]), l["sig"]) if "lineup_sig" not in SKIP else None
    if bad:
        return fail("lineup", bad)
    if l["signer"] != r["signer_pubkey"] and "signer" not in SKIP:
        return fail("lineup", "lineup_signer_mismatch")
    if not describes(l, r):
        return fail("lineup", "lineup_schedule_mismatch")
    if guide is None:
        return f
    g = read_guide(guide)
    if g is None:
        return fail("guide", "guide_malformed")
    sched = r["schedule"]

    def aired(s, e):
        return any(sched[i]["start_ms"] < e and sched[i]["start_ms"] + sched[i]["duration_ms"] > s
                   for i in r["aired"] if i < len(sched))

    progs = all_programmes(r, l)
    if g["format"] == "xmltv":
        shown = g["programmes"]
        f["guide"] = {"format": "xmltv", "from_ms": g["from"], "to_ms": g["to"], "at_ms": None}
    else:
        shown = [p for p in (g["now"], g["next"]) if p is not None]
        f["guide"] = {"format": "now-next", "from_ms": None, "to_ms": None, "at_ms": g["at"]}
    f["guide"]["programmes"] = [{"start_ms": s, "stop_ms": e, "asset_id": i, "aired": aired(s, e)} for s, e, i in shown]
    if (g["root"] != r["schedule_root"].hex() or g["lineup"] != f["lineup"]) and "source" not in SKIP:
        return fail("guide", "guide_source_mismatch")
    key = (lambda p: (p[0], p[1], p[2]["asset_id"]))
    if g["format"] == "xmltv":
        want = [key(p) for p in progs if p[0] < g["to"] and p[1] > g["from"]]
        for i, p in enumerate(g["programmes"]):
            if p not in want and "entries" not in SKIP:
                f["entry"] = i
                return fail("guide", "guide_entry_unscheduled")
        expect = xmltv_guide(r, l, g["from"], g["to"])
    else:
        wn, wx = now_and_next(progs, g["at"])
        for i, (got, w) in enumerate(((g["now"], wn), (g["next"], wx))):
            if got != (key(w) if w else None) and "entries" not in SKIP:
                f["entry"] = i
                return fail("guide", "guide_entry_unscheduled")
        expect = now_next_doc(r, l, g["at"])
    if guide != expect and "bytes" not in SKIP:
        return fail("guide", "guide_mismatch")
    return f


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

# ---------------------------------------------------------------- §7.4 the cue tags, §8.4

PL_VERSION = 3                                                                    # RFC 8216 §7
MAX_URI, MAX_SEGMENT, WINDOW_ENDS_BEFORE = 2048, U32, 253402300800000             # §7.4


def read_cues(b):
    """§8.4 step 4: the cues as `cues.json` holds them, or None."""
    v = strict_json(b)
    if not isinstance(v, list):
        return None
    out = []
    for c in v:
        if not isinstance(c, dict) or set(c) != {"break", "start_ms", "payload"}:
            return None
        if not isinstance(c["break"], str) or not jint(c["start_ms"]):
            return None
        p = c["payload"]
        if not isinstance(p, str) or len(p) % 2 or any(ch not in "0123456789abcdef" for ch in p):
            return None
        out.append({"break": c["break"], "start_ms": c["start_ms"], "payload": bytes.fromhex(p)})
    return out


def schedule_breaks(r):
    """§7.4: the receipt's ad entries of a duration above 0, in schedule order."""
    return [(e["asset_id"], e["start_ms"], e["duration_ms"]) for e in r["schedule"] if e["kind"] == "ad" and e["duration_ms"] > 0]


def payload_section(p):
    """The section a cue's payload carries: (True, section) of form 1, (True, None) of form 0,
    (False, None) when neither."""
    if len(p) < 10:
        return False, None
    if p[9] == 0:
        return True, None
    if p[9] == 1 and len(p) >= 12:
        n = p[10] | (p[11] << 8)
        sec = p[12:12 + n]
        if len(sec) == n and section_reads(sec):
            return True, sec
    return False, None


def bind_cues(r, cues):
    """§8.4 step 5: the breaks, each (id, start, duration, section); or Refuse with the index."""
    breaks = schedule_breaks(r)
    out = []
    for i in range(max(len(breaks), len(cues))):
        if i >= len(breaks) or i >= len(cues):
            check("cues_count", False, ("cues_mismatch", i))
            break
        bid, start, dur = breaks[i]
        c = cues[i]
        check("cues_break", c["break"] == bid and c["start_ms"] == start, ("cues_mismatch", i))
        ok, sec = payload_section(c["payload"])
        check("cues_payload", ok and dur <= U32 and cue_payload({"id": bid, "duration": dur, "section": sec}) == c["payload"], ("cues_mismatch", i))
        out.append({"id": bid, "start": start, "duration": dur, "end": start + dur, "section": sec})
    return out


def window_holds(w):
    """§7.4's bounds, each a check of its own."""
    segs = w["segments"]
    end = w["start"] + sum(d for _, d in segs)
    bounds = {
        "pl_segments": len(segs) > 0,
        "pl_uri_empty": all(len(u) > 0 for u, _ in segs),
        "pl_uri_long": all(len(u.encode()) <= MAX_URI for u, _ in segs),
        "pl_uri_chars": all(0x21 <= ord(c) <= 0x7E for u, _ in segs for c in u),
        "pl_uri_hash": not any(u.startswith("#") for u, _ in segs),
        "pl_duration_zero": all(d >= 1 for _, d in segs),
        "pl_duration_long": all(d <= MAX_SEGMENT for _, d in segs),
        "pl_sequence": w["sequence"] <= MAX_INSTANT,
        "pl_end": end < WINDOW_ENDS_BEFORE,
    }
    return all(ok for name, ok in bounds.items() if name not in SKIP)


def seg_spans(w):
    out, at = [], w["start"]
    for _, d in w["segments"]:
        out.append((at, at + d))
        at += d
    return out


def first_in(b, sp):
    """The first segment that puts a break in the window (§7.4), or None."""
    for k, (s, e) in enumerate(sp):
        if e == b["start"] or b["start"] <= s < b["end"] or s == b["end"]:
            return k
    return None


def pl_span(ms):
    return "%d.%03d" % (ms // 1000, ms % 1000)


def pl_date(ms):
    d = datetime.datetime(1970, 1, 1) + datetime.timedelta(milliseconds=ms)
    return d.strftime("%Y-%m-%dT%H:%M:%S") + ".%03dZ" % (ms % 1000)


def render(breaks, w):
    """§7.4: the playlist of a window, or Refuse((reason, index))."""
    check("pl_window", window_holds(w), ("playlist_malformed", None))
    sp = seg_spans(w)
    for k, (s, e) in enumerate(sp):
        check("aligned", not any(s < x < e for b in breaks for x in (b["start"], b["end"])), ("cue_unaligned", k))
    for i, b in enumerate(breaks):
        if first_in(b, sp) is not None and b["section"] is not None:
            for name, c in (("writable_quote", '"'), ("writable_line_feed", "\n"), ("writable_carriage_return", "\r")):
                check(name, c not in b["id"], ("cue_unwritable", i))
    target = max((d + 999) // 1000 for _, d in w["segments"])
    out = ["#EXTM3U", "#EXT-X-VERSION:%d" % PL_VERSION, "#EXT-X-TARGETDURATION:%d" % target,
           "#EXT-X-MEDIA-SEQUENCE:%d" % w["sequence"], "#EXT-X-PROGRAM-DATE-TIME:" + pl_date(w["start"])]
    for k, ((uri, d), (s, _)) in enumerate(zip(w["segments"], sp)):
        for b in breaks:
            if b["section"] is None:
                continue
            rid = "%s@%d" % (b["id"], b["start"])
            if first_in(b, sp) == k:
                out.append('#EXT-X-DATERANGE:ID="%s",START-DATE="%s",PLANNED-DURATION=%s,SCTE35-OUT=0x%s'
                           % (rid, pl_date(b["start"]), pl_span(b["duration"]), b["section"].hex().upper()))
            if s == b["end"]:
                out.append('#EXT-X-DATERANGE:ID="%s",START-DATE="%s",DURATION=%s' % (rid, pl_date(b["start"]), pl_span(b["duration"])))
        if any(b["end"] == s for b in breaks):
            out.append("#EXT-X-CUE-IN")
        starting = [b for b in breaks if b["start"] == s]
        inside = [b for b in breaks if b["start"] < s < b["end"]]
        if starting:
            out.append("#EXT-X-CUE-OUT:" + pl_span(starting[0]["duration"]))
        elif inside:
            b = inside[0]
            out.append("#EXT-X-CUE-OUT-CONT:ElapsedTime=%s,Duration=%s" % (pl_span(s - b["start"]), pl_span(b["duration"])))
        out += ["#EXTINF:%s," % pl_span(d), uri]
    if w["ended"]:
        out.append("#EXT-X-ENDLIST")
    return "".join(ln + "\n" for ln in out)


def read_cue_playlist(b):
    """§8.4 step 6: a playlist's window, or None."""
    try:
        t = b.decode("utf-8")
    except UnicodeDecodeError:
        return None
    if not t.endswith("\n"):
        return None
    lines = t[:-1].split("\n")
    if lines[0] != "#EXTM3U":
        return None
    seq, first, ended, segs, i = None, None, False, [], 1
    try:
        while i < len(lines):
            ln = lines[i]
            if ln.startswith("#EXT-X-MEDIA-SEQUENCE:"):
                v = ln[len("#EXT-X-MEDIA-SEQUENCE:"):]
                if seq is not None or not plain_uint(v) or int(v) > MAX_INSTANT:
                    return None
                seq = int(v)
            elif ln.startswith("#EXT-X-PROGRAM-DATE-TIME:"):
                if first is None and not segs:
                    first = hls_date_time(ln[len("#EXT-X-PROGRAM-DATE-TIME:"):])
            elif ln == "#EXT-X-ENDLIST":
                ended = True
            elif ln.startswith("#EXTINF:"):
                v = ln[len("#EXTINF:"):]
                if not v.endswith(",") or i + 1 >= len(lines):
                    return None
                segs.append((lines[i + 1], duration(v[:-1])))
                i += 1
            i += 1
    except Refuse:
        return None
    if seq is None or first is None:
        return None
    w = {"sequence": seq, "start": first, "segments": segs, "ended": ended}
    return w if window_holds(w) else None


def verify_playlist(receipt, cues, playlist):
    """§8.4: the facts."""
    f = {"verdict": "verified", "part": None, "receipt": None, "window": None, "index": None}

    def fail(part, reason, index=None):
        f["verdict"], f["part"], f["index"] = reason, part, index
        return f

    r = read_receipt(receipt)
    if r is None:
        return fail("receipt", "channel_receipt_malformed")
    f["receipt"] = receipt_identity(r).hex()
    bad = ED.verify(r["signer_pubkey"], receipt_payload(r), r["sig"]) if "receipt_sig" not in SKIP else None
    if bad:
        return fail("receipt", bad)
    if not receipt_rules(r):
        return fail("receipt", "channel_receipt_invalid")
    c = read_cues(cues)
    if c is None:
        return fail("cues", "cues_malformed")
    try:
        breaks = bind_cues(r, c)
    except Refuse as e:
        return fail("cues", e.reason[0], e.reason[1])
    w = read_cue_playlist(playlist)
    if w is None:
        return fail("playlist", "playlist_malformed")
    sp = seg_spans(w)
    f["window"] = {"media_sequence": w["sequence"], "start_ms": w["start"], "segments": len(w["segments"]), "ended": w["ended"],
                   "breaks": [{"break": b["id"], "start_ms": b["start"], "duration_ms": b["duration"], "scte35": b["section"] is not None}
                              for b in breaks if first_in(b, sp) is not None]}
    try:
        text = render(breaks, w)
    except Refuse as e:
        return fail("playlist", e.reason[0], e.reason[1])
    if "pl_bytes" not in SKIP and text.encode("utf-8") != playlist:
        return fail("playlist", "playlist_mismatch")
    return f


def window_of(v):
    return {"sequence": v["media_sequence"], "start": v["start_ms"], "segments": [(s["uri"], s["duration_ms"]) for s in v["segments"]],
            "ended": v["ended"]}


def check_playlist(v, files):
    """None when a window renders what the corpus states, else why not."""
    r = read_receipt(files[v["receipt"]])
    c = read_cues(files[v["cues"]])
    if r is None or c is None:
        return "the receipt or the cues do not read"
    o = v["outcome"]
    try:
        text = render(bind_cues(r, c), window_of(v["window"]))
    except Refuse as e:
        got = {"refused": e.reason[0], "index": e.reason[1]}
        return None if o == got else "refused %s, the corpus states %s" % (got, o)
    if "playlist" not in o or files.get(o["playlist"]) != text.encode("utf-8"):
        return "renders another playlist than the corpus's %s" % o
    return None


def check_playlist_verification(v, files):
    files = apply_edits(files, v["edits"])
    facts = verify_playlist(files[v["receipt"]], files[v["cues"]], files[v["playlist"]])
    return None if facts == v["facts"] else "states %s, the corpus %s" % (facts, v["facts"])


def apply_edits(files, edits):
    out = dict(files)
    for e in edits:
        b = out[e["file"]]
        if "pad" in e:
            out[e["file"]] = b + b" " * e["pad"]
            continue
        s = b.decode("utf-8")
        if s.count(e["find"]) != 1:
            raise ValueError("%s: an edit's find does not occur once" % e["file"])
        with_ = e["find"] + e["replace"] * e["repeat"] if "repeat" in e else e["replace"]
        out[e["file"]] = s.replace(e["find"], with_, 1).encode("utf-8")
    return out


def import_inputs(v, files):
    inp = dict(v["inputs"])
    inp["artwork_files"] = {n: files[n] for n in inp["artwork"]}
    if inp["window"] is not None:
        inp["window"] = tuple(inp["window"])
    return inp


def check_import(doc, v, files, op_key):
    """None when the import gets the outcome the corpus states, else why not."""
    files = apply_edits(files, v["edits"])
    o = v["outcome"]
    try:
        got = do_import(import_inputs(v, files), files)
    except Refuse as e:
        return None if o.get("refused") == e.reason else "refused %s, the corpus states %s" % (e.reason, o)
    if "refused" in o:
        return "imported, the corpus states %s" % o
    r = read_receipt(files[o["receipt"]])
    if r is None:
        return "the receipt does not read"
    inp = v["inputs"]
    want_grant = {"grant_hash": bytes(32), "capability": "video.channel", "joule_ceiling_micro": U64, "funds_ceiling": None}
    ws, we = got["window"]
    same = (r["channel_id"] == inp["channel"] and r["schedule"] == got["entries"] and r["schedule_root"] == schedule_root(got["entries"])
            and (r["window_start_ms"], r["window_end_ms"]) == (ws, we) and r["aired"] == aired_of(got["entries"], ws, we)
            and r["work_schedule_ops"] == len(got["entries"]) and r["joules_micro"] == 0 and r["class"] is None
            and r["grant"] == want_grant and r["parent"] == (bytes.fromhex(inp["parent_receipt_hash"]) if inp["parent_receipt_hash"] else None)
            and r["signer_pubkey"] == op_key and r["signer_id"] == inp["signer_id"])
    if not same:
        return "the receipt is not the import's"
    if ED.verify(r["signer_pubkey"], receipt_payload(r), r["sig"]) or receipt_identity(r).hex() != o["receipt_hash"]:
        return "the receipt's signature or identity"
    lb = bytes.fromhex(o["lineup"])
    body = lineup_body(got["lineup"], op_key)
    if lb[:-64] != body or ED.verify(op_key, LINEUP_DOMAIN + body, lb[-64:]) or sha(body).hex() != o["lineup_digest"]:
        return "the lineup is not the import's, or its signature does not verify"
    if got["cues"] != o["cues"]:
        return "the cues differ"
    return None


def check_verification(v, files):
    files = apply_edits(files, v["edits"])
    g = files[v["guide"]] if v["guide"] else None
    facts = verify(files[v["receipt"]], bytes.fromhex(v["lineup"]), g)
    return None if facts == v["facts"] else "states %s, the corpus %s" % (facts, v["facts"])


def corpus_files(doc):
    names = {"vectors.json", "README.md"}
    for v in doc["imports"]:
        i = v["inputs"]
        names |= {x for x in (i["xmltv"], i["json"], i["playlist"]) if x} | {m["file"] for m in i["mrss"]} | set(i["artwork"])
        if "receipt" in v["outcome"]:
            names.add(v["outcome"]["receipt"])
    for v in doc["verifications"]:
        names |= {x for x in (v["receipt"], v["guide"]) if x}
    for v in doc["playlists"]:
        names |= {v["receipt"], v["cues"]} | ({v["outcome"]["playlist"]} if "playlist" in v["outcome"] else set())
    for v in doc["playlist_verifications"]:
        names |= {v["receipt"], v["cues"], v["playlist"]}
    return names


def check_all(doc, files, listing, quiet=False):
    """The failures of every vector and of the corpus as a whole."""
    fails = []
    op_key = bytes.fromhex(doc["operator"]["key"])
    for v in doc["imports"]:
        try:
            why = check_import(doc, v, files, op_key)
        except (KeyError, ValueError, TypeError) as e:
            why = "broken: %r" % (e,)
        if why:
            fails.append("import %s: %s" % (v["name"], why))
    for v in doc["verifications"]:
        try:
            why = check_verification(v, files)
        except (KeyError, ValueError, TypeError) as e:
            why = "broken: %r" % (e,)
        if why:
            fails.append("verification %s: %s" % (v["name"], why))
    for v in doc["playlists"]:
        try:
            why = check_playlist(v, files)
        except (KeyError, ValueError, TypeError) as e:
            why = "broken: %r" % (e,)
        if why:
            fails.append("playlist %s: %s" % (v["name"], why))
    for v in doc["playlist_verifications"]:
        try:
            why = check_playlist_verification(v, files)
        except (KeyError, ValueError, TypeError) as e:
            why = "broken: %r" % (e,)
        if why:
            fails.append("playlist verification %s: %s" % (v["name"], why))
    added, reused = reasons()
    seen = {v["outcome"]["refused"] for v in doc["imports"] if "refused" in v["outcome"]}
    seen |= {v["facts"]["verdict"] for v in doc["verifications"]}
    seen |= {v["outcome"]["refused"] for v in doc["playlists"] if "refused" in v["outcome"]}
    seen |= {v["facts"]["verdict"] for v in doc["playlist_verifications"]}
    for r in added + reused:
        if r not in seen:
            fails.append("the corpus gives no %s" % r)
    for r in seen - set(added) - set(reused) - {"verified"}:
        fails.append("%s is not a reason of §10" % r)
    if listing != corpus_files(doc):
        fails.append("the files are not the corpus's: %s" % sorted(listing ^ corpus_files(doc)))
    if not quiet:
        for f in fails:
            print("FAIL", f)
    return fails


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


def flip(h, at):
    c = "0" if h[at] != "0" else "1"
    return h[:at] + c + h[at + 1:]


def mutations(doc, files, listing):
    """Copies of the corpus with one defect each: every one must fail."""
    out = []

    def m(name, f):
        d, fs, ls = copy.deepcopy(doc), dict(files), set(listing)
        f(d, fs, ls)
        out.append((name, d, fs, ls))

    imp = lambda d, n: next(v for v in d["imports"] if v["name"] == n)          # noqa: E731
    ver = lambda d, n: next(v for v in d["verifications"] if v["name"] == n)    # noqa: E731
    m("a lineup byte of the sample's outcome", lambda d, fs, ls: imp(d, "sample")["outcome"].__setitem__("lineup", flip(imp(d, "sample")["outcome"]["lineup"], 120)))
    m("the sample's receipt hash", lambda d, fs, ls: imp(d, "sample")["outcome"].__setitem__("receipt_hash", flip(imp(d, "sample")["outcome"]["receipt_hash"], 0)))
    m("a cue's payload", lambda d, fs, ls: imp(d, "sample")["outcome"]["cues"][0].__setitem__("payload", flip(imp(d, "sample")["outcome"]["cues"][0]["payload"], 4)))
    m("a cue's start", lambda d, fs, ls: imp(d, "sample")["outcome"]["cues"][1].__setitem__("start_ms", 1))
    m("a refusal's code", lambda d, fs, ls: imp(d, "gap")["outcome"].__setitem__("refused", "import_overlap"))
    m("an edit dropped", lambda d, fs, ls: imp(d, "no_stop").__setitem__("edits", []))
    m("a verdict", lambda d, fs, ls: ver(d, "guide_mismatch")["facts"].__setitem__("verdict", "verified"))
    m("an aired flag", lambda d, fs, ls: ver(d, "xmltv_guide")["facts"]["guide"]["programmes"][2].__setitem__("aired", True))
    m("the unscheduled entry's index", lambda d, fs, ls: ver(d, "guide_entry_unscheduled")["facts"].__setitem__("entry", 1))
    m("a guide's title", lambda d, fs, ls: fs.__setitem__("sample.guide.xml", fs["sample.guide.xml"].replace(b"Tidepool Kitchen", b"Tidepool Kitchens")))
    m("a now/next title", lambda d, fs, ls: fs.__setitem__("sample.now-next.json", fs["sample.now-next.json"].replace(b"Tidepool Kitchen", b"Tidepool Kitchens")))
    m("an artwork file", lambda d, fs, ls: fs.__setitem__("art-tk-7.svg", fs["art-tk-7.svg"] + b" "))
    m("a catalogue's duration", lambda d, fs, ls: fs.__setitem__("sample-a.mrss.xml", fs["sample-a.mrss.xml"].replace(b'duration="1620"/>', b'duration="1621"/>', 1)))
    m("a break's start", lambda d, fs, ls: fs.__setitem__("sample.m3u8", fs["sample.m3u8"].replace(b"18:45:30.500Z", b"18:45:31.500Z")))
    m("an SCTE-35 section", lambda d, fs, ls: fs.__setitem__("sample.m3u8", fs["sample.m3u8"].replace(b"03E97FEFFE", b"03EA7FEFFE", 1)))
    m("the receipt's window", lambda d, fs, ls: fs.__setitem__("sample.channel.json", fs["sample.channel.json"].replace(b'"window_end_ms": 1793300400000', b'"window_end_ms": 1793300401000')))
    m("a file added", lambda d, fs, ls: (fs.__setitem__("extra.txt", b"x"), ls.add("extra.txt")))
    m("a vector of a reason dropped", lambda d, fs, ls: d.__setitem__("imports", [v for v in d["imports"] if v["name"] != "window_outside"]))
    pl = lambda d, n: next(v for v in d["playlists"] if v["name"] == n)                 # noqa: E731
    plv = lambda d, n: next(v for v in d["playlist_verifications"] if v["name"] == n)   # noqa: E731
    m("a playlist's EXT-X-CUE-OUT duration", lambda d, fs, ls: fs.__setitem__("sample.out-in.m3u8", fs["sample.out-in.m3u8"].replace(b"CUE-OUT:90.000", b"CUE-OUT:90.001")))
    m("a playlist's elapsed time", lambda d, fs, ls: fs.__setitem__("sample.inside.m3u8", fs["sample.inside.m3u8"].replace(b"ElapsedTime=40.000", b"ElapsedTime=41.000")))
    m("a date range's section", lambda d, fs, ls: fs.__setitem__("sample.offset.m3u8", fs["sample.offset.m3u8"].replace(b"SCTE35-OUT=0xFC30", b"SCTE35-OUT=0xFC31")))
    m("the ended window's end", lambda d, fs, ls: fs.__setitem__("sample.offset.m3u8", fs["sample.offset.m3u8"].replace(b"#EXT-X-ENDLIST\n", b"")))
    m("a window's first segment", lambda d, fs, ls: pl(d, "out_in")["window"].__setitem__("start_ms", pl(d, "out_in")["window"]["start_ms"] + 10000))
    m("a refusal's index", lambda d, fs, ls: pl(d, "unaligned_end")["outcome"].__setitem__("index", 9))
    m("a cues file's payload", lambda d, fs, ls: fs.__setitem__("no-section.cues.json", fs["no-section.cues.json"].replace(b'"payload": "04', b'"payload": "03', 1)))
    m("a playlist verification's break", lambda d, fs, ls: plv(d, "playlist_inside")["facts"]["window"]["breaks"][0].__setitem__("scte35", False))
    m("a playlist verification's index", lambda d, fs, ls: plv(d, "cues_missing")["facts"].__setitem__("index", 2))
    m("a target duration rounded down", lambda d, fs, ls: fs.__setitem__("sample.fractional.m3u8", fs["sample.fractional.m3u8"].replace(b"TARGETDURATION:7\n", b"TARGETDURATION:6\n")))
    m("a span's milliseconds dropped", lambda d, fs, ls: fs.__setitem__("sample.fractional.m3u8", fs["sample.fractional.m3u8"].replace(b"ElapsedTime=6.006,", b"ElapsedTime=6.000,")))
    m("the last instant's window a millisecond later", lambda d, fs, ls: pl(d, "last_instant")["window"].__setitem__("start_ms", pl(d, "last_instant")["window"]["start_ms"] + 1))
    m("a refused URI's bound", lambda d, fs, ls: pl(d, "uri_too_long")["window"]["segments"][0].__setitem__("uri", pl(d, "uri_too_long")["window"]["segments"][0]["uri"][1:]))
    return out


def main(corpus_dir):
    global SKIP
    doc, files, listing = load(corpus_dir)
    if check_all(doc, files, listing):
        return 1
    print("channel lineup corpus: %d imports, %d verifications, %d playlists and %d playlist verifications hold"
          % (len(doc["imports"]), len(doc["verifications"]), len(doc["playlists"]), len(doc["playlist_verifications"])))
    bad = 0
    for name, d, fs, ls in mutations(doc, files, listing):
        if not check_all(d, fs, ls, quiet=True):
            print("FAIL mutation not caught:", name)
            bad += 1
    print("mutations: %d copies, %d not caught" % (len(mutations(doc, files, listing)), bad))
    checks = ["inputs", "size", "fields", "timezone", "no_stop", "asset", "duplicate", "counts", "doctype", "conflict",
              "scte35", "break_duration", "duration_unknown", "gap", "overlap", "whole_seconds", "owner", "artwork",
              "break_overlap", "break_outside", "window", "receipt_sig", "receipt_rules", "lineup_sig", "signer",
              "describes_root", "describes_starts", "describes_assets", "describes_set", "source", "entries", "bytes",
              "cues_count", "cues_break", "cues_payload", "pl_window", "pl_segments", "pl_uri_empty", "pl_uri_long",
              "pl_uri_chars", "pl_uri_hash", "pl_duration_zero", "pl_duration_long", "pl_sequence", "pl_end", "aligned",
              "writable_quote", "writable_line_feed", "writable_carriage_return", "pl_bytes"]
    weak = 0
    for c in checks:
        SKIP = {c}
        try:
            caught = bool(check_all(doc, files, listing, quiet=True))
        except Exception:
            caught = True
        if not caught:
            print("FAIL weakened verifier not caught: without", c)
            weak += 1
    SKIP = set()
    print("weakened verifiers: %d checks dropped one at a time, %d not caught" % (len(checks), weak))
    return 1 if bad or weak else 0


if __name__ == "__main__":
    if len(sys.argv) != 2:
        print(__doc__)
        sys.exit(2)
    sys.exit(main(sys.argv[1]))