be8585f926
extensions/robustness-evidence.md: a corpus named item by item by the SHA-256 of its samples, the attack battery as registered integer transforms with parameters and seeds (synthesis §7.5.7's battery with its parameters free, and gamma), the signed robustness record whose table the harness recomputes and whose rates are exact fractions, and the robustness claim that holds only where every item of a cited record's corpus survived the attack and no negative was read as marked. wai-rs module robustness (feature robustness), wai_robustness_vectors, the corpus with wai-mark:tent64's records over the measurement sets and its claim, and robustness_verify.py, which reruns the conformance battery end to end with _wai_mark.py. Co-Authored-By: Claude Opus 5.5 <[email protected]>
948 lines
40 KiB
Python
948 lines
40 KiB
Python
#!/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 _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"))
|
||
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")
|
||
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") 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"))
|
||
for k in ("items", "negatives"):
|
||
need(isinstance(v[k], list) and len(v[k]) >= 1)
|
||
for i in v[k]:
|
||
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"}
|
||
|
||
|
||
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 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")
|
||
# 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.
|
||
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"])
|
||
need(s is None or len(s) == i["frames"] * i["width"] * i["height"] * 3, "samples_mismatch")
|
||
# 8.
|
||
held = all(i["samples"] in case["samples"] for i in every)
|
||
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)
|
||
|
||
|
||
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]))
|