"""Offline safety and protocol regression tests; never contact a real device.""" from __future__ import annotations import contextlib import hashlib import importlib.util import io import json from pathlib import Path import struct import sys import tempfile import unittest from unittest.mock import patch from types import SimpleNamespace ROOT = Path(__file__).resolve().parents[1] SCRIPT = ROOT / "scripts" / "device_adapter.py" SN = "SYNTHETIC-TEST-SN-0001" MODEL = "SYNTHETIC-MODEL" TARGET = "9.8.7.6" # RFC 5737 documentation address; every socket is replaced with an offline fixture. TEST_IP = "192.0.2.10" def response(sn=SN, model=MODEL, version=TARGET): return { "cmd": 30101, "session": 123, "code": 0, "data": { "sn": sn, "inner": {"type": model, "version": version}, "outer": {"type": model, "version": "1.0.0.0"}, "hardware": {"type": model, "version": TARGET}, }, } class FragmentedInfoSocket: """Transport-only double: creates a protocol response for the actual request.""" def __init__(self, reply): self.reply = reply self.pending = bytearray() self.request = None self.receives = 0 def __enter__(self): return self def __exit__(self, *_args): return False def settimeout(self, _timeout): pass def sendall(self, packet): assert packet[:4] == b"EPTP" self.request = json.loads(packet[8:]) reply = dict(self.reply, session=self.request["session"]) payload = json.dumps(reply).encode("utf-8") self.pending.extend(b"EPTP" + struct.pack("!I", len(payload)) + payload) def recv(self, size): self.receives += 1 count = min(size, 3, len(self.pending)) chunk = bytes(self.pending[:count]) del self.pending[:count] return chunk class UpgradeStatusSocket: """Run the real transfer state machine with controlled writes and status timing.""" def __init__(self, rom_size, success_after_wire_bytes): self.success_after_wire_bytes = success_after_wire_bytes payload = struct.pack("!BIIIII", 2, 123, rom_size, rom_size, rom_size, rom_size) self.pending = bytearray(b"EPFP" + struct.pack("!II", 4001, len(payload)) + payload) self.sent = bytearray() def __enter__(self): return self def __exit__(self, *_args): return False def setblocking(self, _blocking): pass def send(self, packet): count = min(len(packet), 7) self.sent.extend(packet[:count]) return count def recv(self, size): count = min(size, 3, len(self.pending)) chunk = bytes(self.pending[:count]) del self.pending[:count] return chunk def ready(self, readable, writable, exceptional, timeout): if len(self.sent) >= self.success_after_wire_bytes: return ([self], [], []) return ([], [self], []) class DeviceAdapterTests(unittest.TestCase): def setUp(self): self.assertTrue(SCRIPT.is_file(), "device_adapter.py has not been implemented") spec = importlib.util.spec_from_file_location("device_adapter_under_test", SCRIPT) self.adapter = importlib.util.module_from_spec(spec) spec.loader.exec_module(self.adapter) self.temp = tempfile.TemporaryDirectory() self.addCleanup(self.temp.cleanup) self.rom = Path(self.temp.name) / "synthetic.rom" self.rom.write_bytes(b"synthetic firmware fixture\x00" * 30) self.sha = hashlib.sha256(self.rom.read_bytes()).hexdigest() def run_cli(self, argv, reply=None): info_socket = FragmentedInfoSocket(reply or response()) connections = [] def connect(address, timeout): connections.append(address) if address[1] != 8942: raise AssertionError("Unexpected write-port connection") return info_socket stdout = io.StringIO() with patch.object(self.adapter.protocol.socket, "create_connection", side_effect=connect), contextlib.redirect_stdout(stdout): code = self.adapter.main(argv) return code, json.loads(stdout.getvalue()), connections, info_socket def identity_args(self): return ["--expected-sn", SN, "--expected-model", MODEL] def upgrade_args(self): return ["upgrade", "--host", TEST_IP, *self.identity_args(), "--rom", str(self.rom), "--target-version", TARGET, "--version-field", "inner.version"] def run_transfer_cli(self, success_after_wire_bytes, host=TEST_IP): info_socket = FragmentedInfoSocket(response()) transfer_socket = UpgradeStatusSocket(self.rom.stat().st_size, success_after_wire_bytes) connections = [] def connect(address, timeout): connections.append(address) return info_socket if address[1] == 8942 else transfer_socket stdout = io.StringIO() args = self.upgrade_args() + ["--execute", "--rom-sha256", self.sha] args[2] = host with patch.object(self.adapter.protocol.socket, "create_connection", side_effect=connect), \ patch.object(self.adapter.protocol.select, "select", side_effect=transfer_socket.ready), \ contextlib.redirect_stdout(stdout): code = self.adapter.main(args) return code, json.loads(stdout.getvalue()), connections, transfer_socket def test_v2_original_replaced_at_upgrade_connection_sends_approved_snapshot(self): approved = self.rom.read_bytes() for replacement in (b"B" * len(approved), b"C", b"D" * (len(approved) + 5)): self.rom.write_bytes(approved) info_socket = FragmentedInfoSocket(response()) transfer_socket = UpgradeStatusSocket(len(approved), 12 + len(approved)) def connect(address, timeout): if address[1] == 8942: return info_socket self.rom.write_bytes(replacement) return transfer_socket out = io.StringIO() with patch.object(self.adapter.protocol.socket, "create_connection", side_effect=connect), \ patch.object(self.adapter.protocol.select, "select", side_effect=transfer_socket.ready), \ contextlib.redirect_stdout(out): code = self.adapter.main(self.upgrade_args() + ["--execute", "--rom-sha256", self.sha]) self.assertEqual(code, 0, out.getvalue()) self.assertEqual(bytes(transfer_socket.sent)[12:], approved) def verify_args(self, reply): path = Path(self.temp.name) / "response.json" path.write_text(json.dumps(reply), encoding="utf-8") return ["verify-version", "--response", str(path), *self.identity_args(), "--target-version", TARGET, "--version-field", "inner.version"] def test_capabilities_is_offline_and_lists_explicit_v1_transfer(self): stdout = io.StringIO() with patch.object(self.adapter.protocol.socket, "socket", side_effect=AssertionError("Network forbidden")), \ patch.object(self.adapter.protocol.socket, "create_connection", side_effect=AssertionError("Network forbidden")), \ contextlib.redirect_stdout(stdout): code = self.adapter.main(["capabilities"]) result = json.loads(stdout.getvalue()) self.assertEqual(code, 0) self.assertEqual(result["status"], "CAPABILITIES") self.assertFalse(result["network_used"]) self.assertFalse(result["firmware_verified"]) self.assertTrue(result["protocols"]["v1"]["implemented"]) self.assertEqual(result["protocols"]["v1"]["operations"], ["discover", "identify", "info", "upgrade", "verify-version"]) self.assertTrue(result["protocols"]["v1"]["upgrade_preflight"]) self.assertTrue(result["protocols"]["v1"]["upgrade_transfer"]) def test_v1_execute_requires_reviewed_digest_before_io_without_v2_fallback(self): commands = [ self.upgrade_args() + ["--execute"], ] for command in commands: with self.subTest(command=command[0], execute="--execute" in command): stdout = io.StringIO() with patch.object(self.adapter, "load_vendor", side_effect=AssertionError("No fallback")), \ patch.object(self.adapter, "inspect_rom", side_effect=AssertionError("No ROM access")), \ patch.object(self.adapter, "read_response", side_effect=AssertionError("No response access")), \ patch.object(self.adapter.protocol, "query_device_info", side_effect=AssertionError("No V2 query")), \ patch.object(self.adapter.protocol, "transfer_rom", side_effect=AssertionError("No transfer")), \ patch.object(self.adapter.protocol.socket, "socket", side_effect=AssertionError("No UDP")), \ patch.object(self.adapter.protocol.socket, "create_connection", side_effect=AssertionError("No TCP")), \ contextlib.redirect_stdout(stdout): code = self.adapter.main(command + ["--protocol", "v1"]) result = json.loads(stdout.getvalue()) self.assertEqual(code, 1) self.assertEqual(result["status"], "ERROR") self.assertIn("requires an explicitly reviewed", result["message"]) self.assertEqual(result["protocol"], "v1") self.assertFalse(result["write_may_have_started"]) self.assertFalse(result["automatic_retry"]) self.assertEqual([e["event"] for e in result["events"]], ["upgrade_target_bound"]) def test_discover_preserves_native_arguments_with_default_or_explicit_v2(self): native = ["--expected-sn", SN, "--expected-model", MODEL, "--timeout", "3"] for selection in ([], ["--protocol", "v2"]): with self.subTest(selection=selection): discovery = SimpleNamespace(main=lambda arguments: (arguments, 17)) with patch.object(self.adapter, "load_vendor", return_value=discovery) as load: result = self.adapter.main(["discover", *native, *selection]) self.assertEqual(result, (native, 17)) load.assert_called_once_with("v2_discover.py", "artemis_device_vendor_discover") def test_explicit_v2_preflight_uses_only_identity_connection(self): code, result, connections, _ = self.run_cli(self.upgrade_args() + ["--protocol", "v2"]) self.assertEqual(code, 0) self.assertEqual(result["protocol"], "v2") self.assertEqual(result["status"], "PREFLIGHT_ONLY") self.assertFalse(result["write_executed"]) self.assertEqual(connections, [(TEST_IP, 8942)]) def test_info_needs_no_rom_and_reads_fragmented_response(self): code, result, connections, sock = self.run_cli(["info", "--host", "synthetic.invalid", *self.identity_args()]) self.assertEqual(code, 0) self.assertEqual(result["status"], "IDENTITY_CONFIRMED") self.assertEqual(result["response"]["data"]["sn"], SN) self.assertEqual(connections, [("synthetic.invalid", 8942)]) self.assertGreater(sock.receives, 10) self.assertEqual(sock.request["cmd"], 30101) def test_v2_identify_observes_model_with_only_full_sn(self): code, result, connections, _ = self.run_cli(["identify", "--host", TEST_IP, "--expected-sn", SN]) self.assertEqual(code, 0) self.assertEqual(result["status"], "IDENTITY_OBSERVED") self.assertIn(MODEL, result["models"]) self.assertTrue(result["sn_verified"]) self.assertFalse(result["model_verified"]) self.assertFalse(result["write_executed"]) self.assertEqual(connections, [(TEST_IP, 8942)]) def test_v2_rejects_v1_native_version_fields_before_network(self): for field in ("firmware", "dev_ver"): with self.subTest(field=field): args = self.upgrade_args() args[-1] = field code, result, connections, _ = self.run_cli(args) self.assertEqual(code, 1) self.assertEqual(result["status"], "ERROR") self.assertEqual(connections, []) def test_full_serial_mismatch_blocks_upgrade_before_write_port(self): code, result, connections, _ = self.run_cli(self.upgrade_args() + ["--execute", "--rom-sha256", self.sha], response(sn=SN + "-OTHER")) self.assertNotEqual(code, 0) self.assertEqual(result["status"], "ERROR") self.assertEqual([port for _, port in connections], [8942]) def test_model_mismatch_blocks_upgrade_before_write_port(self): code, result, connections, _ = self.run_cli(self.upgrade_args() + ["--execute", "--rom-sha256", self.sha], response(model="SYNTHETIC-OTHER-MODEL")) self.assertNotEqual(code, 0) self.assertEqual(result["status"], "ERROR") self.assertEqual([port for _, port in connections], [8942]) def test_upgrade_without_execute_never_connects_8943(self): code, result, connections, _ = self.run_cli(self.upgrade_args()) self.assertEqual(code, 0) self.assertEqual(result["status"], "PREFLIGHT_ONLY") self.assertFalse(result["firmware_verified"]) self.assertEqual(result["rom"]["sha256"], self.sha) self.assertEqual(result["target_version"], TARGET) self.assertEqual(result["version_field"], "inner.version") self.assertEqual([port for _, port in connections], [8942]) def test_upgrade_rejects_hostname_even_without_execute(self): args = self.upgrade_args() args[2] = "synthetic.invalid" code, result, connections, _ = self.run_cli(args) self.assertNotEqual(code, 0) self.assertEqual(result["status"], "ERROR") self.assertEqual(connections, []) def test_upgrade_rejects_determinable_non_unicast_targets(self): for host in ("224.0.0.1", "0.0.0.0", "255.255.255.255", "::", "ff02::1", "::ffff:0.0.0.0", "::ffff:224.0.0.1", "::ffff:255.255.255.255"): with self.subTest(host=host): args = self.upgrade_args() args[2] = host code, result, connections, _ = self.run_cli(args) self.assertNotEqual(code, 0) self.assertEqual(result["status"], "ERROR") self.assertEqual(connections, []) def test_numeric_ip_is_bound_for_info_and_transfer_and_saved_as_evidence(self): expected_wire_bytes = self.rom.stat().st_size + 12 for host, normalized in ((TEST_IP, TEST_IP), ("2001:0db8:0:0::10", "2001:db8::10")): with self.subTest(host=host): code, result, connections, sock = self.run_transfer_cli(expected_wire_bytes, host) self.assertEqual(code, 0) self.assertEqual(result.get("bound_ip"), normalized) self.assertEqual(connections, [(normalized, 8942), (normalized, 8943)]) self.assertEqual(bytes(sock.sent[:12]), b"EPFP" + struct.pack("!II", 4001, self.rom.stat().st_size)) self.assertEqual(bytes(sock.sent[12:]), self.rom.read_bytes()) self.assertEqual(result["transfer"]["bytes_sent"], self.rom.stat().st_size) self.assertTrue(result["protocol_success"]) self.assertFalse(result["firmware_verified"]) def test_zero_or_partial_upload_status_2_cannot_claim_protocol_success(self): for wire_bytes in (0, 12 + self.rom.stat().st_size // 2): with self.subTest(wire_bytes=wire_bytes): code, result, connections, sock = self.run_transfer_cli(wire_bytes) self.assertNotEqual(code, 0) self.assertEqual(result["status"], "ERROR") self.assertFalse(result.get("protocol_success", False)) self.assertFalse(result["firmware_verified"]) self.assertTrue(result["write_may_have_started"]) self.assertTrue(result["postcheck_required"]) self.assertFalse(result["automatic_retry"]) self.assertEqual(connections, [(TEST_IP, 8942), (TEST_IP, 8943)]) self.assertLess(max(0, len(sock.sent) - 12), self.rom.stat().st_size) self.assertEqual(result.get("bound_ip"), TEST_IP) self.assertNotIn("protocol_success", [event["event"] for event in result["events"]]) def test_execute_requires_expected_rom_digest_before_network(self): code, result, connections, _ = self.run_cli(self.upgrade_args() + ["--execute"]) self.assertNotEqual(code, 0) self.assertEqual(result["status"], "ERROR") self.assertEqual(connections, []) def test_wrong_rom_digest_blocks_before_network(self): code, _result, connections, _ = self.run_cli(self.upgrade_args() + ["--execute", "--rom-sha256", "0" * 64]) self.assertNotEqual(code, 0) self.assertEqual(connections, []) def test_offline_only_hashes_local_rom(self): code, result, connections, _ = self.run_cli(["offline", "--rom", str(self.rom)]) self.assertEqual(code, 0) self.assertEqual(result["status"], "OFFLINE_READY") self.assertEqual(result["rom"]["sha256"], self.sha) self.assertFalse(result["firmware_verified"]) self.assertEqual(connections, []) def test_protocol_success_does_not_pass_firmware_acceptance(self): with patch.object(self.adapter.protocol, "transfer_rom", return_value={"status": 2, "bytes_sent": self.rom.stat().st_size}) as transfer: code, result, _connections, _ = self.run_cli(self.upgrade_args() + ["--execute", "--rom-sha256", self.sha]) self.assertEqual(code, 0) self.assertEqual(transfer.call_count, 1) self.assertEqual(result["status"], "PROTOCOL_SUCCESS_POSTCHECK_REQUIRED") self.assertFalse(result["firmware_verified"]) self.assertTrue(result["postcheck_required"]) def test_write_error_is_not_retried(self): with patch.object(self.adapter.protocol, "transfer_rom", side_effect=OSError("synthetic disconnect")) as transfer: code, result, _connections, _ = self.run_cli(self.upgrade_args() + ["--execute", "--rom-sha256", self.sha]) self.assertNotEqual(code, 0) self.assertEqual(transfer.call_count, 1) self.assertEqual(result["status"], "ERROR") self.assertTrue(result.get("write_may_have_started")) self.assertTrue(result.get("postcheck_required")) def test_cli_configures_utf8_without_import_side_effects(self): class ConfigStream(io.StringIO): def __init__(self): super().__init__() self.configurations = [] def reconfigure(self, **kwargs): self.configurations.append(kwargs) stdout, stderr = ConfigStream(), ConfigStream() with contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr): code = self.adapter.main(["offline", "--rom", str(self.rom)]) self.assertEqual(code, 0) self.assertIn({"encoding": "utf-8", "errors": "backslashreplace"}, stdout.configurations) self.assertIn({"encoding": "utf-8", "errors": "backslashreplace"}, stderr.configurations) def test_recorded_version_mismatch_cannot_pass(self): code, result, connections, _ = self.run_cli(self.verify_args(response(version="0.0.0.0"))) self.assertNotEqual(code, 0) self.assertEqual(result["status"], "FAIL") self.assertFalse(result["firmware_verified"]) self.assertEqual(connections, []) def test_missing_firmware_version_cannot_pass_even_if_hardware_matches(self): reply = response() del reply["data"]["inner"]["version"] code, result, _connections, _ = self.run_cli(self.verify_args(reply)) self.assertNotEqual(code, 0) self.assertEqual(result["status"], "INCOMPLETE") self.assertFalse(result["firmware_verified"]) def test_hardware_version_is_rejected_as_acceptance_field(self): args = self.verify_args(response()) args[-1] = "hardware.version" with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit) as error: self.adapter.main(args) self.assertEqual(error.exception.code, 2) def test_missing_data_cannot_pass(self): code, result, _connections, _ = self.run_cli(self.verify_args({"cmd": 30101, "code": 0})) self.assertNotEqual(code, 0) self.assertEqual(result["status"], "ERROR") self.assertFalse(result["firmware_verified"]) def test_wrong_identity_in_recorded_response_cannot_pass(self): code, result, _connections, _ = self.run_cli(self.verify_args(response(sn="SYNTHETIC-WRONG-SN"))) self.assertNotEqual(code, 0) self.assertFalse(result["firmware_verified"]) def test_matching_recorded_response_verifies_only_recorded_version(self): code, result, connections, _ = self.run_cli(self.verify_args(response())) self.assertEqual(code, 0) self.assertEqual(result["status"], "PASS") self.assertTrue(result["firmware_verified"]) self.assertEqual(result["evidence_source"], "recorded_response") self.assertFalse(result["live_readback"]) self.assertEqual(connections, []) def test_online_version_verification_reads_once(self): args = ["verify-version", "--host", "synthetic.invalid", *self.identity_args(), "--target-version", TARGET, "--version-field", "inner.version"] code, result, connections, _ = self.run_cli(args) self.assertEqual(code, 0) self.assertEqual(result["status"], "PASS") self.assertTrue(result["live_readback"]) self.assertEqual([port for _, port in connections], [8942]) def test_import_preserves_stdout_and_stderr_configuration(self): class GuardStream(io.StringIO): def reconfigure(self, **_kwargs): raise AssertionError("Import must not reconfigure caller output") # Reload the dependencies too: a cached import would hide the regression. names = [name for name in sys.modules if name.startswith("artemis_device_vendor")] removed = {name: sys.modules.pop(name) for name in names} try: with contextlib.redirect_stdout(GuardStream()), contextlib.redirect_stderr(GuardStream()): spec = importlib.util.spec_from_file_location("safe_import_probe", SCRIPT) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) module.load_vendor("v2_discover.py", "artemis_device_vendor_discover") finally: sys.modules.update(removed) if __name__ == "__main__": unittest.main()