28af66a74f
Merged head 3ec983a533.
By meaning:
- codecs/registry.rs: WireKind gains both new kinds, frame-metadata
(browser sink, in the base) and gltf-extension (SPLAT); all() holds
sixteen; the schema's enum lists both labels.
- synthesis §2's origin-claim table: the base's subject row (with the
samples subject of exact-upscale) and SPLAT's media row (volumetric).
- int_kernels_vectors.rs: the live-conformance exclusion reason names
both the vendored VDAF vectors (TEL) and splat/'s decoded attribute
tensors (SPLAT).
- wai-ops-gpu.yml: paths unioned; the build step builds the ops_gpu,
exact-upscale and splat GPU test targets (SPLAT's name, "test
targets"), DRED's build stays its own step; on Metal, exact-upscale's,
DRED's and the splat step, each whole.
- wai-conformance.yml: paths unioned; the step that conflicted is rebuilt
as two whole steps, QoE aggregation's then exact splats'.
- REGISTRY's prose names exploration, browser-sink, robustness-evidence
and splat-exact, and every wire kind.
- Shared lists kept both sides: CHANGELOG (SPLAT's entry on top),
registry constants, rows, records and wire identifiers, Cargo
features, bins and tests, the live-conformance index (splat/ row, its
provenance sentence, its verifier line). The export is regenerated next.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
1060 lines
46 KiB
Python
1060 lines
46 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 _int_sr as SR # 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"]
|
||
ORIGIN_MEDIA = MEDIA + ["volumetric"] # §3.1
|
||
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))
|
||
if t == "samples":
|
||
members(v, ["type", "samples"])
|
||
return ("samples", hexb(v["samples"], 32))
|
||
raise Bad("a subject of no type")
|
||
|
||
|
||
def origin_fields(v):
|
||
c = {"subject": subject_of(v["subject"]), "media": one_of(v["media"], ORIGIN_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)}
|
||
# The derivations of exact-upscale §3, each with the factor its output's samples hold over its
|
||
# input's (§4.3), checked with the prior its evidence names.
|
||
SR_DERIVATIONS = {SR.SR2: 4, SR.SR4: 16}
|
||
|
||
|
||
def evidence_prior(e):
|
||
"""§4.3: the prior an exact-upscale derivation's `params` names, `{"prior"}`, or None."""
|
||
p = e["params"]
|
||
if not isinstance(p, dict) or set(p) != {"prior"} or not isinstance(p["prior"], str):
|
||
return None
|
||
try:
|
||
return hexb(p["prior"], 32)
|
||
except Bad:
|
||
return None
|
||
|
||
|
||
def replicate_admitted(e):
|
||
"""§4.3: a pinned derivation with its parameters, and an output of at most 2^26 bytes."""
|
||
w, h = e["width"], e["height"]
|
||
if e["derivation"] in DERIVATIONS:
|
||
return e["params"] == {} and DERIVATIONS[e["derivation"]][1] * w * h * 3 <= (1 << 26)
|
||
if e["derivation"] in SR_DERIVATIONS:
|
||
return evidence_prior(e) is not None and SR_DERIVATIONS[e["derivation"]] * w * h * 3 <= (1 << 26)
|
||
return False
|
||
|
||
|
||
# ---------------------------------------------------------------- §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: the registered scheme a record names, or None; 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 in SCHEME_DIGESTS:
|
||
if digest != SCHEME_DIGESTS[scheme]:
|
||
raise Refuse("scheme_mismatch")
|
||
if frames is None or not MARK.admits(frames["width"], frames["height"], frames["frames"]):
|
||
raise Refuse("mark_evidence_invalid")
|
||
return scheme
|
||
if scheme == "wai-mark:conformance" and frames is not None:
|
||
raise Refuse("mark_evidence_invalid")
|
||
return None
|
||
|
||
|
||
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, scheme=MARK.SCHEME):
|
||
"""§7.5.4 or §7.9.3, memoised: the SHA-256 of the marked frames."""
|
||
k = (scheme, hashlib.sha256(px).digest(), e["width"], e["height"], e["frames"], payload)
|
||
if k not in _EMBEDDED:
|
||
fn = MARK.embed_s if scheme == MARK.SCHEME_S else MARK.embed
|
||
_EMBEDDED[k] = hashlib.sha256(fn(px, e["width"], e["height"], e["frames"], payload)[0]).digest()
|
||
return _EMBEDDED[k]
|
||
|
||
|
||
def detected(px, r, scheme=MARK.SCHEME):
|
||
"""§7.5.5 or §7.9.4, memoised: the payload read, or None."""
|
||
k = (scheme, hashlib.sha256(px).digest(), r["width"], r["height"], r["frames"])
|
||
if k not in _DETECTED:
|
||
fn = MARK.detect_s if scheme == MARK.SCHEME_S else MARK.detect
|
||
got = fn(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, scheme=MARK.SCHEME):
|
||
"""§7.6 step 2: the embedding of the unmarked frames is the marked frames."""
|
||
return embedded(px, e, payload, scheme) == e["marked"]
|
||
|
||
|
||
def reproduces(px, r, payload, scheme=MARK.SCHEME):
|
||
"""§7.6: the detector, run again, reads the detection's payload."""
|
||
return detected(px, r, scheme) == 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
|
||
priors = {sha(bytes.fromhex(p)): bytes.fromhex(p) for p in case["priors"]}
|
||
|
||
# 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 not replicate_admitted(e):
|
||
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 e["derivation"] in DERIVATIONS:
|
||
out = DERIVATIONS[e["derivation"]][0](px, w, hh)
|
||
else:
|
||
prior = priors.get(evidence_prior(e))
|
||
try:
|
||
out = None if prior is None else SR.derive(e["derivation"], prior, px, w, hh)
|
||
except SR.Unreadable:
|
||
no("evidence_invalid")
|
||
if out is not None:
|
||
if sha(out) != e["output_samples"]:
|
||
no("assistive_recompute_mismatch")
|
||
status = "proved"
|
||
else:
|
||
status = "asserted"
|
||
subject = sha(decoded[h]) if kind == "object" and h in decoded else h if kind == "samples" else None
|
||
if subject is not None and subject != 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"], registered):
|
||
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"], registered):
|
||
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""
|
||
# §7.9.1: the tiled scheme's digest.
|
||
_d = re.search(r"its digest is\s+`([0-9a-f]{64})`", section(_TEXT, "#### 7.9.1 "))
|
||
SCHEME_DIGEST_S = bytes.fromhex(_d.group(1)) if _d else b""
|
||
SCHEME_DIGESTS = {MARK.SCHEME: SCHEME_DIGEST, MARK.SCHEME_S: SCHEME_DIGEST_S}
|
||
|
||
|
||
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, sink = vn["chain"], vn["generated"], vn["ai_room"], vn["certified"], vn["sink_upscale"]
|
||
|
||
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:]))
|
||
|
||
def prior_byte(vs):
|
||
p = vs[sink]["priors"][0]
|
||
vs[sink]["priors"][0] = p[:40] + ("0" if p[40] != "0" else "1") + p[41:]
|
||
|
||
return [
|
||
mutate("a byte of the sink's prior: the prior its evidence names is not held", prior_byte),
|
||
mutate("a byte of the decoded picture the sink upscaled",
|
||
lambda vs: vs[sink]["decoded"][0].update(samples=flip_hex_digit(vs[sink]["decoded"][0]["samples"], 2))),
|
||
mutate("the sink's upscale stated as not recomputed",
|
||
lambda vs: vs[sink]["statements"]["claims"][2].update(assistive="not-recomputed")),
|
||
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 or len(SCHEME_DIGEST_S) != 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", "sink_upscale", "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]))
|