385 lines
18 KiB
Python
385 lines
18 KiB
Python
"""Offline V1 wire/identity tests: synthetic devices, no real sockets or upgrades."""
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import hashlib
|
|
import importlib.util
|
|
import io
|
|
import ipaddress
|
|
import json
|
|
from pathlib import Path
|
|
import struct
|
|
import tempfile
|
|
from types import SimpleNamespace
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
SN = "SYNTHETIC-V1-FULL-SN-0001"
|
|
MODEL = "SYNTHETIC-V1-MODEL"
|
|
IP = "192.0.2.10"
|
|
TARGET = "9.8.7"
|
|
INFO = 0x5A0E2104
|
|
REQUEST = 0x5A0EFF00
|
|
RESPONSE = 0x5A0EFF01
|
|
SESSION = 1234
|
|
|
|
|
|
def wire(command, body):
|
|
data = json.dumps(body, separators=(",", ":")).encode("utf-8")
|
|
return struct.pack("!II", command, len(data)) + data
|
|
|
|
|
|
def info_body(**changes):
|
|
body = dict(code=0, desc="successed", sn=SN, dev_type=MODEL,
|
|
firmware=TARGET, dev_ver=TARGET, hardware="HW-TEST")
|
|
body.update(changes)
|
|
return body
|
|
|
|
|
|
def discovery_body(**changes):
|
|
body = dict(session=SESSION, sn=SN, dev_type=MODEL, ip=IP,
|
|
mac="00:00:5E:00:53:01", firmware=TARGET, hardware="HW-TEST")
|
|
body.update(changes)
|
|
return body
|
|
|
|
|
|
def interface(address="192.0.2.20/24"):
|
|
parsed = ipaddress.ip_interface(address)
|
|
return SimpleNamespace(address=parsed.ip, prefix_length=parsed.network.prefixlen,
|
|
network=parsed.network)
|
|
|
|
|
|
class TcpSocket:
|
|
def __init__(self, packet):
|
|
self.pending = bytearray(packet)
|
|
self.sent = []
|
|
self.timeouts = []
|
|
self.closed = False
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
self.closed = True
|
|
|
|
def settimeout(self, value):
|
|
self.timeouts.append(value)
|
|
|
|
def sendall(self, packet):
|
|
self.sent.append(packet)
|
|
|
|
def recv(self, count):
|
|
count = min(count, 3, len(self.pending))
|
|
result = bytes(self.pending[:count])
|
|
del self.pending[:count]
|
|
return result
|
|
|
|
|
|
class UdpSocket:
|
|
def __init__(self, replies=(), bind_failure=False, send_failure=False):
|
|
self.replies = list(replies)
|
|
self.bind_failure = bind_failure
|
|
self.send_failure = send_failure
|
|
self.sent, self.bindings, self.options = [], [], []
|
|
self.closed = False
|
|
|
|
def setsockopt(self, *args):
|
|
self.options.append(args)
|
|
|
|
def bind(self, address):
|
|
self.bindings.append(address)
|
|
if self.bind_failure:
|
|
raise OSError("synthetic occupied port")
|
|
|
|
def setblocking(self, value):
|
|
assert value is False
|
|
|
|
def sendto(self, packet, destination):
|
|
self.sent.append((packet, destination))
|
|
if self.send_failure:
|
|
raise OSError("synthetic route failure")
|
|
return len(packet)
|
|
|
|
def recvfrom(self, _size):
|
|
return self.replies.pop(0)
|
|
|
|
def select(self, readable, *_args):
|
|
assert readable == [self]
|
|
return ([self] if self.replies else [], [], [])
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
|
|
class V1ProtocolTests(unittest.TestCase):
|
|
def setUp(self):
|
|
spec = importlib.util.spec_from_file_location("adapter_v1_test", ROOT / "scripts" / "device_adapter.py")
|
|
self.adapter = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(self.adapter)
|
|
self.v1 = self.adapter.v1
|
|
self.logger = self.adapter.EventCollector()
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.rom = Path(self.temp.name) / "synthetic-v1.rom"
|
|
self.rom.write_bytes(b"synthetic v1 ROM fixture" * 20)
|
|
self.sha = hashlib.sha256(self.rom.read_bytes()).hexdigest()
|
|
|
|
def tcp_query(self, body=None, packet=None):
|
|
sock = TcpSocket(packet if packet is not None else wire(INFO, body or info_body()))
|
|
with patch.object(self.v1.socket, "create_connection", return_value=sock) as connect:
|
|
result = self.v1.query_device_info(IP, 5, self.logger)
|
|
self.assertEqual(connect.call_count, 1)
|
|
self.assertEqual(connect.call_args.args[0], (IP, 4941))
|
|
return result, sock
|
|
|
|
def cli(self, args, body=None):
|
|
sock = TcpSocket(wire(INFO, body or info_body()))
|
|
stdout = io.StringIO()
|
|
with patch.object(self.v1.socket, "create_connection", return_value=sock) as connect, \
|
|
patch.object(self.adapter.protocol, "query_device_info", side_effect=AssertionError("V2 forbidden")), \
|
|
patch.object(self.adapter.protocol, "transfer_rom", side_effect=AssertionError("transfer forbidden")), \
|
|
contextlib.redirect_stdout(stdout):
|
|
code = self.adapter.main(args + ["--protocol", "v1"])
|
|
return code, json.loads(stdout.getvalue()), sock, connect
|
|
|
|
def identity(self):
|
|
return ["--expected-sn", SN, "--expected-model", MODEL]
|
|
|
|
def discover(self, replies=(), **kwargs):
|
|
sock = UdpSocket(replies, kwargs.pop("bind_failure", False), kwargs.pop("send_failure", False))
|
|
with patch.object(self.v1.socket, "socket", return_value=sock), \
|
|
patch.object(self.v1.select, "select", side_effect=sock.select):
|
|
result = self.v1.discover(kwargs.pop("interfaces", [interface()]), 1, 4942,
|
|
self.logger, session=SESSION, **kwargs)
|
|
return result, sock
|
|
|
|
def test_exact_big_endian_empty_info_header(self):
|
|
self.assertEqual(self.v1.encode_frame(INFO), b"\x5a\x0e\x21\x04\x00\x00\x00\x00")
|
|
packet = self.v1.encode_frame(REQUEST, {"session": SESSION})
|
|
self.assertEqual(packet, wire(REQUEST, {"session": SESSION}))
|
|
|
|
def test_fragmented_tcp_and_documented_field_filter(self):
|
|
body = info_body()
|
|
body["undocumented_field"] = {"synthetic": "not retained"}
|
|
result, sock = self.tcp_query(body)
|
|
self.assertEqual(sock.sent, [bytes.fromhex("5a0e210400000000")])
|
|
self.assertTrue(sock.closed)
|
|
self.assertEqual(result["data"]["sn"], SN)
|
|
self.assertNotIn("undocumented_field", result["data"])
|
|
self.assertNotIn("undocumented_field", json.dumps(self.logger.events))
|
|
|
|
def test_decode_rejects_bad_command_length_json_and_utf8(self):
|
|
cases = [b"1234567", wire(INFO + 1, {}), struct.pack("!II", INFO, 0),
|
|
struct.pack("!II", INFO, self.v1.MAX_JSON_SIZE + 1),
|
|
wire(INFO, {}) + b"extra", wire(INFO, []),
|
|
struct.pack("!II", INFO, 1) + b"{",
|
|
struct.pack("!II", INFO, 1) + b"\xff"]
|
|
for packet in cases:
|
|
with self.subTest(packet=packet[:12]), self.assertRaises(self.v1.ProtocolError):
|
|
self.v1.decode_frame(packet, INFO)
|
|
|
|
def test_early_tcp_eof_closes_without_retry(self):
|
|
for packet in [b"", wire(INFO, info_body())[:-1]]:
|
|
sock = TcpSocket(packet)
|
|
with patch.object(self.v1.socket, "create_connection", return_value=sock) as connect, \
|
|
self.assertRaises(self.v1.ProtocolError):
|
|
self.v1.query_device_info(IP, 5, self.logger)
|
|
self.assertEqual(connect.call_count, 1)
|
|
self.assertTrue(sock.closed)
|
|
|
|
def test_tcp_invalid_headers_are_rejected_before_reading_body(self):
|
|
for command, size in [(REQUEST, 10), (INFO, 0), (INFO, self.v1.MAX_JSON_SIZE + 1)]:
|
|
with self.subTest(command=command, size=size), self.assertRaises(self.v1.ProtocolError):
|
|
self.tcp_query(packet=struct.pack("!II", command, size))
|
|
|
|
def test_tcp_uses_one_total_deadline_across_fragments(self):
|
|
sock = TcpSocket(wire(INFO, info_body()))
|
|
with patch.object(self.v1.socket, "create_connection", return_value=sock) as connect, \
|
|
patch.object(self.v1.time, "monotonic", side_effect=[0, 0.1, 0.2, 0.7, 1.1]), \
|
|
self.assertRaises(TimeoutError):
|
|
self.v1.query_device_info(IP, 1, self.logger)
|
|
self.assertEqual(connect.call_count, 1)
|
|
self.assertTrue(sock.closed)
|
|
self.assertEqual(len(sock.sent), 1)
|
|
|
|
def test_native_code_must_be_integer_zero_and_fields_strings(self):
|
|
for key, value in [("code", False), ("code", "0"), ("code", 1), ("dev_type", {}), ("sn", 123)]:
|
|
body = info_body()
|
|
body[key] = value
|
|
with self.subTest(key=key, value=value), self.assertRaises(self.v1.ProtocolError):
|
|
self.tcp_query(body)
|
|
|
|
def test_identity_requires_explicit_successful_v1_envelope_and_exact_sn(self):
|
|
result, _ = self.tcp_query()
|
|
invalid = [dict(result, code=False), dict(result, code=1), dict(result, protocol="v2"),
|
|
dict(result, cmd=30101), {"cmd": INFO, "code": 0, "data": result["data"]}]
|
|
for value in invalid:
|
|
with self.subTest(value=value), self.assertRaises(self.v1.ProtocolError):
|
|
self.v1.validate_identity(value, SN, MODEL)
|
|
with self.assertRaises(self.v1.ProtocolError):
|
|
self.v1.validate_identity(result, "0001", MODEL)
|
|
with self.assertRaises(self.v1.ProtocolError):
|
|
self.v1.validate_identity(result, SN, "wrong-model")
|
|
|
|
def test_dev_id_cannot_substitute_for_dev_type(self):
|
|
body = info_body()
|
|
body["dev_type"], body["dev_id"] = "OTHER-MODEL", MODEL
|
|
result, _ = self.tcp_query(body)
|
|
with self.assertRaises(self.v1.ProtocolError):
|
|
self.v1.validate_identity(result, SN, MODEL)
|
|
|
|
def test_identify_observes_model_with_only_full_sn(self):
|
|
code, result, sock, connect = self.cli(["identify", "--host", IP, "--expected-sn", SN])
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(result["status"], "IDENTITY_OBSERVED")
|
|
self.assertEqual(result["models"], [MODEL])
|
|
self.assertTrue(result["sn_verified"])
|
|
self.assertFalse(result["model_verified"])
|
|
self.assertFalse(result["write_executed"])
|
|
self.assertEqual(connect.call_args.args[0], (IP, 4941))
|
|
self.assertEqual(sock.sent, [bytes.fromhex("5a0e210400000000")])
|
|
|
|
def test_info_model_mismatch_fails_after_one_read(self):
|
|
body = info_body()
|
|
body["dev_type"] = "WRONG-MODEL"
|
|
code, result, sock, connect = self.cli(["info", "--host", IP, *self.identity()], body)
|
|
self.assertEqual(code, 1)
|
|
self.assertEqual(result["status"], "ERROR")
|
|
self.assertEqual(connect.call_count, 1)
|
|
self.assertTrue(sock.closed)
|
|
|
|
def test_verify_native_fields_and_missing_outer_without_alias(self):
|
|
for field, expected in [("firmware", "PASS"), ("dev_ver", "PASS"), ("outer.version", "INCOMPLETE")]:
|
|
with self.subTest(field=field):
|
|
code, result, _, _ = self.cli(["verify-version", "--host", IP, *self.identity(),
|
|
"--target-version", TARGET, "--version-field", field])
|
|
self.assertEqual(result["status"], expected)
|
|
self.assertEqual(result["firmware_verified"], expected == "PASS")
|
|
self.assertEqual(result["evidence_source"], "live_v1_2104")
|
|
self.assertEqual(code, 0 if expected == "PASS" else 1)
|
|
|
|
def test_version_mismatch_and_missing_firmware_do_not_use_hardware(self):
|
|
for actual, status in [("OLD-VERSION", "FAIL"), ("", "INCOMPLETE")]:
|
|
body = info_body()
|
|
body.update(firmware=actual, hardware=TARGET)
|
|
code, result, _, _ = self.cli(["verify-version", "--host", IP, *self.identity(),
|
|
"--target-version", TARGET, "--version-field", "firmware"], body)
|
|
self.assertEqual(code, 1)
|
|
self.assertEqual(result["status"], status)
|
|
self.assertFalse(result["firmware_verified"])
|
|
|
|
def test_literal_outer_version_can_be_observed_without_alias_or_other_nested_data(self):
|
|
for literal, expected in [(TARGET, "PASS"), ("OTHER", "FAIL"), ({}, "INCOMPLETE")]:
|
|
body = info_body(outer={"version": literal, "unrelated": "discard"})
|
|
code, result, _, _ = self.cli(["verify-version", "--host", IP, *self.identity(),
|
|
"--target-version", TARGET, "--version-field", "outer.version"], body)
|
|
self.assertEqual(result["status"], expected)
|
|
self.assertEqual(code, 0 if expected == "PASS" else 1)
|
|
self.assertNotIn("unrelated", result["response"]["data"].get("outer", {}))
|
|
|
|
def test_recorded_response_requires_v1_envelope_and_no_network(self):
|
|
path = Path(self.temp.name) / "recorded.json"
|
|
path.write_text(json.dumps({"cmd": 30101, "code": 0, "data": info_body()}), encoding="utf-8")
|
|
code, result, _, connect = self.cli(["verify-version", "--response", str(path), *self.identity(),
|
|
"--target-version", TARGET, "--version-field", "firmware"])
|
|
self.assertEqual(code, 1)
|
|
self.assertEqual(result["status"], "ERROR")
|
|
self.assertFalse(connect.called)
|
|
|
|
def test_v1_upgrade_preflight_hashes_and_reads_but_never_sends_upgrade(self):
|
|
code, result, sock, connect = self.cli(["upgrade", "--host", IP, *self.identity(),
|
|
"--rom", str(self.rom), "--rom-sha256", self.sha,
|
|
"--target-version", TARGET, "--version-field", "outer.version"])
|
|
self.assertEqual(code, 0)
|
|
self.assertEqual(result["status"], "PREFLIGHT_ONLY")
|
|
self.assertTrue(result["preflight_complete"])
|
|
self.assertFalse(result["write_executed"])
|
|
self.assertTrue(result["write_supported"])
|
|
self.assertFalse(result["selected_version_field_available"])
|
|
self.assertEqual(result["rom"]["sha256"], self.sha)
|
|
self.assertEqual(connect.call_count, 1)
|
|
self.assertEqual(sock.sent, [bytes.fromhex("5a0e210400000000")])
|
|
|
|
def test_udp_request_reply_session_prefix_and_dedup(self):
|
|
reply = (wire(RESPONSE, discovery_body()), (IP, 4942))
|
|
own_request = (wire(REQUEST, {"session": SESSION}), ("192.0.2.20", 4942))
|
|
result, sock = self.discover([own_request, reply, reply], expected_sn=SN,
|
|
interfaces=[interface(), interface("192.0.2.21/24")])
|
|
self.assertEqual(sock.bindings, [("0.0.0.0", 4942)])
|
|
self.assertEqual(sock.sent, [(wire(REQUEST, {"session": SESSION}), ("192.0.2.255", 4942))])
|
|
self.assertTrue(sock.closed)
|
|
self.assertEqual(result["candidate_count"], 1)
|
|
self.assertTrue(result["identity_verified"])
|
|
self.assertEqual(result["matches"][0]["source_ip"], IP)
|
|
self.assertIn("v1_own_request_ignored", [event["event"] for event in self.logger.events])
|
|
|
|
def test_udp_rejects_unrelated_sessions_malformed_mac_and_frames(self):
|
|
bodies = [discovery_body(session=SESSION + 1), discovery_body(session=True),
|
|
discovery_body(sn=""), discovery_body(mac=[]), discovery_body(code=False)]
|
|
replies = [(wire(RESPONSE, body), (IP, 4942)) for body in bodies]
|
|
replies += [(b"bad", (IP, 4942)), (wire(RESPONSE, discovery_body()), (IP, 4942))]
|
|
result, _ = self.discover(replies, expected_sn=SN)
|
|
self.assertTrue(result["identity_verified"])
|
|
self.assertEqual(sum(e["event"] == "v1_invalid_response" for e in self.logger.events), 6)
|
|
|
|
def test_udp_same_sn_conflicting_ip_mac_or_model_blocks_selection(self):
|
|
for body, peer in [(discovery_body(), ("192.0.2.11", 4942)),
|
|
(discovery_body(mac="00:00:5E:00:53:02"), (IP, 4942)),
|
|
(discovery_body(dev_type="OTHER-MODEL"), (IP, 4942))]:
|
|
replies = [(wire(RESPONSE, discovery_body()), (IP, 4942)), (wire(RESPONSE, body), peer)]
|
|
with self.subTest(body=body, peer=peer):
|
|
result, _ = self.discover(replies, expected_sn=SN)
|
|
self.assertEqual(result["status"], "BLOCKED")
|
|
self.assertEqual(result["matches"], [])
|
|
self.assertEqual(result["identity_conflicts"], [SN])
|
|
|
|
def test_udp_exact_sn_and_model_filters_no_suffix_match(self):
|
|
reply = [(wire(RESPONSE, discovery_body()), (IP, 4942))]
|
|
for filters in [dict(expected_sn="0001"), dict(expected_sn=SN, expected_model="wrong")]:
|
|
with self.subTest(filters=filters):
|
|
result, _ = self.discover(reply, **filters)
|
|
self.assertEqual(result["status"], "BLOCKED")
|
|
self.assertFalse(result["identity_verified"])
|
|
|
|
def test_udp_no_response_is_blocked_for_target_not_an_offline_claim(self):
|
|
result, sock = self.discover(expected_sn=SN)
|
|
self.assertEqual(result["status"], "BLOCKED")
|
|
self.assertEqual(result["reason"], "target_not_uniquely_found")
|
|
self.assertTrue(sock.closed)
|
|
|
|
def test_udp_port_conflict_and_send_failure_close_without_retry(self):
|
|
for options in [dict(bind_failure=True), dict(send_failure=True)]:
|
|
sock = UdpSocket(**options)
|
|
with patch.object(self.v1.socket, "socket", return_value=sock) as create, \
|
|
self.assertRaises(self.v1.ProtocolError):
|
|
self.v1.discover([interface()], 1, 4942, self.logger, session=SESSION)
|
|
self.assertEqual(create.call_count, 1)
|
|
self.assertEqual(len(sock.bindings), 1)
|
|
self.assertTrue(sock.closed)
|
|
|
|
def test_discovery_cli_routes_actual_v1_frames(self):
|
|
sock = UdpSocket([(wire(RESPONSE, discovery_body()), (IP, 4942))])
|
|
output = io.StringIO()
|
|
with patch.object(self.v1.socket, "socket", return_value=sock), \
|
|
patch.object(self.v1.secrets, "randbelow", return_value=SESSION - 1), \
|
|
patch.object(self.v1.select, "select", side_effect=sock.select), contextlib.redirect_stdout(output):
|
|
code = self.adapter.main(["discover", "--protocol", "v1", "--interface", "192.0.2.20/24",
|
|
"--expected-sn", SN])
|
|
result = json.loads(output.getvalue())
|
|
self.assertEqual(code, 0)
|
|
self.assertTrue(result["identity_verified"])
|
|
self.assertEqual(sock.sent[0][0][:4], bytes.fromhex("5a0eff00"))
|
|
|
|
def test_invalid_v1_interface_reports_structured_error_before_network(self):
|
|
output = io.StringIO()
|
|
with patch.object(self.v1.socket, "socket", side_effect=AssertionError("No network")), \
|
|
contextlib.redirect_stdout(output):
|
|
code = self.adapter.main(["discover", "--protocol", "v1", "--interface", "invalid"])
|
|
self.assertEqual(code, 1)
|
|
self.assertEqual(json.loads(output.getvalue())["status"], "ERROR")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|