#!/usr/bin/env python3 """Second implementation of the Smart Byte v1 conformance pack. The pack in ../v1 is generated by the Rust reference implementation (smart-byte-rs, crate smart-byte-conformance). This script is a separate implementation of everything the pack exercises, written in another language from the spec text rather than ported from the Rust: canonical CBOR and its strict decoder (identity spec §2.3), SAID derivation and verification (§2.1, §2.2), Ed25519 (RFC 8032), key-event-log replay (identity spec §3, §4), envelope verification (envelope wire format §8), the cluster state machine (cluster state spec), the lockstep commit rule and dissent log (treatise §13, §14), and the content-addressed gossip intake the gossip vectors describe. It uses the Python standard library only. BLAKE3 and Ed25519 are implemented here from their specifications, and both are checked against published known-answer values before any vector is. python3 verify_v1.py [PACK_DIR] Exits 0 when every vector agrees with this implementation, 1 otherwise. """ import base64 import hashlib import json import pathlib import re import sys # --------------------------------------------------------------------------- # BLAKE3 (hash mode, 32-byte output) # --------------------------------------------------------------------------- _M32 = 0xFFFFFFFF _IV = [0x6A09E667, 0xBB67AE85, 0x3C6EF372, 0xA54FF53A, 0x510E527F, 0x9B05688C, 0x1F83D9AB, 0x5BE0CD19] _PERM = [2, 6, 3, 10, 7, 0, 4, 13, 1, 11, 12, 5, 9, 14, 15, 8] _CHUNK_START, _CHUNK_END, _PARENT, _ROOT = 1, 2, 4, 8 _BLOCK, _CHUNK = 64, 1024 def _rotr(x, n): return ((x >> n) | (x << (32 - n))) & _M32 def _g(s, a, b, c, d, x, y): s[a] = (s[a] + s[b] + x) & _M32 s[d] = _rotr(s[d] ^ s[a], 16) s[c] = (s[c] + s[d]) & _M32 s[b] = _rotr(s[b] ^ s[c], 12) s[a] = (s[a] + s[b] + y) & _M32 s[d] = _rotr(s[d] ^ s[a], 8) s[c] = (s[c] + s[d]) & _M32 s[b] = _rotr(s[b] ^ s[c], 7) def _compress(cv, block, counter, block_len, flags): m = [int.from_bytes(block[4 * i:4 * i + 4], "little") for i in range(16)] s = list(cv) + _IV[:4] + [counter & _M32, (counter >> 32) & _M32, block_len, flags] for r in range(7): _g(s, 0, 4, 8, 12, m[0], m[1]) _g(s, 1, 5, 9, 13, m[2], m[3]) _g(s, 2, 6, 10, 14, m[4], m[5]) _g(s, 3, 7, 11, 15, m[6], m[7]) _g(s, 0, 5, 10, 15, m[8], m[9]) _g(s, 1, 6, 11, 12, m[10], m[11]) _g(s, 2, 7, 8, 13, m[12], m[13]) _g(s, 3, 4, 9, 14, m[14], m[15]) if r < 6: m = [m[i] for i in _PERM] return [s[i] ^ s[i + 8] for i in range(8)] + [s[i + 8] ^ cv[i] for i in range(8)] def _chunk_output(chunk, index): """(cv, last block, block_len, flags) of a chunk, before its last compression.""" blocks = [chunk[i:i + _BLOCK] for i in range(0, len(chunk), _BLOCK)] or [b""] cv = list(_IV) for i, block in enumerate(blocks): flags = (_CHUNK_START if i == 0 else 0) | (_CHUNK_END if i == len(blocks) - 1 else 0) padded = block.ljust(_BLOCK, b"\0") if i == len(blocks) - 1: return cv, padded, len(block), flags, index cv = _compress(cv, padded, index, _BLOCK, flags)[:8] def blake3(data): """BLAKE3-256 of `data`.""" chunks = [data[i:i + _CHUNK] for i in range(0, len(data), _CHUNK)] or [b""] stack = [] for index, chunk in enumerate(chunks[:-1]): cv, block, n, flags, ctr = _chunk_output(chunk, index) node = _compress(cv, block, ctr, n, flags)[:8] total = index + 1 while total & 1 == 0: left = stack.pop() node = _compress(_IV, _words(left) + _words(node), 0, _BLOCK, _PARENT)[:8] total >>= 1 stack.append(node) out = _chunk_output(chunks[-1], len(chunks) - 1) while stack: cv, block, n, flags, ctr = out right = _compress(cv, block, ctr, n, flags)[:8] out = (_IV, _words(stack.pop()) + _words(right), _BLOCK, _PARENT, 0) cv, block, n, flags, ctr = out words = _compress(cv, block, ctr, n, flags | _ROOT)[:8] return b"".join(w.to_bytes(4, "little") for w in words) def _words(cv): return b"".join(w.to_bytes(4, "little") for w in cv) # --------------------------------------------------------------------------- # Ed25519 (RFC 8032), with the strict checks the pack expects # --------------------------------------------------------------------------- _P = 2 ** 255 - 19 _L = 2 ** 252 + 27742317777372353535851937790883648493 _D = -121665 * pow(121666, _P - 2, _P) % _P _SQRT_M1 = pow(2, (_P - 1) // 4, _P) def _recover_x(y, sign): if y >= _P: return None x2 = (y * y - 1) * pow(_D * y * y + 1, _P - 2, _P) % _P if x2 == 0: return None if sign else 0 x = pow(x2, (_P + 3) // 8, _P) if (x * x - x2) % _P != 0: x = x * _SQRT_M1 % _P if (x * x - x2) % _P != 0: return None if x & 1 != sign: x = _P - x return x _BY = 4 * pow(5, _P - 2, _P) % _P _BX = _recover_x(_BY, 0) _B = (_BX, _BY, 1, _BX * _BY % _P) _ZERO = (0, 1, 1, 0) def _add(p, q): a = (p[1] - p[0]) * (q[1] - q[0]) % _P b = (p[1] + p[0]) * (q[1] + q[0]) % _P c = 2 * p[3] * q[3] * _D % _P d = 2 * p[2] * q[2] % _P e, f, g, h = b - a, d - c, d + c, b + a return (e * f % _P, g * h % _P, f * g % _P, e * h % _P) def _mul(k, p): q = _ZERO while k: if k & 1: q = _add(q, p) p = _add(p, p) k >>= 1 return q def _eq(p, q): return (p[0] * q[2] - q[0] * p[2]) % _P == 0 and (p[1] * q[2] - q[1] * p[2]) % _P == 0 def _encode_point(p): zi = pow(p[2], _P - 2, _P) x, y = p[0] * zi % _P, p[1] * zi % _P return (y | ((x & 1) << 255)).to_bytes(32, "little") def _decode_point(b): n = int.from_bytes(b, "little") y, sign = n & ((1 << 255) - 1), n >> 255 x = _recover_x(y, sign) return None if x is None else (x, y, 1, x * y % _P) def _small_order(p): return _eq(_mul(8, p), _ZERO) def _sha512_int(*parts): return int.from_bytes(hashlib.sha512(b"".join(parts)).digest(), "little") def ed25519_public(seed): h = hashlib.sha512(seed).digest() a = int.from_bytes(h[:32], "little") & ((1 << 254) - 8) | (1 << 254) return _encode_point(_mul(a, _B)) def ed25519_sign(seed, msg): h = hashlib.sha512(seed).digest() a = int.from_bytes(h[:32], "little") & ((1 << 254) - 8) | (1 << 254) pub = _encode_point(_mul(a, _B)) r = _sha512_int(h[32:], msg) % _L big_r = _encode_point(_mul(r, _B)) k = _sha512_int(big_r, pub, msg) % _L return big_r + ((r + k * a) % _L).to_bytes(32, "little") def ed25519_verify(pub, msg, sig): """None if valid, else the pack's error class.""" if len(sig) != 64: return "malformed_signature" if len(pub) != 32: return "malformed_verifying_key" a = _decode_point(pub) if a is None: return "malformed_verifying_key" r = _decode_point(sig[:32]) s = int.from_bytes(sig[32:], "little") if r is None or s >= _L or _small_order(a) or _small_order(r): return "signature_invalid" k = _sha512_int(sig[:32], pub, msg) % _L if not _eq(_mul(s, _B), _add(r, _mul(k, a))): return "signature_invalid" return None # --------------------------------------------------------------------------- # Canonical CBOR (identity spec §2.3) # # Values: int, bytes, str, list, Map (a list of (key, value) pairs), bool, # None. Python's bool is an int subclass, so bools are tested first. # --------------------------------------------------------------------------- class Map(list): """A CBOR map as (key, value) pairs, in the order they were given.""" class Refused(Exception): def __init__(self, cls): super().__init__(cls) self.cls = cls def _head(major, n): if n < 24: return bytes([major << 5 | n]) for ai, width in ((24, 1), (25, 2), (26, 4), (27, 8)): if n < 1 << (8 * width): return bytes([major << 5 | ai]) + n.to_bytes(width, "big") raise ValueError("argument does not fit in 64 bits") def encode(v): if v is None: return b"\xf6" if v is True: return b"\xf5" if v is False: return b"\xf4" if isinstance(v, int): return _head(0, v) if v >= 0 else _head(1, -1 - v) if isinstance(v, bytes): return _head(2, len(v)) + v if isinstance(v, str): b = v.encode("utf-8") return _head(3, len(b)) + b if isinstance(v, Map): pairs = sorted((encode(k), encode(x)) for k, x in v) if any(pairs[i][0] == pairs[i + 1][0] for i in range(len(pairs) - 1)): raise Refused("duplicate_map_key") return _head(5, len(pairs)) + b"".join(k + x for k, x in pairs) if isinstance(v, list): return _head(4, len(v)) + b"".join(encode(x) for x in v) raise TypeError(f"no canonical form for {type(v).__name__}") def decode(data, strict=True): """One canonical item, or Refused(class). strict=False accepts over-long heads and unsorted maps; it exists only to show what a lenient implementation would compute.""" pos = 0 def need(n): nonlocal pos if n > len(data) - pos: raise Refused("truncated") out = data[pos:pos + n] pos += n return out def item(depth): nonlocal pos if depth > 128: raise Refused("too_deep") ib = need(1)[0] major, ai = ib >> 5, ib & 31 if major == 7: if ai == 20: return False if ai == 21: return True if ai == 22: return None if ai in (25, 26, 27): raise Refused("float_forbidden") if ai in (28, 29, 30, 31): raise Refused("malformed") if ai == 24 and need(1)[0] < 32: raise Refused("malformed") # RFC 8949 §3.3: not well-formed raise Refused("simple_value_forbidden") if major == 6: raise Refused("tag_forbidden") if ai in (28, 29, 30): raise Refused("malformed") if ai == 31: raise Refused("malformed" if major <= 1 else "indefinite_length") if ai < 24: n = ai else: width = {24: 1, 25: 2, 26: 4, 27: 8}[ai] n = int.from_bytes(need(width), "big") if strict and n < (24 if width == 1 else 1 << (8 * width // 2)): raise Refused("non_shortest_integer" if major <= 1 else "non_shortest_length") if major == 0: return n if major == 1: return -1 - n if major in (2, 3): raw = need(n) if major == 2: return raw try: return raw.decode("utf-8") except UnicodeDecodeError: raise Refused("invalid_utf8") from None if major == 4: return [item(depth + 1) for _ in range(n)] out, prev = Map(), None for _ in range(n): start = pos k = item(depth + 1) kb = data[start:pos] if strict and prev is not None: if kb == prev: raise Refused("duplicate_map_key") if kb < prev: raise Refused("unsorted_map_keys") prev = kb out.append((k, item(depth + 1))) return out v = item(0) if pos != len(data): raise Refused("trailing_bytes") return v def from_json(j): """The pack's typed JSON: {"$bytes": hex} is a byte string; `$bytes` is reserved and never a map member.""" if isinstance(j, dict): if "$bytes" in j: h = j["$bytes"] if len(j) != 1 or not isinstance(h, str): raise ValueError("`$bytes` is reserved: a lone member holding hex") if len(h) % 2 or any(c not in "0123456789abcdef" for c in h): raise ValueError(f"$bytes must be lowercase hex: {h!r}") return bytes.fromhex(h) return Map((k, from_json(v)) for k, v in j.items()) if isinstance(j, list): return [from_json(x) for x in j] if isinstance(j, float): raise ValueError("floats are forbidden") return j def load_json(path): """Parse JSON, refusing any object that names a member twice.""" def no_duplicates(pairs): out = {} for k, v in pairs: if k in out: raise ValueError(f"duplicate member {k!r}") out[k] = v return out return json.loads(path.read_text(encoding="utf-8"), object_pairs_hook=no_duplicates) # --------------------------------------------------------------------------- # SAIDs (identity spec §2.1, §2.2) and receiving them (§2.6) # --------------------------------------------------------------------------- PLACEHOLDER = "#" * 53 def said_text(digest): return "E" + base64.b32encode(digest).decode().rstrip("=").lower() def parse_said(text): if not (isinstance(text, str) and len(text) == 53 and text[0] == "E" and all(c in "abcdefghijklmnopqrstuvwxyz234567" for c in text[1:])): return None digest = base64.b32decode(text[1:].upper() + "====") return digest if said_text(digest) == text else None def _with(obj, field, value): return Map((k, value if k == field else v) for k, v in obj) def derive(obj, field): """(hashed bytes, digest, stamped object bytes).""" if not isinstance(obj, Map): raise Refused("not_a_map") if not any(k == field for k, _ in obj): raise Refused("missing_said_field") hashed = encode(_with(obj, field, PLACEHOLDER)) digest = blake3(hashed) return hashed, digest, encode(_with(obj, field, said_text(digest))) def verify_said(data, field): """The digest the bytes state and hash to, or Refused(class).""" obj = decode(data) if not isinstance(obj, Map): raise Refused("not_a_map") stated = [v for k, v in obj if k == field] if not stated: raise Refused("missing_said_field") digest = parse_said(stated[0]) if digest is None: raise Refused("malformed_said") if derive(obj, field)[1] != digest: raise Refused("said_mismatch") return digest def receive(held, claimed, data): """Identity spec §2.6, in order: canonical bytes, SAID match, the claimed SAID, and only then deduplication. Returns the result object.""" try: digest = verify_said(data, "d") claim = parse_said(claimed) if claim is None: raise Refused("malformed_said") if claim != digest: raise Refused("said_mismatch") except Refused as e: return {"outcome": "rejected", "error": e.cls} text = said_text(digest) if any(h["said"] == text for h in held): return {"outcome": "duplicate", "said": text} held.append({"said": text, "cbor_hex": data.hex()}) return {"outcome": "accepted", "said": text} # --------------------------------------------------------------------------- # Controller signatures (identity spec §3.5; envelope wire format §7) # --------------------------------------------------------------------------- ED25519 = 0x01 U64 = 1 << 64 def is_uint(x): return isinstance(x, int) and not isinstance(x, bool) and 0 <= x < U64 def parse_aid(text): """The digest a controller AID spells — `B` and a SAID's 52 characters — or None.""" if not (isinstance(text, str) and len(text) == 53 and text[0] == "B"): return None return parse_said("E" + text[1:]) def aid_text(digest): return "B" + said_text(digest)[1:] def verify_indexed(keys, threshold, message, signatures): """Every listed (index, algorithm byte ‖ signature) valid over `message`, no key index twice, and at least the threshold — never fewer than one — of keys signing. None, or the error class.""" seen = set() for index, sig in signatures: if not 0 <= index < len(keys): return "key_index_out_of_range" if index in seen: return "duplicate_key_index" seen.add(index) key = keys[index] if not key: return "malformed_verifying_key" if not sig or sig[0] != key[0]: return "algorithm_mismatch" if key[0] != ED25519: return "unsupported_algorithm" err = ed25519_verify(key[1:], message, sig[1:]) if err: return err if len(seen) < max(threshold, 1): return "below_threshold" return None # --------------------------------------------------------------------------- # Key-event logs (identity spec §3, §4) # --------------------------------------------------------------------------- EVENT_MEMBERS = {"a", "d", "i", "k", "kt", "n", "nt", "p", "s", "t", "v"} def members(obj, names, cls): """`obj` as a dict, if it is a map with exactly the text members `names`.""" if not isinstance(obj, Map) or {k for k, _ in obj} != names or not all(isinstance(k, str) for k, _ in obj): raise Refused(cls) return dict(obj) def decode_event(data): """Identity spec §4 step 1: canonical bytes, a v1 event type, exactly the §3.3 members of their types, the version, the SAID, and an inception's own AID. Returns the event's members.""" obj = decode(data) t = dict(obj).get("t") if isinstance(obj, Map) else None if not isinstance(t, str): raise Refused("malformed_event") if t in ("rec", "dip"): raise Refused("unsupported_event_type") if t not in ("icp", "rot", "ixn"): raise Refused("malformed_event") e = members(obj, EVENT_MEMBERS, "malformed_event") well_typed = ( isinstance(e["v"], str) and parse_said(e["d"]) is not None and parse_aid(e["i"]) is not None and all(is_uint(e[f]) for f in ("s", "kt", "nt")) and (isinstance(e["p"], str) if t == "icp" else parse_said(e["p"]) is not None) and isinstance(e["k"], list) and all(isinstance(k, bytes) for k in e["k"]) and isinstance(e["n"], list) and all(parse_said(x) is not None for x in e["n"]) and isinstance(e["a"], list) and all(parse_said(x) is not None for x in e["a"]) ) if not well_typed: raise Refused("malformed_event") if e["v"] != "SBYTE10JSON": raise Refused("unsupported_version") placeholdered = {"d", "i"} if t == "icp" else {"d"} digest = blake3(encode(Map((k, PLACEHOLDER if k in placeholdered else v) for k, v in obj))) if said_text(digest) != e["d"]: raise Refused("said_mismatch") if t == "icp" and e["i"] != aid_text(digest): raise Refused("controller_mismatch") return e def per_key_said(key): """Identity spec §3.4.""" return said_text(blake3(encode(Map([("d", PLACEHOLDER), ("k", key)])))) def check_key_fields(e): """An establishment event's key fields (§3.3): no key or digest twice, and each threshold in range.""" if len(set(e["k"])) != len(e["k"]): raise Refused("duplicate_key") if not 1 <= e["kt"] <= len(e["k"]): raise Refused("invalid_threshold") if len(set(e["n"])) != len(e["n"]): raise Refused("duplicate_key") if (not e["n"] and e["nt"] != 0) or (e["n"] and not 1 <= e["nt"] <= len(e["n"])): raise Refused("invalid_threshold") class LogRefused(Exception): def __init__(self, cls, at): super().__init__(f"{cls} at event {at}") self.cls, self.at = cls, at def kel_step(state, e, data, sigs): """Identity spec §4 steps 2-6 for one decoded event `e` (its bytes `data` and signatures `sigs`), against the state its log has reached — None before the inception. Returns the state the event establishes, with every establishment event's key state; raises Refused.""" t = e["t"] if (state is None) != (t == "icp"): raise Refused("malformed_event") if e["s"] != (0 if state is None else state["s"] + 1): raise Refused("out_of_sequence") if e["p"] != ("" if state is None else state["d"]): raise Refused("prior_mismatch") if state is not None and e["i"] != state["i"]: raise Refused("controller_mismatch") if t == "icp": if e["a"]: raise Refused("malformed_event") check_key_fields(e) keys, threshold = e["k"], e["kt"] elif t == "rot": if not state["n"]: raise Refused("rotation_not_permitted") if len(e["k"]) != len(state["n"]) or any( per_key_said(k) != n for k, n in zip(e["k"], state["n"])): raise Refused("pre_rotation_mismatch") if e["a"]: raise Refused("malformed_event") check_key_fields(e) # Signed by the keys it reveals, under the committed threshold and # its own (§3.5). keys, threshold = e["k"], max(state["nt"], e["kt"]) else: if [e[f] for f in ("k", "kt", "n", "nt")] != [state[f] for f in ("k", "kt", "n", "nt")]: raise Refused("interaction_changes_keys") keys, threshold = state["k"], state["kt"] err = verify_indexed(keys, threshold, data, sigs) if err: raise Refused(err) if t == "ixn": return dict(state, s=e["s"], d=e["d"]) establishments = (state["establishments"] if state else []) + [ {"sequence": e["s"], "keys": e["k"], "threshold": e["kt"]}] return dict(e, establishments=establishments) def replay(log): """Identity spec §4 over [(bytes, [(index, signature)])]. Returns the final state, with every establishment event's key state; raises LogRefused at the first event refused.""" state = None for at, (data, sigs) in enumerate(log): try: state = kel_step(state, decode_event(data), data, sigs) except Refused as r: raise LogRefused(r.cls, at) from None if state is None: raise LogRefused("malformed_event", 0) return state # --------------------------------------------------------------------------- # Envelopes (envelope wire format §2–§8) # --------------------------------------------------------------------------- def controller_signature(x, cls): sig = members(x, {"establishment", "signatures"}, cls) if not is_uint(sig["establishment"]) or not isinstance(sig["signatures"], list): raise Refused(cls) return sig["establishment"], signature_list(sig["signatures"], cls) def signature_list(items, cls): """[{index, signature}] as [(index, signature)].""" if not isinstance(items, list): raise Refused(cls) out = [] for item in items: one = members(item, {"index", "signature"}, cls) if not is_uint(one["index"]) or not isinstance(one["signature"], bytes): raise Refused(cls) out.append((one["index"], one["signature"])) return out def provenance_parts(obj, cls): """A provenance (envelope wire format §3) of exactly its form: its members and the issuer's signature.""" prov = members(obj, {"d", "issuer", "recipient", "nonce", "cargo", "signature"}, cls) cargo = members(prov["cargo"], {"type", "bytes"}, cls) if not (parse_said(prov["d"]) and parse_aid(prov["issuer"]) and parse_aid(prov["recipient"]) and isinstance(prov["nonce"], bytes) and len(prov["nonce"]) == 16 and is_uint(cargo["type"]) and isinstance(cargo["bytes"], bytes)): raise Refused(cls) return prov, controller_signature(prov["signature"], cls) TRANSITION_MEMBERS = {"previous_history_hash", "from", "to", "frame", "signature"} def transition_parts(obj, cls): """A transition (envelope wire format §5) of exactly its form — with a `destination` only when it moves the byte to another cluster: its members and its signature.""" present = {k for k, _ in obj} if isinstance(obj, Map) else set() tm = members(obj, TRANSITION_MEMBERS | ({"destination"} & present), cls) if not (isinstance(tm["previous_history_hash"], bytes) and len(tm["previous_history_hash"]) == 32 and parse_aid(tm["from"]) and parse_aid(tm["to"]) and is_uint(tm["frame"]) and ("destination" not in tm or parse_said(tm["destination"]) is not None)): raise Refused(cls) return tm, controller_signature(tm["signature"], cls) def with_destination(pairs, t): """`pairs`, plus the transition's destination when it has one.""" return Map(pairs + ([("destination", t["destination"])] if "destination" in t else [])) def attested(prov): """What the issuer signs: {cargo, issuer, nonce, recipient} (§3).""" return encode(Map([("cargo", prov["cargo"]), ("issuer", prov["issuer"]), ("nonce", prov["nonce"]), ("recipient", prov["recipient"])])) def transition_message(identity, t): """What an owner signs to transfer a byte (§5).""" return encode(with_destination([("byte", identity), ("previous_history_hash", t["previous_history_hash"]), ("from", t["from"]), ("to", t["to"]), ("frame", t["frame"])], t)) def next_history(history, t): """The history hash after transition `t` (§5).""" return blake3(history + encode(with_destination([("from", t["from"]), ("to", t["to"]), ("frame", t["frame"]), ("signature", t["signature"])], t))) def decode_envelope(data): """§8 step 1: canonical bytes that are exactly an envelope, both SAIDs checking. Returns (envelope members, provenance members, its signature, [(transition members, its signature)]).""" obj = decode(data) env = members(obj, {"d", "provenance", "ownership_chain", "joule_cost"}, "not_an_envelope") cost = members(env["joule_cost"], {"measured_microjoules", "estimated_microjoules"}, "not_an_envelope") prov, issuer_sig = provenance_parts(env["provenance"], "not_an_envelope") if not (parse_said(env["d"]) and all(is_uint(x) for x in cost.values()) and isinstance(env["ownership_chain"], list)): raise Refused("not_an_envelope") chain = [transition_parts(t, "not_an_envelope") for t in env["ownership_chain"]] if derive(obj, "d")[1] != parse_said(env["d"]): raise Refused("said_mismatch") if derive(env["provenance"], "d")[1] != parse_said(prov["d"]): raise Refused("said_mismatch") return env, prov, issuer_sig, chain def verify_envelope(data, key_states): """§8: decode, the issuer's signature, then every transition in order. `key_states` maps (AID, establishment sequence) to (keys, threshold).""" env, prov, issuer_sig, chain = decode_envelope(data) def signed_by(aid, signature, message): establishment, sigs = signature state = key_states.get((aid, establishment)) if state is None: raise Refused("unknown_key_state") err = verify_indexed(state[0], state[1], message, sigs) if err: raise Refused(err) signed_by(prov["issuer"], issuer_sig, attested(prov)) identity = prov["d"] owner, history, last_frame = prov["recipient"], parse_said(identity), None for t, signature in chain: if t["previous_history_hash"] != history: raise Refused("broken_history_chain") if t["from"] != owner: raise Refused("not_current_owner") if last_frame is not None and t["frame"] <= last_frame: raise Refused("frame_not_increasing") signed_by(t["from"], signature, transition_message(identity, t)) history = next_history(history, t) # Frames rise within a cluster; a move to another starts its count. owner, last_frame = t["to"], (None if "destination" in t else t["frame"]) return {"identity": identity, "snapshot": env["d"], "issuer": prov["issuer"], "owner": owner, "history_hash": history.hex(), "transfers": len(chain)} # --------------------------------------------------------------------------- # The cluster state machine (cluster state spec) # # A state is a dict of the spec's members. Maps keyed by AID, identity or # cluster are dicts of records; records are dicts of their members; cargo # and change maps stay Maps, as they were decoded. # --------------------------------------------------------------------------- STATE_MEMBERS = {"frame", "cluster", "peers", "registry", "governance", "controllers", "bytes", "retired", "exported", "imported"} CONTROLLER_MEMBERS = {"establishment", "keys", "threshold", "next_keys", "next_threshold", "sequence", "last_event"} BYTE_MEMBERS = {"issuer", "cargo", "minted", "owner", "history_hash"} INPUT_CLASSES = ("key_event", "governance", "mint", "import", "transfer", "burn") def record_map(r): return Map(r.items()) def encode_state(st): """§2: the state's canonical encoding — the checkpoint.""" return encode(Map([ ("frame", st["frame"]), ("cluster", st["cluster"]), ("peers", Map((c, Map([("keys", keys)])) for c, keys in st["peers"].items())), ("registry", Map((aid, sorted(types)) for aid, types in st["registry"].items())), ("governance", Map([("delay", st["delay"]), ("sequence", st["sequence"]), ("pending", [Map([("change", ch), ("effective", eff)]) for ch, eff in st["pending"]])])), ("controllers", Map((aid, record_map(r)) for aid, r in st["controllers"].items())), ("bytes", Map((i, record_map(r)) for i, r in st["bytes"].items())), ("retired", Map((i, record_map(r)) for i, r in st["retired"].items())), ("exported", Map((i, record_map(r)) for i, r in st["exported"].items())), ("imported", sorted(st["imported"])), ])) def decode_state(data): """A state from its canonical encoding, which it must reproduce.""" m = members(decode(data), STATE_MEMBERS, "malformed_state") gov = members(m["governance"], {"delay", "sequence", "pending"}, "malformed_state") st = { "frame": m["frame"], "cluster": m["cluster"], "peers": {c: dict(members(p, {"keys"}, "malformed_state"))["keys"] for c, p in m["peers"]}, "registry": {aid: list(types) for aid, types in m["registry"]}, "delay": gov["delay"], "sequence": gov["sequence"], "pending": [(dict(p)["change"], dict(p)["effective"]) for p in gov["pending"]], "controllers": {aid: dict(members(r, CONTROLLER_MEMBERS, "malformed_state")) for aid, r in m["controllers"]}, "bytes": {i: dict(members(r, BYTE_MEMBERS, "malformed_state")) for i, r in m["bytes"]}, "retired": {i: dict(members(r, BYTE_MEMBERS | {"retired"}, "malformed_state")) for i, r in m["retired"]}, "exported": {i: dict(members(r, BYTE_MEMBERS | {"destination", "exported"}, "malformed_state")) for i, r in m["exported"]}, "imported": list(m["imported"]), } check(encode_state(st) == data, "the state does not reproduce its encoding") return st def is_ed25519_key(key): """An Ed25519 key a signature could verify under: 01, then 32 bytes that decode to a point not of small order.""" if not (isinstance(key, bytes) and len(key) == 33 and key[0] == ED25519): return False point = _decode_point(key[1:]) return point is not None and not _small_order(point) def change_parts(obj): """A proposal's change (§4.5) of exactly its form.""" action = dict(obj).get("action") if isinstance(obj, Map) else None forms = {"add_issuer": {"action", "cargo_type", "issuer"}, "remove_issuer": {"action", "cargo_type", "issuer"}, "set_peer": {"action", "peer", "keys"}, "remove_peer": {"action", "peer"}} if action not in forms: raise Refused("malformed_input") ch = members(obj, forms[action], "malformed_input") if ("cargo_type" in ch and not is_uint(ch["cargo_type"])) or ("issuer" in ch and parse_aid(ch["issuer"]) is None) \ or ("peer" in ch and parse_said(ch["peer"]) is None) \ or ("keys" in ch and not (isinstance(ch["keys"], list) and all(isinstance(k, bytes) for k in ch["keys"]))): raise Refused("malformed_input") return ch def parse_input(data): """§4.1: the class an input names (None if none), and its parts or why it is not an input (a Refused).""" try: obj = decode(data) except Refused as e: return None, e if not (isinstance(obj, Map) and len(obj) == 1 and isinstance(obj[0][0], str) and obj[0][0] in INPUT_CLASSES): return None, Refused("malformed_input") cls, body = obj[0] try: if cls == "key_event": ke = members(body, {"event", "signatures"}, "malformed_input") sigs = signature_list(ke["signatures"], "malformed_input") event_bytes = encode(ke["event"]) return cls, (decode_event(event_bytes), event_bytes, sigs) if cls == "governance": g = members(body, {"proposal", "approvals"}, "malformed_input") pr = members(g["proposal"], {"cluster", "sequence", "change"}, "malformed_input") if parse_said(pr["cluster"]) is None or not is_uint(pr["sequence"]) or not isinstance(g["approvals"], list): raise Refused("malformed_input") change = change_parts(pr["change"]) approvals = [] for a in g["approvals"]: ap = members(a, {"approver", "signature"}, "malformed_input") if parse_aid(ap["approver"]) is None: raise Refused("malformed_input") approvals.append((ap["approver"], controller_signature(ap["signature"], "malformed_input"))) return cls, (pr, change, encode(g["proposal"]), approvals) if cls == "mint": prov, sig = provenance_parts(body, "malformed_input") return cls, (body, prov, sig) if cls == "import": im = members(body, {"certificate", "envelope", "logs"}, "malformed_input") cert = members(im["certificate"], {"statement", "signatures"}, "malformed_input") stm = members(cert["statement"], {"source", "byte", "export"}, "malformed_input") export = members(stm["export"], BYTE_MEMBERS | {"destination", "exported"}, "malformed_input") cargo = members(export["cargo"], {"type", "bytes"}, "malformed_input") if not (parse_said(stm["source"]) and parse_said(stm["byte"]) and parse_aid(export["issuer"]) and is_uint(export["minted"]) and parse_aid(export["owner"]) and isinstance(export["history_hash"], bytes) and len(export["history_hash"]) == 32 and parse_said(export["destination"]) and is_uint(export["exported"]) and is_uint(cargo["type"]) and isinstance(cargo["bytes"], bytes) and isinstance(im["logs"], list) and all(isinstance(log, list) for log in im["logs"])): raise Refused("malformed_input") cert_sigs = signature_list(cert["signatures"], "malformed_input") logs = [] for log in im["logs"]: events = [] for e in log: le = members(e, {"event", "signatures"}, "malformed_input") events.append((le["event"], signature_list(le["signatures"], "malformed_input"))) logs.append(events) return cls, (stm, export, encode(cert["statement"]), cert_sigs, im["envelope"], logs) if cls == "transfer": tr = members(body, {"byte", "transition"}, "malformed_input") if parse_said(tr["byte"]) is None: raise Refused("malformed_input") t, sig = transition_parts(tr["transition"], "malformed_input") return cls, (tr["byte"], t, sig) b = members(body, {"byte", "previous_history_hash", "owner", "frame", "signature"}, "malformed_input") if not (parse_said(b["byte"]) and isinstance(b["previous_history_hash"], bytes) and len(b["previous_history_hash"]) == 32 and parse_aid(b["owner"]) and is_uint(b["frame"])): raise Refused("malformed_input") return cls, (b, controller_signature(b["signature"], "malformed_input")) except Refused as e: return cls, e def kel_state_of(aid, r): """The key-event-log state a controller record resumes from.""" return {"i": aid, "s": r["sequence"], "d": r["last_event"], "k": r["keys"], "kt": r["threshold"], "n": r["next_keys"], "nt": r["next_threshold"], "establishments": [{"sequence": r["establishment"], "keys": r["keys"], "threshold": r["threshold"]}]} def record_of(state): """The controller record of a key-event-log state.""" return {"establishment": state["establishments"][-1]["sequence"], "keys": state["k"], "threshold": state["kt"], "next_keys": state["n"], "next_threshold": state["nt"], "sequence": state["s"], "last_event": state["d"]} def signed_in_force(st, signer, signature, message): """§4.8: a signature by `signer` under the establishment event in force.""" establishment, sigs = signature r = st["controllers"].get(signer) if r is None: raise Refused("unknown_controller") if establishment != r["establishment"]: raise Refused("establishment_not_in_force") err = verify_indexed(r["keys"], r["threshold"], message, sigs) if err: raise Refused(err) def live(st, identity): """The live record of a byte, or why there is none (§4.4 step 1).""" if identity in st["bytes"]: return st["bytes"][identity] if identity in st["retired"]: raise Refused("byte_retired") if identity in st["exported"]: raise Refused("byte_exported") raise Refused("unknown_byte") def supermajority(n): return 2 * n // 3 + 1 def apply_governance(st, pr, change, message, approvals): """§4.5.""" if pr["cluster"] != st["cluster"]: raise Refused("wrong_cluster") if pr["sequence"] != st["sequence"]: raise Refused("stale_sequence") action, target = change["action"], None registered = lambda issuer, t: t in st["registry"].get(issuer, []) if action == "add_issuer": if registered(change["issuer"], change["cargo_type"]): raise Refused("already_registered") elif action == "remove_issuer": if not registered(change["issuer"], change["cargo_type"]): raise Refused("not_registered") target = change["issuer"] elif action == "set_peer": if change["peer"] == st["cluster"]: raise Refused("wrong_cluster") if not change["keys"]: raise Refused("invalid_threshold") for i, k in enumerate(change["keys"]): if k in change["keys"][:i]: raise Refused("duplicate_key") if not is_ed25519_key(k): raise Refused("malformed_verifying_key") elif change["peer"] not in st["peers"]: raise Refused("not_a_peer") every = {a for a in st["registry"] if a != target} if action in ("add_issuer", "remove_issuer"): of_type = {a for a, types in st["registry"].items() if change["cargo_type"] in types and a != target} electorate = of_type or every else: electorate = every if not electorate: raise Refused("no_electorate") seen = set() for approver, signature in approvals: if approver not in electorate: raise Refused("not_an_elector") if approver in seen: raise Refused("duplicate_approver") seen.add(approver) signed_in_force(st, approver, signature, message) if len(seen) < supermajority(len(electorate)): raise Refused("below_quorum") st["sequence"] += 1 st["pending"].append((pr["change"], st["frame"] + 1 + st["delay"])) def enact_pending(st): """§4 step 1: changes in force from this frame take effect, in order.""" due = [ch for ch, eff in st["pending"] if eff == st["frame"]] st["pending"] = [(ch, eff) for ch, eff in st["pending"] if eff != st["frame"]] for ch_map in due: ch = dict(ch_map) if ch["action"] == "add_issuer": types = st["registry"].setdefault(ch["issuer"], []) if ch["cargo_type"] not in types: types.append(ch["cargo_type"]) elif ch["action"] == "remove_issuer": types = st["registry"].get(ch["issuer"], []) if ch["cargo_type"] in types: types.remove(ch["cargo_type"]) if not types: st["registry"].pop(ch["issuer"], None) elif ch["action"] == "set_peer": st["peers"][ch["peer"]] = ch["keys"] else: st["peers"].pop(ch["peer"], None) def apply_import(st, stm, export, statement_bytes, cert_sigs, envelope, logs): """§5: a byte from another cluster.""" if export["destination"] != st["cluster"]: raise Refused("wrong_cluster") keys = st["peers"].get(stm["source"]) if keys is None: raise Refused("unknown_cluster") err = verify_indexed(keys, supermajority(len(keys)), statement_bytes, cert_sigs) if err: raise Refused(err) digest = blake3(statement_bytes) if digest in st["imported"]: raise Refused("duplicate_import") if stm["byte"] in st["bytes"]: raise Refused("duplicate_byte") if stm["byte"] in st["retired"]: raise Refused("byte_retired") env_bytes = encode(envelope) env, prov, _, chain = decode_envelope(env_bytes) owner, history = prov["recipient"], parse_said(prov["d"]) for t, _ in chain: owner, history = t["to"], next_history(history, t) if (prov["d"], history, owner, prov["issuer"], encode(prov["cargo"])) != \ (stm["byte"], export["history_hash"], export["owner"], export["issuer"], encode(export["cargo"])): raise Refused("certificate_mismatch") key_states = {} for log in logs: try: state = replay([(encode(event), sigs) for event, sigs in log]) except LogRefused as r: raise Refused(r.cls) from None for x in state["establishments"]: key_states[(state["i"], x["sequence"])] = (x["keys"], x["threshold"]) verify_envelope(env_bytes, key_states) st["exported"].pop(stm["byte"], None) st["bytes"][stm["byte"]] = {f: export[f] for f in BYTE_MEMBERS} st["imported"].append(digest) def apply_input(st, cls, parts): """§4.2-§4.7: apply one decoded input, or raise Refused leaving the state as it was.""" if cls == "key_event": e, data, sigs = parts held = st["controllers"].get(e["i"]) if e["t"] == "icp" and held is not None: raise Refused("controller_exists") if e["t"] != "icp" and held is None: raise Refused("unknown_controller") state = kel_step(None if held is None else kel_state_of(e["i"], held), e, data, sigs) st["controllers"][e["i"]] = record_of(state) elif cls == "governance": apply_governance(st, *parts) elif cls == "mint": obj, prov, sig = parts if derive(obj, "d")[1] != parse_said(prov["d"]): raise Refused("said_mismatch") if dict(prov["cargo"])["type"] not in st["registry"].get(prov["issuer"], []): raise Refused("issuer_not_registered") signed_in_force(st, prov["issuer"], sig, attested(prov)) if any(prov["d"] in st[m] for m in ("bytes", "retired", "exported")): raise Refused("duplicate_byte") st["bytes"][prov["d"]] = {"issuer": prov["issuer"], "cargo": prov["cargo"], "minted": st["frame"], "owner": prov["recipient"], "history_hash": parse_said(prov["d"])} elif cls == "import": apply_import(st, *parts) elif cls == "transfer": identity, t, sig = parts b = live(st, identity) if t["frame"] != st["frame"]: raise Refused("wrong_frame") if t["previous_history_hash"] != b["history_hash"]: raise Refused("broken_history_chain") if t["from"] != b["owner"]: raise Refused("not_current_owner") if "destination" in t and t["destination"] not in st["peers"]: raise Refused("unknown_cluster") signed_in_force(st, t["from"], sig, transition_message(identity, t)) history = next_history(b["history_hash"], t) if "destination" in t: del st["bytes"][identity] st["exported"][identity] = dict(b, owner=t["to"], history_hash=history, destination=t["destination"], exported=st["frame"]) else: b["owner"], b["history_hash"] = t["to"], history else: burn, sig = parts b = live(st, burn["byte"]) if burn["frame"] != st["frame"]: raise Refused("wrong_frame") if burn["previous_history_hash"] != b["history_hash"]: raise Refused("broken_history_chain") if burn["owner"] != b["owner"]: raise Refused("not_current_owner") message = encode(Map([("burn", burn["byte"]), ("previous_history_hash", burn["previous_history_hash"]), ("owner", burn["owner"]), ("frame", burn["frame"])])) signed_in_force(st, burn["owner"], sig, message) del st["bytes"][burn["byte"]] st["retired"][burn["byte"]] = dict(b, retired=st["frame"]) def apply_frame(st, inputs): """§4: one result per distinct input, in the order they apply; the frame number advances.""" enact_pending(st) distinct = {blake3(b): b for b in inputs} rank = {c: i for i, c in enumerate(INPUT_CLASSES)} rank[None] = len(INPUT_CLASSES) ordered = sorted(((*parse_input(b), digest) for digest, b in distinct.items()), key=lambda x: (rank[x[0]], x[2])) results = [] for cls, parts, digest in ordered: result = {"input": digest.hex(), "class": cls} try: if isinstance(parts, Refused): raise parts apply_input(st, cls, parts) result["outcome"] = "applied" except Refused as e: result["outcome"], result["error"] = "dropped", e.cls results.append(result) st["frame"] += 1 return results # --------------------------------------------------------------------------- # The checks. Each raises Mismatch; none uses `assert`, so `python -O` # cannot switch them off. # --------------------------------------------------------------------------- class Mismatch(Exception): pass def check(condition, message): if not condition: raise Mismatch(message) def refused_class(fn, *args): try: fn(*args) except Refused as e: return e.cls return None def check_canonicalisation(v): if "input" in v: b = encode(from_json(v["input"])) check(b.hex() == v["expected_cbor_hex"], f"expected_cbor_hex: computed {b.hex()}") check(encode(decode(b)) == b, "decode then encode changed the bytes") else: got = refused_class(decode, bytes.fromhex(v["input_cbor_hex"])) check(got == v["error"], f"refused with {got}, vector says {v['error']}") def check_said(v): field = v["said_field"] if "input" in v: obj = from_json(v["input"]) check([x for k, x in obj if k == field] == [PLACEHOLDER], "input must hold the placeholder") hashed, digest, stamped = derive(obj, field) check(hashed.hex() == v["canonical_cbor_hex"], "canonical_cbor_hex") check(digest.hex() == v["blake3_hex"], "blake3_hex") check(said_text(digest) == v["expected_said"], f"expected_said: computed {said_text(digest)}") check(stamped.hex() == v["said_cbor_hex"], "said_cbor_hex") check(verify_said(stamped, field) == digest, "said_cbor_hex does not verify") else: data = bytes.fromhex(v["input_cbor_hex"]) got = refused_class(verify_said, data, field) check(got == v["error"], f"refused with {got}, vector says {v['error']}") if "said_of_received_bytes" in v or "said_after_reencoding" in v: lenient = decode(data, strict=False) stated = [x for k, x in lenient if k == field] check(len(stated) == 1 and isinstance(stated[0], str) and len(stated[0]) == 53, "no 53-character SAID to substitute") if "said_of_received_bytes" in v: at = data.find(stated[0].encode()) check(at >= 0, "the stated SAID is not in the bytes") raw = data[:at] + PLACEHOLDER.encode() + data[at + 53:] check(said_text(blake3(raw)) == v["said_of_received_bytes"], "said_of_received_bytes") if "said_after_reencoding" in v: check(said_text(derive(lenient, field)[1]) == v["said_after_reencoding"], "said_after_reencoding") def check_signature(v): message = bytes.fromhex(v["message_hex"]) seed = v["signing_key_seed"] if seed is not None: check(ed25519_public(bytes.fromhex(seed)).hex() == v["verifying_key"], "verifying_key is not the key the seed derives") sig = bytes.fromhex(v["signature"]) got = ed25519_verify(bytes.fromhex(v["verifying_key"]), message, sig) if v["expected_verify"] is True: check(got is None and v["error"] is None, f"a positive vector fails: {got}") check(seed is not None, "a positive vector must carry its seed") check(ed25519_sign(bytes.fromhex(seed), message) == sig, "the deterministic signature differs") else: check(v["expected_verify"] is False, "expected_verify is not a boolean") check(got == v["error"], f"refused with {got}, vector says {v['error']}") def typed_matches(v, member, data): if member in v: check(encode(from_json(v[member])) == data, f"`{member}` does not encode to the vector's bytes") def check_kel(v, logs): """Replay the vector's log; an accepted log's key states are recorded in `logs` for the envelope vectors.""" log = [] for i, e in enumerate(v["events"]): data = bytes.fromhex(e["cbor_hex"]) typed_matches(e, "event", data) sigs = [] for s in e["signatures"]: sig = bytes.fromhex(s["signature"]) if s["seed"] is not None: check(sig == b"\x01" + ed25519_sign(bytes.fromhex(s["seed"]), data), f"event {i}: a signature is not the one its seed makes") sigs.append((s["index"], sig)) log.append((data, sigs)) check(("expected" in v) != ("error" in v), "a vector has exactly one of expected and error") try: state = replay(log) except LogRefused as r: check("error" in v, f"refused with {r.cls} at event {r.at}") check((r.cls, r.at) == (v["error"], v["error_at"]), f"refused with {r.cls} at event {r.at}, vector says {v['error']} at {v['error_at']}") return check("expected" in v, f"accepted, vector says {v.get('error')}") computed = { "controller": state["i"], "event_saids": [decode_event(data)["d"] for data, _ in log], "establishments": [{"sequence": x["sequence"], "keys": [k.hex() for k in x["keys"]], "threshold": x["threshold"]} for x in state["establishments"]], "next_key_digests": state["n"], "next_threshold": state["nt"], } check(v["expected"] == computed, f"expected: computed {computed}") logs[v["name"]] = state def check_envelope(v, logs): key_states = {} for name in v["key_event_logs"]: check(name in logs, f"{name} is not an accepted log of kel_vectors.json") for x in logs[name]["establishments"]: key_states[(logs[name]["i"], x["sequence"])] = (x["keys"], x["threshold"]) data = bytes.fromhex(v["cbor_hex"]) typed_matches(v, "envelope", data) check(("expected" in v) != ("error" in v), "a vector has exactly one of expected and error") try: computed = verify_envelope(data, key_states) except Refused as r: check("error" in v, f"refused with {r.cls}") check(r.cls == v["error"], f"refused with {r.cls}, vector says {v['error']}") return check("expected" in v, f"accepted, vector says {v.get('error')}") check(v["expected"] == computed, f"expected: computed {computed}") def tally(n, reports): """(supermajority hash or None, sorted dissenters) — treatise §13.1.""" threshold = 2 * n // 3 + 1 votes = {} for h in reports.values(): votes[h] = votes.get(h, 0) + 1 winners = [h for h, c in votes.items() if c >= threshold] winner = winners[0] if winners else None dissenters = sorted(node for node, h in reports.items() if winner and h != winner) return winner, dissenters def check_state(v): """Apply each frame from the genesis state; every result, state and hash must be the vector's.""" g = v["genesis"] data = bytes.fromhex(g["cbor_hex"]) typed_matches(g, "state", data) st = decode_state(data) check(st["frame"] == 0, "a genesis state is at frame 0") check(blake3(data).hex() == g["state_hash"], "genesis state_hash") for i, f in enumerate(v["frames"]): check(f["frame"] == st["frame"], f"frame {i}: the vector's frame number is not the state's") inputs = [] for inp in f["inputs"]: b = bytes.fromhex(inp["cbor_hex"]) typed_matches(inp, "input", b) inputs.append(b) results = apply_frame(st, inputs) check(results == f["expected_results"], f"frame {i}: expected_results: computed {results}") after = encode_state(st) typed_matches(f, "expected_state", after) check(after.hex() == f["expected_state_cbor_hex"], f"frame {i}: expected_state_cbor_hex") check(blake3(after).hex() == f["expected_state_hash"], f"frame {i}: expected_state_hash") def check_lockstep(v): n, frame = v["node_count"], v["frame_number"] reported, computed = v["node_reported_hashes"], v["node_computed_hashes"] check(sorted(reported) == sorted(computed), "reported and computed name different nodes") check(0 < len(reported) <= n, "more reports than nodes") check(v["expected_threshold"] == 2 * n // 3 + 1, "expected_threshold") winner, _ = tally(n, reported) check(v["expected_commit"] == (winner is not None), "expected_commit") check(v["expected_state_hash"] == winner, "expected_state_hash") outcomes, logs = {}, {} for node, own in computed.items(): # A node's view: its peers' reports and the hash it computed (§13.4). view = dict(reported) view[node] = own best, dissent = tally(n, view) if best is None: outcomes[node] = "cluster_split" elif own != best: outcomes[node] = "minority_divergence" else: outcomes[node] = "commit_with_dissent" if dissent else "clean_commit" logs[node] = [{"frame": frame, "dissenting_peers": dissent, "local_hash": own, "supermajority_hash": best}] \ if outcomes[node] == "commit_with_dissent" else [] check(v["expected_outcomes"] == outcomes, f"expected_outcomes: computed {outcomes}") check(v["expected_dissent_logs"] == logs, f"expected_dissent_logs: computed {logs}") def check_gossip(v): check(len(v["deliveries"]) == len(v["expected_results"]), "one result per delivery") held = [] for d, want in zip(v["deliveries"], v["expected_results"]): got = receive(held, d["claimed_said"], bytes.fromhex(d["cbor_hex"])) check(got == want, f"published {want}, computed {got}") check(v["expected_held"] == held, f"expected_held: computed {held}") def self_test(): # BLAKE3 known answers: the empty input, and the official test-vector input # pattern (byte i is i % 251) at lengths that cross block and chunk edges. check(blake3(b"").hex() == "af1349b9f5f9a1a6a0404dea36dcc9499bcb25c9adc112b7cc9a93cae41f3262", "BLAKE3 of the empty input") for n, want in BLAKE3_KAT: got = blake3(bytes(i % 251 for i in range(n))).hex() check(got == want, f"BLAKE3 known answer for length {n}: {got}") # Ed25519 known answers: RFC 8032 §7.1 tests 1 and 2. s1 = bytes.fromhex("9d61b19deffd5a60ba844af492ec2cc44449c5697b326919703bac031cae7f60") check(ed25519_public(s1).hex() == "d75a980182b10ab7d54bfed3c964073a0ee172f3daa62325af021a68f707511a", "RFC 8032 test 1 key") check(ed25519_sign(s1, b"").hex() == ( "e5564300c360ac729086e2cc806e828a84877f1eb8e5d974d873e065224901555" "fb8821590a33bacc61e39701cf9b46bd25bf5f0595bbe24655141438e7a100b"), "RFC 8032 test 1 signature") s2 = bytes.fromhex("4ccd089b28ff96da9db6c346ec114e0f5b8a319f35aba624da8cf6ed4fb8a6fb") sig2 = ed25519_sign(s2, b"\x72") check(sig2.hex() == ( "92a009a9f0d4cab8720e820b5f642540a2b27b5416503f8fb3762223ebdb69da" "085ac1e43e15996e458f3613d0f11d8c387b2eaeb4302aeeb00d291612bb0c00"), "RFC 8032 test 2 signature") check(ed25519_verify(ed25519_public(s2), b"\x72", sig2) is None, "RFC 8032 test 2 verifies") check(ed25519_verify(ed25519_public(s2), b"\x73", sig2) == "signature_invalid", "a changed message fails") # BLAKE3 hashes of the official test-vector input pattern (byte i is i % 251), # hash mode, 32 bytes, at lengths that cross block, chunk and tree edges. # Computed with the BLAKE3 team's reference implementation (Rust crate # `blake3` 1.8.7). BLAKE3_KAT = [ (1, "2d3adedff11b61f14c886e35afa036736dcd87a74d27b5c1510225d0f592e213"), (63, "e9bc37a594daad83be9470df7f7b3798297c3d834ce80ba85d6e207627b7db7b"), (64, "4eed7141ea4a5cd4b788606bd23f46e212af9cacebacdc7d1f4c6dc7f2511b98"), (65, "de1e5fa0be70df6d2be8fffd0e99ceaa8eb6e8c93a63f2d8d1c30ecb6b263dee"), (1023, "10108970eeda3eb932baac1428c7a2163b0e924c9a9e25b35bba72b28f70bd11"), (1024, "42214739f095a406f3fc83deb889744ac00df831c10daa55189b5d121c855af7"), (1025, "d00278ae47eb27b34faecf67b4fe263f82d5412916c1ffd97c8cb7fb814b8444"), (2048, "e776b6028c7cd22a4d0ba182a8bf62205d2ef576467e838ed6f2529b85fba24a"), (2049, "5f4d72f40d7a5f82b15ca2b2e44b1de3c2ef86c426c95c1af0b6879522563030"), (3072, "b98cb0ff3623be03326b373de6b9095218513e64f1ee2edd2525c7ad1e5cffd2"), (3073, "7124b49501012f81cc7f11ca069ec9226cecb8a2c850cfe644e327d22d3e1cd3"), (4096, "015094013f57a5277b59d8475c0501042c0b642e531b0a1c8f58d2163229e969"), (4097, "9b4052b38f1c5fc8b1f9ff7ac7b27cd242487b3d890d15c96a1c25b8aa0fb995"), (5120, "9cadc15fed8b5d854562b26a9536d9707cadeda9b143978f319ab34230535833"), (8193, "bab6c09cb8ce8cf459261398d2e7aef35700bf488116ceb94a36d0f5f1b7bc3b"), (31744, "62b6960e1a44bcc1eb1a611a8d6235b6b4b78f32e7abc4fb4c6cdcce94895c47"), ] # In order: the envelope vectors draw key states from the accepted logs. CHECKS = { "canonicalisation_vectors.json": check_canonicalisation, "said_vectors.json": check_said, "signature_vectors.json": check_signature, "kel_vectors.json": check_kel, "envelope_vectors.json": check_envelope, "state_vectors.json": check_state, "lockstep_vectors.json": check_lockstep, "gossip_vectors.json": check_gossip, } HERE = pathlib.Path(__file__).resolve().parent def spec_cited_said_vectors(): """The SAID vector prefixes the identity spec cites, ranges expanded.""" spec = (HERE.parent.parent / "spec" / "identity_and_key_rotation.md").read_text(encoding="utf-8") words = [w for w in re.split(r"[^A-Za-z0-9-]+", spec) if w] cited = set() for i, w in enumerate(words): m = re.fullmatch(r"said-(\d+)", w) if not m: continue cited.add(f"said-{int(m.group(1)):03d}") end = re.fullmatch(r"said-(\d+)", words[i + 2]) if i + 2 < len(words) and words[i + 1] == "to" else None if end: cited.update(f"said-{k:03d}" for k in range(int(m.group(1)), int(end.group(1)) + 1)) return cited def readme_error_classes(): readme = (HERE.parent / "README.md").read_text(encoding="utf-8") section = readme.split("### Error classes", 1)[1] rows = [] for line in section.splitlines(): if line.startswith("|"): rows.append(line) elif rows: break return {line.split("`")[1] for line in rows if line.count("`") >= 2} def check_complete(files): """Every file non-empty; every vector the spec cites present; every README error class exercised, and no other class used.""" for name, vectors in files.items(): check(len(vectors) > 0, f"{name} holds no vectors") names = [v["name"] for v in files["said_vectors.json"]] missing = sorted(p for p in spec_cited_said_vectors() if not any(n.startswith(p + "-") for n in names)) check(not missing, f"the identity spec cites vectors the pack lacks: {missing}") used = set() for vectors in files.values(): for v in vectors: if isinstance(v.get("error"), str): used.add(v["error"]) for holder in [v] + v.get("frames", []): for r in holder.get("expected_results", []): if "error" in r: used.add(r["error"]) listed = readme_error_classes() check(listed and used == listed, f"README classes unexercised: {sorted(listed - used)}; unlisted classes used: {sorted(used - listed)}") def main(pack): self_test() failed = total = 0 files = {} for name in CHECKS: try: data = load_json(pack / name) check(isinstance(data, list), "the top-level value must be an array") names = [v["name"] for v in data] check(len(set(names)) == len(names), "duplicate vector names") check(all(isinstance(v.get("description"), str) for v in data), "a vector lacks a description") files[name] = data except (Mismatch, ValueError, KeyError, OSError) as e: failed += 1 print(f"FAIL {name}: {e}") logs = {} for name, vectors in files.items(): for v in vectors: total += 1 try: if name in ("kel_vectors.json", "envelope_vectors.json"): CHECKS[name](v, logs) else: CHECKS[name](v) print(f"ok {name:<32} {v['name']}") except (Mismatch, Refused, LogRefused, KeyError, ValueError, TypeError) as e: failed += 1 print(f"FAIL {name:<32} {v.get('name')}: {e!r}") if len(files) == len(CHECKS): try: check_complete(files) print("ok (pack) complete") except Mismatch as e: failed += 1 print(f"FAIL (pack): {e}") print(f"{total} vectors, {failed} failed") return 1 if failed else 0 if __name__ == "__main__": default = HERE.parent / "v1" sys.exit(main(pathlib.Path(sys.argv[1]) if len(sys.argv) > 1 else default))