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

#!/usr/bin/env python3
"""Second implementation of extensions/robustness-evidence.md (`wai.media.robustness`), over the
corpus `live-conformance/robustness/`.

Written from the extension's text. It shares no code with wai-rs: it reads every record as §2
has it read (strict JSON, the encodings of jwp-receipts §2.4.3, rates read last as exact
fractions), applies the transforms of §4, runs the harness of §6 with `wai-mark:tent64` as the
synthesis corpus's second implementation has it (_wai_mark.py, written from synthesis §7.5),
and rerun end to end on the samples the corpus holds, requires the conformance record's table
bit for bit. It verifies every signature with WAI's strict Ed25519 rule
(claim-conformance/_ed25519.py) and every link hash with BLAKE3 written from its specification
(claim-conformance/_blake3.py). It reads §12's codes from the extension, checks the records of
`wai-mark:tent64` against the synthesis corpus's battery.json — a measurement another program
wrote — and the claim against synthesis §7.5.8's text.

    python3 -I robustness_verify.py <live-conformance/robustness>

It gives every case the outcome, the reason, the record and the statements the corpus states,
and checks that every reason of §12 refuses a case. Then it mutates copies of the corpus and
requires every copy to fail, and drops each of its own checks in turn and requires the corpus
to catch the weakened verifier. Exit status 0 only if all of that holds. Python 3.8 or later;
standard library only.
"""

import copy
import hashlib
import json
import math
import os
import re
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 _wai_mark as M  # noqa: E402  (synthesis §7.5, written from the extension's text)
import _wai_audio_mark as AM  # noqa: E402  (synthesis §7.10, written from the extension's text)
import _ed25519 as ED  # noqa: E402
from _blake3 import blake3  # noqa: E402

WAI = os.path.join(HERE, "..")
EXTENSION = os.path.join(WAI, "extensions", "robustness-evidence.md")
SYNTHESIS = os.path.join(WAI, "extensions", "synthesis.md")

SIGN_DOMAIN = b"wai:robustness\x01"           # §2
LINK_DOMAIN = b"wai:robustness-link\x01"      # §2
RECORD, CLAIM, CORPUS = "wai-robustness", "wai-robustness-claim", "wai-robustness-corpus"
HARNESS = "wai-robustness-harness:1"          # §6
MAX_INT = (1 << 53) - 1
FRAME_CAP, ITEM_CAP = 1 << 26, 1 << 30        # §9.1
RECORD_MEMBERS = ["kind", "algorithm", "algorithm_digest", "binding_type", "media", "corpus", "values", "battery",
                  "harness", "table", "issued_at", "signer_pubkey", "sig"]
ROW_MEMBERS = ["attack", "trials", "survived", "wrong", "negatives", "false_positives", "survival",
               "false_positive_rate", "balanced_accuracy", "tpr_at_1pct_fpr"]
RATES = ["survival", "false_positive_rate", "balanced_accuracy", "tpr_at_1pct_fpr"]
CLAIM_MEMBERS = ["kind", "algorithm", "algorithm_digest", "survives", "evidence", "issued_at", "signer_pubkey", "sig"]
WHITE_SPACE = set("\t\n\x0b\x0c\r \x85\xa0           "
                  " 

  　")

# A check named here is left out: the verifier weakened on purpose, to show the corpus catches it.
SKIP = set()


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


class Bad(Exception):
    """A record that does not read: `record_malformed`."""


class Number:
    """A JSON number that is not an unsigned integer of §2: kept, so that a rate member holding
    one is `rate_not_exact` and any other member holding one is `record_malformed`."""


def need(c, reason=None):
    if not c:
        raise Refuse(reason) if reason else Bad()


def sha(b):
    return hashlib.sha256(b).digest()


# ---------------------------------------------------------------- §2 reading

def strict_json(text):
    """RFC 8259, no member named twice, no boolean, no unpaired surrogate, at most 32 levels;
    a number that is not an unsigned integer of at most 2^53 − 1 is kept as a Number."""
    def pairs(ps):
        d = {}
        for k, x in ps:
            if k in d:
                raise Bad()
            d[k] = x
        return d

    def no(s):
        raise Bad()

    def integer(s):
        if s.startswith("-"):
            return Number()
        n = int(s)
        return n if n <= MAX_INT else Number()

    try:
        v = json.loads(text, object_pairs_hook=pairs, parse_float=lambda s: Number(), parse_int=integer, parse_constant=no)
    except (ValueError, RecursionError):
        raise Bad()

    def walk(x, depth):
        if isinstance(x, (dict, list)):
            need(depth <= 32)
            for y in (x.values() if isinstance(x, dict) else x):
                walk(y, depth + 1)
        elif isinstance(x, str):
            need(not any(0xD800 <= ord(c) <= 0xDFFF for c in x))
        elif isinstance(x, bool):
            raise Bad()
    walk(v, 1)
    return v


def subset(x):
    """§2's values: strings, unsigned integers, null, and arrays and objects of them."""
    if x is None or isinstance(x, str):
        return True
    if isinstance(x, bool) or isinstance(x, Number):
        return False
    if isinstance(x, int):
        return 0 <= x <= MAX_INT
    if isinstance(x, list):
        return all(subset(y) for y in x)
    if isinstance(x, dict):
        return all(subset(y) for y in x.values())
    return False


def jstr(s):
    out = ['"']
    for c in s:
        o = ord(c)
        if c == '"':
            out.append('\\"')
        elif c == "\\":
            out.append("\\\\")
        elif c == "\b":
            out.append("\\b")
        elif c == "\f":
            out.append("\\f")
        elif c == "\n":
            out.append("\\n")
        elif c == "\r":
            out.append("\\r")
        elif c == "\t":
            out.append("\\t")
        elif o < 0x20:
            out.append("\\u%04x" % o)
        else:
            out.append(c)
    out.append('"')
    return "".join(out)


def jcs(v):
    """RFC 8785 over strings, unsigned integers, null, arrays and objects."""
    if v is None:
        return "null"
    if isinstance(v, bool) or isinstance(v, Number):
        raise Bad()
    if isinstance(v, int):
        return str(v)
    if isinstance(v, str):
        return jstr(v)
    if isinstance(v, list):
        return "[" + ",".join(jcs(x) for x in v) + "]"
    keys = sorted(v, key=lambda k: k.encode("utf-16-be"))
    return "{" + ",".join(jstr(k) + ":" + jcs(v[k]) for k in keys) + "}"


def link_hash(v):
    return blake3(LINK_DOMAIN + jcs(v).encode("utf-8"))


def signing_bytes(v):
    return SIGN_DOMAIN + jcs({k: x for k, x in v.items() if k != "sig"}).encode("utf-8")


def is_uint(x):
    return isinstance(x, int) and not isinstance(x, bool) and 0 <= x <= MAX_INT


def is_hex(x, n=None):
    return (isinstance(x, str) and len(x) % 2 == 0 and all(c in "0123456789abcdef" for c in x)
            and (n is None or len(x) == 2 * n))


def is_term(s):
    return (isinstance(s, str) and 1 <= len(s.encode("utf-8")) <= 2048
            and not any(c in WHITE_SPACE or ord(c) < 0x20 or 0x7F <= ord(c) <= 0x9F for c in s))


def date_time(s):
    """RFC 3339 date-time, the day within its month (jwp-receipts §2.4.3)."""
    if not (isinstance(s, str) and len(s) >= 20 and s.isascii()):
        return False
    if not all(s[i].isdigit() for i in (0, 1, 2, 3, 5, 6, 8, 9, 11, 12, 14, 15, 17, 18)):
        return False
    if not (s[4] == "-" and s[7] == "-" and s[10] in "Tt" and s[13] == ":" and s[16] == ":"):
        return False
    yr, mo, da, hh, mi, ss = int(s[0:4]), int(s[5:7]), int(s[8:10]), int(s[11:13]), int(s[14:16]), int(s[17:19])
    leap = yr % 4 == 0 and (yr % 100 != 0 or yr % 400 == 0)
    days = [31, 29 if leap else 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
    if not (1 <= mo <= 12 and 1 <= da <= days[mo - 1] and hh <= 23 and mi <= 59 and ss <= 60):
        return False
    i = 19
    if i < len(s) and s[i] == ".":
        j = i + 1
        while j < len(s) and s[j].isdigit():
            j += 1
        if j == i + 1:
            return False
        i = j
    rest = s[i:]
    if rest in ("Z", "z"):
        return True
    return (len(rest) == 6 and rest[0] in "+-" and rest[1:3].isdigit() and rest[3] == ":" and rest[4:6].isdigit()
            and int(rest[1:3]) <= 23 and int(rest[4:6]) <= 59)


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


def read_frame(text, kind, names):
    v = strict_json(text)
    exact(v, names)
    need(v["kind"] == kind)
    need(is_hex(v["signer_pubkey"], 32) and is_hex(v["sig"], 64))
    need(date_time(v["issued_at"]))
    return v


def check_signature(v):
    if "signature" in SKIP:
        return
    r = ED.verify(bytes.fromhex(v["signer_pubkey"]), signing_bytes(v), bytes.fromhex(v["sig"]))
    if r:
        raise Refuse(r)


def rate(x):
    """§7: [n, d], 1 <= d, n <= d, in lowest terms."""
    ok = (isinstance(x, list) and len(x) == 2 and all(is_uint(y) for y in x) and x[1] >= 1 and x[0] <= x[1]
          and math.gcd(x[0], x[1]) == 1)
    if not ok and "exact-rates" not in SKIP:
        raise Refuse("rate_not_exact")
    return tuple(x) if isinstance(x, list) else x


def read_evidence(text):
    """§2 and §5: the record, every member but the rates, then the rates, then the signature."""
    v = read_frame(text, RECORD, RECORD_MEMBERS)
    rest = copy.deepcopy(v)
    if isinstance(rest["table"], list):
        for r in rest["table"]:
            if isinstance(r, dict):
                for k in RATES:
                    r.pop(k, None)
    need(subset(rest))
    need(v["binding_type"] in ("watermark", "fingerprint", "detector"))   # §5; detection §4
    need(isinstance(v["values"], list) and all(is_hex(x) and x != "" for x in v["values"]))
    need(isinstance(v["battery"], list))
    for a in v["battery"]:
        exact(a, ["attack", "transform", "params", "seed"])
        need(is_term(a["attack"]) and is_term(a["transform"]) and isinstance(a["params"], list)
             and all(is_uint(p) for p in a["params"]) and is_uint(a["seed"]))
    need(isinstance(v["table"], list))
    for r in v["table"]:
        exact(r, ROW_MEMBERS)
        need(is_term(r["attack"]) and all(is_uint(r[k]) for k in ROW_MEMBERS[1:6]))
    need(is_term(v["algorithm"]) and is_hex(v["algorithm_digest"], 32) and v["media"] in ("image", "video", "audio")
         and is_hex(v["corpus"], 32) and is_term(v["harness"]))
    rates = []
    for r in v["table"]:
        rates.append(tuple(None if (k == "tpr_at_1pct_fpr" and r[k] is None) else rate(r[k]) for k in RATES))
    check_signature(v)
    return v, rates


def read_claim(text):
    v = read_frame(text, CLAIM, CLAIM_MEMBERS)
    need(subset(v))
    need(isinstance(v["survives"], list) and len(v["survives"]) >= 1)
    for c in v["survives"]:
        exact(c, ["media", "transform", "params"])
        need(c["media"] in ("image", "video", "audio") and is_term(c["transform"]) and isinstance(c["params"], list)
             and all(is_uint(p) for p in c["params"]))
    need(len({jcs(c) for c in v["survives"]}) == len(v["survives"]))
    need(isinstance(v["evidence"], list) and all(is_hex(e, 32) for e in v["evidence"]))
    need(len(set(v["evidence"])) == len(v["evidence"]))
    need(is_term(v["algorithm"]) and is_hex(v["algorithm_digest"], 32))
    check_signature(v)
    return v


# ---------------------------------------------------------------- §3 the corpus

def read_corpus(text):
    """§3: the document, or `corpus_malformed`."""
    try:
        v = strict_json(text)
        exact(v, ["kind", "name", "media", "items", "negatives"])
        need(subset(v) and v["kind"] == CORPUS and is_term(v["name"]) and v["media"] in ("image", "video", "audio"))
        for k in ("items", "negatives"):
            need(isinstance(v[k], list) and len(v[k]) >= 1)
            for i in v[k]:
                if v["media"] == "audio":
                    exact(i, ["name", "rate", "channels", "length", "samples"])
                    need(is_term(i["name"]) and all(is_uint(i[x]) and i[x] >= 1 for x in ("rate", "channels", "length"))
                         and is_hex(i["samples"], 32))
                    need(i["length"] * i["channels"] * 2 <= MAX_INT)
                else:
                    exact(i, ["name", "frames", "width", "height", "samples"])
                    need(is_term(i["name"]) and all(is_uint(i[x]) and i[x] >= 1 for x in ("frames", "width", "height"))
                         and is_hex(i["samples"], 32))
                    need(v["media"] == "video" or i["frames"] == 1)
                    need(i["frames"] * i["width"] * i["height"] * 3 <= MAX_INT)
            need(len({i["name"] for i in v[k]}) == len(v[k]))
    except Bad:
        raise Refuse("corpus_malformed")
    return v


# ---------------------------------------------------------------- §4 the transforms

def r(lo, hi):
    return lambda x: lo <= x <= hi


# term: (parameter checks, media); a check per parameter, then any relation across them.
TRANSFORMS = {
    "wai-attack:none": ([], None),
    "wai-attack:int-motion": ([r(1, 255)], None),
    "wai-attack:staged-base": ([r(1, 255)], None),
    "wai-attack:denoise3": ([r(1, 16)], None),
    "wai-attack:scale": ([r(1, 16), r(1, 16)], lambda p: p[0] <= 2 * p[1]),
    "wai-attack:aspect": ([r(1, 16), r(1, 16)], lambda p: p[0] <= 2 * p[1]),
    "wai-attack:upscale2": ([], None),
    "wai-attack:half-then-int-motion": ([r(1, 255)], None),
    "wai-attack:crop": ([r(0, 99)], None),
    "wai-attack:mirror": ([], None),
    "wai-attack:rotate-90": ([], None),
    "wai-attack:rotate": ([r(0, 65536), r(0, 65536), r(0, 1)], None),
    "wai-attack:noise": ([r(0, 255)], None),
    "wai-attack:brightness": ([r(0, 255), r(0, 1)], None),
    "wai-attack:contrast": ([r(1, 16), r(1, 16)], None),
    "wai-attack:gamma": ([r(1, 6), r(1, 6)], None),
    "wai-attack:local-blur": ([], None),
    "wai-attack:clip-int-motion": ([r(1, 255)], None),
    "wai-attack:frame-drop": ([], None),
    "wai-attack:one-frame": ([], None),
}
VIDEO_ONLY = {"wai-attack:clip-int-motion", "wai-attack:frame-drop", "wai-attack:one-frame"}
PICTURE = set(TRANSFORMS) - {"wai-attack:none"}
# §4.1: the pinned audio transforms.
_RATE = lambda x: x in AM.RATES  # noqa: E731
AUDIO = {
    "wai-attack:audio-gain": ([r(1, 16), r(1, 16)], None),
    "wai-attack:audio-trim": ([r(0, 60000), r(0, 60000)], None),
    "wai-attack:audio-resample": ([_RATE], None),
    "wai-attack:audio-resample-trip": ([_RATE], None),
    "wai-attack:audio-noise": ([r(0, 32767)], None),
    "wai-attack:audio-mix-noise": ([r(1, 1000), r(1, 1000)], None),
    "wai-attack:audio-mulaw": ([], None),
    "wai-attack:audio-requantize": ([r(1, 16)], None),
    "wai-attack:audio-echo": ([r(1, 1000), r(1, 16), r(1, 16)], lambda p: p[1] <= p[2]),
    "wai-attack:audio-mono": ([], None),
}
# §4.2: the measured transforms, registered, never run here.
MEASURED = {
    "wai-attack:opus": ([r(6000, 510000)], None),
    "wai-attack:mp3": ([r(8000, 320000)], None),
    "wai-attack:aac": ([r(8000, 320000)], None),
    "wai-attack:g722": ([], None),
    "wai-attack:g726": ([lambda x: x in (16000, 24000, 32000, 40000)], None),
    "wai-attack:g723-1": ([], None),
    "wai-attack:resample-band": ([_RATE], None),
}
TRANSFORMS.update(AUDIO)
TRANSFORMS.update(MEASURED)


def transform_ok(term, params):
    """§4: True for a registered transform whose parameters fit, None for an unregistered term;
    `transform_invalid` otherwise."""
    if term not in TRANSFORMS:
        return None
    checks, rel = TRANSFORMS[term]
    ok = len(params) == len(checks) and all(c(p) for c, p in zip(checks, params)) and (rel is None or rel(params))
    if not ok and "transform-ranges" not in SKIP:
        raise Refuse("transform_invalid")
    return True


def gamma_table(n, d):
    """§4 Gamma: the greatest y with y = 0 or (2y - 1)^d * 255^n <= 2^d * v^n * 255^d."""
    t = []
    for v in range(256):
        rhs = 2 ** d * v ** n * 255 ** d
        y = 0
        while y < 255 and (2 * (y + 1) - 1) ** d * 255 ** n <= rhs:
            y += 1
        t.append(y)
    return t


def per_frame(px, w, h, n, f):
    """Apply f(frame) -> (frame', w', h') to each of n frames."""
    fb = w * h * 3
    out, nw, nh = [], w, h
    for i in range(n):
        g, nw, nh = f(px[i * fb:(i + 1) * fb])
        out.append(g)
    return b"".join(out), nw, nh, n


def apply(term, p, px, w, h, n, seed):
    """§4: (samples, width, height, frames) of a registered transform applied at `seed`."""
    empty = (b"", 0, 0, n)
    if term == "wai-attack:none":
        return px, w, h, n
    if term == "wai-attack:int-motion":
        return per_frame(px, w, h, n, lambda fr: (M.int_motion(fr, p[0]), w, h))
    if term in ("wai-attack:staged-base", "wai-attack:half-then-int-motion") and (w < 2 or h < 2):
        return empty
    if term == "wai-attack:staged-base":
        def staged(fr):
            hw, hh = w // 2, h // 2
            half = bytearray(hw * hh * 3)
            for y in range(hh):
                for x in range(hw):
                    for c in range(3):
                        s = sum(fr[((2 * y + dy) * w + 2 * x + dx) * 3 + c] for dy in (0, 1) for dx in (0, 1))
                        half[(y * hw + x) * 3 + c] = (s + 2) >> 2
            return M.upscale2(M.int_motion(bytes(half), p[0]), hw, hh), 2 * hw, 2 * hh
        return per_frame(px, w, h, n, staged)
    if term == "wai-attack:denoise3":
        def dn(fr):
            for _ in range(p[0]):
                fr = M.denoise3(fr, w, h)
            return fr, w, h
        return per_frame(px, w, h, n, dn)
    if term == "wai-attack:scale":
        nw, nh = w * p[0] // p[1], h * p[0] // p[1]
        if nw == 0 or nh == 0:
            return empty
        return per_frame(px, w, h, n, lambda fr: (M.area(fr, w, h, nw, nh), nw, nh))
    if term == "wai-attack:aspect":
        nw = w * p[0] // p[1]
        if nw == 0:
            return empty
        return per_frame(px, w, h, n, lambda fr: (M.area(fr, w, h, nw, h), nw, h))
    if term == "wai-attack:upscale2":
        return per_frame(px, w, h, n, lambda fr: (M.upscale2(fr, w, h), 2 * w, 2 * h))
    if term == "wai-attack:half-then-int-motion":
        nw, nh = w // 2, h // 2
        return per_frame(px, w, h, n, lambda fr: (M.int_motion(M.area(fr, w, h, nw, nh), p[0]), nw, nh))
    if term == "wai-attack:crop":
        dx, dy = w * p[0] // 200, h * p[0] // 200
        nw, nh = w - 2 * dx, h - 2 * dy
        return per_frame(px, w, h, n, lambda fr: (b"".join(fr[(y * w + dx) * 3:(y * w + dx + nw) * 3] for y in range(dy, dy + nh)), nw, nh))
    if term == "wai-attack:mirror":
        return per_frame(px, w, h, n, lambda fr: (b"".join(fr[(y * w + x) * 3:(y * w + x) * 3 + 3]
                                                           for y in range(h) for x in range(w - 1, -1, -1)), w, h))
    if term == "wai-attack:rotate-90":
        return per_frame(px, w, h, n, lambda fr: (b"".join(fr[((h - 1 - xp) * w + yp) * 3:((h - 1 - xp) * w + yp) * 3 + 3]
                                                           for yp in range(w) for xp in range(h)), h, w))
    if term == "wai-attack:rotate":
        s = -p[1] if p[2] == 1 else p[1]
        return per_frame(px, w, h, n, lambda fr: (M.rotate(fr, w, h, p[0], s), w, h))
    if term == "wai-attack:noise":
        st = M.stream(b"wai-mark:battery noise", seed)
        return bytes(M.clamp(v + next(st) % (2 * p[0] + 1) - p[0]) for v in px), w, h, n
    if term == "wai-attack:brightness":
        b = -p[0] if p[1] == 1 else p[0]
        return bytes(M.clamp(v + b) for v in px), w, h, n
    if term == "wai-attack:contrast":
        return bytes(M.clamp(128 + M.rnd((v - 128) * p[0], p[1])) for v in px), w, h, n
    if term == "wai-attack:gamma":
        t = gamma_table(p[0], p[1])
        return bytes(t[v] for v in px), w, h, n
    if term == "wai-attack:local-blur":
        return per_frame(px, w, h, n, lambda fr: M.attack("local-blur", fr, w, h, seed))
    if term == "wai-attack:frame-drop":
        fb = w * h * 3
        kept = list(range(0, n, 2))
        return b"".join(px[i * fb:(i + 1) * fb] for i in kept), w, h, len(kept)
    if term == "wai-attack:one-frame":
        return px[:w * h * 3], w, h, 1
    raise NotImplementedError(term)


# ---------------------------------------------------------------- the algorithm and §6

def tent64_digest():
    """synthesis §7.4: the definition's digest, as its text states it, which is the SHA-256 of
    the synthesis corpus's scheme.json."""
    s = " ".join(open(SYNTHESIS, encoding="utf-8").read().split())
    m = re.search(r"and its digest is `([0-9a-f]{64})`", s)
    stated = m.group(1) if m else None
    with open(os.path.join(HERE, "synthesis-mark", "scheme.json"), "rb") as f:
        computed = hashlib.sha256(f.read()).hexdigest()
    return stated if stated == computed else None


TENT64 = M.SCHEME
TENT64_DIGEST = tent64_digest()


def value_ok(v):
    """§5 check 5: a payload of synthesis §7.2."""
    b = bytes.fromhex(v)
    return len(b) == 8 and b[0] <= 3


HARNESS_MEMO = {}


def band64_digest():
    """synthesis §7.10.1: the definition's digest, as its text states it, which is the SHA-256 of
    the audio corpus's scheme-band64.json."""
    s = open(SYNTHESIS, encoding="utf-8").read()
    at = s.find("\n#### 7.10.1 ")
    m = re.search(r"its digest is\s+`([0-9a-f]{64})`", s[at:at + 4000]) if at >= 0 else None
    stated = m.group(1) if m else None
    try:
        with open(os.path.join(HERE, "audio-mark", "scheme-band64.json"), "rb") as f:
            computed = hashlib.sha256(f.read()).hexdigest()
    except OSError:
        return None
    return stated if stated == computed else None


BAND64_DIGEST = band64_digest()


def run_audio_harness(doc, samples, values, battery):
    """§6 for wai-mark:band64 over audio items whose samples are held."""
    key = ("audio", tuple(i["samples"] for i in doc["items"]), tuple(i["samples"] for i in doc["negatives"]), tuple(values),
           jcs([[a["transform"], a["params"], a["seed"]] for a in battery]))
    if key in HARNESS_MEMO:
        return HARNESS_MEMO[key]

    def chans(i):
        return i["rate"], AM.unpack(samples[i["samples"]], i["channels"])
    marked = []
    for i, val in zip(doc["items"], values):
        rate, c = chans(i)
        if AM.admits(rate, len(c), len(c[0])):
            c = AM.embed(rate, c, bytes.fromhex(val))[0]
        marked.append((rate, c))
    negs = [chans(i) for i in doc["negatives"]]
    rows = []
    for a in battery:
        name = a["transform"].split(":", 1)[1]
        survived = wrong = fp = 0
        for (rate, c), val in zip(marked, values):
            r2, c2 = AM.attack(name, a["params"], rate, c, a["seed"])
            got = AM.detect(r2, len(c2), AM.pack(c2))
            if got is not None:
                if got[0].hex() == val:
                    survived += 1
                else:
                    wrong += 1
        for rate, c in negs:
            r2, c2 = AM.attack(name, a["params"], rate, c, a["seed"])
            if AM.detect(r2, len(c2), AM.pack(c2)) is not None:
                fp += 1
        rows.append((len(marked), survived, wrong, len(negs), fp))
    HARNESS_MEMO[key] = rows
    return rows


def run_harness(items, negatives, values, battery):
    """§6 for wai-mark:tent64: the counts of each attack, in order. `items` and `negatives` are
    (samples, width, height, frames)."""
    key = (tuple(sha(i[0]) for i in items), tuple(sha(i[0]) for i in negatives), tuple(values),
           jcs([[a["transform"], a["params"], a["seed"]] for a in battery]))
    if key in HARNESS_MEMO:
        return HARNESS_MEMO[key]
    marked = []
    for (px, w, h, n), v in zip(items, values):
        payload = bytes.fromhex(v)
        marked.append((M.embed(px, w, h, n, payload)[0] if M.admits(w, h, n) else px, w, h, n))
    rows = []
    for a in battery:
        t, p, s = a["transform"], a["params"], a["seed"]
        survived = wrong = fp = 0
        for (px, w, h, n), v in zip(marked, values):
            got = M.detect(*apply(t, p, px, w, h, n, s))
            if got is not None:
                if got[0].hex() == v:
                    survived += 1
                else:
                    wrong += 1
        for px, w, h, n in negatives:
            if M.detect(*apply(t, p, px, w, h, n, s)) is not None:
                fp += 1
        rows.append((len(items), survived, wrong, len(negatives), fp))
    HARNESS_MEMO[key] = rows
    return rows


# ---------------------------------------------------------------- §7 rates

def reduce_(n, d):
    g = math.gcd(n, d) or 1
    return (n // g, d // g)


def rates_of(trials, survived, negatives, fp):
    sv = reduce_(survived, trials)
    return (sv, reduce_(fp, negatives), reduce_(survived * negatives + (negatives - fp) * trials, 2 * trials * negatives),
            sv if 100 * fp <= negatives else None)


# ---------------------------------------------------------------- §5 and §8 checks

def check_evidence(v, rates, corpus_text, case):
    """§5's checks 1 to 8: the record's status."""
    battery, table = v["battery"], v["table"]
    # 1.
    need(len(battery) >= 1 and len({a["attack"] for a in battery}) == len(battery)
         and all(a["seed"] < 2 ** 32 for a in battery), "battery_malformed")
    reg = [transform_ok(a["transform"], a["params"]) for a in battery]
    if "media" not in SKIP:
        need(not any(ok and v["media"] == "image" and a["transform"] in VIDEO_ONLY for ok, a in zip(reg, battery)),
             "battery_not_applicable")
        need(not any(ok and v["media"] == "audio" and a["transform"] in PICTURE for ok, a in zip(reg, battery)),
             "battery_not_applicable")
        need(not any(ok and v["media"] != "audio" and (a["transform"] in AUDIO or a["transform"] in MEASURED)
                     for ok, a in zip(reg, battery)), "battery_not_applicable")
    # 2.
    need(len(table) == len(battery) and all(r["attack"] == a["attack"] for r, a in zip(table, battery)), "table_malformed")
    # 3.
    neg = table[0]["negatives"]
    if "counts" not in SKIP:
        for row in table:
            need(row["trials"] == len(v["values"]) and row["trials"] >= 1 and row["survived"] + row["wrong"] <= row["trials"]
                 and row["negatives"] >= 1 and row["negatives"] == neg and row["false_positives"] <= row["negatives"],
                 "table_inconsistent")
    # 4.
    if "rates" not in SKIP:
        for row, rt in zip(table, rates):
            need(rates_of(row["trials"], row["survived"], row["negatives"], row["false_positives"]) == rt, "rate_mismatch")
    # 5.
    if v["algorithm"] == AM.SCHEME:
        need(v["algorithm_digest"] == BAND64_DIGEST and v["binding_type"] == "watermark", "algorithm_mismatch")
        if "values" not in SKIP:
            need(all(value_ok(x) for x in v["values"]), "values_invalid")
    registered = v["algorithm"] == TENT64
    if registered:
        need(v["algorithm_digest"] == TENT64_DIGEST and v["binding_type"] == "watermark", "algorithm_mismatch")
        if "values" not in SKIP:
            need(all(value_ok(x) for x in v["values"]), "values_invalid")
    # 6.
    if corpus_text is None:
        return "stated"
    doc = read_corpus(corpus_text)
    if "corpus-digest" not in SKIP:
        need(hashlib.sha256(jcs(doc).encode("utf-8")).hexdigest() == v["corpus"] and doc["media"] == v["media"], "corpus_mismatch")
    need(len(doc["items"]) == len(v["values"]) and len(doc["negatives"]) == neg, "table_inconsistent")
    # 7.
    every = doc["items"] + doc["negatives"]
    if "sizes" not in SKIP:
        for i in every:
            s = case["samples"].get(i["samples"])
            size = i["length"] * i["channels"] * 2 if doc["media"] == "audio" else i["frames"] * i["width"] * i["height"] * 3
            need(s is None or len(s) == size, "samples_mismatch")
    # 8.
    held = all(i["samples"] in case["samples"] for i in every)
    if doc["media"] == "audio":
        pinned = all(a["transform"] == "wai-attack:none" or a["transform"] in AUDIO for a in battery)
        bounded = all(i["length"] * i["channels"] * 2 <= ITEM_CAP for i in every)
        if not (v["algorithm"] == AM.SCHEME and AM.SCHEME in case["implements"] and held and bounded
                and v["harness"] == HARNESS and pinned) or "rerun" in SKIP:
            return "stated"
        got = run_audio_harness(doc, case["samples"], v["values"], battery)
        want = [(r["trials"], r["survived"], r["wrong"], r["negatives"], r["false_positives"]) for r in table]
        need(got == want, "table_irreproducible")
        return "reproduced"
    bounded = all(i["width"] * i["height"] * 3 <= FRAME_CAP and i["frames"] * i["width"] * i["height"] * 3 <= ITEM_CAP
                  for i in every)
    runnable = registered and TENT64 in case["implements"]
    if not (runnable and held and bounded and v["harness"] == HARNESS and all(reg)) or "rerun" in SKIP:
        return "stated"
    frames = lambda i: (case["samples"][i["samples"]], i["width"], i["height"], i["frames"])  # noqa: E731
    got = run_harness([frames(i) for i in doc["items"]], [frames(i) for i in doc["negatives"]], v["values"], battery)
    want = [(r["trials"], r["survived"], r["wrong"], r["negatives"], r["false_positives"]) for r in table]
    need(got == want, "table_irreproducible")
    return "reproduced"


def check_claim(v, held, require_reproduced):
    """§8: the claim's status."""
    cited = []
    for link in v["evidence"]:
        need(link in held, "evidence_not_held")
        h = held[link]
        need(h["v"]["algorithm"] == v["algorithm"] and h["v"]["algorithm_digest"] == v["algorithm_digest"], "evidence_mismatch")
        cited.append(h)
    if require_reproduced:
        need(all(h["status"] == "reproduced" for h in cited), "evidence_not_reproduced")
    for c in v["survives"]:
        measured = False
        for h in cited:
            if h["v"]["media"] != c["media"]:
                continue
            for a, row in zip(h["v"]["battery"], h["v"]["table"]):
                if a["transform"] == c["transform"] and a["params"] == c["params"]:
                    measured = True
                    if "survival" not in SKIP:
                        need(row["survived"] == row["trials"] and row["wrong"] == 0 and row["false_positives"] == 0,
                             "claim_exceeds_evidence")
        need(measured, "claim_unevidenced")
    return "reproduced" if all(h["status"] == "reproduced" for h in cited) else "stated"


def verify_case(case):
    """§9.2: ("accept", statements) or ("refuse", reason, position)."""
    held, out = {}, []
    for i, text in enumerate(case["records"]):
        try:
            try:
                kind = strict_json(text).get("kind")
            except (Bad, AttributeError):
                kind = None
            if kind == RECORD:
                v, rates = read_evidence(text)
                status = check_evidence(v, rates, case["corpora"].get(i), case)
                link = link_hash(v).hex()
                held[link] = {"v": v, "status": status}
                out.append({"kind": RECORD, "link": link, "status": status, "algorithm": v["algorithm"], "corpus": v["corpus"],
                            "attacks": len(v["battery"])})
            elif kind == CLAIM:
                v = read_claim(text)
                status = check_claim(v, held, case["require_reproduced"])
                out.append({"kind": CLAIM, "link": link_hash(v).hex(), "status": status, "algorithm": v["algorithm"],
                            "attacks": len(v["survives"])})
            else:
                raise Bad()
        except Bad:
            return ("refuse", "record_malformed", i)
        except Refuse as e:
            return ("refuse", e.reason, i)
    return ("accept", out)


# ---------------------------------------------------------------- the corpus's cases

def load(corpus_dir):
    files = {}
    for root, _, names in os.walk(corpus_dir):
        for n in names:
            p = os.path.join(root, n)
            rel = os.path.relpath(p, corpus_dir).replace(os.sep, "/")
            if rel != "README.md":
                with open(p, "rb") as f:
                    files[rel] = f.read()
    return files


def case_of(c, files):
    corpora = {}
    for g in c["corpora"]:
        corpora[g["record"]] = files[g["file"]].decode("utf-8") if "file" in g else g["text"]
    samples = {}
    for h in c["samples"]:
        b = files.get("frames/%s.rgb" % h)
        if b is not None and hashlib.sha256(b).hexdigest() == h:
            samples[h] = b
    return {"records": c["records"], "corpora": corpora, "samples": samples, "implements": c["implements"],
            "require_reproduced": c["policy"]["evidence"] == "reproduced"}


def outcome(r):
    if r[0] == "accept":
        return {"outcome": "accept", "statements": r[1]}
    return {"outcome": "refuse", "reason": r[1], "record": r[2]}


def check(doc, files, quiet=False, only=None):
    fails = []
    for c in doc["cases"]:
        if only is not None and c["name"] not in only:
            continue
        got = outcome(verify_case(case_of(c, files)))
        ok = got == c["expected"]
        if not quiet:
            print(("PASS" if ok else "FAIL") + " %s" % c["name"])
        if not ok:
            fails.append("%s: expected %s, got %s" % (c["name"], json.dumps(c["expected"])[:160], json.dumps(got)[:160]))
    return fails


# ---------------------------------------------------------------- the corpus's files and texts

def reasons_of_extension():
    s = open(EXTENSION, encoding="utf-8").read()
    at = s.index("\n## 12. Reasons")
    end = s.index("\n## 13.", at)
    return re.findall(r"^\| `([a-z_]+)` \|", s[at:end], re.M)


def transforms_of_extension():
    s = open(EXTENSION, encoding="utf-8").read()
    at = s.index("\n## 4. The transforms")
    end = s.index("\n## 5.", at)
    return re.findall(r"^\| `(wai-attack:[a-z0-9-]+)` \|", s[at:end], re.M)  # §4, §4.1 and §4.2, in order


def synthesis_claims():
    """synthesis §7.5.8's claims: (pictures, clips), by name."""
    s = open(SYNTHESIS, encoding="utf-8").read()
    at = s.index("\n#### 7.5.8 ")
    end = s.index("\n### 7.6", at)
    t = " ".join(s[at:end].split())
    m = re.search(r"For pictures, these attacks, each with the parameter measured: (.*?)\. For clips: (.*?)\.", t)
    return re.findall(r"`([^`]+)`", m.group(1)), re.findall(r"`([^`]+)`", m.group(2))


ROTATIONS = {"rotate-0.5": [65534, 572, 0], "rotate-1": [65526, 1144, 0], "rotate-2": [65496, 2287, 0]}


def transform_of_name(name):
    """§10's table: a name of synthesis §7.5.7 as a transform and its parameters."""
    m = re.fullmatch(r"(clip-int-motion|int-motion|staged-base|half-then-int-motion)-q(\d+)", name)
    if m:
        return "wai-attack:" + m.group(1), [int(m.group(2))]
    m = re.fullmatch(r"(scale|aspect|contrast)-(\d+)/(\d+)", name)
    if m:
        return "wai-attack:" + m.group(1), [int(m.group(2)), int(m.group(3))]
    m = re.fullmatch(r"crop-(\d+)%", name)
    if m:
        return "wai-attack:crop", [int(m.group(1))]
    m = re.fullmatch(r"noise-(\d+)", name)
    if m:
        return "wai-attack:noise", [int(m.group(1))]
    if name in ROTATIONS:
        return "wai-attack:rotate", ROTATIONS[name]
    if name == "denoise3":
        return "wai-attack:denoise3", [1]
    if name == "denoise3-x4":
        return "wai-attack:denoise3", [4]
    if name in ("brightness+32", "brightness-32"):
        return "wai-attack:brightness", [32, 1 if name.endswith("-32") else 0]
    if name in ("none", "upscale2", "mirror", "rotate-90", "local-blur", "frame-drop", "one-frame"):
        return "wai-attack:" + name, []
    raise ValueError(name)


def check_tent64(files):
    """§10: the records' batteries are synthesis §7.5.7's lists by §10's table and their seeds the
    index; their tables and values are battery.json's; the claim is synthesis §7.5.8's list."""
    fails = []
    with open(os.path.join(HERE, "synthesis-mark", "battery.json"), encoding="utf-8") as f:
        battery = json.load(f)
    payload = {p["name"]: p["payload"] for p in battery["pictures"]}
    lists = {"image": battery["image_attacks"], "video": battery["clip_attacks"]}
    links = []
    for name in ("tent64-spectral", "tent64-leaves", "tent64-clips"):
        v = strict_json(files["records/%s.json" % name].decode("utf-8"))
        links.append(link_hash(v).hex())
        doc = strict_json(files["corpus/%s.json" % name].decode("utf-8"))
        st = name[len("tent64-"):]
        want_b = [{"attack": n, "transform": transform_of_name(n)[0], "params": transform_of_name(n)[1], "seed": i}
                  for i, n in enumerate(lists[v["media"]])]
        if v["battery"] != want_b:
            fails.append("%s: its battery is not synthesis 7.5.7's list" % name)
        if v["values"] != [payload[i["name"]] for i in doc["items"]]:
            fails.append("%s: its values are not battery.json's payloads" % name)
        if hashlib.sha256(jcs(doc).encode()).hexdigest() != v["corpus"]:
            fails.append("%s: its corpus file is not its corpus" % name)
        for row in v["table"]:
            t = next(x for x in battery["table"] if x["attack"] == row["attack"] and x["set"] == st)
            u = next(x for x in battery["unmarked"] if x["attack"] == row["attack"] and x["set"] == st)
            if (row["trials"], row["survived"], row["wrong"], row["negatives"], row["false_positives"]) != \
                    (t["trials"], t["detected"], t["wrong"], u["trials"], u["detections"]):
                fails.append("%s %s: the row is not battery.json's" % (name, row["attack"]))
    claim = strict_json(files["records/tent64-claim.json"].decode("utf-8"))
    pics, clips = synthesis_claims()
    want = ([{"media": "image", "transform": transform_of_name(n)[0], "params": transform_of_name(n)[1]} for n in pics]
            + [{"media": "video", "transform": transform_of_name(n)[0], "params": transform_of_name(n)[1]} for n in clips])
    if claim["survives"] != want:
        fails.append("the tent64 claim is not synthesis 7.5.8's list")
    if claim["evidence"] != links:
        fails.append("the tent64 claim does not cite the three records")
    return fails


def check_files(doc, files):
    """The corpus documents are canonical, each frame file is named by its SHA-256, the records'
    files are the texts the accepted cases verify, and nothing else is there."""
    fails = []
    for k, b in files.items():
        if k.startswith("corpus/"):
            if jcs(strict_json(b.decode("utf-8"))).encode("utf-8") != b:
                fails.append("%s is not its JCS" % k)
        elif k.startswith("frames/"):
            if k != "frames/%s.rgb" % hashlib.sha256(b).hexdigest():
                fails.append("%s is not named by its SHA-256" % k)
        elif k.startswith("records/"):
            pass
        elif k != "vectors.json":
            fails.append("an extra file: %s" % k)
    by_name = {c["name"]: c for c in doc["cases"]}
    want = {
        "records/conformance.json": by_name["conformance_reproduced"]["records"][0],
        "records/conformance-claim.json": by_name["conformance_claim_reproduced"]["records"][1],
        "records/tent64-spectral.json": by_name["tent64_records_and_claim"]["records"][0],
        "records/tent64-leaves.json": by_name["tent64_records_and_claim"]["records"][1],
        "records/tent64-clips.json": by_name["tent64_records_and_claim"]["records"][2],
        "records/tent64-claim.json": by_name["tent64_records_and_claim"]["records"][3],
    }
    for k, t in want.items():
        if files.get(k) != t.encode("utf-8"):
            fails.append("%s is not the text its accepted case verifies" % k)
    for k in files:
        if k.startswith("records/") and k not in want:
            fails.append("an extra record: %s" % k)
    return fails


# ---------------------------------------------------------------- mutations and weakened checks

def edit_case(doc, name, f):
    d = copy.deepcopy(doc)
    c = next(c for c in d["cases"] if c["name"] == name)
    f(c)
    return d, name


def edit_record(doc, name, i, f):
    """A case's record edited by f and left with its signature, which then fails, or with
    rates and counts that no longer agree."""
    def g(c):
        v = json.loads(c["records"][i])
        f(v)
        c["records"][i] = json.dumps(v, separators=(",", ":"))
    return edit_case(doc, name, g)


def mutations(doc, files):
    """Copies of the corpus, or of its files, with one defect each: (label, doc, files, case)."""
    out = []

    def m(label, pair, fs=None):
        out.append((label, pair[0], fs if fs is not None else files, pair[1]))

    m("a reproduced record's survived count changed", edit_record(doc, "conformance_reproduced", 0,
      lambda v: v["table"][0].__setitem__("survived", 2)))
    m("a reproduced record's seed changed", edit_record(doc, "conformance_reproduced", 0,
      lambda v: v["battery"][7].__setitem__("seed", 99)))
    m("a stated record's rate written as a decimal", edit_case(doc, "tent64_records_and_claim",
      lambda c: c["records"].__setitem__(0, c["records"][0].replace('"survival":[1,1]', '"survival":1.0', 1))))
    m("the claim given a stated record's corpus file from another set", edit_case(doc, "tent64_records_and_claim",
      lambda c: c["corpora"][0].__setitem__("file", "corpus/tent64-leaves.json")))
    m("the policy switched to require reproduced evidence", edit_case(doc, "tent64_records_and_claim",
      lambda c: c["policy"].__setitem__("evidence", "reproduced")))
    m("the samples dropped from a reproduced case", edit_case(doc, "conformance_reproduced", lambda c: c.__setitem__("samples", [])))
    m("an accepted statement's status changed", edit_case(doc, "conformance_stated_without_samples",
      lambda c: c["expected"]["statements"][0].__setitem__("status", "reproduced")))
    m("a refusal's reason changed", edit_case(doc, "claim_crop_10_exceeds",
      lambda c: c["expected"].__setitem__("reason", "claim_unevidenced")))
    m("the claim's records reordered after it", edit_case(doc, "tent64_records_and_claim",
      lambda c: c["records"].insert(0, c["records"].pop())))
    frame = next(k for k in files if k.startswith("frames/"))
    bad = dict(files)
    bad[frame] = bytes([files[frame][0] ^ 1]) + files[frame][1:]
    m("a frame file's first byte changed", (doc, "conformance_reproduced"), bad)
    return out


CHECKS = {
    "signature": "§2 the signature",
    "exact-rates": "§7 rates as exact fractions",
    "transform-ranges": "§4 the parameters' ranges",
    "media": "§4 the transforms' media",
    "counts": "§5 check 3 the counts",
    "rates": "§5 check 4 the rates",
    "values": "§5 check 5 the values",
    "corpus-digest": "§5 check 6 the corpus digest",
    "sizes": "§5 check 7 the samples' sizes",
    "rerun": "§5 check 8 the rerun",
    "survival": "§8 step 3 survival",
}


def main(corpus_dir):
    fails = []
    st = ED._self_test()
    print(("PASS" if not st else "FAIL") + " Ed25519 strict rule self-test " + ",".join(st))
    fails += ["ed25519 self-test"] if st else []
    print(("PASS" if TENT64_DIGEST else "FAIL") + " wai-mark:tent64's digest, as synthesis 7.4 states it, is scheme.json's")
    fails += [] if TENT64_DIGEST else ["the tent64 digest"]
    reasons = reasons_of_extension()
    print(("PASS" if len(reasons) == 21 else "FAIL") + " §12 read from the extension: %d reasons" % len(reasons))
    fails += [] if len(reasons) == 21 else ["§12 does not read"]
    listed = transforms_of_extension()
    ok = listed == list(TRANSFORMS)
    print(("PASS" if ok else "FAIL") + " §4 registers exactly the transforms this verifier implements")
    fails += [] if ok else ["§4's transforms"]
    files = load(corpus_dir)
    doc = json.loads(files["vectors.json"])
    fails += check(doc, files)
    f = check_files(doc, files)
    print(("PASS" if not f else "FAIL") + " the corpus documents, frames and records are what the cases verify, and nothing else is there")
    fails += f
    f = check_tent64(files)
    print(("PASS" if not f else "FAIL") + " the tent64 records are synthesis 7.5.7's battery over battery.json's tallies, and the claim is synthesis 7.5.8's list")
    fails += f
    given = {c["expected"]["reason"] for c in doc["cases"] if c["expected"]["outcome"] == "refuse"}
    missing = [x for x in reasons if x not in given]
    unknown = sorted(given - set(reasons))
    print(("PASS" if not missing and not unknown else "FAIL") + " every reason of §12 refuses a case, and no other reason does")
    fails += ["no case gives %s" % x for x in missing] + ["a reason §12 does not list: %s" % x for x in unknown]
    for label_, d, fs, case in mutations(doc, files):
        caught = check(d, fs, quiet=True, only={case})
        print(("PASS" if caught else "FAIL") + " mutated copy fails: %s%s" % (label_, " (%s)" % caught[0][:90] if caught else ""))
        if not caught:
            fails.append("a mutated copy verifies: %s" % label_)
    for name, what in CHECKS.items():
        SKIP.clear()
        SKIP.add(name)
        caught = check(doc, files, quiet=True)
        SKIP.clear()
        print(("PASS" if caught else "FAIL") + " a verifier without %s fails the corpus%s" % (what, " (%s)" % caught[0][:70] if caught else ""))
        if not caught:
            fails.append("the corpus does not catch a verifier without %s" % what)
    for x in fails:
        print("FAIL " + x)
    print(("PASS" if not fails else "FAIL") + " robustness corpus: %d cases" % len(doc["cases"]))
    return 1 if fails else 0


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