0a62b9394e
Written from the extension's text; ports the pinned generator and checks it against the generator's own conformance vector first; mutates copies of the corpus and drops each of its own checks in turn. Co-Authored-By: Claude Opus 5.5 <[email protected]>
1273 lines
53 KiB
Python
1273 lines
53 KiB
Python
#!/usr/bin/env python3
|
|
"""Second implementation of extensions/contribution.md (`wai.media.contribution`), over the
|
|
corpus `live-conformance/contribution/`.
|
|
|
|
Written from the extension's text. It shares no code with wai-rs: it reads every record as §2
|
|
has it read, with the strict JSON reading and the encodings of the synthesis corpus's second
|
|
implementation (synthesis_verify.py), which also reads each origin claim (§9.1); it verifies
|
|
every signature with WAI's strict Ed25519 rule (claim-conformance/_ed25519.py) and every link
|
|
hash and tree node with BLAKE3 written from its specification (claim-conformance/_blake3.py);
|
|
it evaluates every pinned operation of §4.5 and runs the pinned generator in its own
|
|
arithmetic, which it first checks against the generator's conformance vector
|
|
(wai-rs/tests/vectors/generate/sha256.json); and it computes every authorship map, every
|
|
share and every C2PA statement itself.
|
|
|
|
python3 -I contribution_verify.py <live-conformance/contribution>
|
|
|
|
It gives every vector of the corpus the outcome, the reason, the place and the statements the
|
|
corpus states and checks that every reason of §11 refuses a vector. Then it mutates copies of
|
|
the corpus and requires every copy to fail, and drops each of its own checks in turn and
|
|
requires the corpus to catch the weakened verifier. Exit status 0 only if all of that holds.
|
|
Python 3.8 or later; standard library only.
|
|
"""
|
|
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import math
|
|
import os
|
|
import re
|
|
import sys
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, HERE)
|
|
sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance"))
|
|
import synthesis_verify as SYN # noqa: E402
|
|
import _ed25519 as ED # noqa: E402
|
|
from _blake3 import blake3 # noqa: E402
|
|
|
|
EXTENSION = os.path.join(HERE, "..", "extensions", "contribution.md")
|
|
GENERATOR_VECTOR = os.path.join(HERE, "..", "wai-rs", "tests", "vectors", "generate")
|
|
|
|
SIGN_DOMAIN = b"wai:contribution\x01" # §2
|
|
LINK_DOMAIN = b"wai:contribution-link\x01" # §2
|
|
LEAF_DOMAIN = b"wai:contribution-leaf\x01" # §5.2
|
|
NODE_DOMAIN = b"wai:contribution-node\x01" # §5.2
|
|
MAP_DOMAIN = b"wai:contribution-map\x01" # §7.3
|
|
SPECTRAL = "wai.generate.spectral"
|
|
LUT = "wai-contribution:lut"
|
|
CROP, ROTATION, MIRROR, SCALING = "wai.derive.crop", "wai.derive.rotation", "wai.derive.mirror", "wai.derive.scaling"
|
|
OVERLAY, GRID = "wai.derive.overlay_composition", "wai.derive.grid_composition"
|
|
UPSCALE2, DENOISE3 = "wai-synthesis:upscale2", "wai-synthesis:denoise3"
|
|
PINNED_TRANSFORMS = [CROP, ROTATION, MIRROR, UPSCALE2, DENOISE3, LUT]
|
|
PINNED_COMPOSITIONS = [OVERLAY, GRID]
|
|
EMPTY = "http://c2pa.org/digitalsourcetype/empty"
|
|
MAX_SAMPLES = 1 << 26
|
|
ACTS = ["author", "generate", "candidates", "select", "edit", "inpaint", "transform", "compose", "reject"]
|
|
TIERS = ["stated", "checked", "recomputed"]
|
|
UNKNOWN, UNDISCLOSED = "unknown", "undisclosed"
|
|
|
|
# The checks a mutation run may drop, by the reason each raises (§11).
|
|
DROPPED = set()
|
|
|
|
|
|
class Refuse(Exception):
|
|
def __init__(self, reason, at=None):
|
|
super().__init__(reason)
|
|
self.reason, self.at = reason, at
|
|
|
|
|
|
Bad = SYN.Bad
|
|
need, hexb, opt_hex, integer, term, one_of, members = SYN.need, SYN.hexb, SYN.opt_hex, SYN.integer, SYN.term, SYN.one_of, SYN.members
|
|
jcs = SYN.jcs
|
|
|
|
|
|
def fail(reason, at):
|
|
"""Refuse with `reason` at `at`, unless a mutation run dropped that check."""
|
|
if reason not in DROPPED:
|
|
raise Refuse(reason, at)
|
|
|
|
|
|
def sha(*parts):
|
|
h = hashlib.sha256()
|
|
for p in parts:
|
|
h.update(p)
|
|
return h.digest()
|
|
|
|
|
|
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 size_ok(w, h):
|
|
return w >= 1 and h >= 1 and w * h * 3 <= MAX_SAMPLES
|
|
|
|
|
|
# ---------------------------------------------------------------- §2-§6 reading
|
|
|
|
def read(text, kind, names, fields):
|
|
"""§2: a signed record of `kind`: its members, then its kind's own, then its signature."""
|
|
try:
|
|
v = SYN.strict_json(text)
|
|
members(v, names)
|
|
need(v["kind"] == kind)
|
|
signer = hexb(v["signer_pubkey"], 32)
|
|
sig = hexb(v["sig"], 64)
|
|
SYN.date_time(v["issued_at"])
|
|
r = fields(v, signer)
|
|
except (Bad, KeyError, TypeError, AttributeError, ValueError):
|
|
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
|
|
|
|
|
|
def session_fields(v, signer):
|
|
"""§3.1."""
|
|
hexb(v["session"], 32)
|
|
tool = term(v["tool"])
|
|
one_of(v["media"], ["image"])
|
|
need(isinstance(v["parties"], list) and 1 <= len(v["parties"]) <= 64)
|
|
parties = []
|
|
for p in v["parties"]:
|
|
need(isinstance(p, dict))
|
|
role = p.get("role")
|
|
if role in ("human", "audience"):
|
|
members(p, ["role", "key"])
|
|
parties.append({"role": role, "key": hexb(p["key"], 32), "digest": None, "trained": False})
|
|
elif role == "model":
|
|
members(p, ["role", "digest", "type"])
|
|
parties.append({"role": role, "key": None, "digest": hexb(p["digest"], 32),
|
|
"trained": one_of(p["type"], ["trained", "algorithmic"]) == "trained"})
|
|
else:
|
|
raise Bad("a party of no role")
|
|
keys = [p["key"] for p in parties if p["key"] is not None]
|
|
digests = [p["digest"] for p in parties if p["digest"] is not None]
|
|
need(len(set(keys)) == len(keys) and len(set(digests)) == len(digests) and signer not in keys)
|
|
return {"tool": tool, "parties": parties}
|
|
|
|
|
|
def membership_fields(v, signer):
|
|
"""§3.2."""
|
|
return {"session": hexb(v["session"], 32), "party": integer(v["party"])}
|
|
|
|
|
|
def output_of(v, act):
|
|
"""§4.1's output of an act."""
|
|
if act == "reject":
|
|
need(v is None)
|
|
return None
|
|
if act == "candidates":
|
|
members(v, ["set", "width", "height"])
|
|
need(isinstance(v["set"], list) and 2 <= len(v["set"]) <= 16)
|
|
s = [hexb(d, 32) for d in v["set"]]
|
|
need(len(set(s)) == len(s))
|
|
w, h = integer(v["width"]), integer(v["height"])
|
|
need(size_ok(w, h))
|
|
return {"set": s, "w": w, "h": h}
|
|
members(v, ["samples", "width", "height"])
|
|
w, h = integer(v["width"]), integer(v["height"])
|
|
need(size_ok(w, h))
|
|
return {"samples": hexb(v["samples"], 32), "w": w, "h": h}
|
|
|
|
|
|
def params_of(p, act, n_members):
|
|
"""§4.3."""
|
|
if act == "author":
|
|
members(p, ["source"])
|
|
return {"source": one_of(p["source"], SYN.CAPTURE_SOURCES)}
|
|
if act in ("generate", "candidates", "inpaint"):
|
|
names = {"generate": ["generator", "prompt", "prompt_by", "seed", "steps"],
|
|
"candidates": ["generator", "prompt", "prompt_by", "seeds", "steps"],
|
|
"inpaint": ["generator", "prompt", "prompt_by", "mask_by", "seed", "steps"]}[act]
|
|
members(p, names)
|
|
r = {"generator": term(p["generator"]), "prompt": opt_hex(p["prompt"], 32),
|
|
"prompt_by": None if p["prompt_by"] is None else integer(p["prompt_by"]),
|
|
"steps": None if p["steps"] is None else integer(p["steps"]), "seed": None, "seeds": None, "mask_by": None}
|
|
need((r["prompt"] is None) == (r["prompt_by"] is None))
|
|
if "seed" in p:
|
|
r["seed"] = None if p["seed"] is None else integer(p["seed"])
|
|
if "seeds" in p and p["seeds"] is not None:
|
|
need(isinstance(p["seeds"], list) and len(p["seeds"]) == n_members)
|
|
r["seeds"] = [integer(s) for s in p["seeds"]]
|
|
if "mask_by" in p:
|
|
r["mask_by"] = integer(p["mask_by"])
|
|
return r
|
|
if act in ("select", "edit", "reject"):
|
|
members(p, [])
|
|
return {}
|
|
members(p, ["derivation", "params"])
|
|
need(isinstance(p["params"], dict))
|
|
return {"derivation": term(p["derivation"]), "params": p["params"]}
|
|
|
|
|
|
def step_fields(v, signer):
|
|
"""§4.1-§4.3. A step of the reserved act is refused once its members are read."""
|
|
act = v["act"]
|
|
need(isinstance(act, str))
|
|
if act == "direct":
|
|
raise Refuse("act_reserved")
|
|
one_of(act, ACTS)
|
|
seq = integer(v["seq"])
|
|
need(isinstance(v["inputs"], list))
|
|
inputs = []
|
|
for i in v["inputs"]:
|
|
members(i, ["step", "samples"])
|
|
x = (integer(i["step"]), hexb(i["samples"], 32))
|
|
need(x[0] < seq and x not in inputs)
|
|
inputs.append(x)
|
|
out = output_of(v["output"], act)
|
|
lo, hi = {"author": (0, 0), "generate": (0, 0), "candidates": (0, 0), "compose": (1, 16)}.get(act, (1, 1))
|
|
need(lo <= len(inputs) <= hi)
|
|
region = None
|
|
if act in ("edit", "inpaint"):
|
|
need(isinstance(v["region"], list) and 1 <= len(v["region"]) <= 256)
|
|
region = []
|
|
for r in v["region"]:
|
|
members(r, ["x", "y", "w", "h"])
|
|
x, y, w, h = (integer(r[k]) for k in ("x", "y", "w", "h"))
|
|
need(w >= 1 and h >= 1 and x + w <= out["w"] and y + h <= out["h"])
|
|
region.append((x, y, w, h))
|
|
else:
|
|
need(v["region"] is None)
|
|
n_members = len(out["set"]) if out is not None and "set" in out else 0
|
|
return {"session": hexb(v["session"], 32), "seq": seq, "prev": opt_hex(v["prev"], 32), "party": integer(v["party"]),
|
|
"act": act, "inputs": inputs, "output": out, "region": region,
|
|
"params": params_of(v["params"], act, n_members),
|
|
"replicate": one_of(v["check"], ["replicate", "attest"]) == "replicate"}
|
|
|
|
|
|
def record_fields(v, signer):
|
|
"""§6."""
|
|
steps = integer(v["steps"])
|
|
need(steps >= 1)
|
|
o = v["output"]
|
|
members(o, ["step", "samples", "width", "height", "content_hash"])
|
|
w, h = integer(o["width"]), integer(o["height"])
|
|
need(size_ok(w, h) and integer(o["step"]) < steps)
|
|
content_hash = opt_hex(o["content_hash"], 32)
|
|
origin = opt_hex(v["origin"], 32)
|
|
need(origin is None or content_hash is not None)
|
|
need(isinstance(v["estimates"], list) and len(v["estimates"]) <= 16)
|
|
for e in v["estimates"]:
|
|
members(e, ["estimator", "class", "party", "basis", "share_ppm"])
|
|
term(e["estimator"])
|
|
term(e["class"])
|
|
integer(e["party"])
|
|
one_of(e["basis"], ["made_by", "directed_by"])
|
|
need(integer(e["share_ppm"]) <= 1000000)
|
|
return {"session": hexb(v["session"], 32), "steps": steps, "root": hexb(v["root"], 32), "out_step": o["step"],
|
|
"out_samples": hexb(o["samples"], 32), "w": w, "h": h, "content_hash": content_hash,
|
|
"map": hexb(v["map"], 32), "origin": origin, "estimates": v["estimates"]}
|
|
|
|
|
|
def read_set(a, n):
|
|
"""§7.3: parties in ascending order, then "unknown" where it holds it."""
|
|
need(isinstance(a, list))
|
|
out, last = set(), -1
|
|
for i, e in enumerate(a):
|
|
if e == UNKNOWN:
|
|
need(i == len(a) - 1)
|
|
out.add(UNKNOWN)
|
|
continue
|
|
p = integer(e)
|
|
need(p < n and p > last)
|
|
last = p
|
|
out.add(p)
|
|
return frozenset(out)
|
|
|
|
|
|
def read_map(text, n):
|
|
"""§7.3: the stated map, in its one encoding; `map_malformed` otherwise."""
|
|
try:
|
|
v = SYN.strict_json(text)
|
|
members(v, ["kind", "output", "width", "height", "palette", "rows"])
|
|
need(v["kind"] == "wai-authorship-map")
|
|
w, h = integer(v["width"]), integer(v["height"])
|
|
need(size_ok(w, h))
|
|
need(isinstance(v["palette"], list))
|
|
palette = []
|
|
for l in v["palette"]:
|
|
members(l, ["made_by", "directed_by"])
|
|
palette.append((read_set(l["made_by"], n), read_set(l["directed_by"], n)))
|
|
need(len(set(palette)) == len(palette))
|
|
need(isinstance(v["rows"], list) and len(v["rows"]) == h)
|
|
labels, seen = [], []
|
|
for row in v["rows"]:
|
|
need(isinstance(row, list))
|
|
total, prev = 0, None
|
|
for run in row:
|
|
need(isinstance(run, list) and len(run) == 2)
|
|
i, k = integer(run[0]), integer(run[1])
|
|
need(k >= 1 and i < len(palette) and i != prev)
|
|
if i not in seen:
|
|
need(i == len(seen))
|
|
seen.append(i)
|
|
prev = i
|
|
total += k
|
|
need(total <= w)
|
|
labels.extend([palette[i]] * k)
|
|
need(total == w)
|
|
need(len(seen) == len(palette))
|
|
return {"digest": blake3(MAP_DOMAIN + jcs(v).encode("utf-8")), "output": hexb(v["output"], 32), "w": w, "h": h,
|
|
"labels": labels}
|
|
except (Bad, KeyError, TypeError, AttributeError, ValueError):
|
|
raise Refuse("map_malformed")
|
|
|
|
|
|
def read_opening(text):
|
|
"""§8.1."""
|
|
try:
|
|
v = SYN.strict_json(text)
|
|
members(v, ["commitment", "salt", "text"])
|
|
need(isinstance(v["text"], str))
|
|
return {"commitment": hexb(v["commitment"], 32), "salt": hexb(v["salt"], 32), "text": v["text"]}
|
|
except (Bad, KeyError, TypeError, AttributeError, ValueError):
|
|
raise Refuse("record_malformed")
|
|
|
|
|
|
# ---------------------------------------------------------------- §4.5 the pinned operations
|
|
|
|
def parse_op(derivation, p):
|
|
"""A pinned derivation's parameters, exactly (§4.5; deriving-edge §4.2), or None."""
|
|
try:
|
|
def pos(x):
|
|
n = integer(x)
|
|
need(n >= 1)
|
|
return n
|
|
if derivation == CROP:
|
|
members(p, ["x", "y", "w", "h"])
|
|
return ("crop", integer(p["x"]), integer(p["y"]), pos(p["w"]), pos(p["h"]))
|
|
if derivation == ROTATION:
|
|
members(p, ["quarter_turns"])
|
|
q = integer(p["quarter_turns"])
|
|
need(1 <= q <= 3)
|
|
return ("rotation", q)
|
|
if derivation == MIRROR:
|
|
members(p, ["flip"])
|
|
return ("mirror", one_of(p["flip"], ["left-right", "top-bottom"]) == "left-right")
|
|
if derivation in (UPSCALE2, DENOISE3):
|
|
members(p, [])
|
|
return ("upscale2",) if derivation == UPSCALE2 else ("denoise3",)
|
|
if derivation == LUT:
|
|
members(p, ["table"])
|
|
t = p["table"]
|
|
need(isinstance(t, str) and re.fullmatch(r"(?:[0-9a-f]{2}){768}", t) is not None)
|
|
return ("lut", bytes.fromhex(t))
|
|
if derivation == OVERLAY:
|
|
members(p, ["width", "height", "fill", "at"])
|
|
fill = hexb(p["fill"], 3)
|
|
need(isinstance(p["at"], list))
|
|
at = []
|
|
for a in p["at"]:
|
|
members(a, ["x", "y"])
|
|
at.append((integer(a["x"]), integer(a["y"])))
|
|
w, h = pos(p["width"]), pos(p["height"])
|
|
need(size_ok(w, h))
|
|
return ("overlay", w, h, fill, at)
|
|
if derivation == GRID:
|
|
members(p, ["columns", "rows"])
|
|
return ("grid", pos(p["columns"]), pos(p["rows"]))
|
|
except (Bad, KeyError, TypeError, AttributeError):
|
|
return None
|
|
return None
|
|
|
|
|
|
def out_size(op, sizes):
|
|
"""The output size over inputs of these sizes, or None where the parameters do not hold."""
|
|
k = op[0]
|
|
if k in ("crop", "rotation", "mirror", "upscale2", "denoise3", "lut"):
|
|
if len(sizes) != 1:
|
|
return None
|
|
iw, ih = sizes[0]
|
|
if k == "crop":
|
|
_, x, y, w, h = op
|
|
if x + w > iw or y + h > ih:
|
|
return None
|
|
r = (w, h)
|
|
elif k == "rotation":
|
|
r = (iw, ih) if op[1] == 2 else (ih, iw)
|
|
elif k == "upscale2":
|
|
r = (2 * iw, 2 * ih)
|
|
else:
|
|
r = (iw, ih)
|
|
elif k == "overlay":
|
|
if not sizes or len(op[4]) != len(sizes):
|
|
return None
|
|
r = (op[1], op[2])
|
|
else:
|
|
_, columns, rows = op
|
|
if len(sizes) != columns * rows or any(s != sizes[0] for s in sizes):
|
|
return None
|
|
r = (columns * sizes[0][0], rows * sizes[0][1])
|
|
return r if size_ok(*r) else None
|
|
|
|
|
|
def sources(op, sizes, x, y):
|
|
"""§4.5: the input pixels output pixel (x, y) reads, as (input, x, y)."""
|
|
k = op[0]
|
|
if k == "crop":
|
|
return [(0, op[1] + x, op[2] + y)]
|
|
if k == "rotation":
|
|
iw, ih = sizes[0]
|
|
return [(0,) + {1: (y, ih - 1 - x), 2: (iw - 1 - x, ih - 1 - y), 3: (iw - 1 - y, x)}[op[1]]]
|
|
if k == "mirror":
|
|
iw, ih = sizes[0]
|
|
return [(0, iw - 1 - x, y)] if op[1] else [(0, x, ih - 1 - y)]
|
|
if k == "lut":
|
|
return [(0, x, y)]
|
|
if k == "upscale2":
|
|
iw, ih = sizes[0]
|
|
sx, sy = x // 2, y // 2
|
|
x1, y1 = min(sx + 1, iw - 1), min(sy + 1, ih - 1)
|
|
pts = [(sx, sy)] + ([(x1, sy)] if x % 2 else []) + ([(sx, y1)] if y % 2 else []) + ([(x1, y1)] if x % 2 and y % 2 else [])
|
|
return sorted({(0, a, b) for a, b in pts})
|
|
if k == "denoise3":
|
|
iw, ih = sizes[0]
|
|
return sorted({(0, min(max(x + i, 0), iw - 1), min(max(y + j, 0), ih - 1)) for i in (-1, 0, 1) for j in (-1, 0, 1)})
|
|
if k == "overlay":
|
|
for i in reversed(range(len(sizes))):
|
|
ax, ay = op[4][i]
|
|
iw, ih = sizes[i]
|
|
if ax <= x < ax + iw and ay <= y < ay + ih:
|
|
return [(i, x - ax, y - ay)]
|
|
return []
|
|
cw, ch = sizes[0]
|
|
return [((y // ch) * op[1] + x // cw, x % cw, y % ch)]
|
|
|
|
|
|
def evaluate(op, bufs):
|
|
"""§4.5: the operation over its inputs' samples: (w, h, samples)."""
|
|
sizes = [(w, h) for w, h, _ in bufs]
|
|
size = out_size(op, sizes)
|
|
if size is None:
|
|
return None
|
|
w, h = size
|
|
k = op[0]
|
|
if k == "upscale2":
|
|
return w, h, SYN.upscale2(bufs[0][2], *sizes[0])
|
|
if k == "denoise3":
|
|
return w, h, SYN.denoise3(bufs[0][2], *sizes[0])
|
|
if k == "lut":
|
|
t = op[1]
|
|
return w, h, bytes(t[256 * (i % 3) + b] for i, b in enumerate(bufs[0][2]))
|
|
fill = op[3] if k == "overlay" else b"\0\0\0"
|
|
out = bytearray()
|
|
for y in range(h):
|
|
for x in range(w):
|
|
s = sources(op, sizes, x, y)
|
|
if s:
|
|
i, sx, sy = s[0]
|
|
o = (sy * sizes[i][0] + sx) * 3
|
|
out += bufs[i][2][o:o + 3]
|
|
else:
|
|
out += fill
|
|
return w, h, bytes(out)
|
|
|
|
|
|
# ---------------------------------------------------------------- the pinned generator
|
|
|
|
M64 = (1 << 64) - 1
|
|
|
|
|
|
def tdiv(a, b):
|
|
"""Integer division truncated toward zero, as the reference generator divides."""
|
|
q = abs(a) // abs(b)
|
|
return q if (a >= 0) == (b > 0) else -q
|
|
|
|
|
|
def fnv1a(data):
|
|
h = 14695981039346656037
|
|
for b in data:
|
|
h = ((h ^ b) * 1099511628211) & M64
|
|
return h
|
|
|
|
|
|
def splitmix(state):
|
|
state = (state + 0x9E3779B97F4A7C15) & M64
|
|
z = state
|
|
z = ((z ^ (z >> 30)) * 0xBF58476D1CE4E5B9) & M64
|
|
z = ((z ^ (z >> 27)) * 0x94D049BB133111EB) & M64
|
|
return state, z ^ (z >> 31)
|
|
|
|
|
|
def iatan2(dy, dx, t):
|
|
if dx == 0:
|
|
return 0 if dy == 0 else (tdiv(t, 4) if dy > 0 else tdiv(3 * t, 4))
|
|
if dy == 0:
|
|
return 0 if dx > 0 else tdiv(t, 2)
|
|
x, y, ang = abs(dx) << 12, abs(dy) << 12, 0
|
|
for i, a in enumerate([128, 76, 40, 20, 10, 5, 3, 1, 1, 0, 0, 0, 0, 0]):
|
|
if y > 0:
|
|
x, y, ang = x + (y >> i), y - (x >> i), ang + a
|
|
elif y < 0:
|
|
x, y, ang = x - (y >> i), y + (x >> i), ang - a
|
|
if dx >= 0 and dy >= 0:
|
|
q = ang
|
|
elif dx < 0 and dy >= 0:
|
|
q = tdiv(t, 2) - ang
|
|
elif dx < 0:
|
|
q = tdiv(t, 2) + ang
|
|
else:
|
|
q = t - ang
|
|
return q % t
|
|
|
|
|
|
def read_model(data):
|
|
"""The generator's model file, where it has every member the generator reads, each in a
|
|
range it computes with; None otherwise."""
|
|
try:
|
|
m = json.loads(data)
|
|
lut = m["sine_lut"]
|
|
ok = (isinstance(m["version"], str) and isinstance(m["lut_size"], int) and m["lut_size"] >= 1
|
|
and isinstance(lut, list) and len(lut) == m["lut_size"]
|
|
and all(isinstance(v, int) and abs(v) <= 1 << 20 for v in lut)
|
|
and all(isinstance(m[k], int) and 0 <= m[k] <= c for k, c in
|
|
(("freq_max", 1 << 12), ("amp_bits", 16), ("base_components", 1024), ("step_components", 1024)))
|
|
and (("lut_bits" not in m) or (isinstance(m["lut_bits"], int) and 0 <= m["lut_bits"] <= 20))
|
|
and isinstance(m["color_offsets"], list) and len(m["color_offsets"]) == 3
|
|
and all(isinstance(v, int) and abs(v) <= 1 << 20 for v in m["color_offsets"]))
|
|
except (ValueError, KeyError, TypeError):
|
|
return None
|
|
if not ok:
|
|
return None
|
|
m.setdefault("lut_bits", 14)
|
|
lex = {}
|
|
for k, e in (m.get("lexicon") or {}).items():
|
|
lex[k] = ([v if isinstance(v, int) else 0 for v in e] + [0] * 10)[:10]
|
|
m["lexicon"] = lex
|
|
return m
|
|
|
|
|
|
def generate(m, prompt, seed, steps, h, w):
|
|
"""`generate(model, prompt, seed, steps, h, w)` of `wai.generate.spectral` (SPEC §5): an
|
|
h x w RGB8 picture."""
|
|
k = m["base_components"] + steps * m["step_components"]
|
|
hs, f, o, r, mi, gr, an, sp, n = [0, 0, 0], 0, 0, 0, 0, 0, 0, 0, 0
|
|
for tok in re.split(r"[^0-9a-z]", prompt.lower()):
|
|
e = m["lexicon"].get(tok) if tok else None
|
|
if e is not None:
|
|
hs = [hs[0] + e[0], hs[1] + e[1], hs[2] + e[2]]
|
|
f, o, r, mi, gr, an, sp, n = f + e[3], o + e[4], r + e[5], mi + e[6], gr + e[7], an + e[8], sp + e[9], n + 1
|
|
if n == 0:
|
|
hue, freq, orient, radial, mirror, gradient, angular, spiral = [256] * 3, 256, 0, 0, 0, 0, 0, 0
|
|
else:
|
|
mx = max(max(hs), 1)
|
|
hue = [tdiv(c * 256, mx) for c in hs]
|
|
clamp = lambda v, lo, hi: min(max(v, lo), hi) # noqa: E731
|
|
freq, orient = tdiv(f, n), clamp(tdiv(o, n), -256, 256)
|
|
radial, mirror, gradient = clamp(tdiv(r, n), 0, 256), clamp(tdiv(mi, n), 0, 256), clamp(tdiv(gr, n), -256, 256)
|
|
angular, spiral = clamp(tdiv(an, n), 0, 256), clamp(tdiv(sp, n), 0, 256)
|
|
fx_scale, fy_scale = (max(256 + orient, 0), 256) if orient < 0 else (256, max(256 - orient, 0))
|
|
fmax = max(tdiv(m["freq_max"] * freq, 256), 1)
|
|
rng = fnv1a(prompt.encode("utf-8")) ^ seed ^ 0xD1B54A32D192ED03
|
|
amp_max = 1 << m["amp_bits"]
|
|
span = 2 * fmax + 1
|
|
thr = [radial, mirror, angular, spiral]
|
|
plane_w = max(256 - sum(thr), 48)
|
|
total = plane_w + sum(thr)
|
|
t = m["lut_size"]
|
|
comps = []
|
|
for _ in range(k):
|
|
rng, pick = splitmix(rng)
|
|
pick %= total
|
|
typ = 0
|
|
if pick >= plane_w:
|
|
bound = plane_w
|
|
for i, wgt in enumerate(thr):
|
|
bound += wgt
|
|
if pick < bound:
|
|
typ = i + 1
|
|
break
|
|
rng, a = splitmix(rng)
|
|
rng, b = splitmix(rng)
|
|
rng, c = splitmix(rng)
|
|
rng, d = splitmix(rng)
|
|
fxr, fyr, amp, phase = a % span - fmax, b % span - fmax, c % amp_max + 1, d % t
|
|
if typ == 1:
|
|
comps.append((1, max(abs(fxr), 2), 0, amp, phase))
|
|
elif typ == 3:
|
|
comps.append((3, abs(fxr) % 11 + 2, 0, amp, phase))
|
|
elif typ == 4:
|
|
comps.append((4, abs(fxr) % 9 + 2, max(abs(fyr), 2), amp, phase))
|
|
else:
|
|
comps.append((typ, tdiv(fxr * fx_scale, 256), tdiv(fyr * fy_scale, 256), amp, phase))
|
|
cx, cy = w // 2, h // 2
|
|
need_r = any(c[0] in (1, 4) for c in comps)
|
|
need_th = any(c[0] in (3, 4) for c in comps)
|
|
gscale = tdiv((1 << m["lut_bits"]) * amp_max * k, 2048 * max(cy, 1))
|
|
lut = m["sine_lut"]
|
|
acc = [0] * (3 * h * w)
|
|
for ch, coff in enumerate(m["color_offsets"]):
|
|
for y in range(h):
|
|
grad = gradient * (cy - y) * gscale
|
|
for x in range(w):
|
|
dx, dy = x - cx, y - cy
|
|
rr = math.isqrt(dx * dx + dy * dy) if need_r else 0
|
|
th = iatan2(dy, dx, t) if need_th else 0
|
|
v = grad
|
|
for typ, fx, fy, amp, phase in comps:
|
|
if typ == 1:
|
|
arg = fx * rr + phase + coff
|
|
elif typ == 2:
|
|
arg = fx * abs(dx) + fy * abs(dy) + phase + coff
|
|
elif typ == 3:
|
|
arg = fx * th + phase + coff
|
|
elif typ == 4:
|
|
arg = fx * th + fy * rr + phase + coff
|
|
else:
|
|
arg = fx * x + fy * y + phase + coff
|
|
v += amp * lut[arg % t]
|
|
acc[ch * h * w + y * w + x] = v
|
|
gmin, gmax = min(acc), max(acc)
|
|
rng_ = max(gmax - gmin, 1)
|
|
eh = [64 + tdiv(c * (256 - 64), 256) for c in hue]
|
|
out = bytearray(h * w * 3)
|
|
for ch in range(3):
|
|
for p in range(h * w):
|
|
q = tdiv((acc[ch * h * w + p] - gmin) * 255 + tdiv(rng_, 2), rng_)
|
|
out[p * 3 + ch] = min(max(tdiv(q * eh[ch], 256), 0), 255)
|
|
return bytes(out)
|
|
|
|
|
|
# ---------------------------------------------------------------- §5.2 the tree
|
|
|
|
def leaf(link):
|
|
return blake3(LEAF_DOMAIN + link)
|
|
|
|
|
|
def node(a, b):
|
|
return blake3(NODE_DOMAIN + a + b)
|
|
|
|
|
|
def proof_len(n):
|
|
return 0 if n <= 1 else (n - 1).bit_length()
|
|
|
|
|
|
def walk(link, seq, siblings):
|
|
h = leaf(link)
|
|
for i, s in enumerate(siblings):
|
|
h = node(s, h) if (seq >> i) & 1 else node(h, s)
|
|
return h
|
|
|
|
|
|
# ---------------------------------------------------------------- §7 maps
|
|
|
|
def label(made, directed):
|
|
return (frozenset(made), frozenset(directed))
|
|
|
|
|
|
L_UNKNOWN = label([UNKNOWN], [UNKNOWN])
|
|
L_UNDISCLOSED = label([UNDISCLOSED], [UNDISCLOSED])
|
|
|
|
|
|
class StateMap:
|
|
"""One label for every pixel, or a label for each."""
|
|
|
|
def __init__(self, const=None, w=None, labels=None):
|
|
self.const, self.w, self.labels = const, w, labels
|
|
|
|
def at(self, x, y):
|
|
return self.const if self.const is not None else self.labels[y * self.w + x]
|
|
|
|
def expand(self, w, h):
|
|
return [self.const] * (w * h) if self.const is not None else list(self.labels)
|
|
|
|
|
|
def shares(labels, n):
|
|
"""§7.4, as §10.4 states them."""
|
|
def per(i):
|
|
return [{"party": p, "samples": c} for p in range(n) for c in [sum(1 for l in labels if p in l[i])] if c > 0]
|
|
return {"samples": len(labels), "made_by": per(0),
|
|
"made_by_unknown": sum(1 for l in labels if UNKNOWN in l[0]),
|
|
"made_by_undisclosed": sum(1 for l in labels if UNDISCLOSED in l[0]),
|
|
"directed_by": per(1),
|
|
"directed_by_unknown": sum(1 for l in labels if UNKNOWN in l[1]),
|
|
"directed_by_undisclosed": sum(1 for l in labels if UNDISCLOSED in l[1])}
|
|
|
|
|
|
# ---------------------------------------------------------------- §10 verification
|
|
|
|
def size_of(out):
|
|
return None if out is None else (out["w"], out["h"])
|
|
|
|
|
|
def holds(out, d):
|
|
if out is None:
|
|
return False
|
|
return d in out["set"] if "set" in out else out["samples"] == d
|
|
|
|
|
|
def states_of(out):
|
|
if out is None:
|
|
return []
|
|
return list(out["set"]) if "set" in out else [out["samples"]]
|
|
|
|
|
|
def in_region(region, x, y):
|
|
return any(rx <= x < rx + rw and ry <= y < ry + rh for rx, ry, rw, rh in region)
|
|
|
|
|
|
def expected_replicate(s):
|
|
"""§4.4: the step's check."""
|
|
a, p = s["act"], s["params"]
|
|
if a in ("author", "edit", "select", "reject"):
|
|
return True
|
|
if a in ("generate", "candidates", "inpaint"):
|
|
return p["generator"] == SPECTRAL
|
|
if a == "transform":
|
|
return p["derivation"] in PINNED_TRANSFORMS
|
|
return p["derivation"] in PINNED_COMPOSITIONS
|
|
|
|
|
|
def replicate_params_ok(s):
|
|
"""§4.4: a replicate step's parameters."""
|
|
p = s["params"]
|
|
if s["act"] in ("generate", "candidates", "inpaint"):
|
|
w, h = size_of(s["output"])
|
|
if s["act"] == "candidates":
|
|
seeds_ok = p["seeds"] is not None and all(x <= 0xFFFFFFFF for x in p["seeds"])
|
|
else:
|
|
seeds_ok = p["seed"] is not None and p["seed"] <= 0xFFFFFFFF
|
|
return seeds_ok and p["steps"] is not None and 1 <= p["steps"] <= 64 and w <= 4096 and h <= 4096 and w * h <= 1 << 23
|
|
if s["act"] in ("transform", "compose"):
|
|
return parse_op(p["derivation"], p["params"]) is not None
|
|
return True
|
|
|
|
|
|
def verify(case, models_by_digest):
|
|
"""§10.2: the statements of §10.4, or Refuse with its reason and place."""
|
|
# Step 1: reading.
|
|
def at(place, f):
|
|
try:
|
|
return f()
|
|
except Refuse as e:
|
|
raise Refuse(e.reason, place)
|
|
|
|
sess = at(["session"], lambda: read(case["session"], "wai-session",
|
|
["kind", "session", "tool", "media", "parties", "issued_at", "signer_pubkey", "sig"], session_fields))
|
|
parties, n, tool_key = sess["parties"], len(sess["parties"]), sess["signer"]
|
|
mems = [at(["membership", i], lambda t=t: read(t, "wai-membership", ["kind", "session", "party", "issued_at", "signer_pubkey", "sig"],
|
|
membership_fields)) for i, t in enumerate(case["memberships"])]
|
|
rec = at(["record"], lambda: read(case["record"], "wai-contribution",
|
|
["kind", "session", "steps", "root", "output", "map", "origin", "estimates", "issued_at",
|
|
"signer_pubkey", "sig"], record_fields))
|
|
steps, proofs = [], []
|
|
for i, d in enumerate(case["steps"]):
|
|
steps.append(at(["step", i], lambda d=d: read(d["step"], "wai-step",
|
|
["kind", "session", "seq", "prev", "party", "act", "inputs", "output", "region",
|
|
"params", "check", "issued_at", "signer_pubkey", "sig"], step_fields)))
|
|
try:
|
|
proofs.append([hexb(p, 32) for p in d["proof"]])
|
|
except Bad:
|
|
raise Refuse("record_malformed", ["step", i])
|
|
smap = at(["map"], lambda: read_map(case["map"], n))
|
|
openings = [at(["opening", i], lambda t=t: read_opening(t)) for i, t in enumerate(case["openings"])]
|
|
origins = []
|
|
for i, t in enumerate(case["origins"]):
|
|
try:
|
|
origins.append(SYN.read(t, "wai-origin", SYN.ORIGIN, SYN.origin_fields))
|
|
except SYN.Refuse as e:
|
|
raise Refuse(e.reason, ["origin", i])
|
|
|
|
# Step 2: openings.
|
|
for i, o in enumerate(openings):
|
|
if sha(o["salt"], o["text"].encode("utf-8")) != o["commitment"]:
|
|
fail("opening_mismatch", ["opening", i])
|
|
|
|
# Step 3: membership.
|
|
signed = set()
|
|
for i, m in enumerate(mems):
|
|
p = parties[m["party"]] if m["party"] < n else None
|
|
if m["session"] != sess["link"] or p is None or p["key"] != m["signer"]:
|
|
fail("membership_invalid", ["membership", i])
|
|
signed.add(m["party"])
|
|
for i, p in enumerate(parties):
|
|
if p["key"] is not None and i not in signed:
|
|
fail("party_not_countersigned", ["session", "party", i])
|
|
|
|
# Step 4: the record.
|
|
if rec["session"] != sess["link"]:
|
|
fail("session_mismatch", ["record"])
|
|
if rec["signer"] != tool_key:
|
|
fail("signer_mismatch", ["record"])
|
|
if any(e["party"] >= n for e in rec["estimates"]):
|
|
fail("party_unknown", ["record"])
|
|
|
|
# Step 5: each step, in the order given.
|
|
first = {}
|
|
for i, s in enumerate(steps):
|
|
first.setdefault(s["seq"], i)
|
|
for i, s in enumerate(steps):
|
|
place = ["step", i]
|
|
if s["session"] != sess["link"]:
|
|
fail("session_mismatch", place)
|
|
p = s["params"]
|
|
named = [x for x in (p.get("prompt_by"), p.get("mask_by")) if x is not None]
|
|
if s["party"] >= n or any(x >= n for x in named):
|
|
fail("party_unknown", place)
|
|
continue
|
|
performer = parties[s["party"]]
|
|
if s["act"] in ("author", "edit") and performer["role"] != "human":
|
|
fail("party_role_mismatch", place)
|
|
if s["act"] in ("generate", "candidates", "inpaint") and performer["role"] != "model":
|
|
fail("party_role_mismatch", place)
|
|
if s["signer"] != (performer["key"] if performer["key"] is not None else tool_key):
|
|
fail("signer_mismatch", place)
|
|
if s["replicate"] != expected_replicate(s):
|
|
fail("check_mismatch", place)
|
|
if s["replicate"] and not replicate_params_ok(s):
|
|
fail("params_invalid", place)
|
|
before = first.get(s["seq"] - 1) if s["seq"] > 0 else None
|
|
if (s["seq"] >= rec["steps"] or first[s["seq"]] != i or (s["seq"] == 0) != (s["prev"] is None)
|
|
or (before is not None and s["prev"] != steps[before]["link"])):
|
|
fail("chain_broken", place)
|
|
if len(proofs[i]) != proof_len(rec["steps"]) or walk(s["link"], s["seq"], proofs[i]) != rec["root"]:
|
|
fail("inclusion_failed", place)
|
|
|
|
# Step 6: each step against the states, in seq order.
|
|
states = {sha(bytes.fromhex(b)): bytes.fromhex(b) for b in case["states"]}
|
|
models = {}
|
|
for d in case["models"]:
|
|
data = models_by_digest[d]
|
|
models[sha(data)] = data
|
|
opened = {o["commitment"]: o["text"] for o in openings}
|
|
by_seq = {seq: steps[i] for seq, i in first.items()}
|
|
|
|
def held_state(d, w, h):
|
|
b = states.get(d)
|
|
return b if b is not None and len(b) == w * h * 3 else None
|
|
|
|
def prompt_text(c):
|
|
return "" if c is None else opened.get(c)
|
|
|
|
status = {}
|
|
for seq in sorted(by_seq):
|
|
s = by_seq[seq]
|
|
status[seq] = check_step(s, by_seq, parties, states, models, held_state, prompt_text, ["step", first[seq]])
|
|
|
|
# Step 7: the output.
|
|
out_step = by_seq.get(rec["out_step"])
|
|
if out_step is not None and out_step["output"] != {"samples": rec["out_samples"], "w": rec["w"], "h": rec["h"]}:
|
|
fail("output_mismatch", ["record"])
|
|
b = states.get(rec["out_samples"])
|
|
if b is not None and len(b) != rec["w"] * rec["h"] * 3:
|
|
fail("samples_mismatch", ["record"])
|
|
|
|
# Step 8: the map.
|
|
if (smap["digest"], smap["output"], smap["w"], smap["h"]) != (rec["map"], rec["out_samples"], rec["w"], rec["h"]):
|
|
fail("map_unbound", ["map"])
|
|
computed = computed_map(by_seq, rec["out_step"], rec["out_samples"], rec["w"], rec["h"])
|
|
for y in range(rec["h"]):
|
|
for x in range(rec["w"]):
|
|
c, st = computed[y * rec["w"] + x], smap["labels"][y * rec["w"] + x]
|
|
for cs, ss in ((c[0], st[0]), (c[1], st[1])):
|
|
ok = (cs - {UNDISCLOSED}) <= ss if UNDISCLOSED in cs else cs == ss
|
|
if not ok:
|
|
fail("unknown_attributed" if UNKNOWN in cs and UNKNOWN not in ss else "map_mismatch", ["map", x, y])
|
|
|
|
# Step 9: the origin claim.
|
|
origin = None
|
|
if rec["origin"] is not None:
|
|
origin = "not-checked"
|
|
for i, o in enumerate(origins):
|
|
if o["link"] != rec["origin"]:
|
|
continue
|
|
if o["subject"] != ("object", rec["content_hash"]):
|
|
fail("origin_mismatch", ["origin", i])
|
|
model_parties = {k for k, p in enumerate(parties) if p["role"] == "model"}
|
|
if any(l[0] & model_parties for l in smap["labels"]) and o["class"] != "synthesized":
|
|
fail("origin_understated", ["origin", i])
|
|
origin = "agree"
|
|
break
|
|
|
|
# §10.3: the tier.
|
|
behind, todo, all_checked = set(), [rec["out_step"]], True
|
|
while todo:
|
|
seq = todo.pop()
|
|
if seq in behind:
|
|
continue
|
|
behind.add(seq)
|
|
s = by_seq.get(seq)
|
|
if s is None:
|
|
all_checked = False
|
|
else:
|
|
all_checked = all_checked and status[seq] == "checked"
|
|
todo.extend(i[0] for i in s["inputs"])
|
|
tier = "stated" if not steps else ("recomputed" if all_checked else "checked")
|
|
committed = rec["steps"] - len(steps)
|
|
c2pa = jcs(c2pa_statement(sess, by_seq, behind)) if committed == 0 else None
|
|
|
|
# Step 10: policy.
|
|
if TIERS.index(tier) < TIERS.index(case["policy"]["tier"]):
|
|
fail("tier_not_met", ["policy"])
|
|
|
|
return {"tier": tier, "session": sess["link"].hex(), "record": rec["link"].hex(),
|
|
"steps": [{"seq": seq, "act": by_seq[seq]["act"], "party": by_seq[seq]["party"],
|
|
"check": "replicate" if by_seq[seq]["replicate"] else "attest", "status": status[seq]} for seq in sorted(by_seq)],
|
|
"committed": committed, "computed": shares(computed, n), "stated": shares(smap["labels"], n),
|
|
"estimates": rec["estimates"], "origin": origin, "c2pa": c2pa}
|
|
|
|
|
|
def check_step(s, by_seq, parties, states, models, held_state, prompt_text, place):
|
|
"""§10.2 step 6 for one step; its status (§10.3)."""
|
|
out = s["output"]
|
|
# 6.1
|
|
if out is not None:
|
|
for d in states_of(out):
|
|
b = states.get(d)
|
|
if b is not None and len(b) != out["w"] * out["h"] * 3:
|
|
fail("samples_mismatch", place)
|
|
# 6.2
|
|
inputs_held = all(i[0] in by_seq for i in s["inputs"])
|
|
if s["act"] == "select":
|
|
step, d = s["inputs"][0]
|
|
if not holds(out, d):
|
|
fail("selection_not_in_set", place)
|
|
c = by_seq.get(step)
|
|
if c is not None and (c["act"] != "candidates" or not holds(c["output"], d)):
|
|
fail("selection_not_in_set", place)
|
|
else:
|
|
for step, d in s["inputs"]:
|
|
if step in by_seq and not holds(by_seq[step]["output"], d):
|
|
fail("input_unknown", place)
|
|
w, h = size_of(out) or (0, 0)
|
|
# 6.3
|
|
if s["act"] in ("edit", "inpaint"):
|
|
src = by_seq.get(s["inputs"][0][0])
|
|
if src is not None and size_of(src["output"]) != (w, h):
|
|
fail("region_invalid", place)
|
|
# 6.4
|
|
p = s["params"]
|
|
op = parse_op(p["derivation"], p["params"]) if s["act"] in ("transform", "compose") and s["replicate"] else None
|
|
sizes = [size_of(by_seq[i[0]]["output"]) for i in s["inputs"]] if inputs_held else None
|
|
if op is not None and sizes is not None:
|
|
size = out_size(op, sizes)
|
|
if size is None:
|
|
fail("params_invalid", place)
|
|
elif size != (w, h):
|
|
fail("recompute_mismatch", place)
|
|
# 6.5
|
|
performer = parties[s["party"]]
|
|
model = None
|
|
if performer["digest"] is not None and performer["digest"] in models:
|
|
model = read_model(models[performer["digest"]])
|
|
ran = False
|
|
a = s["act"]
|
|
if a in ("author", "reject"):
|
|
ran = True
|
|
elif a == "select":
|
|
ran = inputs_held
|
|
elif a in ("edit", "inpaint"):
|
|
b = states.get(s["inputs"][0][1])
|
|
if b is not None and len(b) != w * h * 3:
|
|
fail("region_invalid", place)
|
|
src, dst = held_state(s["inputs"][0][1], w, h), held_state(out["samples"], w, h)
|
|
if src is not None and dst is not None:
|
|
for y in range(h):
|
|
for x in range(w):
|
|
k = (y * w + x) * 3
|
|
if not in_region(s["region"], x, y) and src[k:k + 3] != dst[k:k + 3]:
|
|
fail("outside_region_changed", place)
|
|
ran = True
|
|
if a == "inpaint" and s["replicate"]:
|
|
text = prompt_text(p["prompt"])
|
|
if model is not None and text is not None and dst is not None:
|
|
fill = generate(model, text, p["seed"], p["steps"], h, w)
|
|
for y in range(h):
|
|
for x in range(w):
|
|
k = (y * w + x) * 3
|
|
if in_region(s["region"], x, y) and fill[k:k + 3] != dst[k:k + 3]:
|
|
fail("regeneration_mismatch", place)
|
|
else:
|
|
ran = False
|
|
elif a in ("generate", "candidates") and s["replicate"]:
|
|
text = prompt_text(p["prompt"])
|
|
if model is not None and text is not None:
|
|
seeds = p["seeds"] if a == "candidates" else [p["seed"]]
|
|
for d, seed in zip(states_of(out), seeds):
|
|
if sha(generate(model, text, seed, p["steps"], h, w)) != d:
|
|
fail("regeneration_mismatch", place)
|
|
ran = True
|
|
elif a in ("transform", "compose") and s["replicate"] and sizes is not None:
|
|
bufs = [(iw, ih, held_state(d, iw, ih)) for (_, d), (iw, ih) in zip(s["inputs"], sizes)]
|
|
if all(b is not None for _, _, b in bufs):
|
|
r = evaluate(op, bufs)
|
|
if r is None:
|
|
fail("params_invalid", place)
|
|
elif (r[0], r[1]) != (w, h) or sha(r[2]) != out["samples"]:
|
|
fail("recompute_mismatch", place)
|
|
ran = True
|
|
if not s["replicate"]:
|
|
return "attested"
|
|
return "checked" if ran and inputs_held else "not-checked"
|
|
|
|
|
|
def computed_map(by_seq, out_step, out_samples, w, h):
|
|
"""§7.2: the output's map from the steps held."""
|
|
memo = {}
|
|
|
|
def state(step, d):
|
|
if (step, d) in memo:
|
|
return memo[(step, d)]
|
|
s = by_seq.get(step)
|
|
if s is None:
|
|
return StateMap(const=L_UNDISCLOSED)
|
|
a, p = s["party"], s["params"]
|
|
sw, sh = size_of(s["output"]) or (0, 0)
|
|
directors = frozenset(x for x in (p.get("prompt_by"), p.get("mask_by")) if x is not None)
|
|
act = s["act"]
|
|
if act == "author":
|
|
m = StateMap(const=label([a], []))
|
|
elif act in ("generate", "candidates"):
|
|
m = StateMap(const=(frozenset([a]), directors))
|
|
elif act in ("edit", "inpaint"):
|
|
inner = (frozenset([a]), frozenset() if act == "edit" else directors)
|
|
src = state(*s["inputs"][0])
|
|
m = StateMap(w=sw, labels=[inner if in_region(s["region"], x, y) else src.at(x, y) for y in range(sh) for x in range(sw)])
|
|
elif act == "select":
|
|
src = state(*s["inputs"][0])
|
|
if src.const is not None:
|
|
m = StateMap(const=(src.const[0], src.const[1] | {a}))
|
|
else:
|
|
m = StateMap(w=src.w, labels=[(l[0], l[1] | {a}) for l in src.labels])
|
|
elif act in ("transform", "compose") and not s["replicate"]:
|
|
m = StateMap(const=L_UNKNOWN)
|
|
elif act in ("transform", "compose"):
|
|
op = parse_op(p["derivation"], p["params"])
|
|
if not all(i[0] in by_seq for i in s["inputs"]):
|
|
m = StateMap(const=label([UNDISCLOSED], [UNDISCLOSED, a]))
|
|
else:
|
|
sizes = [size_of(by_seq[i[0]]["output"]) for i in s["inputs"]]
|
|
srcs = [state(*i) for i in s["inputs"]]
|
|
labels = []
|
|
for y in range(sh):
|
|
for x in range(sw):
|
|
src = sources(op, sizes, x, y)
|
|
if not src:
|
|
labels.append(label([a], [a]))
|
|
continue
|
|
made, directed = set(), {a}
|
|
for k, sx, sy in src:
|
|
l = srcs[k].at(sx, sy)
|
|
made |= l[0]
|
|
directed |= l[1]
|
|
labels.append(label(made, directed))
|
|
m = StateMap(w=sw, labels=labels)
|
|
else:
|
|
m = StateMap(const=L_UNDISCLOSED)
|
|
memo[(step, d)] = m
|
|
return m
|
|
|
|
if out_step not in by_seq:
|
|
return StateMap(const=L_UNDISCLOSED).expand(w, h)
|
|
return state(out_step, out_samples).expand(w, h)
|
|
|
|
|
|
def c2pa_statement(sess, by_seq, behind):
|
|
"""§9.2."""
|
|
dst = lambda code: SYN.IPTC + code # noqa: E731
|
|
actions = [{"action": "c2pa.created", "digitalSourceType": EMPTY, "softwareAgent": {"name": sess["tool"]}}]
|
|
by_derivation = {CROP: ("c2pa.cropped", None), ROTATION: ("c2pa.orientation", None), MIRROR: ("c2pa.orientation", None),
|
|
UPSCALE2: ("c2pa.resized", "algorithmicallyEnhanced"), DENOISE3: ("c2pa.enhanced", "algorithmicallyEnhanced"),
|
|
LUT: ("c2pa.adjustedColor", None), SCALING: ("c2pa.resized", None)}
|
|
for seq in sorted(behind):
|
|
s = by_seq[seq]
|
|
p = sess["parties"][s["party"]]
|
|
agent = SYN.MODEL_ID + p["digest"].hex() if p["digest"] is not None else sess["tool"]
|
|
act = s["act"]
|
|
if act in ("select", "reject"):
|
|
continue
|
|
if act == "author":
|
|
name, source = "c2pa.edited", s["params"]["source"]
|
|
elif act in ("generate", "candidates"):
|
|
name, source = "c2pa.edited", "trainedAlgorithmicMedia" if p["trained"] else "algorithmicMedia"
|
|
elif act == "inpaint":
|
|
name, source = "c2pa.edited", "compositeWithTrainedAlgorithmicMedia" if p["trained"] else "algorithmicMedia"
|
|
elif act == "edit":
|
|
name, source = "c2pa.drawing", "digitalCreation"
|
|
elif act == "transform":
|
|
name, source = by_derivation.get(s["params"]["derivation"], ("c2pa.edited", None))
|
|
else:
|
|
name, source = "c2pa.edited", None
|
|
a = {"action": name, "softwareAgent": {"name": agent}}
|
|
if source is not None:
|
|
a["digitalSourceType"] = dst(source)
|
|
if s["region"] is not None:
|
|
a["changes"] = [{"region": [{"type": "spatial", "shape": {"type": "rectangle", "unit": "pixel", "origin": {"x": x, "y": y},
|
|
"width": w, "height": h}} for x, y, w, h in s["region"]]}]
|
|
actions.append(a)
|
|
return {"actions": actions}
|
|
|
|
|
|
# ---------------------------------------------------------------- the corpus
|
|
|
|
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()
|
|
REASONS = [l.split("`")[1] for l in section(_TEXT, "## 11. Reasons").splitlines() if l.startswith("| `")]
|
|
|
|
|
|
def outcome(v, models):
|
|
try:
|
|
return {"outcome": "accept", "statements": verify(v, models)}
|
|
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 models_of(doc):
|
|
out = {}
|
|
for m in doc["models"]:
|
|
data = m["file"].encode("utf-8")
|
|
if sha(data).hex() != m["sha256"]:
|
|
raise ValueError("a model file whose SHA-256 is not the one stated")
|
|
out[m["sha256"]] = data
|
|
return out
|
|
|
|
|
|
def check(doc):
|
|
"""Failures of a corpus document: an outcome other than the one stated."""
|
|
fails = []
|
|
try:
|
|
models = models_of(doc)
|
|
except (ValueError, KeyError, TypeError) as e:
|
|
return ["the model files do not read: %r" % (e,)]
|
|
for v in doc["vectors"]:
|
|
try:
|
|
got = outcome(v, models)
|
|
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
|
|
|
|
def retext(vs, name, key, k, f):
|
|
v = json.loads(vs[vn[name]][key][k]["step"] if key == "steps" else vs[vn[name]][key][k])
|
|
f(v)
|
|
if key == "steps":
|
|
vs[vn[name]][key][k]["step"] = json.dumps(v)
|
|
else:
|
|
vs[vn[name]][key][k] = json.dumps(v)
|
|
|
|
def held_state(vs, name, k, at):
|
|
vs[vn[name]]["states"][k] = flip_hex_digit(vs[vn[name]]["states"][k], at)
|
|
|
|
def stmt(name, f):
|
|
return lambda vs: f(vs[vn[name]]["statements"])
|
|
|
|
first_refusal = next(k for k, v in enumerate(doc["vectors"]) if v["outcome"] == "refuse")
|
|
return [
|
|
mutate("a digit of the brush edit's signature", lambda vs: retext(vs, "brush", "steps", 1, lambda v: v.update(sig=flip_hex_digit(v["sig"], 9)))),
|
|
mutate("a byte of the brushed picture", lambda vs: held_state(vs, "brush", 1, 0)),
|
|
mutate("the inpainting's prompt opened to another text",
|
|
lambda vs: vs[vn["inpaint"]]["openings"].__setitem__(0, json.dumps(dict(json.loads(vs[vn["inpaint"]]["openings"][0]), text="a calm ocean")))),
|
|
mutate("the generator's model file withheld from the candidates", lambda vs: vs[vn["candidates"]].update(models=[])),
|
|
mutate("a sibling of the candidates step's proof", lambda vs: vs[vn["candidates"]]["steps"][0]["proof"].__setitem__(
|
|
0, flip_hex_digit(vs[vn["candidates"]]["steps"][0]["proof"][0], 2))),
|
|
mutate("the brush map's rows reordered", lambda vs: vs[vn["brush"]].update(map=json.dumps(dict(
|
|
json.loads(vs[vn["brush"]]["map"]), rows=list(reversed(json.loads(vs[vn["brush"]]["map"])["rows"])))))),
|
|
mutate("the collaborator's membership withheld", lambda vs: vs[vn["brush"]]["memberships"].pop(1)),
|
|
mutate("the inpainting's origin claim withheld", lambda vs: vs[vn["inpaint"]].update(origins=[])),
|
|
mutate("the compose arranger's count raised by one",
|
|
stmt("compose", lambda s: s["computed"]["directed_by"][0].update(samples=s["computed"]["directed_by"][0]["samples"] + 1))),
|
|
mutate("the scaling's unknown samples stated as none", stmt("scaling_unknown", lambda s: s["computed"].update(made_by_unknown=0))),
|
|
mutate("the partial disclosure stated as recomputed", stmt("candidates_partial", lambda s: s.update(tier="recomputed"))),
|
|
mutate("the transforms' first action stated with another source",
|
|
stmt("transforms", lambda s: s.update(c2pa=s["c2pa"].replace("digitalsourcetype/empty", "digitalsourcetype/digitalCapture")))),
|
|
mutate("the estimate stated as another share", stmt("estimate", lambda s: s["estimates"][0].update(share_ppm=1))),
|
|
mutate("the first refusal stated at another place", lambda vs: vs[first_refusal].update(at=["record"])),
|
|
mutate("a regeneration refusal stated as a recomputation", lambda vs: vs[vn["regeneration_mismatch"]].update(reason="recompute_mismatch")),
|
|
("the model file changed by one byte", dict(copy.deepcopy(doc), models=[dict(doc["models"][0], file=doc["models"][0]["file"].replace("1", "2", 1))])),
|
|
]
|
|
|
|
|
|
def self_test_generator():
|
|
"""The generator's own conformance vector: the picture of its instruction object, by
|
|
SHA-256 (wai-rs/tests/vectors/generate)."""
|
|
with open(os.path.join(GENERATOR_VECTOR, "generate_spectral_model.json"), "rb") as f:
|
|
m = read_model(f.read())
|
|
with open(os.path.join(GENERATOR_VECTOR, "generate.json"), encoding="utf-8") as f:
|
|
g = json.load(f)
|
|
with open(os.path.join(GENERATOR_VECTOR, "sha256.json"), encoding="utf-8") as f:
|
|
want = json.load(f)["generate"]["sha256"]
|
|
return sha(generate(m, g["prompt"], g["seed"], g["steps"], g["h"], g["w"])).hex() == want
|
|
|
|
|
|
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")
|
|
gen_ok = self_test_generator()
|
|
print(("PASS" if gen_ok else "FAIL") + " the generator's arithmetic gives its conformance vector's SHA-256")
|
|
if not gen_ok:
|
|
fails.append("the generator's conformance vector")
|
|
with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as f:
|
|
doc = json.load(f)
|
|
models = models_of(doc)
|
|
reasons_seen = set()
|
|
for v in doc["vectors"]:
|
|
try:
|
|
got = outcome(v, models)
|
|
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 " " + got.get("statements", {}).get("tier", ""))))
|
|
if not ok:
|
|
fails.append("%s: %s, the corpus states %s" % (v["name"], json.dumps(got)[:600], json.dumps(stated(v))[:600]))
|
|
missing = [r for r in REASONS if r not in reasons_seen]
|
|
tiers = {v["statements"]["tier"] for v in doc["vectors"] if v["outcome"] == "accept"}
|
|
if missing:
|
|
fails.append("no vector refused with %s" % ", ".join(missing))
|
|
if tiers != set(TIERS):
|
|
fails.append("accepted vectors do not reach every tier: %s" % sorted(tiers))
|
|
print("%s every reason of §11 refuses a vector, and accepted vectors reach every tier" % ("PASS" if not missing and tiers == set(TIERS) else "FAIL"))
|
|
for label_, d in mutations(doc):
|
|
caught = check(d)
|
|
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 r in REASONS:
|
|
DROPPED.clear()
|
|
DROPPED.add(r)
|
|
caught = check(doc)
|
|
DROPPED.clear()
|
|
if r in ("record_malformed", "bad_key", "bad_signature", "act_reserved", "map_malformed"):
|
|
continue # reading: a record that does not read has nothing to continue with
|
|
print(("PASS" if caught else "FAIL") + " the corpus catches a verifier without its %s check" % r)
|
|
if not caught:
|
|
fails.append("a verifier without its %s check passes the corpus" % r)
|
|
for f in fails:
|
|
print("FAIL " + f)
|
|
print(("PASS" if not fails else "FAIL") + " contribution corpus: %d vectors" % len(doc["vectors"]))
|
|
return 1 if fails else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) != 2:
|
|
print("usage: contribution_verify.py <live-conformance/contribution>", file=sys.stderr)
|
|
sys.exit(2)
|
|
sys.exit(main(sys.argv[1]))
|