3bbac2e2b1
C3's synthesis subject `samples` (a picture a party presents and does not publish) met takedown's exhaustive match once both slices were in one tree (E0004). Such samples are no item a holder serves or keeps for others, so a closure claim whose subject they are adds no subject and needs no purge: takedown §4.2 rule 3, the reference and its second implementation say so, and the corpus's shared world gains such a claim, so every case exercises it. Co-Authored-By: Claude Opus 5.5 <[email protected]>
854 lines
37 KiB
Python
854 lines
37 KiB
Python
#!/usr/bin/env python3
|
|
"""takedown_verify.py — the second implementation of extensions/takedown.md (wai.media.takedown).
|
|
|
|
Written from the extension's text, sharing no code with wai-rs. It reads every record of §5
|
|
under §2's encodings, computes the derivation closure of §6 from the derivation records (read
|
|
with the deriving-edge corpus's second implementation, edge-conformance/edge_verify.py), the
|
|
origin claims (read with the synthesis corpus's, live-conformance/synthesis_verify.py), the
|
|
delivery receipts of the groups it reaches (read with edge_verify.py too) and the
|
|
acknowledgements that show who served or received an object (read with the settlement corpus's,
|
|
live-conformance/settlement_verify.py), holds the takedown's
|
|
stated set to it, finds the holders and the purges that cover them (§7), computes the time to
|
|
removal against the clock the registry gives (§4, §7.3), and recomputes the counts (§9). Every
|
|
signature is checked with WAI's strict Ed25519 rule (claim-conformance/_ed25519.py).
|
|
|
|
Then it requires every mutated copy of the corpus to fail, and, with each of its own checks
|
|
dropped in turn, requires the corpus to catch the weakened verifier.
|
|
|
|
python3 -I takedown_verify.py <live-conformance/takedown>
|
|
|
|
Standard library only. Exit status 0 when everything holds, 1 otherwise.
|
|
"""
|
|
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"))
|
|
sys.path.insert(0, os.path.join(HERE, "..", "edge-conformance"))
|
|
import _ed25519 as ED # noqa: E402
|
|
from _blake3 import blake3 # noqa: E402
|
|
import edge_verify as EV # noqa: E402
|
|
import synthesis_verify as SYV # noqa: E402
|
|
import settlement_verify as SEV # noqa: E402
|
|
|
|
EXTENSION = os.path.join(HERE, "..", "extensions", "takedown.md")
|
|
PUBLISHED = os.path.join(HERE, "..", "registry", "takedown-clocks.json")
|
|
|
|
SIGN = b"wai:takedown\x01" # §2
|
|
LINK = b"wai:takedown-link\x01" # §2
|
|
MAX_INT = (1 << 53) - 1
|
|
ROLES = ("origin", "store", "relay", "cache", "edge")
|
|
|
|
CHECKS = (
|
|
"registry-shape", "signature", "registry-link", "policy-clocks", "takedown-policy", "takedown-signer",
|
|
"notice-held", "notice-targets", "clock-in-policy", "clock-consistent", "records", "claims", "acknowledgements",
|
|
"purge-takedown", "rule-group", "rule-derivation", "rule-claim", "group-expanded",
|
|
"exact-supported", "match-not-exact", "exact-complete", "match-held", "match-anchored",
|
|
"store-holders", "deriver-holders", "claimant-holders", "publisher-holders", "served-holders",
|
|
"holder-only", "group-coverage", "deadline", "purges-complete", "window",
|
|
"counts-as-of", "counts-rows",
|
|
)
|
|
DROPPED = set()
|
|
|
|
|
|
def on(check):
|
|
assert check in CHECKS, check
|
|
return check not in DROPPED
|
|
|
|
|
|
class Refuse(Exception):
|
|
def __init__(self, reason, at):
|
|
super().__init__(reason)
|
|
self.reason, self.at = reason, at
|
|
|
|
|
|
class Bad(Exception):
|
|
"""A member that breaks its rule: the record is record_malformed."""
|
|
|
|
|
|
def need(cond):
|
|
if not cond:
|
|
raise Bad()
|
|
|
|
|
|
# ---------------------------------------------------------------- §2 encodings
|
|
|
|
def is_hex(v, n):
|
|
return isinstance(v, str) and re.fullmatch("[0-9a-f]{%d}" % (2 * n), v) is not None
|
|
|
|
|
|
def is_int(v):
|
|
return isinstance(v, int) and not isinstance(v, bool) and 0 <= v <= MAX_INT
|
|
|
|
|
|
def text(v, lo, hi):
|
|
need(isinstance(v, str) and lo <= len(v) <= hi)
|
|
return v
|
|
|
|
|
|
def days_in(y, m):
|
|
if m == 2:
|
|
return 29 if y % 4 == 0 and (y % 100 != 0 or y % 400 == 0) else 28
|
|
return 30 if m in (4, 6, 9, 11) else 31
|
|
|
|
|
|
def date(s):
|
|
m = isinstance(s, str) and re.fullmatch(r"([0-9]{4})-([0-9]{2})-([0-9]{2})", s)
|
|
need(bool(m))
|
|
y, mo, d = (int(m.group(i)) for i in (1, 2, 3))
|
|
need(1 <= mo <= 12 and 1 <= d <= days_in(y, mo))
|
|
return y, mo, d
|
|
|
|
|
|
def time_value(s):
|
|
"""§2: `YYYY-MM-DDTHH:MM:SSZ`, a year of 1970 or later, a second of at most 59: its value."""
|
|
m = isinstance(s, str) and re.fullmatch(r"([0-9]{4}-[0-9]{2}-[0-9]{2})T([0-9]{2}):([0-9]{2}):([0-9]{2})Z", s)
|
|
need(bool(m))
|
|
y, mo, d = date(m.group(1))
|
|
hh, mi, ss = int(m.group(2)), int(m.group(3)), int(m.group(4))
|
|
need(y >= 1970 and hh <= 23 and mi <= 59 and ss <= 59)
|
|
return calendar.timegm((y, mo, d, hh, mi, ss, 0, 0, 0))
|
|
|
|
|
|
def clock_id(s):
|
|
need(isinstance(s, str) and 1 <= len(s.encode()) <= 64 and re.fullmatch(r"[a-z0-9]+(-[a-z0-9]+)*", s) is not None)
|
|
return s
|
|
|
|
|
|
_SIGS = {}
|
|
|
|
|
|
def ed_verify(pk, msg, sig):
|
|
"""WAI's strict Ed25519 rule (_ed25519.verify), remembered: the mutations and the dropped
|
|
checks verify the corpus many times over, and a verdict is a function of its three inputs."""
|
|
k = (pk, msg, sig)
|
|
if k not in _SIGS:
|
|
_SIGS[k] = ED.verify(pk, msg, sig)
|
|
return _SIGS[k]
|
|
|
|
|
|
def link_of(v):
|
|
return blake3(LINK + EV.jcs(v).encode("utf-8"))
|
|
|
|
|
|
def members(v, names):
|
|
need(isinstance(v, dict) and set(v) == set(names))
|
|
return v
|
|
|
|
|
|
# ---------------------------------------------------------------- §3 subjects
|
|
|
|
def subject(v):
|
|
need(isinstance(v, dict))
|
|
t = v.get("type")
|
|
if t == "object":
|
|
members(v, ["type", "content_hash"])
|
|
need(is_hex(v["content_hash"], 32))
|
|
elif t == "group":
|
|
members(v, ["type", "receipt"])
|
|
need(is_hex(v["receipt"], 32))
|
|
elif t == "track":
|
|
members(v, ["type", "track", "from_group"])
|
|
need(is_hex(v["track"], 32) and is_int(v["from_group"]))
|
|
elif t == "take":
|
|
members(v, ["type", "take"])
|
|
need(is_hex(v["take"], 32))
|
|
else:
|
|
raise Bad()
|
|
return v
|
|
|
|
|
|
def key(s):
|
|
"""A subject's JCS form: lists of subjects are in its byte order (§3)."""
|
|
return EV.jcs(s)
|
|
|
|
|
|
def subject_list(v, lo, hi):
|
|
need(isinstance(v, list) and lo <= len(v) <= hi)
|
|
out = [subject(x) for x in v]
|
|
ks = [key(x).encode() for x in out]
|
|
need(all(a < b for a, b in zip(ks, ks[1:])))
|
|
return out
|
|
|
|
|
|
def obj(h):
|
|
return {"type": "object", "content_hash": h}
|
|
|
|
|
|
# ---------------------------------------------------------------- records
|
|
|
|
def strict(text_):
|
|
v = EV.strict(text_)
|
|
need(EV.in_subset(v))
|
|
return v
|
|
|
|
|
|
def read_registry(t):
|
|
try:
|
|
v = strict(t)
|
|
members(v, ["kind", "clocks"])
|
|
need(v["kind"] == "wai-takedown-clocks")
|
|
cs = v["clocks"]
|
|
need(isinstance(cs, list) and 1 <= len(cs) <= 64)
|
|
clocks = {}
|
|
ids = []
|
|
for c in cs:
|
|
members(c, ["id", "type", "duration_s", "starts", "instrument", "section", "applies_from"])
|
|
clock_id(c["id"])
|
|
need(c["type"] in ("deadline", "reply_window", "unfixed"))
|
|
d = c["duration_s"]
|
|
need(d is None or (is_int(d) and d >= 1))
|
|
need((c["type"] == "unfixed") == (d is None))
|
|
need(c["starts"] in ("receipt", "statement"))
|
|
text(c["instrument"], 1, 500)
|
|
text(c["section"], 1, 200)
|
|
date(c["applies_from"])
|
|
clocks[c["id"]] = c
|
|
ids.append(c["id"].encode())
|
|
need(all(a < b for a, b in zip(ids, ids[1:])))
|
|
except (Bad, ValueError, KeyError, TypeError, AttributeError):
|
|
if on("registry-shape"):
|
|
raise Refuse("record_malformed", ["registry"])
|
|
raise Refuse("record_malformed", ["registry", "unread"]) from None
|
|
return {"link": link_of(v), "clocks": clocks}
|
|
|
|
|
|
def read_signed(t, kind, names, fields, at):
|
|
"""§2: the shape, the kind's own members, then the signature."""
|
|
try:
|
|
v = strict(t)
|
|
members(v, names)
|
|
need(v["kind"] == kind and is_hex(v["signer_pubkey"], 32) and is_hex(v["sig"], 64))
|
|
r = fields(v)
|
|
except (Bad, ValueError, KeyError, TypeError, AttributeError):
|
|
raise Refuse("record_malformed", at)
|
|
bad = ed_verify(bytes.fromhex(v["signer_pubkey"]), SIGN + EV.jcs({k: x for k, x in v.items() if k != "sig"}).encode("utf-8"),
|
|
bytes.fromhex(v["sig"]))
|
|
if bad and on("signature"):
|
|
raise Refuse(bad, at)
|
|
r.update(link=link_of(v), signer=v["signer_pubkey"])
|
|
return r
|
|
|
|
|
|
def policy_fields(v):
|
|
u = v["use"]
|
|
need(isinstance(u, list) and len(u) <= 64)
|
|
for x in u:
|
|
clock_id(x)
|
|
need(all(a.encode() < b.encode() for a, b in zip(u, u[1:])))
|
|
st = v["stores"]
|
|
need(isinstance(st, list) and 1 <= len(st) <= 256)
|
|
keys = []
|
|
for s in st:
|
|
members(s, ["key", "role"])
|
|
need(is_hex(s["key"], 32) and s["role"] in ROLES)
|
|
keys.append(s["key"])
|
|
need(all(a < b for a, b in zip(keys, keys[1:])))
|
|
need(is_hex(v["clocks"], 32))
|
|
time_value(v["issued_at"])
|
|
return {"clocks": v["clocks"], "use": list(u), "stores": keys}
|
|
|
|
|
|
def notice_fields(v):
|
|
need(v["type"] in ("request", "order") and v["ground"] in ("illegal", "terms"))
|
|
text(v["ground_ref"], 0, 500)
|
|
text(v["statement"], 1, 5000)
|
|
text(v["contact"], 0, 500)
|
|
return {"type": v["type"], "targets": subject_list(v["targets"], 1, 16), "issued_at": time_value(v["issued_at"])}
|
|
|
|
|
|
def takedown_fields(v):
|
|
m = v["matched"]
|
|
need(isinstance(m, list) and len(m) <= 256)
|
|
for e in m:
|
|
members(e, ["match", "candidate"])
|
|
need(is_hex(e["match"], 32) and is_hex(e["candidate"], 32))
|
|
need(all(a["match"] < b["match"] for a, b in zip(m, m[1:])))
|
|
need(v["verdict"] in ("remove", "disable"))
|
|
a = members(v["automated"], ["detection", "decision"])
|
|
need(a["detection"] in ("yes", "no") and a["decision"] in ("not", "partially", "fully"))
|
|
text(v["explanation"], 1, 2000)
|
|
text(v["facts"], 1, 5000)
|
|
need(is_hex(v["policy"], 32) and (v["notice"] is None or is_hex(v["notice"], 32)))
|
|
if v["clock"] is not None:
|
|
clock_id(v["clock"])
|
|
cs = None if v["clock_start"] is None else time_value(v["clock_start"])
|
|
ra = None if v["reply_at"] is None else time_value(v["reply_at"])
|
|
need(v["ground"] in ("illegal", "terms") and is_int(v["category"]) and 1 <= v["category"] <= 16)
|
|
return {"policy": v["policy"], "notice": v["notice"], "clock": v["clock"], "clock_start": cs, "reply_at": ra,
|
|
"targets": subject_list(v["targets"], 1, 16), "exact": subject_list(v["exact"], 1, 4096), "matched": m,
|
|
"ground": v["ground"], "category": v["category"], "automated": a["decision"] != "not",
|
|
"decided_at": time_value(v["decided_at"])}
|
|
|
|
|
|
def match_fields(v):
|
|
need(is_hex(v["source"], 32) and is_hex(v["candidate"], 32) and v["source"] != v["candidate"])
|
|
SYV.term(v["method"])
|
|
need(is_int(v["distance"]) and is_int(v["threshold"]) and v["distance"] <= v["threshold"])
|
|
time_value(v["issued_at"])
|
|
return {"source": v["source"], "candidate": v["candidate"]}
|
|
|
|
|
|
def purge_fields(v):
|
|
need(is_hex(v["takedown"], 32))
|
|
return {"takedown": v["takedown"], "subjects": subject_list(v["subjects"], 1, 4096), "purged_at": time_value(v["purged_at"])}
|
|
|
|
|
|
ROW = ["clock", "ground", "category", "source", "takedowns", "met", "missed", "pending", "unfixed", "automated", "median_s"]
|
|
|
|
|
|
def row_order(r):
|
|
return (r["clock"] is not None, (r["clock"] or "").encode(), r["ground"], r["category"], r["source"])
|
|
|
|
|
|
def counts_fields(v):
|
|
td = v["takedowns"]
|
|
need(isinstance(td, list) and len(td) <= 4096 and all(is_hex(x, 32) for x in td))
|
|
need(all(a < b for a, b in zip(td, td[1:])))
|
|
rows = v["rows"]
|
|
need(isinstance(rows, list))
|
|
for r in rows:
|
|
members(r, ROW)
|
|
if r["clock"] is not None:
|
|
clock_id(r["clock"])
|
|
need(r["ground"] in ("illegal", "terms") and is_int(r["category"]) and 1 <= r["category"] <= 16)
|
|
need(r["source"] in ("order", "request", "own"))
|
|
need(all(is_int(r[k]) for k in ROW[4:10]) and (r["median_s"] is None or is_int(r["median_s"])))
|
|
need(r["takedowns"] >= 1 and r["met"] + r["missed"] + r["pending"] + r["unfixed"] == r["takedowns"])
|
|
need(r["automated"] <= r["takedowns"])
|
|
need(all(row_order(a) < row_order(b) for a, b in zip(rows, rows[1:])))
|
|
f, t = time_value(v["from"]), time_value(v["to"])
|
|
need(f < t and is_hex(v["policy"], 32))
|
|
time_value(v["issued_at"])
|
|
return {"policy": v["policy"], "from": f, "to": t, "as_of": time_value(v["as_of"]), "takedowns": td, "rows": rows}
|
|
|
|
|
|
# ---------------------------------------------------------------- §6 group expansion
|
|
|
|
def expand(receipts, g):
|
|
"""§6.3: (publisher, objects) or None."""
|
|
gr = None
|
|
for r in receipts:
|
|
if isinstance(r, dict) and r.get("kind") == "group":
|
|
x = EV.read_group(r)
|
|
if x is not None and EV.link(x).hex() == g:
|
|
gr = x
|
|
break
|
|
if gr is None or not EV.group_verifies(gr) or gr["leaf_count"] > 4096:
|
|
return None
|
|
objs = [EV.read_object(r) for r in receipts if isinstance(r, dict) and r.get("kind") == "object"]
|
|
out = []
|
|
for p in range(gr["leaf_count"]):
|
|
found = next((o for o in objs if o is not None and o["track_alias"] == gr["track_alias"] and o["group_id"] == gr["group_id"]
|
|
and o["object_id"] == p and EV.included(o, gr)), None)
|
|
if found is None:
|
|
return None
|
|
out.append(found["content_hash"])
|
|
return gr["signer_pubkey"], out
|
|
|
|
|
|
# ---------------------------------------------------------------- §8 one case
|
|
|
|
def evaluate(case, as_of):
|
|
reg = read_registry(case["registry"])
|
|
pol = read_signed(case["policy"], "wai-takedown-policy", ["kind", "clocks", "use", "stores", "issued_at", "signer_pubkey", "sig"],
|
|
policy_fields, ["policy"])
|
|
if pol["clocks"] != reg["link"].hex() and on("registry-link"):
|
|
raise Refuse("registry_mismatch", ["policy"])
|
|
if any(u not in reg["clocks"] for u in pol["use"]) and on("policy-clocks"):
|
|
raise Refuse("clock_unknown", ["policy"])
|
|
notices = [read_signed(t, "wai-takedown-notice", ["kind", "type", "targets", "ground", "ground_ref", "statement", "contact",
|
|
"issued_at", "signer_pubkey", "sig"], notice_fields, ["notice", i])
|
|
for i, t in enumerate(case.get("notices", []))]
|
|
at = ["takedown"]
|
|
td = read_signed(case["takedown"], "wai-takedown",
|
|
["kind", "policy", "notice", "clock", "clock_start", "reply_at", "targets", "verdict", "exact", "matched", "ground",
|
|
"category", "automated", "explanation", "facts", "decided_at", "signer_pubkey", "sig"], takedown_fields, at)
|
|
if td["policy"] != pol["link"].hex() and on("takedown-policy"):
|
|
raise Refuse("policy_mismatch", at)
|
|
if td["signer"] != pol["signer"] and on("takedown-signer"):
|
|
raise Refuse("not_authority", at)
|
|
notice = None
|
|
if td["notice"] is not None:
|
|
notice = next((n for n in notices if n["link"].hex() == td["notice"]), None)
|
|
if notice is None and on("notice-held"):
|
|
raise Refuse("notice_unknown", at)
|
|
if notice is not None:
|
|
nk = {key(x) for x in notice["targets"]}
|
|
if (not all(key(x) in nk for x in td["targets"]) or notice["issued_at"] > td["decided_at"]) and on("notice-targets"):
|
|
raise Refuse("notice_mismatch", at)
|
|
clock = None
|
|
if td["clock"] is not None:
|
|
if td["clock"] not in pol["use"] and on("clock-in-policy"):
|
|
raise Refuse("clock_unknown", at)
|
|
clock = reg["clocks"].get(td["clock"])
|
|
cs, ra = td["clock_start"], td["reply_at"]
|
|
ok = (td["clock"] is None) == (cs is None)
|
|
ok = ok and (cs is None or cs <= td["decided_at"])
|
|
if ok and clock is not None and cs is not None and clock["starts"] == "receipt":
|
|
ok = notice is not None and notice["issued_at"] <= cs
|
|
if ok and ra is not None:
|
|
ok = clock is not None and clock["type"] == "reply_window" and cs is not None and ra >= cs
|
|
if not ok and on("clock-consistent"):
|
|
raise Refuse("clock_invalid", at)
|
|
|
|
records = []
|
|
for i, t in enumerate(case.get("records", [])):
|
|
try:
|
|
r = EV.read_record(t, i)
|
|
good = ed_verify(bytes.fromhex(r["signer_pubkey"]), EV.signing_input(r["raw"]), bytes.fromhex(r["sig"])) is None
|
|
except EV.Refuse:
|
|
r, good = None, False
|
|
if not good and on("records"):
|
|
raise Refuse("evidence_invalid", ["record", i])
|
|
if r is not None:
|
|
records.append(r)
|
|
claims = []
|
|
for i, t in enumerate(case.get("claims", [])):
|
|
try:
|
|
c = SYV.read(t, "wai-origin", SYV.ORIGIN, SYV.origin_fields)
|
|
except SYV.Refuse:
|
|
if on("claims"):
|
|
raise Refuse("evidence_invalid", ["claim", i])
|
|
continue
|
|
claims.append(c)
|
|
acks = []
|
|
for i, a in enumerate(case.get("acknowledgements", [])):
|
|
try:
|
|
rec = SEV.read_ack(bytes.fromhex(a["record"]))
|
|
lists = [bytes.fromhex(x) for x in a["items"]]
|
|
need(len(lists) == len(rec["entries"]))
|
|
hashes = []
|
|
for e, raw in zip(rec["entries"], lists):
|
|
items = SEV.read_items(raw)
|
|
need(SEV.recomputes(e, items, raw))
|
|
hashes += [h.hex() for h, _, _ in items]
|
|
hop = rec["downstream"].hex() if rec["mode"] == SEV.ACK and rec["role"] == SEV.HOP else None
|
|
acks.append((rec["upstream"].hex(), hop, hashes))
|
|
except (SEV.Refuse, Bad, ValueError, KeyError, TypeError):
|
|
if on("acknowledgements"):
|
|
raise Refuse("evidence_invalid", ["acknowledgement", i])
|
|
matches = [read_signed(t, "wai-takedown-match", ["kind", "source", "candidate", "method", "distance", "threshold", "issued_at",
|
|
"signer_pubkey", "sig"], match_fields, ["match", i])
|
|
for i, t in enumerate(case.get("matches", []))]
|
|
purges = []
|
|
for i, t in enumerate(case.get("purges", [])):
|
|
p = read_signed(t, "wai-takedown-purge", ["kind", "takedown", "subjects", "purged_at", "signer_pubkey", "sig"], purge_fields,
|
|
["purge", i])
|
|
if (p["takedown"] != td["link"].hex() or p["purged_at"] > as_of) and on("purge-takedown"):
|
|
raise Refuse("purge_mismatch", ["purge", i])
|
|
purges.append(p)
|
|
|
|
# §6.1: the closure, by its rules until none adds a subject.
|
|
closure = {key(x): x for x in td["targets"]}
|
|
groups = {}
|
|
via_records, via_claims = set(), set()
|
|
|
|
def claim_subject(c):
|
|
# Samples a party presents and does not publish (synthesis §3.1) are no item a holder
|
|
# serves or keeps for others: such a claim adds nothing (§4.2 rule 3).
|
|
s = c["value"]["subject"]
|
|
return None if s.get("type") == "samples" else {k: s[k] for k in s}
|
|
|
|
while True:
|
|
before = len(closure)
|
|
for s in list(closure.values()):
|
|
if s["type"] == "group" and s["receipt"] not in groups:
|
|
e = expand(case.get("receipts", []), s["receipt"])
|
|
groups[s["receipt"]] = e
|
|
if e is not None and on("rule-group"):
|
|
for h in e[1]:
|
|
closure[key(obj(h))] = obj(h)
|
|
objects = {s["content_hash"] for s in closure.values() if s["type"] == "object"}
|
|
for i, r in enumerate(records):
|
|
if any(x["content_hash"] in objects for x in r["inputs"]) and on("rule-derivation"):
|
|
via_records.add(i)
|
|
closure[key(obj(r["output"]["content_hash"]))] = obj(r["output"]["content_hash"])
|
|
for i, c in enumerate(claims):
|
|
if any(x["content_hash"].hex() in objects for x in c["inputs"]) and on("rule-claim"):
|
|
via_claims.add(i)
|
|
s = claim_subject(c)
|
|
if s is not None:
|
|
closure[key(s)] = s
|
|
unexpanded = any(s["type"] == "group" and s["receipt"] not in groups for s in closure.values())
|
|
if len(closure) == before and not unexpanded:
|
|
break
|
|
for g in sorted(groups):
|
|
if groups[g] is None and on("group-expanded"):
|
|
raise Refuse("group_unexpanded", ["group", g])
|
|
objects = {s["content_hash"] for s in closure.values() if s["type"] == "object"}
|
|
|
|
# §6.4, §6.5: the stated set and the matches.
|
|
candidates = {m["candidate"] for m in matches}
|
|
for k, s in enumerate(td["exact"]):
|
|
if key(s) not in closure:
|
|
if s["type"] == "object" and s["content_hash"] in candidates and on("match-not-exact"):
|
|
raise Refuse("match_as_exact", ["exact", k])
|
|
if on("exact-supported"):
|
|
raise Refuse("closure_unsupported", ["exact", k])
|
|
stated = {key(s) for s in td["exact"]}
|
|
for k, ks in enumerate(sorted(closure, key=lambda x: x.encode())):
|
|
if ks not in stated and on("exact-complete"):
|
|
raise Refuse("closure_incomplete", ["closure", k])
|
|
for k, e in enumerate(td["matched"]):
|
|
m = next((m for m in matches if m["link"].hex() == e["match"] and m["candidate"] == e["candidate"]), None)
|
|
if m is None:
|
|
if on("match-held"):
|
|
raise Refuse("match_unknown", ["matched", k])
|
|
continue
|
|
if m["source"] not in objects and on("match-anchored"):
|
|
raise Refuse("match_unanchored", ["matched", k])
|
|
|
|
# §7.1: the required pairs.
|
|
required, roles = {}, {}
|
|
|
|
def needs(k, s, role):
|
|
roles.setdefault(k, set()).add(role)
|
|
required[(k, key(s))] = s
|
|
|
|
if on("store-holders"):
|
|
for k in pol["stores"]:
|
|
for s in closure.values():
|
|
needs(k, s, "store")
|
|
if on("deriver-holders"):
|
|
for i in sorted(via_records):
|
|
r = records[i]
|
|
for x in r["inputs"]:
|
|
if x["content_hash"] in objects:
|
|
needs(r["signer_pubkey"], obj(x["content_hash"]), "deriver")
|
|
needs(r["signer_pubkey"], obj(r["output"]["content_hash"]), "deriver")
|
|
if on("claimant-holders"):
|
|
for i in sorted(via_claims):
|
|
c = claims[i]
|
|
for x in c["inputs"]:
|
|
if x["content_hash"].hex() in objects:
|
|
needs(c["signer"].hex(), obj(x["content_hash"].hex()), "claimant")
|
|
if claim_subject(c) is not None:
|
|
needs(c["signer"].hex(), claim_subject(c), "claimant")
|
|
if on("publisher-holders"):
|
|
for g, e in groups.items():
|
|
if key({"type": "group", "receipt": g}) in closure:
|
|
needs(e[0], {"type": "group", "receipt": g}, "publisher")
|
|
if on("served-holders"):
|
|
for up, hop, hashes in acks:
|
|
for h in hashes:
|
|
if h in objects:
|
|
needs(up, obj(h), "served")
|
|
if hop is not None:
|
|
needs(hop, obj(h), "received")
|
|
matched_required = {}
|
|
cands = {e["candidate"] for e in td["matched"]}
|
|
for k in pol["stores"]:
|
|
for c in cands:
|
|
matched_required[(k, key(obj(c)))] = obj(c)
|
|
for up, hop, hashes in acks:
|
|
for h in hashes:
|
|
if h in cands:
|
|
for k in [up] + ([hop] if hop is not None else []):
|
|
matched_required[(k, key(obj(h)))] = obj(h)
|
|
|
|
# §7.2: what a purge covers.
|
|
def covers(p, s):
|
|
named = {key(x) for x in p["subjects"]}
|
|
if key(s) in named:
|
|
return True
|
|
if not on("group-coverage"):
|
|
return False
|
|
if s["type"] == "object":
|
|
return any(x["type"] == "group" and groups.get(x["receipt"]) and s["content_hash"] in groups[x["receipt"]][1]
|
|
for x in p["subjects"])
|
|
if s["type"] == "group" and groups.get(s["receipt"]):
|
|
return all(key(obj(h)) in named for h in groups[s["receipt"]][1])
|
|
return False
|
|
|
|
def pair_time(k, s):
|
|
ts = [p["purged_at"] for p in purges if (p["signer"] == k or not on("holder-only")) and covers(p, s)]
|
|
return min(ts) if ts else None
|
|
|
|
missing, times = [], []
|
|
for (k, ks) in sorted(required, key=lambda x: (x[0], x[1].encode())):
|
|
tt = pair_time(k, required[(k, ks)])
|
|
if tt is None:
|
|
missing.append({"holder": k, "subject": required[(k, ks)]})
|
|
else:
|
|
times.append(tt)
|
|
m_missing, m_times = [], []
|
|
for (k, ks) in sorted(matched_required, key=lambda x: (x[0], x[1].encode())):
|
|
tt = pair_time(k, matched_required[(k, ks)])
|
|
if tt is None:
|
|
m_missing.append({"holder": k, "subject": matched_required[(k, ks)]})
|
|
else:
|
|
m_times.append(tt)
|
|
uncounted = []
|
|
for i, p in enumerate(purges):
|
|
for s in p["subjects"]:
|
|
if (p["signer"], key(s)) not in required and (p["signer"], key(s)) not in matched_required:
|
|
uncounted.append({"purge": i, "subject": s})
|
|
|
|
# §7.3: the time to removal and the clock.
|
|
complete = not missing
|
|
first = min(times) if times else None
|
|
last = max(times) if times else None
|
|
start = cs if cs is not None else td["decided_at"]
|
|
ttr = max(0, last - start) if complete and last is not None else None
|
|
deadline = window_end = None
|
|
reason = None
|
|
if clock is not None and clock["type"] == "deadline":
|
|
deadline = start + clock["duration_s"]
|
|
if complete and last is not None and (last <= deadline or not on("deadline")):
|
|
verdict = "met"
|
|
elif complete:
|
|
verdict, reason = "missed", "deadline_missed"
|
|
elif as_of > deadline and on("purges-complete"):
|
|
verdict, reason = "missed", "purge_missing"
|
|
else:
|
|
verdict = "pending"
|
|
elif clock is not None and clock["type"] == "reply_window":
|
|
window_end = start + clock["duration_s"] if ra is None else min(ra, start + clock["duration_s"])
|
|
if first is not None and first < window_end and on("window"):
|
|
verdict, reason = "missed", "window_not_observed"
|
|
else:
|
|
verdict = "met" if complete else "pending"
|
|
else:
|
|
verdict = "unfixed" if complete else "pending"
|
|
|
|
facts = {
|
|
"closure": [closure[k] for k in sorted(closure, key=lambda x: x.encode())],
|
|
"holders": [{"key": k, "roles": sorted(roles[k]), "required": sum(1 for (kk, _) in required if kk == k)} for k in sorted(roles)],
|
|
"missing": missing,
|
|
"uncounted": uncounted,
|
|
"clock": td["clock"],
|
|
"start": start,
|
|
"deadline": deadline,
|
|
"window_end": window_end,
|
|
"first_purge": first,
|
|
"last_purge": last,
|
|
"time_to_removal_s": ttr,
|
|
"verdict": verdict,
|
|
"matched": {"candidates": sorted({e["candidate"] for e in td["matched"]}), "missing": m_missing,
|
|
"last_purge": max(m_times) if m_times else None},
|
|
}
|
|
if reason is not None:
|
|
out = {"outcome": "missed", "reason": reason, "facts": facts}
|
|
elif verdict == "pending":
|
|
out = {"outcome": "pending", "facts": facts}
|
|
else:
|
|
out = {"outcome": "accept", "facts": facts}
|
|
source = "own" if notice is None else notice["type"]
|
|
return {"policy": pol, "takedown": td, "source": source, "outcome": out}
|
|
|
|
|
|
def verify_case(case):
|
|
try:
|
|
return evaluate(case, case["as_of"])["outcome"]
|
|
except Refuse as e:
|
|
return {"outcome": "refuse", "reason": e.reason, "at": e.at}
|
|
|
|
|
|
# ---------------------------------------------------------------- §9 counts
|
|
|
|
def rows_of(done):
|
|
by = {}
|
|
for e in done:
|
|
t = e["takedown"]
|
|
k = (t["clock"] is not None, (t["clock"] or "").encode(), t["ground"], t["category"], e["source"])
|
|
r, ttr = by.setdefault(k, ({"clock": t["clock"], "ground": t["ground"], "category": t["category"], "source": e["source"],
|
|
"takedowns": 0, "met": 0, "missed": 0, "pending": 0, "unfixed": 0, "automated": 0,
|
|
"median_s": None}, []))
|
|
r["takedowns"] += 1
|
|
r["automated"] += 1 if t["automated"] else 0
|
|
f = e["outcome"]["facts"]
|
|
r[f["verdict"]] += 1
|
|
if f["time_to_removal_s"] is not None:
|
|
ttr.append(f["time_to_removal_s"])
|
|
out = []
|
|
for k in sorted(by):
|
|
r, ttr = by[k]
|
|
ttr.sort()
|
|
r["median_s"] = ttr[(len(ttr) - 1) // 2] if ttr else None
|
|
out.append(r)
|
|
return out
|
|
|
|
|
|
def verify_counts(counts_text, cases):
|
|
c = read_signed(counts_text, "wai-takedown-counts", ["kind", "policy", "from", "to", "as_of", "takedowns", "rows", "issued_at",
|
|
"signer_pubkey", "sig"], counts_fields, ["counts"])
|
|
done = []
|
|
for i, case in enumerate(cases):
|
|
try:
|
|
done.append(evaluate(case, c["as_of"] if on("counts-as-of") else case["as_of"]))
|
|
except Refuse as e:
|
|
raise Refuse(e.reason, ["case", i] + e.at)
|
|
for e in done:
|
|
if e["policy"]["link"].hex() != c["policy"]:
|
|
raise Refuse("policy_mismatch", ["counts"])
|
|
if e["policy"]["signer"] != c["signer"]:
|
|
raise Refuse("not_authority", ["counts"])
|
|
links = [e["takedown"]["link"].hex() for e in done]
|
|
if set(links) != set(c["takedowns"]) or len(set(links)) != len(links) \
|
|
or any(not c["from"] <= e["takedown"]["decided_at"] < c["to"] for e in done):
|
|
raise Refuse("counts_mismatch", ["counts"])
|
|
rows = rows_of(done)
|
|
if rows != c["rows"] and on("counts-rows"):
|
|
raise Refuse("counts_mismatch", ["counts"])
|
|
return rows
|
|
|
|
|
|
# ---------------------------------------------------------------- the corpus
|
|
|
|
def case_of(world, over):
|
|
c = dict(world)
|
|
c.update(over)
|
|
c["as_of"] = time_value(c["as_of"])
|
|
return c
|
|
|
|
|
|
def section(text_, head):
|
|
at = text_.index("\n" + head)
|
|
nxt = re.search(r"\n#{2} ", 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 = [line.split("`")[1] for line in section(_TEXT, "## 11. Reasons").splitlines() if line.startswith("| `")]
|
|
|
|
|
|
def check(doc, quiet=True):
|
|
fails = []
|
|
world = doc["world"]
|
|
by = {}
|
|
for v in doc["cases"]:
|
|
try:
|
|
case = case_of(world, v["case"])
|
|
by[v["name"]] = case
|
|
got = verify_case(case)
|
|
except Exception as e: # noqa: BLE001 — a crash on a mutated or weakened run is a failure
|
|
got = {"outcome": "crash: %r" % (e,)}
|
|
if got != v["expect"]:
|
|
fails.append("%s: got %s" % (v["name"], json.dumps(got)[:240]))
|
|
if not quiet:
|
|
print("FAIL case %s: got %s, the corpus states %s" % (v["name"], json.dumps(got)[:300], json.dumps(v["expect"])[:300]))
|
|
elif not quiet:
|
|
print("PASS case %s: %s" % (v["name"], got["outcome"] + (" %s" % got["reason"] if "reason" in got else "")
|
|
+ (" at %s" % got["at"] if "at" in got else "")))
|
|
for v in doc["counts"]:
|
|
try:
|
|
rows = verify_counts(v["counts"], [by[n] for n in v["cases"]])
|
|
got = {"outcome": "accept", "rows": rows}
|
|
except Refuse as e:
|
|
got = {"outcome": "refuse", "reason": e.reason, "at": e.at}
|
|
except Exception as e: # noqa: BLE001
|
|
got = {"outcome": "crash: %r" % (e,)}
|
|
if got != v["expect"]:
|
|
fails.append("%s: got %s" % (v["name"], json.dumps(got)[:240]))
|
|
if not quiet:
|
|
print("FAIL counts %s: got %s" % (v["name"], json.dumps(got)[:300]))
|
|
elif not quiet:
|
|
print("PASS counts %s: %s" % (v["name"], got["outcome"] + (" %s" % got["reason"] if "reason" in got else "")))
|
|
return fails
|
|
|
|
|
|
def flip_hex_digit(s, at):
|
|
return s[:at] + ("1" if s[at] != "1" else "2") + s[at + 1:]
|
|
|
|
|
|
def mutations(doc):
|
|
"""Copies of the corpus, each with one defect a conforming verifier must catch."""
|
|
by = {v["name"]: i for i, v in enumerate(doc["cases"])}
|
|
|
|
def mutate(label, f):
|
|
d = copy.deepcopy(doc)
|
|
f(d)
|
|
return label, d
|
|
|
|
def case(d, name):
|
|
return d["cases"][by[name]]
|
|
|
|
def retext(texts, i, f):
|
|
v = json.loads(texts[i])
|
|
f(v)
|
|
texts[i] = EV.jcs(v)
|
|
|
|
def facts(d, name):
|
|
return case(d, name)["expect"]["facts"]
|
|
|
|
return [
|
|
mutate("a digit of the met request's takedown signature",
|
|
lambda d: case(d, "request-48h-met")["case"].update(takedown=EV.jcs(dict(json.loads(case(d, "request-48h-met")["case"]["takedown"]), sig=flip_hex_digit(json.loads(case(d, "request-48h-met")["case"]["takedown"])["sig"], 7))))),
|
|
mutate("the met request's last purge withheld", lambda d: case(d, "request-48h-met")["case"]["purges"].pop()),
|
|
mutate("the late purge of the missed order moved an hour earlier",
|
|
lambda d: retext(case(d, "order-3h-missed")["case"]["purges"], 5, lambda v: v.update(purged_at="2026-10-21T10:57:40Z"))),
|
|
mutate("the met request's time to removal stated otherwise", lambda d: facts(d, "request-48h-met").update(time_to_removal_s=42899)),
|
|
mutate("the met request's closure stated without its live track", lambda d: facts(d, "request-48h-met")["closure"].pop()),
|
|
mutate("the stranger's purge stated as counted", lambda d: facts(d, "stranger-purge-not-counted").update(uncounted=[])),
|
|
mutate("the matched candidate stated as purged by nobody", lambda d: facts(d, "request-48h-met")["matched"].update(last_purge=None)),
|
|
mutate("a registry duration changed", lambda d: d["world"].update(registry=d["world"]["registry"].replace("172800", "172799"))),
|
|
mutate("the generated group's receipts withheld from the world",
|
|
lambda d: d["world"].update(receipts=[r for r in d["world"]["receipts"] if r.get("group_id") != 5])),
|
|
mutate("the thumbnail's derivation record withheld", lambda d: d["world"]["records"].pop(1)),
|
|
mutate("the upscale's claim withheld", lambda d: d["world"]["claims"].pop(1)),
|
|
mutate("the cache's acknowledgement of the thumbnail withheld", lambda d: d["world"]["acknowledgements"].pop(0)),
|
|
mutate("a byte of an acknowledgement's signature",
|
|
lambda d: d["world"]["acknowledgements"][1].update(record=flip_hex_digit(d["world"]["acknowledgements"][1]["record"], len(d["world"]["acknowledgements"][1]["record"]) - 9))),
|
|
mutate("the unfixed request stated met", lambda d: facts(d, "request-unfixed").update(verdict="met")),
|
|
mutate("the reply window's end moved", lambda d: facts(d, "own-reply-early").update(window_end=facts(d, "own-reply-early")["window_end"] + 1)),
|
|
mutate("a refused case's place moved", lambda d: case(d, "derivative-left-out")["expect"].update(at=["closure", 5])),
|
|
mutate("a counts row's median stated otherwise", lambda d: d["counts"][0]["expect"]["rows"][0].update(median_s=1199)),
|
|
mutate("the counts verified without the pending order", lambda d: d["counts"][0]["cases"].remove("order-3h-pending")),
|
|
]
|
|
|
|
|
|
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 []
|
|
ok = len(REASONS) == 22 and len(set(REASONS)) == len(REASONS)
|
|
print(("PASS" if ok else "FAIL") + " §11 reads from the extension: %d reasons" % len(REASONS))
|
|
fails += [] if ok else ["the extension's §11 does not read"]
|
|
with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as f:
|
|
doc = json.load(f)
|
|
with open(PUBLISHED, encoding="utf-8") as f:
|
|
published = f.read()
|
|
same = doc["world"]["registry"] == published
|
|
print(("PASS" if same else "FAIL") + " the world's registry is registry/takedown-clocks.json as published")
|
|
fails += [] if same else ["the world's registry is not the published one"]
|
|
fails += check(doc, quiet=False)
|
|
extra = sorted(set(os.listdir(corpus_dir)) - {"vectors.json", "README.md", "__pycache__"})
|
|
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 = {v["expect"].get("reason") for v in doc["cases"]} | {v["expect"].get("reason") for v in doc["counts"]}
|
|
missing = [r for r in REASONS if r not in seen]
|
|
print(("PASS" if not missing else "FAIL") + " every reason of §11 appears in the corpus")
|
|
fails += ["no case gives %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][:90] if caught else ""))
|
|
if not caught:
|
|
fails.append("a mutated copy verifies: %s" % label)
|
|
for name in CHECKS:
|
|
DROPPED.clear()
|
|
DROPPED.add(name)
|
|
caught = check(doc)
|
|
DROPPED.clear()
|
|
print(("PASS" if caught else "FAIL") + " the corpus catches the verifier without its check %s%s"
|
|
% (name, " (%s)" % caught[0][:90] if caught else ""))
|
|
if not caught:
|
|
fails.append("the check %s can be dropped unnoticed" % name)
|
|
for f in fails:
|
|
print("FAIL " + f)
|
|
print(("PASS" if not fails else "FAIL") + " takedown corpus: %d cases, %d counts" % (len(doc["cases"]), len(doc["counts"])))
|
|
return 1 if fails else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) != 2:
|
|
print("usage: takedown_verify.py <live-conformance/takedown>", file=sys.stderr)
|
|
sys.exit(2)
|
|
sys.exit(main(sys.argv[1]))
|