Transaction-Science/open-standards
Watch
1
0
Fork
You've already forked open-standards
 0
Code Issues Pull requests Projects Releases 15 Packages Wiki Activity Actions 128
open-standards/wai/live-conformance/capture_take_verify.py

683 lines
28 KiB
Python
Raw Permalink Blame History

#!/usr/bin/env python3
"""Independent verifier of the capture-binding corpus (extensions/capture-binding.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. The room is verified by the live-rooms corpus's own
second implementation, live_rooms_verify.py (live-rooms §9), which verifies every event
with live_events_verify.py (live-events §6); both are likewise written from their texts
alone. Everything else is here:
- the take tree (§4): leaves of the take's domain, the Merkle tree hash of RFC 6962 §2.1
with staged-measurement Appendix C's node, and consistency checked as RFC 9162 §2.1.4.2
checks it;
- the take_commit payload and its chain (§5), the upload receipts and their state machine
(§6), the take manifest (§7) and the consent record (§8), each read and checked here;
- take verification (§9): the take in the room, the store, the receipts, the consents,
the verdicts of single-take, aligned, complete and publishable, and whether the
manifest stands; the master position of every marker (§10); the stream id (§11.1).
For every vector it requires the outcome, reason and place, and for one with verdicts the
take id, stream id, commitments, size, root, verdicts, standing and markers the corpus
states; of the corpus, that every reason of §12 and delivery_gap appear in it. Then it
mutates copies of the corpus, one defect each, and requires every copy to fail.
python3 capture_take_verify.py <live-conformance/takes>
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 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 _ed25519 as ED # noqa: E402
Refuse = LEV.Refuse
TAKE_DOMAIN = b"wai:capture-take\x01"
LEAF_DOMAIN = b"wai:take-leaf\x01"
NODE_DOMAIN = b"wai:out-node\x01"
MANIFEST_DOMAIN = b"wai:capture-manifest\x01"
RECEIPT_DOMAIN = b"wai:take-upload\x01"
CONSENT_DOMAIN = b"wai:capture-consent\x01"
STORE_AUTHORITY = "capture/store"
ZERO = bytes(32)
OPEN, RUNNING, CLOSE = 1, 2, 3
R_OPEN, R_PROGRESS, R_RESUME, R_SEALED = 1, 2, 3, 4
PURPOSES = [(1, "live"), (2, "record"), (4, "distribute"), (8, "train")]
RECORD = 2
SINGLE_TAKE, ALIGNED, COMPLETE = 1, 2, 4
TAKE_COMMIT, MARKER, CONTROL = 0x0C, 0x08, 0x09
# §12, in its order.
REASONS = [
"recording_not_permitted", "take_unknown", "commit_malformed", "device_not_bound",
"manifest_invalid", "manifest_mismatch", "store_not_listed", "upload_receipt_invalid",
"upload_receipt_mismatch", "store_outside_region", "consent_invalid",
"commit_chain_broken", "root_not_extending", "take_not_closed", "manifest_not_committed",
"clock_unstable", "capture_discontinuous", "chunk_too_long", "commit_early", "commit_late",
"chunk_extra", "chunk_reordered", "chunk_altered", "misaligned", "chunk_missing",
"upload_not_sealed", "consent_missing", "guardian_missing", "consent_revoked",
"consent_expired", "purpose_not_consented",
]
def sha(*parts):
h = hashlib.sha256()
for p in parts:
h.update(p)
return h.digest()
class Reader:
"""Reads little-endian fields in order; running out is `out`."""
def __init__(self, b, p, out):
self.b, self.p, self.out = b, p, out
def take(self, n):
if self.p + n > len(self.b):
raise Refuse(self.out)
s = self.b[self.p:self.p + n]
self.p += n
return s
def u8(self):
return self.take(1)[0]
def uint(self, n):
return int.from_bytes(self.take(n), "little")
def i64(self):
return int.from_bytes(self.take(8), "little", signed=True)
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)
# ---------------------------------------------------------------- §4 the take tree
def leaf(desc):
return sha(LEAF_DOMAIN, desc)
def node(left, right):
return sha(NODE_DOMAIN, left, right)
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 6962 §2.1, n >= 1."""
if len(ls) == 1:
return ls[0]
k = split(len(ls))
return node(mth(ls[:k]), mth(ls[k:]))
def root_of(chunks):
return ZERO if not chunks else mth([leaf(c["desc"]) for c in chunks])
def consistent(first, second, first_root, second_root, path):
"""RFC 9162 §2.1.4.2, with node() for HASH(0x01 || ...); 0 < first < second."""
if not path:
return False
if first & (first - 1) == 0:
path = [first_root] + list(path)
fn, sn = first - 1, second - 1
while fn & 1:
fn >>= 1
sn >>= 1
fr = sr = path[0]
for c in path[1:]:
if sn == 0:
return False
if fn & 1 or fn == sn:
fr = node(c, fr)
sr = node(c, sr)
if not fn & 1:
while not fn & 1 and fn != 0:
fn >>= 1
sn >>= 1
else:
sr = node(sr, c)
fn >>= 1
sn >>= 1
return fr == first_root and sr == second_root and sn == 0
def extends(first, old, size, new, proof):
"""§4.3, with the two cases RFC 9162 leaves to the caller."""
if first == 0:
return not proof
if first == size:
return not proof and old == new
return first < size and consistent(first, size, old, new, proof)
# ---------------------------------------------------------------- §5, §6, §7, §8 records
def read_commit(b):
"""§5.2."""
r = Reader(b, 0, "commit_malformed")
c = {"take": r.take(32), "n": r.uint(4), "phase": r.u8()}
if c["phase"] not in (OPEN, RUNNING, CLOSE):
raise Refuse("commit_malformed")
c["prev"], c["first"], c["size"] = r.take(32), r.uint(8), r.uint(8)
if c["size"] < c["first"]:
raise Refuse("commit_malformed")
c["root"] = r.take(32)
if c["size"] == 0 and c["root"] != ZERO:
raise Refuse("commit_malformed")
c["offset"] = r.i64()
k = r.u8()
if k > 64:
raise Refuse("commit_malformed")
c["proof"] = [r.take(32) for _ in range(k)]
if not r.done():
raise Refuse("commit_malformed")
return c
def is_opening(c):
return (c["n"] == 0 and c["phase"] == OPEN and c["prev"] == ZERO and c["first"] == 0 and c["size"] == 0
and not c["proof"])
def read_chunk(r):
desc = r.take(64)
return {"desc": desc, "index": int.from_bytes(desc[0:8], "little"), "start": int.from_bytes(desc[8:16], "little"),
"end": int.from_bytes(desc[16:24], "little"), "len": int.from_bytes(desc[24:32], "little"), "hash": desc[32:]}
def is_chunk(c, b):
return len(b) == c["len"] and sha(b) == c["hash"]
def read_receipt(b):
"""§6.2."""
read_head(b, b"WUR1")
r = Reader(b, 5, "truncated")
x = {"store": r.take(32), "take": r.take(32), "seq": r.uint(8), "prev": r.take(32), "at": r.uint(8),
"state": r.u8(), "held": r.uint(8), "root": r.take(32), "region": r.take(2), "sig": r.take(64)}
if not r.done():
raise Refuse("malformed")
x["body"] = b[:-64]
x["digest"] = sha(x["body"])
return x
def read_manifest(b, room):
"""§7.2: the manifest, or Refuse."""
read_head(b, b"WTM1")
r = Reader(b, 5, "truncated")
m = {k: r.take(32) for k in ("room", "device", "principal", "start", "take", "store")}
m["claims"], m["purposes"] = r.u8(), r.u8()
m["chunks"] = [read_chunk(r) for _ in range(r.uint(4))]
m["root"] = r.take(32)
m["commits"] = [r.take(32) for _ in range(r.uint(4))]
m["depicted"] = [(r.take(32), r.take(32)) for _ in range(r.u8())]
m["sig"] = r.take(64)
if not r.done():
raise Refuse("malformed")
keys = [p for p, _ in m["depicted"]]
ok = (m["claims"] & ~0b111 == 0 and m["purposes"] & ~0b1111 == 0 and len(m["chunks"]) >= 1
and all(c["index"] == i and c["len"] >= 1 for i, c in enumerate(m["chunks"]))
and m["root"] == root_of(m["chunks"])
and len(m["commits"]) >= 1 and len(set(m["commits"])) == len(m["commits"])
and 1 <= len(keys) <= 32 and all(a < z for a, z in zip(keys, keys[1:])) and m["principal"] in keys
and m["take"] == sha(TAKE_DOMAIN, m["room"], m["device"], m["start"]))
if not ok:
raise Refuse("manifest_invalid")
if m["room"] != room:
raise Refuse("wrong_room")
check_sig(m["device"], MANIFEST_DOMAIN, b[:-64], m["sig"])
return m
def read_consent(b):
"""§8.1, §8.2."""
read_head(b, b"WCN1")
r = Reader(b, 5, "truncated")
c = {"subject": r.take(32), "minor": r.u8(), "guardian": r.take(32), "purposes": r.u8(), "room": r.take(32),
"nbf": r.uint(8), "exp": r.uint(8), "nonce": r.take(16), "sig": r.take(64)}
if not r.done():
raise Refuse("malformed")
ok = (c["minor"] in (0, 1) and (c["minor"] == 1 or c["guardian"] == ZERO) and c["guardian"] != c["subject"]
and c["purposes"] != 0 and c["purposes"] & ~0b1111 == 0 and c["nbf"] < c["exp"])
if not ok:
raise Refuse("consent_invalid")
signer = c["subject"] if c["guardian"] == ZERO else c["guardian"]
check_sig(signer, CONSENT_DOMAIN, b[:-64], c["sig"])
c["digest"] = sha(b[:-64])
return c
def stream_id(take):
"""§11.1: an RFC 9562 version-8 UUID of the take id's first 16 bytes."""
u = bytearray(take[:16])
u[6] = (u[6] & 0x0F) | 0x80
u[8] = (u[8] & 0x3F) | 0x80
h = u.hex()
return "%s-%s-%s-%s-%s" % (h[:8], h[8:12], h[12:16], h[16:20], h[20:])
# ---------------------------------------------------------------- §9 verification
class Stop(Exception):
def __init__(self, at, reason):
Exception.__init__(self, reason)
self.at, self.reason = at, reason
def attrs_of(g):
return dict(LRV.read_attrs(g["attrs"]) or [])
def binds(g, principal):
"""§3.1."""
a = attrs_of(g)
return a["subject"] == principal.hex() if "subject" in a else g["principal"] == principal
def control(ev, device, command):
"""The control payload of `ev`, when it is `command` addressed to `device`."""
if ev["kind"] != CONTROL:
return None
try:
p = LEV.read_payload(CONTROL, ev["payload"], 65535)
except Refuse:
return None
return p if p["target"] == device and p["command"] == command else None
def find(d, grants, seen, gaps, m):
"""§9.2: the start, its count-in, the stop and the held commitments."""
if d["mode"] == 0:
raise Stop(["manifest"], "recording_not_permitted")
pos = next((k for k, s in enumerate(seen) if s["ev"]["digest"] == m["start"]), None)
start = None if pos is None else control(seen[pos]["ev"], m["device"], 1)
if start is None:
raise Stop(["manifest"], "take_unknown")
stop = next((s for s in seen[pos + 1:] if control(s["ev"], m["device"], 2) is not None), None)
mine = sorted((s for s in seen if s["ev"]["principal"] == m["device"] and s["ev"]["kind"] == TAKE_COMMIT),
key=lambda s: s["ev"]["seq"])
commits = []
for s in mine:
try:
c = read_commit(s["ev"]["payload"])
except Refuse as e:
raise Stop(["record", s["chain"], s["record"]], e.reason)
if c["take"] == m["take"]:
commits.append((s, c))
for s, _ in commits:
g = grants.get(s["grant"])
if g is None or not binds(g, m["principal"]):
raise Stop(["record", s["chain"], s["record"]], "device_not_bound")
device_gap = any(g["principal"] == m["device"].hex() for g in gaps)
held = [s["ev"]["digest"] for s, _ in commits]
every = {s["ev"]["digest"] for s in seen}
listed_held = [x for x in m["commits"] if x in held]
unheld = [x for x in m["commits"] if x not in held]
if listed_held != held or any(x in every or not device_gap for x in unheld):
raise Stop(["manifest"], "manifest_mismatch")
return seen[pos], start["count_in_ms"], stop, commits
def head_of(text):
return json.loads(text) if text is not None else None
def check_upload(d, head, m, receipts):
"""§6.4, §6.3: the last receipt, or Stop."""
if d["origin"] != b"":
h = head_of(head)
listed = h is not None and any(
e["key"] == m["store"].hex() and STORE_AUTHORITY in e["may_sign"] and e["status"] != "revoked" for e in h["keys"])
if not listed:
raise Stop(["store"], "store_not_listed")
prev = None
for i, b in enumerate(receipts):
at = ["receipt", i]
try:
x = read_receipt(b)
except Refuse as e:
raise Stop(at, e.reason)
fields = (x["state"] in (1, 2, 3, 4) and all(0x41 <= c <= 0x5A for c in x["region"])
and (x["held"] != 0 or x["root"] == ZERO))
if not fields or x["store"] != m["store"] or x["take"] != m["take"]:
raise Stop(at, "upload_receipt_invalid")
bad = ED.verify(x["store"], RECEIPT_DOMAIN + x["body"], x["sig"])
if bad:
raise Stop(at, bad)
if prev is None:
chained = x["seq"] == 0 and x["prev"] == ZERO and x["state"] == R_OPEN and x["held"] == 0
else:
step = {R_PROGRESS: x["held"] > prev["held"], R_RESUME: x["held"] == prev["held"],
R_SEALED: x["held"] == prev["held"]}.get(x["state"], False)
chained = (x["seq"] == i and x["prev"] == prev["digest"] and x["at"] >= prev["at"]
and x["region"] == prev["region"] and prev["state"] != R_SEALED and step)
if not chained:
raise Stop(at, "upload_receipt_invalid")
if x["held"] > len(m["chunks"]) or x["root"] != root_of(m["chunks"][:x["held"]]):
raise Stop(at, "upload_receipt_mismatch")
if d["regions"] and x["region"] not in d["regions"]:
raise Stop(at, "store_outside_region")
prev = x
return prev
def single_take(f, m, clock, parts):
"""§9.4 single-take: None when it holds, else its reason."""
start, _, stop, commits = f
interval, jitter, o = clock
held = {s["ev"]["digest"] for s, _ in commits}
if any(x not in held for x in m["commits"]):
return "delivery_gap"
cs = [c for _, c in commits]
if not cs or not is_opening(cs[0]):
return "commit_chain_broken"
for (ps, p), (_, c) in zip(commits, commits[1:]):
if p["phase"] == CLOSE or c["phase"] == OPEN or c["n"] != p["n"] + 1 or c["prev"] != ps["ev"]["digest"] or c["first"] != p["size"]:
return "commit_chain_broken"
for p, c in zip(cs, cs[1:]):
if not extends(c["first"], p["root"], c["size"], c["root"], c["proof"]):
return "root_not_extending"
last = cs[-1]
if last["phase"] != CLOSE or stop is None:
return "take_not_closed"
ch = m["chunks"]
if len(ch) != last["size"] or any(c["size"] > 0 and root_of(ch[:c["size"]]) != c["root"] for c in cs):
return "manifest_not_committed"
if any(abs(c["offset"] - o) > jitter for c in cs):
return "clock_unstable"
if any(c["start"] >= c["end"] for c in ch) or any(b["start"] != a["end"] for a, b in zip(ch, ch[1:])):
return "capture_discontinuous"
if any(c["end"] - c["start"] > interval for c in ch):
return "chunk_too_long"
added = [(s["ev"]["t"], ch[c["first"]:c["size"]]) for s, c in commits]
if (commits[0][0]["ev"]["t"] < start["ev"]["t"] or commits[-1][0]["ev"]["t"] < stop["ev"]["t"]
or any(t + jitter < x["end"] + o for t, xs in added for x in xs)):
return "commit_early"
if any(t > x["end"] + o + interval + jitter for t, xs in added for x in xs):
return "commit_late"
for index, b in parts:
if index >= len(ch):
return "chunk_extra"
if not is_chunk(ch[index], b):
return "chunk_reordered" if any(c["index"] != index and is_chunk(c, b) for c in ch) else "chunk_altered"
return None
def aligned(f, m, clock):
"""§9.4 aligned."""
start, count_in_ms, _, commits = f
_, jitter, o = clock
if not commits or commits[0][0]["ev"]["digest"] != m["commits"][0]:
return "delivery_gap"
if not is_opening(commits[0][1]):
return "commit_chain_broken"
adds = next((c for _, c in commits if c["size"] > 0), None)
if adds is None or adds["size"] > len(m["chunks"]) or root_of(m["chunks"][:adds["size"]]) != adds["root"]:
return "manifest_not_committed"
if abs(m["chunks"][0]["start"] + o - (start["ev"]["t"] + count_in_ms * 1000)) > jitter:
return "misaligned"
return None
def complete(f, m, parts, last_receipt):
"""§9.4 complete."""
commits = f[3]
s, r = (commits[-1][1]["size"], commits[-1][1]["root"]) if commits else (0, ZERO)
if s == 0 or s > len(m["chunks"]) or root_of(m["chunks"][:s]) != r:
return "manifest_not_committed"
for c in m["chunks"][:s]:
if not any(i == c["index"] and is_chunk(c, b) for i, b in parts):
return "chunk_missing"
if last_receipt is None or last_receipt["state"] != R_SEALED or last_receipt["held"] != s:
return "upload_not_sealed"
return None
def publishable(p, m, consents, room, revoked, principal_minor, start_s, at):
"""§9.4 publishable for purpose bit p."""
for subject, digest in m["depicted"]:
c = next((c for c in consents if digest != ZERO and c["digest"] == digest and c["subject"] == subject
and c["room"] in (ZERO, room)), None)
if c is None:
return "consent_missing"
if (c["minor"] == 1 or (subject == m["principal"] and principal_minor)) and c["guardian"] == ZERO:
return "guardian_missing"
if c["digest"] in revoked:
return "consent_revoked"
if not (c["nbf"] <= start_s < c["exp"]) or not (c["nbf"] <= at < c["exp"]):
return "consent_expired"
if not c["purposes"] & RECORD or not c["purposes"] & p:
return "purpose_not_consented"
return None
def position(chunks, o, t):
"""§10: (chunk, into the chunk, into the master), or None outside the master."""
dv = t - o
for c in chunks:
if c["start"] <= dv < c["end"]:
return c["index"], dv - c["start"], dv - chunks[0]["start"]
return None
def verify_take(v):
"""§9: a dict of what the corpus states for the vector."""
descriptor = bytes.fromhex(v["descriptor"])
grants = [bytes.fromhex(g) for g in v["grants"]]
head = v["key_log_head"]
chains = [[bytes.fromhex(r) for r in c] for c in v["chains"]]
room_out = LRV.verify_room(descriptor, grants, head, chains)
if room_out["outcome"] == "refuse":
return {"outcome": "refuse", "reason": room_out["reason"], "at": ["room"] + room_out["at"]}
d = LRV.read_descriptor(descriptor)
held_grants, _ = LRV.check_grants(d, grants)
by_digest = {g["digest"]: g for g in held_grants}
seen, digests = [], set()
for c, i, g in room_out["order"]:
ev = LEV.read_record(chains[c][i])
if ev["digest"] not in digests:
digests.add(ev["digest"])
seen.append({"chain": c, "record": i, "ev": ev, "grant": bytes.fromhex(g) if g else None})
try:
try:
m = read_manifest(bytes.fromhex(v["manifest"]), d["room"])
except Refuse as e:
raise Stop(["manifest"], e.reason)
f = find(d, by_digest, seen, room_out.get("gaps", []), m)
last_receipt = check_upload(d, head, m, [bytes.fromhex(r) for r in v["receipts"]])
consents = []
for k, b in enumerate(v["consents"]):
try:
consents.append(read_consent(bytes.fromhex(b)))
except Refuse as e:
raise Stop(["consent", k], e.reason)
except Stop as e:
return {"outcome": "refuse", "reason": e.reason, "at": e.at}
_, _, _, commits = f
o = commits[0][1]["offset"] if commits else 0
clock = (d["commit_interval_ms"] * 1000, d["jitter_ms"] * 1000, o)
parts = [(p["index"], bytes.fromhex(p["bytes"])) for p in v["parts"]]
revoked = set()
if d["origin"] != b"":
revoked = {bytes.fromhex(x) for x in head_of(head)["revoked_modules"]}
first_grant = by_digest.get(commits[0][0]["grant"]) if commits else None
principal_minor = first_grant is not None and attrs_of(first_grant).get("minor") == "true"
start_s = (d["t0"] + f[0]["ev"]["t"]) // 1000000
verdicts = {
"single_take": single_take(f, m, clock, parts) or "holds",
"aligned": aligned(f, m, clock) or "holds",
"complete": complete(f, m, parts, last_receipt) or "holds",
"publishable": {n: publishable(b, m, consents, d["room"], revoked, principal_minor, start_s, v["evaluated_at"]) or "holds"
for b, n in PURPOSES},
}
claimed = [(SINGLE_TAKE, "single_take"), (ALIGNED, "aligned"), (COMPLETE, "complete")]
stands = (all(not m["claims"] & b or verdicts[k] == "holds" for b, k in claimed)
and all(not m["purposes"] & b or verdicts["publishable"][n] == "holds" for b, n in PURPOSES))
markers = []
if verdicts["aligned"] == "holds":
for s in seen:
if s["ev"]["kind"] == MARKER:
at = position(m["chunks"], o, s["ev"]["t"])
if at is not None:
markers.append({"event": s["ev"]["digest"].hex(), "t": s["ev"]["t"], "chunk": at[0],
"into_chunk": at[1], "into_master": at[2]})
last = commits[-1][1] if commits else None
return {"outcome": "verdicts", "take": m["take"].hex(), "stream_id": stream_id(m["take"]),
"commits": [s["ev"]["digest"].hex() for s, _ in commits],
"size": last["size"] if last else 0, "root": (last["root"] if last else ZERO).hex(),
"verdicts": verdicts, "stands": stands, "markers": markers}
# ---------------------------------------------------------------- the corpus
def problems_of(v):
"""What differs between the vector's statement and this verifier's outcome."""
try:
got = verify_take(v)
except Exception as e: # a crash on a mutated copy is a failure to read it
return ["the vector does not verify: %s: %s" % (type(e).__name__, e)], None
out = [k for k in got if v.get(k) != got[k]]
out += [k for k in ("outcome", "reason", "at") if k in v and k not in got]
return out, got
def check(doc, quiet=False):
"""Every vector against its statement: the failures."""
fails = []
for v in doc["vectors"]:
problems, got = problems_of(v)
if not quiet:
what = "?" if got is None else got["outcome"] + (" " + got["reason"] if got["outcome"] == "refuse" else "")
print(("PASS" if not problems else "FAIL") + " vector %s: %s" % (v["name"], what))
for k in problems:
fails.append("%s: %s is %s, the corpus states %s" % (
v["name"], k, json.dumps(got.get(k))[:300] if got else k, json.dumps(v.get(k))[:300]))
return fails
def flip(h, at):
"""Hexadecimal h with the byte at `at` (from the end when negative) changed."""
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."""
names = {v["name"]: k for k, v in enumerate(doc["vectors"])}
lesson = names["lesson"]
def mutate(label, f):
d = copy.deepcopy(doc)
f(d["vectors"])
return label, d
def device_chain(vs, k):
dev = doc["principals"]["device"]
return next(c for c in vs[k]["chains"] if LEV.read_record(bytes.fromhex(c[0]))["principal"].hex() == dev)
out = [
mutate("a byte of the lesson's chunk 2 part", lambda vs: vs[lesson]["parts"][2].update(bytes=flip(vs[lesson]["parts"][2]["bytes"], 5))),
mutate("the lesson's parts 0 and 1 swapped", lambda vs: vs[lesson]["parts"][0].update(bytes=vs[lesson]["parts"][1]["bytes"])),
mutate("the lesson's last part dropped", lambda vs: vs[lesson]["parts"].pop()),
mutate("a byte of the lesson's manifest signature", lambda vs: vs[lesson].update(manifest=flip(vs[lesson]["manifest"], -10))),
# Chunk 0's descriptor begins at byte 203 of a manifest (§7.1); byte 8 of it is `start`'s.
mutate("a byte of the lesson's chunk 0 descriptor", lambda vs: vs[lesson].update(manifest=flip(vs[lesson]["manifest"], 203 + 8))),
mutate("the lesson's sealed receipt removed", lambda vs: vs[lesson]["receipts"].pop()),
mutate("a byte of the lesson's consent", lambda vs: vs[lesson]["consents"].__setitem__(0, flip(vs[lesson]["consents"][0], 70))),
mutate("the lesson's second commitment not delivered", lambda vs: device_chain(vs, lesson).pop(1)),
mutate("the lesson evaluated a year on", lambda vs: vs[lesson].update(evaluated_at=vs[lesson]["evaluated_at"] + 365 * 86400)),
mutate("the lesson's stated root", lambda vs: vs[lesson].update(root="00" * 32)),
mutate("the lesson's stated stream id", lambda vs: vs[lesson].update(stream_id="0" + vs[lesson]["stream_id"][1:])),
mutate("the lesson's stated marker position", lambda vs: vs[lesson]["markers"][0].update(into_master=vs[lesson]["markers"][0]["into_master"] + 1)),
mutate("the missing chunk stated complete", lambda vs: vs[names["missing_chunk"]]["verdicts"].update(complete="holds")),
mutate("the late commitment stated single", lambda vs: vs[names["late_commit"]]["verdicts"].update(single_take="holds")),
mutate("the misaligned take stated aligned", lambda vs: vs[names["misaligned"]]["verdicts"].update(aligned="holds")),
mutate("the revoked consent stated publishable", lambda vs: vs[names["consent_revoked"]]["verdicts"]["publishable"].update(record="holds")),
mutate("the impostor's refusal at the manifest", lambda vs: vs[names["device_not_bound"]].update(at=["manifest"])),
mutate("the store outside its region stated listed", lambda vs: vs[names["store_outside_region"]].update(reason="store_not_listed")),
]
return out
def main(corpus_dir):
fails = []
st = ED._self_test()
print(("PASS" if not st else "FAIL") + " Ed25519 strict rule self-test " + ",".join(st))
if st:
fails.append("ed25519 self-test")
with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as f:
doc = json.load(f)
fails += check(doc)
seen = set()
for v in doc["vectors"]:
if v.get("reason"):
seen.add(v["reason"])
if "verdicts" in v:
vd = v["verdicts"]
seen |= {vd["single_take"], vd["aligned"], vd["complete"]} | set(vd["publishable"].values())
missing = [r for r in REASONS + ["delivery_gap"] if r not in seen]
if missing:
fails.append("no vector gives %s" % ", ".join(missing))
takes = ["lesson", "resumed_upload", "minor_guardian"]
absent = [t for t in takes if not any(v["name"] == t and v.get("stands") for v in doc["vectors"])]
if absent:
fails.append("good takes that do not stand: %s" % ", ".join(absent))
print("%s every reason of §12 appears, and every good take of §13 stands" % ("PASS" if not (missing or absent) else "FAIL"))
for label, mutated in mutations(doc):
changed = [v for v, o in zip(mutated["vectors"], doc["vectors"]) if v != o]
caught = check({"vectors": changed}, quiet=True) if changed else []
print(("PASS" if caught else "FAIL") + " mutated copy fails: %s%s" % (label, " (%s)" % caught[0][:120] 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") + " capture-binding corpus: %d vectors" % len(doc["vectors"]))
return 1 if fails else 0
if __name__ == "__main__":
if len(sys.argv) != 2:
print("usage: capture_take_verify.py <live-conformance/takes>", file=sys.stderr)
sys.exit(2)
sys.exit(main(sys.argv[1]))
Reference in a new issue View git blame Copy permalink