1140 lines
49 KiB
Python
1140 lines
49 KiB
Python
#!/usr/bin/env python3
|
|
"""Second implementation of extensions/influence.md (`wai.media.influence`), over the corpus
|
|
`live-conformance/influence/`.
|
|
|
|
Written from the extension's text. It shares no code with wai-rs: it verifies each case's
|
|
contribution session with the contribution corpus's second implementation
|
|
(contribution_verify.py), and runs the pinned generator of §4.4 with that implementation's
|
|
arithmetic of `wai.generate.spectral`; it reads usage statements with ai_usage_verify.py and
|
|
statements with settlement_verify.py, whose attribution lines (settlement §7.8) it gives the
|
|
bases it verified itself; it 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); and it computes every step map, share, line and report itself.
|
|
|
|
python3 -I influence_verify.py <live-conformance/influence>
|
|
|
|
It gives every case, statement and report of the corpus the outcome, the reason, the place and
|
|
the statements the corpus states, and checks that every reason of §9 and every code it reuses
|
|
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 calendar
|
|
import copy
|
|
import json
|
|
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 contribution_verify as CON # noqa: E402
|
|
import ai_usage_verify as AIU # noqa: E402
|
|
import settlement_verify as SET # noqa: E402
|
|
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", "influence.md")
|
|
|
|
SIGN_DOMAIN = b"wai:influence\x01" # §2
|
|
LINK_DOMAIN = b"wai:influence-link\x01" # §2
|
|
BLEND = "wai-influence:spectral-blend" # §4.4
|
|
RULE = "wai-influence:share/1" # §5.2
|
|
SPECTRAL = "wai.generate.spectral"
|
|
UNIT = 10 ** 6 # §5.2
|
|
KINDS = ("image", "voice", "adapter", "prior", "taste", "catalogue")
|
|
GEN_ACTS = ("candidates", "generate", "inpaint")
|
|
TIERS = ["stated", "checked", "recomputed"]
|
|
VOCABULARIES = ("aipref", "cawg", "tdmrep")
|
|
REUSED = ["record_malformed", "bad_key", "bad_signature", "session_mismatch", "signer_mismatch", "output_mismatch",
|
|
"party_unknown", "params_invalid", "unknown_attributed", "tier_not_met", "usage_unanswered", "usage_disallowed",
|
|
"record_missing"]
|
|
|
|
# The checks a mutation run may drop: reasons this verifier raises, and the names of the
|
|
# attribution-line checks of settlement §7.6 it gives settlement_verify.py.
|
|
DROPPED = set()
|
|
|
|
Bad = SYN.Bad
|
|
need, hexb, integer, term, members = SYN.need, SYN.hexb, SYN.integer, SYN.term, SYN.members
|
|
jcs = SYN.jcs
|
|
sha = CON.sha
|
|
|
|
|
|
class Refuse(Exception):
|
|
def __init__(self, reason, at=None):
|
|
super().__init__(reason)
|
|
self.reason, self.at = reason, at
|
|
|
|
|
|
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 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")
|
|
|
|
|
|
DT = re.compile(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}))")
|
|
|
|
|
|
def instant(s):
|
|
"""Seconds since 1970 of a date-time of jwp-receipts §2.4.3, its offset applied, a fraction
|
|
dropped, a 60th second the next minute's first."""
|
|
need(isinstance(s, str))
|
|
SYN.date_time(s)
|
|
m = DT.fullmatch(s)
|
|
y, mo, d, h, mi, sec = (int(m.group(i)) for i in range(1, 7))
|
|
t = calendar.timegm((y, mo, d, h, mi, 0, 0, 0, 0)) + sec
|
|
if m.group(9) is not None:
|
|
off = int(m.group(10)) * 3600 + int(m.group(11)) * 60
|
|
t -= off if m.group(9) == "+" else -off
|
|
return t
|
|
|
|
|
|
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)
|
|
except (Bad, AIU.Refuse, 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 is_term(vocabulary, t):
|
|
"""ai-usage §6.2 rule 4."""
|
|
if vocabulary == "aipref":
|
|
return t in AIU.LABELS
|
|
if vocabulary == "tdmrep":
|
|
return t == AIU.TDM_TERM
|
|
return t in AIU.CAWG_KEYS or (AIU.cawg_key(t) and not t.startswith("cawg."))
|
|
|
|
|
|
def usage_of(a):
|
|
"""§4.1's `usage`: ai-usage §6.1's entries, in strictly ascending order of vocabulary."""
|
|
need(isinstance(a, list) and len(a) <= 3)
|
|
out = []
|
|
for e in a:
|
|
members(e, ["vocabulary", "term"])
|
|
need(e["vocabulary"] in VOCABULARIES)
|
|
t = e["term"]
|
|
need(t is None or (isinstance(t, str) and 1 <= len(t.encode("utf-8")) <= 255 and is_term(e["vocabulary"], t)))
|
|
need(not out or out[-1][0] < e["vocabulary"])
|
|
out.append((e["vocabulary"], t))
|
|
return out
|
|
|
|
|
|
def ascending(xs):
|
|
return all(a < b for a, b in zip(xs, xs[1:]))
|
|
|
|
|
|
# ---------------------------------------------------------------- §3 terms and catalogues
|
|
|
|
TERMS = ["kind", "owner", "work", "usage", "acts", "tariff", "attribution", "valid_from", "valid_until", "issued_at",
|
|
"signer_pubkey", "sig"]
|
|
|
|
|
|
def terms_fields(v):
|
|
w = members(v["work"], ["kind", "digest", "id"])
|
|
need(w["kind"] in KINDS and isinstance(w["id"], str) and 1 <= len(w["id"].encode("utf-8")) <= 256)
|
|
usage = None if v["usage"] is None else AIU.read_statement(v["usage"])
|
|
acts = v["acts"]
|
|
need(isinstance(acts, list) and 1 <= len(acts) and all(a in GEN_ACTS for a in acts) and ascending(acts))
|
|
t = members(v["tariff"], ["currency", "per_use", "share_ppm"])
|
|
cur = t["currency"]
|
|
need(cur is None or (isinstance(cur, str) and re.fullmatch(r"[A-Z]{3}", cur) is not None))
|
|
per_use, share = integer(t["per_use"]), integer(t["share_ppm"])
|
|
need(share <= UNIT and (cur is not None or (per_use == 0 and share == 0)))
|
|
need(v["attribution"] in TIERS)
|
|
frm = instant(v["valid_from"])
|
|
until = None if v["valid_until"] is None else instant(v["valid_until"])
|
|
need(until is None or until > frm)
|
|
return {"owner": hexb(v["owner"], 32), "kind": w["kind"], "digest": hexb(w["digest"], 32), "usage": usage, "acts": acts,
|
|
"currency": cur, "per_use": per_use, "share": share, "attribution": v["attribution"], "from": frm, "until": until}
|
|
|
|
|
|
def read_terms(text):
|
|
t = read(text, "wai-terms", TERMS, terms_fields)
|
|
if t["owner"] != t["signer"]:
|
|
fail("not_signed_by_owner", None)
|
|
return t
|
|
|
|
|
|
def catalogue_fields(v):
|
|
works = v["works"]
|
|
need(isinstance(works, list) and 1 <= len(works) <= 1024)
|
|
ws = [hexb(x, 32) for x in works]
|
|
need(ascending(ws))
|
|
return {"owner": hexb(v["owner"], 32), "works": ws}
|
|
|
|
|
|
def read_catalogue(text):
|
|
c = read(text, "wai-catalogue", ["kind", "owner", "works", "issued_at", "signer_pubkey", "sig"], catalogue_fields)
|
|
if c["owner"] != c["signer"]:
|
|
fail("not_signed_by_owner", None)
|
|
return c
|
|
|
|
|
|
def in_force(t, at_):
|
|
return t["from"] <= at_ and (t["until"] is None or at_ < t["until"])
|
|
|
|
|
|
# ---------------------------------------------------------------- §4 conditioning
|
|
|
|
def reference_of(r):
|
|
members(r, ["kind", "work", "size", "catalogue", "terms", "weight", "usage"])
|
|
need(r["kind"] in KINDS)
|
|
size = None
|
|
if r["size"] is not None:
|
|
s = members(r["size"], ["width", "height"])
|
|
w, h = integer(s["width"]), integer(s["height"])
|
|
need(w >= 1 and h >= 1 and w * h * 3 <= CON.MAX_SAMPLES)
|
|
size = (w, h)
|
|
need((r["kind"] == "image") == (size is not None))
|
|
cat = None if r["catalogue"] is None else hexb(r["catalogue"], 32)
|
|
need(not (r["kind"] == "catalogue" and cat is not None))
|
|
weight = None if r["weight"] is None else integer(r["weight"])
|
|
need(weight is None or 1 <= weight <= UNIT)
|
|
return {"kind": r["kind"], "work": hexb(r["work"], 32), "size": size, "catalogue": cat,
|
|
"terms": None if r["terms"] is None else hexb(r["terms"], 32), "weight": weight, "usage": usage_of(r["usage"])}
|
|
|
|
|
|
def conditioning_fields(v):
|
|
refs = v["references"]
|
|
need(isinstance(refs, list) and 1 <= len(refs) <= 16)
|
|
rs = [reference_of(r) for r in refs]
|
|
need(len({r["work"] for r in rs}) == len(rs))
|
|
weighted = [r["weight"] is not None for r in rs]
|
|
need(all(weighted) or not any(weighted))
|
|
need(sum(r["weight"] or 0 for r in rs) <= UNIT)
|
|
return {"session": hexb(v["session"], 32), "step": integer(v["step"]), "step_link": hexb(v["link"], 32), "refs": rs,
|
|
"weighted": all(weighted)}
|
|
|
|
|
|
def read_conditioning(text):
|
|
return read(text, "wai-conditioning", ["kind", "session", "step", "link", "references", "issued_at", "signer_pubkey", "sig"],
|
|
conditioning_fields)
|
|
|
|
|
|
def blend(model, prompt, seed, steps, w, h, refs):
|
|
"""§4.4: the picture, byte by byte, of the base and each reference's picture by weight."""
|
|
base = CON.generate(model, prompt, seed, steps, h, w)
|
|
pics = []
|
|
for r, x in refs:
|
|
if r["kind"] == "image":
|
|
buf, rw, rh = x
|
|
p = bytearray()
|
|
for y in range(h):
|
|
for xx in range(w):
|
|
k = ((y % rh) * rw + (xx % rw)) * 3
|
|
p += buf[k:k + 3]
|
|
else:
|
|
p = CON.generate(x, prompt, seed, steps, h, w)
|
|
pics.append((r["weight"], p))
|
|
total = sum(wt for wt, _ in pics)
|
|
return bytes((base[i] * (UNIT - total) + sum(wt * p[i] for wt, p in pics) + 500000) // UNIT for i in range(len(base)))
|
|
|
|
|
|
def regenerate(s, c, parties, states, models, openings):
|
|
"""§4.4: True or False where the step regenerates or not, None where something it reads is
|
|
not held."""
|
|
p = s["params"]
|
|
model = models.get(parties[s["party"]]["digest"])
|
|
model = CON.read_model(model) if model is not None else None
|
|
if model is None:
|
|
return None
|
|
prompt = "" if p["prompt"] is None else openings.get(p["prompt"])
|
|
if prompt is None:
|
|
return None
|
|
refs = []
|
|
for r in c["refs"]:
|
|
if r["kind"] == "image":
|
|
rw, rh = r["size"]
|
|
b = states.get(r["work"])
|
|
if b is None or len(b) != rw * rh * 3:
|
|
return None
|
|
refs.append((r, (b, rw, rh)))
|
|
else:
|
|
data = models.get(r["work"])
|
|
m = CON.read_model(data) if data is not None else None
|
|
if m is None:
|
|
return None
|
|
refs.append((r, m))
|
|
out = s["output"]
|
|
w, h = out["w"], out["h"]
|
|
if s["act"] == "generate":
|
|
return sha(blend(model, prompt, p["seed"], p["steps"], w, h, refs)) == out["samples"]
|
|
if s["act"] == "candidates":
|
|
return all(sha(blend(model, prompt, sd, p["steps"], w, h, refs)) == d for d, sd in zip(out["set"], p["seeds"]))
|
|
b = states.get(out["samples"])
|
|
if b is None or len(b) != w * h * 3:
|
|
return None
|
|
pic = blend(model, prompt, p["seed"], p["steps"], w, h, refs)
|
|
for y in range(h):
|
|
for x in range(w):
|
|
k = (y * w + x) * 3
|
|
if CON.in_region(s["region"], x, y) and pic[k:k + 3] != b[k:k + 3]:
|
|
return False
|
|
return True
|
|
|
|
|
|
def blend_params_ok(s):
|
|
"""§4.4: contribution §4.4's parameters of a step of the exact generator."""
|
|
p, out = s["params"], s["output"]
|
|
seeds = p["seeds"] if s["act"] == "candidates" else ([p["seed"]] if p["seed"] is not None else None)
|
|
return (seeds is not None and all(x <= 0xFFFFFFFF for x in seeds) and p["steps"] is not None and 1 <= p["steps"] <= 64
|
|
and out["w"] <= 4096 and out["h"] <= 4096 and out["w"] * out["h"] <= 1 << 23)
|
|
|
|
|
|
# ---------------------------------------------------------------- §5 attribution
|
|
|
|
ATTRIBUTION = ["kind", "session", "record", "output", "samples", "rule", "conditioning", "models", "shares", "estimates",
|
|
"issued_at", "signer_pubkey", "sig"]
|
|
|
|
|
|
def bucket_of(b):
|
|
need(isinstance(b, dict))
|
|
kind = b.get("bucket")
|
|
if kind == "owner":
|
|
members(b, ["bucket", "owner", "terms", "works", "units"])
|
|
need(isinstance(b["works"], list) and b["works"])
|
|
ws = tuple(hexb(x, 32) for x in b["works"])
|
|
need(ascending(list(ws)))
|
|
out = ("owner", hexb(b["owner"], 32), hexb(b["terms"], 32), ws)
|
|
elif kind == "party":
|
|
members(b, ["bucket", "party", "units"])
|
|
out = ("party", integer(b["party"]))
|
|
elif kind == "unowned":
|
|
members(b, ["bucket", "work", "units"])
|
|
out = ("unowned", hexb(b["work"], 32))
|
|
elif kind in ("unknown", "undisclosed"):
|
|
members(b, ["bucket", "units"])
|
|
out = (kind,)
|
|
else:
|
|
raise Bad("a bucket of no kind")
|
|
units = integer(b["units"])
|
|
need(units >= 1)
|
|
return out + (units,)
|
|
|
|
|
|
RANK = {"owner": 0, "party": 1, "unowned": 2, "unknown": 3, "undisclosed": 4}
|
|
|
|
|
|
def bucket_key(b):
|
|
"""§5.3's order, and a bucket's identity."""
|
|
if b[0] == "owner":
|
|
return (0, b[1], b[2])
|
|
if b[0] == "party":
|
|
return (1, b[1].to_bytes(32, "big"), b"")
|
|
if b[0] == "unowned":
|
|
return (2, b[1], b"")
|
|
return (RANK[b[0]], b"", b"")
|
|
|
|
|
|
def bucket_json(b):
|
|
if b[0] == "owner":
|
|
return {"bucket": "owner", "owner": b[1].hex(), "terms": b[2].hex(), "works": [x.hex() for x in b[3]], "units": b[4]}
|
|
if b[0] == "party":
|
|
return {"bucket": "party", "party": b[1], "units": b[2]}
|
|
if b[0] == "unowned":
|
|
return {"bucket": "unowned", "work": b[1].hex(), "units": b[2]}
|
|
return {"bucket": b[0], "units": b[1]}
|
|
|
|
|
|
def attribution_fields(v):
|
|
conds = v["conditioning"]
|
|
need(isinstance(conds, list) and len(conds) <= 256)
|
|
cs = [hexb(x, 32) for x in conds]
|
|
need(ascending(cs))
|
|
need(isinstance(v["models"], list) and len(v["models"]) <= 64)
|
|
models = []
|
|
for m in v["models"]:
|
|
members(m, ["party", "terms", "usage"])
|
|
models.append({"party": integer(m["party"]), "terms": hexb(m["terms"], 32), "usage": usage_of(m["usage"])})
|
|
need(ascending([m["party"] for m in models]))
|
|
need(isinstance(v["shares"], list))
|
|
shares = [bucket_of(b) for b in v["shares"]]
|
|
need(ascending([bucket_key(b) for b in shares]))
|
|
need(isinstance(v["estimates"], list) and len(v["estimates"]) <= 16)
|
|
est = []
|
|
for e in v["estimates"]:
|
|
members(e, ["estimator", "class", "owner", "share_ppm"])
|
|
term(e["estimator"])
|
|
term(e["class"])
|
|
need(integer(e["share_ppm"]) <= UNIT)
|
|
est.append((hexb(e["owner"], 32), e["share_ppm"]))
|
|
return {"session": hexb(v["session"], 32), "record": hexb(v["record"], 32), "output": hexb(v["output"], 32),
|
|
"samples": integer(v["samples"]), "rule": term(v["rule"]), "conditioning": cs, "models": models, "shares": shares,
|
|
"estimates": est, "estimates_json": v["estimates"]}
|
|
|
|
|
|
def units_of(b):
|
|
return b[-1]
|
|
|
|
|
|
def maker_units(by_seq, out_step, out_samples, w, h):
|
|
"""§5.1, §5.2 step 1: each sample's step set, then what each maker takes."""
|
|
memo = {}
|
|
UNK, UND = ("unknown",), ("undisclosed",)
|
|
|
|
def state(step, d):
|
|
if (step, d) in memo:
|
|
return memo[(step, d)]
|
|
s = by_seq.get(step)
|
|
if s is None:
|
|
return ("const", frozenset([UND]))
|
|
act, sw, sh = s["act"], (s["output"] or {}).get("w", 0), (s["output"] or {}).get("h", 0)
|
|
me = frozenset([s["seq"]])
|
|
if act in ("author", "generate", "candidates"):
|
|
m = ("const", me)
|
|
elif act in ("edit", "inpaint"):
|
|
src = state(*s["inputs"][0])
|
|
m = ("px", sw, [me if CON.in_region(s["region"], x, y) else at_(src, x, y) for y in range(sh) for x in range(sw)])
|
|
elif act == "select":
|
|
m = state(*s["inputs"][0])
|
|
elif act in ("transform", "compose") and not s["replicate"]:
|
|
m = ("const", frozenset([UNK]))
|
|
elif act in ("transform", "compose"):
|
|
if not all(i[0] in by_seq for i in s["inputs"]):
|
|
m = ("const", frozenset([UND]))
|
|
else:
|
|
op = CON.parse_op(s["params"]["derivation"], s["params"]["params"])
|
|
sizes = [(by_seq[i[0]]["output"]["w"], by_seq[i[0]]["output"]["h"]) for i in s["inputs"]]
|
|
srcs = [state(*i) for i in s["inputs"]]
|
|
px = []
|
|
for y in range(sh):
|
|
for x in range(sw):
|
|
src = CON.sources(op, sizes, x, y)
|
|
px.append(me if not src else frozenset().union(*(at_(srcs[k], sx, sy) for k, sx, sy in src)))
|
|
m = ("px", sw, px)
|
|
else:
|
|
m = ("const", frozenset([UND]))
|
|
memo[(step, d)] = m
|
|
return m
|
|
|
|
def at_(m, x, y):
|
|
return m[1] if m[0] == "const" else m[2][y * m[1] + x]
|
|
|
|
root = state(out_step, out_samples) if out_step in by_seq else ("const", frozenset([UND]))
|
|
units = {}
|
|
for y in range(h):
|
|
for x in range(w):
|
|
made = at_(root, x, y)
|
|
makers = sorted(e for e in made if isinstance(e, int)) + [e for e in (UNK, UND) if e in made]
|
|
n = len(makers)
|
|
for i, mk in enumerate(makers):
|
|
units[mk] = units.get(mk, 0) + UNIT // n + (UNIT % n if i == 0 else 0)
|
|
return units
|
|
|
|
|
|
def compute_shares(by_seq, parties, units, conds, terms, models, step_tier):
|
|
"""§5.2 steps 2 and 3: the buckets in §5.3's order, and each owner bucket's tier."""
|
|
owner, party, unowned, tiers = {}, {}, {}, {}
|
|
|
|
def give(tl, work, take, tier):
|
|
if take == 0:
|
|
return
|
|
t = terms.get(tl) if tl is not None else None
|
|
if t is None:
|
|
unowned[work] = unowned.get(work, 0) + take
|
|
return
|
|
k = (t["owner"], tl)
|
|
e = owner.setdefault(k, [set(), 0])
|
|
e[0].add(work)
|
|
e[1] += take
|
|
tiers[k] = min(tiers.get(k, tier), tier, key=TIERS.index)
|
|
|
|
for mk, u in units.items():
|
|
if not isinstance(mk, int) or u == 0:
|
|
continue
|
|
s = by_seq[mk]
|
|
p = parties[s["party"]]
|
|
if p["role"] != "model":
|
|
party[s["party"]] = party.get(s["party"], 0) + u
|
|
continue
|
|
tier, left = step_tier(mk), u
|
|
c = conds.get(mk)
|
|
if c is not None:
|
|
n = len(c["refs"])
|
|
for r in c["refs"]:
|
|
take = u * r["weight"] // UNIT if r["weight"] is not None else u // (n + 1)
|
|
left -= take
|
|
give(r["terms"], r["work"], take, tier)
|
|
give(models.get(s["party"]), p["digest"], left, tier)
|
|
out = [("owner", o, l, tuple(sorted(ws)), n) for (o, l), (ws, n) in owner.items()]
|
|
out += [("party", k, n) for k, n in party.items()]
|
|
out += [("unowned", k, n) for k, n in unowned.items()]
|
|
if units.get(("unknown",), 0):
|
|
out.append(("unknown", units[("unknown",)]))
|
|
if units.get(("undisclosed",), 0):
|
|
out.append(("undisclosed", units[("undisclosed",)]))
|
|
out.sort(key=bucket_key)
|
|
return out, tiers
|
|
|
|
|
|
def compare_shares(stated, computed, estimates, samples):
|
|
"""§5.4, with no undisclosed units computed."""
|
|
def of(bs, rank):
|
|
return next((units_of(b) for b in bs if RANK[b[0]] == rank), 0)
|
|
for rank in (3, 4):
|
|
if of(computed, rank) > of(stated, rank):
|
|
fail("unknown_attributed", ["shares"])
|
|
by = {bucket_key(b): units_of(b) for b in computed}
|
|
for b in stated:
|
|
if b[0] != "owner":
|
|
continue
|
|
excess = b[4] - by.get(bucket_key(b), 0)
|
|
if excess <= 0:
|
|
continue
|
|
mine = [p * samples for o, p in estimates if o == b[1]]
|
|
for mask in range(1, 1 << len(mine)):
|
|
if sum(mine[i] for i in range(len(mine)) if mask >> i & 1) == excess:
|
|
fail("estimate_mixed", ["shares"])
|
|
if stated != computed:
|
|
fail("shares_mismatch", ["shares"])
|
|
|
|
|
|
# ---------------------------------------------------------------- §8 verification of a case
|
|
|
|
def preference(t, vocabulary, tm):
|
|
if tm is None:
|
|
return "outside"
|
|
if t is None or t["usage"] is None:
|
|
return "absent"
|
|
p = AIU.pref(t["usage"], vocabulary, tm)
|
|
return "absent" if p is None else p
|
|
|
|
|
|
def usage_ok(t, usage, place):
|
|
"""§4.5 step 4."""
|
|
if t["usage"] is None:
|
|
return
|
|
for voc in t["usage"]["prefs"]:
|
|
if not any(u[0] == voc for u in usage):
|
|
fail("usage_unanswered", place)
|
|
for voc, tm in usage:
|
|
if tm is not None and AIU.pref(t["usage"], voc, tm) == "disallow":
|
|
fail("usage_disallowed", place)
|
|
|
|
|
|
def verify_case(case, models_by_digest):
|
|
"""§8.2: (the statements of §8.3, the basis), or Refuse with its reason and place."""
|
|
# Step 1: contribution.
|
|
cc = dict(case["contribution"], policy={"tier": "stated"})
|
|
try:
|
|
con_out = CON.verify(cc, models_by_digest)
|
|
except CON.Refuse as e:
|
|
raise Refuse(e.reason, ["contribution"] + (e.at or []))
|
|
sess = CON.read(cc["session"], "wai-session", ["kind", "session", "tool", "media", "parties", "issued_at", "signer_pubkey", "sig"],
|
|
CON.session_fields)
|
|
parties, tool = sess["parties"], sess["signer"]
|
|
rec = CON.read(cc["record"], "wai-contribution", ["kind", "session", "steps", "root", "output", "map", "origin", "estimates",
|
|
"issued_at", "signer_pubkey", "sig"], CON.record_fields)
|
|
steps = [CON.read(d["step"], "wai-step", ["kind", "session", "seq", "prev", "party", "act", "inputs", "output", "region",
|
|
"params", "check", "issued_at", "signer_pubkey", "sig"], CON.step_fields)
|
|
for d in cc["steps"]]
|
|
by_seq = {s["seq"]: s for s in steps}
|
|
index_of = {s["seq"]: i for i, s in enumerate(steps)}
|
|
status = {s["seq"]: s["status"] for s in con_out["steps"]}
|
|
|
|
def issued(s):
|
|
return instant(s["value"]["issued_at"])
|
|
|
|
# Step 2: reading.
|
|
def at(place, f):
|
|
try:
|
|
return f()
|
|
except Refuse as e:
|
|
raise Refuse(e.reason, place if e.at is None else e.at)
|
|
terms_l = [at(["terms", i], lambda t=t: read_terms(t)) for i, t in enumerate(case["terms"])]
|
|
cats_l = [at(["catalogue", i], lambda t=t: read_catalogue(t)) for i, t in enumerate(case["catalogues"])]
|
|
conds_l = [at(["conditioning", i], lambda t=t: read_conditioning(t)) for i, t in enumerate(case["conditioning"])]
|
|
attr = at(["attribution"], lambda: read(case["attribution"], "wai-attribution", ATTRIBUTION, attribution_fields))
|
|
terms, cats = {}, {}
|
|
for t in terms_l:
|
|
terms.setdefault(t["link"], t)
|
|
for c in cats_l:
|
|
cats.setdefault(c["link"], c)
|
|
|
|
# Step 3: the attribution.
|
|
pa = ["attribution"]
|
|
if attr["session"] != sess["link"]:
|
|
fail("session_mismatch", pa)
|
|
if attr["signer"] != tool:
|
|
fail("signer_mismatch", pa)
|
|
if attr["record"] != rec["link"] or attr["output"] != rec["content_hash"] or attr["samples"] != rec["w"] * rec["h"]:
|
|
fail("output_mismatch", pa)
|
|
if attr["rule"] != RULE:
|
|
fail("rule_unknown", pa)
|
|
whole = attr["samples"] * UNIT
|
|
if sum(units_of(b) for b in attr["shares"]) != whole:
|
|
fail("shares_not_whole", pa)
|
|
named = []
|
|
for l in attr["conditioning"]:
|
|
i = next((k for k, c in enumerate(conds_l) if c["link"] == l), None)
|
|
if i is None:
|
|
fail("record_missing", pa)
|
|
continue
|
|
named.append((i, conds_l[i]))
|
|
for k, m in enumerate(attr["models"]):
|
|
pm = ["attribution", "model", k]
|
|
if m["party"] >= len(parties) or parties[m["party"]]["role"] != "model":
|
|
fail("party_unknown", pm)
|
|
if m["terms"] not in terms:
|
|
fail("record_missing", pm)
|
|
|
|
# Step 4: each conditioning record.
|
|
states = {sha(bytes.fromhex(b)): bytes.fromhex(b) for b in cc["states"]}
|
|
models = {}
|
|
for d in cc["models"]:
|
|
models[sha(models_by_digest[d])] = models_by_digest[d]
|
|
openings = {}
|
|
for t in cc["openings"]:
|
|
o = CON.read_opening(t)
|
|
openings[o["commitment"]] = o["text"]
|
|
by_step, regenerated, cond_out = {}, set(), []
|
|
for i, c in named:
|
|
pc = ["conditioning", i]
|
|
if c["session"] != sess["link"]:
|
|
fail("session_mismatch", pc)
|
|
if c["signer"] != tool:
|
|
fail("signer_mismatch", pc)
|
|
s = by_seq.get(c["step"])
|
|
if s is not None:
|
|
if c["step_link"] != s["link"]:
|
|
fail("conditioning_unbound", pc)
|
|
if s["act"] not in GEN_ACTS:
|
|
fail("conditioning_invalid", pc)
|
|
if c["step"] in by_step:
|
|
fail("conditioning_invalid", pc)
|
|
generator = s["params"].get("generator") if s is not None and s["act"] in GEN_ACTS else None
|
|
if generator == SPECTRAL:
|
|
fail("conditioning_mismatch", pc)
|
|
if generator == BLEND:
|
|
if not c["weighted"] or any(r["kind"] not in ("image", "adapter") for r in c["refs"]):
|
|
fail("conditioning_invalid", pc)
|
|
if not blend_params_ok(s):
|
|
fail("params_invalid", pc)
|
|
refs_out = []
|
|
for j, r in enumerate(c["refs"]):
|
|
pr = ["conditioning", i, "reference", j]
|
|
t, owner = None, None
|
|
if r["terms"] is not None:
|
|
t = terms.get(r["terms"])
|
|
if t is None:
|
|
fail("record_missing", pr)
|
|
continue
|
|
if r["catalogue"] is None:
|
|
if t["kind"] != r["kind"] or t["digest"] != r["work"]:
|
|
fail("terms_mismatch", pr)
|
|
else:
|
|
cat = cats.get(r["catalogue"])
|
|
if cat is None:
|
|
fail("record_missing", pr)
|
|
else:
|
|
if t["kind"] != "catalogue" or t["digest"] != cat["link"] or cat["owner"] != t["owner"]:
|
|
fail("terms_mismatch", pr)
|
|
if r["work"] not in cat["works"]:
|
|
fail("not_in_catalogue", pr)
|
|
if s is not None and not in_force(t, issued(s)):
|
|
fail("terms_not_in_force", pr)
|
|
usage_ok(t, r["usage"], pr)
|
|
if s is not None and s["act"] not in t["acts"]:
|
|
fail("usage_disallowed", pr)
|
|
owner = t["owner"].hex()
|
|
refs_out.append({"work": r["work"].hex(), "owner": owner,
|
|
"usage": [{"vocabulary": v, "term": tm, "preference": preference(t, v, tm)} for v, tm in r["usage"]]})
|
|
if s is None:
|
|
st = "committed"
|
|
elif generator == BLEND:
|
|
ok = regenerate(s, c, parties, states, models, openings)
|
|
if ok is False:
|
|
fail("conditioning_mismatch", pc)
|
|
if ok:
|
|
regenerated.add(s["seq"])
|
|
st = "recomputed" if ok else "checked"
|
|
else:
|
|
st = "stated"
|
|
by_step[c["step"]] = c
|
|
cond_out.append({"step": c["step"], "status": st, "weights": "declared" if c["weighted"] else "equal", "references": refs_out})
|
|
|
|
# Step 5.
|
|
for s in steps:
|
|
if s["act"] in GEN_ACTS and s["params"]["generator"] == BLEND and s["seq"] not in by_step:
|
|
fail("conditioning_missing", ["step", index_of[s["seq"]]])
|
|
|
|
# Step 6: each models entry.
|
|
model_terms = {}
|
|
for k, m in enumerate(attr["models"]):
|
|
pm = ["attribution", "model", k]
|
|
t = terms.get(m["terms"])
|
|
if t is None or m["party"] >= len(parties):
|
|
continue
|
|
if t["kind"] != "prior" or t["digest"] != parties[m["party"]]["digest"]:
|
|
fail("terms_mismatch", pm)
|
|
usage_ok(t, m["usage"], pm)
|
|
for s in steps:
|
|
if s["party"] == m["party"] and s["act"] in GEN_ACTS:
|
|
if not in_force(t, issued(s)):
|
|
fail("terms_not_in_force", pm)
|
|
if s["act"] not in t["acts"]:
|
|
fail("usage_disallowed", pm)
|
|
model_terms[m["party"]] = m["terms"]
|
|
|
|
# Step 7: the shares.
|
|
def is_blend(seq):
|
|
s = by_seq[seq]
|
|
return s["act"] in GEN_ACTS and s["params"]["generator"] == BLEND
|
|
|
|
def step_tier(seq):
|
|
if is_blend(seq):
|
|
return "recomputed" if seq in regenerated else "checked"
|
|
return {"checked": "recomputed", "not-checked": "checked"}.get(status.get(seq), "stated")
|
|
|
|
units = maker_units(by_seq, rec["out_step"], rec["out_samples"], rec["w"], rec["h"])
|
|
computed, tiers = compute_shares(by_seq, parties, units, by_step, terms, model_terms, step_tier)
|
|
compared = not any(b[0] == "undisclosed" for b in computed)
|
|
if compared:
|
|
compare_shares(attr["shares"], computed, attr["estimates"], attr["samples"])
|
|
|
|
# Step 8: the owners' buckets.
|
|
owners_out, payees = [], []
|
|
for k, b in enumerate(attr["shares"]):
|
|
if b[0] != "owner":
|
|
continue
|
|
ps = ["shares", k]
|
|
t = terms.get(b[2])
|
|
if t is None:
|
|
fail("record_missing", ps)
|
|
continue
|
|
if t["owner"] != b[1]:
|
|
fail("terms_mismatch", ps)
|
|
tier = tiers.get((b[1], b[2]), "stated") if compared else "stated"
|
|
if TIERS.index(tier) < TIERS.index(t["attribution"]):
|
|
fail("attribution_not_met", ps)
|
|
owners_out.append({"owner": b[1].hex(), "terms": b[2].hex(), "tier": tier})
|
|
payees.append({"key": b[1], "units": b[4], "per_use": t["per_use"], "share": t["share"], "currency": t["currency"],
|
|
"terms": b[2], "works": b[3]})
|
|
|
|
# Step 9: policy.
|
|
behind, todo, every = 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:
|
|
every = False
|
|
continue
|
|
every = every and ((seq in regenerated) if is_blend(seq) else status.get(seq) == "checked")
|
|
todo.extend(i[0] for i in s["inputs"])
|
|
tier = "stated" if not steps else ("recomputed" if every else "checked")
|
|
if TIERS.index(tier) < TIERS.index(case["policy"]["tier"]):
|
|
fail("tier_not_met", ["policy"])
|
|
statements = {
|
|
"tier": tier, "record": rec["link"].hex(), "attribution": attr["link"].hex(), "conditioning": cond_out,
|
|
"shares": {"computed": [bucket_json(b) for b in computed], "stated": [bucket_json(b) for b in attr["shares"]],
|
|
"compared": compared},
|
|
"owners": owners_out, "estimates": attr["estimates_json"],
|
|
}
|
|
basis = {"link": attr["link"], "output": attr["output"], "whole": whole, "payees": payees}
|
|
return statements, basis
|
|
|
|
|
|
def case_outcome(v, models):
|
|
try:
|
|
return {"outcome": "accept", "statements": verify_case(v, models)[0]}
|
|
except Refuse as e:
|
|
return {"outcome": "refuse", "reason": e.reason, "at": e.at}
|
|
|
|
|
|
# ---------------------------------------------------------------- §6.2 a statement
|
|
|
|
class Stop(Exception):
|
|
def __init__(self, reason, at):
|
|
super().__init__(reason)
|
|
self.reason, self.at = reason, at
|
|
|
|
|
|
def settlement_inputs(v):
|
|
"""settlement_verify.py's pool and vector, from a statement vector's records."""
|
|
doc = {"records": [{"name": str(k), "record": r["record"], "items": r["items"]} for k, r in enumerate(v["records"])],
|
|
"proofs": [], "hops": [], "rooms": {}}
|
|
vec = {"statement": v["statement"], "withhold": [], "extra": {"records": [], "proofs": []}, "rooms": []}
|
|
return doc, vec
|
|
|
|
|
|
def attribution_link(case):
|
|
try:
|
|
return link_hash(SYN.strict_json(case["attribution"]))
|
|
except Bad:
|
|
return None
|
|
|
|
|
|
def revenue(acks, prices, output, c):
|
|
paid = 0
|
|
for a in acks:
|
|
if a["mode"] != SET.ACK:
|
|
continue
|
|
for e, its in zip(a["entries"], a["items_read"]):
|
|
if e["class"] == c and not (c in SET.RETRIEVAL and e["grant"] == SET.ZERO):
|
|
paid += sum(end - start for h, start, end in its if h == output)
|
|
return paid * prices[c - 1] // SET.PER
|
|
|
|
|
|
def verify_statement(v, cases, models):
|
|
"""§6.2: (the accepted outcome, the statement, its acks, its bases), or Stop."""
|
|
doc, vec = settlement_inputs(v)
|
|
try:
|
|
s = SET.read_statement(bytes.fromhex(v["statement"]))
|
|
except SET.Refuse as e:
|
|
raise Stop(e.reason, ["statement"])
|
|
given = [cases[n] for n in v["given"]]
|
|
links = [attribution_link(c) for c in given]
|
|
bases, used = {}, set()
|
|
for k, d in enumerate(SET.basis_order(s)):
|
|
j = next((i for i, l in enumerate(links) if l == d), None)
|
|
if j is None:
|
|
raise Stop("record_missing", ["basis", k])
|
|
try:
|
|
_, b = verify_case(given[j], models)
|
|
except Refuse as e:
|
|
raise Stop(e.reason, ["basis", k] + e.at)
|
|
cur = s["currency"].decode("latin-1") if s["currency"] != bytes(3) else None
|
|
if any(p["currency"] is not None and p["currency"] != cur for p in b["payees"]) and "currency_mismatch" not in DROPPED:
|
|
raise Stop("currency_mismatch", ["basis", k])
|
|
used.add(j)
|
|
bases[d] = b
|
|
set_bases = {d: {"output": b["output"], "whole": b["whole"], "payees": [(p["key"], p["units"], p["per_use"], p["share"]) for p in b["payees"]]}
|
|
for d, b in bases.items()}
|
|
SET.SKIP.clear()
|
|
SET.SKIP.update(x for x in DROPPED if x.startswith("attribution-"))
|
|
try:
|
|
out = SET.verify_statement(doc, vec, set_bases)
|
|
except SET.Stop as e:
|
|
raise Stop(e.reason, e.at)
|
|
finally:
|
|
SET.SKIP.clear()
|
|
acks = []
|
|
for d in s["records"]:
|
|
r = next(r for r in v["records"] if SET.given_digest(bytes.fromhex(r["record"])) == d)
|
|
a = SET.read_ack(bytes.fromhex(r["record"]))
|
|
a["items_read"] = [SET.read_items(bytes.fromhex(x)) for x in r["items"][:len(a["entries"])]]
|
|
acks.append(a)
|
|
for j, c in enumerate(given):
|
|
if j in used:
|
|
continue
|
|
try:
|
|
_, b = verify_case(c, models)
|
|
except Refuse as e:
|
|
raise Stop(e.reason, ["given", j] + e.at)
|
|
delivered = any(a["mode"] == SET.ACK and any(it[0] == b["output"] for its in a["items_read"] for it in its) for a in acks)
|
|
if b["payees"] and delivered and "attribution_unsettled" not in DROPPED:
|
|
raise Stop("attribution_unsettled", ["given", j])
|
|
out["bases"] = [d.hex() for d in SET.basis_order(s)]
|
|
return out, s, acks, bases
|
|
|
|
|
|
def statement_outcome(v, cases, models):
|
|
try:
|
|
return verify_statement(v, cases, models)[0]
|
|
except Stop as e:
|
|
return {"outcome": "refuse", "reason": e.reason, "at": e.at}
|
|
|
|
|
|
# ---------------------------------------------------------------- §7 the report
|
|
|
|
REPORT = ["kind", "owner", "party", "statement", "start", "end", "outputs", "amount", "issued_at", "signer_pubkey", "sig"]
|
|
|
|
|
|
def report_fields(v):
|
|
need(isinstance(v["outputs"], list) and len(v["outputs"]) <= 1024)
|
|
outs = []
|
|
for o in v["outputs"]:
|
|
members(o, ["basis", "output", "whole", "buckets", "lines"])
|
|
need(isinstance(o["buckets"], list) and 1 <= len(o["buckets"]) <= 16)
|
|
bs = []
|
|
for b in o["buckets"]:
|
|
members(b, ["terms", "works", "units"])
|
|
need(isinstance(b["works"], list) and b["works"])
|
|
ws = [hexb(x, 32) for x in b["works"]]
|
|
need(ascending(ws) and integer(b["units"]) >= 1)
|
|
bs.append((hexb(b["terms"], 32), tuple(ws), b["units"]))
|
|
need(ascending([b[0] for b in bs]))
|
|
need(isinstance(o["lines"], list) and 1 <= len(o["lines"]) <= 4)
|
|
ls = []
|
|
for l in o["lines"]:
|
|
members(l, ["class", "deliveries", "bytes", "revenue", "amount"])
|
|
need(1 <= integer(l["class"]) <= 4)
|
|
ls.append(tuple(integer(l[k]) for k in ("class", "deliveries", "bytes", "revenue", "amount")))
|
|
need(ascending([l[0] for l in ls]))
|
|
need(integer(o["whole"]) >= 1)
|
|
outs.append({"basis": hexb(o["basis"], 32), "output": hexb(o["output"], 32), "whole": o["whole"], "buckets": bs, "lines": ls})
|
|
need(ascending([o["basis"] for o in outs]))
|
|
return {"owner": hexb(v["owner"], 32), "party": hexb(v["party"], 32), "statement": hexb(v["statement"], 32),
|
|
"start": integer(v["start"]), "end": integer(v["end"]), "outputs": outs, "amount": integer(v["amount"])}
|
|
|
|
|
|
def verify_report(v, statements, cases, models):
|
|
"""§7.2."""
|
|
try:
|
|
r = read(v["report"], "wai-influence-report", REPORT, report_fields)
|
|
except Refuse as e:
|
|
raise Stop(e.reason, ["report"])
|
|
if r["signer"] != r["party"] and "signer_mismatch" not in DROPPED:
|
|
raise Stop("signer_mismatch", ["report"])
|
|
terms = {}
|
|
for i, t in enumerate(v["terms"]):
|
|
try:
|
|
x = read_terms(t)
|
|
except Refuse as e:
|
|
raise Stop(e.reason, ["terms", i])
|
|
terms.setdefault(x["link"], x)
|
|
total = 0
|
|
for i, o in enumerate(r["outputs"]):
|
|
po = ["report", "output", i]
|
|
ts = []
|
|
for tl, _, u in o["buckets"]:
|
|
t = terms.get(tl)
|
|
if t is None:
|
|
raise Stop("record_missing", po)
|
|
if t["owner"] != r["owner"] and "terms_mismatch" not in DROPPED:
|
|
raise Stop("terms_mismatch", po)
|
|
ts.append((t, u))
|
|
for _, n, _, rev, amount in o["lines"]:
|
|
want = sum(t["per_use"] * n + rev * t["share"] * u // (UNIT * o["whole"]) for t, u in ts)
|
|
if want != amount and "report_mismatch" not in DROPPED:
|
|
raise Stop("report_mismatch", po)
|
|
total += amount
|
|
if total != r["amount"] and "report_mismatch" not in DROPPED:
|
|
raise Stop("report_mismatch", ["report"])
|
|
if v["audit"] is not None:
|
|
sv = statements[v["audit"]]
|
|
try:
|
|
out, s, acks, bases = verify_statement(sv, cases, models)
|
|
except Stop as e:
|
|
raise Stop(e.reason, ["statement"] + e.at)
|
|
if (r["party"], r["statement"], r["start"], r["end"]) != (s["party"], s["digest"], s["start"], s["end"]) and "report_mismatch" not in DROPPED:
|
|
raise Stop("report_mismatch", ["report"])
|
|
want = {}
|
|
for x in s["lines"]:
|
|
if x["kind"] != SET.ATTRIBUTION_LINE or x["payee"] != r["owner"]:
|
|
continue
|
|
b = bases[x["payer"]]
|
|
w = want.setdefault(x["payer"], {"basis": x["payer"], "output": b["output"], "whole": b["whole"],
|
|
"buckets": sorted((p["terms"], p["works"], p["units"]) for p in b["payees"] if p["key"] == r["owner"]),
|
|
"lines": []})
|
|
w["lines"].append((x["class"], x["objects"], x["bytes"], revenue(acks, s["prices"], b["output"], x["class"]), x["amount"]))
|
|
got = {o["basis"]: o for o in r["outputs"]}
|
|
if "report_incomplete" not in DROPPED:
|
|
for d, w in want.items():
|
|
if d not in got or any(l[0] not in [g[0] for g in got[d]["lines"]] for l in w["lines"]):
|
|
raise Stop("report_incomplete", ["report"])
|
|
if "report_mismatch" not in DROPPED and (len(got) != len(want) or any(want.get(d) != o for d, o in got.items())):
|
|
raise Stop("report_mismatch", ["report"])
|
|
return {"outcome": "accept", "owner": r["owner"].hex(), "amount": r["amount"],
|
|
"outputs": [{"basis": o["basis"].hex(), "amount": sum(l[4] for l in o["lines"])} for o in r["outputs"]]}
|
|
|
|
|
|
def report_outcome(v, statements, cases, models):
|
|
try:
|
|
return verify_report(v, statements, cases, models)
|
|
except Stop as e:
|
|
return {"outcome": "refuse", "reason": e.reason, "at": e.at}
|
|
|
|
|
|
# ---------------------------------------------------------------- the corpus
|
|
|
|
def section(text, head):
|
|
at = text.index("\n" + head)
|
|
nxt = re.search(r"\n## ", 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, "## 9. Reasons").splitlines() if l.startswith("| `")]
|
|
|
|
|
|
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 stated(v):
|
|
return {k: v[k] for k in ("outcome", "reason", "at", "statements") if k in v}
|
|
|
|
|
|
def outcomes(doc):
|
|
"""Every vector's outcome as this verifier reaches it: (section, name, got, stated)."""
|
|
models = models_of(doc)
|
|
cases = {v["name"]: v for v in doc["cases"]}
|
|
statements = {v["name"]: v for v in doc["statements"]}
|
|
out = []
|
|
for v in doc["cases"]:
|
|
try:
|
|
got = case_outcome(v, models)
|
|
except Exception as e: # noqa: BLE001 — a malformed copy fails, whatever it breaks
|
|
got = {"outcome": "unreadable: %r" % (e,)}
|
|
out.append(("cases", v["name"], got, stated(v)))
|
|
for v in doc["statements"]:
|
|
try:
|
|
got = statement_outcome(v, cases, models)
|
|
except Exception as e: # noqa: BLE001
|
|
got = {"outcome": "unreadable: %r" % (e,)}
|
|
out.append(("statements", v["name"], got, v["outcome"]))
|
|
for v in doc["reports"]:
|
|
try:
|
|
got = report_outcome(v, statements, cases, models)
|
|
except Exception as e: # noqa: BLE001
|
|
got = {"outcome": "unreadable: %r" % (e,)}
|
|
out.append(("reports", v["name"], got, v["outcome"]))
|
|
return out
|
|
|
|
|
|
def check(doc):
|
|
try:
|
|
return ["%s/%s: %s" % (sec, n, json.dumps(got)[:300]) for sec, n, got, want in outcomes(doc) if got != want]
|
|
except (ValueError, KeyError, TypeError) as e:
|
|
return ["the corpus does not read: %r" % (e,)]
|
|
|
|
|
|
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."""
|
|
def named(d, sec, name):
|
|
return next(v for v in d[sec] if v["name"] == name)
|
|
|
|
def m(label, f):
|
|
d = copy.deepcopy(doc)
|
|
f(d)
|
|
return label, d
|
|
|
|
def retext(d, name, key, k, f):
|
|
v = named(d, "cases", name)
|
|
x = json.loads(v[key][k])
|
|
f(x)
|
|
v[key][k] = json.dumps(x)
|
|
|
|
def reattr(d, name, f):
|
|
v = named(d, "cases", name)
|
|
x = json.loads(v["attribution"])
|
|
f(x)
|
|
v["attribution"] = json.dumps(x)
|
|
|
|
def stmts(name, f):
|
|
return lambda d: f(named(d, "cases", name)["statements"])
|
|
|
|
return [
|
|
m("a digit of the style conditioning record's signature", lambda d: retext(d, "style", "conditioning", 0, lambda x: x.update(sig=flip_hex_digit(x["sig"], 5)))),
|
|
m("the style reference's weight", lambda d: retext(d, "style", "conditioning", 0, lambda x: x["references"][0].update(weight=299999))),
|
|
m("the reference image withheld from the style case", lambda d: named(d, "cases", "style")["contribution"].update(
|
|
states=named(d, "cases", "style")["contribution"]["states"][1:])),
|
|
m("the adapter's model file withheld", lambda d: named(d, "cases", "adapter_catalogue")["contribution"].update(
|
|
models=named(d, "cases", "adapter_catalogue")["contribution"]["models"][:1])),
|
|
m("a unit moved between two of the style shares", lambda d: reattr(d, "style", lambda x: (x["shares"][0].update(units=x["shares"][0]["units"] + 1),
|
|
x["shares"][1].update(units=x["shares"][1]["units"] - 1)))),
|
|
m("the catalogue's work list changed", lambda d: retext(d, "adapter_catalogue", "catalogues", 0, lambda x: x.update(works=x["works"][1:]))),
|
|
m("an owner's computed units stated one more", stmts("adapter_catalogue", lambda s: s["shares"]["computed"][0].update(units=s["shares"]["computed"][0]["units"] + 1))),
|
|
m("a reference's preference stated otherwise", stmts("style", lambda s: s["conditioning"][0]["references"][0]["usage"][0].update(preference="unknown"))),
|
|
m("the partial shares stated as compared", stmts("attested_partial", lambda s: s["shares"].update(compared=True))),
|
|
m("the equal split stated as declared weights", stmts("attested_equal", lambda s: s["conditioning"][0].update(weights="declared"))),
|
|
m("an owner's tier stated higher", stmts("attested_equal", lambda s: s["owners"][0].update(tier="checked"))),
|
|
m("the estimate stated as another share", stmts("style_estimate", lambda s: s["estimates"][0].update(share_ppm=1))),
|
|
m("a refusal stated at another place", lambda d: named(d, "cases", "not_in_catalogue").update(at=["conditioning", 0, "reference", 0])),
|
|
m("a refusal stated with another reason", lambda d: named(d, "cases", "conditioning_reference_omitted").update(reason="conditioning_invalid")),
|
|
m("an attribution line's amount stated otherwise", lambda d: named(d, "statements", "publisher_pays_owners")["outcome"]["attribution"][0].update(amount=1)),
|
|
m("a statement's given attributions emptied", lambda d: named(d, "statements", "publisher_pays_owners").update(given=[])),
|
|
m("a statement's byte changed", lambda d: named(d, "statements", "publisher_pays_owners").update(
|
|
statement=flip_hex_digit(named(d, "statements", "publisher_pays_owners")["statement"], 90))),
|
|
m("a report's amount stated otherwise", lambda d: named(d, "reports", "label_report_owner")["outcome"].update(amount=0)),
|
|
m("a report's audit withdrawn", lambda d: named(d, "reports", "report_leaves_output_out").update(audit=None)),
|
|
m("the owner's terms withdrawn from a report", lambda d: named(d, "reports", "label_report_owner").update(terms=[])),
|
|
("the second model file changed by one byte", dict(copy.deepcopy(doc), models=[doc["models"][0], dict(doc["models"][1], file=doc["models"][1]["file"].replace("1", "2", 1))])),
|
|
]
|
|
|
|
|
|
# The checks a weakened verifier drops: this extension's reasons it raises with fail(), and the
|
|
# attribution-line checks of settlement §7.6.
|
|
DROPPABLE = [r for r in REASONS] + [r for r in REUSED if r not in ("record_malformed", "bad_key", "bad_signature")] + \
|
|
["attribution-covered", "attribution-figures"]
|
|
|
|
|
|
def main(corpus_dir):
|
|
fails = []
|
|
st = ED._self_test()
|
|
print(("PASS" if not st else "FAIL") + " Ed25519 strict rule self-test " + ",".join(st))
|
|
fails += ["ed25519 self-test"] if st else []
|
|
gen_ok = CON.self_test_generator()
|
|
print(("PASS" if gen_ok else "FAIL") + " the generator's arithmetic gives its conformance vector's SHA-256")
|
|
fails += [] if gen_ok else ["the generator's conformance vector"]
|
|
print(("PASS" if len(REASONS) == 17 else "FAIL") + " §9 read from the extension: %d reasons" % len(REASONS))
|
|
fails += [] if len(REASONS) == 17 else ["§9 does not read"]
|
|
with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as f:
|
|
doc = json.load(f)
|
|
extra = sorted(set(os.listdir(corpus_dir)) - {"vectors.json", "README.md"})
|
|
print(("PASS" if not extra else "FAIL") + " no file but vectors.json and README.md")
|
|
fails += ["an extra file: %s" % n for n in extra]
|
|
seen = set()
|
|
for sec, name, got, want in outcomes(doc):
|
|
ok = got == want
|
|
if want.get("reason"):
|
|
seen.add(want["reason"])
|
|
print(("PASS" if ok else "FAIL") + " %s %s: %s" % (sec, name, got["outcome"] + (" %s at %s" % (got.get("reason"), got.get("at")) if "reason" in got else "")))
|
|
if not ok:
|
|
fails.append("%s/%s: %s, the corpus states %s" % (sec, name, json.dumps(got)[:600], json.dumps(want)[:600]))
|
|
missing = [r for r in REASONS + REUSED if r not in seen]
|
|
print(("PASS" if not missing else "FAIL") + " every reason of §9, and every code it reuses, refuses a vector")
|
|
fails += ["no vector refused with %s" % r for r in missing]
|
|
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 DROPPABLE:
|
|
DROPPED.clear()
|
|
DROPPED.add(r)
|
|
caught = check(doc)
|
|
DROPPED.clear()
|
|
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 x in fails:
|
|
print("FAIL " + x)
|
|
print(("PASS" if not fails else "FAIL") + " influence corpus: %d cases, %d statements, %d reports" % (len(doc["cases"]), len(doc["statements"]), len(doc["reports"])))
|
|
return 1 if fails else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) != 2:
|
|
print("usage: influence_verify.py <live-conformance/influence>", file=sys.stderr)
|
|
sys.exit(2)
|
|
sys.exit(main(sys.argv[1]))
|