df1004f187
Resolved by meaning: synthesis::Case holds both the priors (exact-upscale) and the unmarked frames (§7.6); synthesis §11.1 lists both inputs and the three policy members; the synthesis-mark corpus's cases hold no prior; exact-upscale passes synthesis's second_layer policy through; the live-conformance table gains the upscale row in its generator column; both CI steps and both CHANGELOG entries are kept. Generated files are main's here and are regenerated in the next commit. Co-Authored-By: Claude Opus 5.5 <[email protected]>
404 lines
18 KiB
Python
404 lines
18 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
|
||
|
||
|
||
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 px, e, payload: True),
|
||
("the detection not run again", "reproduces", lambda px, r, payload: 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)
|
||
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]))
|