diff --git a/.github/workflows/run-unit-tests.yml b/.github/workflows/run-unit-tests.yml index 4b56eb79..7b84b987 100644 --- a/.github/workflows/run-unit-tests.yml +++ b/.github/workflows/run-unit-tests.yml @@ -341,6 +341,7 @@ jobs: python3 -B test/test_morse_auth.py -v python3 -B test/test_morse_reply_handoff.py -v python3 -B test/test_morse_remote.py -v + python3 -B test/test_morse_hil_safety.py -v python3 -B test/test_morse_loop_policy.py -v python3 -B test/test_morse_terminal_input.py -v python3 -B test/test_radio_profile_cli.py diff --git a/docs/cli_commands.md b/docs/cli_commands.md index ee1655cc..a46354dd 100644 --- a/docs/cli_commands.md +++ b/docs/cli_commands.md @@ -27,8 +27,9 @@ For timed Morse RX, use `morse listen ` (fractions allowed, 1 second through 5 minutes). `morse stop` ends TX or RX. All roles retain the bounded capture in RAM for `morse read [offset]`; follow each page's `next` value until `next=END`. `morse clear` removes it when idle. A new successful session or -reboot replaces/removes the capture. Full Companion ASCII terminals show -decoded progress; binary clients request it through the CLI, without adding +reboot replaces/removes the capture. Buffered USB/browser Full Companion ASCII +terminals show decoded progress; other terminals and binary clients request it +through the CLI, without adding unidentified text to authenticated channel/chat messages. Normal LoRa is paused during listening, so a remote administrator must wait for the selected session to end before reading the capture over LoRa. See [timed reception](carrier_wave.md#timed-morse-reception-and-retained-captures) diff --git a/examples/companion_radio/MyMesh.cpp b/examples/companion_radio/MyMesh.cpp index 2ebd9b8a..ad3d52c9 100644 --- a/examples/companion_radio/MyMesh.cpp +++ b/examples/companion_radio/MyMesh.cpp @@ -51,6 +51,16 @@ #include "helpers/radiolib/RxBoostedGainDefaults.h" #include "helpers/radiolib/CadTiming.h" +#if COMPANION_FEATURE_READER +static bool isCompanionReaderRequest(const char* command) { + if (!command) return false; + mesh::bible::skipSpace(command); + if (!mesh::bible::token(command, "get")) return false; + mesh::bible::skipSpace(command); + return mesh::bible::token(command, "reader"); +} +#endif + #if defined(MESHCORE_EXTRAFS_HIL) \ && (!defined(NRF52_PLATFORM) || !defined(EXTRAFS) || defined(QSPIFLASH)) #error "MESHCORE_EXTRAFS_HIL requires an nRF52 internal ExtraFS build" @@ -8753,6 +8763,11 @@ void MyMesh::serviceMorseTerminal() { || _terminal_input_editing || _terminal_output->available() > 0 || _terminal_login_pending || _terminal_trace_pending || _terminal_command.isPending()) return; +#if COMPANION_FEATURE_READER + // A chapter cursor owns its entire logical output, including pauses under + // backpressure. Progress must not split a header, verse or final prompt. + if (mesh::readerCommandActive(_terminal_output)) return; +#endif // The existing USB facade owns all queued suffixes, including replies and // echo. Never append progress in the middle of an earlier text producer. if (_terminal_mode && _terminal_output == &mesh::usbTerminalPort() @@ -9485,6 +9500,11 @@ void MyMesh::handleTerminalCommand(char* command, bool reader_prompt, mesh::cli::normalizeCommandVerb(command); #if COMPANION_FEATURE_READER + if (mesh::morseTimingActive(_radio) && isCompanionReaderRequest(command)) { + mesh::cancelReaderCommand(); + terminalOutput().print(" ERROR: Reader unavailable during Morse\r\n"); + return; + } if (mesh::handleReaderCommand(command, terminalOutput(), reader_prompt, reader_try_write)) return; #endif @@ -10340,6 +10360,11 @@ bool MyMesh::handleCommand(const char* command, uint32_t sender_timestamp, } #if COMPANION_FEATURE_READER + if (sender_timestamp == 0 && mesh::morseTimingActive(_radio) + && isCompanionReaderRequest(command)) { + snprintf(reply, reply_capacity, "Error: Reader unavailable during Morse"); + return true; + } if (sender_timestamp == 0 && mesh::handleReaderReply(command, reply, reply_capacity)) return true; #endif @@ -10765,7 +10790,10 @@ complete_line: { reply_buf[0] = 0; #if COMPANION_FEATURE_READER - if (mesh::handleReaderCommand(cli_command, output)) { + if (mesh::morseTimingActive(_radio) && isCompanionReaderRequest(cli_command)) { + mesh::cancelReaderCommand(); + output.print(" ERROR: Reader unavailable during Morse\r\n"); + } else if (mesh::handleReaderCommand(cli_command, output)) { // USB rescue streams chapters/ranges cooperatively like the terminal. } else #endif @@ -11103,7 +11131,9 @@ void MyMesh::loop() { #endif servicePendingSerialReply(); #if COMPANION_FEATURE_READER - mesh::serviceReaderCommand(); + // Every verse chunk reinflates its 2 KiB flash block. Retain the cursor but + // defer that optional decode/output work until envelope timing is finished. + if (!mesh::morseTimingActive(_radio)) mesh::serviceReaderCommand(); #endif serviceBinaryTraceReply(); if (!mesh::morseTimingActive(_radio)) serviceSigningSession(); diff --git a/test/fixtures/morse_rx_terminal.cpp b/test/fixtures/morse_rx_terminal.cpp index d029e859..e9f4ba9f 100644 --- a/test/fixtures/morse_rx_terminal.cpp +++ b/test/fixtures/morse_rx_terminal.cpp @@ -23,6 +23,8 @@ namespace mesh { Stream usb_terminal; Stream& usbTerminalPort() { return usb_terminal; } bool hasPendingUsbTerminalOutput() { return false; } +Stream* active_reader_output = nullptr; +bool readerCommandActive(const Stream* output) { return output == active_reader_output; } } struct Radio { bool receiving = true; @@ -72,6 +74,23 @@ static size_t occurrences(const std::string& haystack, const std::string& needle return count; } int main() { +#if COMPANION_FEATURE_READER + { + MyMesh m; Stream terminal, other; m._terminal_output = &terminal; + m.radio.capture = "SOS"; terminal.available_write = 0; + m.serviceMorseTerminal(); + assert(m._morse_terminal_pending_chars == 3); + mesh::active_reader_output = &terminal; terminal.available_write = 128; + m.radio.capture += " HELP"; + for (int i = 0; i < 10; ++i) { m.clock.now += 250; m.serviceMorseTerminal(); } + assert(terminal.output.empty() && m._morse_terminal_offset == 0); + assert(m._morse_terminal_pending_chars == 3); + mesh::active_reader_output = &other; // only the exact output owner blocks progress + drain(m); + assert(terminal.output == "\r\n[Morse RX] SOS\r\n> \r\n[Morse RX] HELP\r\n> "); + mesh::active_reader_output = nullptr; + } +#endif { MyMesh m; Stream terminal, binary; m.radio.capture = "SOS"; diff --git a/test/test_companion_reader.py b/test/test_companion_reader.py index 50ac9347..def17ce4 100644 --- a/test/test_companion_reader.py +++ b/test/test_companion_reader.py @@ -181,13 +181,21 @@ int main() { # handling and local-console gate, against the real Reader handler. handler = handler[:handler.index("#if defined(NRF52_POWER_MANAGEMENT)")] handler += " return false;\n}\n" + predicate = extract_braced(source, "static bool isCompanionReaderRequest(") harness = r''' #include #include #include #include #include +struct Radio { bool active = false; }; +namespace mesh { +bool morseTimingActive(const Radio* radio) { return radio && radio->active; } +} +@PREDICATE@ struct MyMesh { + Radio radio; + Radio* _radio = &radio; bool handleCommand(const char*, uint32_t, char*); }; @HANDLER@ @@ -212,8 +220,19 @@ int main() { assert(!mesh.handleCommand("get reader 3:16", 123, out.reply)); assert(!mesh.handleCommand("AA|get reader 3:16", 123, out.reply)); assert(!mesh.handleCommand("get name", 0, out.reply)); + // Block synchronous 2 KiB decompression before it can stall Morse sampling, + // including tagged bounded replies. Ordinary readers resume unchanged. + mesh.radio.active = true; + assert(mesh.handleCommand("E5|get reader 21:17 2", 0, out.reply)); + assert(std::string(out.reply) == "E5|Error: Reader unavailable during Morse"); + assert(strlen(out.reply) < sizeof(out.reply) && out.after == 0xaabbccdd); + assert(!mesh.handleCommand("get name", 0, out.reply)); + assert(!mesh.handleCommand("get reader 3:16", 123, out.reply)); + mesh.radio.active = false; + assert(mesh.handleCommand("F6|get reader 3:16", 0, out.reply)); + assert(std::string(out.reply).find("F6|Reader 3:16 (WEB)\n") == 0); } -'''.replace("@HANDLER@", handler) +'''.replace("@HANDLER@", handler).replace("@PREDICATE@", predicate) with tempfile.TemporaryDirectory(prefix="meshcore-reader-dispatch-") as directory: directory = Path(directory) generated = directory / "dispatch.cpp" @@ -250,6 +269,114 @@ int main() { str(obj), "-o", str(binary)]) self.run_checked([str(binary)]) + def test_actual_reader_stream_pauses_and_decode_entrypoints_reject_during_morse(self): + cc, cxx = shutil.which("gcc"), shutil.which("g++") + if not cc or not cxx: + self.skipTest("host GCC and G++ required") + source = (ROOT / "examples/companion_radio/MyMesh.cpp").read_text(encoding="utf-8") + predicate = extract_braced(source, "static bool isCompanionReaderRequest(") + + def reader_block(method, marker): + body = extract_braced(source, method) + position = body.index(marker) + start = body.rfind("#if COMPANION_FEATURE_READER", 0, position) + return body[start:body.index("#endif", position) + len("#endif")] + + terminal = reader_block("void MyMesh::handleTerminalCommand(", "mesh::handleReaderCommand(") + rescue = reader_block("void MyMesh::checkCLIRescueCmd()", "mesh::handleReaderCommand(cli_command, output)") + service = reader_block("void MyMesh::loop()", "mesh::serviceReaderCommand();") + harness = r''' +#include +#include +#include +#include +#include +struct Radio { bool active = false; }; +namespace mesh { +bool morseTimingActive(const Radio* radio) { return radio && radio->active; } +} +@PREDICATE@ +struct MyMesh { + Radio radio; + Radio* _radio = &radio; + Stream output; + unsigned ordinary = 0; + Stream& terminalOutput() { return output; } + void terminalCommand(const char* command, bool reader_prompt = true, + mesh::ReaderTryWrite reader_try_write = nullptr) { +@TERMINAL@ + ++ordinary; + } + void rescueCommand(const char* cli_command) { + Stream& output = this->output; +@RESCUE@ + { ++ordinary; } + } + void loopReader() { +@SERVICE@ + } +}; +int main() { + assert(isCompanionReaderRequest(" GET\tREADER 3:16 2")); + assert(isCompanionReaderRequest("get reader")); + assert(!isCompanionReaderRequest("get readership")); + assert(!isCompanionReaderRequest("morse stop")); + assert(!isCompanionReaderRequest(nullptr)); + MyMesh mesh; + mesh.terminalCommand("get reader 3:16"); + assert(mesh::readerCommandActive(&mesh.output)); + mesh.loopReader(); // A real chapter header is partially/fully admitted. + const auto before = mesh.output.text; + const auto writes = mesh.output.writes; + mesh.radio.active = true; + for (int n = 0; n < 1000; ++n) mesh.loopReader(); + assert(mesh.output.text == before && mesh.output.writes == writes); + assert(mesh::readerCommandActive(&mesh.output)); // Pausing retains the cursor. + mesh.radio.active = false; + for (int n = 0; mesh::readerCommandActive(); ++n) { + assert(n < 1000); mesh.loopReader(); + } + assert(mesh.output.text.find("eternal life.") != std::string::npos); + assert(mesh.output.text.find("Reader 3:16", mesh.output.text.find("Reader 3:16") + 1) == std::string::npos); + assert(mesh.output.text.size() > before.size()); + + mesh.terminalCommand("get reader 3:16"); + assert(mesh::readerCommandActive()); + mesh.radio.active = true; mesh.output.text.clear(); + mesh.terminalCommand("get reader 21:17 2"); // Synchronous page must not inflate. + assert(mesh.output.text == " ERROR: Reader unavailable during Morse\r\n"); + assert(!mesh::readerCommandActive()); + mesh.output.text.clear(); mesh.rescueCommand("get reader 3:16 1"); + assert(mesh.output.text == " ERROR: Reader unavailable during Morse\r\n"); + assert(!mesh::readerCommandActive()); + mesh.output.text.clear(); mesh.terminalCommand("morse stop"); + mesh.rescueCommand("morse stop"); + assert(mesh.ordinary == 2 && mesh.output.text.empty()); // Stop/input still dispatch. + + mesh.radio.active = false; + mesh.rescueCommand("get reader 3:16"); + assert(mesh::readerCommandActive()); + for (int n = 0; mesh::readerCommandActive(); ++n) { + assert(n < 1000); mesh.loopReader(); + } + assert(mesh.output.text.find("eternal life.") != std::string::npos); +} +'''.replace("@PREDICATE@", predicate).replace("@TERMINAL@", terminal).replace("@RESCUE@", rescue).replace("@SERVICE@", service) + with tempfile.TemporaryDirectory(prefix="meshcore-reader-morse-") as directory: + directory = Path(directory) + generated = directory / "interlock.cpp" + generated.write_text(harness) + for platform in ("ESP32_PLATFORM", "NRF52_PLATFORM"): + flags = [f"-D{platform}=1", "-DCOMPANION_RADIO_FULL=1", "-DENABLE_USB_INTERFACE=1"] + obj, binary = directory / "tinf.o", directory / "interlock.exe" + self.run_checked([cc, "-std=c11", "-Os", *flags, "-c", + str(ROOT / "src/helpers/ota/OtaTinf.c"), "-o", str(obj)]) + self.run_checked([cxx, "-std=c++17", "-Os", "-Wall", "-Wextra", "-Werror", + *flags, "-I" + str(ROOT / "src"), "-I" + str(FIXTURE), + str(generated), str(ROOT / "src/helpers/CompanionReader.cpp"), + str(obj), "-o", str(binary)]) + self.run_checked([str(binary)]) + def test_reader_button_routing_and_profile_gate(self): ui = (ROOT / "examples/companion_radio/ui-new/UITask.cpp").read_text(encoding="utf-8") long_press = ui[ui.index("char UITask::handleLongPress("):] diff --git a/test/test_morse_hil_safety.py b/test/test_morse_hil_safety.py new file mode 100644 index 00000000..d6f0a64d --- /dev/null +++ b/test/test_morse_hil_safety.py @@ -0,0 +1,303 @@ +"""Fail-closed lab-tool tests. No real ports, sudo, flashing or radio access.""" +from contextlib import redirect_stdout +import copy +import importlib.util +import io +import json +from pathlib import Path +import stat +import struct +import subprocess +import sys +import tempfile +import types +import unittest +from unittest.mock import Mock, patch +import warnings +import zipfile + +from tools.hil import verified_nrf_flash as guard + +ROOT = Path(__file__).resolve().parents[1] +serial_stub = types.ModuleType("serial") +serial_tools_stub = types.ModuleType("serial.tools") +serial_tools_stub.list_ports = types.SimpleNamespace(comports=Mock(return_value=[])) +serial_stub.Serial = Mock() +with patch.dict(sys.modules, {"serial": serial_stub, "serial.tools": serial_tools_stub}): + spec = importlib.util.spec_from_file_location("morse_inventory_safety", ROOT / "tools/hil/morse_pi_inventory.py") + inventory_tool = importlib.util.module_from_spec(spec) + spec.loader.exec_module(inventory_tool) + + +class InventorySafetyTests(unittest.TestCase): + def setUp(self): + self.port_info = types.SimpleNamespace(serial_number="9AB3B64C641BA927", + device="/dev/ttyACM-test", interface="MeshCore") + self.serial_factory = Mock() + self.comports = Mock(return_value=[self.port_info]) + self.serial_patch = patch.object(inventory_tool.serial, "Serial", self.serial_factory) + self.ports_patch = patch.object(inventory_tool.list_ports, "comports", self.comports) + self.serial_patch.start() + self.ports_patch.start() + self.addCleanup(self.serial_patch.stop) + self.addCleanup(self.ports_patch.stop) + + @staticmethod + def result(command, rc=0, out="", err=""): + return subprocess.CompletedProcess(command, rc, out, err) + + def run_inventory(self, sudo=None, fuser=None): + commands = [] + + def execute(command, **kwargs): + commands.append(command) + selected = sudo if command == ["sudo", "-n", "true"] else fuser if "fuser" in command else None + if isinstance(selected, Exception): + raise selected + return selected or self.result(command) + + with patch.object(inventory_tool.subprocess, "run", side_effect=execute), redirect_stdout(io.StringIO()) as output: + status = inventory_tool.main() + return status, commands, [json.loads(line) for line in output.getvalue().splitlines()] + + def test_sudo_refusal_timeout_and_missing_binary_never_enumerate_or_open(self): + for failure in (self.result([], 1, err="sudo denied"), + subprocess.TimeoutExpired("sudo", 15), OSError("sudo missing")): + with self.subTest(failure=failure): + self.comports.reset_mock() + status, commands, records = self.run_inventory(sudo=failure) + self.assertEqual(status, 1) + self.comports.assert_not_called() + self.serial_factory.assert_not_called() + self.assertFalse(any("fuser" in command for command in commands)) + self.assertTrue(records) + + def test_only_explicit_clean_fuser_no_users_result_allows_access(self): + for rc, out, err, expected in ((0, "1234", "", False), (0, "", "", False), + (1, "", "", True), (1, " \n", "\n", True)): + with self.subTest(rc=rc, out=out, err=err): + with patch.object(inventory_tool.subprocess, "run", return_value=self.result([], rc, out, err)) as call: + self.assertIs(inventory_tool.port_is_unowned("/dev/test"), expected) + self.assertEqual(call.call_args.args[0], ["sudo", "-n", "fuser", "/dev/test"]) + for rc, out, err in ((1, "1234", ""), (1, "", "permission denied"), + (2, "", ""), (127, "", "fuser missing"), (-9, "", "")): + with self.subTest(rc=rc, out=out, err=err): + with patch.object(inventory_tool.subprocess, "run", return_value=self.result([], rc, out, err)): + with self.assertRaisesRegex(RuntimeError, "Cannot verify"): + inventory_tool.port_is_unowned("/dev/test") + + def test_uncertain_fuser_and_busy_ports_never_construct_a_serial_port(self): + for failure in (self.result([], 1, err="sudo denied"), self.result([], 2), + self.result([], 127, err="fuser missing"), + subprocess.TimeoutExpired("fuser", 10), OSError("fuser missing")): + with self.subTest(failure=failure): + status, _, records = self.run_inventory(fuser=failure) + self.assertEqual(status, 1) + self.serial_factory.assert_not_called() + self.assertTrue(records[-1]["refused"]) + status, _, records = self.run_inventory(fuser=self.result([], 0)) + self.assertEqual(status, 0) + self.serial_factory.assert_not_called() + self.assertTrue(records[-1]["busy"]) + + def test_unreviewed_and_logging_interfaces_are_never_probed(self): + self.comports.return_value = [types.SimpleNamespace(serial_number="UNKNOWN", device="/dev/other", interface="MeshCore"), + types.SimpleNamespace(serial_number="9AB3B64C641BA927", device="/dev/log", interface="MeshCore Logging")] + status, commands, _ = self.run_inventory() + self.assertEqual(status, 0) + self.assertFalse(any("fuser" in command for command in commands)) + self.serial_factory.assert_not_called() + + def test_verified_free_port_uses_exclusive_access_and_confirmed_terminal(self): + port = Mock() + self.serial_factory.return_value = port + with patch.object(inventory_tool, "read_quiet", side_effect=["", "WELCOME test", *["OK"] * 6]): + status, _, _ = self.run_inventory(fuser=self.result([], 1)) + self.assertEqual(status, 0) + self.assertTrue(self.serial_factory.call_args.kwargs["exclusive"]) + port.open.assert_called_once_with() + port.close.assert_called_once_with() + self.assertEqual([item.args[0] for item in port.write.call_args_list], + [b"+++MESHCORE-TERM-START\n", b"ver\n", b"board\n", b"get name\n", + b"get radio\n", b"get tx\n", b"get radio2.status\n", b"+++MESHCORE-TERM-STOP\n"]) + + def test_failed_terminal_entry_sends_no_commands_and_closes_port(self): + port = Mock() + self.serial_factory.return_value = port + with patch.object(inventory_tool, "read_quiet", side_effect=["", "binary mode"]): + status, _, records = self.run_inventory(fuser=self.result([], 1)) + self.assertEqual(status, 1) + self.assertTrue(records[-1]["refused"]) + self.assertEqual([item.args[0] for item in port.write.call_args_list], [b"+++MESHCORE-TERM-START\n"]) + port.close.assert_called_once_with() + + def test_terminal_stop_failure_still_closes_port(self): + port = Mock() + port.write.side_effect = lambda data: (_ for _ in ()).throw(OSError("USB vanished")) if b"TERM-STOP" in data else None + self.serial_factory.return_value = port + with patch.object(inventory_tool, "read_quiet", side_effect=["", "WELCOME test", *["OK"] * 6]): + with self.assertRaisesRegex(OSError, "USB vanished"): + self.run_inventory(fuser=self.result([], 1)) + port.close.assert_called_once_with() + + +class ApplicationZipSafetyTests(unittest.TestCase): + def setUp(self): + self.directory = tempfile.TemporaryDirectory() + self.addCleanup(self.directory.cleanup) + self.root = Path(self.directory.name) + self.application = {"bin_file": "firmware.bin", "dat_file": "firmware.dat", + "init_packet_data": {"device_type": 82, "softdevice_req": [182]}} + self.document = {"manifest": {"application": self.application, "dfu_version": 0.5}} + + def make_zip(self, document=None, entries=None, raw_manifest=None, name="RAK_4631_companion-test.zip"): + artifact = self.root / name + body = raw_manifest if raw_manifest is not None else json.dumps(document or self.document).encode() + selected = [("manifest.json", body), ("firmware.bin", b"application bytes"), ("firmware.dat", b"init bytes")] + with warnings.catch_warnings(), zipfile.ZipFile(artifact, "w", compression=zipfile.ZIP_DEFLATED) as package: + warnings.simplefilter("ignore", UserWarning) + for member, data in selected if entries is None else entries: + package.writestr(member, data) + return artifact + + def test_current_legacy_and_minimal_application_packages_are_accepted(self): + self.assertEqual(guard.verify_application_zip(self.make_zip()), ("firmware.bin", "firmware.dat")) + self.assertEqual(guard.verify_application_zip(self.make_zip({"manifest": {"application": { + "bin_file": "firmware.bin", "dat_file": "firmware.dat"}}})), ("firmware.bin", "firmware.dat")) + + def test_any_bootloader_softdevice_or_unknown_component_is_refused_even_if_null(self): + for component in ("bootloader", "softdevice", "softdevice_bootloader", "unknown"): + for value in (None, {"bin_file": "firmware.bin", "dat_file": "firmware.dat"}): + document = copy.deepcopy(self.document) + document["manifest"][component] = value + with self.subTest(component=component, value=value), self.assertRaisesRegex(ValueError, "forbids"): + guard.verify_application_zip(self.make_zip(document)) + + def test_missing_malformed_ambiguous_manifest_objects_are_refused(self): + for document in ([], {"other": {}}, {"manifest": []}, {"manifest": {}}, + {"manifest": {"application": None}}, {"manifest": {"application": []}}, + {"manifest": {"application": {"bin_file": "firmware.bin"}}}, + {"manifest": {"application": dict(self.application, sd_size=100)}}): + with self.subTest(document=document), self.assertRaises(ValueError): + guard.verify_application_zip(self.make_zip(raw_manifest=json.dumps(document).encode())) + for raw in (b"not JSON", b"\xff", b'{"manifest":{"application":{},"application":{}}}', + b'{"manifest":{"application":{},"dfu_version":NaN}}'): + with self.subTest(raw=raw), self.assertRaises(ValueError): + guard.verify_application_zip(self.make_zip(raw_manifest=raw)) + + def test_invalid_metadata_and_version_are_refused(self): + for metadata in (None, [], "init"): + document = copy.deepcopy(self.document) + document["manifest"]["application"]["init_packet_data"] = metadata + with self.subTest(metadata=metadata), self.assertRaisesRegex(ValueError, "metadata"): + guard.verify_application_zip(self.make_zip(document)) + for version in (None, "0.5", [], True, 0, -1, float("inf")): + document = copy.deepcopy(self.document) + document["manifest"]["dfu_version"] = version + with self.subTest(version=version), self.assertRaises(ValueError): + guard.verify_application_zip(self.make_zip(document)) + + def test_application_references_must_be_safe_existing_root_bin_and_dat_files(self): + for key in ("bin_file", "dat_file"): + for name in (None, [], "", "../firmware.bin", "folder/firmware.bin", r"folder\firmware.bin", + "/firmware.bin", "C:firmware.bin", "firmware.hex", "missing.bin"): + document = copy.deepcopy(self.document) + document["manifest"]["application"][key] = name + with self.subTest(key=key, name=name), self.assertRaises(ValueError): + guard.verify_application_zip(self.make_zip(document)) + + def test_extra_duplicate_missing_empty_and_oversized_members_are_refused(self): + body = json.dumps(self.document).encode() + good = [("manifest.json", body), ("firmware.bin", b"app"), ("firmware.dat", b"init")] + for entries in (good + [("bootloader.bin", b"boot")], good + [("firmware.bin", b"duplicate")], + good[:2], good[1:], [good[0], ("other.bin", b"app"), good[2]], + [good[0], ("firmware.bin", b""), good[2]], + [good[0], good[1], ("firmware.dat", b"")], + [good[0], ("firmware.bin", b"x" * (1024 * 1024 + 1)), good[2]], + [good[0], good[1], ("firmware.dat", b"x" * 65537)]): + with self.subTest(members=[entry[0] for entry in entries]), self.assertRaises(ValueError): + guard.verify_application_zip(self.make_zip(entries=entries)) + + def test_symlink_directory_encryption_and_unsupported_compression_are_refused(self): + body = json.dumps(self.document).encode() + for kind in (stat.S_IFLNK, stat.S_IFDIR, stat.S_IFIFO, stat.S_IFCHR): + special = zipfile.ZipInfo("firmware.bin") + special.create_system = 3 + special.external_attr = (kind | 0o777) << 16 + with self.subTest(kind=kind), self.assertRaisesRegex(ValueError, "unsafe"): + guard.verify_application_zip(self.make_zip(entries=[("manifest.json", body), (special, b"target"), ("firmware.dat", b"init")])) + with self.assertRaisesRegex(ValueError, "unsafe"): + guard.verify_application_zip(self.make_zip(entries=[("manifest.json", body), ("folder/", b"folder"), ("firmware.dat", b"init")])) + for property_name, value in (("flag_bits", 1), ("compress_type", zipfile.ZIP_BZIP2)): + artifact = self.make_zip() + with zipfile.ZipFile(artifact) as package: + members = package.infolist() + setattr(members[1], property_name, value) + fake_package = Mock() + fake_package.__enter__ = Mock(return_value=fake_package) + fake_package.__exit__ = Mock(return_value=False) + fake_package.infolist.return_value = members + with self.subTest(property=property_name), patch.object(guard.zipfile, "ZipFile", return_value=fake_package): + with self.assertRaisesRegex(ValueError, "unsafe"): + guard.verify_application_zip(artifact) + fake_package.read.assert_not_called() + + def test_corrupt_zip_and_bin_crc_are_refused(self): + artifact = self.root / "bad.zip" + artifact.write_bytes(b"not a ZIP") + with self.assertRaisesRegex(ValueError, "unreadable"): + guard.verify_application_zip(artifact) + artifact = self.make_zip() + with zipfile.ZipFile(artifact) as package: + entry = package.getinfo("firmware.bin") + data = bytearray(artifact.read_bytes()) + filename_size, extra_size = struct.unpack_from(" OK - Morse stopped\r\n"); diff --git a/tools/hil/morse_bench.py b/tools/hil/morse_bench.py index 441866d8..68b8f1c4 100644 --- a/tools/hil/morse_bench.py +++ b/tools/hil/morse_bench.py @@ -259,7 +259,7 @@ def main(): results = {"frequency_mhz": args.cw_frequency, "packet_radio": PACKET_RADIO, "requested_chip_power_dbm": -9, "partner": args.partner, "direction": args.direction, - "scope": "Actual production applications on this pair, not all radio families.", + "scope": "Experimental HIL applications on this pair; not production speed qualification or all radio families.", "trials": [], "finished": False} devices = {} original_gain = None diff --git a/tools/hil/morse_nrf_flash.sh b/tools/hil/morse_nrf_flash.sh index 2f2f008c..04407b49 100644 --- a/tools/hil/morse_nrf_flash.sh +++ b/tools/hil/morse_nrf_flash.sh @@ -24,6 +24,7 @@ morse_assert_unowned() { } } python3 "$(dirname "$bench_inventory")/verified_nrf_flash.py" \ + --application-only \ --inventory "$bench_inventory" --serial "$bench_serial" --board "$bench_board" \ --artifact "$bench_image" --port "$bench_port" morse_assert_unowned "$bench_port" diff --git a/tools/hil/morse_pi_inventory.py b/tools/hil/morse_pi_inventory.py index 7802db55..665cb4c1 100644 --- a/tools/hil/morse_pi_inventory.py +++ b/tools/hil/morse_pi_inventory.py @@ -1,7 +1,5 @@ """Read-only CLI/tool inventory for the identity-bound Mercerwood Morse bench.""" import json -import os -from pathlib import Path import subprocess import time @@ -11,6 +9,17 @@ from serial.tools import list_ports EXPECTED = {"9AB3B64C641BA927": "RAK4631", "651F8E496197F882": "Heltec T096"} +def port_is_unowned(device): + """Only an explicit, error-free fuser 'no users' result permits access.""" + owner = subprocess.run(["sudo", "-n", "fuser", device], + capture_output=True, text=True, timeout=10) + if owner.returncode == 0: + return False + if owner.returncode == 1 and not owner.stdout.strip() and not owner.stderr.strip(): + return True + raise RuntimeError(f"Cannot verify that {device} is unowned (fuser rc {owner.returncode})") + + def read_quiet(port, limit=2.0): data = bytearray() until = time.monotonic() + limit @@ -24,23 +33,37 @@ def read_quiet(port, limit=2.0): def main(): + refused = False for command in (["hostname"], ["sudo", "-n", "true"], ["find", "/home/mikec/meshfirmware", "-maxdepth", "2", "-iname", "*custom*"], ["ls", "-la", "/home/mikec/meshfirmware/.customfirmware"]): - result = subprocess.run(command, capture_output=True, text=True, timeout=15) + try: + result = subprocess.run(command, capture_output=True, text=True, timeout=15) + except (OSError, subprocess.TimeoutExpired) as exc: + print(json.dumps({"command": command, "refused": True, "err": str(exc)}), flush=True) + if command == ["sudo", "-n", "true"]: + return 1 + continue print(json.dumps({"command": command, "rc": result.returncode, "out": result.stdout, "err": result.stderr}), flush=True) + if command == ["sudo", "-n", "true"] and result.returncode != 0: + return 1 for item in list_ports.comports(): ident = (item.serial_number or "").upper() if ident not in EXPECTED or item.interface == "MeshCore Logging": continue - owner = subprocess.run(["sudo", "-n", "fuser", "-v", item.device], - capture_output=True, text=True, timeout=10) - if owner.stdout.strip(): + try: + unowned = port_is_unowned(item.device) + except (RuntimeError, OSError, subprocess.TimeoutExpired) as exc: + print(json.dumps({"board": EXPECTED[ident], "serial": ident, + "port": item.device, "refused": True, "err": str(exc)}), flush=True) + refused = True + continue + if not unowned: print(json.dumps({"board": EXPECTED[ident], "serial": ident, "port": item.device, "busy": True}), flush=True) continue - port = serial.Serial(port=None, baudrate=115200, timeout=0.03) + port = serial.Serial(port=None, baudrate=115200, timeout=0.03, exclusive=True) port.dtr = True port.rts = False port.port = item.device @@ -51,6 +74,12 @@ def main(): port.write(b"+++MESHCORE-TERM-START\n") hello = read_quiet(port) entered = "WELCOME" in hello + if not entered: + print(json.dumps({"board": EXPECTED[ident], "serial": ident, + "port": item.device, "refused": True, + "err": "Text terminal entry was not confirmed"}), flush=True) + refused = True + continue results = {} for command in ("ver", "board", "get name", "get radio", "get tx", "get radio2.status"): port.write((command + "\n").encode()) @@ -59,10 +88,13 @@ def main(): "port": item.device, "hello": hello, "commands": results}), flush=True) finally: - if entered: - port.write(b"+++MESHCORE-TERM-STOP\n") - port.close() + try: + if entered: + port.write(b"+++MESHCORE-TERM-STOP\n") + finally: + port.close() + return int(refused) if __name__ == "__main__": - main() + raise SystemExit(main()) diff --git a/tools/hil/verified_nrf_flash.py b/tools/hil/verified_nrf_flash.py index f27a2eaa..18077157 100644 --- a/tools/hil/verified_nrf_flash.py +++ b/tools/hil/verified_nrf_flash.py @@ -9,8 +9,13 @@ Run it before either serial DFU or a UF2 copy. import argparse import json +import math import os from pathlib import Path +import re +import stat +import zipfile +import zlib def verify(inventory, serial_number, board, artifact_name): @@ -49,20 +54,98 @@ def verify_port(port_path, serial_number): raise ValueError(f"USB port {port_path} has serial {observed!r}, expected {serial_number}") +def _unique_json_object(pairs): + value = {} + for key, item in pairs: + if key in value: + raise ValueError(f"duplicate manifest key {key!r}") + value[key] = item + return value + + +def _reject_json_constant(value): + raise ValueError(f"invalid manifest JSON constant {value}") + + +def verify_application_zip(artifact): + """Reject every DFU component except one application and its init packet. + + This checks the package's DFU selection, not firmware authenticity. The + existing reviewed board/artifact identity gate remains mandatory too. + Nothing is extracted and no hardware is queried by this check. + """ + try: + with zipfile.ZipFile(artifact) as package: + entries = package.infolist() + names = [entry.filename for entry in entries] + if len(entries) != 3 or len(set(names)) != 3 or "manifest.json" not in names: + raise ValueError("application-only ZIP must contain exactly manifest.json, one BIN and one DAT") + for entry in entries: + if (entry.is_dir() or entry.flag_bits & 1 + or entry.compress_type not in {zipfile.ZIP_STORED, zipfile.ZIP_DEFLATED} + or stat.S_IFMT(entry.external_attr >> 16) not in {0, stat.S_IFREG}): + raise ValueError("application-only ZIP contains an unsupported or unsafe member") + limit = 1024 * 1024 if entry.filename.lower().endswith(".bin") else 65536 + if entry.file_size <= 0 or entry.file_size > limit: + raise ValueError("application-only ZIP members are empty or exceed the size limit") + document = json.loads(package.read("manifest.json").decode("utf-8"), + object_pairs_hook=_unique_json_object, + parse_constant=_reject_json_constant) + if not isinstance(document, dict) or set(document) != {"manifest"}: + raise ValueError("application-only ZIP has an invalid manifest root") + manifest = document["manifest"] + if (not isinstance(manifest, dict) or "application" not in manifest + or set(manifest) - {"application", "dfu_version"}): + raise ValueError("application-only ZIP forbids bootloader, SoftDevice and unknown DFU components") + if "dfu_version" in manifest: + version = manifest["dfu_version"] + if (type(version) not in {int, float} or not math.isfinite(version) + or version <= 0): + raise ValueError("application-only ZIP has an invalid DFU version") + application = manifest["application"] + if (not isinstance(application, dict) + or not {"bin_file", "dat_file"}.issubset(application) + or set(application) - {"bin_file", "dat_file", "init_packet_data"}): + raise ValueError("application-only ZIP has an invalid application entry") + if ("init_packet_data" in application + and not isinstance(application["init_packet_data"], dict)): + raise ValueError("application-only ZIP has invalid init-packet metadata") + selected = [] + for key, extension in (("bin_file", "bin"), ("dat_file", "dat")): + name = application[key] + if (not isinstance(name, str) + or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*\." + extension, name, + flags=re.IGNORECASE)): + raise ValueError("application BIN/DAT references must be safe root-level filenames") + selected.append(name) + if set(names) != {"manifest.json", *selected}: + raise ValueError("application-only ZIP members do not match the application BIN/DAT references") + if package.testzip() is not None: + raise ValueError("application-only ZIP has a corrupt member") + return tuple(selected) + except (zipfile.BadZipFile, UnicodeDecodeError, NotImplementedError, RuntimeError, + EOFError, zlib.error) as exc: + raise ValueError(f"invalid or unreadable application-only ZIP: {exc}") from exc + + def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--inventory", type=Path, required=True) parser.add_argument("--serial", required=True, help="Stable USB chip serial") parser.add_argument("--board", required=True, help="Reviewed physical board name") parser.add_argument("--artifact", type=Path, required=True) + parser.add_argument("--application-only", action="store_true", + help="Require a ZIP with only one application BIN/DAT; reject bootloader/SoftDevice updates") parser.add_argument("--port", required=True, help="Current application or DFU serial port") args = parser.parse_args(argv) try: inventory = json.loads(args.inventory.read_text()) verify(inventory, args.serial, args.board, args.artifact.name) - verify_port(args.port, args.serial) if not args.artifact.is_file(): raise ValueError(f"artifact does not exist: {args.artifact}") + if args.application_only: + verify_application_zip(args.artifact) + verify_port(args.port, args.serial) except (ValueError, OSError, KeyError, json.JSONDecodeError) as exc: parser.exit(1, f"REFUSED: {exc}\n") print(f"VERIFIED: {args.serial} = {args.board}; {args.artifact.name}; {args.port}")