7f08895096
§7.4 registers wai-mark:tent64 by its definition and digest and keeps wai-mark:conformance as the scheme with no algorithm. §7.5 is the scheme: frames and luma, the grid and the tent pattern, the stream, the shuffle, the signs and the mask, the shortened BCH code, the informed embedding, the detector in eight orientations, what it guarantees (no other channel, a PSNR floor, a false-positive bound derived and checked), the battery and the table of what it survives, with the claims it makes. §7.6 proves a marking by recomputation and reproduces a detection; §7.7 gives the signpost in the content and in the C2PA statement, and a well-known resolver; §7.8 has a sink that generates mark as part of the reconstruction it is held to. _wai_mark.py is the scheme and the battery in Python from the text; synthesis_verify.py runs §7.6's checks with it; synthesis_mark_verify.py verifies the scheme's corpus, mutates it and weakens itself in turn. Co-Authored-By: Claude Opus 5.5 <[email protected]>
1005 lines
43 KiB
Python
1005 lines
43 KiB
Python
#!/usr/bin/env python3
|
||
"""Second implementation of extensions/synthesis.md (`wai.media.synthesis`), over the corpus
|
||
`live-conformance/synthesis/`.
|
||
|
||
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), 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), runs the two pinned derivations
|
||
of §4 in its own arithmetic, writes the digital source type of §5 and the C2PA statement of §6
|
||
itself, and verifies each room with the live-rooms corpus's second implementation
|
||
(live_rooms_verify.py). It reads §10.2's pinned icons and §12's codes from the extension.
|
||
|
||
python3 -I synthesis_verify.py <live-conformance/synthesis>
|
||
|
||
It gives every vector of the corpus the outcome, the reason, the place and the statements the
|
||
corpus states, checks that every reason of §12 refuses a vector, and then mutates copies of the
|
||
corpus and requires every copy to fail. Exit status 0 only if all of that holds. Python 3.8 or
|
||
later; standard library only.
|
||
"""
|
||
|
||
import copy
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import unicodedata
|
||
|
||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
sys.path.insert(0, HERE)
|
||
sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance"))
|
||
import live_events_verify as LEV # noqa: E402
|
||
import live_rooms_verify as LRV # noqa: E402
|
||
import _wai_mark as MARK # noqa: E402
|
||
import _ed25519 as ED # noqa: E402
|
||
from _blake3 import blake3 # noqa: E402
|
||
|
||
EXTENSION = os.path.join(HERE, "..", "extensions", "synthesis.md")
|
||
REGISTRY = os.path.join(HERE, "..", "registry", "capabilities.json")
|
||
|
||
SIGN_DOMAIN = b"wai:synthesis\x01" # §2
|
||
LINK_DOMAIN = b"wai:synthesis-link\x01" # §2
|
||
OBJECT_DOMAIN = b"moq-jwp:object\x01" # jwp-receipts §2.1
|
||
IPTC = "http://cv.iptc.org/newscodes/digitalsourcetype/"
|
||
MODEL_ID = "tag:transaction.science,2026:wai/model/" # §6
|
||
UPSCALE2, DENOISE3 = "wai-synthesis:upscale2", "wai-synthesis:denoise3"
|
||
MAX_INT = (1 << 53) - 1
|
||
|
||
CLASSES = ["captured", "reconstructed", "assistive", "synthesized"] # §3.2, in order
|
||
STEP_CLASS = {"capture": "captured", "convert": "reconstructed", "assist": "assistive", "generate": "synthesized"}
|
||
CAPTURE_SOURCES = ["digitalCapture", "computationalCapture", "screenCapture", "digitalCreation"]
|
||
SOURCES = CAPTURE_SOURCES + ["algorithmicallyEnhanced", "trainedAlgorithmicMedia", "compositeWithTrainedAlgorithmicMedia",
|
||
"algorithmicMedia", "composite", "compositeCapture", "compositeSynthetic"] # §5
|
||
OVERSIGHT = ["fully_autonomous", "prompt_guided", "human_validated"]
|
||
MEDIA = ["audio", "video", "image", "text"]
|
||
PATH_TIERS = ["decode-equivalence", "generation-equivalence", "entropy-consistency", "none"]
|
||
REVIEW_TIERS = ["decode-equivalence", "generation-equivalence"]
|
||
|
||
# Unicode's White_Space property (a term holds none of these, jwp-receipts §2.4.3).
|
||
WHITE_SPACE = set("\t\n\x0b\x0c\r \x85\xa0 "
|
||
"
")
|
||
|
||
|
||
class Refuse(Exception):
|
||
def __init__(self, reason, at=None):
|
||
super().__init__(reason)
|
||
self.reason, self.at = reason, at
|
||
|
||
|
||
class Bad(Exception):
|
||
"""A record that does not read: `record_malformed`."""
|
||
|
||
|
||
# ---------------------------------------------------------------- §2 records
|
||
|
||
def strict_json(text):
|
||
"""RFC 8259, no member named twice, no number but an unsigned integer, no unpaired
|
||
surrogate, at most 32 levels."""
|
||
def pairs(ps):
|
||
d = {}
|
||
for k, v in ps:
|
||
if k in d:
|
||
raise Bad("a member named twice")
|
||
d[k] = v
|
||
return d
|
||
|
||
def no_float(s):
|
||
raise Bad("a number that is not an unsigned integer")
|
||
|
||
def integer(s):
|
||
if s.startswith("-"):
|
||
raise Bad("a negative number")
|
||
return int(s)
|
||
|
||
try:
|
||
v = json.loads(text, object_pairs_hook=pairs, parse_float=no_float, parse_int=integer, parse_constant=no_float)
|
||
except (ValueError, RecursionError):
|
||
raise Bad("not JSON")
|
||
|
||
def walk(x, depth):
|
||
if isinstance(x, (dict, list)):
|
||
if depth > 32:
|
||
raise Bad("too deep")
|
||
for y in (x.values() if isinstance(x, dict) else x):
|
||
walk(y, depth + 1)
|
||
if isinstance(x, dict):
|
||
for k in x:
|
||
walk(k, depth)
|
||
elif isinstance(x, str):
|
||
if any(0xD800 <= ord(c) <= 0xDFFF for c in x):
|
||
raise Bad("an unpaired surrogate")
|
||
elif isinstance(x, bool):
|
||
raise Bad("a boolean")
|
||
elif isinstance(x, int):
|
||
if x > MAX_INT:
|
||
raise Bad("an integer past 2^53 - 1")
|
||
walk(v, 1)
|
||
return v
|
||
|
||
|
||
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 on the values a record carries."""
|
||
if v is None:
|
||
return "null"
|
||
if isinstance(v, bool):
|
||
raise Bad("a boolean")
|
||
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 need(cond):
|
||
if not cond:
|
||
raise Bad("a member breaks its rule")
|
||
|
||
|
||
def hexb(v, n):
|
||
need(isinstance(v, str) and len(v) == 2 * n and re.fullmatch(r"[0-9a-f]*", v) is not None)
|
||
return bytes.fromhex(v)
|
||
|
||
|
||
def opt_hex(v, n):
|
||
return None if v is None else hexb(v, n)
|
||
|
||
|
||
def integer(v):
|
||
need(isinstance(v, int) and not isinstance(v, bool) and 0 <= v <= MAX_INT)
|
||
return v
|
||
|
||
|
||
def term(v):
|
||
need(isinstance(v, str) and 0 < len(v.encode("utf-8")) <= 2048
|
||
and not any(c in WHITE_SPACE or unicodedata.category(c) == "Cc" for c in v))
|
||
return v
|
||
|
||
|
||
def opt_term(v):
|
||
return None if v is None else term(v)
|
||
|
||
|
||
def one_of(v, values):
|
||
need(isinstance(v, str) and v in values)
|
||
return v
|
||
|
||
|
||
def members(v, names):
|
||
need(isinstance(v, dict) and set(v) == set(names))
|
||
return v
|
||
|
||
|
||
def date_time(s):
|
||
"""RFC 3339 §5.6 date-time, with §5.7's days (jwp-receipts §2.4.3)."""
|
||
m = re.fullmatch(r"([0-9]{4})-([0-9]{2})-([0-9]{2})[Tt]([0-9]{2}):([0-9]{2}):([0-9]{2})(\.[0-9]+)?([Zz]|[+-]([0-9]{2}):([0-9]{2}))", s or "")
|
||
need(m is not None)
|
||
y, mo, d, h, mi, sec = (int(m.group(i)) for i in range(1, 7))
|
||
leap = y % 4 == 0 and (y % 100 != 0 or y % 400 == 0)
|
||
days = [31, 29 if leap else 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31]
|
||
need(1 <= mo <= 12 and 1 <= d <= days[mo - 1] and h <= 23 and mi <= 59 and sec <= 60)
|
||
if m.group(9) is not None:
|
||
need(int(m.group(9)) <= 23 and int(m.group(10)) <= 59)
|
||
|
||
|
||
def read(text, kind, names, fields):
|
||
"""§2: read one record, then its signature. `fields` reads the kind's own members."""
|
||
try:
|
||
v = strict_json(text)
|
||
members(v, names)
|
||
need(v["kind"] == kind)
|
||
signer = hexb(v["signer_pubkey"], 32)
|
||
sig = hexb(v["sig"], 64)
|
||
if "issued_at" in v:
|
||
date_time(v["issued_at"])
|
||
r = fields(v)
|
||
except (Bad, KeyError, TypeError, AttributeError):
|
||
raise Refuse("record_malformed")
|
||
bad = ED.verify(signer, signing_bytes(v), sig)
|
||
if bad:
|
||
raise Refuse(bad)
|
||
r.update(value=v, signer=signer, link=link_hash(v))
|
||
return r
|
||
|
||
|
||
# ---------------------------------------------------------------- §3 the origin claim
|
||
|
||
ORIGIN = ["kind", "subject", "media", "class", "step", "operation", "record", "source", "models", "oversight", "inputs",
|
||
"evidence", "issued_at", "signer_pubkey", "sig"]
|
||
|
||
|
||
def subject_of(v):
|
||
t = v.get("type")
|
||
if t == "object":
|
||
members(v, ["type", "content_hash"])
|
||
return ("object", hexb(v["content_hash"], 32))
|
||
if t == "group":
|
||
members(v, ["type", "receipt"])
|
||
return ("group", hexb(v["receipt"], 32))
|
||
if t == "track":
|
||
members(v, ["type", "track", "from_group"])
|
||
integer(v["from_group"])
|
||
return ("track", hexb(v["track"], 32))
|
||
raise Bad("a subject of no type")
|
||
|
||
|
||
def origin_fields(v):
|
||
c = {"subject": subject_of(v["subject"]), "media": one_of(v["media"], MEDIA), "class": one_of(v["class"], CLASSES),
|
||
"step": one_of(v["step"], list(STEP_CLASS)), "operation": term(v["operation"]), "record": opt_hex(v["record"], 32),
|
||
"source": one_of(v["source"], SOURCES),
|
||
"oversight": None if v["oversight"] is None else one_of(v["oversight"], OVERSIGHT)}
|
||
need(isinstance(v["models"], list) and len(v["models"]) <= 16)
|
||
c["models"] = []
|
||
for m in v["models"]:
|
||
members(m, ["digest", "type"])
|
||
c["models"].append({"digest": hexb(m["digest"], 32), "trained": one_of(m["type"], ["trained", "algorithmic"]) == "trained"})
|
||
need(len({m["digest"] for m in c["models"]}) == len(c["models"]))
|
||
need(isinstance(v["inputs"], list) and len(v["inputs"]) <= 16)
|
||
c["inputs"] = []
|
||
for i in v["inputs"]:
|
||
members(i, ["content_hash", "origin", "class", "source"])
|
||
c["inputs"].append({"content_hash": hexb(i["content_hash"], 32), "origin": opt_hex(i["origin"], 32),
|
||
"class": one_of(i["class"], CLASSES), "source": one_of(i["source"], SOURCES)})
|
||
need(len({i["content_hash"] for i in c["inputs"]}) == len(c["inputs"]))
|
||
e = v["evidence"]
|
||
if e is None:
|
||
c["evidence"] = None
|
||
else:
|
||
members(e, ["tier", "derivation", "params", "width", "height", "input_samples", "output_samples"])
|
||
need(isinstance(e["params"], dict))
|
||
c["evidence"] = {"tier": one_of(e["tier"], ["replicate", "attest"]), "derivation": term(e["derivation"]),
|
||
"params": e["params"], "width": integer(e["width"]), "height": integer(e["height"]),
|
||
"input_samples": hexb(e["input_samples"], 32), "output_samples": hexb(e["output_samples"], 32)}
|
||
need(c["evidence"]["width"] >= 1 and c["evidence"]["height"] >= 1)
|
||
n, m, s = len(c["inputs"]), len(c["models"]), c["step"]
|
||
plain = c["oversight"] is None and c["evidence"] is None
|
||
if s == "capture":
|
||
need(n == 0 and m == 0 and plain and c["source"] in CAPTURE_SOURCES)
|
||
elif s == "convert":
|
||
need(n >= 1 and m == 0 and plain)
|
||
elif s == "assist":
|
||
need(n == 1 and c["oversight"] is None and c["evidence"] is not None)
|
||
else:
|
||
need(m >= 1 and c["oversight"] is not None and c["evidence"] is None)
|
||
return c
|
||
|
||
|
||
def class_of(c):
|
||
"""§3.2: the greatest of the step's class and the inputs'."""
|
||
return max([STEP_CLASS[c["step"]]] + [i["class"] for i in c["inputs"]], key=CLASSES.index)
|
||
|
||
|
||
def generated_source(c):
|
||
"""§5."""
|
||
if not any(m["trained"] for m in c["models"]):
|
||
return "algorithmicMedia"
|
||
if c["oversight"] == "human_validated" or any(i["class"] != "synthesized" for i in c["inputs"]):
|
||
return "compositeWithTrainedAlgorithmicMedia"
|
||
return "trainedAlgorithmicMedia"
|
||
|
||
|
||
def source_of(c):
|
||
"""§5's table."""
|
||
ins = c["inputs"]
|
||
if c["step"] == "capture":
|
||
return c["source"]
|
||
if c["step"] == "convert":
|
||
if len(ins) == 1:
|
||
return ins[0]["source"]
|
||
if any(i["class"] == "synthesized" for i in ins):
|
||
return "compositeSynthetic"
|
||
if all(i["source"] in ("digitalCapture", "computationalCapture") for i in ins):
|
||
return "compositeCapture"
|
||
return "composite"
|
||
if c["step"] == "assist":
|
||
return ins[0]["source"] if ins[0]["class"] == "synthesized" else "algorithmicallyEnhanced"
|
||
return generated_source(c)
|
||
|
||
|
||
def statement(c):
|
||
"""§6, as its JCS text."""
|
||
if len(c["inputs"]) != 1:
|
||
actions = [{"action": "c2pa.created", "digitalSourceType": IPTC + c["source"]}]
|
||
elif c["step"] == "convert":
|
||
actions = [{"action": "c2pa.opened"}, {"action": "c2pa.converted"}]
|
||
elif c["step"] == "assist":
|
||
actions = [{"action": "c2pa.opened"}, {"action": "c2pa.edited", "digitalSourceType": IPTC + "algorithmicallyEnhanced"}]
|
||
else:
|
||
actions = [{"action": "c2pa.opened"}, {"action": "c2pa.edited", "digitalSourceType": IPTC + generated_source(c)}]
|
||
disclosures = []
|
||
for m in [m for m in c["models"] if m["trained"]]:
|
||
data = {"modelType": "c2pa.types.model", "modelIdentifier": MODEL_ID + m["digest"].hex()}
|
||
if c["step"] == "generate":
|
||
data["contentProfile"] = {"humanOversightLevel": c["oversight"]}
|
||
k = len(disclosures)
|
||
disclosures.append({"label": "c2pa.ai-disclosure" if k == 0 else "c2pa.ai-disclosure__%d" % k, "data": data})
|
||
return jcs({"actions": actions, "assertions": disclosures})
|
||
|
||
|
||
def payload_of(c):
|
||
"""§7.2: the class's code, then seven bytes of the claim's link hash."""
|
||
return bytes([CLASSES.index(c["class"])]) + c["link"][:7]
|
||
|
||
|
||
# ---------------------------------------------------------------- §4 the derivations
|
||
|
||
def upscale2(px, w, h):
|
||
out = bytearray(4 * w * h * 3)
|
||
ow = 2 * w
|
||
|
||
def at(x, y, k):
|
||
return px[(y * w + x) * 3 + k]
|
||
for y in range(h):
|
||
for x in range(w):
|
||
x1, y1 = min(x + 1, w - 1), min(y + 1, h - 1)
|
||
for k in range(3):
|
||
a, b, c, d = at(x, y, k), at(x1, y, k), at(x, y1, k), at(x1, y1, k)
|
||
out[((2 * y) * ow + 2 * x) * 3 + k] = a
|
||
out[((2 * y) * ow + 2 * x + 1) * 3 + k] = (a + b + 1) >> 1
|
||
out[((2 * y + 1) * ow + 2 * x) * 3 + k] = (a + c + 1) >> 1
|
||
out[((2 * y + 1) * ow + 2 * x + 1) * 3 + k] = (a + b + c + d + 2) >> 2
|
||
return bytes(out)
|
||
|
||
|
||
def denoise3(px, w, h):
|
||
out = bytearray(w * h * 3)
|
||
for y in range(h):
|
||
for x in range(w):
|
||
for k in range(3):
|
||
s = 8
|
||
for j in (-1, 0, 1):
|
||
for i in (-1, 0, 1):
|
||
weight = (2 if i == 0 else 1) * (2 if j == 0 else 1)
|
||
xx, yy = min(max(x + i, 0), w - 1), min(max(y + j, 0), h - 1)
|
||
s += weight * px[(yy * w + xx) * 3 + k]
|
||
out[(y * w + x) * 3 + k] = s >> 4
|
||
return bytes(out)
|
||
|
||
|
||
DERIVATIONS = {UPSCALE2: (upscale2, 4), DENOISE3: (denoise3, 1)}
|
||
|
||
|
||
# ---------------------------------------------------------------- §7, §9, §10 the other records
|
||
|
||
def positive(v):
|
||
need(integer(v) >= 1)
|
||
return v
|
||
|
||
|
||
def frames_of(v, digests):
|
||
"""§7.2, §7.3: an embedding or a read: the frames' count and size, and digests."""
|
||
if v is None:
|
||
return None
|
||
members(v, ["frames", "width", "height"] + digests)
|
||
f = {k: positive(v[k]) for k in ("frames", "width", "height")}
|
||
f.update({k: hexb(v[k], 32) for k in digests})
|
||
return f
|
||
|
||
|
||
def marking_fields(v):
|
||
return {"subject": hexb(v["subject"], 32), "origin": hexb(v["origin"], 32), "scheme": term(v["scheme"]),
|
||
"scheme_digest": hexb(v["scheme_digest"], 32), "payload": hexb(v["payload"], 8),
|
||
"embedding": frames_of(v["embedding"], ["unmarked", "marked"])}
|
||
|
||
|
||
def detection_fields(v):
|
||
return {"subject": hexb(v["subject"], 32), "scheme": term(v["scheme"]), "scheme_digest": hexb(v["scheme_digest"], 32),
|
||
"payload": opt_hex(v["payload"], 8), "read": frames_of(v["read"], ["samples"])}
|
||
|
||
|
||
def scheme_check(scheme, digest, frames):
|
||
"""§7.4: whether the record's scheme is the registered one; refuses a registered term under
|
||
another digest, a record of it with no frames the scheme admits, and frames stated for
|
||
wai-mark:conformance."""
|
||
if scheme == MARK.SCHEME:
|
||
if digest != SCHEME_DIGEST:
|
||
raise Refuse("scheme_mismatch")
|
||
if frames is None or not MARK.admits(frames["width"], frames["height"], frames["frames"]):
|
||
raise Refuse("mark_evidence_invalid")
|
||
return True
|
||
if scheme == "wai-mark:conformance" and frames is not None:
|
||
raise Refuse("mark_evidence_invalid")
|
||
return False
|
||
|
||
|
||
def marking_statement(m):
|
||
"""§7.7: the marking's C2PA statement, as its JCS text."""
|
||
return jcs({"actions": [{"action": "c2pa.watermarked.bound"}],
|
||
"assertions": [{"label": "c2pa.soft-binding",
|
||
"data": {"alg": m["scheme"], "alg-params": m["scheme_digest"].hex(),
|
||
"blocks": [{"scope": {}, "value": m["payload"].hex()}]}}]})
|
||
|
||
|
||
_EMBEDDED, _DETECTED = {}, {}
|
||
|
||
|
||
def embedded(px, e, payload):
|
||
"""§7.5.4, memoised: the SHA-256 of the marked frames."""
|
||
k = (hashlib.sha256(px).digest(), e["width"], e["height"], e["frames"], payload)
|
||
if k not in _EMBEDDED:
|
||
_EMBEDDED[k] = hashlib.sha256(MARK.embed(px, e["width"], e["height"], e["frames"], payload)[0]).digest()
|
||
return _EMBEDDED[k]
|
||
|
||
|
||
def detected(px, r):
|
||
"""§7.5.5, memoised: the payload read, or None."""
|
||
k = (hashlib.sha256(px).digest(), r["width"], r["height"], r["frames"])
|
||
if k not in _DETECTED:
|
||
got = MARK.detect(px, r["width"], r["height"], r["frames"])
|
||
_DETECTED[k] = None if got is None else got[0]
|
||
return _DETECTED[k]
|
||
|
||
|
||
def recomputes(px, e, payload):
|
||
"""§7.6 step 2: the embedding of the unmarked frames is the marked frames."""
|
||
return embedded(px, e, payload) == e["marked"]
|
||
|
||
|
||
def reproduces(px, r, payload):
|
||
"""§7.6: the detector, run again, reads the detection's payload."""
|
||
return detected(px, r) == payload
|
||
|
||
|
||
def inherits(c, p, by_link):
|
||
"""§7.3 step 4: p is the payload of a held claim c reaches through inputs along steps
|
||
convert and assist."""
|
||
seen, stack = set(), [c]
|
||
while stack:
|
||
x = stack.pop()
|
||
if x["step"] not in ("convert", "assist") or x["link"] in seen:
|
||
continue
|
||
seen.add(x["link"])
|
||
for i in x["inputs"]:
|
||
held = by_link.get(i["origin"])
|
||
if held is not None:
|
||
if payload_of(held) == p:
|
||
return True
|
||
stack.append(held)
|
||
return False
|
||
|
||
|
||
def review_fields(v):
|
||
verdict = one_of(v["verdict"], ["approved", "approved-with-notes", "rejected"])
|
||
notes = opt_hex(v["notes"], 32)
|
||
need((notes is None) == (verdict == "approved"))
|
||
term(v["decoder"])
|
||
opt_term(v["variant"])
|
||
opt_term(v["role"])
|
||
one_of(v["scope"], ["accuracy", "rendering", "translation"])
|
||
return {"subject": hexb(v["subject"], 32), "origin": hexb(v["origin"], 32), "tier": one_of(v["tier"], REVIEW_TIERS),
|
||
"reconstruction": hexb(v["reconstruction"], 32), "verdict": verdict}
|
||
|
||
|
||
def policy_fields(v):
|
||
need(isinstance(v["classes"], list) and len(v["classes"]) >= 1)
|
||
idx = [CLASSES.index(one_of(c, CLASSES)) for c in v["classes"]]
|
||
need(all(a < b for a, b in zip(idx, idx[1:])))
|
||
interval = None if v["interval_us"] is None else integer(v["interval_us"])
|
||
need(interval != 0)
|
||
return {"classes": list(v["classes"]), "visual": hexb(v["visual"], 32), "audio": opt_hex(v["audio"], 32),
|
||
"interval_us": interval, "grace_us": integer(v["grace_us"])}
|
||
|
||
|
||
def entry_of(e):
|
||
t = e.get("type")
|
||
if t == "present":
|
||
members(e, ["type", "subject", "origin", "class", "media", "tier", "decoder", "variant", "output"])
|
||
term(e["decoder"])
|
||
opt_term(e["variant"])
|
||
return {"type": t, "subject": hexb(e["subject"], 32), "origin": opt_hex(e["origin"], 32),
|
||
"class": None if e["class"] is None else one_of(e["class"], CLASSES), "media": one_of(e["media"], MEDIA),
|
||
"tier": one_of(e["tier"], PATH_TIERS), "output": opt_hex(e["output"], 32)}
|
||
if t == "shown":
|
||
members(e, ["type", "label", "asset", "trigger"])
|
||
return {"type": t, "audio": one_of(e["label"], ["visual", "audio"]) == "audio", "asset": hexb(e["asset"], 32),
|
||
"trigger": one_of(e["trigger"], ["first-exposure", "resume", "interval"])}
|
||
if t == "resume":
|
||
members(e, ["type", "after"])
|
||
one_of(e["after"], ["ad-break", "other"])
|
||
return {"type": t}
|
||
if t == "review":
|
||
members(e, ["type", "subject", "review"])
|
||
return {"type": t, "subject": hexb(e["subject"], 32), "review": hexb(e["review"], 32)}
|
||
raise Bad("an entry of no type")
|
||
|
||
|
||
def player_fields(v):
|
||
return {"session": hexb(v["session"], 32), "seq": integer(v["seq"]), "prev": opt_hex(v["prev"], 32), "at": integer(v["at"]),
|
||
"entry": entry_of(v["entry"])}
|
||
|
||
|
||
# ---------------------------------------------------------------- §11 verification
|
||
|
||
def generates(env, classes):
|
||
"""§3.3: an envelope whose capability's registry class is generative-exact or behavioral, or
|
||
whose intent is create."""
|
||
if len(env) < 8 or env[:4] != b"WAI1":
|
||
return False
|
||
n = int.from_bytes(env[4:8], "little")
|
||
try:
|
||
m = json.loads(env[8:8 + n])
|
||
except ValueError:
|
||
return False
|
||
cap = m.get("model_requirement", {}).get("capability", "")
|
||
return classes.get(cap) in ("generative-exact", "behavioral") or m.get("intent") == "create"
|
||
|
||
|
||
def track_digest(room, owner, media, layer):
|
||
"""live-rooms §6.2."""
|
||
return hashlib.sha256(("-".join([room.hex(), owner, media]) + "\x00" + str(layer)).encode()).digest()
|
||
|
||
|
||
def verify(case, classes):
|
||
"""§11.2: the statements of §11.3, or Refuse with its reason and place."""
|
||
st = {"claims": [], "markings": [], "detections": [], "ai_tracks": [], "reviews_presented": [], "labels": [], "visual_icon": None,
|
||
"room": None}
|
||
|
||
def each(key, kind, names, fields, place):
|
||
out = []
|
||
for i, t in enumerate(case[key]):
|
||
try:
|
||
out.append(read(t, kind, names, fields))
|
||
except Refuse as e:
|
||
raise Refuse(e.reason, [place, i])
|
||
return out
|
||
|
||
# Step 1.
|
||
claims = each("claims", "wai-origin", ORIGIN, origin_fields, "claim")
|
||
markings = each("markings", "wai-marking", ["kind", "subject", "origin", "scheme", "scheme_digest", "payload", "embedding", "issued_at",
|
||
"signer_pubkey", "sig"], marking_fields, "marking")
|
||
detections = each("detections", "wai-detection", ["kind", "subject", "scheme", "scheme_digest", "payload", "read", "issued_at",
|
||
"signer_pubkey", "sig"], detection_fields, "detection")
|
||
reviews = each("reviews", "wai-review", ["kind", "subject", "origin", "tier", "reconstruction", "decoder", "variant", "scope", "verdict", "notes",
|
||
"role", "issued_at", "signer_pubkey", "sig"], review_fields, "review")
|
||
policy = None
|
||
if case["label_policy"] is not None:
|
||
try:
|
||
policy = read(case["label_policy"], "wai-label-policy", ["kind", "classes", "visual", "audio", "interval_us", "grace_us", "issued_at",
|
||
"signer_pubkey", "sig"], policy_fields)
|
||
except Refuse as e:
|
||
raise Refuse(e.reason, ["label_policy"])
|
||
log = each("player", "wai-player", ["kind", "session", "seq", "prev", "at", "entry", "signer_pubkey", "sig"], player_fields, "player")
|
||
|
||
by_link = {c["link"]: c for c in claims}
|
||
objects = {blake3(OBJECT_DOMAIN + bytes.fromhex(o)): bytes.fromhex(o) for o in case["objects"]}
|
||
decoded = {bytes.fromhex(d["object"]): bytes.fromhex(d["samples"]) for d in case["decoded"]}
|
||
unmarked = {bytes.fromhex(d["object"]): bytes.fromhex(d["samples"]) for d in case["unmarked"]}
|
||
prove = case["policy"]["second_layer"] == "require-proof"
|
||
sha = lambda b: hashlib.sha256(b).digest() # noqa: E731
|
||
|
||
# Step 2.
|
||
seen = set()
|
||
for i, c in enumerate(claims):
|
||
def no(reason):
|
||
raise Refuse(reason, ["claim", i])
|
||
key = jcs(c["value"]["subject"])
|
||
if key in seen:
|
||
no("claim_conflict")
|
||
seen.add(key)
|
||
if c["class"] != class_of(c):
|
||
no("class_mismatch")
|
||
if c["source"] != source_of(c):
|
||
no("source_mismatch")
|
||
for inp in c["inputs"]:
|
||
held = by_link.get(inp["origin"])
|
||
if held is not None and (held["subject"] != ("object", inp["content_hash"]) or held["class"] != inp["class"]
|
||
or held["source"] != inp["source"]):
|
||
no("input_mismatch")
|
||
kind, h = c["subject"]
|
||
if kind == "object" and h in objects and generates(objects[h], classes) and c["step"] != "generate":
|
||
no("class_understated")
|
||
status = None
|
||
e = c["evidence"]
|
||
if e is not None:
|
||
w, hh = e["width"], e["height"]
|
||
if e["tier"] == "replicate":
|
||
if e["derivation"] not in DERIVATIONS or e["params"] != {} or DERIVATIONS[e["derivation"]][1] * w * hh * 3 > (1 << 26):
|
||
no("evidence_invalid")
|
||
status = "not-recomputed"
|
||
px = decoded.get(c["inputs"][0]["content_hash"])
|
||
if px is not None:
|
||
if len(px) != w * hh * 3 or sha(px) != e["input_samples"]:
|
||
no("input_samples_mismatch")
|
||
if sha(DERIVATIONS[e["derivation"]][0](px, w, hh)) != e["output_samples"]:
|
||
no("assistive_recompute_mismatch")
|
||
status = "proved"
|
||
else:
|
||
status = "asserted"
|
||
if kind == "object" and h in decoded and sha(decoded[h]) != e["output_samples"]:
|
||
no("samples_mismatch")
|
||
if case["policy"]["assistive"] == "require-proof" and status != "proved":
|
||
no("assistive_not_proved")
|
||
st["claims"].append({"link": c["link"].hex(), "class": c["class"], "source": c["source"], "c2pa": statement(c), "assistive": status})
|
||
|
||
# Step 3: §7.2 and §7.6 for the markings.
|
||
for i, m in enumerate(markings):
|
||
def no(reason):
|
||
raise Refuse(reason, ["marking", i])
|
||
c = by_link.get(m["origin"])
|
||
if c is None:
|
||
no("claim_unknown")
|
||
if c["subject"] != ("object", m["subject"]) or payload_of(c) != m["payload"]:
|
||
no("marking_mismatch")
|
||
e = m["embedding"]
|
||
try:
|
||
registered = scheme_check(m["scheme"], m["scheme_digest"], e)
|
||
except Refuse as r:
|
||
no(r.reason)
|
||
status = "stated"
|
||
if registered and m["subject"] in unmarked:
|
||
px = unmarked[m["subject"]]
|
||
if len(px) != e["frames"] * e["width"] * e["height"] * 3 or sha(px) != e["unmarked"]:
|
||
no("mark_input_mismatch")
|
||
if not recomputes(px, e, m["payload"]):
|
||
no("mark_recompute_mismatch")
|
||
status = "proved"
|
||
if e is not None and m["subject"] in decoded and sha(decoded[m["subject"]]) != e["marked"]:
|
||
no("samples_mismatch")
|
||
if prove and status != "proved":
|
||
no("layer_not_proved")
|
||
st["markings"].append({"subject": m["subject"].hex(), "scheme": m["scheme"], "status": status, "c2pa": marking_statement(m)})
|
||
# §7.3 and §7.6 for the detections.
|
||
for i, d in enumerate(detections):
|
||
def no(reason):
|
||
raise Refuse(reason, ["detection", i])
|
||
claim = next((c for c in claims if c["subject"] == ("object", d["subject"])), None)
|
||
marking = next((m for m in markings if m["subject"] == d["subject"]), None)
|
||
if marking is not None and (marking["scheme"], marking["scheme_digest"]) != (d["scheme"], d["scheme_digest"]):
|
||
no("detection_mismatch")
|
||
r = d["read"]
|
||
try:
|
||
registered = scheme_check(d["scheme"], d["scheme_digest"], r)
|
||
except Refuse as x:
|
||
no(x.reason)
|
||
rerun = "not-rerun"
|
||
if r is not None and d["subject"] in decoded:
|
||
px = decoded[d["subject"]]
|
||
if len(px) != r["frames"] * r["width"] * r["height"] * 3 or sha(px) != r["samples"]:
|
||
no("samples_mismatch")
|
||
if registered:
|
||
if not reproduces(px, r, d["payload"]):
|
||
no("detection_irreproducible")
|
||
rerun = "reproduced"
|
||
if prove and rerun != "reproduced":
|
||
no("layer_not_proved")
|
||
if d["payload"] is None:
|
||
if marking is not None:
|
||
no("watermark_absent")
|
||
result = "unmarked"
|
||
elif claim is None:
|
||
result = "not-checked"
|
||
elif d["payload"] == payload_of(claim):
|
||
result = "agree"
|
||
elif inherits(claim, d["payload"], by_link):
|
||
result = "agree-inherited"
|
||
else:
|
||
no("layers_disagree")
|
||
st["detections"].append({"subject": d["subject"].hex(), "result": result, "rerun": rerun})
|
||
if case["policy"]["dual_layer"] == "required":
|
||
for i, c in enumerate(claims):
|
||
if c["class"] == "synthesized" and c["subject"][0] == "object" and not any(m["origin"] == c["link"] for m in markings):
|
||
raise Refuse("layer_missing", ["claim", i])
|
||
|
||
# Step 4.
|
||
for i, r in enumerate(reviews):
|
||
c = by_link.get(r["origin"])
|
||
if c is None:
|
||
raise Refuse("claim_unknown", ["review", i])
|
||
if c["subject"] != ("object", r["subject"]) or c["class"] != "synthesized":
|
||
raise Refuse("review_subject_mismatch", ["review", i])
|
||
|
||
# Step 5.
|
||
room = case["room"]
|
||
if room is not None:
|
||
chains = [[bytes.fromhex(e) for e in ch] for ch in room["chains"]]
|
||
grants = [bytes.fromhex(g) for g in room["grants"]]
|
||
out = LRV.verify_room(bytes.fromhex(room["descriptor"]), grants, room["key_log"], chains)
|
||
if out["outcome"] == "refuse":
|
||
raise Refuse(out["reason"], ["room"] + list(out["at"]))
|
||
st["room"] = out["outcome"]
|
||
d = LRV.read_descriptor(bytes.fromhex(room["descriptor"]))
|
||
held, _ = LRV.check_grants(d, grants)
|
||
by_digest = {g["digest"]: g for g in held}
|
||
current = {}
|
||
for c_i, r_i, under in out["order"]:
|
||
if under is not None:
|
||
continue
|
||
ev = LEV.read_record(chains[c_i][r_i])
|
||
if not 0x10 <= ev["kind"] <= 0x16:
|
||
continue
|
||
subject, x = ev["payload"][:32], LRV.read_detail(ev["kind"], ev["payload"][32:])
|
||
if ev["kind"] == 0x10:
|
||
current[subject] = x["grant"]
|
||
continue
|
||
if ev["kind"] == 0x14:
|
||
current[subject] = x["new"]
|
||
continue
|
||
if ev["kind"] == 0x11:
|
||
current.pop(subject, None)
|
||
continue
|
||
if ev["kind"] not in (0x12, 0x15):
|
||
continue
|
||
g = by_digest.get(current.get(subject))
|
||
model = None
|
||
for k, v in (LRV.read_attrs(g["attrs"]) or []) if g else []:
|
||
if k == "ai":
|
||
model = bytes.fromhex(v)
|
||
if model is None:
|
||
continue
|
||
owner = subject.hex() if not x["slot"] else "_" + x["slot"]
|
||
media = {1: "audio", 2: "video", 3: "screen"}[x["media"]]
|
||
for layer in range(x["layers"]):
|
||
t = track_digest(d["room"], owner, media, layer)
|
||
found = next((c for c in claims if c["subject"] == ("track", t) and c["class"] == "synthesized" and c["step"] == "generate"
|
||
and any(m["digest"] == model for m in c["models"])
|
||
and (c["signer"] == subject or LRV.is_room_key(d, c["signer"]))), None)
|
||
if found is None:
|
||
raise Refuse("ai_track_unlabelled", ["room", "record", c_i, r_i])
|
||
st["ai_tracks"].append({"track": t.hex(), "claim": found["link"].hex()})
|
||
|
||
# Step 6.
|
||
for j, p in enumerate(log):
|
||
ok = p["seq"] == j and p["session"] == log[0]["session"] and p["signer"] == log[0]["signer"]
|
||
ok = ok and (p["prev"] is None if j == 0 else (p["prev"] == log[j - 1]["link"] and p["at"] >= log[j - 1]["at"]))
|
||
if not ok:
|
||
raise Refuse("player_chain_broken", ["player", j])
|
||
for j, p in enumerate(log):
|
||
e = p["entry"]
|
||
if e["type"] == "present":
|
||
c = by_link.get(e["origin"]) if e["origin"] is not None else None
|
||
if c is not None and (c["subject"] != ("object", e["subject"]) or e["class"] != c["class"]):
|
||
raise Refuse("present_mismatch", ["player", j])
|
||
if e["output"] is not None and e["subject"] in decoded and sha(decoded[e["subject"]]) != e["output"]:
|
||
raise Refuse("present_output_irreproducible", ["player", j])
|
||
elif e["type"] == "review":
|
||
r = next((r for r in reviews if r["link"] == e["review"]), None)
|
||
if r is None:
|
||
raise Refuse("claim_unknown", ["player", j])
|
||
last = next((q["entry"] for q in reversed(log[:j]) if q["entry"]["type"] == "present" and q["entry"]["subject"] == e["subject"]), None)
|
||
if last is None or r["subject"] != e["subject"]:
|
||
raise Refuse("review_subject_mismatch", ["player", j])
|
||
if r["verdict"] == "rejected":
|
||
raise Refuse("review_rejected", ["player", j])
|
||
if last["tier"] != r["tier"]:
|
||
raise Refuse("review_path_not_exact", ["player", j])
|
||
if last["output"] != r["reconstruction"]:
|
||
raise Refuse("review_reconstruction_mismatch", ["player", j])
|
||
st["reviews_presented"].append({"entry": j, "review": e["review"].hex()})
|
||
if policy is not None:
|
||
st["visual_icon"] = ICONS.get(policy["visual"].hex())
|
||
for j, p in enumerate(log):
|
||
e = p["entry"]
|
||
if e["type"] == "shown" and (policy["audio"] if e["audio"] else policy["visual"]) != e["asset"]:
|
||
raise Refuse("label_asset_mismatch", ["player", j])
|
||
|
||
def shown_in(audio, lo, hi):
|
||
return next((k for k, q in enumerate(log) if q["entry"]["type"] == "shown" and q["entry"]["audio"] == audio and lo <= q["at"] <= hi), None)
|
||
|
||
last = {False: None, True: None}
|
||
for j, p in enumerate(log):
|
||
e = p["entry"]
|
||
if e["type"] != "present" or e["class"] not in policy["classes"]:
|
||
continue
|
||
audio = e["media"] == "audio"
|
||
hi = p["at"] + policy["grace_us"]
|
||
prev = last[audio]
|
||
if prev is None:
|
||
trigger, k = "first-exposure", shown_in(audio, 0, hi)
|
||
else:
|
||
resumes = [q["at"] for q in log[prev + 1:j] if q["entry"]["type"] == "resume"]
|
||
if resumes:
|
||
trigger, k = "resume", shown_in(audio, resumes[-1], hi)
|
||
elif policy["interval_us"] is not None:
|
||
trigger, k = "interval", shown_in(audio, max(p["at"] - policy["interval_us"], 0), hi)
|
||
else:
|
||
last[audio] = j
|
||
continue
|
||
if k is None:
|
||
raise Refuse("label_not_shown", ["player", j])
|
||
st["labels"].append({"entry": j, "trigger": trigger, "shown": k})
|
||
last[audio] = j
|
||
return st
|
||
|
||
|
||
# ---------------------------------------------------------------- the extension's tables
|
||
|
||
def section(text, head):
|
||
at = text.index("\n" + head)
|
||
nxt = re.search(r"\n#{2,3} ", text[at + 1:])
|
||
return text[at: at + 1 + nxt.start()] if nxt else text[at:]
|
||
|
||
|
||
with open(EXTENSION, encoding="utf-8") as _f:
|
||
_TEXT = _f.read()
|
||
ICONS = {}
|
||
for _line in section(_TEXT, "### 10.2 ").splitlines():
|
||
_m = re.fullmatch(r"\| `(LABEL_AI[^`]*)` \| `([0-9a-f]{64})` \|", _line.strip())
|
||
if _m:
|
||
ICONS[_m.group(2)] = _m.group(1)
|
||
REASONS = [l.split("`")[1] for l in section(_TEXT, "## 12. Reasons").splitlines() if l.startswith("| `")]
|
||
# §13: the codes only the scheme's corpus gives.
|
||
_m = re.search(r"The codes ((?:`[a-z_]+`(?:, | and )?)+) need frames", " ".join(section(_TEXT, "## 13. Conformance").split()))
|
||
MARK_ONLY = re.findall(r"`([a-z_]+)`", _m.group(1)) if _m else []
|
||
# §7.4: the registered scheme's digest.
|
||
_d = re.search(r"its digest is\s+`([0-9a-f]{64})`", section(_TEXT, "### 7.4 "))
|
||
SCHEME_DIGEST = bytes.fromhex(_d.group(1)) if _d else b""
|
||
|
||
|
||
def outcome(v, classes):
|
||
try:
|
||
return {"outcome": "accept", "statements": verify(v, classes)}
|
||
except Refuse as e:
|
||
return {"outcome": "refuse", "reason": e.reason, "at": e.at}
|
||
|
||
|
||
def stated(v):
|
||
return {k: v[k] for k in ("outcome", "reason", "at", "statements") if k in v}
|
||
|
||
|
||
def check(doc, classes):
|
||
"""Failures of a corpus document: an outcome other than the one stated, or a table that
|
||
differs from the extension's."""
|
||
fails = []
|
||
pins = {d["sha256"]: d["file"] for d in doc["eu_icons"]}
|
||
if pins != ICONS or len(doc["eu_icons"]) != len(ICONS):
|
||
fails.append("the corpus's EU icons differ from §10.2's table")
|
||
for v in doc["vectors"]:
|
||
try:
|
||
got = outcome(v, classes)
|
||
except Exception as e: # noqa: BLE001 — a malformed copy fails, whatever it breaks
|
||
got = {"outcome": "unreadable: %r" % (e,)}
|
||
if got != stated(v):
|
||
fails.append("%s: %s" % (v["name"], json.dumps(got)[:300]))
|
||
return fails
|
||
|
||
|
||
def flip_hex_digit(s, at):
|
||
c = "0" if s[at] != "0" else "1"
|
||
return s[:at] + c + s[at + 1:]
|
||
|
||
|
||
def mutations(doc):
|
||
"""Copies of the corpus, each with one defect a conforming verifier must catch."""
|
||
vn = {v["name"]: k for k, v in enumerate(doc["vectors"])}
|
||
|
||
def mutate(label, f):
|
||
d = copy.deepcopy(doc)
|
||
f(d["vectors"])
|
||
return label, d
|
||
|
||
chain, gen, room, cert = vn["chain"], vn["generated"], vn["ai_room"], vn["certified"]
|
||
|
||
def retext(vs, i, key, k, f):
|
||
"""Rewrite record k of list `key` of vector i as JSON, applying f to its value."""
|
||
v = json.loads(vs[i][key][k])
|
||
f(v)
|
||
vs[i][key][k] = json.dumps(v)
|
||
|
||
def decoded_byte(vs):
|
||
d = vs[chain]["decoded"][0]
|
||
d["samples"] = flip_hex_digit(d["samples"], 0)
|
||
|
||
def upscale_digit(vs):
|
||
retext(vs, chain, "claims", 2, lambda v: v["evidence"].update(output_samples=flip_hex_digit(v["evidence"]["output_samples"], 5)))
|
||
|
||
def detection_class(vs):
|
||
retext(vs, gen, "detections", 0, lambda v: v.update(payload="00" + v["payload"][2:]))
|
||
|
||
return [
|
||
mutate("a digit of the upscale's signature", lambda vs: retext(vs, chain, "claims", 2, lambda v: v.update(sig=flip_hex_digit(v["sig"], 9)))),
|
||
mutate("the upscale's output digest changed", upscale_digit),
|
||
mutate("a byte of the reconstruction's decoded samples", decoded_byte),
|
||
mutate("the reconstruction's claim withheld from the chain", lambda vs: vs[chain]["claims"].pop(1)),
|
||
mutate("the generated video's detection reading class captured", detection_class),
|
||
mutate("the generated video's marking withheld", lambda vs: vs[gen]["markings"].pop(0)),
|
||
mutate("the composite's source stated compositeCapture",
|
||
lambda vs: vs[gen]["statements"]["claims"][4].update(source="compositeCapture")),
|
||
mutate("the transcode's statement stated as created",
|
||
lambda vs: vs[gen]["statements"]["claims"][5].update(c2pa=vs[gen]["statements"]["claims"][1]["c2pa"])),
|
||
mutate("the AI host's track claim withheld", lambda vs: vs[room].update(claims=[])),
|
||
mutate("the AI host's grant withheld from the room", lambda vs: vs[room]["room"]["grants"].pop(0)),
|
||
mutate("the certified log's first label withheld", lambda vs: vs[cert]["player"].pop(0)),
|
||
mutate("the label after the break stated as an interval",
|
||
lambda vs: vs[cert]["statements"]["labels"][1].update(trigger="interval")),
|
||
mutate("the review stated at generation-equivalence",
|
||
lambda vs: retext(vs, cert, "reviews", 0, lambda v: v.update(tier="generation-equivalence"))),
|
||
mutate("the presented video's decoded frame changed",
|
||
lambda vs: vs[cert]["decoded"][0].update(samples=flip_hex_digit(vs[cert]["decoded"][0]["samples"], 3))),
|
||
mutate("the policy's visual stated as another pinned icon",
|
||
lambda vs: vs[cert]["statements"].update(visual_icon="LABEL_AI_black.svg")),
|
||
mutate("the first refusal stated at another place", lambda vs: vs[next(k for k, v in enumerate(vs) if v["outcome"] == "refuse")].update(at=["claim", 1])),
|
||
("an EU icon's pin changed", dict(copy.deepcopy(doc), eu_icons=[dict(doc["eu_icons"][0], sha256="00" * 32)] + doc["eu_icons"][1:])),
|
||
]
|
||
|
||
|
||
def main(corpus_dir):
|
||
fails = []
|
||
st = ED._self_test()
|
||
print(("PASS" if not st else "FAIL") + " Ed25519 strict rule self-test " + ",".join(st))
|
||
if st:
|
||
fails.append("ed25519 self-test")
|
||
with open(REGISTRY, 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)
|
||
if len(ICONS) != 24:
|
||
fails.append("§10.2's table does not read as 24 pinned icons")
|
||
reasons_seen = set()
|
||
for v in doc["vectors"]:
|
||
try:
|
||
got = outcome(v, classes)
|
||
except Exception as e: # noqa: BLE001
|
||
got = {"outcome": "unreadable: %r" % (e,)}
|
||
ok = got == stated(v)
|
||
if v.get("reason"):
|
||
reasons_seen.add(v["reason"])
|
||
print(("PASS" if ok else "FAIL") + " vector %s: %s" % (v["name"], got["outcome"] + (" %s at %s" % (got["reason"], got["at"]) if "reason" in got else "")))
|
||
if not ok:
|
||
fails.append("%s: %s, the corpus states %s" % (v["name"], json.dumps(got)[:400], json.dumps(stated(v))[:400]))
|
||
table = check(doc, classes)
|
||
if any("EU icons" in t for t in table):
|
||
fails.append("the corpus's EU icons differ from §10.2's table")
|
||
missing = [r for r in REASONS if r not in reasons_seen and r not in MARK_ONLY]
|
||
if len(MARK_ONLY) != 3 or not set(MARK_ONLY) <= set(REASONS) or len(SCHEME_DIGEST) != 32:
|
||
fails.append("§13's codes of the scheme's corpus, or §7.4's digest, do not read")
|
||
if missing:
|
||
fails.append("no vector refused with %s" % ", ".join(missing))
|
||
accepted = {v["name"] for v in doc["vectors"] if v["outcome"] == "accept"}
|
||
want = {"chain", "assistive_asserted", "generated", "ai_room", "certified"}
|
||
if not want <= accepted:
|
||
fails.append("vectors not accepted: %s" % ", ".join(sorted(want - accepted)))
|
||
print("%s every reason of §12 but %s refuses a vector, and every case of §13 is accepted"
|
||
% ("PASS" if not missing and want <= accepted else "FAIL", ", ".join(MARK_ONLY)))
|
||
for label, d in mutations(doc):
|
||
caught = check(d, 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 f in fails:
|
||
print("FAIL " + f)
|
||
print(("PASS" if not fails else "FAIL") + " synthesis corpus: %d vectors" % len(doc["vectors"]))
|
||
return 1 if fails else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
if len(sys.argv) != 2:
|
||
print("usage: synthesis_verify.py <live-conformance/synthesis>", file=sys.stderr)
|
||
sys.exit(2)
|
||
sys.exit(main(sys.argv[1]))
|