57e426c940
The contribution link record and the live-recording clip record both began WCL1. The clip record had it first (live-recording §5.1, the registry's only WCL1 row); the link record took the same four bytes when it was added, so a reader that tells records apart by their magic could not tell one from the other. The link record is now WLK1, a magic nothing else in the registry, the code or an open branch uses. live-contribution §6.2 and SPEC name it, §6.5 step 2 says the clip record's WCL1 is refused as malformed, and codecs/registry.rs registers it (REGISTRY.md and registry/ exported from it). joule-moq-ingest writes and reads it. The corpus live-conformance/links/ is regenerated with it, its files now .wlk1, plus clip_record_magic, a part with the clip record's magic, refused. live_contribution_verify.py reads WLK1 and also refuses each accepted part mutated to WCL1. A wire change: a link record written before this one is malformed to a reader of this revision. The CHANGELOG says so and acknowledges every corpus file that changed. Co-Authored-By: Claude Opus 5.5 <[email protected]>
660 lines
30 KiB
Python
660 lines
30 KiB
Python
#!/usr/bin/env python3
|
|
"""Independent verifier of the contribution link-record corpus (extensions/live-contribution.md
|
|
§6, the record `WLK1`, and §7's refusals of it).
|
|
|
|
Written from extensions/live-contribution.md alone. It shares no code with the reference
|
|
implementation (joule-moq-ingest), which generates the corpus, and never runs it. It checks
|
|
every Ed25519 signature with ../claim-conformance/_ed25519.py, WAI's strict rule (SPEC §7
|
|
*Signatures*) written from RFC 8032 with the standard library. Everything else is here, from
|
|
the text, with Python's integers:
|
|
|
|
- a part (§6.5), in the stated order: the size bound (§6.7); the reading of §6.2's layout,
|
|
field by field, with its reasons; the signature by `hop` over `wai:contrib-link` 0x01 and
|
|
the bytes before `sig`; the link rules; the account rules (§6.3); the loss runs;
|
|
- a series (§6.6): each part verified, then chained to the one before (`chain_broken`) and
|
|
continuing its sequence numbers and its span (`sequence_gap`).
|
|
|
|
For every vector it requires the outcome and the reason the corpus states, and for a series
|
|
the index of the part refused; for an accepted part its digest and every summary field the
|
|
corpus states, each recomputed from the record's bytes; for an accepted series each part's
|
|
digest and summary. Of the corpus, it requires that every reason of §7's last list refuses a
|
|
vector and that the index lists exactly those reasons. Then it runs the Ed25519 helper's own
|
|
self-test; mutates accepted records and series in memory, one defect each, and requires every
|
|
mutated copy to be refused; and leaves out its own checks one at a time and requires the
|
|
corpus to catch the absence of each one it lists as required, noting the others it does not.
|
|
|
|
python3 -I live_contribution_verify.py <live-conformance/links>
|
|
|
|
Exit status 0 only if every vector verifies as stated, every mutated copy is refused and the
|
|
corpus catches every required dropped check; 1 otherwise; 2 for a wrong command line.
|
|
Python 3.8 or later; standard library only.
|
|
"""
|
|
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 _ed25519 as ED # noqa: E402
|
|
|
|
FORMAT = "wai-live-contribution-links/1"
|
|
MAGIC = b"WLK1"
|
|
CLIP_MAGIC = b"WCL1" # live-recording §5.1: the clip record, another record's magic
|
|
VERSION = 1
|
|
DOMAIN = b"wai:contrib-link\x01" # §6.5: the 17 bytes the signature covers before the record
|
|
FIXED = 415 # §6.7: the fixed fields and the signature
|
|
TRACK = 80 # §6.2: digest 32, track_alias 8, groups 8, head 32
|
|
RUN = 20 # §6.2: first 8, count 4, recovered 4, dropped 4
|
|
MAX_TRACKS = 8
|
|
MAX_RUNS = 512
|
|
MAX_SIZE = FIXED + TRACK * MAX_TRACKS + RUN * MAX_RUNS # §6.7: 11 295
|
|
SIG = 64
|
|
ZERO = bytes(32)
|
|
# §6.5 step 4: the key lengths each protocol allows, in bits.
|
|
CIPHERS = {1: (0, 128, 192, 256), 2: (0,), 3: (0, 128, 256)}
|
|
ENDED_MAX = 7
|
|
|
|
# §7's last list, in its order: the reasons this verifier raises.
|
|
REASONS = [
|
|
"record_too_large", "truncated", "malformed", "unknown_version", "bad_key", "bad_signature",
|
|
"link_invalid", "account_invalid", "runs_invalid", "chain_broken", "sequence_gap",
|
|
]
|
|
|
|
# Every check this verifier makes, by name. A self-test run may leave one out to show the
|
|
# corpus catches its absence; a verification never does.
|
|
CHECKS = (
|
|
# §6.5 step 1
|
|
"size-bound",
|
|
# step 2
|
|
"magic", "version", "last-flag", "tracks-bound", "runs-bound", "truncated-flag", "trailing-byte",
|
|
# step 3
|
|
"signature", "small-order-key", "small-order-r",
|
|
# step 4
|
|
"protocol", "mode", "cipher", "last-ended", "ended-max", "span", "window", "parent-zero",
|
|
"track-zero-digest", "track-head-groups", "tracks-order",
|
|
# step 5
|
|
"first-try", "account-identity", "outstanding-last", "recovery-samples", "rtt-empty", "rtt-order",
|
|
"recovery-empty", "recovery-order", "margin-recovered", "margin-window", "requested", "bytes-delivered",
|
|
# step 6
|
|
"truncated-full", "run-count", "run-overfull", "run-within", "run-order", "runs-sums",
|
|
# §6.6
|
|
"series-first-part", "series-same-room", "series-same-hop", "series-same-principal", "series-same-grant",
|
|
"series-same-session", "series-same-protocol", "series-same-mode", "series-same-cipher",
|
|
"series-same-window", "series-after-last", "series-part-index", "series-parent", "series-track-kept",
|
|
"series-track-alias", "series-track-advance", "series-time", "series-seq",
|
|
)
|
|
|
|
# The checks whose absence the corpus must catch: with any one of them left out, at least one
|
|
# vector must come out other than the corpus states. Every other check is left out too, and
|
|
# those the corpus does not catch are listed in a NOTE line. `first-try` can never be caught
|
|
# alone: `recovery.samples` is unsigned, so `recovery.samples = delivered - first_try`
|
|
# already holds `first_try` to at most `delivered`, with the same reason.
|
|
REQUIRED_CAUGHT = (
|
|
"size-bound", "magic", "version", "trailing-byte", "signature", "small-order-key",
|
|
"protocol", "mode", "cipher", "last-ended", "ended-max", "span", "window", "parent-zero",
|
|
"track-zero-digest", "track-head-groups", "tracks-order",
|
|
"account-identity", "outstanding-last", "recovery-samples", "rtt-empty", "rtt-order",
|
|
"margin-window", "requested",
|
|
"truncated-full", "run-count", "run-within", "run-order", "runs-sums",
|
|
"series-first-part", "series-same-session", "series-after-last", "series-part-index", "series-parent",
|
|
"series-track-kept", "series-track-advance", "series-time", "series-seq",
|
|
)
|
|
|
|
FIXED_U64 = (
|
|
"start", "end", "window", "seq_first", "seq_count", "delivered", "first_try", "dropped", "outstanding",
|
|
"late", "duplicates", "bytes", "requests", "requested", "retransmitted",
|
|
)
|
|
|
|
|
|
class Refuse(Exception):
|
|
def __init__(self, reason, check):
|
|
super().__init__(reason)
|
|
self.reason, self.check = reason, check
|
|
|
|
|
|
def need(ok, reason, check, skip):
|
|
"""Refuse with `reason` unless `ok`, or unless a self-test run left `check` out."""
|
|
assert check in CHECKS, check
|
|
if not ok and check not in skip:
|
|
raise Refuse(reason, check)
|
|
|
|
|
|
class Cursor:
|
|
"""§6.5 step 2: a record read field by field; running out is `truncated`."""
|
|
|
|
def __init__(self, b):
|
|
self.b, self.p = b, 0
|
|
|
|
def take(self, n):
|
|
if self.p + n > len(self.b):
|
|
raise Refuse("truncated", "read")
|
|
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 read_part(b, skip):
|
|
"""§6.5 step 2: the fields of §6.2 in their order, to exactly the last byte."""
|
|
c = Cursor(b)
|
|
rec = {}
|
|
need(c.take(4) == MAGIC, "malformed", "magic", skip)
|
|
rec["version"] = c.uint(1)
|
|
need(rec["version"] == VERSION, "unknown_version", "version", skip)
|
|
for name, n in (("room", 32), ("hop", 32), ("principal", 32), ("grant", 32), ("session", 16)):
|
|
rec[name] = c.take(n)
|
|
rec["part"] = c.uint(4)
|
|
rec["parent"] = c.take(32)
|
|
rec["protocol"] = c.uint(1)
|
|
rec["mode"] = c.uint(1)
|
|
rec["cipher"] = c.uint(2)
|
|
rec["last"] = c.uint(1)
|
|
need(rec["last"] in (0, 1), "malformed", "last-flag", skip)
|
|
rec["ended"] = c.uint(1)
|
|
for name in FIXED_U64:
|
|
rec[name] = c.uint(8)
|
|
rec["rtt"] = tuple(c.uint(4) for _ in range(4))
|
|
rec["recovery"] = tuple(c.uint(4) for _ in range(4))
|
|
rec["margin_min"] = c.uint(4)
|
|
n = c.uint(1)
|
|
need(n <= MAX_TRACKS, "malformed", "tracks-bound", skip)
|
|
rec["tracks"] = []
|
|
for _ in range(n):
|
|
rec["tracks"].append({"digest": c.take(32), "alias": c.uint(8), "groups": c.uint(8), "head": c.take(32)})
|
|
k = c.uint(2)
|
|
need(k <= MAX_RUNS, "malformed", "runs-bound", skip)
|
|
rec["truncated"] = c.uint(1)
|
|
need(rec["truncated"] in (0, 1), "malformed", "truncated-flag", skip)
|
|
rec["runs"] = []
|
|
for _ in range(k):
|
|
rec["runs"].append({"first": c.uint(8), "count": c.uint(4), "recovered": c.uint(4), "dropped": c.uint(4)})
|
|
rec["signed"] = b[:c.p]
|
|
rec["sig"] = c.take(SIG)
|
|
need(c.p == len(b), "malformed", "trailing-byte", skip)
|
|
rec["digest"] = hashlib.sha256(rec["signed"]).digest()
|
|
return rec
|
|
|
|
|
|
_SIGNATURES = {}
|
|
|
|
|
|
def signature_verdict(pk, msg, sig, ed_skip):
|
|
"""The shared strict rule, remembered so repeated runs over the corpus verify each signature once."""
|
|
key = (pk, msg, sig, ed_skip)
|
|
if key not in _SIGNATURES:
|
|
_SIGNATURES[key] = ED.verify(pk, msg, sig, skip=ed_skip)
|
|
return _SIGNATURES[key]
|
|
|
|
|
|
def verify_part(b, skip=frozenset()):
|
|
"""§6.5: the steps in order, refusing at the first that fails; the record read."""
|
|
# 1. The size bound, before anything is read.
|
|
need(len(b) <= MAX_SIZE, "record_too_large", "size-bound", skip)
|
|
# 2. The reading.
|
|
rec = read_part(b, skip)
|
|
# 3. The signature, under `hop`, strictly.
|
|
if "signature" not in skip:
|
|
ed_skip = tuple(s for s in ED.SKIPPABLE if s in skip)
|
|
bad = signature_verdict(rec["hop"], DOMAIN + rec["signed"], rec["sig"], ed_skip)
|
|
if bad is not None:
|
|
raise Refuse(bad, "signature")
|
|
# 4. The link.
|
|
L = "link_invalid"
|
|
p = rec["protocol"]
|
|
need(p in (1, 2, 3), L, "protocol", skip)
|
|
need(rec["mode"] in (1, 2), L, "mode", skip)
|
|
if p in CIPHERS:
|
|
need(rec["cipher"] in CIPHERS[p], L, "cipher", skip)
|
|
need((rec["last"] == 1) == (rec["ended"] != 0), L, "last-ended", skip)
|
|
need(rec["ended"] <= ENDED_MAX, L, "ended-max", skip)
|
|
need(rec["start"] < rec["end"], L, "span", skip)
|
|
need(rec["window"] > 0, L, "window", skip)
|
|
need((rec["part"] == 0) == (rec["parent"] == ZERO), L, "parent-zero", skip)
|
|
prev = None
|
|
for t in rec["tracks"]:
|
|
need(t["digest"] != ZERO, L, "track-zero-digest", skip)
|
|
need((t["head"] == ZERO) == (t["groups"] == 0), L, "track-head-groups", skip)
|
|
if prev is not None: # a digest is a byte string: ordered byte by byte, the first byte first
|
|
need(prev < t["digest"], L, "tracks-order", skip)
|
|
prev = t["digest"]
|
|
# 5. The account.
|
|
A = "account_invalid"
|
|
delivered, first_try = rec["delivered"], rec["first_try"]
|
|
recovered = delivered - first_try
|
|
need(first_try <= delivered, A, "first-try", skip)
|
|
need(delivered + rec["dropped"] + rec["outstanding"] == rec["seq_count"], A, "account-identity", skip)
|
|
need(rec["outstanding"] == 0 or rec["last"] == 1, A, "outstanding-last", skip)
|
|
need(rec["recovery"][0] == recovered, A, "recovery-samples", skip)
|
|
for name in ("rtt", "recovery"): # with no sample all four are 0; with samples, min <= median <= max
|
|
samples, lo, mid, hi = rec[name]
|
|
if samples == 0:
|
|
need((lo, mid, hi) == (0, 0, 0), A, name + "-empty", skip)
|
|
else:
|
|
need(lo <= mid <= hi, A, name + "-order", skip)
|
|
need(recovered != 0 or rec["margin_min"] == 0, A, "margin-recovered", skip)
|
|
need(rec["margin_min"] <= rec["window"], A, "margin-window", skip)
|
|
need(rec["requested"] >= rec["requests"], A, "requested", skip)
|
|
need(delivered != 0 or rec["bytes"] == 0, A, "bytes-delivered", skip)
|
|
# 6. The runs.
|
|
R = "runs_invalid"
|
|
runs, truncated = rec["runs"], rec["truncated"] == 1
|
|
if truncated:
|
|
need(len(runs) == MAX_RUNS, R, "truncated-full", skip)
|
|
lo, hi = rec["seq_first"], rec["seq_first"] + rec["seq_count"]
|
|
end = None
|
|
for r in runs:
|
|
need(r["count"] > 0, R, "run-count", skip)
|
|
need(r["recovered"] + r["dropped"] <= r["count"], R, "run-overfull", skip)
|
|
need(lo <= r["first"] and r["first"] + r["count"] <= hi, R, "run-within", skip)
|
|
if end is not None:
|
|
need(r["first"] >= end + 1, R, "run-order", skip)
|
|
end = r["first"] + r["count"]
|
|
sums = (
|
|
(sum(r["count"] for r in runs), rec["seq_count"] - first_try),
|
|
(sum(r["recovered"] for r in runs), recovered),
|
|
(sum(r["dropped"] for r in runs), rec["dropped"]),
|
|
(sum(r["count"] - r["recovered"] - r["dropped"] for r in runs), rec["outstanding"]),
|
|
)
|
|
need(all(got <= of if truncated else got == of for got, of in sums), R, "runs-sums", skip)
|
|
return rec
|
|
|
|
|
|
SERIES_SAME = ("room", "hop", "principal", "grant", "session", "protocol", "mode", "cipher", "window")
|
|
|
|
|
|
def chain(prev, rec, skip):
|
|
"""§6.6 steps 1 and 2: a part after the first, against the part before it."""
|
|
C = "chain_broken"
|
|
for f in SERIES_SAME:
|
|
need(rec[f] == prev[f], C, "series-same-" + f, skip)
|
|
need(prev["last"] == 0, C, "series-after-last", skip)
|
|
need(rec["part"] == prev["part"] + 1, C, "series-part-index", skip)
|
|
need(rec["parent"] == prev["digest"], C, "series-parent", skip)
|
|
named = {t["digest"]: t for t in rec["tracks"]}
|
|
for t in prev["tracks"]:
|
|
again = named.get(t["digest"])
|
|
need(again is not None, C, "series-track-kept", skip)
|
|
if again is None:
|
|
continue
|
|
need(again["alias"] == t["alias"], C, "series-track-alias", skip)
|
|
need(again["groups"] > t["groups"] or (again["groups"] == t["groups"] and again["head"] == t["head"]),
|
|
C, "series-track-advance", skip)
|
|
G = "sequence_gap"
|
|
need(rec["start"] == prev["end"], G, "series-time", skip)
|
|
need(rec["seq_first"] == prev["seq_first"] + prev["seq_count"], G, "series-seq", skip)
|
|
|
|
|
|
def verify_series(parts, skip=frozenset()):
|
|
"""§6.6: the records in order, each verified, then chained to the one before; the records
|
|
read, or a Refuse with `at` set to the index of the part refused."""
|
|
out, prev = [], None
|
|
for i, b in enumerate(parts):
|
|
try:
|
|
rec = verify_part(b, skip)
|
|
if prev is None:
|
|
need(rec["part"] == 0, "chain_broken", "series-first-part", skip)
|
|
else:
|
|
chain(prev, rec, skip)
|
|
except Refuse as e:
|
|
e.at = i
|
|
raise
|
|
out.append(rec)
|
|
prev = rec
|
|
return out
|
|
|
|
|
|
def summary(rec, b):
|
|
"""The figures a corpus may state of an accepted part, each from the record's bytes."""
|
|
return {
|
|
"part": rec["part"], "protocol": rec["protocol"], "mode": rec["mode"], "cipher": rec["cipher"],
|
|
"last": rec["last"] == 1, "ended": rec["ended"], "start": rec["start"], "end": rec["end"],
|
|
"window": rec["window"], "seq_first": rec["seq_first"], "seq_count": rec["seq_count"],
|
|
"delivered": rec["delivered"], "first_try": rec["first_try"],
|
|
"recovered": rec["delivered"] - rec["first_try"], "dropped": rec["dropped"],
|
|
"outstanding": rec["outstanding"], "late": rec["late"], "duplicates": rec["duplicates"],
|
|
"bytes": rec["bytes"], "requests": rec["requests"], "requested": rec["requested"],
|
|
"retransmitted": rec["retransmitted"], "margin_min": rec["margin_min"],
|
|
"tracks": len(rec["tracks"]), "runs": len(rec["runs"]), "truncated": rec["truncated"] == 1,
|
|
"size": len(b),
|
|
}
|
|
|
|
|
|
def outcome(kind, parts, skip=frozenset()):
|
|
"""What this verifier makes of a vector's records: the outcome and its particulars."""
|
|
if kind == "part":
|
|
if len(parts) != 1:
|
|
raise ValueError("a part vector names %d files" % len(parts))
|
|
try:
|
|
rec = verify_part(parts[0], skip)
|
|
except Refuse as e:
|
|
return {"outcome": "refuse", "reason": e.reason, "check": e.check}
|
|
return {"outcome": "accept", "digest": rec["digest"].hex(), "summary": summary(rec, parts[0]), "records": [rec]}
|
|
if kind == "series":
|
|
try:
|
|
recs = verify_series(parts, skip)
|
|
except Refuse as e:
|
|
return {"outcome": "refuse", "reason": e.reason, "at": e.at, "check": e.check}
|
|
return {"outcome": "accept", "digests": [r["digest"].hex() for r in recs],
|
|
"summaries": [summary(r, b) for r, b in zip(recs, parts)], "records": recs}
|
|
raise ValueError("kind %r" % kind)
|
|
|
|
|
|
def same(a, b):
|
|
"""Equal and of the same JSON type: a stated `true` is not a computed 1."""
|
|
return type(a) is type(b) and a == b
|
|
|
|
|
|
def compare_summary(stated, got, where):
|
|
problems = []
|
|
if not isinstance(stated, dict):
|
|
return ["%s: the summary is not an object" % where]
|
|
for k, v in sorted(stated.items()):
|
|
if k not in got:
|
|
problems.append("%s: the corpus states %s, which this verifier does not compute" % (where, k))
|
|
elif not same(v, got[k]):
|
|
problems.append("%s: %s is %s, the corpus states %s" % (where, k, json.dumps(got[k]), json.dumps(v)))
|
|
return problems
|
|
|
|
|
|
def compare(v, got):
|
|
"""The ways `got` differs from what vector `v` states."""
|
|
problems = []
|
|
if v.get("outcome") != got["outcome"]:
|
|
what = got["outcome"] + (" " + got["reason"] if got["outcome"] == "refuse" else "")
|
|
return ["outcome is %s (%s), the corpus states %s %s" % (what, got.get("check", ""), v.get("outcome"), v.get("reason", ""))]
|
|
if got["outcome"] == "refuse":
|
|
if v.get("reason") != got["reason"]:
|
|
problems.append("reason is %s (check %s), the corpus states %s" % (got["reason"], got["check"], v.get("reason")))
|
|
if v["kind"] == "series" and v.get("at") != got["at"]:
|
|
problems.append("at is %s, the corpus states %s" % (got["at"], v.get("at")))
|
|
if v["kind"] == "part" and "at" in v:
|
|
problems.append("a part vector states `at`")
|
|
return problems
|
|
if v["kind"] == "part":
|
|
if "digest" not in v or "summary" not in v:
|
|
problems.append("an accepted part states no digest or no summary")
|
|
if "digest" in v and v["digest"] != got["digest"]:
|
|
problems.append("digest is %s, the corpus states %s" % (got["digest"], v["digest"]))
|
|
if "summary" in v:
|
|
problems += compare_summary(v["summary"], got["summary"], "summary")
|
|
else:
|
|
if "digests" not in v or "summaries" not in v:
|
|
problems.append("an accepted series states no digests or no summaries")
|
|
if "digests" in v and v["digests"] != got["digests"]:
|
|
problems.append("digests are %s, the corpus states %s" % (got["digests"], v["digests"]))
|
|
if "summaries" in v:
|
|
if len(v["summaries"]) != len(got["summaries"]):
|
|
problems.append("the corpus states %d summaries for %d parts" % (len(v["summaries"]), len(got["summaries"])))
|
|
for i, (s, g) in enumerate(zip(v["summaries"], got["summaries"])):
|
|
problems += compare_summary(s, g, "summaries[%d]" % i)
|
|
return problems
|
|
|
|
|
|
def load(corpus_dir, v):
|
|
if v.get("kind") not in ("part", "series") or not v.get("files"):
|
|
raise ValueError("vector %s: kind %r with files %r" % (v.get("name"), v.get("kind"), v.get("files")))
|
|
parts = []
|
|
for f in v["files"]:
|
|
if os.path.basename(f) != f:
|
|
raise ValueError("vector %s: file %r outside the corpus directory" % (v["name"], f))
|
|
with open(os.path.join(corpus_dir, f), "rb") as fh:
|
|
parts.append(fh.read())
|
|
return parts
|
|
|
|
|
|
def stated(v):
|
|
return (v.get("outcome"), v.get("reason"), v.get("at"))
|
|
|
|
|
|
def result(got):
|
|
return (got["outcome"], got.get("reason"), got.get("at"))
|
|
|
|
|
|
def check_vectors(doc, loaded):
|
|
fails = []
|
|
for v in doc["vectors"]:
|
|
try:
|
|
got = outcome(v["kind"], loaded[v["name"]])
|
|
problems = compare(v, got)
|
|
except Exception as e: # a crash is a failure to verify the vector
|
|
got, problems = None, ["the vector does not verify: %s: %s" % (type(e).__name__, e)]
|
|
if got is None:
|
|
what = "?"
|
|
elif got["outcome"] == "refuse":
|
|
what = "refuse %s" % got["reason"] + (" at %d" % got["at"] if "at" in got else "")
|
|
else:
|
|
what = "accept"
|
|
print(("PASS" if not problems else "FAIL") + " vector %s: %s" % (v["name"], what))
|
|
fails += ["vector %s: %s" % (v["name"], p) for p in problems]
|
|
if got is not None and got["outcome"] == "accept" and "hop" in doc:
|
|
for rec in got["records"]:
|
|
if rec["hop"].hex() != doc["hop"]:
|
|
fails.append("vector %s: signed by %s, not the corpus's hop key" % (v["name"], rec["hop"].hex()))
|
|
return fails
|
|
|
|
|
|
def reasons_listed():
|
|
"""The reasons of §7's last list, in its order."""
|
|
with open(os.path.join(HERE, "..", "extensions", "live-contribution.md"), encoding="utf-8") as f:
|
|
text = f.read()
|
|
at = text.index("\n## 7. Refusals")
|
|
end = text.find("\n## ", at + 1)
|
|
section = text[at:end if end != -1 else len(text)].strip()
|
|
last = re.split(r"\n\s*\n", section)[-1]
|
|
if not last.startswith("Of the link record"):
|
|
raise ValueError("§7 does not end with the link record's list")
|
|
return re.findall(r"`([a-z_]+)`", last)
|
|
|
|
|
|
# ---------------------------------------------------------------------------------------------
|
|
# The mutation self-test: copies of accepted records and series, one defect each, in memory.
|
|
|
|
FIELD_AT = {} # §6.2: the offset and size of each fixed field
|
|
|
|
|
|
def _offsets():
|
|
at = 0
|
|
for name, n in (("magic", 4), ("version", 1), ("room", 32), ("hop", 32), ("principal", 32), ("grant", 32),
|
|
("session", 16), ("part", 4), ("parent", 32), ("protocol", 1), ("mode", 1), ("cipher", 2),
|
|
("last", 1), ("ended", 1)) + tuple((f, 8) for f in FIXED_U64) + (
|
|
("rtt", 16), ("recovery", 16), ("margin_min", 4), ("tracks", 1)):
|
|
FIELD_AT[name] = (at, n)
|
|
at += n
|
|
|
|
|
|
_offsets()
|
|
NUMERIC = ("part", "protocol", "mode", "cipher", "last", "ended") + FIXED_U64 + ("rtt", "recovery", "margin_min", "tracks")
|
|
IDENTITY = ("room", "principal", "grant", "session", "parent")
|
|
|
|
|
|
def flipped(b, at, bit=0x01):
|
|
m = bytearray(b)
|
|
m[at] ^= bit
|
|
return bytes(m)
|
|
|
|
|
|
def mutations(records, series):
|
|
"""(kind, [(label, kind of input, the records)]) for every defect; each must be refused."""
|
|
kinds = []
|
|
|
|
def each(label, f):
|
|
out = []
|
|
for name, b in records:
|
|
for tag, m in f(b):
|
|
out.append(("%s %s" % (name, tag), "part", [m]))
|
|
kinds.append((label, out))
|
|
|
|
def runs_at(b):
|
|
n = b[FIELD_AT["tracks"][0]]
|
|
return FIELD_AT["tracks"][0] + 1 + TRACK * n
|
|
|
|
each("a byte of the signature flipped", lambda b: [("sig[%d]" % i, flipped(b, len(b) - SIG + i)) for i in (0, 31, 32, 63)])
|
|
each("a byte of each fixed numeric field flipped",
|
|
lambda b: [("%s+%d" % (f, j), flipped(b, FIELD_AT[f][0] + j)) for f in NUMERIC for j in (0, FIELD_AT[f][1] - 1)])
|
|
each("a byte of each fixed identity field flipped",
|
|
lambda b: [("%s" % f, flipped(b, FIELD_AT[f][0] + 7)) for f in IDENTITY])
|
|
each("a byte of each track head flipped",
|
|
lambda b: [("track %d %s" % (i, part), flipped(b, FIELD_AT["tracks"][0] + 1 + TRACK * i + off))
|
|
for i in range(b[FIELD_AT["tracks"][0]]) for part, off in (("digest", 0), ("alias", 32), ("groups", 40), ("head", 79))])
|
|
each("a byte of the runs flipped",
|
|
lambda b: [("runs count", flipped(b, runs_at(b))), ("truncated", flipped(b, runs_at(b) + 2))]
|
|
+ [("run %d %s" % (i, part), flipped(b, runs_at(b) + 3 + RUN * i + off))
|
|
for i in range(min(3, b[runs_at(b)] | b[runs_at(b) + 1] << 8)) for part, off in (("first", 0), ("count", 8), ("recovered", 12), ("dropped", 16))])
|
|
each("truncated at every length", lambda b: [("to %d" % n, b[:n]) for n in range(len(b))])
|
|
each("a byte appended", lambda b: [("0x00", b + b"\x00"), ("0xff", b + b"\xff")])
|
|
each("the magic changed", lambda b: [("byte %d" % i, flipped(b, i, 0x20)) for i in range(4)] + [("WLK2", b"WLK2" + b[4:]), ("the clip record's", CLIP_MAGIC + b[4:])])
|
|
each("the version changed", lambda b: [("to %d" % n, b[:4] + bytes([n]) + b[5:]) for n in (0, 2, 255)])
|
|
each("a byte of the hop key changed", lambda b: [("hop[%d]" % i, flipped(b, FIELD_AT["hop"][0] + i, bit)) for i, bit in ((0, 1), (15, 0x10), (31, 0x80))])
|
|
each("the record padded past its bound", lambda b: [("to %d" % (MAX_SIZE + 1), b + bytes(MAX_SIZE + 1 - len(b)))])
|
|
|
|
def each_series(label, f):
|
|
out = []
|
|
for name, parts in series:
|
|
for tag, m in f(parts):
|
|
out.append(("%s %s" % (name, tag), "series", m))
|
|
kinds.append((label, out))
|
|
|
|
def swaps(p):
|
|
return [("%d<->%d" % (i, j), [p[j] if k == i else p[i] if k == j else x for k, x in enumerate(p)])
|
|
for i in range(len(p)) for j in range(i + 1, len(p))]
|
|
|
|
def others(p):
|
|
out = []
|
|
for i in range(len(p)):
|
|
for name, b in records:
|
|
if b not in p:
|
|
out.append(("[%d]=%s" % (i, name), p[:i] + [b] + p[i + 1:]))
|
|
return out
|
|
|
|
each_series("two parts of a series swapped", swaps)
|
|
each_series("the middle part of a series dropped", lambda p: [("[%d] dropped" % i, p[:i] + p[i + 1:]) for i in range(1, len(p) - 1)])
|
|
each_series("the first part of a series dropped", lambda p: [("[0] dropped", p[1:])])
|
|
each_series("a part of a series repeated", lambda p: [("[%d] twice" % i, p[:i + 1] + p[i:]) for i in range(len(p))])
|
|
each_series("a series reversed", lambda p: [("reversed", p[::-1])])
|
|
each_series("a part of a series replaced by a part of another accepted vector", others)
|
|
each_series("a byte of a later part's parent flipped",
|
|
lambda p: [("[%d] parent" % i, p[:i] + [flipped(p[i], FIELD_AT["parent"][0])] + p[i + 1:]) for i in range(1, len(p))])
|
|
return kinds
|
|
|
|
|
|
def run_mutations(records, series):
|
|
fails = []
|
|
for label, copies in mutations(records, series):
|
|
bad, reasons = [], set()
|
|
for tag, kind, parts in copies:
|
|
try:
|
|
got = outcome(kind, parts)
|
|
except Exception as e: # a crash is not a refusal
|
|
bad.append("%s crashed: %s: %s" % (tag, type(e).__name__, e))
|
|
continue
|
|
if got["outcome"] != "refuse":
|
|
bad.append("%s accepted" % tag)
|
|
else:
|
|
reasons.add(got["reason"])
|
|
ok = copies and not bad
|
|
print(("PASS" if ok else "FAIL") + " mutated copies refused: %s (%s)" % (label, ", ".join(sorted(reasons)) or "none made"))
|
|
if not copies:
|
|
fails.append("mutation %s: no copy made" % label)
|
|
fails += ["mutation %s: %s" % (label, b) for b in bad]
|
|
return fails
|
|
|
|
|
|
def run_dropped_checks(doc, loaded):
|
|
"""Each check left out in turn: for a required one, the corpus must then come out otherwise
|
|
somewhere; the others it does not catch are noted."""
|
|
fails, uncaught = [], []
|
|
for check in CHECKS:
|
|
skip = frozenset([check])
|
|
caught = []
|
|
for v in doc["vectors"]:
|
|
try:
|
|
got = outcome(v["kind"], loaded[v["name"]], skip)
|
|
except Exception as e:
|
|
caught.append("%s (%s)" % (v["name"], type(e).__name__))
|
|
continue
|
|
if result(got) != stated(v):
|
|
caught.append("%s (%s)" % (v["name"], " ".join(str(x) for x in result(got) if x is not None)))
|
|
if check not in REQUIRED_CAUGHT:
|
|
if not caught:
|
|
uncaught.append(check)
|
|
continue
|
|
print(("PASS" if caught else "FAIL") + " the corpus catches the check %s left out%s" % (
|
|
check, ": " + caught[0] if caught else ""))
|
|
if not caught:
|
|
fails.append("with the check %s left out, every vector still comes out as stated" % check)
|
|
if uncaught:
|
|
print("NOTE no vector isolates the checks %s" % ", ".join(uncaught))
|
|
return fails
|
|
|
|
|
|
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: %s" % ",".join(st))
|
|
with open(os.path.join(corpus_dir, "vectors.json"), encoding="utf-8") as f:
|
|
doc = json.load(f)
|
|
if doc.get("format") != FORMAT:
|
|
fails.append("the index's format is %r" % doc.get("format"))
|
|
names = [v.get("name") for v in doc["vectors"]]
|
|
if len(set(names)) != len(names):
|
|
fails.append("vector names repeat")
|
|
loaded = {}
|
|
for v in doc["vectors"]:
|
|
try:
|
|
loaded[v["name"]] = load(corpus_dir, v)
|
|
except (OSError, ValueError) as e:
|
|
fails.append("vector %s: %s" % (v.get("name"), e))
|
|
if len(loaded) != len(doc["vectors"]):
|
|
for x in fails:
|
|
print("FAIL " + x)
|
|
print("FAIL live-contribution link corpus: the vectors do not load")
|
|
return 1
|
|
fails += check_vectors(doc, loaded)
|
|
|
|
listed = reasons_listed()
|
|
ok = listed == REASONS
|
|
print("%s §7's last list is exactly the reasons this verifier raises" % ("PASS" if ok else "FAIL"))
|
|
if not ok:
|
|
fails.append("§7's last list is %s" % listed)
|
|
ok = doc.get("reasons") == listed
|
|
print("%s the index lists exactly §7's last list" % ("PASS" if ok else "FAIL"))
|
|
if not ok:
|
|
fails.append("the index lists %s" % doc.get("reasons"))
|
|
refused = {v.get("reason") for v in doc["vectors"] if v.get("outcome") == "refuse"}
|
|
missing = [r for r in listed if r not in refused]
|
|
print("%s every reason of §7's last list refuses a vector" % ("PASS" if not missing else "FAIL"))
|
|
if missing:
|
|
fails.append("no vector is refused %s" % ", ".join(missing))
|
|
|
|
records, series = [], []
|
|
for v in doc["vectors"]:
|
|
if v.get("outcome") != "accept":
|
|
continue
|
|
if v["kind"] == "part":
|
|
records.append((v["name"], loaded[v["name"]][0]))
|
|
else:
|
|
series.append((v["name"], loaded[v["name"]]))
|
|
records += [("%s[%d]" % (v["name"], i), b) for i, b in enumerate(loaded[v["name"]])]
|
|
if not records or not series:
|
|
fails.append("the corpus holds no accepted part or no accepted series to mutate")
|
|
fails += run_mutations(records, series)
|
|
fails += run_dropped_checks(doc, loaded)
|
|
|
|
for x in fails:
|
|
print("FAIL " + x)
|
|
print(("PASS" if not fails else "FAIL") + " live-contribution link corpus")
|
|
return 1 if fails else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) != 2:
|
|
print("usage: live_contribution_verify.py <live-conformance/links>", file=sys.stderr)
|
|
sys.exit(2)
|
|
sys.exit(main(sys.argv[1]))
|