diff --git a/test/test_host_kiss_ingress.py b/test/test_host_kiss_ingress.py index 394fbd2b0..9486cfac5 100644 --- a/test/test_host_kiss_ingress.py +++ b/test/test_host_kiss_ingress.py @@ -177,8 +177,8 @@ class HostKissIngressTest(unittest.TestCase): signature = ('static void __attribute__((noinline)) serviceCommandInterfaces()' if role == 'simple_repeater' else 'void loop()') pump = body(source, signature) - self.assertLess(pump.index('mesh::hostkiss::usbInputReserved(console.peek())'), - pump.index('console.read()')) + read = 'const int value = console_read()' if role == 'simple_repeater' else 'console.read()' + self.assertLess(pump.index('mesh::hostkiss::usbInputReserved(console.peek())'), pump.index(read)) if __name__ == '__main__': diff --git a/test/test_repeater_ota_serial_control.py b/test/test_repeater_ota_serial_control.py index c56577a5e..447fde45a 100644 --- a/test/test_repeater_ota_serial_control.py +++ b/test/test_repeater_ota_serial_control.py @@ -15,12 +15,16 @@ HARNESS = r''' #include #include using namespace mesh::ota; +namespace mesh { +bool kiss_owned=false; +bool isUsbBinaryStreamOwned(){return kiss_owned;} +} struct Wire:Stream { std::deque input; std::vector next_response; - int writes=0; + int writes=0,reads=0; int available()override{return input.size();} - int read()override{if(input.empty())return -1;auto b=input.front();input.pop_front();return b;} + int read()override{++reads;if(input.empty())return -1;auto b=input.front();input.pop_front();return b;} size_t write(const uint8_t* data,size_t n)override{ assert(n>=4&&data[0]=='M'&&data[1]=='S');++writes; input.insert(input.end(),next_response.begin(),next_response.end());next_response.clear();return n; @@ -39,24 +43,33 @@ static std::string controls(SerialMotaSource& source){ } return output; } +static bool kissClaimAllowed(SerialMotaSource& control_source){ + @KISS_GATE@ + return kiss_can_read(); +} int main(){ const std::string command="ota.peer cmd "+std::string(64,'1')+" ota key add "+std::string(64,'A')+"\r"; const std::string dangerous="\rota.peer cmd "+std::string(64,'2')+" reboot\r\nota folder off\r"; - const std::vector payload(dangerous.begin(),dangerous.end()); + std::vector payload={0xc0,0,0x41,0xc0}; + payload.insert(payload.end(),dangerous.begin(),dangerous.end()); const auto response=frame(payload); for(size_t prefix=0;prefix target(payload.size()); assert(!source.read(0,0,target.data(),target.size())); assert(source.hasPendingResponse()); + assert(!kissClaimAllowed(source)); for(size_t i=prefix;i str: """A harmless text probe avoids feeding Binary commands to a repeater CLI.""" import serial connection = serial.Serial(port=None, baudrate=115200, timeout=.1, write_timeout=2) - connection.dtr, connection.rts = True, False - connection.port = port try: - connection.open() + open_serial_session(connection, port) time.sleep(.75) connection.write(b"\rver\rota status\r") deadline = time.monotonic() + 3 diff --git a/tools/lora_ota/ota_wizard_companion.py b/tools/lora_ota/ota_wizard_companion.py index 00f2b6c73..f61cecdb2 100644 --- a/tools/lora_ota/ota_wizard_companion.py +++ b/tools/lora_ota/ota_wizard_companion.py @@ -68,6 +68,52 @@ def _deflate(data): return compressor.compress(data) + compressor.flush() +async def _create_usb_sdk(port, baudrate=115200, *, only_error=True, + default_timeout=15, auto_reconnect=False): + """Use SDK framing with control lines selected before the USB port opens.""" + import serial + import serial_asyncio_fast + from meshcore import MeshCore, SerialConnection + from ota_wizard_usb import open_serial_session + + class ConfiguredConnection(SerialConnection): + async def connect(self, timeout=10): + self._connected_event.clear() + wire = serial.Serial(port=None, baudrate=self.baudrate, write_timeout=2) + try: + # The pinned SDK factory cannot configure DTR before opening. + # Adopt an already configured connection through the serial + # transport's public API; keep the SDK protocol and parser. + await asyncio.to_thread(open_serial_session, wire, self.port) + loop = asyncio.get_running_loop() + transport, _ = await serial_asyncio_fast.connection_for_serial( + loop, lambda: self.MCSerialClientProtocol(self), wire) + self.transport = transport + await asyncio.wait_for(self._connected_event.wait(), timeout) + return self.port + except BaseException: + if self.transport is not None: + self.transport.close() + self.transport = None + await asyncio.to_thread(wire.close) + raise + + connection = ConfiguredConnection(port, baudrate, cx_dly=.1) + sdk = MeshCore(connection, only_error=only_error, default_timeout=default_timeout, + auto_reconnect=auto_reconnect) + try: + if await sdk.connect() is None: + await sdk.disconnect() + return None + return sdk + except BaseException: + try: + await sdk.disconnect() + except Exception: + pass + raise + + class CompanionTransport: def __init__(self, port, baud=115200, *, command_timeout=15, remote_timeout=60, discovery_seconds=4, sdk_factory=None, @@ -126,10 +172,10 @@ class CompanionTransport: return self if self._sdk_factory is None: try: - from meshcore import MeshCore, EventType + from meshcore import EventType except ImportError as exc: raise CompanionError("Install the meshcore Python package for USB Companion support") from exc - self._sdk_factory, self._event_types = MeshCore.create_serial, EventType + self._sdk_factory, self._event_types = _create_usb_sdk, EventType if self._event_types is None: raise CompanionError("Companion event definitions are unavailable") try: diff --git a/tools/lora_ota/ota_wizard_repeater.py b/tools/lora_ota/ota_wizard_repeater.py index 04285e53e..61976f60a 100644 --- a/tools/lora_ota/ota_wizard_repeater.py +++ b/tools/lora_ota/ota_wizard_repeater.py @@ -15,6 +15,7 @@ import time import zlib import lora_ota as ota +from ota_wizard_usb import open_serial_session class RepeaterError(ota.OtaError): @@ -178,9 +179,7 @@ class RepeaterTransport: def create(): connection = factory(port=None, baudrate=self.baud, timeout=.1, write_timeout=3, rtscts=False, dsrdtr=False) - connection.dtr, connection.rts = True, False - connection.port = self.port - connection.open() + open_serial_session(connection, self.port) return connection try: self._serial = await asyncio.to_thread(create) diff --git a/tools/lora_ota/ota_wizard_usb.py b/tools/lora_ota/ota_wizard_usb.py new file mode 100644 index 000000000..5a21e547f --- /dev/null +++ b/tools/lora_ota/ota_wizard_usb.py @@ -0,0 +1,74 @@ +"""Select USB control-line states before opening a radio's serial port.""" +from __future__ import annotations + +import os +import re + + +def _device_path(value) -> str: + text = str(value) + # Windows callers may use either COM7 or the equivalent \\.\COM7. + match = re.fullmatch(r"(?:\\\\\.\\)?COM(\d+)", text, re.I) + if match: + return "com" + str(int(match[1])) + return os.path.realpath(text).casefold() + + +def serial_line_states(port, *, ports=None) -> tuple[bool, bool]: + """Return (DTR, RTS), resolving aliases against the selected USB device. + + ESP32 Serial/JTAG 303A:1001 and UART bridges must leave DTR released + across firmware reboots. Other Espressif CDC, Seeed and Adafruit/nRF52 + CDC devices need DTR asserted. Unknown ports use the released state. + """ + if ports is None: + try: + from serial.tools import list_ports + ports = list_ports.comports() + except (ImportError, OSError): + ports = () + selected = _device_path(port) + for info in ports: + device = getattr(info, "device", None) + if device is None or _device_path(device) != selected: + continue + vid, pid = getattr(info, "vid", None), getattr(info, "pid", None) + return vid in (0x303A, 0x2886, 0x239A) and (vid, pid) != (0x303A, 0x1001), False + return False, False + + +def configure_serial_session(connection, port=None, *, ports=None) -> tuple[bool, bool]: + """Configure a closed pySerial-compatible object without toggling a device.""" + if getattr(connection, "is_open", False): + raise ValueError("USB control lines must be configured before opening the port") + if port is not None: + connection.port = port + dtr, rts = serial_line_states(connection.port, ports=ports) + connection.rts = rts + connection.dtr = dtr + return dtr, rts + + +def open_serial_session(connection, port=None, *, ports=None) -> None: + """Open configured lines without a POSIX USB-UART bridge reset pulse.""" + configure_serial_session(connection, port, ports=ports) + # POSIX pySerial normally applies DTR before RTS. Releasing DTR while + # the driver still asserts RTS can pulse ESP32 EN. Defer DTR until RTS + # has been released, matching tools/hil/serial_session.py. + defer_dtr = os.name == "posix" and not connection.dtr + previous_dsrdtr = getattr(connection, "dsrdtr", False) + try: + if defer_dtr: + connection.dsrdtr = True + connection.open() + if defer_dtr: + connection.rts = False + connection.dtr = False + connection.dsrdtr = previous_dsrdtr + except BaseException: + try: + connection.close() + finally: + if defer_dtr: + connection.dsrdtr = previous_dsrdtr + raise diff --git a/tools/lora_ota/test_ota_wizard_companion.py b/tools/lora_ota/test_ota_wizard_companion.py index 3b9b1715b..b496c3327 100644 --- a/tools/lora_ota/test_ota_wizard_companion.py +++ b/tools/lora_ota/test_ota_wizard_companion.py @@ -3,6 +3,7 @@ import asyncio from copy import deepcopy from enum import Enum from types import SimpleNamespace +import sys import unittest from unittest import mock import time @@ -460,5 +461,112 @@ class PureHelpers(unittest.TestCase): companion.packet_airtime(20, (910, 62.5, 7, 5))) +class UsbConstructionTests(unittest.IsolatedAsyncioTestCase): + async def create_sdk(self, dtr, *, handshake=True): + import ota_wizard_usb as usb + history = [] + + class Wire: + def __init__(self, port, baudrate, write_timeout): + self.port, self.baudrate = port, baudrate + self.is_open, self.dsrdtr = False, False + self._dtr = self._rts = True + history.append(("construct", port)) + + @property + def dtr(self): + return self._dtr + + @dtr.setter + def dtr(self, value): + history.append(("dtr", value, self.is_open)) + self._dtr = value + + @property + def rts(self): + return self._rts + + @rts.setter + def rts(self, value): + history.append(("rts", value, self.is_open)) + self._rts = value + + def open(self): + history.append(("open", self.dtr, self.rts)) + self.is_open = True + + def close(self): + history.append(("close",)) + self.is_open = False + + class SerialConnection: + def __init__(self, port, baudrate, cx_dly): + self.port, self.baudrate = port, baudrate + self.transport = None + self._connected_event = asyncio.Event() + + class MCSerialClientProtocol: + def __init__(self, connection): + self.connection = connection + + def connection_made(self, transport): + self.connection.transport = transport + transport.serial.rts = False + self.connection._connected_event.set() + + class MeshCore: + last = None + def __init__(self, connection, **settings): + MeshCore.last = self + self.connection, self.settings = connection, settings + + async def connect(self): + await self.connection.connect() + return event() if handshake else None + + async def disconnect(self): + if self.connection.transport: + self.connection.transport.close() + self.connection.transport = None + + async def adopt(loop, protocol_factory, wire): + self.assertTrue(wire.is_open) + self.assertEqual((wire.dtr, wire.rts), (dtr, False)) + history.append(("adopt", wire.dtr, wire.rts)) + protocol = protocol_factory() + transport = SimpleNamespace(serial=wire, close=wire.close) + protocol.connection_made(transport) + return transport, protocol + + modules = {"meshcore": SimpleNamespace(MeshCore=MeshCore, SerialConnection=SerialConnection), + "serial": SimpleNamespace(Serial=Wire), + "serial_asyncio_fast": SimpleNamespace(connection_for_serial=adopt)} + with mock.patch.dict(sys.modules, modules), \ + mock.patch.object(usb, "serial_line_states", return_value=(dtr, False)): + sdk = await companion._create_usb_sdk("USB-TEST", default_timeout=.1) + return sdk, MeshCore.last, history + + async def test_native_usb_dtr_is_false_before_sdk_adopts_connection(self): + sdk, instance, history = await self.create_sdk(False) + self.assertIs(sdk, instance) + self.assertLess(history.index(("dtr", False, False)), history.index(("open", False, False))) + self.assertFalse(any(item[0] == "dtr" and item[1] for item in history)) + self.assertFalse(sdk.settings["auto_reconnect"]) + await sdk.disconnect() + self.assertEqual(history[-1], ("close",)) + + async def test_tinyusb_nrf_dtr_is_true_before_open(self): + sdk, _, history = await self.create_sdk(True) + self.assertLess(history.index(("dtr", True, False)), history.index(("open", True, False))) + await sdk.disconnect() + self.assertEqual(history[-1], ("close",)) + + async def test_failed_binary_handshake_closes_the_configured_port(self): + sdk, instance, history = await self.create_sdk(False, handshake=False) + self.assertIsNone(sdk) + self.assertIsNone(instance.connection.transport) + self.assertEqual(history[-1], ("close",)) + + if __name__ == "__main__": unittest.main() diff --git a/tools/lora_ota/test_ota_wizard_session.py b/tools/lora_ota/test_ota_wizard_session.py index 249e5fce2..f1ea9a91f 100644 --- a/tools/lora_ota/test_ota_wizard_session.py +++ b/tools/lora_ota/test_ota_wizard_session.py @@ -262,6 +262,9 @@ class SessionTests(unittest.IsolatedAsyncioTestCase): await self.session.cleanup() async def test_destination_temporarily_missing_after_reboot_is_probed_again(self): + import itertools + from types import SimpleNamespace + from unittest import mock from ota_wizard_companion import CompanionError before = await self.session.inspect() original = self.link.remote_command @@ -273,7 +276,11 @@ class SessionTests(unittest.IsolatedAsyncioTestCase): raise CompanionError("No directly reachable destination replied yet") return await original(key, password, text) self.link.remote_command = reboot_discovery - after = await self.session.install(wizard.full_package(application(), 1024), before) + # Advance only this module's deadline clock; compiler load must not + # exhaust the tiny test reboot budget between three fake probes. + clock = SimpleNamespace(monotonic=itertools.count(0, .001).__next__) + with mock.patch.object(wizard, "time", clock): + after = await self.session.install(wizard.full_package(application(), 1024), before) self.assertEqual(after.version, NEW) self.assertEqual(missing, 0) self.assertEqual(len(self.link.reset_calls), 3) diff --git a/tools/lora_ota/test_ota_wizard_usb.py b/tools/lora_ota/test_ota_wizard_usb.py new file mode 100644 index 000000000..4eec3ae80 --- /dev/null +++ b/tools/lora_ota/test_ota_wizard_usb.py @@ -0,0 +1,186 @@ +"""USB enumeration policy and pre-open line ordering without hardware access.""" +import sys +from types import ModuleType, SimpleNamespace +import unittest +from unittest import mock + +import ota_wizard as wizard +import ota_wizard_usb as usb +from ota_wizard_repeater import RepeaterTransport +from test_ota_wizard_repeater import FakeSerial + + +def metadata(device="COM7", vid=0x303A, pid=0x1001): + return SimpleNamespace(device=device, vid=vid, pid=pid) + + +class SerialFixture: + def __init__(self, **kwargs): + self.is_open = False + self.events = [] + self.port = None + self._rts = self._dtr = True + self._dsrdtr = False + self.open_error = None + + @property + def dtr(self): + return self._dtr + + @dtr.setter + def dtr(self, value): + self.events.append(("dtr", value, self.is_open)) + self._dtr = value + + @property + def rts(self): + return self._rts + + @rts.setter + def rts(self, value): + self.events.append(("rts", value, self.is_open)) + self._rts = value + + @property + def dsrdtr(self): + return self._dsrdtr + + @dsrdtr.setter + def dsrdtr(self, value): + self.events.append(("dsrdtr", value, self.is_open)) + self._dsrdtr = value + + def open(self): + self.events.append(("open", self.dtr, self.rts, self.dsrdtr, self.port)) + self.is_open = True + if self.open_error: + raise self.open_error + + def close(self): + self.events.append(("close",)) + self.is_open = False + + def write(self, data): + self.events.append(("write", data)) + return len(data) + + def read(self, length): + return b" -> v1.18.0.1\r\n" + + +class UsbPolicyTests(unittest.TestCase): + def test_vid_pid_matrix_dtr_with_rts_always_released(self): + cases = ((0x303A, 0x1001, False), (0x303A, 0x4001, True), + (0x2886, 0x0044, True), (0x239A, 0x8029, True), + (0x10C4, 0xEA60, False), (0x1A86, 0x7523, False), + (0x0403, 0x6001, False), (None, None, False)) + for vid, pid, dtr in cases: + with self.subTest(vid=vid, pid=pid): + ports = [metadata(vid=vid, pid=pid)] + self.assertEqual(usb.serial_line_states("COM7", ports=ports), (dtr, False)) + self.assertEqual(usb.serial_line_states("COM8", ports=[metadata(vid=0x239A)]), + (False, False)) + + def test_windows_com_aliases_and_posix_by_id_resolve_same_device(self): + self.assertEqual(usb.serial_line_states(r"\\.\COM7", ports=[metadata(vid=0x239A)]), + (True, False)) + self.assertEqual(usb.serial_line_states("com7", ports=[metadata(vid=0x2886)]), + (True, False)) + def resolved(value): + return "/dev/ttyACM0" if value == "/dev/serial/by-id/radio" else value + with mock.patch.object(usb.os.path, "realpath", side_effect=resolved): + self.assertEqual(usb.serial_line_states("/dev/serial/by-id/radio", + ports=[metadata("/dev/ttyACM0", vid=0x2886)]), (True, False)) + + def test_lazy_enumeration_and_missing_metadata_default_to_released(self): + serial, tools = ModuleType("serial"), ModuleType("serial.tools") + listing = mock.Mock(return_value=[metadata(vid=0x239A)]) + tools.list_ports = SimpleNamespace(comports=listing) + serial.tools = tools + with mock.patch.dict(sys.modules, {"serial": serial, "serial.tools": tools}): + self.assertEqual(usb.serial_line_states("COM7"), (True, False)) + listing.assert_called_once_with() + listing.side_effect = OSError("USB device disappeared") + self.assertEqual(usb.serial_line_states("COM7"), (False, False)) + + def test_configuration_happens_before_open_and_rejects_open_object(self): + port = SerialFixture() + self.assertEqual(usb.configure_serial_session(port, "COM7", ports=[metadata()]), + (False, False)) + self.assertEqual(port.events, [("rts", False, False), ("dtr", False, False)]) + self.assertEqual(port.port, "COM7") + port.is_open = True + before = list(port.events) + with self.assertRaisesRegex(ValueError, "before opening"): + usb.configure_serial_session(port, "COM8", ports=[]) + self.assertEqual(port.events, before) + self.assertEqual(port.port, "COM7") + + def test_posix_release_defers_dtr_until_after_rts_without_handshaking_left_on(self): + port = SerialFixture() + with mock.patch.object(usb, "os", SimpleNamespace(name="posix")), \ + mock.patch.object(usb, "serial_line_states", return_value=(False, False)): + usb.open_serial_session(port, "/dev/ttyACM0") + self.assertEqual(port.events, [ + ("rts", False, False), ("dtr", False, False), ("dsrdtr", True, False), + ("open", False, False, True, "/dev/ttyACM0"), + ("rts", False, True), ("dtr", False, True), ("dsrdtr", False, True)]) + self.assertFalse(port.dsrdtr) + + def test_native_cdc_and_windows_open_without_posix_bridge_sequence(self): + for platform, dtr in (("posix", True), ("nt", True), ("nt", False)): + port = SerialFixture() + with self.subTest(platform=platform, dtr=dtr), \ + mock.patch.object(usb, "os", SimpleNamespace(name=platform)), \ + mock.patch.object(usb, "serial_line_states", return_value=(dtr, False)): + usb.open_serial_session(port, "COM7") + self.assertEqual(port.events, [("rts", False, False), ("dtr", dtr, False), + ("open", dtr, False, False, "COM7")]) + + def test_open_failure_closes_device_and_restores_handshaking(self): + port = SerialFixture() + port.open_error = OSError("open failed") + with mock.patch.object(usb, "os", SimpleNamespace(name="posix")), \ + mock.patch.object(usb, "serial_line_states", return_value=(False, False)), \ + self.assertRaisesRegex(OSError, "open failed"): + usb.open_serial_session(port, "/dev/ttyUSB0") + self.assertFalse(port.is_open) + self.assertFalse(port.dsrdtr) + self.assertIn(("close",), port.events) + + def test_detect_mode_obeys_shared_policy_before_first_probe(self): + serial = ModuleType("serial") + connection = SerialFixture() + serial.Serial = mock.Mock(return_value=connection) + with mock.patch.dict(sys.modules, {"serial": serial}), \ + mock.patch.object(usb, "serial_line_states", return_value=(False, False)), \ + mock.patch.object(wizard.time, "sleep"), \ + mock.patch.object(wizard.time, "monotonic", side_effect=[0, 0, 4]): + self.assertEqual(wizard.detect_mode("COM7"), "repeater") + opened = next(event for event in connection.events if event[0] == "open") + self.assertEqual(opened[1:3], (False, False)) + self.assertFalse(connection.is_open) + + +class RepeaterUsbPolicyTests(unittest.IsolatedAsyncioTestCase): + async def test_repeater_opens_with_shared_policy_for_both_device_types(self): + for dtr in (False, True): + connection = FakeSerial() + observed = [] + def open_port(): + observed.append((connection.dtr, connection.rts, connection.port)) + connection.is_open = True + connection.open = open_port + link = RepeaterTransport("COM7", open_settle=0, + serial_factory=lambda **kwargs: connection) + with self.subTest(dtr=dtr), \ + mock.patch.object(usb, "serial_line_states", return_value=(dtr, False)): + try: + await link.open() + self.assertEqual(observed, [(dtr, False, "COM7")]) + finally: + await link.close() + + +if __name__ == "__main__": + unittest.main()