ad1f47472b
Verifies each room with live_rooms_verify.py and each clip with live_record_verify.py (which gains an exempt parameter, unchanged by default), writes every export from §9, reads §9.5, §9.6 and §10 from the extension, and requires 19 mutated copies to fail. Co-Authored-By: Claude Opus 5.5 <[email protected]>
810 lines
38 KiB
Python
810 lines
38 KiB
Python
#!/usr/bin/env python3
|
|
"""Independent verifier of the live-moderation corpus (extensions/live-moderation.md).
|
|
|
|
Written from the extension's text. It shares no code with wai-rs, which generates the corpus,
|
|
and never runs it. It checks Ed25519 signatures with ../claim-conformance/_ed25519.py, WAI's
|
|
strict rule (SPEC §7 *Signatures*) written from RFC 8032 with the standard library. Each room
|
|
is verified by the live-rooms corpus's own second implementation, live_rooms_verify.py
|
|
(live-rooms §9), and each clip by the live-recording corpus's, live_record_verify.py
|
|
(live-recording §5.4), each written from its text alone. Everything else is here:
|
|
|
|
- the review policy (§3): its layout, rules, signature, chain and the room's review field;
|
|
- the statement (§4.2, §4.3), lines of decisions (§4.5) and fallback decisions (§4.6);
|
|
- notices and complaints (§5);
|
|
- items under review and their timelines (§6);
|
|
- verification, with the gaps restricted events leave explained (§7);
|
|
- clips under moderation (§8.3);
|
|
- the statement-of-reasons export (§9), written byte for byte from §9.2 and §9.3, with the
|
|
categories and content types of §9.5 and §9.6 read from the extension's own tables.
|
|
|
|
For every room it requires the outcome, reason and place, and everything else the corpus
|
|
states; every export file, byte for byte; for every clip, its outcome; of the corpus, that
|
|
every reason of §10 appears in it and that no other file is there. Then it mutates copies of the
|
|
corpus, one defect each, and requires every copy to fail.
|
|
|
|
python3 live_moderation_verify.py <live-conformance/moderation>
|
|
|
|
Exit status 0 only if every case verifies and every mutated copy fails. Python 3.8 or later;
|
|
standard library only.
|
|
"""
|
|
|
|
import copy
|
|
import datetime
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import struct
|
|
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 live_events_verify as LEV # noqa: E402
|
|
import live_rooms_verify as LRV # noqa: E402
|
|
import live_record_verify as LREC # noqa: E402
|
|
import _ed25519 as ED # noqa: E402
|
|
|
|
Refuse = LEV.Refuse
|
|
EXTENSION = os.path.join(HERE, "..", "extensions", "live-moderation.md")
|
|
|
|
POLICY_MAGIC, STATEMENT_MAGIC, NOTICE_MAGIC = b"WMP1", b"WMS1", b"WMN1"
|
|
POLICY_DOMAIN = b"wai:live-moderation-policy\x01"
|
|
NOTICE_DOMAIN = b"wai:live-moderation-notice\x01"
|
|
ZERO = bytes(32)
|
|
META, CUE, CHAT, REACTION, POLL_OPEN, VOTE, POLL_CLOSE, MARKER, CAPTION, DECISION = 1, 2, 3, 4, 5, 6, 7, 8, 0x0A, 0x0B
|
|
DECIDABLE = {META, CUE, CHAT, REACTION, POLL_OPEN, VOTE, POLL_CLOSE, MARKER, CAPTION} # §4.1
|
|
MODERATE = 1 # live-rooms §3.3
|
|
ALLOW, DENY, REDACT = 1, 2, 3
|
|
EVENT, OBJECT = 1, 2
|
|
OWN, NOTICE, COMPLAINT, FALLBACK = 1, 2, 3, 4
|
|
ILLEGAL, INCOMPATIBLE = 1, 2
|
|
PRE, POST = 1, 2
|
|
F_ALLOW, F_DENY, F_HOLD = 1, 2, 3
|
|
VERDICTS = {1: "allow", 2: "deny", 3: "redact"}
|
|
BASES = {1: "own", 2: "notice", 3: "complaint", 4: "fallback"}
|
|
MEDIA = {1: "audio", 2: "video", 3: "screen"}
|
|
TERRITORY = ["AT", "BE", "BG", "CY", "CZ", "DE", "DK", "EE", "ES", "FI", "FR", "GR", "HR", "HU", "IE", "IS", "IT", "LI",
|
|
"LT", "LU", "LV", "MT", "NL", "NO", "PL", "PT", "RO", "SE", "SI", "SK"] # §9.3
|
|
VISIBLE = {"shown", "pending", "allowed", "fallback_allow"} # §6.2
|
|
|
|
|
|
class Stop(Exception):
|
|
def __init__(self, reason, at):
|
|
super().__init__(reason)
|
|
self.reason, self.at = reason, at
|
|
|
|
|
|
def sha(*parts):
|
|
h = hashlib.sha256()
|
|
for p in parts:
|
|
h.update(p)
|
|
return h.digest()
|
|
|
|
|
|
# ---------------------------------------------------------------- the extension's own tables
|
|
|
|
def section(text, heading):
|
|
at = text.index("\n" + heading)
|
|
end = text.find("\n## ", at + 1)
|
|
end2 = text.find("\n### ", at + 1)
|
|
ends = [e for e in (end, end2) if e != -1]
|
|
return text[at:min(ends) if ends else len(text)]
|
|
|
|
|
|
def table_rows(sec):
|
|
return [[c.strip() for c in line.strip().strip("|").split("|")] for line in sec.splitlines()
|
|
if line.startswith("| ") and not line.startswith("|---")][1:]
|
|
|
|
|
|
def tables():
|
|
"""§9.5's categories, §9.6's content types and §10's reasons, from the text."""
|
|
with open(EXTENSION, encoding="utf-8") as f:
|
|
text = f.read()
|
|
categories = [row[1].strip("`") for row in table_rows(section(text, "### 9.5 Categories"))]
|
|
types = {}
|
|
for target, ctype, other in table_rows(section(text, "### 9.6 Content types")):
|
|
ctype, other = ctype.strip("`"), other.strip("`") or None
|
|
if target.startswith("an object of media 1"):
|
|
types[(OBJECT, 1)] = (ctype, other)
|
|
elif target.startswith("an object of media 2"):
|
|
types[(OBJECT, 2)] = types[(OBJECT, 3)] = (ctype, other)
|
|
else:
|
|
names = {v: k for k, v in LEV.KINDS.items()}
|
|
for name in re.findall(r"`([a-z_]+)`", target):
|
|
types[(EVENT, names[name])] = (ctype, other)
|
|
reasons = [row[0].strip("`") for row in table_rows(section(text, "## 10. Reasons"))]
|
|
return categories, types, reasons
|
|
|
|
|
|
CATEGORIES, CONTENT_TYPES, REASONS = tables()
|
|
|
|
|
|
# ---------------------------------------------------------------- reading
|
|
|
|
class Reader(LEV.Reader):
|
|
def text(self):
|
|
return self.take(self.u16())
|
|
|
|
|
|
def read_head(b, magic):
|
|
"""§3.3 step 1."""
|
|
if len(b) < 4:
|
|
raise Refuse("truncated")
|
|
if b[:4] != magic:
|
|
raise Refuse("malformed")
|
|
if len(b) < 5:
|
|
raise Refuse("truncated")
|
|
if b[4] != 1:
|
|
raise Refuse("unknown_version")
|
|
|
|
|
|
def utf8(b):
|
|
try:
|
|
return b.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
return None
|
|
|
|
|
|
def read_policy(b):
|
|
"""§3.1, §3.3: the policy, or Refuse."""
|
|
read_head(b, POLICY_MAGIC)
|
|
r = Reader(b, "truncated")
|
|
r.take(5)
|
|
p = {"room": r.take(32), "prev": r.take(32), "effective": r.u64()}
|
|
p["reviewers"] = [r.take(32) for _ in range(r.u8())]
|
|
p["flaggers"] = [r.take(32) for _ in range(r.u8())]
|
|
p["entries"] = []
|
|
for _ in range(r.u8()):
|
|
p["entries"].append({"scope": r.u8(), "code": r.u8(), "mode": r.u8(), "deadline_ms": r.u32(), "fallback": r.u8(),
|
|
"fallback_reason": r.u16()})
|
|
p["reasons"] = []
|
|
for _ in range(r.u8()):
|
|
p["reasons"].append({"code": r.u16(), "ground": r.u8(), "category": r.u8(), "reference": r.text(), "url": r.text()})
|
|
p["territory"] = [r.take(2) for _ in range(r.u8())]
|
|
p["redress"] = r.text()
|
|
body_len = r.p
|
|
p["sig"] = r.take(64)
|
|
if not r.done():
|
|
raise Refuse("malformed")
|
|
p["body"] = b[:body_len]
|
|
p["digest"] = sha(p["body"])
|
|
bad = Refuse("policy_malformed")
|
|
counts_ok = (len(p["reviewers"]) <= 16 and len(p["flaggers"]) <= 16 and len(p["entries"]) <= 32
|
|
and 1 <= len(p["reasons"]) <= 64 and 1 <= len(p["territory"]) <= 30)
|
|
if not counts_ok:
|
|
raise bad
|
|
asc = lambda xs: all(a < b for a, b in zip(xs, xs[1:])) # noqa: E731
|
|
if not asc(p["reviewers"]) or not asc(p["flaggers"]):
|
|
raise bad
|
|
if not asc([(e["scope"], e["code"]) for e in p["entries"]]) or not asc([c["code"] for c in p["reasons"]]):
|
|
raise bad
|
|
codes = {c["code"] for c in p["reasons"]}
|
|
for e in p["entries"]:
|
|
code_ok = (e["scope"] == 1 and e["code"] in DECIDABLE) or (e["scope"] == 2 and e["code"] in (1, 2, 3))
|
|
reason_ok = e["fallback_reason"] == 0 if e["fallback"] == F_ALLOW else e["fallback_reason"] in codes
|
|
if not (code_ok and e["mode"] in (PRE, POST) and e["deadline_ms"] > 0 and e["fallback"] in (1, 2, 3) and reason_ok):
|
|
raise bad
|
|
for c in p["reasons"]:
|
|
ref, url = utf8(c["reference"]), utf8(c["url"])
|
|
if ref is None or url is None:
|
|
raise bad
|
|
url_ok = url == "" or (len(c["url"]) <= 500 and url.startswith("https://") and all(0x21 <= x <= 0x7E for x in c["url"]))
|
|
if not (c["code"] > 0 and c["ground"] in (1, 2) and 1 <= c["category"] <= 16 and 1 <= len(ref) <= 500 and url_ok):
|
|
raise bad
|
|
c["reference"], c["url"] = ref, url
|
|
terr = [t.decode("latin-1") for t in p["territory"]]
|
|
if not asc(terr) or any(t not in TERRITORY for t in terr):
|
|
raise bad
|
|
p["territory"] = terr
|
|
if utf8(p["redress"]) is None:
|
|
raise bad
|
|
return p
|
|
|
|
|
|
def read_statement(b):
|
|
"""§4.2, §4.3: the statement, or Refuse("statement_malformed")."""
|
|
try:
|
|
r = Reader(b, "statement_malformed")
|
|
if r.take(4) != STATEMENT_MAGIC or r.u8() != 1:
|
|
raise Refuse("statement_malformed")
|
|
s = {"target_type": r.u8(), "target_prev": r.take(32), "supersedes": r.take(32), "basis": r.u8(),
|
|
"notice": r.take(32), "automated": r.u8(), "ground": r.u8()}
|
|
texts = [utf8(r.text()) for _ in range(3)]
|
|
if not r.done():
|
|
raise Refuse("statement_malformed")
|
|
except Refuse:
|
|
raise Refuse("statement_malformed")
|
|
s["ground_ref"], s["explanation"], s["facts"] = texts
|
|
decided = (s["automated"] >> 1) & 3
|
|
ok = (s["target_type"] in (EVENT, OBJECT) and (s["target_type"] == EVENT or s["target_prev"] == ZERO)
|
|
and s["basis"] in (1, 2, 3, 4) and ((s["basis"] in (NOTICE, COMPLAINT)) == (s["notice"] != ZERO))
|
|
and s["automated"] & ~0b111 == 0 and decided != 3
|
|
and (s["basis"] != FALLBACK or (decided == 2 and s["supersedes"] == ZERO))
|
|
and s["ground"] in (1, 2) and None not in texts)
|
|
if not ok:
|
|
raise Refuse("statement_malformed")
|
|
return s
|
|
|
|
|
|
def read_notice(b):
|
|
"""§5.1, §5.2 (all but the list's rule): the notice, or Refuse."""
|
|
read_head(b, NOTICE_MAGIC)
|
|
r = Reader(b, "truncated")
|
|
r.take(5)
|
|
n = {"room": r.take(32), "type": r.u8(), "target_type": r.u8(), "target": r.take(32), "t": r.u64(), "ground": r.u8(),
|
|
"good_faith": r.u8(), "ground_ref": r.text(), "statement": r.text(), "contact": r.text(), "submitter": r.take(32)}
|
|
body_len = r.p
|
|
n["sig"] = r.take(64)
|
|
if not r.done():
|
|
raise Refuse("malformed")
|
|
n["body"] = b[:body_len]
|
|
n["digest"] = sha(n["body"])
|
|
if n["type"] == 1:
|
|
ok = n["target_type"] in (1, 2) and n["ground"] in (1, 2)
|
|
elif n["type"] == 2:
|
|
ok = n["target_type"] == 1 and n["ground"] == 0
|
|
else:
|
|
ok = False
|
|
if not ok or n["good_faith"] > 1 or any(utf8(n[k]) is None for k in ("ground_ref", "statement", "contact")):
|
|
raise Refuse("notice_malformed")
|
|
return n
|
|
|
|
|
|
# ---------------------------------------------------------------- §7 verification
|
|
|
|
def policy_at(policies, t):
|
|
"""§3.4."""
|
|
for p in reversed(policies):
|
|
if p["effective"] <= t:
|
|
return p
|
|
return None
|
|
|
|
|
|
def entry_of(p, scope, code):
|
|
return next((e for e in p["entries"] if e["scope"] == scope and e["code"] == code), None) if p else None
|
|
|
|
|
|
def reason_of(p, code):
|
|
return next((c for c in p["reasons"] if c["code"] == code), None)
|
|
|
|
|
|
def has_kind(bm, k):
|
|
return bm[k // 8] >> (k % 8) & 1 == 1
|
|
|
|
|
|
def moderate(v):
|
|
"""§7: a dict of what the corpus states for room vector v."""
|
|
descriptor = bytes.fromhex(v["descriptor"])
|
|
grants = [bytes.fromhex(g) for g in v["grants"]]
|
|
chains = [[bytes.fromhex(h) for h in c] for c in v["chains"]]
|
|
room = LRV.verify_room(descriptor, grants, v["key_log"], chains) # step 1
|
|
if room["outcome"] == "refuse":
|
|
return {"outcome": "refuse", "reason": room["reason"], "at": ["room"] + room["at"]}, None
|
|
d = LRV.read_descriptor(descriptor)
|
|
held_grants, _ = LRV.check_grants(d, grants)
|
|
powers = {g["digest"]: g["powers"] for g in held_grants}
|
|
room_gaps = [(bytes.fromhex(g["principal"]), g["first"], g["last"]) for g in room.get("gaps", [])]
|
|
policies = []
|
|
for i, h in enumerate(v["policies"]): # step 2
|
|
try:
|
|
p = read_policy(bytes.fromhex(h))
|
|
if p["room"] != d["room"]:
|
|
raise Refuse("wrong_room")
|
|
bad = ED.verify(d["authority"], POLICY_DOMAIN + p["body"], p["sig"])
|
|
if bad:
|
|
raise Refuse(bad)
|
|
q = policies[-1] if policies else None
|
|
if not ((q is None and p["prev"] == ZERO and p["effective"] == 0)
|
|
or (q is not None and p["prev"] == q["digest"] and p["effective"] > q["effective"])):
|
|
raise Refuse("policy_unlinked")
|
|
kinds, deadline, fb = d["review"]
|
|
for k in range(256): # §3.5
|
|
if has_kind(kinds, k):
|
|
e = entry_of(p, 1, k)
|
|
if e is None or e["mode"] != PRE or e["deadline_ms"] != deadline or e["fallback"] != fb:
|
|
raise Refuse("policy_conflicts_room")
|
|
except Refuse as e:
|
|
return {"outcome": "refuse", "reason": e.reason, "at": ["policy", i]}, None
|
|
policies.append(p)
|
|
notices = []
|
|
for i, h in enumerate(v["notices"]): # step 3
|
|
try:
|
|
n = read_notice(bytes.fromhex(h))
|
|
if any(m["digest"] == n["digest"] for m in notices):
|
|
raise Refuse("notice_malformed")
|
|
if n["room"] != d["room"]:
|
|
raise Refuse("wrong_room")
|
|
bad = ED.verify(n["submitter"], NOTICE_DOMAIN + n["body"], n["sig"])
|
|
if bad:
|
|
raise Refuse(bad)
|
|
except Refuse as e:
|
|
return {"outcome": "refuse", "reason": e.reason, "at": ["notice", i]}, None
|
|
notices.append(n)
|
|
objects = [{"digest": bytes.fromhex(o["digest"]), "media": o["media"], "t": o["t"]} for o in v["objects"]]
|
|
# The events the room accepted, once each, in the merged order.
|
|
held, seen = [], set()
|
|
for c, i, under in room["order"]:
|
|
ev = LEV.read_record(chains[c][i])
|
|
if ev["digest"] in seen:
|
|
continue
|
|
seen.add(ev["digest"])
|
|
ev["payload_read"] = LEV.read_payload(ev["kind"], ev["payload"], 1 << 20)
|
|
ev["chain"], ev["record"], ev["under"] = c, i, bytes.fromhex(under) if under else None
|
|
held.append(ev)
|
|
by_seq = {(e["principal"], e["seq"]): k for k, e in enumerate(held)}
|
|
by_digest = {e["digest"]: k for k, e in enumerate(held)}
|
|
|
|
def room_key(p):
|
|
return p in (d["authority"], d["clock"])
|
|
|
|
def is_statement(e):
|
|
return e["kind"] == META and e["payload"][:4] == STATEMENT_MAGIC
|
|
|
|
statements = {}
|
|
for e in held: # step 4
|
|
if not is_statement(e):
|
|
continue
|
|
at = ["record", e["chain"], e["record"]]
|
|
if e["seq"] == 0:
|
|
return {"outcome": "refuse", "reason": "statement_misplaced", "at": at}, None
|
|
b = by_seq.get((e["principal"], e["seq"] - 1))
|
|
if b is None:
|
|
continue
|
|
if held[b]["kind"] != DECISION or held[b]["t"] != e["t"]:
|
|
return {"outcome": "refuse", "reason": "statement_misplaced", "at": at}, None
|
|
try:
|
|
statements[b] = read_statement(e["payload"])
|
|
except Refuse as x:
|
|
return {"outcome": "refuse", "reason": x.reason, "at": at}, None
|
|
last_on = {}
|
|
for k, e in enumerate(held):
|
|
if e["kind"] == DECISION:
|
|
last_on[e["payload_read"]["target"]] = k
|
|
explained, remaining, withheld = [], [], set() # step 5, §7.3
|
|
for g in room_gaps:
|
|
ex = explain(g, held, by_seq, statements, last_on)
|
|
if ex is None:
|
|
remaining.append(g)
|
|
else:
|
|
explained.append(ex[0])
|
|
withheld |= ex[1]
|
|
gap_remains = bool(remaining)
|
|
|
|
def event_entry(e):
|
|
"""§6.1."""
|
|
if e["kind"] not in DECIDABLE or is_statement(e) or room_key(e["principal"]) or powers.get(e["under"], 0) & MODERATE:
|
|
return None
|
|
return entry_of(policy_at(policies, e["t"]), 1, e["kind"])
|
|
|
|
def object_entry(o):
|
|
return entry_of(policy_at(policies, o["t"]), 2, o["media"])
|
|
|
|
decisions, line, decided = [], {}, {}
|
|
for k, e in enumerate(held): # step 6, §7.4
|
|
if e["kind"] != DECISION:
|
|
continue
|
|
at = ["record", e["chain"], e["record"]]
|
|
p, pl = e["principal"], e["payload_read"]
|
|
v_, x, code = pl["verdict"], pl["target"], pl["reason"]
|
|
if k in statements:
|
|
said = "known"
|
|
elif (p, e["seq"] + 1) in by_seq:
|
|
said = "none"
|
|
elif any(q == p and a <= e["seq"] + 1 <= z for q, a, z in remaining):
|
|
said = "unknown"
|
|
else:
|
|
said = "none"
|
|
st = statements.get(k)
|
|
tt = st["target_type"] if st else EVENT
|
|
try:
|
|
pol = policy_at(policies, e["t"]) # 1
|
|
rc = reason_of(pol, code) if pol else None
|
|
if rc is None or (st is not None and st["ground"] != rc["ground"]):
|
|
raise Refuse("reason_not_in_policy")
|
|
target, entry = None, None # 2
|
|
if tt == OBJECT:
|
|
o = next((o for o in objects if o["digest"] == x and o["t"] <= e["t"]), None)
|
|
if o is None:
|
|
raise Refuse("target_unknown")
|
|
target, entry = (OBJECT, o["media"], o["t"]), object_entry(o)
|
|
elif x in by_digest:
|
|
ti = by_digest[x]
|
|
te = held[ti]
|
|
if ti >= k:
|
|
raise Refuse("target_unknown")
|
|
if te["kind"] not in DECIDABLE or is_statement(te):
|
|
raise Refuse("target_not_decidable")
|
|
if st is not None and st["target_prev"] != te["prev"]:
|
|
raise Refuse("target_prev_mismatch")
|
|
target, entry = (EVENT, te["kind"], te["t"]), event_entry(te)
|
|
elif x not in withheld and not gap_remains:
|
|
raise Refuse("target_unknown")
|
|
if entry is not None and pol["reviewers"] and not room_key(p) and p not in pol["reviewers"]: # 3
|
|
raise Refuse("not_reviewer")
|
|
if said != "unknown": # 4
|
|
sup = st["supersedes"] if st else ZERO
|
|
head = line.get(x)
|
|
if sup == ZERO:
|
|
if head is not None:
|
|
raise Refuse("supersede_gap")
|
|
elif sup in decided:
|
|
if decisions[decided[sup]]["target"] != x:
|
|
raise Refuse("supersede_mismatch")
|
|
if head != sup:
|
|
raise Refuse("supersede_gap")
|
|
elif not gap_remains:
|
|
raise Refuse("supersede_unknown")
|
|
if st is not None and st["basis"] in (NOTICE, COMPLAINT): # 5
|
|
want = 1 if st["basis"] == NOTICE else 2
|
|
n = next((n for n in notices if n["digest"] == st["notice"] and n["type"] == want), None)
|
|
if n is None:
|
|
raise Refuse("notice_unknown")
|
|
aims = (n["target"] == x and n["target_type"] == tt) if st["basis"] == NOTICE else n["target"] == st["supersedes"]
|
|
if not aims or n["t"] > e["t"]:
|
|
raise Refuse("notice_mismatch")
|
|
if st is not None and st["basis"] == FALLBACK:
|
|
if not room_key(p):
|
|
raise Refuse("fallback_not_in_policy")
|
|
if target is not None:
|
|
if entry is None or entry["fallback"] == F_HOLD or v_ != entry["fallback"] or code != entry["fallback_reason"]:
|
|
raise Refuse("fallback_not_in_policy")
|
|
if e["t"] < target[2] + entry["deadline_ms"] * 1000:
|
|
raise Refuse("fallback_early")
|
|
except Refuse as r:
|
|
return {"outcome": "refuse", "reason": r.reason, "at": at}, None
|
|
line[x] = e["digest"]
|
|
decided[e["digest"]] = len(decisions)
|
|
decisions.append({"digest": e["digest"], "decider": p, "t": e["t"], "target": x, "target_type": tt, "verdict": v_,
|
|
"reason": code, "said": said, "st": st, "target_of": target})
|
|
for x in decisions:
|
|
x["standing"] = line[x["target"]] == x["digest"]
|
|
# Step 7 and what a verification states (§7.5).
|
|
out = {"outcome": "gap" if gap_remains else "accept", "room_outcome": room["outcome"],
|
|
"gaps": [{"principal": p.hex(), "first": a, "last": z} for p, a, z in remaining],
|
|
"explained": [{"principal": p.hex(), "first": a, "last": z, "decisions": [h.hex() for h in ds]} for p, a, z, ds in explained]}
|
|
out["decisions"] = [{
|
|
"digest": x["digest"].hex(), "decider": x["decider"].hex(), "t": x["t"], "target": x["target"].hex(),
|
|
"target_type": "event" if x["target_type"] == EVENT else "object", "verdict": VERDICTS[x["verdict"]], "reason": x["reason"],
|
|
"statement": x["said"], "basis": None if x["said"] == "unknown" else BASES[x["st"]["basis"] if x["st"] else OWN],
|
|
"supersedes": x["st"]["supersedes"].hex() if x["st"] and x["st"]["supersedes"] != ZERO else None,
|
|
"notice": x["st"]["notice"].hex() if x["st"] and x["st"]["notice"] != ZERO else None,
|
|
"standing": x["standing"]} for x in decisions]
|
|
items = []
|
|
for e in held:
|
|
ds = [x for x in decisions if x["target"] == e["digest"] and x["target_type"] == EVENT]
|
|
en = event_entry(e)
|
|
if en is not None or ds:
|
|
items.append(item(e["digest"], "event", LEV.KINDS[e["kind"]], e["t"], en, ds))
|
|
for o in objects:
|
|
ds = [x for x in decisions if x["target"] == o["digest"] and x["target_type"] == OBJECT]
|
|
en = object_entry(o)
|
|
if en is not None or ds:
|
|
items.append(item(o["digest"], "object", MEDIA[o["media"]], o["t"], en, ds))
|
|
out["items"] = items
|
|
ctx = {"t0": d["t0"], "policies": policies, "notices": notices, "decisions": decisions}
|
|
exports, files = [], {}
|
|
for k, x in enumerate(decisions):
|
|
rec = export(ctx, k)
|
|
if rec is None:
|
|
exports.append({"decision": x["digest"].hex(), "file": None})
|
|
elif isinstance(rec, str):
|
|
exports.append({"decision": x["digest"].hex(), "file": None, "refuse": "sor_incomplete", "member": rec})
|
|
else:
|
|
name = "%s.d%d.sor.json" % (v["name"], k)
|
|
files[name] = rec
|
|
exports.append({"decision": x["digest"].hex(), "file": name})
|
|
out["exports"] = exports
|
|
restricted = {x["target"]: x["digest"] for x in decisions
|
|
if x["standing"] and x["target_type"] == OBJECT and x["verdict"] != ALLOW}
|
|
return out, {"files": files, "restricted": restricted}
|
|
|
|
|
|
def explain(g, held, by_seq, statements, last_on):
|
|
"""§7.3: (explained gap, withheld digests), or None."""
|
|
p, a, z = g
|
|
after = by_seq.get((p, z + 1))
|
|
if after is None:
|
|
return None
|
|
dg, ds, digests = held[after]["prev"], [], set()
|
|
for _ in range(a, z + 1):
|
|
k = last_on.get(dg)
|
|
if k is None or k not in statements:
|
|
return None
|
|
s = statements[k]
|
|
if s["target_type"] != EVENT or held[k]["payload_read"]["verdict"] == ALLOW:
|
|
return None
|
|
ds.append(held[k]["digest"])
|
|
digests.add(dg)
|
|
dg = s["target_prev"]
|
|
if a == 0:
|
|
start = ZERO
|
|
else:
|
|
b = by_seq.get((p, a - 1))
|
|
if b is None:
|
|
return None
|
|
start = held[b]["digest"]
|
|
if dg != start:
|
|
return None
|
|
return (p, a, z, list(reversed(ds))), digests
|
|
|
|
|
|
def item(digest, tt, what, t, entry, ds):
|
|
"""§6.2: the item and its timeline."""
|
|
first = "shown" if entry is None else ("held" if entry["mode"] == PRE else "pending")
|
|
tl = [(t, first, None)]
|
|
if entry is not None:
|
|
due = t + entry["deadline_ms"] * 1000
|
|
if not any(x["t"] <= due and not (x["st"] and x["st"]["basis"] == FALLBACK) for x in ds):
|
|
tl.append((due, {F_ALLOW: "fallback_allow", F_DENY: "fallback_deny", F_HOLD: "fallback_hold"}[entry["fallback"]], None))
|
|
for x in ds:
|
|
tl.append((x["t"], {ALLOW: "allowed", DENY: "denied", REDACT: "redacted"}[x["verdict"]], x["digest"]))
|
|
return {"digest": digest.hex(), "target_type": tt, "kind": what, "t": t,
|
|
"entry": None if entry is None else {"mode": "pre" if entry["mode"] == PRE else "post",
|
|
"deadline_ms": entry["deadline_ms"],
|
|
"fallback": {1: "allow", 2: "deny", 3: "hold"}[entry["fallback"]]},
|
|
"timeline": [{"t": a, "state": s, "decision": h.hex() if h else None, "visible": s in VISIBLE} for a, s, h in tl]}
|
|
|
|
|
|
# ---------------------------------------------------------------- §9 the export
|
|
|
|
def js(s):
|
|
"""§9.2's string."""
|
|
out = ['"']
|
|
for c in s:
|
|
o = ord(c)
|
|
if c == '"':
|
|
out.append('\\"')
|
|
elif c == "\\":
|
|
out.append("\\\\")
|
|
elif o in (8, 9, 10, 12, 13):
|
|
out.append({8: "\\b", 9: "\\t", 10: "\\n", 12: "\\f", 13: "\\r"}[o])
|
|
elif o < 0x20:
|
|
out.append("\\u%04x" % o)
|
|
else:
|
|
out.append(c)
|
|
out.append('"')
|
|
return "".join(out)
|
|
|
|
|
|
def utc_date(t0, t):
|
|
try:
|
|
return (datetime.datetime(1970, 1, 1) + datetime.timedelta(microseconds=t0 + t)).date().isoformat()
|
|
except OverflowError:
|
|
return None
|
|
|
|
|
|
def source(ctx, k):
|
|
"""§9.3 source_type, or None."""
|
|
x = ctx["decisions"][k]
|
|
if x["said"] == "unknown":
|
|
return None
|
|
st = x["st"]
|
|
if st is None or st["basis"] in (OWN, FALLBACK):
|
|
return "SOURCE_VOLUNTARY"
|
|
if st["basis"] == NOTICE:
|
|
n = next((n for n in ctx["notices"] if n["digest"] == st["notice"]), None)
|
|
if n is None:
|
|
return None
|
|
if n["ground"] != ILLEGAL:
|
|
return "SOURCE_TYPE_OTHER_NOTIFICATION"
|
|
pol = policy_at(ctx["policies"], x["t"])
|
|
return "SOURCE_TRUSTED_FLAGGER" if n["submitter"] in pol["flaggers"] else "SOURCE_ARTICLE_16"
|
|
y = next((j for j in range(k) if ctx["decisions"][j]["digest"] == st["supersedes"]), None)
|
|
return None if y is None else source(ctx, y)
|
|
|
|
|
|
def export(ctx, k):
|
|
"""§9: None for an allow, the member's name when §9.8 refuses, or the record's bytes."""
|
|
x = ctx["decisions"][k]
|
|
if x["verdict"] == ALLOW:
|
|
return None
|
|
st = x["st"]
|
|
if st is None or not 1 <= len(st["facts"]) <= 5000:
|
|
return "decision_facts"
|
|
pol = policy_at(ctx["policies"], x["t"])
|
|
rc = reason_of(pol, x["reason"])
|
|
illegal = st["ground"] == ILLEGAL
|
|
gname = "illegal_content_legal_ground" if illegal else "incompatible_content_ground"
|
|
ename = "illegal_content_explanation" if illegal else "incompatible_content_explanation"
|
|
gref = st["ground_ref"] or rc["reference"]
|
|
if not 1 <= len(gref) <= 500:
|
|
return gname
|
|
if not 1 <= len(st["explanation"]) <= 2000:
|
|
return ename
|
|
if x["target_of"] is None:
|
|
return "content_type"
|
|
tt, code, target_t = x["target_of"]
|
|
ctype, other = CONTENT_TYPES[(tt, code)]
|
|
cdate = utc_date(ctx["t0"], target_t)
|
|
if cdate is None or not "2000-01-01" <= cdate <= "2038-01-01":
|
|
return "content_date"
|
|
adate = utc_date(ctx["t0"], x["t"])
|
|
if adate is None or not "2020-01-01" <= adate <= "2038-01-01":
|
|
return "application_date"
|
|
src = source(ctx, k)
|
|
if src is None:
|
|
return "source_type"
|
|
m = [("decision_visibility", "[%s]" % js("DECISION_VISIBILITY_CONTENT_DISABLED" if x["verdict"] == DENY else "DECISION_VISIBILITY_CONTENT_REMOVED")),
|
|
("decision_facts", js(st["facts"])),
|
|
("decision_ground", js("DECISION_GROUND_ILLEGAL_CONTENT" if illegal else "DECISION_GROUND_INCOMPATIBLE_CONTENT"))]
|
|
if rc["url"]:
|
|
m.append(("decision_ground_reference_url", js(rc["url"])))
|
|
m += [(gname, js(gref)), (ename, js(st["explanation"])), ("content_type", "[%s]" % js(ctype))]
|
|
if other:
|
|
m.append(("content_type_other", js(other)))
|
|
m += [("category", js(CATEGORIES[rc["category"] - 1])),
|
|
("territorial_scope", "[%s]" % ",".join(js(t) for t in pol["territory"])),
|
|
("content_date", js(cdate)), ("application_date", js(adate)), ("source_type", js(src)),
|
|
("automated_detection", js("Yes" if st["automated"] & 1 else "No")),
|
|
("automated_decision", js(["AUTOMATED_DECISION_NOT_AUTOMATED", "AUTOMATED_DECISION_PARTIALLY",
|
|
"AUTOMATED_DECISION_FULLY"][(st["automated"] >> 1) & 3])),
|
|
("puid", js(x["digest"].hex()))]
|
|
return ("{" + ",".join("%s:%s" % (js(a), b) for a, b in m) + "}").encode("utf-8")
|
|
|
|
|
|
# ---------------------------------------------------------------- §8.3 clips
|
|
|
|
def clip_outcome(c, restricted):
|
|
"""§8.3: a dict of what the corpus states for clip c."""
|
|
exempt = frozenset(restricted)
|
|
got = LREC.verify_clip(c, exempt=exempt)
|
|
if got["outcome"] == "refuse":
|
|
return got
|
|
clip = LREC.read_clip(bytes.fromhex(c["record"]))
|
|
withheld = []
|
|
for k, (p, g) in enumerate(zip(clip["parts"], c["parts"])):
|
|
w = []
|
|
for j, x in enumerate(p["chunks"]):
|
|
leaf = LREC.leaf(x["desc"])
|
|
if leaf in restricted:
|
|
if j < len(g["bytes"]) and g["bytes"][j] is not None:
|
|
return {"outcome": "refuse", "reason": "redaction_not_applied", "at": ["chunk", k, j]}
|
|
w.append({"index": x["index"], "decision": restricted[leaf].hex()})
|
|
withheld.append(w)
|
|
facts = []
|
|
for f, w in zip(got["facts"], withheld):
|
|
f = {k: f[k] for k in ("take", "first", "count", "offset", "span", "device_window", "bytes", "sha256")}
|
|
f["withheld"] = w
|
|
facts.append(f)
|
|
got["facts"] = facts
|
|
return got
|
|
|
|
|
|
# ---------------------------------------------------------------- the corpus
|
|
|
|
def check(doc, files, quiet=False):
|
|
"""Every room and clip against its statement, and every export file: the failures."""
|
|
fails, runs, written = [], {}, {}
|
|
for v in doc.get("vectors", []):
|
|
try:
|
|
got, extra = moderate(v)
|
|
except Exception as e: # a crash on a mutated copy is a failure to read it
|
|
got, extra = None, None
|
|
bad = ["the vector does not verify: %s: %s" % (type(e).__name__, e)]
|
|
else:
|
|
bad = [k for k in got if v.get(k) != got[k]] + [k for k in ("outcome", "reason", "at") if k in v and k not in got]
|
|
if extra:
|
|
runs[v["name"]] = extra
|
|
for name, b in extra["files"].items():
|
|
written[name] = b
|
|
if files.get(name) != b:
|
|
bad.append("its export %s" % name)
|
|
if not quiet:
|
|
what = "?" if got is None else got["outcome"] + (" " + got["reason"] if got["outcome"] == "refuse" else "")
|
|
print(("PASS" if not bad else "FAIL") + " room %s: %s" % (v["name"], what))
|
|
fails += ["room %s: %s differs from the corpus" % (v["name"], k) for k in bad]
|
|
for c in doc.get("clips", []):
|
|
run = runs.get(c.get("room"))
|
|
try:
|
|
got = clip_outcome(c, run["restricted"] if run else {})
|
|
except Exception as e:
|
|
got, bad = None, ["the clip does not verify: %s: %s" % (type(e).__name__, e)]
|
|
else:
|
|
bad = [k for k in got if c.get(k) != got[k]] + [k for k in ("outcome", "reason", "at") if k in c and k not in got]
|
|
if not quiet:
|
|
what = "?" if got is None else got["outcome"] + (" " + got["reason"] if got["outcome"] == "refuse" else "")
|
|
print(("PASS" if not bad else "FAIL") + " clip %s: %s" % (c["name"], what))
|
|
fails += ["clip %s: %s differs from the corpus" % (c["name"], k) for k in bad]
|
|
return fails, written
|
|
|
|
|
|
def flip(h, at):
|
|
b = bytearray(bytes.fromhex(h))
|
|
b[at] ^= 0x01
|
|
return b.hex()
|
|
|
|
|
|
def mutations(doc, files):
|
|
"""Copies of the corpus, each with one defect a conforming verifier must catch."""
|
|
vn = {v["name"]: k for k, v in enumerate(doc["vectors"])}
|
|
cn = {c["name"]: k for k, c in enumerate(doc["clips"])}
|
|
|
|
def mutate(label, f):
|
|
d, fs = copy.deepcopy(doc), dict(files)
|
|
f(d["vectors"], d["clips"], fs)
|
|
return label, d, fs
|
|
|
|
pre, sub, notice, appeal, red = vn["premoderated_chat"], vn["subscriber_view"], vn["notice_and_decision"], vn["appeal_reverses"], vn["redacted_segment"]
|
|
|
|
def drop_statement(vs, cs, fs):
|
|
"""The moderator's first statement taken out of its chain."""
|
|
ch = vs[pre]["chains"]
|
|
for c in ch:
|
|
for i, h in enumerate(c):
|
|
if bytes.fromhex(h)[4 + 1 + 32 + 32 + 8 + 32 + 8 + 1 + 1 + 2:][:4] == STATEMENT_MAGIC:
|
|
del c[i]
|
|
return
|
|
|
|
def first_export(fs, name):
|
|
return next(k for k in sorted(fs) if k.startswith(name + "."))
|
|
|
|
return [
|
|
mutate("a byte of the first policy's signature", lambda vs, cs, fs: vs[pre]["policies"].__setitem__(0, flip(vs[pre]["policies"][0], -3))),
|
|
mutate("the policies given in each other's place", lambda vs, cs, fs: vs[pre]["policies"].reverse()),
|
|
mutate("the moderator's first statement dropped", drop_statement),
|
|
mutate("the denied message stated allowed", lambda vs, cs, fs: vs[pre]["decisions"][1].update(verdict="allow")),
|
|
mutate("a fallback stated at the wrong instant", lambda vs, cs, fs: vs[pre]["items"][3]["timeline"][1].update(t=34000000)),
|
|
mutate("the held reaction stated visible", lambda vs, cs, fs: vs[pre]["items"][3]["timeline"][1].update(visible=True)),
|
|
mutate("a byte of an export file", lambda vs, cs, fs: fs.update({first_export(fs, "premoderated_chat"): fs[first_export(fs, "premoderated_chat")].replace(b"Yes", b"No ")})),
|
|
mutate("the subscriber's explained gap stated with its decisions swapped",
|
|
lambda vs, cs, fs: vs[sub]["explained"][0]["decisions"].reverse()),
|
|
mutate("the subscriber's view stated without its gap explained", lambda vs, cs, fs: vs[sub].update(explained=[])),
|
|
mutate("a byte of the notice's cited provision", lambda vs, cs, fs: vs[notice]["notices"].__setitem__(0, flip(vs[notice]["notices"][0], 120))),
|
|
mutate("the notice's decision stated answering the flagger's notice", lambda vs, cs, fs: vs[notice]["decisions"][1].update(notice=vs[notice]["decisions"][3]["notice"])),
|
|
mutate("the trusted flagger's redaction stated from an Article 16 notice",
|
|
lambda vs, cs, fs: fs.update({"notice_and_decision.d3.sor.json": fs["notice_and_decision.d3.sor.json"].replace(b"SOURCE_TRUSTED_FLAGGER", b"SOURCE_ARTICLE_16")})),
|
|
mutate("the complaint withheld", lambda vs, cs, fs: vs[appeal].update(notices=[])),
|
|
mutate("the appeal's reversal stated not standing", lambda vs, cs, fs: vs[appeal]["decisions"][1].update(standing=False)),
|
|
mutate("the redacted chunk's object stated after the decision",
|
|
lambda vs, cs, fs: next(o for o in vs[red]["objects"] if o["digest"] == vs[red]["decisions"][0]["target"]).update(t=21000000)),
|
|
mutate("the clip given the redacted chunk's bytes", lambda vs, cs, fs: cs[cn["clip_withheld"]]["parts"][0]["bytes"].__setitem__(1, cs[cn["clip_restored_whole"]]["parts"][0]["bytes"][1])),
|
|
mutate("the trimmed clip's stated digest", lambda vs, cs, fs: cs[cn["clip_trimmed_withheld"]]["facts"][0].update(sha256="00" * 32)),
|
|
mutate("the withheld clip stated over the reversed room", lambda vs, cs, fs: cs[cn["clip_withheld"]].update(room="redaction_reversed")),
|
|
mutate("a stated export file removed", lambda vs, cs, fs: fs.pop(first_export(fs, "appeal_reverses"))),
|
|
]
|
|
|
|
|
|
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_tables = len(CATEGORIES) == 16 and len(REASONS) == 20 and len(CONTENT_TYPES) == 12
|
|
print(("PASS" if ok_tables else "FAIL") + " §9.5, §9.6 and §10 read from the extension: %d categories, %d content types, %d reasons"
|
|
% (len(CATEGORIES), len(CONTENT_TYPES), len(REASONS)))
|
|
fails += [] if ok_tables else ["the extension's tables do not read"]
|
|
with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as f:
|
|
doc = json.load(f)
|
|
files = {}
|
|
for name in sorted(os.listdir(corpus_dir)):
|
|
if name.endswith(".sor.json"):
|
|
with open(os.path.join(corpus_dir, name), "rb") as f:
|
|
files[name] = f.read()
|
|
got, written = check(doc, files)
|
|
fails += got
|
|
extra = sorted(set(os.listdir(corpus_dir)) - set(written) - {"vectors.json", "README.md"})
|
|
print(("PASS" if not extra else "FAIL") + " no file but vectors.json, README.md and the exports the rooms write")
|
|
fails += ["an extra file: %s" % n for n in extra]
|
|
seen = {v.get("reason") for v in doc["vectors"] + doc["clips"]}
|
|
seen |= {e.get("refuse") for v in doc["vectors"] for e in v.get("exports", [])}
|
|
missing = [r for r in REASONS if r not in seen]
|
|
print(("PASS" if not missing else "FAIL") + " every reason of §10 appears in the corpus")
|
|
fails += ["no case gives %s" % r for r in missing]
|
|
for label, d, fs in mutations(doc, files):
|
|
caught, _ = check(d, fs, quiet=True)
|
|
print(("PASS" if caught else "FAIL") + " mutated copy fails: %s%s" % (label, " (%s)" % caught[0][:100] if caught else ""))
|
|
if not caught:
|
|
fails.append("a mutated copy verifies: %s" % label)
|
|
for f in fails:
|
|
print("FAIL " + f)
|
|
print(("PASS" if not fails else "FAIL") + " live-moderation corpus: %d rooms, %d clips, %d exports"
|
|
% (len(doc["vectors"]), len(doc["clips"]), len(files)))
|
|
return 1 if fails else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) != 2:
|
|
print("usage: live_moderation_verify.py <live-conformance/moderation>", file=sys.stderr)
|
|
sys.exit(2)
|
|
sys.exit(main(sys.argv[1]))
|