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

#!/usr/bin/env python3
"""Independent verifier of the sink-evaluated predicate corpus (extensions/sink-predicate.md).

Written from the extension's text. It shares no code with wai-rs, which generates the corpus, and
never runs it. It takes from the second implementations of the texts this extension builds on what
they define: documents, their canonical form, digest and tokens from route-record's
(`route_record_verify.py`, route-record §2); region codes, their country and *holds* from rung
price's (`rung_price_verify.py`, rung-price §2); a blackout's form, a location aggregated against
blackouts and what follows from them, and §5.5's territory, from the audience decision corpus's
(`audience_decision_verify.py`, audience-decision §4.2, §4.3, §4.6 and §5.5); and Ed25519
verification as SPEC §7 *Signatures* states it from the routine every Python verifier of WAI's
corpora shares (`../claim-conformance/_ed25519.py`). Everything this extension adds is written here:

- §3: a predicate's members and rules, the members by which it would send something away from the
  device, the state classes it reads, its digest and its signature;
- §4: a sink's state, the programme's verdict, a candidate's regions and caps, a slot's fill, and
  the evaluation;
- §5: a sink's log of a slot, checked.

For every case it requires the outcome the corpus states, member for member; that every reason of
§6's table appears in the corpus and that the corpus states no other; that no evaluation depends on
the order of the sink's exposures, on exposures of a counter no cap names, or on the order of a
document's members; that a candidate is shown exactly where audience-decision §5.5 puts the location
within its regions; and that each slot's fill, given as a log with the state the sink held before
that slot, conforms. It checks what no vector carries: the bound of a text. 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 sink_predicate_verify.py <live-conformance/sink-predicate>

Exit status 0 only if every case verifies, every mutated copy fails and every weakened verifier is
caught. Python 3.8 or later; the `cryptography` package, which the audience decision verifier's
readers import.
"""

import copy
import hashlib
import json
import os
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
for d in ("claim-conformance", "jwp-conformance", "signer-conformance"):
    sys.path.insert(0, os.path.join(HERE, "..", d))
import _ed25519 as ED  # noqa: E402
import route_record_verify as RR  # noqa: E402
import rung_price_verify as RP  # noqa: E402
import audience_decision_verify as AD  # noqa: E402

EXTENSION = os.path.join(HERE, "..", "extensions", "sink-predicate.md")
KIND = "wai-sink-predicate"                         # §3.1
DOMAIN = b"wai:sink-predicate\x01"                  # §3.4
MAX_TEXT = 1 << 20                                  # §3.5: 1,048,576 bytes
MAX_BLACKOUTS = 256                                 # §3.1
MAX_SLOTS = 64                                      # §3.1
MAX_CANDIDATES = 64                                 # §3.2
MAX_CAPS = 8                                        # §3.3
MAX_REGIONS = 8192                                  # §3.3
MAX_AVAIL = 3600000                                 # §3.2
MAX_LIMIT = 1000                                    # §3.3
PREDICATE = ["kind", "version", "source", "programme", "start", "end", "state", "blackouts", "slots", "sig"]  # §3.1
SLOT = ["id", "at", "avail_ms", "candidates", "filler"]                                       # §3.2
CANDIDATE = ["ad", "creative", "duration_ms", "show", "regions", "caps"]                      # §3.3
CAP = ["counter", "limit", "window"]                                                          # §3.3
SEND = ["report", "beacon", "callback", "uri"]                                                # §3.5
ACROSS = ["household", "account"]                                                             # §3.5
SHOWS = ["everywhere", "only", "except"]
EXPOSURES, LOCATION = "exposures", "location"                                                 # §4.1

MALFORMED = "predicate_malformed"
NEEDS_REPORT = "predicate_needs_report"
SIGNATURE = "predicate_signature_invalid"
SOURCE_UNKNOWN = "predicate_source_unknown"
NOT_TARGETED = "predicate_not_targeted"
CAP_BREACHED = "predicate_cap_breached"
FILL_MISMATCH = "predicate_fill_mismatch"

# A weakened run drops one of these checks (see `WEAKENINGS`); None checks everything.
DROPPED = None


def drops(name):
    return DROPPED == name


def reasons_table():
    """§6's codes, in order."""
    text = open(EXTENSION, encoding="utf-8").read()
    at = text.index("\n## 6. Reasons")
    end = text.find("\n## ", at + 1)
    return [r.split("`")[1] for r in text[at:end].splitlines() if r.startswith("| `")]


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


def need(ok, reason, check=None):
    """Refuse with `reason` unless `ok`, or unless `check` is the check a weakened run drops."""
    if not ok and not (check is not None and drops(check)):
        raise Refused(reason)


# ------------------------------------------------------------------ §3

def names_a_report(v, depth=0):
    """§3.5: a member, at any depth, by which an evaluation would send something away."""
    names = SEND if drops("report_across") else SEND + ACROSS
    if isinstance(v, dict):
        if any(k in names for k in v) and (depth == 0 or not drops("report_nested")):
            return True
        return any(names_a_report(x, depth + 1) for x in v.values())
    if isinstance(v, list):
        return any(names_a_report(x, depth + 1) for x in v)
    return False


def reads(blackouts, slots):
    """§3.1: the state classes a predicate of these blackouts and slots reads."""
    cands = [c for s in slots for c in s["candidates"]]
    out = []
    if any(c["caps"] for c in cands):
        out.append(EXPOSURES)
    if blackouts or any(c["show"] != "everywhere" for c in cands):
        out.append(LOCATION)
    return out


def cap_ok(k):
    if not (RR.exactly(k, CAP) and RR.is_token(k["counter"]) and RR.is_int(k["limit"]) and RR.is_int(k["window"])):
        return False
    if not 1 <= k["limit"] <= MAX_LIMIT and not drops("cap_limit"):
        return False
    return k["window"] >= 1 or drops("cap_window")


def candidate_ok(c, avail):
    if not (RR.exactly(c, CANDIDATE) and RR.is_token(c["ad"]) and RR.is_hex(c["creative"], 64)
            and RR.is_int(c["duration_ms"]) and c["duration_ms"] >= 1):
        return False
    if c["duration_ms"] > avail and not drops("duration_in_avail"):
        return False
    if c["show"] not in SHOWS or not AD.regions_ok(c["regions"], 0, MAX_REGIONS):
        return False
    if (len(c["regions"]) == 0) != (c["show"] == "everywhere") and not drops("everywhere_empty"):
        return False
    caps = c["caps"]
    if not (isinstance(caps, list) and len(caps) <= MAX_CAPS and all(cap_ok(k) for k in caps)):
        return False
    return AD.ascending([k["counter"] for k in caps]) or drops("caps_ascending")


def slot_ok(s, start, end):
    if not (RR.exactly(s, SLOT) and RR.is_token(s["id"]) and RR.is_int(s["at"]) and RR.is_int(s["avail_ms"])):
        return False
    if not start <= s["at"] < end and not drops("at_in_span"):
        return False
    if not 1 <= s["avail_ms"] <= MAX_AVAIL:
        return False
    cs = s["candidates"]
    if not (isinstance(cs, list) and 1 <= len(cs) <= MAX_CANDIDATES and all(candidate_ok(c, s["avail_ms"]) for c in cs)):
        return False
    if len({c["ad"] for c in cs}) != len(cs) and not drops("ads_distinct"):
        return False
    return AD.null_or_token(s["filler"])


def predicate_form(v):
    """Rules 1 to 4."""
    if not RR.exactly(v, PREDICATE) or v["kind"] != KIND or not RR.same(v["version"], 1):
        return False
    if not (RR.is_hex(v["source"], 64) and RR.is_token(v["programme"]) and RR.is_int(v["start"])
            and RR.is_int(v["end"]) and RR.is_hex(v["sig"], 128)):
        return False
    start, end = v["start"], v["end"]
    if start >= end and not drops("start_below_end"):
        return False
    st = v["state"]
    if not (isinstance(st, list) and all(x in (EXPOSURES, LOCATION) for x in st) and AD.ascending(st)):
        return False
    bs = v["blackouts"]
    if not (isinstance(bs, list) and len(bs) <= MAX_BLACKOUTS and all(AD.blackout_ok(b) for b in bs)):
        return False
    if not AD.ascending([b["id"] for b in bs]) and not drops("blackouts_ascending"):
        return False
    ss = v["slots"]
    if not (isinstance(ss, list) and len(ss) <= MAX_SLOTS and all(slot_ok(s, start, end) for s in ss)):
        return False
    if not all(ss[i]["at"] < ss[i + 1]["at"] for i in range(len(ss) - 1)) and not drops("slots_ascending"):
        return False
    if len({s["id"] for s in ss}) != len(ss) and not drops("slot_ids_distinct"):
        return False
    return st == reads(bs, ss) or drops("state_exact")


_VERIFIED = {}


def signed(v):
    """§3.4: an Ed25519 signature by `source` over the domain and the canonical form of `v` without
    `sig`, verified strictly."""
    if drops("signature"):
        return True
    body = {k: x for k, x in v.items() if k != "sig"}
    form = json.dumps(body, separators=(",", ":"), ensure_ascii=False) if drops("signature_as_written") else RR.canonical(body)
    domain = b"" if drops("signature_domain") else AD.DECISION_DOMAIN if drops("signature_other_domain") else DOMAIN
    msg = domain + form.encode("utf-8")
    k = (v["source"], hashlib.sha256(msg).digest(), v["sig"])
    if k not in _VERIFIED:
        try:
            _VERIFIED[k] = ED.verify(bytes.fromhex(v["source"]), msg, bytes.fromhex(v["sig"])) is None
        except ValueError:
            _VERIFIED[k] = False
    return _VERIFIED[k]


def read_predicate(text):
    """§3.5: the predicate's value, or Refused."""
    v = AD.read_text(text)
    need(v is not None, MALFORMED)
    need(not names_a_report(v), NEEDS_REPORT, "report")
    need(predicate_form(v), MALFORMED)
    need(signed(v), SIGNATURE)
    return v


def digest(v):
    return RR.digest({k: x for k, x in v.items() if k != "sig"})


def predicate_outcome(text):
    try:
        v = read_predicate(text)
    except Refused as r:
        return {"outcome": "refuse", "reason": r.reason}
    return {"outcome": "read", "digest": digest(v), "source": v["source"], "programme": v["programme"], "state": v["state"]}


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

def shown(c, loc):
    """§4.3."""
    if c["show"] == "everywhere":
        return True
    if loc is None:
        return drops("no_location_shown")
    held = RP.holds(c["regions"], loc)
    if c["show"] == "only":
        return held
    country_of_listed = len(loc) == 2 and any(len(r) > 2 and RP.country(r) == loc for r in c["regions"])
    return (not held or drops("except_held")) and (not country_of_listed or drops("undetermined"))


def reached(k, exposures, t):
    """§4.4."""
    if drops("cap"):
        return False
    n = 0
    for e in exposures:
        if e["counter"] != k["counter"]:
            continue
        if e["at"] > t and not drops("window_lower"):
            continue
        if not (t < e["at"] + k["window"] or (drops("window_upper") and t == e["at"] + k["window"])):
            continue
        n += 1
    return n >= k["limit"]


def admits(c, exposures, t):
    """§4.4: with no cap always; otherwise only when the exposures are kept and no cap is reached."""
    if not c["caps"]:
        return True
    if exposures is None:
        if not drops("not_kept"):
            return False
        exposures = []
    return not any(reached(k, exposures, t) for k in c["caps"])


def expose(c, exposures, t):
    if exposures is not None and not drops("expose"):
        exposures.extend({"counter": k["counter"], "at": t} for k in c["caps"])


def fill(s, loc, exposures):
    """§4.5: (ads, rest) of the slot, the exposures it adds appended to `exposures`."""
    left, ads = s["avail_ms"], []
    for c in s["candidates"]:
        if shown(c, loc) and admits(c, exposures, s["at"]) and (c["duration_ms"] <= left or drops("avail")):
            left -= c["duration_ms"]
            ads.append(c["ad"])
            expose(c, exposures, s["at"])
    return ads, left


def programme(v, loc):
    """§4.2: audience-decision §4.6 at the location's aggregation (§4.3 there)."""
    e = AD.aggregate(v["blackouts"], loc)
    return AD.follows(v["blackouts"], e["region"], e["resolved"])


def held_exposures(state):
    x = state["exposures"]
    return None if x is None else [dict(e) for e in x]


def read_for(text, state):
    """§4.6 steps 1 and 2."""
    v = read_predicate(text)
    need(v["source"] in state["sources"], SOURCE_UNKNOWN, "source")
    return v


def evaluate(text, state):
    """§4.6: the evaluation, or Refused."""
    v = read_for(text, state)
    loc = state["location"]
    applied, verdict, alternate = programme(v, loc)
    fills = []
    if verdict == "show" or drops("verdict_gates"):
        x = held_exposures(state)
        for s in v["slots"]:
            ads, rest = fill(s, loc, held_exposures(state) if drops("across_slots") else x)
            fills.append({"slot": s["id"], "ads": ads, "rest_ms": rest})
    return {"outcome": "evaluated", "verdict": verdict, "applied": applied, "alternate": alternate, "fills": fills}


def evaluation_outcome(case):
    try:
        return evaluate(case["text"], case["state"])
    except Refused as r:
        return {"outcome": "refuse", "reason": r.reason}


# ------------------------------------------------------------------ §5

def check_log(text, state, slot_id, log):
    v = read_for(text, state)
    loc = state["location"]
    verdict = programme(v, loc)[1]
    s = next((x for x in v["slots"] if x["id"] == slot_id), None)
    need(s is not None, FILL_MISMATCH)
    want = fill(s, loc, held_exposures(state))[0] if verdict == "show" else []
    x = held_exposures(state)
    seen = set()
    for a in log:
        c = next((k for k in s["candidates"] if k["ad"] == a), None)
        need(c is not None, FILL_MISMATCH)
        need(a not in seen, FILL_MISMATCH, "log_once")
        seen.add(a)
        need(shown(c, loc), NOT_TARGETED, "log_target")
        need(admits(c, x, s["at"]), CAP_BREACHED, "log_cap")
        expose(c, x, s["at"])
    need(list(log) == want, FILL_MISMATCH, "log_fill")
    return {"outcome": "conforms"}


def log_outcome(case):
    try:
        return check_log(case["text"], case["state"], case["slot"], case["log"])
    except Refused as r:
        return {"outcome": "refuse", "reason": r.reason}


RUNS = (("predicates", lambda c: predicate_outcome(c["text"])), ("evaluations", evaluation_outcome),
        ("logs", log_outcome))
KINDS = tuple(k for k, _ in RUNS)


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

def reordered(text):
    """`text`'s document with its members in reverse order, or None when it is not a document object."""
    v = AD.read_text(text)
    if not isinstance(v, dict):
        return None
    return json.dumps(dict(reversed(list(v.items()))), indent=1, ensure_ascii=False)


def counters(text):
    v = AD.read_text(text) or {}
    return {k["counter"] for s in v.get("slots") or [] for c in s["candidates"] for k in c["caps"]}


def invariances(doc, fail):
    for c in doc["evaluations"]:
        st, out = c["state"], c["outcome"]
        # The order of the exposures, and exposures of a counter no cap names.
        if st["exposures"] is not None:
            unnamed = "zz/unnamed"
            assert unnamed not in counters(c["text"])
            for x in (list(reversed(st["exposures"])), st["exposures"] + [{"counter": unnamed, "at": 0}]):
                if not RR.same(evaluation_outcome(dict(c, state=dict(st, exposures=x))), out):
                    fail(f"evaluations/{c['name']}: other exposures of no consequence give another evaluation")
        # The order of a document's members.
        t = reordered(c["text"])
        if t is not None and not RR.same(evaluation_outcome(dict(c, text=t)), out):
            fail(f"evaluations/{c['name']}: its members in another order give another evaluation")
        if out["outcome"] != "evaluated":
            continue
        v = read_predicate(c["text"])
        # §4.3 is audience-decision §5.5's within.
        for s in v["slots"]:
            for k in s["candidates"]:
                if shown(k, st["location"]) != (AD.territory_word(k, st["location"]) == "within"):
                    fail(f"evaluations/{c['name']}: {k['ad']} is shown otherwise than §5.5 words it")
        # Each slot's fill, as a log with the state the sink held before the slot, conforms.
        before = held_exposures(st)
        for s, f in zip(v["slots"], out["fills"]):
            got = log_outcome({"text": c["text"], "state": dict(st, exposures=copy.deepcopy(before)), "slot": s["id"],
                               "log": f["ads"]})
            if got != {"outcome": "conforms"}:
                fail(f"evaluations/{c['name']}: the fill of {s['id']} as a log gives {got}")
            fill(s, st["location"], before)
    for c in doc["predicates"]:
        t = reordered(c["text"])
        if t is not None and len(t.encode()) <= MAX_TEXT and not RR.same(predicate_outcome(t), c["outcome"]):
            fail(f"predicates/{c['name']}: its members in another order give another outcome")


def strings(v, name, out):
    if isinstance(v, dict):
        for k, x in v.items():
            if k == name and isinstance(x, str):
                out.append(x)
            strings(x, name, out)
    elif isinstance(v, list):
        for x in v:
            strings(x, name, out)


def check(doc, quiet=False, full=True):
    fails = []

    def fail(msg):
        fails.append(msg)
        if not quiet:
            print("FAIL", msg)

    for kind, run in RUNS:
        cases = doc.get(kind) or []
        if not cases:
            fail(f"no {kind} cases")
        for c in cases:
            try:
                got = run(c)
            except (KeyError, TypeError, ValueError, AttributeError) as e:
                fail(f"{kind}/{c.get('name')}: {e!r}")
                continue
            if not RR.same(got, c["outcome"]):
                fail(f"{kind}/{c['name']}: stated {json.dumps(c['outcome'])} got {json.dumps(got)}")
    table = reasons_table()
    reasons = []
    strings([c["outcome"] for k in KINDS for c in doc.get(k) or []], "reason", reasons)
    for r in table:
        if r not in reasons:
            fail(f"no case gives {r}")
    for r in sorted(set(reasons) - set(table)):
        fail(f"a case gives {r}, which §6 does not list")
    if full:
        try:
            invariances(doc, fail)
        except (KeyError, TypeError, ValueError, AttributeError, AssertionError, Refused) as e:
            fail(f"invariances: {e!r}")
    return fails


WEAKENINGS = (
    "report", "report_nested", "report_across", "start_below_end", "state_exact", "blackouts_ascending",
    "slots_ascending", "slot_ids_distinct", "at_in_span", "ads_distinct", "duration_in_avail", "everywhere_empty",
    "caps_ascending", "cap_limit", "cap_window", "signature", "signature_domain", "signature_other_domain",
    "signature_as_written", "source",
    "no_location_shown", "undetermined", "except_held", "cap", "window_lower", "window_upper", "not_kept", "expose",
    "avail", "across_slots", "verdict_gates", "log_once", "log_target", "log_cap", "log_fill",
)


def mutations(doc):
    """Copies of the corpus, one defect each, every one of which must fail."""
    def case(d, kind, name):
        return next(c for c in d[kind] if c["name"] == name)

    def m_fill(d):
        case(d, "evaluations", "show_and_fill")["outcome"]["fills"][0]["ads"] = ["ad/a", "ad/b"]

    def m_rest(d):
        case(d, "evaluations", "targets_in_a_listed_country")["outcome"]["fills"][0]["rest_ms"] = 0

    def m_verdict(d):
        case(d, "evaluations", "withhold_fills_no_slot")["outcome"]["verdict"] = "show"

    def m_location(d):
        case(d, "evaluations", "only_holds")["state"]["location"] = "XB-2"

    def m_window(d):
        case(d, "evaluations", "an_exposure_a_whole_window_before")["state"]["exposures"][0]["at"] += 1

    def m_kept(d):
        case(d, "evaluations", "exposures_not_kept")["state"]["exposures"] = []

    def m_source(d):
        c = case(d, "evaluations", "show_and_fill")
        c["state"]["sources"] = [d["keys"]["other_source"]]

    def m_log(d):
        case(d, "logs", "the_fill")["log"] = ["ad/a", "ad/b"]

    def m_log_reason(d):
        case(d, "logs", "two_of_one_counter")["outcome"]["reason"] = FILL_MISMATCH

    def m_log_state(d):
        case(d, "logs", "cap_reached_by_the_sinks_exposures")["state"]["exposures"] = []

    def m_report_reason(d):
        case(d, "predicates", "household_in_a_cap")["outcome"]["reason"] = MALFORMED

    def m_digest(d):
        case(d, "predicates", "targeted_and_capped")["outcome"]["digest"] = "00" * 32

    def m_state(d):
        case(d, "predicates", "location_alone")["outcome"]["state"] = []

    def m_sig(d):
        c = case(d, "predicates", "no_state")
        v = json.loads(c["text"])
        v["sig"] = ("01" if v["sig"][:2] != "01" else "02") + v["sig"][2:]
        c["text"] = RR.canonical(v)

    def m_reason_gone(d):
        d["logs"] = [c for c in d["logs"] if c["outcome"].get("reason") != NOT_TARGETED]

    def m_reason_unknown(d):
        case(d, "logs", "passes_one_over")["outcome"]["reason"] = "predicate_passed_over"

    def m_no_kind(d):
        d["logs"] = []

    return [m_fill, m_rest, m_verdict, m_location, m_window, m_kept, m_source, m_log, m_log_reason, m_log_state,
            m_report_reason, m_digest, m_state, m_sig, m_reason_gone, m_reason_unknown, m_no_kind]


def direct(doc):
    """§3.5's bound of 1,048,576 bytes: a text of that many bytes, whitespace added, reads as itself,
    and one of a byte more is refused."""
    fails = []
    c = next(x for x in doc["predicates"] if x["name"] == "targeted_and_capped")
    pad = MAX_TEXT - len(c["text"].encode("utf-8"))
    if not RR.same(predicate_outcome(c["text"] + " " * pad), c["outcome"]):
        fails.append("targeted_and_capped of 1,048,576 bytes, whitespace added, did not read as itself")
    if predicate_outcome(c["text"] + " " * (pad + 1)) != {"outcome": "refuse", "reason": MALFORMED}:
        fails.append(f"targeted_and_capped of 1,048,577 bytes was not refused {MALFORMED}")
    return fails


def main(argv):
    global DROPPED
    if len(argv) != 2:
        print(__doc__)
        return 2
    doc = json.load(open(os.path.join(argv[1], "vectors.json"), encoding="utf-8"))
    extra = sorted(set(os.listdir(argv[1])) - {"vectors.json", "README.md"})
    fails = check(doc)
    if extra:
        fails.append(f"files the corpus does not hold: {extra}")
    print(f"{sum(len(doc[k]) for k in KINDS)} cases ({', '.join(f'{len(doc[k])} {k}' for k in KINDS)}): {len(fails)} failure(s)")
    b = direct(doc)
    print(f"what no vector carries: {len(b)} failure(s)")
    fails += b
    caught = 0
    ms = mutations(doc)
    for m in ms:
        d = copy.deepcopy(doc)
        m(d)
        if check(d, quiet=True):
            caught += 1
        else:
            fails.append(f"mutation {m.__name__} was not caught")
    print(f"{caught} of {len(ms)} mutated copies fail")
    weak = 0
    for w in WEAKENINGS:
        DROPPED = w
        try:
            if check(doc, quiet=True, full=False):
                weak += 1
            else:
                fails.append(f"a verifier without its {w} check passes the corpus")
        finally:
            DROPPED = None
    print(f"{weak} of {len(WEAKENINGS)} weakened verifiers are caught")
    for f in fails:
        print("FAIL", f)
    return 1 if fails else 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))