#!/usr/bin/env python3 """weft_verify.py: the second implementation of the `wai.tensor.weft` corpus (SPEC §5, *WFT1 tensor frames*). Written from the format's text (WEFTCODEC-FORMAT v1.0.0 §1 to §11 and WEFT-AI-CODEC v1.3.0 §2 and §5, as published with weftcodec 1.2.0) and SPEC §5, sharing no code with wai-rs: a decoder of the frames the registration admits (RAW, TENSOR, FLOAT_TENSOR, none naming a dependency), its CRC-32C and its domain hash over BLAKE3 (`claim-conformance/_blake3.py`, itself written from BLAKE3's specification). weft_verify.py It requires, of the corpus: 1. every vector's frame has the SHA-256 vectors.json states; 2. every vector gets its stated outcome: its plaintext's length and SHA-256, or the format's error class it is refused with, under the limits it names; 3. the classes vectors.json lists are the classes SPEC §5 lists, and its admitted modes are the registration's; 4. every frame it decodes is refused once any one of its bytes is changed; 5. every check of this verifier refuses some vector, and every check is needed: with it dropped, some vector gets another outcome. The exceptions are the checks IMPLIED names, each of which another check always backs with the same class; for those, dropping the check alone changes nothing, and dropping it with the check that backs it changes some vector's outcome. So the corpus exercises every rule this verifier reads from the text. Exit 0 when all hold, 1 with one line a finding otherwise. Standard library only, run isolated (-I). """ import hashlib import json import os import re import struct import sys HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, os.path.join(HERE, "..", "claim-conformance")) from _blake3 import blake3 # noqa: E402 ADMITTED = (0, 6, 9) CLASSES = ["Malformed", "Corrupt", "UnsupportedVersion", "UnsupportedMode", "UnsupportedFeature", "LimitExceeded", "NonCanonical"] DEFAULTS = { "max_frame_bytes": 1 << 30, "max_decoded_bytes": 1 << 32, "max_dependency_count": 64, "max_chunks": 1 << 20, "max_structured_depth": 64, "max_structured_items": 1_000_000, "max_string_bytes": 256 << 20, "max_tensor_elements": 1 << 32, "max_tensor_rank": 8, } TENSOR_SIZES = {"f16": 2, "bf16": 2, "f32": 4, "f64": 8, "i8": 1, "u8": 1, "i16": 2, "i32": 4, "i64": 8} FLOAT_WIDTHS = {"bf16": (1, 8, 7), "fp16": (1, 5, 10), "fp32": (1, 8, 23), "fp8-e4m3": (1, 4, 3), "fp8-e5m2": (1, 5, 2)} # The checks this verifier makes, by name. A name in DROPPED is skipped (requirement 5). DROPPED = set() CHECKS = [] # The checks another check always backs, with the class it gives, for every frame of an admitted # mode, and the check that backs each. They stay, so that a frame with several defects is refused # in the format's order; requirement 5 drops each together with the check named here. IMPLIED = { # A domain above 1: the admitted modes all take domain 0. "frame:domain": "mode:payload-domain", # A dependency table: no admitted mode names a dependency. "frame:deps-feature": "mode:no-dependency", "frame:dependency-entry": "mode:no-dependency", # A RAW frame's chunk extents end at the body's end exactly when its chunk lengths sum to # decoded_len, given body_len == decoded_len and each chunk's two lengths equal. "chunk:extent": "chunk:sum", "chunk:sum": "chunk:extent", } class Refuse(Exception): def __init__(self, cls, why, check=None): super().__init__(f"{cls}: {why}") self.cls = cls self.check = check def check(name, bad, cls, why): """Refuse with `cls` when `bad`, unless the check `name` is dropped.""" if name not in CHECKS: CHECKS.append(name) if bad and name not in DROPPED: raise Refuse(cls, why, name) _H = {} def H(domain, data): key = (domain, bytes(data)) if key not in _H: d = domain.encode() _H[key] = blake3(struct.pack("> 1) ^ 0x82F63B78 if c & 1 else c >> 1 _CRC.append(c) c = 0xFFFFFFFF for b in data: c = _CRC[(c ^ b) & 0xFF] ^ (c >> 8) return c ^ 0xFFFFFFFF def u16(b, o): return struct.unpack_from(" lim, "LimitExceeded", f"{name} {value} > {lim}") # ---- WCB (WEFTCODEC-FORMAT §8) ---- class Wcb: def __init__(self, b, lim): self.b, self.pos, self.items, self.lim = b, 0, 0, lim def varint(self): value, shift, count = 0, 0, 0 while True: if self.pos >= len(self.b): raise Refuse("Malformed", "truncated varint") byte = self.b[self.pos] self.pos += 1 count += 1 if count > 10 or (count == 10 and byte > 1): raise Refuse("Malformed", "varint past u64") value |= (byte & 0x7F) << shift if not byte & 0x80: check("wcb:minimal-varint", count > 1 and byte == 0, "NonCanonical", "non-minimal varint") return value shift += 7 def take(self, n, what): limit("max_string_bytes", n, self.lim["max_string_bytes"]) if n > len(self.b) - self.pos: raise Refuse("Malformed", f"truncated {what}") s = self.b[self.pos:self.pos + n] self.pos += n return s def value(self, depth): self.items += 1 limit("max_structured_items", self.items, self.lim["max_structured_items"]) if self.pos >= len(self.b): raise Refuse("Malformed", "truncated value") tag = self.b[self.pos] self.pos += 1 if tag == 0: return None if tag in (1, 2): return tag == 2 if tag == 3: return ("u64", self.varint()) if tag in (4, 9): z = self.varint() return ("i64" if tag == 4 else "ts", (z >> 1) ^ -(z & 1)) if tag == 5: s = self.take(self.varint(), "text") try: return s.decode("utf-8") except UnicodeDecodeError: raise Refuse("NonCanonical", "text not UTF-8") if tag == 6: return bytes(self.take(self.varint(), "bytes")) if tag in (7, 8): limit("max_structured_depth", depth + 1, self.lim["max_structured_depth"]) n = self.varint() limit("max_structured_items", n, self.lim["max_structured_items"]) if n * (1 if tag == 7 else 2) > len(self.b) - self.pos: raise Refuse("Malformed", "truncated container") if tag == 7: return [self.value(depth + 1) for _ in range(n)] out, prev = {}, None for _ in range(n): start = self.pos if self.pos >= len(self.b): raise Refuse("Malformed", "truncated key") kt = self.b[self.pos] self.pos += 1 check("wcb:text-keys", kt != 5, "NonCanonical", "a key that is not text") kb = self.take(self.varint(), "key") enc = bytes(self.b[start:self.pos]) check("wcb:key-order", prev is not None and enc <= prev, "NonCanonical", "keys not in encoded order") prev = enc try: k = kb.decode("utf-8") except UnicodeDecodeError: raise Refuse("NonCanonical", "key not UTF-8") out[k] = self.value(depth + 1) return out raise Refuse("NonCanonical", f"tag {tag:#x}") def wcb(b, lim): if not b: raise Refuse("Malformed", "empty WCB") r = Wcb(b, lim) v = r.value(0) check("wcb:no-trailing", r.pos != len(b), "NonCanonical", "bytes after the value") return v def is_u64(v): return isinstance(v, tuple) and v[0] == "u64" def need_text(m, k): v = m.get(k) if not isinstance(v, str): raise Refuse("Malformed", f"{k} not text") return v def need_u64(m, k): v = m.get(k) if not is_u64(v): raise Refuse("Malformed", f"{k} not u64") return v[1] def need_shape(m): v = m.get("shape") if not isinstance(v, list) or not all(is_u64(x) for x in v): raise Refuse("Malformed", "shape not an array of u64") return [x[1] for x in v] def need_hash(m): v = m.get("payload_hash") if not isinstance(v, bytes): raise Refuse("Malformed", "payload_hash not bytes") check("tensor:hash-length", len(v) != 32, "Malformed", "payload_hash not 32 bytes") return v def keys(m, want): check("desc:keys", set(m) != set(want), "Malformed", "a key the schema lacks, or one it has missing") def descriptor(body, lim): meta_len = u32(body, 0) if 4 + meta_len > len(body): raise Refuse("Malformed", "descriptor past the body") m = wcb(bytes(body[4:4 + meta_len]), lim) if not isinstance(m, dict): raise Refuse("Malformed", "descriptor not a map") return m, body[4 + meta_len:] def tensor(body, decoded_len, lim): m, payload = descriptor(body, lim) check("tensor:payload-length", len(payload) != decoded_len, "Malformed", "payload length") keys(m, ("schema", "dtype", "shape", "layout", "endianness", "element_count", "payload_bytes", "payload_hash")) check("tensor:schema", need_text(m, "schema") != "weft.tensor.v1", "Malformed", "schema") dtype = need_text(m, "dtype") check("tensor:dtype", dtype not in TENSOR_SIZES, "Malformed", "dtype") check("tensor:layout", need_text(m, "layout") != "row_major", "Malformed", "layout") check("tensor:endianness", need_text(m, "endianness") != "little", "Malformed", "endianness") shape = need_shape(m) count, nbytes = need_u64(m, "element_count"), need_u64(m, "payload_bytes") limit("max_tensor_rank", len(shape), lim["max_tensor_rank"]) # A product past u64 is past every element limit a u64 can state: LimitExceeded either way. p = 1 for d in shape: p *= d limit("max_tensor_elements", p, lim["max_tensor_elements"]) check("tensor:element-count", count != p, "Malformed", "element_count") check("tensor:payload-bytes", nbytes != count * TENSOR_SIZES.get(dtype, 1), "Malformed", "payload_bytes") check("tensor:payload-bytes-length", nbytes != len(payload), "Malformed", "payload length") check("tensor:payload-hash", need_hash(m) != H("weft/tensor/payload/v1", payload), "Corrupt", "payload_hash") return bytes(payload) def huffman(coded, n): check("huffman:table", len(coded) < 256, "Corrupt", "shorter than its table") lengths, stream = coded[:256], coded[256:] check("huffman:max-length", any(l > 24 for l in lengths), "NonCanonical", "a code over 24 bits") used = [(l, s) for s, l in enumerate(lengths) if l] if not used: check("huffman:empty", n != 0, "Corrupt", "an empty table for a plane") check("huffman:trailing", len(stream) != 0, "Corrupt", "bytes after the table") return b"" if len(used) == 1: check("huffman:single", used[0][0] != 1 or n == 0, "NonCanonical", "single-symbol form") check("huffman:trailing", len(stream) != 0, "Corrupt", "bytes after the table") return bytes([used[0][1]]) * n if n == 0: # a FLOAT_TENSOR plane is never empty: every extent is above zero raise Refuse("NonCanonical", "codes for an empty plane") kraft = sum(1 << (24 - l) for l, _ in used) check("huffman:kraft", kraft != 1 << 24, "NonCanonical", "over-subscribed or incomplete") table, code, prev = {}, 0, 0 for l, s in sorted(used): code <<= l - prev table[(l, code)] = s code += 1 prev = l maxlen = max(l for l, _ in used) out, bit = bytearray(), 0 while len(out) < n: c, l = 0, 0 while True: if l and (l, c) in table: out.append(table[(l, c)]) break if l >= maxlen: raise Refuse("Corrupt", "an invalid code") if bit >= 8 * len(stream): raise Refuse("Corrupt", "the bitstream ends early") c = (c << 1) | ((stream[bit // 8] >> (7 - bit % 8)) & 1) l += 1 bit += 1 used_bytes = (bit + 7) // 8 if bit % 8: check("huffman:padding", stream[used_bytes - 1] & ((1 << (8 - bit % 8)) - 1), "Corrupt", "padding bits") check("huffman:trailing", used_bytes != len(stream), "Corrupt", "bytes after the last code") return bytes(out) def float_tensor(body, decoded_len, lim): m, stored = descriptor(body, lim) keys(m, ("schema", "dtype", "shape", "element_count", "sign_bits", "exponent_bits", "mantissa_bits", "exponent_coder", "exponent_plane_len", "mantissa_plane_len", "payload_hash")) check("float:schema", need_text(m, "schema") != "weft.ai.float_tensor.v1", "Malformed", "schema") dtype = need_text(m, "dtype") check("float:coder", need_text(m, "exponent_coder") != "huffman-canonical-v1", "Malformed", "coder") widths = tuple(need_u64(m, k) for k in ("sign_bits", "exponent_bits", "mantissa_bits")) # One rule: the widths are the dtype's, so an unknown dtype (none) and a width past a byte # (no dtype's) are refused by it. check("float:widths", widths != FLOAT_WIDTHS.get(dtype), "Malformed", "an unknown dtype, or widths not its") s, e, mbits = FLOAT_WIDTHS.get(dtype, (1, 8, 7)) size = (s + e + mbits) // 8 shape = need_shape(m) count = need_u64(m, "element_count") check("float:rank-zero", not shape, "Malformed", "rank 0") limit("max_tensor_rank", len(shape), lim["max_tensor_rank"]) p = 1 for d in shape: check("float:zero-extent", d == 0, "Malformed", "an extent of 0") p *= d limit("max_tensor_elements", p, lim["max_tensor_elements"]) check("float:element-count", count != p, "Malformed", "element_count") out_len = count * size if out_len >= 1 << 64: # unreachable within the element limit; the text requires it raise Refuse("LimitExceeded", "decoded length overflows") limit("max_decoded_bytes", out_len, lim["max_decoded_bytes"]) check("float:decoded-len", out_len != decoded_len, "Corrupt", "decoded_len") mlen = (count * (s + mbits) + 7) // 8 check("float:mantissa-len", need_u64(m, "mantissa_plane_len") != mlen, "Malformed", "mantissa_plane_len") elen = need_u64(m, "exponent_plane_len") check("float:planes-overflow", elen + mlen >= 1 << 64, "LimitExceeded", "plane lengths overflow") check("float:stored-len", elen + mlen != len(stored), "Malformed", "stored length") exps = huffman(bytes(stored[:elen]), count) plane = stored[elen:] out, bit, w = bytearray(), 0, s + mbits for x in exps: check("float:exponent-range", x >> e, "Corrupt", "an exponent past its bits") v = 0 for _ in range(w): v = (v << 1) | ((plane[bit // 8] >> (7 - bit % 8)) & 1) bit += 1 word = ((v >> mbits) << (e + mbits)) | (x << mbits) | (v & ((1 << mbits) - 1)) out += word.to_bytes(size, "little") if bit % 8: check("float:padding", plane[-1] & ((1 << (8 - bit % 8)) - 1), "Corrupt", "mantissa padding") check("float:payload-hash", need_hash(m) != H("weft/float-tensor/payload/v1", stored), "Corrupt", "payload_hash") return bytes(out) def decode(frame, lim): """The plaintext of an admitted frame, or Refuse with the format's class.""" check("frame:min-length", len(frame) < 100, "Malformed", "below 100 bytes") limit("max_frame_bytes", len(frame), lim["max_frame_bytes"]) check("frame:magic", frame[:4] != b"WFT1", "Malformed", "magic") check("frame:version", frame[4] != 1 or frame[5] != 0, "UnsupportedVersion", "version") check("frame:header-crc", crc32c(frame[:72] + frame[76:96]) != u32(frame, 72), "Corrupt", "header CRC") domain, mode = frame[6], frame[7] check("frame:mode", mode not in ADMITTED, "UnsupportedMode", f"mode {mode}") flags, decoded_len, body_len = u16(frame, 8), u64(frame, 12), u64(frame, 20) deps, chunks, features, table_len = u32(frame, 28), u32(frame, 32), u16(frame, 36), u64(frame, 76) check("frame:reserved", u16(frame, 38) or u32(frame, 84) or u64(frame, 88), "Malformed", "reserved") check("frame:domain", domain > 1, "Malformed", "domain") check("frame:flags", flags & ~0xF, "UnsupportedFeature", "unknown flags") check("frame:original-hash", flags & 8, "UnsupportedFeature", "ORIGINAL_HASH") check("frame:self-contained", flags & 2, "UnsupportedFeature", "SELF_CONTAINED") check("frame:features", features & ~3, "UnsupportedFeature", "unknown features") limit("max_decoded_bytes", decoded_len, lim["max_decoded_bytes"]) limit("max_dependency_count", deps, lim["max_dependency_count"]) limit("max_chunks", chunks, lim["max_chunks"]) check("frame:table-len", table_len != 48 * (deps + chunks), "Malformed", "table_len") chunked = bool(flags & 1) check("frame:chunked-nonempty", chunked and chunks == 0, "Malformed", "CHUNKED with no chunks") check("frame:chunk-hashes", chunked and not features & 2, "Malformed", "CHUNKED without CHUNK_HASHES") check("frame:chunks-need-flag", not chunked and chunks != 0, "Malformed", "chunks without CHUNKED") check("frame:hashes-need-flag", not chunked and features & 2, "Malformed", "CHUNK_HASHES without CHUNKED") check("frame:deps-feature", deps and not features & 1, "Malformed", "dependencies without DEPENDENCIES") check("frame:total-length", 96 + table_len + body_len + 4 != len(frame), "Malformed", "frame length") check("frame:trailer-crc", crc32c(frame[:-4]) != u32(frame, len(frame) - 4), "Corrupt", "trailer CRC") body = frame[96 + table_len:len(frame) - 4] for i in range(deps): o = 96 + 48 * i check("frame:dependency-entry", frame[o + 41] != 1 or u16(frame, o + 42) or frame[o + 40] > 2, "Malformed", "a dependency entry") limit("max_decoded_bytes", u64(frame, o + 32), lim["max_decoded_bytes"]) pieces = [] if chunked: nxt, total = 0, 0 for i in range(chunks): o = 96 + 48 * (deps + i) off, enc, dec = u64(frame, o), u32(frame, o + 8), u32(frame, o + 12) check("chunk:contiguous", off != nxt, "Malformed", "a gap or an overlap") check("chunk:nonempty", enc == 0, "Malformed", "an empty chunk") nxt, total = nxt + enc, total + dec pieces.append((off, enc, dec, frame[o + 16:o + 48])) check("chunk:extent", nxt != body_len, "Malformed", "extents do not end at the body's end") check("chunk:sum", total != decoded_len, "Malformed", "lengths do not sum") check("mode:payload-domain", domain != 0, "Malformed", "the structured domain") check("mode:dict-raw", flags & 4, "Malformed", "DICT_RAW") check("mode:no-dependency", deps != 0, "Malformed", "a dependency") if mode == 0: check("raw:body-length", body_len != decoded_len, "Malformed", "RAW body_len") if chunked: plain = b"" for i, (off, enc, dec, hsh) in enumerate(pieces): check("raw:chunk-length", enc != dec, "Malformed", "a RAW chunk's lengths") piece = body[off:off + enc] check("raw:chunk-hash", H("weft/chunk/v1", piece) != hsh, "Corrupt", f"chunk {i} hash") plain += piece else: plain = bytes(body) else: check("tensor:not-chunked", chunks != 0, "Malformed", "a chunked tensor") check("tensor:body-4", body_len < 4, "Malformed", "a body under 4 bytes") plain = tensor(body, decoded_len, lim) if mode == 6 else float_tensor(body, decoded_len, lim) check("frame:content-hash", H("weft/chunk/v1", plain) != frame[40:72], "Corrupt", "content hash") return plain def outcome(frame, lim): try: p = decode(frame, lim) return {"decode": {"decoded_len": len(p), "sha256": hashlib.sha256(p).hexdigest()}} except Refuse as r: return {"refuse": r.cls} except (IndexError, ValueError, struct.error): # A dropped check lets a short frame through to a read past its end, or a code longer # than the coder's 24 bits to a negative shift. return {"refuse": "crash"} def run(root, doc): """(name, got, want) for every vector whose outcome is not the stated one.""" bad = [] for v in doc["vectors"]: frame = frames[v["name"]] lim = dict(DEFAULTS, **v["limits"]) got = outcome(frame, lim) if got != v["expect"]: bad.append((v["name"], got, v["expect"])) return bad frames = {} def main(): root = sys.argv[1] fails = [] with open(os.path.join(root, "vectors.json"), encoding="utf-8") as fh: doc = json.load(fh) for v in doc["vectors"]: with open(os.path.join(root, v["file"]), "rb") as fh: frames[v["name"]] = fh.read() if hashlib.sha256(frames[v["name"]]).hexdigest() != v["frame_sha256"]: fails.append(f"{v['name']}: the frame is not the one its digest names") # 3. The classes and modes SPEC §5 states. with open(os.path.join(root, "..", "SPEC.md"), encoding="utf-8") as fh: spec = fh.read() m = re.search(r"\*\*WFT1 tensor frames\.\*\*(.*?)\n\n", spec, re.S) if not m: fails.append("SPEC.md has no paragraph *WFT1 tensor frames*") else: stated = re.findall(r"`(Malformed|Corrupt|Unsupported\w+|LimitExceeded|NonCanonical)`", m.group(1)) if sorted(set(stated)) != sorted(CLASSES): fails.append(f"SPEC §5 lists the classes {sorted(set(stated))}, not {sorted(CLASSES)}") if doc.get("classes") != CLASSES: fails.append(f"vectors.json lists the classes {doc.get('classes')}") if doc.get("admitted_modes") != list(ADMITTED): fails.append(f"vectors.json admits the modes {doc.get('admitted_modes')}") # 2. Every outcome. for name, got, want in run(root, doc): fails.append(f"{name}: {got}, the corpus states {want}") print(f"{'PASS' if not fails else 'FAIL'} {len(doc['vectors'])} vectors, each with its stated outcome") # 4. A changed byte of a decoded frame is refused. changed = 0 for v in doc["vectors"]: if "decode" not in v["expect"]: continue frame = frames[v["name"]] lim = dict(DEFAULTS, **v["limits"]) for pos in sorted({0, 7, 12, 40, 72, 96, len(frame) // 2, len(frame) - 5, len(frame) - 1}): f = bytearray(frame) f[pos] ^= 0x01 changed += 1 if "decode" in outcome(bytes(f), lim): fails.append(f"{v['name']}: a change of byte {pos} still decodes") print(f"PASS {changed} changed frames, each refused" if not any("still decodes" in f for f in fails) else f"FAIL changed frames that still decode") # 5. Every check refuses some vector, and is needed: alone, or with the check that backs it. with open(os.path.abspath(__file__), encoding="utf-8") as fh: declared = {n for n in re.findall(r'\bcheck\("([\w:-]+)"', fh.read()) if not n.endswith(":")} fired = set() for v in doc["vectors"]: try: decode(frames[v["name"]], dict(DEFAULTS, **v["limits"])) except Refuse as r: fired.add(r.check) found = [f"the check {n} is reached by no vector" for n in sorted(declared - set(CHECKS))] found += [f"the check {n} refuses no vector" for n in CHECKS if n not in fired] found += [f"IMPLIED names {n} or {b}, which this verifier does not check" for n, b in IMPLIED.items() if n not in CHECKS or b not in CHECKS] for name in list(CHECKS): DROPPED.clear() DROPPED.add(name) alone = bool(run(root, doc)) if name in IMPLIED: DROPPED.add(IMPLIED[name]) if alone: found.append(f"IMPLIED names {name}, which some vector needs alone") if not run(root, doc): found.append(f"dropping {name} with {IMPLIED[name]}, which backs it, changes no vector's outcome") elif not alone: found.append(f"dropping the check {name} changes no vector's outcome") DROPPED.clear() fails += found print(f"{'PASS' if not found else 'FAIL'} {len(CHECKS)} checks, each refusing some vector and needed: " f"{len(CHECKS) - len(IMPLIED)} alone, {len(IMPLIED)} with the check that backs it") for f in fails: print("FAIL", f) return 1 if fails else 0 if __name__ == "__main__": sys.exit(main())