#!/usr/bin/env python3 """Validate the bounded Phase 3A datagram header and fixture corpus.""" from __future__ import annotations import binascii import pathlib ROOT = pathlib.Path(__file__).resolve().parents[1] HEADER_BYTES = 21 MAX_FRAME_BYTES = 65536 CHANNEL_LIMITS = {1: 1024, 2: 2048, 3: 65515, 10: 1179, 11: 1179, 12: 1179} def classify(raw: bytes) -> str: if len(raw) < HEADER_BYTES: return "invalid:truncated" if raw[:2] != b"VD": return "invalid:magic" if raw[2] != 1: return "invalid:unsupported_version" if raw[3] not in CHANNEL_LIMITS: return "invalid:unknown_channel" if raw[4] != 0: return "invalid:flags" fragment_index, fragment_count = raw[17], raw[18] if fragment_count == 0 or fragment_index >= fragment_count: return "invalid:fragment" payload_length = int.from_bytes(raw[19:21], "big") if payload_length > CHANNEL_LIMITS[raw[3]]: return "invalid:payload_limit" if len(raw) != HEADER_BYTES + payload_length: return "invalid:length_mismatch" if len(raw) > MAX_FRAME_BYTES: return "invalid:frame_limit" return "valid" def main() -> None: lines = (ROOT / "fixtures/conformance/datagram-v1.tsv").read_text(encoding="utf-8").splitlines() assert lines[0] == "id\tversion\tkind\tinput\texpected" for line in lines[1:]: identifier, version, kind, input_value, expected = line.split("\t") assert kind == "datagram" and version == "1" encoded = input_value.removeprefix("hex=") try: actual = classify(binascii.unhexlify(encoded)) except binascii.Error: actual = "invalid:hex" assert actual == expected, f"{identifier}: {actual} != {expected}" print("Datagram frame validation passed") if __name__ == "__main__": main()