6b693f1141
Interaction of #425 (wai-mark:tent64s) with #408 (exact-upscale, in the base): #408 made synthesis_verify.py's outcome() read a case's held priors (synthesis §11.1) and gave tent64's cases an empty list; #425's tent64s case loop, written before #408 landed, built its cases without one, so every tiled case read as unreadable (KeyError 'priors'). The tiled cases now state an empty list, as tent64's do. Co-Authored-By: Claude Opus 5.5 <[email protected]>
722 lines
33 KiB
Python
722 lines
33 KiB
Python
#!/usr/bin/env python3
|
||
"""Second implementation of the registered watermark scheme of extensions/synthesis.md
|
||
(§7.4 to §7.8), over the corpus `live-conformance/synthesis-mark/`.
|
||
|
||
Written from the extension's text. It shares no code with wai-rs: the scheme (the stream, the
|
||
shuffle, the signs, the mask, the BCH code with a decoder of its own, the embedding, the detector)
|
||
and the picture attacks of the battery are _wai_mark.py's, written from §7.5; every case is
|
||
verified by synthesis_verify.py, the synthesis corpus's second implementation, with §7.6's checks.
|
||
It reads from the extension the scheme's definition and digest (§7.4), its generator polynomial
|
||
and the shuffle's first entries (§7.5.3), the figures of §7.5.6, and the table and the claims of
|
||
§7.5.8.
|
||
|
||
python3 -I synthesis_mark_verify.py <live-conformance/synthesis-mark>
|
||
|
||
It checks that every frame file is named by its SHA-256; that the definition is §7.4's and has
|
||
§7.4's digest; that the layout and the code are §7.5.3's; that every codeword, decoding,
|
||
embedding and detection of the corpus is what this implementation gives; that every picture
|
||
attack of the battery sample gives the result and the reading the corpus states; that every case
|
||
gets the outcome, the reason, the place and the statements it states, and that the codes only this
|
||
corpus gives each refuse one; that §7.5.8's table is battery.json's and its claims are attacks the
|
||
tallies show survived, and §7.5.6's figures are battery.json's. Then it mutates copies of the
|
||
corpus and requires each to fail, and weakens its own verifier in turn (no recomputation, no
|
||
detection run again, no agreement by inheritance) and requires the corpus to catch each. 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)
|
||
import _wai_mark as MARK # noqa: E402
|
||
import synthesis_verify as SV # noqa: E402
|
||
|
||
TEXT = " ".join(SV._TEXT.split()) # the extension, its white space folded
|
||
|
||
|
||
def sha(b):
|
||
return hashlib.sha256(b).digest()
|
||
|
||
|
||
def sec(head):
|
||
return SV.section(SV._TEXT, head)
|
||
|
||
|
||
# ---------------------------------------------------------------- what the extension states
|
||
|
||
def text_definition():
|
||
"""§7.4: the definition, its lines joined with no white space."""
|
||
s = sec("### 7.4 ")
|
||
block = s.split("```")[1]
|
||
return "".join(line.strip() for line in block.strip().splitlines()).encode("utf-8")
|
||
|
||
|
||
def text_generator():
|
||
m = re.search(r"in hexadecimal, are\s+`([0-9a-f]+)`", sec("#### 7.5.3 "))
|
||
return m.group(1) if m else None
|
||
|
||
|
||
def text_shuffle_head():
|
||
m = re.search(r"π begins ([0-9, ]+?), and the corpus", " ".join(sec("#### 7.5.3 ").split()))
|
||
return [int(x) for x in m.group(1).split(", ")] if m else None
|
||
|
||
|
||
def text_figures():
|
||
"""§7.5.6's false-positive bound and the words' figures, in the order written."""
|
||
s = " ".join(sec("#### 7.5.6 ").split())
|
||
fp = re.search(r"2\^−([0-9.]+), about", s)
|
||
w = re.search(r"have a mean of ([0-9.]+) ones with a standard deviation of ([0-9.]+), and the ([0-9,]+) pairs of those words "
|
||
r"from different pictures differ in a mean of ([0-9.]+) bits with a standard deviation of ([0-9.]+) and in no fewer "
|
||
r"than ([0-9]+);", s)
|
||
return (fp.group(1) if fp else None), (w.groups() if w else None)
|
||
|
||
|
||
def text_tables():
|
||
"""§7.5.8's two tables: {attack: [cell, ...]}, pictures' and clips'."""
|
||
s = sec("#### 7.5.8 ")
|
||
pictures, clips = {}, {}
|
||
for line in s.splitlines():
|
||
cells = [c.strip() for c in line.strip().strip("|").split("|")]
|
||
if len(cells) == 4 and cells[0].startswith("`"):
|
||
pictures[cells[0].strip("`")] = cells[1:]
|
||
elif len(cells) == 2 and cells[0].startswith("`"):
|
||
clips[cells[0].strip("`")] = cells[1:]
|
||
return pictures, clips
|
||
|
||
|
||
def text_claims():
|
||
s = " ".join(sec("#### 7.5.8 ").split())
|
||
m = re.search(r"For pictures, these attacks, each with the parameter measured: (.*?)\. For clips: (.*?)\.", s)
|
||
if not m:
|
||
return None, None
|
||
return re.findall(r"`([^`]+)`", m.group(1)), re.findall(r"`([^`]+)`", m.group(2))
|
||
|
||
|
||
# ---------------------------------------------------------------- memoised work
|
||
|
||
_ATTACKED = {}
|
||
|
||
|
||
def attacked(px, w, h, name, seed):
|
||
k = (sha(px), w, h, name, seed)
|
||
if k not in _ATTACKED:
|
||
_ATTACKED[k] = MARK.attack(name, px, w, h, seed)
|
||
return _ATTACKED[k]
|
||
|
||
|
||
_DET = {}
|
||
|
||
|
||
def detect(px, w, h, n):
|
||
k = (sha(px), w, h, n)
|
||
if k not in _DET:
|
||
_DET[k] = MARK.detect(px, w, h, n)
|
||
return _DET[k]
|
||
|
||
|
||
_EMB = {}
|
||
|
||
|
||
def embed(px, w, h, n, payload):
|
||
k = (sha(px), w, h, n, payload)
|
||
if k not in _EMB:
|
||
marked, amps = MARK.embed(px, w, h, n, payload)
|
||
_EMB[k] = (sha(marked), ["".join("%x" % a for a in f) for f in amps])
|
||
return _EMB[k]
|
||
|
||
|
||
def detected_json(d):
|
||
return None if d is None else {"payload": d[0].hex(), "orientation": d[1], "errors": d[2]}
|
||
|
||
|
||
def bits(b):
|
||
return "".join("1" if x else "0" for x in b)
|
||
|
||
|
||
def fp_log2():
|
||
"""§7.5.6: log2 of 8·2^58·V(248, 29)/2^248, from an exact V."""
|
||
v = sum(math.comb(248, i) for i in range(30))
|
||
return "%.2f" % (3 + 58 + math.log2(v) - 248)
|
||
|
||
|
||
# ---------------------------------------------------------------- the checks
|
||
|
||
def check(doc, battery, scheme_bytes, frames, classes):
|
||
"""Failures of a corpus: every line names one."""
|
||
fails = []
|
||
|
||
def frame(h):
|
||
return frames.get(h)
|
||
|
||
for name, b in frames.items():
|
||
if sha(b).hex() != name:
|
||
fails.append("frame %s is not named by its SHA-256" % name[:16])
|
||
|
||
# §7.4: the definition.
|
||
if scheme_bytes != text_definition():
|
||
fails.append("scheme.json is not §7.4's definition")
|
||
if sha(scheme_bytes) != SV.SCHEME_DIGEST or doc["scheme"]["digest"] != SV.SCHEME_DIGEST.hex():
|
||
fails.append("the definition's digest is not §7.4's")
|
||
try:
|
||
d = json.loads(scheme_bytes)
|
||
want = {"amplitude": MARK.AMPLITUDE, "grid": MARK.GRID, "length": MARK.LENGTH, "cells_per_bit": MARK.CELLS_PER_BIT,
|
||
"margin": list(MARK.MARGIN), "min_width": MARK.MIN_SIDE, "min_height": MARK.MIN_SIDE, "code": [255, 71, MARK.T],
|
||
"max_frame_bytes": MARK.MAX_FRAME_BYTES, "class_max": MARK.CLASS_MAX, "orientations": 8, "scheme": MARK.SCHEME}
|
||
if any(d.get(k) != v for k, v in want.items()):
|
||
fails.append("the definition's parameters are not §7.5's")
|
||
except ValueError:
|
||
fails.append("scheme.json does not read")
|
||
|
||
# §7.5.3: the layout and the code.
|
||
g = "%x" % MARK.G
|
||
if g != text_generator() or doc["scheme"]["generator"] != g:
|
||
fails.append("the generator polynomial is not §7.5.3's")
|
||
if text_shuffle_head() != MARK.PERM[:4]:
|
||
fails.append("§7.5.3's first entries of the shuffle are not the stream's")
|
||
sc = doc["scheme"]
|
||
if (sc["shuffle_first"] != MARK.PERM[:16] or sc["shuffle_last"] != MARK.PERM[-16:] or sc["signs_first"] != MARK.SIGNS[:16]
|
||
or sc["mask"] != bits(MARK.MASK)):
|
||
fails.append("the corpus's layout is not the stream's")
|
||
if sc["false_positive_log2"] != fp_log2() or battery.get("false_positive_log2") != fp_log2():
|
||
fails.append("the false-positive bound is not §7.5.6's")
|
||
if MARK.decode(MARK.MASK) is not None or MARK.decode([1 - b for b in MARK.MASK]) is not None:
|
||
fails.append("the mask or its complement decodes")
|
||
for c in doc["codewords"]:
|
||
if bits(MARK.encode(bytes.fromhex(c["payload"]))) != c["codeword"]:
|
||
fails.append("codeword of %s" % c["payload"])
|
||
for x in doc["decodes"]:
|
||
got = MARK.decode([int(ch) for ch in x["word"]])
|
||
got = None if got is None else {"payload": got[0].hex(), "errors": got[1]}
|
||
if got != x["result"]:
|
||
fails.append("decoding of a word with %s bits flipped: %s" % (x["flipped"], got))
|
||
|
||
# §7.5.4, §7.5.5.
|
||
for e in doc["embeds"]:
|
||
u = frame(e["unmarked"])
|
||
if u is None:
|
||
fails.append("embed %s: no unmarked frames" % e["name"])
|
||
continue
|
||
m, amps = embed(u, e["width"], e["height"], e["frames"], bytes.fromhex(e["payload"]))
|
||
if m.hex() != e["marked"] or amps != e["amplitudes"]:
|
||
fails.append("embed %s: the marked frames or the amplitudes differ" % e["name"])
|
||
mk = frame(e["marked"])
|
||
if mk is None or sum((a - b) ** 2 for a, b in zip(u, mk)) != e["sse"]:
|
||
fails.append("embed %s: the marked frames are not held, or their SSE differs" % e["name"])
|
||
if any(sum(1 for a in f if int(a, 16) == MARK.AMPLITUDE) < s for f, s in zip(e["amplitudes"], e["short"])):
|
||
fails.append("embed %s: a bit left short is not at the cap" % e["name"])
|
||
for x in doc["detects"]:
|
||
px = frame(x["samples"])
|
||
got = None if px is None else detected_json(detect(px, x["width"], x["height"], x["frames"]))
|
||
if px is None or got != x["detected"]:
|
||
fails.append("detect %s: %s" % (x["name"], got))
|
||
|
||
# §7.5.7: the battery's sample, over the corpus's still.
|
||
still = next((e for e in doc["embeds"] if e["name"] == "still"), {"width": 0, "height": 0})
|
||
for x in doc["battery_sample"]:
|
||
px = frame(x["frames"])
|
||
if px is None:
|
||
fails.append("battery sample %s: no frames" % x["attack"])
|
||
continue
|
||
y, w, h = attacked(px, still["width"], still["height"], x["attack"], x["seed"])
|
||
got = detected_json(detect(y, w, h, 1))
|
||
if (sha(y).hex(), w, h, got) != (x["result_samples"], x["width"], x["height"], x["detected"]):
|
||
fails.append("battery sample %s of the %s still: %s" % (x["attack"], x["which"], got))
|
||
|
||
# §11 cases, with §7.6's checks.
|
||
for v in doc["cases"]:
|
||
case = dict(v, room=None)
|
||
try:
|
||
case["decoded"] = [{"object": d["object"], "samples": frame(d["frames"]).hex()} for d in v["decoded"]]
|
||
case["unmarked"] = [{"object": d["object"], "samples": frame(d["frames"]).hex()} for d in v["unmarked"]]
|
||
case["priors"] = [] # the scheme's cases hold no exact-upscale prior (synthesis §11.1)
|
||
got = SV.outcome(case, classes)
|
||
except Exception as e: # noqa: BLE001 — a malformed copy fails, whatever it breaks
|
||
got = {"outcome": "unreadable: %r" % (e,)}
|
||
if got != SV.stated(v):
|
||
fails.append("case %s: %s" % (v["name"], json.dumps(got)[:300]))
|
||
|
||
# §7.5.8: the table and the claims.
|
||
pictures, clips = text_tables()
|
||
rows = {}
|
||
for r in battery["table"]:
|
||
rows.setdefault(r["attack"], {})[r["set"]] = r
|
||
fp = {}
|
||
for u in battery["unmarked"]:
|
||
fp[(u["attack"], u["set"])] = u["detections"]
|
||
|
||
def cell(r):
|
||
return "%d / %d" % (r["detected"], r["trials"]) + (" (%d)" % r["errors_max"] if r["detected"] else "")
|
||
|
||
for a in battery["image_attacks"]:
|
||
if [cell(rows[a]["spectral"]), cell(rows[a]["leaves"])] != pictures.get(a, [None, None, None])[:2]:
|
||
fails.append("§7.5.8's row for %s is not battery.json's" % a)
|
||
for a in battery["clip_attacks"]:
|
||
if [cell(rows[a]["clips"])] != clips.get(a):
|
||
fails.append("§7.5.8's clip row for %s is not battery.json's" % a)
|
||
survived = {"pictures": [], "clips": []}
|
||
for kind, attacks, sets in (("pictures", battery["image_attacks"], ["spectral", "leaves"]), ("clips", battery["clip_attacks"], ["clips"])):
|
||
for a in attacks:
|
||
if all(rows[a][s]["detected"] == rows[a][s]["trials"] and fp[(a, s)] == 0 for s in sets):
|
||
survived[kind].append(a)
|
||
if survived != battery["claims"]:
|
||
fails.append("battery.json's claims are not the attacks its tallies show survived")
|
||
cp, cc = text_claims()
|
||
if cp is None or not set(cp) <= set(survived["pictures"]) or not set(cc) <= set(survived["clips"]):
|
||
fails.append("§7.5.8 claims an attack the measurement does not show survived")
|
||
for a in cp or []:
|
||
cells = pictures.get(a, [])
|
||
if len(cells) != 3 or any(not re.fullmatch(r"(\d+) / \1 \(\d+\)", c) for c in cells):
|
||
fails.append("§7.5.8 claims %s, which a column of its table does not show survived" % a)
|
||
if any(b["wrong"] for b in battery["table"]):
|
||
fails.append("a picture gave another payload")
|
||
fpt, words = text_figures()
|
||
w = battery["words"]
|
||
if fpt is None or "-" + fpt != fp_log2() or words is None or list(words) != [
|
||
w["ones_mean"], w["ones_sd"], "{:,}".format(w["pairs"]), w["distance_mean"], w["distance_sd"], str(w["distance_min"])]:
|
||
fails.append("§7.5.6's figures are not battery.json's")
|
||
return fails
|
||
|
||
|
||
# ================================================================ wai-mark:tent64s, §7.9
|
||
|
||
def text_definition_s():
|
||
"""§7.9.1: the definition, its lines joined with no white space."""
|
||
block = sec("#### 7.9.1 ").split("```")[1]
|
||
return "".join(line.strip() for line in block.strip().splitlines()).encode("utf-8")
|
||
|
||
|
||
def text_shuffle_head_s():
|
||
m = re.search(r"π begins ([0-9, ]+?), and the corpus", " ".join(sec("#### 7.9.2 ").split()))
|
||
return [int(x) for x in m.group(1).split(", ")] if m else None
|
||
|
||
|
||
def text_figures_s():
|
||
s = " ".join(sec("#### 7.9.5 ").split())
|
||
fp = re.search(r"2\^−([0-9.]+), about", s)
|
||
w = re.search(r"([0-9,]+) words have a mean of ([0-9.]+) ones with a standard deviation of ([0-9.]+), and the ([0-9,]+) pairs "
|
||
r"of those words from different pictures differ in a mean of ([0-9.]+) bits with a standard deviation of ([0-9.]+) "
|
||
r"and in no fewer than ([0-9]+);", s)
|
||
return (fp.group(1) if fp else None), (w.groups() if w else None)
|
||
|
||
|
||
def text_tables_s():
|
||
s = sec("#### 7.9.7 ")
|
||
pictures, clips = {}, {}
|
||
for line in s.splitlines():
|
||
cells = [c.strip() for c in line.strip().strip("|").split("|")]
|
||
if len(cells) == 4 and cells[0].startswith("`"):
|
||
pictures[cells[0].strip("`")] = cells[1:]
|
||
elif len(cells) == 2 and cells[0].startswith("`"):
|
||
clips[cells[0].strip("`")] = cells[1:]
|
||
return pictures, clips
|
||
|
||
|
||
def text_claims_s():
|
||
s = " ".join(sec("#### 7.9.7 ").split())
|
||
m = re.search(r"For pictures: (.*?)\. For clips: (.*?)\.", s)
|
||
if not m:
|
||
return None, None
|
||
return re.findall(r"`([^`]+)`", m.group(1)), re.findall(r"`([^`]+)`", m.group(2))
|
||
|
||
|
||
_DET_S = {}
|
||
|
||
|
||
def detect_s(px, w, h, n):
|
||
k = (sha(px), w, h, n)
|
||
if k not in _DET_S:
|
||
_DET_S[k] = MARK.detect_s(px, w, h, n)
|
||
return _DET_S[k]
|
||
|
||
|
||
_TRACE = {}
|
||
|
||
|
||
def trace_s(px, w, h, n):
|
||
k = (sha(px), w, h, n)
|
||
if k not in _TRACE:
|
||
_TRACE[k] = MARK.stage2_trace(px, w, h, n)
|
||
return _TRACE[k]
|
||
|
||
|
||
_DERIVED = {}
|
||
|
||
|
||
def derive(base, dv):
|
||
"""Frames a vector derives (§13): an attack of §7.9.6, or periodic, on each frame."""
|
||
k = (sha(base), dv["attack"], dv["seed"], dv["width"], dv["height"], dv["frames"])
|
||
if k not in _DERIVED:
|
||
_DERIVED[k] = MARK.attack_frames(dv["attack"], base, dv["width"], dv["height"], dv["frames"], dv["seed"])
|
||
return _DERIVED[k]
|
||
|
||
|
||
def detected_s_json(d):
|
||
if d is None:
|
||
return None
|
||
return {"payload": d[0].hex(), "stage": d[1], "orientation": d[2], "errors": d[3], "resync": d[4]}
|
||
|
||
|
||
def fp_log2_s():
|
||
"""§7.9.5: log2 of 12·2^58·V(248, 29)/2^248."""
|
||
v = sum(math.comb(248, i) for i in range(30))
|
||
return "%.2f" % (math.log2(12) + 58 + math.log2(v) - 248)
|
||
|
||
|
||
def check_s(doc, battery, scheme_bytes, frames, classes, photo_text=True):
|
||
"""Failures of the tent64s files."""
|
||
fails = []
|
||
|
||
def frame(h):
|
||
return frames.get(h)
|
||
|
||
if scheme_bytes != text_definition_s():
|
||
fails.append("scheme-tent64s.json is not §7.9.1's definition")
|
||
if sha(scheme_bytes) != SV.SCHEME_DIGEST_S or doc["scheme"]["digest"] != SV.SCHEME_DIGEST_S.hex():
|
||
fails.append("the tiled scheme's digest is not §7.9.1's")
|
||
try:
|
||
d = json.loads(scheme_bytes)
|
||
want = {"scheme": MARK.SCHEME_S, "tiles": [2, 2], "cells_per_tile_bit": MARK.PER_TILE, "cells_per_bit": 16,
|
||
"amplitude": MARK.AMPLITUDE, "margin": list(MARK.MARGIN), "grid": MARK.GRID, "length": MARK.LENGTH,
|
||
"resync": {"bins": [4, 8], "candidates": MARK.CANDIDATES, "decodes": MARK.DECODES, "frames": MARK.STAGE2_FRAMES,
|
||
"robust": [25, 5, MARK.ROBUST_CLAMP], "slope": list(MARK.SLOPE), "span": [list(MARK.SPAN[0]), list(MARK.SPAN[1])],
|
||
"subsample": MARK.SUB, "working_side": MARK.WORKING_SIDE}}
|
||
if any(d.get(k) != v for k, v in want.items()):
|
||
fails.append("the tiled definition's parameters are not §7.9's")
|
||
except ValueError:
|
||
fails.append("scheme-tent64s.json does not read")
|
||
sc = doc["scheme"]
|
||
if text_shuffle_head_s() != MARK.T_PERM[:4]:
|
||
fails.append("§7.9.2's first entries of the shuffle are not the stream's")
|
||
if (sc["shuffle_first"] != MARK.T_PERM[:16] or sc["shuffle_last"] != MARK.T_PERM[-16:] or sc["signs_first"] != MARK.T_SIGNS[:16]
|
||
or sc["mask"] != bits(MARK.T_MASK)):
|
||
fails.append("the tiled layout is not the stream's")
|
||
if sc["false_positive_log2"] != fp_log2_s() or battery.get("false_positive_log2") != fp_log2_s():
|
||
fails.append("the tiled bound is not §7.9.5's")
|
||
if MARK.decode(MARK.T_MASK) is not None or MARK.decode([1 - b for b in MARK.T_MASK]) is not None:
|
||
fails.append("the tiled mask or its complement decodes")
|
||
|
||
for e in doc["embeds"]:
|
||
u = frame(e["unmarked"])
|
||
if u is None:
|
||
fails.append("tent64s embed %s: no unmarked frames" % e["name"])
|
||
continue
|
||
m, amps = MARK.embed_s(u, e["width"], e["height"], e["frames"], bytes.fromhex(e["payload"]))
|
||
if sha(m).hex() != e["marked"] or ["".join("%x" % a for a in f) for f in amps] != e["amplitudes"]:
|
||
fails.append("tent64s embed %s: the marked frames or the amplitudes differ" % e["name"])
|
||
mk = frame(e["marked"])
|
||
if mk is None or sum((a - b) ** 2 for a, b in zip(u, mk)) != e["sse"]:
|
||
fails.append("tent64s embed %s: the marked frames are not held, or their SSE differs" % e["name"])
|
||
for x in doc["detects"]:
|
||
if "derive" in x:
|
||
base = frame(x["base"])
|
||
if base is None:
|
||
fails.append("tent64s detect %s: no base frames" % x["name"])
|
||
continue
|
||
px, w, h = derive(base, x["derive"])
|
||
dims = (w, h, x["derive"]["frames"])
|
||
else:
|
||
px = frame(x["samples"])
|
||
dims = (x["width"], x["height"], x["frames"])
|
||
if px is None or sha(px).hex() != x["samples"] or dims != (x["width"], x["height"], x["frames"]):
|
||
fails.append("tent64s detect %s: the frames are not the stated ones" % x["name"])
|
||
continue
|
||
got = detected_s_json(detect_s(px, *dims))
|
||
if got != x["detected"]:
|
||
fails.append("tent64s detect %s: %s" % (x["name"], json.dumps(got)[:200]))
|
||
tr = trace_s(px, *dims)
|
||
if tr != x["stage2"]:
|
||
fails.append("tent64s detect %s: stage 2 runs otherwise: %s" % (x["name"], json.dumps(tr)[:200]))
|
||
|
||
still = next((e for e in doc["embeds"] if e["name"] == "still"), {"width": 0, "height": 0})
|
||
for x in doc["battery_sample"]:
|
||
px = frame(x["frames"])
|
||
if px is None:
|
||
fails.append("tent64s battery sample %s: no frames" % x["attack"])
|
||
continue
|
||
y, w, h = attacked(px, still["width"], still["height"], x["attack"], x["seed"])
|
||
got = detected_s_json(detect_s(y, w, h, 1))
|
||
if (sha(y).hex(), w, h, got) != (x["result_samples"], x["width"], x["height"], x["detected"]):
|
||
fails.append("tent64s battery sample %s of the %s still: %s" % (x["attack"], x["which"], json.dumps(got)[:200]))
|
||
if [x["attack"] for x in doc["battery_sample"][:len(battery["image_attacks"])]] != battery["image_attacks"] \
|
||
or [x["seed"] for x in doc["battery_sample"][:len(battery["image_attacks"])]] != list(range(len(battery["image_attacks"]))):
|
||
fails.append("the tiled battery sample is not every attack of §7.9.6 at its index")
|
||
|
||
for v in doc["cases"]:
|
||
case = dict(v, room=None)
|
||
try:
|
||
dec = []
|
||
for d in v["decoded"]:
|
||
if "derive" in d:
|
||
px = derive(frame(d["frames"]), d["derive"])[0]
|
||
else:
|
||
px = frame(d["frames"])
|
||
dec.append({"object": d["object"], "samples": px.hex()})
|
||
case["decoded"] = dec
|
||
case["unmarked"] = [{"object": d["object"], "samples": frame(d["frames"]).hex()} for d in v["unmarked"]]
|
||
case["priors"] = [] # the tiled scheme's cases hold no exact-upscale prior (synthesis §11.1)
|
||
got = SV.outcome(case, classes)
|
||
except Exception as e: # noqa: BLE001 — a malformed copy fails, whatever it breaks
|
||
got = {"outcome": "unreadable: %r" % (e,)}
|
||
if got != SV.stated(v):
|
||
fails.append("tent64s case %s: %s" % (v["name"], json.dumps(got)[:300]))
|
||
|
||
pictures, clips = text_tables_s()
|
||
rows = {}
|
||
for r in battery["table"]:
|
||
rows.setdefault(r["attack"], {})[r["set"]] = r
|
||
fp = {(u["attack"], u["set"]): u["detections"] for u in battery["unmarked"]}
|
||
|
||
def cell(r):
|
||
return "%d / %d" % (r["detected"], r["trials"]) + (" (%d)" % r["errors_max"] if r["detected"] else "")
|
||
|
||
for a in battery["image_attacks"]:
|
||
if [cell(rows[a]["spectral"]), cell(rows[a]["leaves"])] != pictures.get(a, [None, None, None])[:2]:
|
||
fails.append("§7.9.7's row for %s is not battery-tent64s.json's" % a)
|
||
for a in battery["clip_attacks"]:
|
||
if [cell(rows[a]["clips"])] != clips.get(a):
|
||
fails.append("§7.9.7's clip row for %s is not battery-tent64s.json's" % a)
|
||
survived = {"pictures": [], "clips": []}
|
||
for kind, attacks, sets in (("pictures", battery["image_attacks"], ["spectral", "leaves"]), ("clips", battery["clip_attacks"], ["clips"])):
|
||
for a in attacks:
|
||
if all(rows[a][s_]["detected"] == rows[a][s_]["trials"] and fp[(a, s_)] == 0 for s_ in sets):
|
||
survived[kind].append(a)
|
||
if survived != battery["claims"]:
|
||
fails.append("battery-tent64s.json's claims are not the attacks its tallies show survived")
|
||
cp, cc = text_claims_s()
|
||
if cp is None or not set(cp) <= set(survived["pictures"]) or not set(cc) <= set(survived["clips"]):
|
||
fails.append("§7.9.7 claims an attack the measurement does not show survived")
|
||
for a in cp or []:
|
||
cells = pictures.get(a, [])
|
||
if len(cells) != 3 or any(not re.fullmatch(r"(\d+) / \1 \(\d+\)", c) for c in cells):
|
||
fails.append("§7.9.7 claims %s, which a column of its table does not show survived" % a)
|
||
if any(b["wrong"] for b in battery["table"]):
|
||
fails.append("a picture gave another payload under the tiled scheme")
|
||
fpt, words = text_figures_s()
|
||
w = battery["words"]
|
||
if fpt is None or "-" + fpt != fp_log2_s() or words is None or list(words) != [
|
||
str(w["words"]), w["ones_mean"], w["ones_sd"], "{:,}".format(w["pairs"]), w["distance_mean"], w["distance_sd"], str(w["distance_min"])]:
|
||
fails.append("§7.9.5's figures are not battery-tent64s.json's")
|
||
return fails
|
||
|
||
|
||
def mutations_s(doc, battery, scheme_bytes, frames):
|
||
out = []
|
||
|
||
def mutate(label, f):
|
||
d, b, s, fr = copy.deepcopy(doc), copy.deepcopy(battery), scheme_bytes, dict(frames)
|
||
r = f(d, b, fr)
|
||
if r is not None:
|
||
s = r
|
||
out.append((label, d, b, s, fr))
|
||
|
||
vn = {v["name"]: k for k, v in enumerate(doc["cases"])}
|
||
dn = {x["name"]: k for k, x in enumerate(doc["detects"])}
|
||
marked = doc["embeds"][0]["marked"]
|
||
|
||
def flip_frame(d, b, fr):
|
||
x = bytearray(fr[marked])
|
||
x[3000] ^= 4
|
||
fr[marked] = bytes(x)
|
||
|
||
def agreement(d, b, fr):
|
||
x = d["detects"][dn["clip marked, cropped 10% from the left"]]["detected"]["resync"]
|
||
x["agreement"] += 1
|
||
|
||
def lattice(d, b, fr):
|
||
x = d["detects"][dn["clip marked, turned 1 degree"]]["detected"]["resync"]
|
||
x["lattice"][1] += 1
|
||
|
||
def other_attack(d, b, fr):
|
||
d["cases"][vn["crop_agrees_s"]]["decoded"][0]["derive"]["attack"] = "crop-right-10%"
|
||
|
||
def forged_reads(d, b, fr):
|
||
d["detects"][dn["still unmarked, tiled by its quarter"]]["detected"] = d["detects"][dn["still marked"]]["detected"]
|
||
|
||
def beyond_accept(d, b, fr):
|
||
v = d["cases"][vn["beyond_range_s"]]
|
||
v.pop("reason")
|
||
v.pop("at")
|
||
v["outcome"] = "accept"
|
||
|
||
def sample_stage(d, b, fr):
|
||
x = next(x for x in d["battery_sample"] if x["attack"] == "rotate-5" and x["which"] == "marked")
|
||
x["detected"]["stage"] = 1
|
||
|
||
def tally(d, b, fr):
|
||
r = next(r for r in b["table"] if r["attack"] == "crop-left-25%" and r["set"] == "leaves")
|
||
r["detected"] -= 1
|
||
r["missed"] += 1
|
||
|
||
mutate("a byte of the still the tiled scheme marked", flip_frame)
|
||
mutate("scheme-tent64s.json with one more space", lambda d, b, fr: scheme_bytes + b" ")
|
||
mutate("the tiled mask's last bit", lambda d, b, fr: d["scheme"].update(mask=flip(d["scheme"]["mask"], 247)))
|
||
mutate("the cropped clip's agreement count", agreement)
|
||
mutate("the turned clip's lattice by 1/64 of a sample", lattice)
|
||
mutate("the crop case derived by another crop", other_attack)
|
||
mutate("the forged lattice stated as read", forged_reads)
|
||
mutate("the crop beyond reach stated as accepted", beyond_accept)
|
||
mutate("a rotation of 5 degrees stated as read in place", sample_stage)
|
||
mutate("a marked leaves picture missed at crop-left-25%", tally)
|
||
mutate("the tiled words' mean", lambda d, b, fr: b["words"].update(ones_mean="124.15"))
|
||
|
||
def phase(d, b, fr):
|
||
c = d["detects"][dn["photo unmarked"]]["stage2"]["candidates"][-1]
|
||
c["phase"][0] = (c["phase"][0] + 1) % c["bins"][0]
|
||
|
||
mutate("a candidate's phase in the unmarked photograph's trace", phase)
|
||
return out
|
||
|
||
|
||
def flip(s, at):
|
||
return s[:at] + ("1" if s[at] == "0" else "0") + s[at + 1:]
|
||
|
||
|
||
def mutations(doc, battery, scheme_bytes, frames):
|
||
"""Copies of the corpus, each with one defect a conforming verifier must catch."""
|
||
out = []
|
||
|
||
def mutate(label, f):
|
||
d, b, s, fr = copy.deepcopy(doc), copy.deepcopy(battery), scheme_bytes, dict(frames)
|
||
r = f(d, b, fr)
|
||
if r is not None:
|
||
s = r
|
||
out.append((label, d, b, s, fr))
|
||
|
||
vn = {v["name"]: k for k, v in enumerate(doc["cases"])}
|
||
still_marked = doc["embeds"][0]["marked"]
|
||
|
||
def flip_frame(d, b, fr):
|
||
x = bytearray(fr[still_marked])
|
||
x[100] ^= 1
|
||
fr[still_marked] = bytes(x)
|
||
|
||
def swap_unmarked(d, b, fr):
|
||
d["cases"][vn["still_proved"]]["unmarked"][0]["frames"] = d["embeds"][1]["unmarked"]
|
||
|
||
def tally(d, b, fr):
|
||
r = next(r for r in b["table"] if r["attack"] == "int-motion-q16" and r["set"] == "leaves")
|
||
r["detected"] -= 1
|
||
r["missed"] += 1
|
||
|
||
def unmarked_hit(d, b, fr):
|
||
next(u for u in b["unmarked"] if u["attack"] == "scale-1/2" and u["set"] == "spectral")["detections"] = 1
|
||
|
||
def inherited_as_agree(d, b, fr):
|
||
d["cases"][vn["copy_inherits"]]["statements"]["detections"][0]["result"] = "agree"
|
||
|
||
mutate("a byte of the marked still's frame file", flip_frame)
|
||
mutate("scheme.json with one more space", lambda d, b, fr: scheme_bytes + b" ")
|
||
mutate("the mask's first bit", lambda d, b, fr: d["scheme"].update(mask=flip(d["scheme"]["mask"], 0)))
|
||
mutate("a codeword's parity bit", lambda d, b, fr: d["codewords"][2].update(codeword=flip(d["codewords"][2]["codeword"], 200)))
|
||
mutate("a decoding of 29 errors stated as none", lambda d, b, fr: next(x for x in d["decodes"] if x["flipped"] == 29).update(result=None))
|
||
mutate("the still's marked digest", lambda d, b, fr: d["embeds"][0].update(marked=doc["embeds"][1]["marked"]))
|
||
mutate("one bit's amplitude", lambda d, b, fr: d["embeds"][0].update(amplitudes=[("1" if doc["embeds"][0]["amplitudes"][0][0] != "1" else "2")
|
||
+ doc["embeds"][0]["amplitudes"][0][1:]]))
|
||
mutate("the clip's detection stated in another orientation", lambda d, b, fr: d["detects"][2]["detected"].update(orientation=1))
|
||
mutate("a crop's sample reading stated as a mark",
|
||
lambda d, b, fr: next(x for x in d["battery_sample"] if x["attack"] == "crop-10%" and x["which"] == "marked").update(
|
||
detected=d["detects"][0]["detected"]))
|
||
mutate("the proved still's unmarked frame swapped for the photograph's", swap_unmarked)
|
||
mutate("the copy's inherited agreement stated as plain agreement", inherited_as_agree)
|
||
mutate("a marked leaves picture missed at int-motion-q16", tally)
|
||
mutate("an unmarked spectral picture read at scale-1/2", unmarked_hit)
|
||
mutate("the words' least distance", lambda d, b, fr: b["words"].update(distance_min=b["words"]["distance_min"] + 1))
|
||
return out
|
||
|
||
|
||
def weakened():
|
||
"""The verifier with one of §7.6's checks, or §7.3's inheritance, dropped."""
|
||
return [
|
||
("the marking not recomputed", "recomputes", lambda *a: True),
|
||
("the detection not run again", "reproduces", lambda *a: True),
|
||
("no agreement by inheritance", "inherits", lambda c, p, by_link: False),
|
||
]
|
||
|
||
|
||
def main(corpus_dir):
|
||
fails = []
|
||
with open(os.path.join(HERE, "..", "registry", "capabilities.json"), encoding="utf-8") as f:
|
||
classes = {c["capability"]: c["replicate"] for c in json.load(f)["capabilities"]}
|
||
with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as f:
|
||
doc = json.load(f)
|
||
with open(os.path.join(corpus_dir, "battery.json"), encoding="utf-8") as f:
|
||
battery = json.load(f)
|
||
with open(os.path.join(corpus_dir, "scheme.json"), "rb") as f:
|
||
scheme_bytes = f.read()
|
||
frames = {}
|
||
for n in sorted(os.listdir(os.path.join(corpus_dir, "frames"))):
|
||
with open(os.path.join(corpus_dir, "frames", n), "rb") as f:
|
||
frames[n[:-len(".rgb")]] = f.read()
|
||
|
||
base = check(doc, battery, scheme_bytes, frames, classes)
|
||
for b in base:
|
||
print("FAIL " + b)
|
||
print(("PASS" if not base else "FAIL") + " the scheme, the code, %d embeddings, %d detections, %d battery samples and %d cases"
|
||
% (len(doc["embeds"]), len(doc["detects"]), len(doc["battery_sample"]), len(doc["cases"])))
|
||
fails += base
|
||
seen = {v.get("reason") for v in doc["cases"]}
|
||
missing = [r for r in SV.MARK_ONLY if r not in seen]
|
||
print(("PASS" if not missing and len(SV.MARK_ONLY) == 3 else "FAIL") + " every code only this corpus gives (§13) refuses a case: "
|
||
+ ", ".join(SV.MARK_ONLY))
|
||
if missing or len(SV.MARK_ONLY) != 3:
|
||
fails.append("codes no case gives: %s" % missing)
|
||
for label, d, b, s, fr in mutations(doc, battery, scheme_bytes, frames):
|
||
caught = check(d, b, s, fr, classes)
|
||
print(("PASS" if caught else "FAIL") + " mutated copy fails: %s%s" % (label, " (%s)" % caught[0][:100] if caught else ""))
|
||
if not caught:
|
||
fails.append("a mutated copy verifies: %s" % label)
|
||
for label, name, fn in weakened():
|
||
saved = getattr(SV, name)
|
||
setattr(SV, name, fn)
|
||
try:
|
||
caught = check(doc, battery, scheme_bytes, frames, classes)
|
||
finally:
|
||
setattr(SV, name, saved)
|
||
print(("PASS" if caught else "FAIL") + " the corpus catches a verifier with %s%s" % (label, " (%s)" % caught[0][:100] if caught else ""))
|
||
if not caught:
|
||
fails.append("a weakened verifier passes: %s" % label)
|
||
# wai-mark:tent64s (§7.9), beside tent64's and sharing its frames.
|
||
with open(os.path.join(corpus_dir, "vectors-tent64s.json"), encoding="utf-8") as f:
|
||
doc_s = json.load(f)
|
||
with open(os.path.join(corpus_dir, "battery-tent64s.json"), encoding="utf-8") as f:
|
||
battery_s = json.load(f)
|
||
with open(os.path.join(corpus_dir, "scheme-tent64s.json"), "rb") as f:
|
||
scheme_s = f.read()
|
||
base_s = check_s(doc_s, battery_s, scheme_s, frames, classes)
|
||
for b in base_s:
|
||
print("FAIL " + b)
|
||
print(("PASS" if not base_s else "FAIL") + " the tiled scheme: %d embeddings, %d detections, %d battery samples and %d cases"
|
||
% (len(doc_s["embeds"]), len(doc_s["detects"]), len(doc_s["battery_sample"]), len(doc_s["cases"])))
|
||
fails += base_s
|
||
for label, d, b, s, fr in mutations_s(doc_s, battery_s, scheme_s, frames):
|
||
caught = check_s(d, b, s, fr, classes)
|
||
print(("PASS" if caught else "FAIL") + " mutated tiled copy fails: %s%s" % (label, " (%s)" % caught[0][:100] if caught else ""))
|
||
if not caught:
|
||
fails.append("a mutated tiled copy verifies: %s" % label)
|
||
for label, name, fn in weakened():
|
||
saved = getattr(SV, name)
|
||
setattr(SV, name, fn)
|
||
try:
|
||
caught = check_s(doc_s, battery_s, scheme_s, frames, classes)
|
||
finally:
|
||
setattr(SV, name, saved)
|
||
print(("PASS" if caught else "FAIL") + " the tiled corpus catches a verifier with %s%s" % (label, " (%s)" % caught[0][:100] if caught else ""))
|
||
if not caught:
|
||
fails.append("a weakened verifier passes the tiled corpus: %s" % label)
|
||
for f in fails:
|
||
print("FAIL " + f)
|
||
print(("PASS" if not fails else "FAIL") + " synthesis-mark corpus")
|
||
return 1 if fails else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
if len(sys.argv) != 2:
|
||
print("usage: synthesis_mark_verify.py <live-conformance/synthesis-mark>", file=sys.stderr)
|
||
sys.exit(2)
|
||
sys.exit(main(sys.argv[1]))
|