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

Written from the extension's text. It shares no code with wai-rs, which generates the corpus, and
never runs it. Acknowledgements and their item lists are read by the settlement corpus's second
implementation, settlement_verify.py (settlement §3.6, the tree root of jwp-receipts §2.2); claims,
their quartile and their run by the ad-presentation corpus's, ad_presentation_verify.py
(ad-presentation §5, §6); the observation log and a figure's acquisition class by the
live-metering corpus's, live_metering_verify.py (live-metering §3.6, §6.1); and every signature
with ../claim-conformance/_ed25519.py, WAI's strict rule (SPEC §7 *Signatures*) written from RFC
8032. Everything else is here, from the standard library:

- the channel id, the session and an origin's upstream (§3, settlement §14.2);
- the session record: reading, rules and signature (§5);
- the bundle and its verification in §10.2's order, with each refusal's place;
- each claim's verdict and facts (§10.2 step 9);
- the channel report, its bytes and its recomputation (§11).

For every session and report it requires the outcome the corpus states; of the corpus, that every
reason of §14 and every code it borrows appears in it and that no other file is there. Then it
mutates copies of the corpus, one defect each, and requires every copy to fail; and it drops each
of its own checks in turn and requires the corpus to catch the weakened verifier.

    python3 stream_audit_verify.py <live-conformance/audit>

Exit status 0 only if every case verifies, every mutated copy fails and every weakened verifier
fails. Python 3.8 or later; standard library only.
"""

import copy
import hashlib
import json
import os
import re
import struct
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance"))
import live_events_verify as LEV  # noqa: E402
import live_metering_verify as LMV  # noqa: E402
import settlement_verify as SV  # noqa: E402
import ad_presentation_verify as APV  # noqa: E402
import _ed25519 as ED  # noqa: E402

EXTENSION = os.path.join(HERE, "..", "extensions", "stream-audit.md")

Refuse = LEV.Refuse

SESSION_MAGIC = b"WAS1"                                   # §5.1
SESSION_DOMAIN = b"wai:audit-session\x01"                 # §5.4
CHANNEL_DOMAIN = b"wai:audit-channel\x01"                 # §3.1
MARKER_DOMAIN = b"wai:audit-marker\x01"                   # §6.1
REPORT_DOMAIN = b"wai:audit-report\x01"                   # §11.3
ORIGIN_DOMAIN = b"wai:settle-origin\x01"                  # settlement §14.2
SESSION_KEY_DOMAIN = b"wai:ad-presentation-session\x01"   # §3.2, ad-presentation §3
NO_STARTUP = 0xFFFFFFFF
QOE_MIN = 50
LIVE, ON_DEMAND = 1, 2
ENDED = ["ended", "user", "hidden", "idle", "error"]                 # §8.2
ROLES = ["origin", "delivery", "sink_decode", "display"]             # §5.3
CLASS_NAMES = {1: "live", 2: "fetch", 3: "fill", 4: "ancillary"}     # settlement §3.3
LABELS = {0: "unmetered", 2: "ModelBased", 3: "Estimator", 4: "OnChipCounter", 5: "CalibratedInstrument"}
AUDIBLE, AD = 1, 2
ZERO = bytes(32)

SKIP = set()


def reasons_of_extension():
    """§14's codes, read from the text."""
    with open(EXTENSION, encoding="utf-8") as f:
        text = f.read()
    at = text.index("\n## 14. Reasons")
    end = text.index("\n## ", at + 1)
    return [line.split("`")[1] for line in text[at:end].splitlines() if line.startswith("| `")]


REASONS = reasons_of_extension()
REUSED = ["truncated", "malformed", "unknown_version", "bad_key", "bad_signature", "record_missing", "energy_invalid",
          "qoe_invalid", "ack_invalid", "self_acknowledged", "items_invalid", "items_mismatch", "ack_outside_period",
          "ack_overlap", "ack_chain_broken", "claim_malformed", "claim_duplicate", "claim_asset_mismatch"]


class Stop(Exception):
    """A refusal and its place (§10.2)."""

    def __init__(self, reason, at):
        super().__init__(reason)
        self.reason, self.at = reason, at


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


def given_digest(b):
    return sha(b[:max(len(b) - 64, 0)])


def channel_id(name):
    """§3.1, or None for a name that is not 1 to 255 bytes of printable ASCII."""
    b = name.encode("utf-8")
    if not 1 <= len(b) <= 255 or any(c < 0x21 or c > 0x7E for c in b):
        return None
    return sha(CHANNEL_DOMAIN, b)


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


def upstream_of(origin):
    return sha(ORIGIN_DOMAIN, origin)


def marker_digest(marker):
    return sha(MARKER_DOMAIN, marker)


def unhex(s):
    if not isinstance(s, str) or len(s) % 2 or not re.fullmatch(r"[0-9a-f]*", s):
        return None
    return bytes.fromhex(s)


def check_sig(key, domain, body, sig):
    bad = ED.verify(key, domain + body, sig)
    if bad:
        raise Refuse(bad)


class R:
    """A cursor for the record and for live-metering's class reader; running out is `truncated`."""

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

    def take(self, n):
        if self.p + n > len(self.b):
            raise Refuse("truncated")
        s = self.b[self.p:self.p + n]
        self.p += n
        return s

    def u8(self):
        return self.take(1)[0]

    def u16(self):
        return struct.unpack("<H", self.take(2))[0]

    def u32(self):
        return struct.unpack("<I", self.take(4))[0]

    def u64(self):
        return struct.unpack("<Q", self.take(8))[0]

    def be32(self):
        return struct.unpack(">I", self.take(4))[0]

    def be64(self):
        return struct.unpack(">Q", self.take(8))[0]


# ---------------------------------------------------------------- §5 the session record

def origin_ok(o):
    """§5.5."""
    if not 1 <= len(o) <= 2048 or any(c < 0x21 or c > 0x7E for c in o):
        return False
    for scheme in (b"https://", b"http://"):
        if o.startswith(scheme):
            rest = o[len(scheme):]
            return rest != b"" and not any(c in b"/?#@" for c in rest)
    return False


def ascending(v):
    return all(x < y for x, y in zip(v, v[1:]))


def read_session(b):
    """§5.5: the record, its rules, its energy entries, then its signature."""
    if len(b) < 4:
        raise Refuse("truncated")
    if b[:4] != SESSION_MAGIC:
        raise Refuse("malformed")
    if len(b) < 5:
        raise Refuse("truncated")
    if b[4] != 1:
        raise Refuse("unknown_version")
    r = R(b, 5)
    s = {"channel": r.take(32), "owner": r.take(32), "player": r.take(32), "session": r.take(16), "presentation": r.u8(),
         "start": r.u64(), "end": r.u64(), "ended": r.u8()}
    s["origins"] = [r.take(r.u16()) for _ in range(r.u8())]
    s["acks"] = [r.take(32) for _ in range(r.u16())]
    s["claims"] = [r.take(32) for _ in range(r.u16())]
    s["spans"] = []
    for _ in range(r.u32()):
        start, length, vis, flags = struct.unpack("<IIHB", r.take(11))
        s["spans"].append({"start": start, "length": length, "visible": vis, "flags": flags})
    s["startup_ms"], s["stalls"], s["stall_ms"] = r.u32(), r.u32(), r.u64()
    s["repeat_objects"], s["repeat_bytes"] = r.u32(), r.u64()
    s["log"], s["notes"] = r.take(32), r.take(32)
    s["energy"] = []
    for _ in range(r.u8()):
        role, joules = r.u8(), r.u64()
        cls = LMV.read_class(r)
        s["energy"].append({"role": role, "joules": joules, "class": cls})
    s["attachments"] = [r.take(32) for _ in range(r.u16())]
    sig = r.take(64)
    if r.p != len(b):
        raise Refuse("malformed")
    if "session-rules" not in SKIP and not session_rules(s):
        raise Refuse("audit_session_invalid")
    if "energy" not in SKIP and not all(LMV.energy_ok(e["joules"], e["class"]) for e in s["energy"]):
        raise Refuse("energy_invalid")
    if "session-signature" not in SKIP:
        check_sig(s["player"], SESSION_DOMAIN, b[:-64], sig)
    s["digest"] = sha(b[:-64])
    return s


def session_rules(s):
    limit = (s["end"] - s["start"]) // 1000 if s["end"] > s["start"] else 0
    prev = 0
    for x in s["spans"]:
        if x["length"] < 1 or x["visible"] > 10000 or x["flags"] > 3 or x["start"] < prev or x["start"] + x["length"] > limit:
            return False
        prev = x["start"] + x["length"]
    roles = [e["role"] for e in s["energy"]]
    return (s["presentation"] in (LIVE, ON_DEMAND) and s["start"] < s["end"] and 1 <= s["ended"] <= 5
            and s["origins"] and all(origin_ok(o) for o in s["origins"]) and ascending(s["origins"])
            and s["acks"] and ascending(s["acks"]) and ascending(s["claims"]) and ascending(s["attachments"])
            and (s["startup_ms"] == NO_STARTUP) == (not s["spans"])
            and (s["stalls"] != 0 or s["stall_ms"] == 0)
            and (s["repeat_bytes"] == 0 if s["repeat_objects"] == 0 else s["repeat_bytes"] >= s["repeat_objects"])
            and s["log"] != ZERO and ascending(roles) and all(1 <= x <= 4 for x in roles))


# ---------------------------------------------------------------- §9 the bundle

def exact(v, members):
    return isinstance(v, dict) and set(v) == set(members)


def read_bundle(raw):
    """§9: the bundle, or Stop(audit_bundle_invalid)."""
    def bad():
        return Stop("audit_bundle_invalid", ["bundle"])
    try:
        v = json.loads(raw.decode("utf-8"))
    except (UnicodeDecodeError, ValueError):
        raise bad()
    if not exact(v, ["v", "session", "acks", "claims", "log", "notes", "attachments"]) or v["v"] != 1 or isinstance(v["v"], bool):
        raise bad()
    out = {"session": unhex(v["session"]), "acks": [], "claims": [], "attachments": []}
    if out["session"] is None or not isinstance(v["acks"], list) or not isinstance(v["claims"], list) or not isinstance(v["attachments"], list):
        raise bad()
    for a in v["acks"]:
        if not exact(a, ["record", "items"]) or not isinstance(a["items"], list):
            raise bad()
        rec, items = unhex(a["record"]), [unhex(i) for i in a["items"]]
        if rec is None or any(i is None for i in items):
            raise bad()
        out["acks"].append((rec, items))
    for c in v["claims"]:
        if not exact(c, ["claim", "marker", "creative"]) or not isinstance(c["marker"], str) or not isinstance(c["creative"], list) or not c["creative"]:
            raise bad()
        objects = []
        for o in c["creative"]:
            if not exact(o, ["content_hash", "duration_ms"]):
                raise bad()
            h, d = unhex(o["content_hash"]), o["duration_ms"]
            if h is None or len(h) != 32 or not isinstance(d, int) or isinstance(d, bool) or not 1 <= d <= 0xFFFFFFFF:
                raise bad()
            objects.append({"content_hash": h, "duration_ms": d})
        claim = unhex(c["claim"])
        if claim is None:
            raise bad()
        out["claims"].append({"claim": claim, "marker": c["marker"].encode("utf-8"), "creative": objects})
    if not isinstance(v["log"], str) or not (v["notes"] is None or isinstance(v["notes"], str)):
        raise bad()
    out["log"] = v["log"].encode("utf-8")
    out["notes"] = None if v["notes"] is None else v["notes"].encode("utf-8")
    for a in v["attachments"]:
        if not exact(a, ["name", "record"]) or not isinstance(a["name"], str) or unhex(a["record"]) is None:
            raise bad()
        out["attachments"].append(unhex(a["record"]))
    return out


# ---------------------------------------------------------------- §10 verification

def stopped(fn, at):
    try:
        return fn()
    except Refuse as e:
        raise Stop(e.reason, at)


def read_one_log(b):
    """§10.2 step 6: a logs document of exactly one log: its sid and its metrics."""
    try:
        doc = json.loads(b.decode("utf-8"))
    except (UnicodeDecodeError, ValueError):
        raise Refuse("qoe_invalid")
    if not isinstance(doc, list) or len(doc) != 1:
        raise Refuse("qoe_invalid")
    LMV.read_logs(b)
    return doc[0][0]["sid"], LMV.session_metrics(doc[0])


def read_notes(b, acknowledged):
    """§7.3: a notes document's count of responses, or None for one that does not read."""
    try:
        doc = json.loads(b.decode("utf-8"))
    except (UnicodeDecodeError, ValueError):
        return None
    if not isinstance(doc, list):
        return None
    for n in doc:
        if not exact(n, ["content_hash", "cmsd_static", "cmsd_dynamic"]):
            return None
        h = n["content_hash"]
        if not isinstance(h, str) or not re.fullmatch(r"[0-9a-f]{64}", h):
            return None
        if "notes-acked" not in SKIP and bytes.fromhex(h) not in acknowledged:
            return None
        if any(not (n[k] is None or isinstance(n[k], str)) for k in ("cmsd_static", "cmsd_dynamic")):
            return None
    return len(doc)


def verify_bundle(raw, owner, channels):
    """§10.2: the outcome of a bundle, as §10.3 states it."""
    try:
        return accept_json(verify(raw, owner, channels))
    except Stop as e:
        return {"outcome": "refuse", "reason": e.reason, "at": e.at}


def verify(raw, owner, channels):
    # 1. The bundle.
    b = read_bundle(raw)
    # 2. The record.
    s = stopped(lambda: read_session(b["session"]), ["session"])
    # 3. Its binding.
    if "owner" not in SKIP and s["owner"] != owner:
        raise Stop("audit_wrong_owner", ["session"])
    if "channel" not in SKIP and not any(channel_id(c) == s["channel"] for c in channels):
        raise Stop("audit_unknown_channel", ["session"])
    if "session-bound" not in SKIP and session_of(s["channel"], s["player"]) != s["session"]:
        raise Stop("audit_session_unbound", ["session"])
    # 4. Each acknowledgement.
    upstreams = {upstream_of(o): o for o in s["origins"]}
    by_digest = {}
    for k, (rec, _) in enumerate(b["acks"]):
        by_digest.setdefault(given_digest(rec), k)
    media = 1 if s["presentation"] == LIVE else 2
    acks, delivery, acknowledged = [], {}, set()
    for i, d in enumerate(s["acks"]):
        if d not in by_digest:
            raise Stop("record_missing", ["ack", i])
        rec, lists = b["acks"][by_digest[d]]
        a = stopped(lambda: SV.read_ack(rec), ["ack", i])
        if "ack-bound" not in SKIP and (a["mode"] != 1 or a["downstream"] != s["player"]):
            raise Stop("audit_ack_unbound", ["ack", i])
        if a["upstream"] not in upstreams and "origin" not in SKIP:
            raise Stop("audit_origin_unknown", ["ack", i])
        origin = upstreams.get(a["upstream"], b"?")
        if "ack-window" not in SKIP and (a["start"] < s["start"] or a["end"] > s["end"]):
            raise Stop("ack_outside_period", ["ack", i])
        if "ack-classes" not in SKIP and (a["role"] != 2 or any(e["grant"] != ZERO or e["proof"] != ZERO or e["class"] not in (4, media) for e in a["entries"])):
            raise Stop("audit_ack_invalid", ["ack", i])
        if len(lists) > len(a["entries"]):
            raise Stop("audit_bundle_invalid", ["ack", i])
        for j, e in enumerate(a["entries"]):
            at = ["ack", i, "entry", j]
            if j >= len(lists):
                raise Stop("record_missing", at)
            items = stopped(lambda: SV.read_items(lists[j]), at)
            if not SV.recomputes(e, items, lists[j]):
                raise Stop("items_mismatch", at)
            if "items-start" not in SKIP and any(st != 0 for _, st, _ in items):
                raise Stop("audit_ack_invalid", at)
            key = (origin, e["class"])
            o, n = delivery.get(key, (0, 0))
            delivery[key] = (o + e["objects"], n + e["bytes"])
            acknowledged.update(h for h, _, _ in items)
        acks.append((i, a))
    named, once = set(s["acks"]), set()
    for k, (rec, _) in enumerate(b["acks"]):
        d = given_digest(rec)
        if "bundle-extra-ack" not in SKIP and (d not in named or d in once):
            raise Stop("audit_bundle_invalid", ["ack", k])
        once.add(d)
    # 5. The chains.
    chains = {}
    for i, a in acks:
        chains.setdefault(a["upstream"], []).append((i, a))
    breaks = []
    for chain in chains.values():
        chain.sort(key=lambda x: (x[1]["start"], x[0]))
        for n, (i, a) in enumerate(chain):
            if n == 0:
                if "chain-first" not in SKIP and a["prev"] != ZERO:
                    breaks.append((i, "ack_chain_broken"))
                continue
            p = chain[n - 1][1]
            if "chain-overlap" not in SKIP and p["end"] > a["start"]:
                breaks.append((i, "ack_overlap"))
            elif "chain-prev" not in SKIP and a["prev"] != p["digest"]:
                breaks.append((i, "ack_chain_broken"))
    if breaks:
        i, r = min(breaks)
        raise Stop(r, ["ack", i])
    # 6. The log.
    if "log-digest" not in SKIP and sha(b["log"]) != s["log"]:
        raise Stop("audit_document_mismatch", ["log"])
    sid, m = stopped(lambda: read_one_log(b["log"]), ["log"])
    startup = None if s["startup_ms"] == NO_STARTUP else s["startup_ms"]
    if "qoe" not in SKIP and (sid != s["session"].hex() or m["vst"] != startup or m["stalls"] != s["stalls"] or m["stall_ms"] != s["stall_ms"]):
        raise Stop("audit_qoe_mismatch", ["log"])
    # 7. The notes.
    if s["notes"] == ZERO and b["notes"] is None:
        notes = 0
    elif s["notes"] != ZERO and b["notes"] is not None and (sha(b["notes"]) == s["notes"] or "notes-digest" in SKIP):
        notes = read_notes(b["notes"], acknowledged)
        if notes is None:
            raise Stop("audit_notes_invalid", ["notes"])
    else:
        raise Stop("audit_document_mismatch", ["notes"])
    # 8. The attachments.
    given = [sha(x) for x in b["attachments"]]
    for i, d in enumerate(s["attachments"]):
        if "attach" not in SKIP and d not in given:
            raise Stop("record_missing", ["attachment", i])
    for j, d in enumerate(given):
        if d not in s["attachments"]:
            raise Stop("audit_document_mismatch", ["attachment", j])
    # 9. The claims.
    by_claim = {}
    for k, c in enumerate(b["claims"]):
        by_claim.setdefault(given_digest(c["claim"]), k)
    order = []
    for i, d in enumerate(s["claims"]):
        if d not in by_claim:
            raise Stop("record_missing", ["claim", i])
        order.append(by_claim[d])
    named, once = set(s["claims"]), set()
    for k, c in enumerate(b["claims"]):
        d = given_digest(c["claim"])
        if "bundle-extra" not in SKIP and (d not in named or d in once):
            raise Stop("audit_bundle_invalid", ["claim", k])
        once.add(d)
    seen = set()
    verdicts = [judge(b["claims"][k], s, acknowledged, seen) for k in order]
    return {"session": s, "delivery": delivery, "claims": verdicts, "notes": notes}


def judge(c, s, acknowledged, seen):
    """§10.2 step 9: one claim's verdict and facts."""
    try:
        x = APV.read_claim(c["claim"])
    except APV.Bad as e:
        return {"verdict": str(e)}
    bad = ED.verify(x["player"], APV.CLAIM_DOMAIN + x["body"], x["sig"])
    if bad:
        return {"verdict": bad}
    if "claim-bound" not in SKIP and (x["room"] != s["channel"] or x["session"] != s["session"] or x["player"] != s["player"]):
        return {"verdict": "audit_claim_unbound"}
    if "marker" not in SKIP and x["cue"] != marker_digest(c["marker"]):
        return {"verdict": "audit_marker_mismatch"}
    if "duplicate" not in SKIP and (x["cue"], x["slot"]) in seen:
        return {"verdict": "claim_duplicate"}
    seen.add((x["cue"], x["slot"]))
    d = sum(o["duration_ms"] for o in c["creative"])
    if "asset" not in SKIP and (x["creative_hash"] != APV.creative_digest(c["creative"]) or any(sp["start"] + sp["length"] > d for sp in x["spans"])
                                or x["objects"] != APV.touched(c["creative"], x["spans"])):
        return {"verdict": "claim_asset_mismatch"}
    if "acked" not in SKIP and any(h not in acknowledged for h in x["objects"]):
        return {"verdict": "audit_ad_not_acknowledged"}
    run, viewable = APV.runs(x, {"rule": APV.VIDEO, "require": APV.CLAIM})
    return {"verdict": "presented", "quartile": APV.quartile(x["spans"], d), "run_ms": run, "viewable_ms": viewable, "viewable": run >= 2000}


def presented(s):
    f = [0, 0, 0, 0]
    for x in s["spans"]:
        f[0] += x["length"]
        f[1] += x["length"] if x["visible"] >= 5000 else 0
        f[2] += x["length"] if x["flags"] & AUDIBLE else 0
        f[3] += x["length"] if x["flags"] & AD else 0
    return f


def tag(cls):
    return 0 if cls is None else cls[0]


def accept_json(v):
    s = v["session"]
    p = presented(s)
    return {
        "outcome": "accept", "digest": s["digest"].hex(), "channel": s["channel"].hex(),
        "presentation": "live" if s["presentation"] == LIVE else "on_demand", "start": s["start"], "end": s["end"],
        "ended": ENDED[s["ended"] - 1], "presented_ms": p[0], "viewable_ms": p[1], "audible_ms": p[2], "ad_ms": p[3],
        "delivery": [{"origin": o.decode(), "class": CLASS_NAMES[c], "objects": n, "bytes": by} for (o, c), (n, by) in sorted(v["delivery"].items())],
        "repeats": {"objects": s["repeat_objects"], "bytes": s["repeat_bytes"]},
        "startup_ms": None if s["startup_ms"] == NO_STARTUP else s["startup_ms"], "stalls": s["stalls"], "stall_ms": s["stall_ms"],
        "energy": [{"role": ROLES[e["role"] - 1], "class": LABELS[tag(e["class"])], "joules_micro": e["joules"]} for e in s["energy"]],
        "claims": v["claims"], "notes": v["notes"], "attachments": len(s["attachments"]),
    }


# ---------------------------------------------------------------- §11 the report

def report(name, owner, frm, to, accepted):
    """§11.1, §11.2: the report over the accepted sessions of the channel, the owner and the period."""
    channel = channel_id(name)
    sessions = {}
    for a in accepted:
        s = a["session"]
        if s["channel"] == channel and s["owner"] == owner and frm <= s["start"] < to:
            sessions[s["digest"]] = a
    sessions = [sessions[d] for d in sorted(sessions)]
    if "report-dup" not in SKIP and len({a["session"]["session"] for a in sessions}) != len(sessions):
        raise Stop("audit_session_duplicate", ["report"])
    pres, ends = [0, 0, 0, 0], [0] * 5
    claims = shown = viewable = 0
    quartiles, refused = [0] * 5, {}
    with_span = startup = stalls = stall_ms = 0
    delivery, rep, energy = {}, [0, 0], {}
    unmetered, absent = [0] * 4, [0] * 4
    for a in sessions:
        s = a["session"]
        pres = [x + y for x, y in zip(pres, presented(s))]
        ends[s["ended"] - 1] += 1
        for c in a["claims"]:
            claims += 1
            if c["verdict"] == "presented":
                shown += 1
                if 1 <= c["quartile"] <= 5:
                    quartiles[c["quartile"] - 1] += 1
                viewable += 1 if c["viewable"] else 0
            else:
                refused[c["verdict"]] = refused.get(c["verdict"], 0) + 1
        if s["spans"]:
            with_span += 1
            startup += s["startup_ms"]
        stalls += s["stalls"]
        stall_ms += s["stall_ms"]
        for k, (n, by) in a["delivery"].items():
            o0, b0 = delivery.get(k, (0, 0))
            delivery[k] = (o0 + n, b0 + by)
        rep[0] += s["repeat_objects"]
        rep[1] += s["repeat_bytes"]
        for r in range(1, 5):
            es = [e for e in s["energy"] if e["role"] == r]
            if not es:
                absent[r - 1] += 1
            elif es[0]["class"] is None and "unmetered-apart" not in SKIP:
                unmetered[r - 1] += 1
            else:
                k = (r, tag(es[0]["class"]))
                f, j = energy.get(k, (0, 0))
                energy[k] = (f + 1, j + es[0]["joules"])
    n = len(sessions)
    sup = n < QOE_MIN and "suppress" not in SKIP

    def q(v):
        return None if sup else v
    return {
        "v": 1, "channel": channel.hex(), "owner": owner.hex(), "from": frm, "to": to,
        "sessions": [a["session"]["digest"].hex() for a in sessions],
        "presented": {"presented_ms": pres[0], "viewable_ms": pres[1], "audible_ms": pres[2], "ad_ms": pres[3]},
        "ended": dict(zip(ENDED, ends)),
        "ads": {"claims": claims, "presented": shown, "quartiles": quartiles, "viewable": viewable, "refused": refused},
        "rebuffering": {"sessions": n, "suppressed": sup, "presented": q(with_span), "startup_ms": q(startup), "stalls": q(stalls), "stall_ms": q(stall_ms)},
        "delivery": [{"origin": o.decode(), "class": CLASS_NAMES[c], "objects": k, "bytes": by} for (o, c), (k, by) in sorted(delivery.items())],
        "repeats": {"objects": rep[0], "bytes": rep[1]},
        "energy": [{"role": ROLES[r - 1], "class": LABELS[t], "figures": f, "joules_micro": j} for (r, t), (f, j) in sorted(energy.items())],
        "unmetered": dict(zip(ROLES, unmetered)), "absent": dict(zip(ROLES, absent)),
    }


def report_bytes(v):
    """§11.3: members in bytewise order, no white space, no escape but \\" and \\\\ (the report is ASCII)."""
    return json.dumps(v, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("ascii")


def check_report(served, bundles, name, owner, channels):
    """§11.4."""
    accepted = []
    for k, raw in enumerate(bundles):
        try:
            accepted.append(verify(raw, owner, channels))
        except Stop as e:
            raise Stop(e.reason, ["bundle", k] + e.at)
    if not exact(served, ["report", "sig"]):
        raise Stop("audit_report_mismatch", ["report"])
    r, sig = served["report"], unhex(served["sig"])
    frm, to, names = r.get("from") if isinstance(r, dict) else None, r.get("to") if isinstance(r, dict) else None, r.get("sessions") if isinstance(r, dict) else None
    if sig is None or len(sig) != 64 or not isinstance(frm, int) or not isinstance(to, int) or not isinstance(names, list):
        raise Stop("audit_report_mismatch", ["report"])
    names = [unhex(x) for x in names]
    if any(x is None or len(x) != 32 for x in names):
        raise Stop("audit_report_mismatch", ["report"])
    recomputed = report(name, owner, frm, to, accepted)
    digests = {a["session"]["digest"] for a in accepted}
    for i, d in enumerate(names):
        if "report-missing" not in SKIP and d not in digests:
            raise Stop("record_missing", ["report", "session", i])
    if "report-bytes" not in SKIP and report_bytes(r) != report_bytes(recomputed):
        raise Stop("audit_report_mismatch", ["report"])
    if "report-signature" not in SKIP and ED.verify(owner, REPORT_DOMAIN + report_bytes(r), sig):
        raise Stop("bad_signature", ["report"])


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

def bundle_bytes(v):
    if "bundle_text" in v:
        return v["bundle_text"].encode("utf-8")
    return json.dumps(v["bundle"]).encode("utf-8")


def report_outcome(doc, v, owner, channels):
    bundles = []
    for n in v["sessions"]:
        for s in doc["sessions"]:
            if s["name"] == n:
                bundles.append(bundle_bytes(s))
                break
    try:
        check_report(v["served"], bundles, v["channel"], owner, channels)
        return {"outcome": "accept", "digest": sha(report_bytes(v["served"]["report"])).hex()}
    except Stop as e:
        return {"outcome": "refuse", "reason": e.reason, "at": e.at}


def check(doc, quiet=False):
    """Every vector's outcome: the failures."""
    fails = []
    owner, channels = bytes.fromhex(doc["owner"]), doc["channels"]
    for v in doc["sessions"]:
        got = verify_bundle(bundle_bytes(v), owner, channels)
        if got != v["outcome"]:
            fails.append("session %s: stated %s, got %s" % (v["name"], json.dumps(v["outcome"])[:300], json.dumps(got)[:300]))
    for v in doc["reports"]:
        got = report_outcome(doc, v, owner, channels)
        if got != v["outcome"]:
            fails.append("report %s: stated %s, got %s" % (v["name"], v["outcome"], got))
    if not quiet:
        for f in fails:
            print("FAIL " + f)
    return fails


def reasons_given(doc):
    seen = set()
    for section in ("sessions", "reports"):
        for v in doc[section]:
            if v["outcome"].get("reason"):
                seen.add(v["outcome"]["reason"])
            for c in v["outcome"].get("claims", []):
                seen.add(c["verdict"])
    return seen


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


def mutations(doc):
    """Copies of the corpus, each with one defect a conforming verifier notices."""
    def find(d, section, name):
        return next(v for v in d[section] if v["name"] == name)

    def m_sig(d):
        v = find(d, "sessions", "live_two_origins")
        v["bundle"]["session"] = flip_hex(v["bundle"]["session"], -1)

    def m_figure(d):
        find(d, "sessions", "live_two_origins")["outcome"]["presented_ms"] += 1

    def m_verdict(d):
        find(d, "sessions", "claim_verdicts")["outcome"]["claims"][0]["quartile"] = 5

    def m_drop_ack(d):
        find(d, "sessions", "on_demand_hidden")["bundle"]["acks"].pop()

    def m_item(d):
        v = find(d, "sessions", "live_two_origins")
        v["bundle"]["acks"][0]["items"][0] = flip_hex(v["bundle"]["acks"][0]["items"][0], 5)

    def m_log(d):
        find(d, "sessions", "nothing_presented")["bundle"]["log"] += " "

    def m_marker(d):
        find(d, "sessions", "live_two_origins")["bundle"]["claims"][0]["marker"] += "x"

    def m_reason(d):
        find(d, "sessions", "ack_chain_broken")["outcome"]["reason"] = "ack_overlap"

    def m_place(d):
        find(d, "sessions", "items_not_whole")["outcome"]["at"] = ["ack", 0]

    def m_channels(d):
        d["channels"] = [c for c in d["channels"] if c != "film-library"]

    def m_owner(d):
        d["owner"] = flip_hex(d["owner"], 3)

    def m_report(d):
        find(d, "reports", "news_east")["served"]["report"]["ads"]["presented"] += 1

    def m_report_sessions(d):
        find(d, "reports", "fifty")["sessions"].pop()

    def m_notes(d):
        v = find(d, "sessions", "live_two_origins")
        v["bundle"]["notes"] = v["bundle"]["notes"].replace("rtt=40", "rtt=41")

    def m_creative(d):
        find(d, "sessions", "live_two_origins")["bundle"]["claims"][1]["creative"][0]["duration_ms"] += 1

    return [m_sig, m_figure, m_verdict, m_drop_ack, m_item, m_log, m_marker, m_reason, m_place, m_channels, m_owner,
            m_report, m_report_sessions, m_notes, m_creative]


CHECKS = ["session-rules", "energy", "session-signature", "owner", "channel", "session-bound", "ack-bound", "origin",
          "ack-window", "ack-classes", "items-start", "chain-first", "chain-overlap", "chain-prev", "log-digest", "qoe",
          "notes-digest", "notes-acked", "attach", "bundle-extra", "bundle-extra-ack", "claim-bound", "marker", "duplicate", "asset", "acked",
          "report-dup", "report-missing", "report-bytes", "report-signature", "suppress", "unmetered-apart"]


def main(corpus_dir):
    with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as f:
        doc = json.load(f)
    ok = True
    files = sorted(os.listdir(corpus_dir))
    if files != ["README.md", "vectors.json"]:
        print("FAIL the corpus directory holds %s" % files)
        ok = False
    fails = check(doc)
    print("%s %d sessions and %d reports get the outcomes the corpus states" % ("PASS" if not fails else "FAIL", len(doc["sessions"]), len(doc["reports"])))
    ok = ok and not fails
    seen = reasons_given(doc)
    missing = [r for r in REASONS + REUSED if r not in seen]
    print("%s every reason of §14 and every borrowed code appears%s" % ("PASS" if not missing else "FAIL", "" if not missing else ": missing " + ", ".join(missing)))
    ok = ok and not missing
    survived = []
    for m in mutations(doc):
        d = copy.deepcopy(doc)
        m(d)
        if not check(d, quiet=True):
            survived.append(m.__name__)
    n = len(mutations(doc))
    print("%s %d of %d mutated copies of the corpus fail%s" % ("PASS" if not survived else "FAIL", n - len(survived), n, "" if not survived else ": " + ", ".join(survived)))
    ok = ok and not survived
    uncaught = []
    for c in CHECKS:
        SKIP.add(c)
        try:
            if not check(doc, quiet=True):
                uncaught.append(c)
        finally:
            SKIP.discard(c)
    print("%s the corpus catches each of %d checks dropped%s" % ("PASS" if not uncaught else "FAIL", len(CHECKS), "" if not uncaught else ": not caught " + ", ".join(uncaught)))
    ok = ok and not uncaught
    return 0 if ok else 1


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