98 lines
3.5 KiB
Python
98 lines
3.5 KiB
Python
#!/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"
|
|
if kind == 3:
|
|
return "valid" if not body else "invalid:length"
|
|
return "invalid:type"
|
|
if kind in (1, 2, 3):
|
|
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()
|