673554abf0
The link corpus refuses a part that bears the clip record's magic; this adds the reverse. The live-recording corpus gains the clip link_record_magic: chunk 4's clip record beginning WLK1 and signed under it, so the magic is its one defect, refused malformed at the clip (§5.2, which now names WLK1 among the other magics). wai_live_record_vectors writes it; the README lists it. live_record_verify.py requires the same refusal, malformed at the clip, of a copy of each accepted clip given WLK1, so a verifier that reads past the magic to the signature fails. The CHANGELOG entry for WLK1 says so. Co-Authored-By: Claude Opus 5.5 <[email protected]>
887 lines
39 KiB
Python
887 lines
39 KiB
Python
#!/usr/bin/env python3
|
|
"""Independent verifier of the live-recording corpus (extensions/live-recording.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 take is verified by the capture-binding corpus's
|
|
own second implementation, capture_take_verify.py (capture-binding §9), and the room by
|
|
the live-rooms corpus's, live_rooms_verify.py (live-rooms §9), each written from its text
|
|
alone. Everything else is here:
|
|
|
|
- the package (§3): the listing of SPEC §3.1, written and checked here; the components and
|
|
their digests; the room's files and the events' chains split into their records;
|
|
- the recording manifest (§4.1, §4.2), who may sign it (§4.3), recording verification
|
|
(§4.4) and the verdicts of complete and publishable (§4.5);
|
|
- the clip record (§5.1, §5.2); the range proof of §5.3, by its recursion, and, for a run
|
|
of one chunk, by the inclusion-proof check of RFC 9162 §2.1.3.2 as that RFC states it,
|
|
which must agree; clip verification (§5.4); the time correspondence and its bound (§6);
|
|
and the FETCH address of §8.
|
|
|
|
For every vector it requires the outcome, reason and place, and the data the corpus states;
|
|
of the corpus, that every reason of §9 appears in it, and that every take it says it reuses
|
|
is capture binding's, byte for byte. Then it mutates copies of the corpus, one defect each,
|
|
and requires every copy to fail, and requires a copy of each accepted clip whose record
|
|
begins with the contribution link record's magic (live-contribution §6.2) to be refused as
|
|
`malformed` at the clip.
|
|
|
|
python3 live_record_verify.py <live-conformance/recordings>
|
|
|
|
Exit status 0 only if every vector verifies and every mutated copy fails. Python 3.8 or
|
|
later; standard library only.
|
|
"""
|
|
import copy
|
|
import hashlib
|
|
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 live_events_verify as LEV # noqa: E402
|
|
import live_rooms_verify as LRV # noqa: E402
|
|
import capture_take_verify as CT # noqa: E402
|
|
import _ed25519 as ED # noqa: E402
|
|
|
|
Refuse = LEV.Refuse
|
|
|
|
MANIFEST_DOMAIN = b"wai:live-recording\x01"
|
|
CLIP_DOMAIN = b"wai:live-clip\x01"
|
|
LEAF_DOMAIN = b"wai:take-leaf\x01"
|
|
NODE_DOMAIN = b"wai:out-node\x01"
|
|
TAKE_DOMAIN = b"wai:capture-take\x01"
|
|
CLIP_MAGIC = b"WCL1" # §5.1
|
|
LINK_MAGIC = b"WLK1" # live-contribution §6.2: the contribution link record, another record's magic
|
|
LISTING, MANIFEST_FILE = "LISTING", "recording.wrm"
|
|
ROOM, EVENTS, TAKE, COMPOSITE = 1, 2, 3, 4
|
|
COMPLETE = 1
|
|
RECORD_POWER = 8
|
|
TAKE_COMMIT, CONTROL = 0x0C, 0x09
|
|
JOIN, LEAVE, GRANT_EXCHANGE = 0x10, 0x11, 0x14
|
|
OPEN, CLOSE = 1, 3
|
|
PURPOSES = [(1, "live"), (2, "record"), (4, "distribute"), (8, "train")]
|
|
ZERO = bytes(32)
|
|
|
|
# §9, in its order.
|
|
REASONS = [
|
|
"listing_mismatch", "recording_manifest_invalid", "component_invalid", "component_digest_mismatch",
|
|
"component_unlisted", "recording_signer_not_permitted", "outside_span", "take_omitted",
|
|
"clip_record_invalid", "clip_device_mismatch", "clip_take_mismatch", "tree_head_not_committed",
|
|
"clip_out_of_range", "clip_not_included", "clip_window_invalid", "clip_chunk_missing",
|
|
]
|
|
|
|
|
|
def sha(*parts):
|
|
h = hashlib.sha256()
|
|
for p in parts:
|
|
h.update(p)
|
|
return h.digest()
|
|
|
|
|
|
class Stop(Exception):
|
|
def __init__(self, at, reason):
|
|
Exception.__init__(self, reason)
|
|
self.at, self.reason = at, reason
|
|
|
|
|
|
class Reader:
|
|
"""Little-endian fields in order; running out is `truncated`."""
|
|
|
|
def __init__(self, b, p):
|
|
self.b, self.p = b, p
|
|
|
|
def take(self, n):
|
|
if self.p + n > len(self.b):
|
|
raise Refuse("truncated")
|
|
s = self.b[self.p:self.p + n]
|
|
self.p += n
|
|
return s
|
|
|
|
def uint(self, n):
|
|
return int.from_bytes(self.take(n), "little")
|
|
|
|
def done(self):
|
|
return self.p == len(self.b)
|
|
|
|
|
|
def read_head(b, magic):
|
|
"""live-events §3.3's first steps, with `magic` for WEV1."""
|
|
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 check_sig(key, domain, body, sig):
|
|
bad = ED.verify(key, domain + body, sig)
|
|
if bad:
|
|
raise Refuse(bad)
|
|
|
|
|
|
# ---------------------------------------------------------------- SPEC §3.1 the listing
|
|
|
|
SEGMENT = re.compile(r"[A-Za-z0-9_-]([A-Za-z0-9_.-]*[A-Za-z0-9_-])?")
|
|
|
|
|
|
def listing(files):
|
|
"""The listing of `files` (path -> bytes), or None when a path cannot be carried."""
|
|
paths = sorted(files, key=lambda p: p.encode())
|
|
if not paths:
|
|
return None
|
|
for p in paths:
|
|
if not all(SEGMENT.fullmatch(s) for s in p.split("/")):
|
|
return None
|
|
folded = [p.lower() for p in paths]
|
|
if len(set(folded)) != len(folded):
|
|
return None
|
|
if any(q.startswith(p + "/") for p in paths for q in paths):
|
|
return None
|
|
return b"".join(sha(files[p]).hex().encode() + b" " + p.encode() + b"\n" for p in paths)
|
|
|
|
|
|
def below(files, d):
|
|
return {p[len(d):]: b for p, b in files.items() if p.startswith(d)}
|
|
|
|
|
|
# ---------------------------------------------------------------- §3, §4 the package
|
|
|
|
def is_hex64(s):
|
|
return len(s) == 64 and all(c in "0123456789abcdef" for c in s)
|
|
|
|
|
|
def component_dir(kind, cid):
|
|
return {ROOM: "room/", EVENTS: "events/", TAKE: "takes/%s/" % cid}.get(kind, "composite/")
|
|
|
|
|
|
def read_manifest(b):
|
|
"""§4.1, §4.2: the recording manifest, or Refuse."""
|
|
read_head(b, b"WRM1")
|
|
r = Reader(b, 5)
|
|
m = {"room": r.take(32), "signer": r.take(32), "from": r.uint(8), "to": r.uint(8), "claims": r.uint(1),
|
|
"purposes": r.uint(1), "region": r.take(2), "components": []}
|
|
for _ in range(r.uint(2)):
|
|
kind = r.uint(1)
|
|
cid = r.take(r.uint(1))
|
|
m["components"].append({"kind": kind, "id": cid, "sha256": r.take(32)})
|
|
m["sig"] = r.take(64)
|
|
if not r.done():
|
|
raise Refuse("malformed")
|
|
c = m["components"]
|
|
ok = (m["claims"] & ~COMPLETE == 0 and m["purposes"] & ~0b1111 == 0 and m["from"] <= m["to"]
|
|
and all(0x41 <= c <= 0x5A for c in m["region"]))
|
|
ok = ok and len(c) >= 2 and (c[0]["kind"], c[0]["id"]) == (ROOM, b"room") and (c[1]["kind"], c[1]["id"]) == (EVENTS, b"events")
|
|
rest = c[2:]
|
|
if ok and rest and rest[-1]["kind"] == COMPOSITE:
|
|
ok = rest[-1]["id"] == b"composite"
|
|
rest = rest[:-1]
|
|
ok = ok and all(x["kind"] == TAKE and is_hex64(x["id"].decode("latin-1")) for x in rest)
|
|
ok = ok and all(a["id"] < z["id"] for a, z in zip(rest, rest[1:]))
|
|
if not ok:
|
|
raise Refuse("recording_manifest_invalid")
|
|
check_sig(m["signer"], MANIFEST_DOMAIN, b[:-64], m["sig"])
|
|
m["digest"] = sha(b[:-64])
|
|
for x in c:
|
|
x["dir"] = component_dir(x["kind"], x["id"].decode())
|
|
return m
|
|
|
|
|
|
def split_chain(b):
|
|
"""§3.3: the records of a chain file, each read by live-events §3.3, or None."""
|
|
out = []
|
|
while b:
|
|
if len(b) < 118 or b[117] not in (0, 1):
|
|
return None
|
|
after = 118 + (48 if b[117] == 1 else 0)
|
|
if len(b) < after + 3:
|
|
return None
|
|
n = after + 3 + int.from_bytes(b[after + 1:after + 3], "little") + 64
|
|
if n > len(b):
|
|
return None
|
|
try:
|
|
LEV.read_record(b[:n])
|
|
except Refuse:
|
|
return None
|
|
out.append(b[:n])
|
|
b = b[n:]
|
|
return out
|
|
|
|
|
|
def room_files(files):
|
|
"""§3.3: (descriptor, grants, head), or None."""
|
|
d, grants, head = None, [], None
|
|
for p, b in sorted(below(files, "room/").items()):
|
|
if p == "descriptor.wrd":
|
|
d = b
|
|
elif p == "key-log.json":
|
|
try:
|
|
head = b.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
return None
|
|
else:
|
|
m = re.fullmatch(r"grant-([0-9]{4})\.wrg", p)
|
|
if not m or int(m.group(1)) != len(grants):
|
|
return None
|
|
grants.append(b)
|
|
return None if d is None else (d, grants, head)
|
|
|
|
|
|
def event_chains(files):
|
|
"""§3.3: each principal's chain, in the order of the files' paths, or None."""
|
|
out = []
|
|
for p, b in sorted(below(files, "events/").items(), key=lambda x: x[0].encode()):
|
|
m = re.fullmatch(r"([0-9a-f]{64})\.wev", p)
|
|
recs = split_chain(b) if m else None
|
|
if not recs or not all(LEV.read_record(x)["principal"].hex() == m.group(1) for x in recs):
|
|
return None
|
|
out.append(recs)
|
|
return out
|
|
|
|
|
|
def take_files(files, d):
|
|
"""§3.4: (manifest, parts, receipts, consents), or None."""
|
|
manifest, parts, receipts, consents = None, [], [], []
|
|
for p, b in sorted(below(files, d).items(), key=lambda x: x[0].encode()):
|
|
if p == "manifest.wtm":
|
|
manifest = b
|
|
continue
|
|
m = re.fullmatch(r"(chunk|receipt|consent)-([0-9]+)\.(bin|wur|wcn)", p)
|
|
if not m or (m.group(1), m.group(3)) not in (("chunk", "bin"), ("receipt", "wur"), ("consent", "wcn")):
|
|
return None
|
|
width = 10 if m.group(1) == "chunk" else 4
|
|
if len(m.group(2)) != width:
|
|
return None
|
|
k = int(m.group(2))
|
|
if m.group(1) == "chunk":
|
|
parts.append((k, b))
|
|
else:
|
|
lst = receipts if m.group(1) == "receipt" else consents
|
|
if k != len(lst):
|
|
return None
|
|
lst.append(b)
|
|
return None if manifest is None else (manifest, parts, receipts, consents)
|
|
|
|
|
|
def session_takes(d, chains, order, span):
|
|
"""§4.5 complete 2: the takes the room's events open within the span, each a start its
|
|
device answered."""
|
|
seen, starts, commits = set(), [], set()
|
|
for c, i, _ in order:
|
|
ev = LEV.read_record(chains[c][i])
|
|
if ev["digest"] in seen:
|
|
continue
|
|
seen.add(ev["digest"])
|
|
if ev["kind"] == TAKE_COMMIT:
|
|
try:
|
|
commits.add((ev["principal"], CT.read_commit(ev["payload"])["take"]))
|
|
except Refuse:
|
|
pass
|
|
elif ev["kind"] == CONTROL:
|
|
try:
|
|
p = LEV.read_payload(CONTROL, ev["payload"], 65535)
|
|
except Refuse:
|
|
continue
|
|
if p["command"] == 1 and span[0] <= ev["t"] <= span[1]:
|
|
starts.append((p["target"], sha(TAKE_DOMAIN, d["room"], p["target"], ev["digest"])))
|
|
return [t for dev, t in starts if (dev, t) in commits]
|
|
|
|
|
|
def signer_lapse(d, chains, order, held, revoked, signer, span):
|
|
"""§4.3: the first instant of the span at which `signer` does not hold the record power,
|
|
or None. The room state at t holds every join, grant_exchange and leave of the signer
|
|
the room verifier accepted with its t at or below t."""
|
|
changes, seen = [], set()
|
|
for c, i, _ in order:
|
|
ev = LEV.read_record(chains[c][i])
|
|
if ev["digest"] in seen or not LRV.is_room_key(d, ev["principal"]):
|
|
continue
|
|
seen.add(ev["digest"])
|
|
if ev["kind"] not in (JOIN, LEAVE, GRANT_EXCHANGE):
|
|
continue
|
|
p = LEV.read_payload(ev["kind"], ev["payload"], 65535)
|
|
if p["subject"] != signer:
|
|
continue
|
|
x = LRV.read_detail(ev["kind"], p["detail"])
|
|
changes.append((ev["t"], x["grant"] if ev["kind"] == JOIN else x["new"] if ev["kind"] == GRANT_EXCHANGE else None))
|
|
lo, hi = span
|
|
current = None
|
|
for t, g in changes:
|
|
if t <= lo:
|
|
current = g
|
|
segments, at = [], lo
|
|
for t, g in changes:
|
|
if lo < t <= hi:
|
|
if t > at:
|
|
segments.append((at, t - 1, current))
|
|
current, at = g, t
|
|
segments.append((at, hi, current))
|
|
by = {g["digest"]: g for g in held}
|
|
for a, z, g in segments:
|
|
g = by.get(g) if g is not None else None
|
|
if g is None or not g["powers"] & RECORD_POWER or LRV.revoked(g, revoked):
|
|
return a
|
|
if not LRV.good_at(g, d, a):
|
|
return a
|
|
if not LRV.good_at(g, d, z):
|
|
return max(a, g["exp"] * 1000000 - d["t0"])
|
|
return None
|
|
|
|
|
|
def verify_recording(files, at):
|
|
"""§4.4, §4.5: a dict of what the corpus states."""
|
|
lst = files.get(LISTING)
|
|
if lst is None or listing({p: b for p, b in files.items() if p != LISTING}) != lst:
|
|
return {"outcome": "refuse", "reason": "listing_mismatch", "at": ["listing"]}
|
|
try:
|
|
if MANIFEST_FILE not in files:
|
|
raise Stop(["manifest"], "recording_manifest_invalid")
|
|
try:
|
|
m = read_manifest(files[MANIFEST_FILE])
|
|
except Refuse as e:
|
|
raise Stop(["manifest"], e.reason)
|
|
for i, c in enumerate(m["components"]):
|
|
inner = below(files, c["dir"])
|
|
if not inner:
|
|
raise Stop(["component", i], "component_invalid")
|
|
if sha(listing(inner) or b"") != c["sha256"]:
|
|
raise Stop(["component", i], "component_digest_mismatch")
|
|
dirs = [c["dir"] for c in m["components"]]
|
|
if any(p not in (LISTING, MANIFEST_FILE) and not any(p.startswith(d) for d in dirs) for p in files):
|
|
raise Stop(["listing"], "component_unlisted")
|
|
rf = room_files(files)
|
|
if rf is None:
|
|
raise Stop(["component", 0], "component_invalid")
|
|
chains = event_chains(files)
|
|
if chains is None:
|
|
raise Stop(["component", 1], "component_invalid")
|
|
descriptor, grants, head = rf
|
|
room = LRV.verify_room(descriptor, grants, head, chains)
|
|
if room["outcome"] == "refuse":
|
|
raise Stop(["room"] + room["at"], room["reason"])
|
|
d = LRV.read_descriptor(descriptor)
|
|
if m["room"] != d["room"]:
|
|
raise Stop(["manifest"], "wrong_room")
|
|
if d["mode"] == 0: # §4.4 step 5: a room never recorded has no recording
|
|
raise Stop(["manifest"], "recording_not_permitted")
|
|
if m["signer"] != d["authority"]:
|
|
held, _ = LRV.check_grants(d, grants)
|
|
revoked = LRV.check_key_log(d, head)
|
|
lapse = signer_lapse(d, chains, room["order"], held, revoked, m["signer"], (m["from"], m["to"]))
|
|
if lapse is not None:
|
|
raise Stop(["signer", lapse], "recording_signer_not_permitted")
|
|
if d["regions"] and m["region"] not in d["regions"]:
|
|
raise Stop(["manifest"], "store_outside_region")
|
|
times = {}
|
|
for c, i, _ in room["order"]:
|
|
ev = LEV.read_record(chains[c][i])
|
|
times[ev["digest"]] = ev["t"]
|
|
takes = []
|
|
for i, c in enumerate(m["components"]):
|
|
if c["kind"] != TAKE:
|
|
continue
|
|
tf = take_files(files, c["dir"])
|
|
if tf is None:
|
|
raise Stop(["component", i], "component_invalid")
|
|
tm, parts, receipts, consents = tf
|
|
v = {"descriptor": descriptor.hex(), "grants": [g.hex() for g in grants], "key_log_head": head,
|
|
"chains": [[r.hex() for r in ch] for ch in chains], "manifest": tm.hex(),
|
|
"parts": [{"index": k, "bytes": b.hex()} for k, b in parts], "receipts": [r.hex() for r in receipts],
|
|
"consents": [x.hex() for x in consents], "evaluated_at": at}
|
|
got = CT.verify_take(v)
|
|
if got["outcome"] == "refuse":
|
|
raise Stop(["take", i] + got["at"], got["reason"])
|
|
if got["take"] != c["id"].decode():
|
|
raise Stop(["component", i], "component_invalid")
|
|
opened = times.get(CT.read_manifest(tm, d["room"])["start"])
|
|
closed = times.get(bytes.fromhex(got["commits"][-1])) if got["commits"] else None
|
|
if opened is None or opened < m["from"] or (closed is not None and closed > m["to"]):
|
|
raise Stop(["component", i], "outside_span")
|
|
takes.append(got)
|
|
except Stop as e:
|
|
return {"outcome": "refuse", "reason": e.reason, "at": e.at}
|
|
# §4.5.
|
|
complete = "holds"
|
|
held = {t["take"] for t in takes}
|
|
if room["outcome"] == "gap":
|
|
complete = "delivery_gap"
|
|
elif any(t.hex() not in held for t in session_takes(d, chains, room["order"], (m["from"], m["to"]))):
|
|
complete = "take_omitted"
|
|
else:
|
|
for t in takes:
|
|
vd = t["verdicts"]
|
|
if vd["single_take"] in ("delivery_gap", "commit_chain_broken", "take_not_closed"):
|
|
complete = vd["single_take"]
|
|
break
|
|
if vd["complete"] != "holds":
|
|
complete = vd["complete"]
|
|
break
|
|
publishable = {n: next((t["verdicts"]["publishable"][n] for t in takes if t["verdicts"]["publishable"][n] != "holds"), "holds")
|
|
for _, n in PURPOSES}
|
|
stands = ((not m["claims"] & COMPLETE or complete == "holds")
|
|
and all(not m["purposes"] & b or publishable[n] == "holds" for b, n in PURPOSES))
|
|
return {"outcome": "verdicts", "package": sha(lst).hex(), "manifest": m["digest"].hex(), "room": d["room"].hex(),
|
|
"complete": complete, "publishable": publishable, "stands": stands,
|
|
"takes": [{"take": t["take"], "single_take": t["verdicts"]["single_take"], "aligned": t["verdicts"]["aligned"],
|
|
"complete": t["verdicts"]["complete"], "publishable": t["verdicts"]["publishable"],
|
|
"stands": t["stands"]} for t in takes]}
|
|
|
|
|
|
# ---------------------------------------------------------------- §5.3 the range proof, RFC 9162
|
|
|
|
def leaf(desc):
|
|
return sha(LEAF_DOMAIN, desc)
|
|
|
|
|
|
def node(a, b):
|
|
return sha(NODE_DOMAIN, a, b)
|
|
|
|
|
|
def split(n):
|
|
"""The largest power of two below n, n >= 2."""
|
|
k = 1
|
|
while k * 2 < n:
|
|
k *= 2
|
|
return k
|
|
|
|
|
|
def mth(ls):
|
|
"""RFC 9162 §2.1.1, with the take tree's node; n >= 1."""
|
|
if len(ls) == 1:
|
|
return ls[0]
|
|
k = split(len(ls))
|
|
return node(mth(ls[:k]), mth(ls[k:]))
|
|
|
|
|
|
def path(m, d):
|
|
"""RFC 6962 §2.1.1's PATH(m, D[n])."""
|
|
if len(d) <= 1:
|
|
return []
|
|
k = split(len(d))
|
|
return path(m, d[:k]) + [mth(d[k:])] if m < k else path(m - k, d[k:]) + [mth(d[:k])]
|
|
|
|
|
|
def range_nodes(a, b, d):
|
|
"""§5.3's RANGE(a, b, D[n])."""
|
|
n = len(d)
|
|
if a == 0 and b == n:
|
|
return []
|
|
k = split(n)
|
|
if b <= k:
|
|
return range_nodes(a, b, d[:k]) + [mth(d[k:])]
|
|
if a >= k:
|
|
return range_nodes(a - k, b - k, d[k:]) + [mth(d[:k])]
|
|
return range_nodes(a, k, d[:k]) + range_nodes(0, b - k, d[k:])
|
|
|
|
|
|
def range_root(a, b, n, ls, nodes):
|
|
"""§5.3's ROOT(a, b, n, L), taking nodes from the front of `nodes`; None when they run out."""
|
|
if a == 0 and b == n:
|
|
return mth(ls)
|
|
k = split(n)
|
|
if b <= k:
|
|
x = range_root(a, b, k, ls, nodes)
|
|
return None if x is None or not nodes else node(x, nodes.pop(0))
|
|
if a >= k:
|
|
x = range_root(a - k, b - k, n - k, ls, nodes)
|
|
return None if x is None or not nodes else node(nodes.pop(0), x)
|
|
x = range_root(a, k, k, ls[:k - a], nodes)
|
|
y = None if x is None else range_root(0, b - k, n - k, ls[k - a:], nodes)
|
|
return None if y is None else node(x, y)
|
|
|
|
|
|
def range_verifies(first, ls, proof, size, root):
|
|
"""§5.3: the proof shows `ls` are leaves `first` onward of the tree (size, root)."""
|
|
if not ls or first + len(ls) > size:
|
|
return False
|
|
nodes = list(proof)
|
|
r = range_root(first, first + len(ls), size, ls, nodes)
|
|
return r is not None and not nodes and r == root
|
|
|
|
|
|
def rfc9162_inclusion(leaf_hash, leaf_index, tree_size, inclusion_path, root_hash):
|
|
"""RFC 9162 §2.1.3.2, as its steps state it, with node() for HASH(0x01 || ...)."""
|
|
if leaf_index >= tree_size:
|
|
return False
|
|
fn, sn = leaf_index, tree_size - 1
|
|
r = leaf_hash
|
|
for p in inclusion_path:
|
|
if sn == 0:
|
|
return False
|
|
if fn & 1 or fn == sn:
|
|
r = node(p, r)
|
|
if not fn & 1:
|
|
while not fn & 1 and fn != 0:
|
|
fn >>= 1
|
|
sn >>= 1
|
|
else:
|
|
r = node(r, p)
|
|
fn >>= 1
|
|
sn >>= 1
|
|
return sn == 0 and r == root_hash
|
|
|
|
|
|
def self_test():
|
|
"""The range check against RFC 9162 §2.1.3.2 and RFC 6962's PATH: the failures."""
|
|
fails = []
|
|
ls = [sha(b"t", bytes([i])) for i in range(22)]
|
|
for n in range(1, 22):
|
|
d, root = ls[:n], mth(ls[:n])
|
|
for a in range(n):
|
|
p = path(a, d)
|
|
if range_nodes(a, a + 1, d) != p:
|
|
fails.append("RANGE(%d, %d) is not PATH at n = %d" % (a, a + 1, n))
|
|
if not rfc9162_inclusion(d[a], a, n, p, root) or not range_verifies(a, [d[a]], p, n, root):
|
|
fails.append("an inclusion proof does not verify at n = %d, a = %d" % (n, a))
|
|
for bad in ([x for x in p[:-1]] + [bytes(32)] if p else [], p + [bytes(32)]):
|
|
if rfc9162_inclusion(d[a], a, n, bad, root) != range_verifies(a, [d[a]], bad, n, root):
|
|
fails.append("the two checks disagree at n = %d, a = %d" % (n, a))
|
|
for b in range(a + 1, n + 1):
|
|
q = range_nodes(a, b, d)
|
|
if not range_verifies(a, d[a:b], q, n, root):
|
|
fails.append("RANGE(%d, %d) does not verify at n = %d" % (a, b, n))
|
|
if b < n and range_verifies(a + 1, d[a:b], q, n, root):
|
|
fails.append("a moved run verifies at n = %d" % n)
|
|
if q and range_verifies(a, d[a:b], q[:-1], n, root):
|
|
fails.append("a short proof verifies at n = %d" % n)
|
|
return fails
|
|
|
|
|
|
# ---------------------------------------------------------------- §5 clips
|
|
|
|
def read_clip(b):
|
|
"""§5.1, §5.2: the clip record, or Refuse."""
|
|
read_head(b, CLIP_MAGIC)
|
|
r = Reader(b, 5)
|
|
c = {"room": r.take(32), "in": r.uint(8), "out": r.uint(8), "parts": []}
|
|
for _ in range(r.uint(1)):
|
|
p = {"device": r.take(32), "take": r.take(32), "head": r.take(32), "size": r.uint(8), "root": r.take(32),
|
|
"first": r.uint(8)}
|
|
count = r.uint(4)
|
|
p["trim_in"], p["trim_out"] = r.uint(8), r.uint(8)
|
|
p["chunks"] = [CT.read_chunk(CT.Reader(r.take(64), 0, "truncated")) for _ in range(count)]
|
|
p["proof"] = [r.take(32) for _ in range(r.uint(1))]
|
|
c["parts"].append(p)
|
|
c["signer"] = r.take(32)
|
|
c["sig"] = r.take(64)
|
|
if not r.done():
|
|
raise Refuse("malformed")
|
|
ps = c["parts"]
|
|
ok = 1 <= len(ps) <= 16 and len({p["take"] for p in ps}) == len(ps) and c["in"] < c["out"]
|
|
trimmed = any(p["trim_in"] or p["trim_out"] for p in ps)
|
|
ok = ok and (not trimmed or len(ps) == 1)
|
|
for p in ps:
|
|
ch = p["chunks"]
|
|
ok = (ok and len(ch) >= 1 and all(x["index"] == p["first"] + i and x["len"] >= 1 for i, x in enumerate(ch))
|
|
and len(p["proof"]) <= 128 and p["trim_in"] < ch[0]["len"] and p["trim_out"] < ch[-1]["len"]
|
|
and (len(ch) > 1 or p["trim_in"] + p["trim_out"] < ch[0]["len"]))
|
|
if not ok:
|
|
raise Refuse("clip_record_invalid")
|
|
check_sig(c["signer"], CLIP_DOMAIN, b[:-64], c["sig"])
|
|
c["digest"] = sha(b[:-64])
|
|
return c
|
|
|
|
|
|
def verify_part(k, p, given, d, clip, exempt=frozenset()):
|
|
"""§5.4 step 3 for part k: its facts, or Stop. A chunk whose leaf is in `exempt` is passed
|
|
over in step (k), as live-moderation §8.3 has a restricted chunk passed over; none is by
|
|
default."""
|
|
I, J = d["commit_interval_ms"] * 1000, d["jitter_ms"] * 1000
|
|
at_part = ["part", k]
|
|
if bytes.fromhex(given["device"]) != p["device"]: # (a)
|
|
raise Stop(at_part, "clip_device_mismatch")
|
|
held = []
|
|
for j, h in enumerate(given["commits"]): # (b)
|
|
at = ["commit", k, j]
|
|
b = bytes.fromhex(h)
|
|
try:
|
|
ev = LEV.read_record(b)
|
|
except Refuse as e:
|
|
raise Stop(at, e.reason)
|
|
if ev["room"] != d["room"]:
|
|
raise Stop(at, "wrong_room")
|
|
if ev["principal"] != p["device"]:
|
|
raise Stop(at, "clip_device_mismatch")
|
|
bad = ED.verify(ev["principal"], LEV.SIGN_DOMAIN + ev["body"], ev["sig"])
|
|
if bad:
|
|
raise Stop(at, bad)
|
|
if ev["kind"] != TAKE_COMMIT:
|
|
raise Stop(at, "commit_malformed")
|
|
try:
|
|
c = CT.read_commit(ev["payload"])
|
|
except Refuse as e:
|
|
raise Stop(at, e.reason)
|
|
if c["take"] != p["take"]:
|
|
raise Stop(at, "clip_take_mismatch")
|
|
held.append((ev, c))
|
|
if not held: # (c)
|
|
raise Stop(at_part, "commit_chain_broken")
|
|
if not CT.is_opening(held[0][1]):
|
|
raise Stop(["commit", k, 0], "commit_chain_broken")
|
|
for j in range(1, len(held)):
|
|
(pe, pc), (_, c) = held[j - 1], held[j]
|
|
if c["n"] > pc["n"] + 1:
|
|
raise Stop(["commit", k, j], "delivery_gap")
|
|
if not (pc["phase"] != CLOSE and c["phase"] != OPEN and c["n"] == pc["n"] + 1 and c["prev"] == pe["digest"]
|
|
and c["first"] == pc["size"]):
|
|
raise Stop(["commit", k, j], "commit_chain_broken")
|
|
for j in range(1, len(held)):
|
|
pc, c = held[j - 1][1], held[j][1]
|
|
if not CT.extends(c["first"], pc["root"], c["size"], c["root"], c["proof"]):
|
|
raise Stop(["commit", k, j], "root_not_extending")
|
|
hev, hc = held[-1] # (d)
|
|
if hev["digest"] != p["head"] or hc["size"] != p["size"] or hc["root"] != p["root"]:
|
|
raise Stop(at_part, "tree_head_not_committed")
|
|
ch = p["chunks"]
|
|
if p["first"] + len(ch) > p["size"]: # (e)
|
|
raise Stop(at_part, "clip_out_of_range")
|
|
if not range_verifies(p["first"], [leaf(x["desc"]) for x in ch], p["proof"], p["size"], p["root"]): # (f)
|
|
raise Stop(at_part, "clip_not_included")
|
|
if len(ch) == 1 and rfc9162_inclusion(leaf(ch[0]["desc"]), p["first"], p["size"], p["proof"], p["root"]) is not True:
|
|
raise Stop(at_part, "clip_not_included")
|
|
o = held[0][1]["offset"]
|
|
for j, (_, c) in enumerate(held): # (g)
|
|
if abs(c["offset"] - o) > J:
|
|
raise Stop(["commit", k, j], "clock_unstable")
|
|
if any(x["start"] >= x["end"] for x in ch) or any(b["start"] != a["end"] for a, b in zip(ch, ch[1:])): # (h)
|
|
raise Stop(at_part, "capture_discontinuous")
|
|
if any(x["end"] - x["start"] > I for x in ch):
|
|
raise Stop(at_part, "chunk_too_long")
|
|
|
|
def adder(x):
|
|
return next(ev for ev, c in held if c["first"] <= x["index"] < c["size"])
|
|
if any(adder(x)["t"] + J < x["end"] + o for x in ch): # (i)
|
|
raise Stop(at_part, "commit_early")
|
|
if any(adder(x)["t"] > x["end"] + o + I + J for x in ch):
|
|
raise Stop(at_part, "commit_late")
|
|
span = (ch[0]["start"] + o, ch[-1]["end"] + o) # (j)
|
|
wi, wo = clip["in"], clip["out"]
|
|
covers = span[0] <= wi and wo <= span[1] and wi < ch[0]["end"] + o and ch[-1]["start"] + o < wo
|
|
if not covers or ((p["trim_in"] or p["trim_out"]) and (wi, wo) != span):
|
|
raise Stop(at_part, "clip_window_invalid")
|
|
shown = b"" # (k)
|
|
trim_in, trim_out = p["trim_in"], p["trim_out"]
|
|
for j, x in enumerate(ch):
|
|
if leaf(x["desc"]) in exempt:
|
|
trim_in = 0 if j == 0 else trim_in
|
|
trim_out = 0 if j == len(ch) - 1 else trim_out
|
|
continue
|
|
h = given["bytes"][j] if j < len(given["bytes"]) else None
|
|
if h is None:
|
|
raise Stop(["chunk", k, j], "clip_chunk_missing")
|
|
b = bytes.fromhex(h)
|
|
if not CT.is_chunk(x, b):
|
|
other = any(y["index"] != x["index"] and CT.is_chunk(y, b) for y in ch)
|
|
raise Stop(["chunk", k, j], "chunk_reordered" if other else "chunk_altered")
|
|
shown += b
|
|
if len(given["bytes"]) > len(ch):
|
|
raise Stop(["chunk", k, len(ch)], "chunk_extra")
|
|
shown = shown[trim_in:len(shown) - trim_out]
|
|
return {"take": p["take"].hex(), "first": p["first"], "count": len(ch), "offset": o, "span": list(span),
|
|
"device_window": [wi - o, wo - o], "bytes": len(shown), "sha256": sha(shown).hex(),
|
|
"fetch": {"namespace": [d["room"].hex(), p["device"].hex(), "take"], "name": p["take"].hex(),
|
|
"start_group": p["first"], "start_object": 0, "end_group_delta": len(ch) - 1}}
|
|
|
|
|
|
def verify_clip(v, exempt=frozenset()):
|
|
"""§5.4: a dict of what the corpus states. `exempt`: as verify_part's."""
|
|
try:
|
|
try:
|
|
d = LRV.read_descriptor(bytes.fromhex(v["descriptor"]))
|
|
except Refuse as e:
|
|
raise Stop(["descriptor"], e.reason)
|
|
try:
|
|
clip = read_clip(bytes.fromhex(v["record"]))
|
|
except Refuse as e:
|
|
raise Stop(["clip"], e.reason)
|
|
if clip["room"] != d["room"]:
|
|
raise Stop(["clip"], "wrong_room")
|
|
if len(v["parts"]) != len(clip["parts"]):
|
|
raise Stop(["clip"], "clip_record_invalid")
|
|
facts = [verify_part(k, p, g, d, clip, exempt) for k, (p, g) in enumerate(zip(clip["parts"], v["parts"]))]
|
|
except Stop as e:
|
|
return {"outcome": "refuse", "reason": e.reason, "at": e.at}
|
|
J = d["jitter_ms"] * 1000
|
|
return {"outcome": "accept", "clip": clip["digest"].hex(), "signer": clip["signer"].hex(), "window": [clip["in"], clip["out"]],
|
|
"bound_us": J if len(facts) == 1 else 2 * J, "facts": facts}
|
|
|
|
|
|
# ---------------------------------------------------------------- the corpus
|
|
|
|
def files_of(v):
|
|
return {p: bytes.fromhex(h) for p, h in v["files"].items()}
|
|
|
|
|
|
def problems(v, got):
|
|
out = [k for k in got if v.get(k) != got[k]]
|
|
return out + [k for k in ("outcome", "reason", "at") if k in v and k not in got]
|
|
|
|
|
|
def check(doc, quiet=False):
|
|
"""Every vector against its statement: the failures."""
|
|
fails = []
|
|
for kind, f in (("recordings", lambda v: verify_recording(files_of(v), v["evaluated_at"])), ("clips", verify_clip)):
|
|
for v in doc.get(kind, []):
|
|
try:
|
|
got = f(v)
|
|
except Exception as e: # a crash on a mutated copy is a failure to read it
|
|
got, bad = None, ["the vector does not verify: %s: %s" % (type(e).__name__, e)]
|
|
else:
|
|
bad = problems(v, 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") + " %s %s: %s" % (kind[:-1], v["name"], what))
|
|
for k in bad:
|
|
fails.append("%s %s: %s is %s, the corpus states %s" % (
|
|
kind[:-1], v["name"], k, json.dumps(got.get(k))[:200] if got else k, json.dumps(v.get(k))[:200]))
|
|
return fails
|
|
|
|
|
|
def reuse_fails(doc, takes):
|
|
"""Every take the corpus says it reuses is capture binding's, byte for byte."""
|
|
by = {v["name"]: v for v in takes["vectors"]}
|
|
fails, n = [], 0
|
|
for v in doc["recordings"]:
|
|
for r in v["reuses"]:
|
|
t, d, f = by[r["vector"]], "takes/%s/" % r["take"], v["files"]
|
|
parts = sorted((int(p[len(d) + 6:len(d) + 16]), h) for p, h in f.items() if p.startswith(d + "chunk-"))
|
|
same = (t["take"] == r["take"] and f.get(d + "manifest.wtm") == t["manifest"]
|
|
and parts == [(p["index"], p["bytes"]) for p in t["parts"]]
|
|
and [h for p, h in sorted(f.items()) if p.startswith(d + "receipt-")] == t["receipts"]
|
|
and [h for p, h in sorted(f.items()) if p.startswith(d + "consent-")] == t["consents"])
|
|
if not same:
|
|
fails.append("recording %s: its take %s is not %s" % (v["name"], r["take"][:12], r["vector"]))
|
|
n += 1
|
|
for c in doc["clips"]:
|
|
for p in c["parts"]:
|
|
if p["take_vector"] is None:
|
|
continue
|
|
t = by[p["take_vector"]]
|
|
records = {x for ch in t["chains"] for x in ch}
|
|
parts = {x["bytes"] for x in t["parts"]}
|
|
if not all(x in records for x in p["commits"]) or not all(b is None or b in parts for b in p["bytes"]):
|
|
fails.append("clip %s: a part is not %s's" % (c["name"], p["take_vector"]))
|
|
n += 1
|
|
if n == 0:
|
|
fails.append("the corpus reuses no take")
|
|
return fails
|
|
|
|
|
|
def flip(h, at):
|
|
b = bytearray(bytes.fromhex(h))
|
|
b[at] ^= 0x01
|
|
return b.hex()
|
|
|
|
|
|
def mutations(doc):
|
|
"""Copies of the corpus, each with one defect a conforming verifier must catch."""
|
|
rn = {v["name"]: k for k, v in enumerate(doc["recordings"])}
|
|
cn = {v["name"]: k for k, v in enumerate(doc["clips"])}
|
|
|
|
def mutate(label, f):
|
|
d = copy.deepcopy(doc)
|
|
f(d["recordings"], d["clips"])
|
|
return label, d
|
|
|
|
def first_chunk(v):
|
|
return next(p for p in sorted(v["files"]) if "/chunk-" in p)
|
|
|
|
good, single, aligned = rn["lesson_session"], cn["single_chunk"], cn["two_takes_aligned"]
|
|
return [
|
|
mutate("a byte of the session's first chunk file", lambda rs, cs: rs[good]["files"].update({first_chunk(rs[good]): flip(rs[good]["files"][first_chunk(rs[good])], 3)})),
|
|
mutate("the session's listing dropped", lambda rs, cs: rs[good]["files"].pop("LISTING")),
|
|
mutate("a byte of the session's manifest signature", lambda rs, cs: rs[good]["files"].update({"recording.wrm": flip(rs[good]["files"]["recording.wrm"], -9)})),
|
|
mutate("the session's stated package digest", lambda rs, cs: rs[good].update(package="00" * 32)),
|
|
mutate("the session evaluated a year on", lambda rs, cs: rs[good].update(evaluated_at=rs[good]["evaluated_at"] + 365 * 86400)),
|
|
mutate("the omitted take stated complete", lambda rs, cs: rs[rn["take_omitted"]].update(complete="holds")),
|
|
mutate("the missing chunk stated complete", lambda rs, cs: rs[rn["missing_chunk"]].update(complete="holds")),
|
|
mutate("the unpermitted signer stated at the listing", lambda rs, cs: rs[rn["signer_not_permitted"]].update(at=["listing"])),
|
|
mutate("the removed signer's power stated lost a second early", lambda rs, cs: rs[rn["signer_removed"]].update(at=["signer", 17000000])),
|
|
mutate("the late recorder's span stated from 0", lambda rs, cs: rs[rn["signer_power_late"]].update(at=["signer", 1000000])),
|
|
mutate("a byte of the single chunk's bytes", lambda rs, cs: cs[single]["parts"][0]["bytes"].__setitem__(0, flip(cs[single]["parts"][0]["bytes"][0], 2))),
|
|
mutate("a byte of the single chunk's record", lambda rs, cs: cs[single].update(record=flip(cs[single]["record"], 200))),
|
|
mutate("the single chunk's last commitment dropped", lambda rs, cs: cs[single]["parts"][0]["commits"].pop()),
|
|
mutate("the single chunk's stated digest", lambda rs, cs: cs[single]["facts"][0].update(sha256="00" * 32)),
|
|
mutate("the aligned clip's stated bound", lambda rs, cs: cs[aligned].update(bound_us=500000)),
|
|
mutate("the aligned clip's parts given in each other's place", lambda rs, cs: cs[aligned]["parts"].reverse()),
|
|
mutate("the byte range's stated fetch", lambda rs, cs: cs[cn["byte_range"]]["facts"][0]["fetch"].update(start_group=3)),
|
|
mutate("the byte range's stated window on the device clock", lambda rs, cs: cs[cn["byte_range"]]["facts"][0]["device_window"].__setitem__(0, 0)),
|
|
mutate("the live head's commitment 2 dropped", lambda rs, cs: cs[cn["live_head"]]["parts"][0]["commits"].pop()),
|
|
mutate("the missing chunk given chunk 5's bytes", lambda rs, cs: cs[cn["missing_chunk"]]["parts"][0]["bytes"].__setitem__(1, cs[cn["missing_chunk"]]["parts"][0]["bytes"][0])),
|
|
mutate("the spliced clip stated accepted", lambda rs, cs: cs[cn["spliced_two_takes"]].update(outcome="accept")),
|
|
mutate("the uncommitted head stated at part 1", lambda rs, cs: cs[cn["uncommitted_head"]].update(at=["part", 1])),
|
|
mutate("the clean span's late commitment dropped", lambda rs, cs: cs[cn["late_take_clean_span"]]["parts"][0]["commits"].pop(3)),
|
|
]
|
|
|
|
|
|
def link_magic_fails(doc):
|
|
"""A copy of each accepted clip whose record begins with the contribution link record's
|
|
magic in place of WCL1: §5.2 reads it as live-events §3.3 does, so it is refused as
|
|
`malformed` at the clip, before its signature is read. The failures."""
|
|
fails, n = [], 0
|
|
for v in doc["clips"]:
|
|
if v["outcome"] != "accept":
|
|
continue
|
|
n += 1
|
|
record = LINK_MAGIC.hex() + v["record"][2 * len(CLIP_MAGIC):]
|
|
try:
|
|
got = verify_clip(dict(v, record=record))
|
|
except Exception as e: # a crash is a failure to refuse it
|
|
got = "%s: %s" % (type(e).__name__, e)
|
|
if got != {"outcome": "refuse", "reason": "malformed", "at": ["clip"]}:
|
|
fails.append("clip %s given the link record's magic: %s" % (v["name"], json.dumps(got)[:160]))
|
|
if n == 0:
|
|
fails.append("no accepted clip to give the link record's magic")
|
|
return fails, n
|
|
|
|
|
|
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 []
|
|
rt = self_test()
|
|
print(("PASS" if not rt else "FAIL") + " range proofs: §5.3's recursion against RFC 9162 §2.1.3.2 and RFC 6962's PATH")
|
|
fails += rt
|
|
with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as f:
|
|
doc = json.load(f)
|
|
with open(os.path.join(corpus_dir, "..", "takes", "vectors.json"), encoding="utf-8") as f:
|
|
takes = json.load(f)
|
|
fails += check(doc)
|
|
seen = {v.get("reason") for v in doc["recordings"] + doc["clips"]} | {v.get("complete") for v in doc["recordings"]}
|
|
missing = [r for r in REASONS if r not in seen]
|
|
if missing:
|
|
fails.append("no vector gives %s" % ", ".join(missing))
|
|
good = [v["name"] for v in doc["recordings"] if v["name"] in ("lesson_session", "signed_by_director", "with_composite", "late_recorder_span") and not v.get("stands")]
|
|
good += [v["name"] for v in doc["clips"] if v["name"] in ("single_chunk", "byte_range", "whole_take", "first_chunk",
|
|
"last_chunk", "two_takes_aligned", "live_head") and v["outcome"] != "accept"]
|
|
if good:
|
|
fails.append("good vectors that do not stand or are not accepted: %s" % ", ".join(good))
|
|
print("%s every reason of §9 appears, the good recordings stand and the good clips are accepted" % ("PASS" if not (missing or good) else "FAIL"))
|
|
rf = reuse_fails(doc, takes)
|
|
print("%s every take the corpus reuses is capture binding's, byte for byte" % ("PASS" if not rf else "FAIL"))
|
|
fails += rf
|
|
lf, n = link_magic_fails(doc)
|
|
print("%s mutated copies refused: each of %d accepted clips given the link record's magic, WLK1, is malformed at the clip" % ("PASS" if not lf else "FAIL", n))
|
|
fails += lf
|
|
for label, mutated in mutations(doc):
|
|
changed = {k: [v for v, o in zip(mutated[k], doc[k]) if v != o] for k in ("recordings", "clips")}
|
|
caught = check(changed, quiet=True) if any(changed.values()) else []
|
|
print(("PASS" if caught else "FAIL") + " mutated copy fails: %s%s" % (label, " (%s)" % caught[0][:110] 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-recording corpus: %d recordings, %d clips" % (len(doc["recordings"]), len(doc["clips"])))
|
|
return 1 if fails else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) != 2:
|
|
print("usage: live_record_verify.py <live-conformance/recordings>", file=sys.stderr)
|
|
sys.exit(2)
|
|
sys.exit(main(sys.argv[1]))
|