feat(protocol): define gateway control envelopes

tools/validate_gateway_envelopes.py

@@ -0,0 +1,95 @@
+#!/usr/bin/env python3
+"""Validate fixed-byte Phase 3C gateway input and feedback envelopes."""
+
+from __future__ import annotations
+
+import binascii
+import pathlib
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+
+
+def classify_input(raw: bytes) -> str:
+    if len(raw) < 6:
+        return "invalid:truncated"
+    if raw[:4] != b"VGI1":
+        return "invalid:magic"
+    kind, length = raw[4], raw[5]
+    if len(raw) != 6 + length:
+        return "invalid:length"
+    body = raw[6:]
+    if kind == 1:
+        return "valid" if len(body) == 4 and body[0] <= 1 and body[2:4] != b"\0\0" else "invalid:field"
+    if kind == 2:
+        return "valid" if len(body) == 3 and body[0] <= 1 and 1 <= body[1] <= 5 and body[2] == 0 else "invalid:reserved"
+    if kind == 3:
+        return "valid" if len(body) == 4 else "invalid:length"
+    if kind == 4:
+        try:
+            decoded = body.decode("utf-8")
+        except UnicodeDecodeError:
+            return "invalid:utf8"
+        return "valid" if 1 <= len(body) <= 4 and len(decoded) == 1 else "invalid:utf8"
+    if kind == 5:
+        if len(body) != 17:
+            return "invalid:length"
+        if body[0] > 15:
+            return "invalid:field"
+        active_mask = int.from_bytes(body[1:3], "big")
+        return "valid" if active_mask or not any(body[3:]) else "invalid:field"
+    return "invalid:kind"
+
+
+def classify_feedback(raw: bytes) -> str:
+    if len(raw) < 8:
+        return "invalid:truncated"
+    if raw[:4] != b"VGF1":
+        return "invalid:magic"
+    direction, kind = raw[4], raw[5]
+    length = int.from_bytes(raw[6:8], "big")
+    if len(raw) != 8 + length:
+        return "invalid:length"
+    if direction not in (0, 1):
+        return "invalid:direction"
+    body = raw[8:]
+    if direction == 0:
+        if kind in (0x10, 0x11, 0x12):
+            return "invalid:direction"
+        if kind == 1:
+            return "valid" if not body else "invalid:length"
+        if kind == 2:
+            return "valid" if valid_fec_status(body) else "invalid:field"
+        return "invalid:type"
+    if kind in (1, 2):
+        return "invalid:direction"
+    if kind == 0x10:
+        return "valid" if len(body) == 4 else "invalid:length"
+    if kind == 0x11:
+        return "valid" if len(body) == 5 and body[0] <= 15 else "invalid:field"
+    if kind == 0x12:
+        return "valid" if len(body) == 1 and body[0] <= 1 else "invalid:field"
+    return "invalid:type"
+
+
+def valid_fec_status(body: bytes) -> bool:
+    return len(body) == 21 and int.from_bytes(body[10:12], "big") > 0 and int.from_bytes(body[14:16], "big") <= int.from_bytes(body[10:12], "big") and int.from_bytes(body[16:18], "big") <= int.from_bytes(body[12:14], "big") and body[18] <= 100 and body[20] > 0 and body[19] < body[20]
+
+
+def main() -> None:
+    lines = (ROOT / "fixtures/conformance/gateway-input-feedback-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 version == "1"
+        try:
+            raw = binascii.unhexlify(input_value.removeprefix("hex="))
+        except binascii.Error:
+            actual = "invalid:hex"
+        else:
+            actual = classify_input(raw) if kind == "gateway_input" else classify_feedback(raw)
+        assert actual == expected, f"{identifier}: {actual} != {expected}"
+    print("Gateway input/feedback envelope validation passed")
+
+
+if __name__ == "__main__":
+    main()