From 6eeab56779bc2294e4fb2dd426876cbcf009c237 Mon Sep 17 00:00:00 2001 From: agessaman Date: Sun, 26 Jul 2026 19:59:15 -0700 Subject: [PATCH] fix(ble): bound the write, and correct two reply-path edge cases Follow-up to review of the two preceding commits. The write lock added in 00135bb prevented concurrent writes from dropping the link, but bounded nothing. CommandHandler's own timeout does not cover the write: send() awaits _sender_func() and only afterwards arms asyncio.wait(futures, timeout=...). So a stalled write -- observed on hardware running to minutes -- held the lock indefinitely while every other command queued behind it, with nothing logged, no error raised and no DISCONNECTED event. Before the lock only the stalled command hung; the others went out and could trip the error-19 disconnect, which at least recovered. The lock turned a bounded, self-healing failure into an unbounded silent one, and also blocked the post-reconnect CMD_APP_START behind the dead connection's holder. The write (lock acquisition included) is now bounded by BLEConnection.WRITE_TIMEOUT. On expiry the link is torn down rather than the lock merely released: the underlying CoreBluetooth write may still be in flight, and a second write racing it re-creates the overlap the lock exists to prevent. Tearing down hands over to the reconnect path, which is bounded and self-healing. The lock is now a lazily-created property, mirroring _mesh_request_lock in commands/base.py, so an instance built without __init__ still works. The two BLE tests previously assigned _write_lock themselves, which meant deleting the __init__ line left them green; there is now a test that __init__ provides it. Also in send_anon_req: - A failed change_contact_path() no longer proceeds. The device still has the contact as flood, so sendAnonReq() floods the request, and the server gates REGIONS/OWNER/BASIC behind isRouteDirect() and drops it -- the caller then waits out a full path-scaled timeout for a reply that cannot arrive. It now returns ERROR path_reset_failed. - encode_reply_path() clamps to the server's 64-byte reply_path buffer, which MyMesh.cpp memcpys into with no length check, and rejects hash mode 3 (the 4-byte hops Packet::isValidPathLen refuses). Not a regression -- the previous encoder overflowed identically -- but this function is the chokepoint and its comment claimed to bound the read. - The hop count saturates at 63 instead of being masked with & 63, which would wrap a 64-hop path to zero hops, i.e. request a zero-hop reply from a distant node. Not reachable from a device-sourced contact (the reader caps the field at 63) but silent if it ever were. - The suggested_timeout multiplier now scales by the hops actually emitted rather than the contact's claimed out_path_len, which can differ once the encoder clamps or truncates. Correction to 30446ed's message: the claim that mode 0 is "byte-for-byte identical" is wrong. Differentially, over 30000 randomised contact fields restricted to what a device can actually emit, mode 0 diverges in 1177 of 10118 cases -- every one of them a path containing a 0x00 byte, and in every one the old encoder was the wrong one. The accurate claim is "unchanged for mode-0 paths containing no 0x00 byte". --- src/meshcore/ble_cx.py | 58 ++++++++--- src/meshcore/commands/base.py | 45 ++++++++- tests/unit/test_error_handling.py | 153 ++++++++++++++++++++++++++---- 3 files changed, 223 insertions(+), 33 deletions(-) diff --git a/src/meshcore/ble_cx.py b/src/meshcore/ble_cx.py index 7cee63f..b0f7db2 100644 --- a/src/meshcore/ble_cx.py +++ b/src/meshcore/ble_cx.py @@ -28,6 +28,11 @@ UART_RX_CHAR_UUID = "6E400002-B5A3-F393-E0A9-E50E24DCCA9E" UART_TX_CHAR_UUID = "6E400003-B5A3-F393-E0A9-E50E24DCCA9E" class BLEConnection: + # Upper bound on a single write (lock acquisition included). Healthy writes + # measured 0.06-0.17s against real hardware; observed stalls ran 20s to + # minutes, so this preempts them rather than waiting for CoreBluetooth. + WRITE_TIMEOUT = 10.0 + def __init__(self, address=None, device=None, client=None, pin=None): """ Constructor: specify address or an existing BleakClient. @@ -53,13 +58,27 @@ class BLEConnection: self.rx_char = None self._disconnect_callback = None self._background_tasks: set[asyncio.Task] = set() - # Serialises write_gatt_char(). Two overlapping writes to the same - # characteristic drop the link outright (observed on macOS/CoreBluetooth: - # "BLE write failed: 19", connection gone). Nothing above this layer - # guarantees callers are sequential -- schedulers, health checks and - # user commands all issue commands independently -- so the transport has - # to enforce it. Lazily created so it binds to the running loop. - self._write_lock: Optional[asyncio.Lock] = None + self._write_lock_obj: Optional[asyncio.Lock] = None + + @property + def _write_lock(self) -> asyncio.Lock: + """Serialises write_gatt_char(). + + Two overlapping writes to the same characteristic drop the link outright + (observed on macOS/CoreBluetooth: "BLE write failed: 19", connection + gone). Nothing above this layer guarantees callers are sequential -- + schedulers, health checks and user commands all issue independently -- + so the transport has to enforce it. + + Lazily created so it binds to the running loop, mirroring the + _mesh_request_lock property in commands/base.py. Read through getattr so + an instance built without __init__ still works. + """ + lock = getattr(self, "_write_lock_obj", None) + if lock is None: + lock = asyncio.Lock() + self._write_lock_obj = lock + return lock def _spawn_background(self, coro) -> asyncio.Task: """Create a tracked background task (prevents GC of fire-and-forget tasks).""" @@ -198,6 +217,10 @@ class BLEConnection: if self.reader is not None: self._spawn_background(self.reader.handle_rx(data)) + async def _write_locked(self, data): + async with self._write_lock: + await self.client.write_gatt_char(self.rx_char, bytes(data), response=True) + async def send(self, data): if not self.client: logger.error("Client is not connected") @@ -207,11 +230,24 @@ class BLEConnection: if not self.rx_char: logger.error("RX characteristic not found") return False - if self._write_lock is None: - self._write_lock = asyncio.Lock() + # Bound the whole acquire-plus-write. A stalled write has been seen to + # hang for minutes, and CommandHandler's own timeout does not cover this + # -- it starts only after _sender_func returns -- so without a bound the + # serialising lock would queue every other command behind the stall + # indefinitely, with nothing logged and no disconnect raised. Turning one + # hung command into a silent whole-client stall would be worse than the + # overlap the lock exists to prevent. try: - async with self._write_lock: - await self.client.write_gatt_char(self.rx_char, bytes(data), response=True) + await asyncio.wait_for(self._write_locked(data), timeout=self.WRITE_TIMEOUT) + except asyncio.TimeoutError: + # Do not simply release and carry on: the underlying write may still + # be in flight, and a second write racing it re-creates the exact + # overlap that kills the link. Tear the connection down so the + # reconnect path takes over -- bounded and self-healing. + logger.warning(f"BLE write timed out after {self.WRITE_TIMEOUT}s") + if self._disconnect_callback: + await self._disconnect_callback("ble_write_timeout") + return False except Exception as exc: logger.warning(f"BLE write failed: {exc}") if self._disconnect_callback: diff --git a/src/meshcore/commands/base.py b/src/meshcore/commands/base.py index 2ed4a59..5e4216a 100644 --- a/src/meshcore/commands/base.py +++ b/src/meshcore/commands/base.py @@ -57,6 +57,13 @@ def _validate_destination(dst: DestinationType, prefix_length: int = 6) -> bytes ) +# Size of the server-side reply_path buffer (uint8_t reply_path[64] in +# simple_repeater/MyMesh.h). It is memcpy'd into without a length check. +MAX_REPLY_PATH_BYTES = 64 +# reply_path_len is the low 6 bits of the header byte. +MAX_REPLY_PATH_HOPS = 63 + + def encode_reply_path(out_path_len: int, out_path_hex: str, out_path_hash_mode: int) -> bytes: """Encode the reply path a server should use when answering us. @@ -71,17 +78,36 @@ def encode_reply_path(out_path_len: int, out_path_hex: str, out_path_hash_mode: The path itself is reversed by *hop*, not by byte: a return path visits the same hops in the opposite order, and each hop's multi-byte hash must stay - intact. (At hash mode 0 the two are indistinguishable, which is why this - went unnoticed - mode 0 is the default.) + intact. (For single-byte hops the two are indistinguishable, which is most + of why this went unnoticed - mode 0 is the default.) """ hash_mode = max(out_path_hash_mode, 0) # -1 means "flood", i.e. no path + if hash_mode > 2: + # The server computes hash_size = mode + 1, and Packet::isValidPathLen + # rejects 4-byte hops outright, so such a path is unusable on the wire. + logger.warning( + f"Unsupported out_path_hash_mode {out_path_hash_mode}; " + "requesting a zero-hop reply path instead" + ) + return b"\x00" hash_size = hash_mode + 1 - hops = max(out_path_len, 0) & 63 + # Saturate rather than mask: `& 63` would silently wrap a 64-hop path to + # zero hops, i.e. a zero-hop reply for a distant node. + hops = min(max(out_path_len, 0), MAX_REPLY_PATH_HOPS) raw = bytes.fromhex(out_path_hex or "") # Never read past what the contact actually carries; a truncated or padded # field would otherwise yield short trailing hops. hops = min(hops, len(raw) // hash_size) + # The server memcpys into a fixed 64-byte reply_path with no bounds check + # (simple_repeater/MyMesh.cpp), so never describe more than fits. + max_hops = min(MAX_REPLY_PATH_HOPS, MAX_REPLY_PATH_BYTES // hash_size) + if hops > max_hops: + logger.warning( + f"Reply path of {hops} hops x {hash_size}B exceeds the " + f"{MAX_REPLY_PATH_BYTES}B the server can hold; truncating to {max_hops}" + ) + hops = max_hops path = b"".join( raw[i * hash_size:(i + 1) * hash_size] for i in range(hops - 1, -1, -1) @@ -362,7 +388,15 @@ class CommandHandlerBase: if contact["out_path_len"] == -1: logger.info("No path set trying zero hop") zero_hop = True - await self.change_contact_path(contact, "") + path_res = await self.change_contact_path(contact, "") + if path_res is not None and path_res.type == EventType.ERROR: + # The device still has this contact as flood, so sendAnonReq + # will flood the request -- and the server gates REGIONS, + # OWNER and BASIC behind isRouteDirect(), silently dropping + # it. Better to fail here than to wait out a full timeout + # for a reply that cannot come. + logger.error("Could not set zero-hop path, aborting anon request") + return Event(EventType.ERROR, {"reason": "path_reset_failed"}) # update_contact() normally reflects the change back onto the dict, so # out_path_len reads 0 here. Clamp anyway: if that call failed (e.g. the # device query inside it errored) the dict is still -1, and the unsigned @@ -388,7 +422,8 @@ class CommandHandlerBase: exp_tag = result.payload["expected_ack"].hex() # Use provided timeout or fallback to suggested timeout (with 5s default) - result.payload["suggested_timeout"] = result.payload.get("suggested_timeout", 4000) * (out_path_len + 1) # update timeout from path_len + emitted_hops = reply_path[0] & 63 # what actually went on the wire + result.payload["suggested_timeout"] = result.payload.get("suggested_timeout", 4000) * (emitted_hops + 1) # update timeout from path_len actual_timeout = timeout if timeout is not None and timeout > 0 else result.payload.get("suggested_timeout", 4000) / 800.0 actual_timeout = min_timeout if actual_timeout < min_timeout else actual_timeout self._reader.register_binary_request(pubkey_prefix.hex(), exp_tag, request_type, actual_timeout, context=context, is_anon=True) diff --git a/tests/unit/test_error_handling.py b/tests/unit/test_error_handling.py index 2206895..3153795 100644 --- a/tests/unit/test_error_handling.py +++ b/tests/unit/test_error_handling.py @@ -368,34 +368,30 @@ async def test_send_anon_req_flood_contact_still_forces_and_restores_zero_hop( assert sent == b"\x39" + bytes.fromhex(VALID_PUBKEY_HEX) + b"\x01" + b"\x00" -async def test_send_anon_req_survives_change_contact_path_not_mutating( +async def test_send_anon_req_aborts_when_change_contact_path_fails( command_handler, mock_connection, mock_dispatcher ): - """A failed change_contact_path must not crash and must still restore the path. + """A failed zero-hop switch must abort, not send an unanswerable request. - update_contact() normally writes out_path_len back onto the dict, but if it - errors the value stays -1. Unclamped, the unsigned to_bytes raises - OverflowError, which skips reset_path and leaves the contact pinned to - zero-hop on the device. + If change_contact_path() fails the device still has the contact as flood, so + sendAnonReq() floods the request -- and the server gates REGIONS/OWNER/BASIC + behind isRouteDirect(), dropping it silently. Sending anyway would burn a + full path-scaled timeout waiting for a reply that cannot arrive. """ contact = {"public_key": VALID_PUBKEY_HEX, "out_path_len": -1, "out_path": ""} command_handler._get_contact_by_prefix = MagicMock(return_value=contact) - # Returns an error without reflecting the change onto the dict. command_handler.change_contact_path = AsyncMock( return_value=Event(EventType.ERROR, {"reason": "device_query_failed"}) ) command_handler.reset_path = AsyncMock() - setup_event_response( - mock_dispatcher, EventType.MSG_SENT, - {"expected_ack": b"\x01\x02\x03\x04", "suggested_timeout": 4000}, - ) result = await command_handler.send_anon_req(VALID_PUBKEY_HEX, MagicMock(value=1)) - assert not result.is_error() - sent = mock_connection.send.await_args.args[0] - assert sent == b"\x39" + bytes.fromhex(VALID_PUBKEY_HEX) + b"\x01" + b"\x00" - command_handler.reset_path.assert_awaited_once() + assert result.is_error() + assert result.payload["reason"] == "path_reset_failed" + mock_connection.send.assert_not_awaited() + # Nothing was changed on the device, so there is nothing to restore. + command_handler.reset_path.assert_not_awaited() # ── send_trace handles unknown path_hash_len without NameError ── @@ -427,7 +423,6 @@ async def test_ble_send_serialises_concurrent_writes(): conn = BLEConnection.__new__(BLEConnection) # bypass bleak availability check conn._disconnect_callback = None - conn._write_lock = None conn.rx_char = object() overlap = {"current": 0, "max": 0} @@ -451,7 +446,6 @@ async def test_ble_send_releases_lock_on_write_failure(): from meshcore.ble_cx import BLEConnection conn = BLEConnection.__new__(BLEConnection) - conn._write_lock = None conn.rx_char = object() reasons = [] @@ -595,3 +589,128 @@ async def test_reader_contact_out_path_keeps_zero_bytes(): assert contact["out_path_len"] == 1 assert contact["out_path_hash_mode"] == 2 assert contact["out_path"] == "aa00bb", "the 0x00 inside the hop hash was lost" + + +# ── BLE write bound (a stalled write must not wedge the client) ── + +async def test_ble_write_timeout_tears_down_the_link(): + """A stalled write must be bounded and must drop the link, not just release. + + CommandHandler's timeout starts only after the write returns, so nothing + else bounds this. Releasing the lock alone would be wrong too: the + underlying write may still be in flight, and a second write racing it + re-creates the overlap the lock exists to prevent. + """ + from meshcore.ble_cx import BLEConnection + + conn = BLEConnection.__new__(BLEConnection) + conn.rx_char = object() + reasons = [] + + async def capture(reason): + reasons.append(reason) + + conn._disconnect_callback = capture + conn.WRITE_TIMEOUT = 0.05 + + class Stalling: + async def write_gatt_char(self, char, data, response=True): + await asyncio.Event().wait() # never completes + + conn.client = Stalling() + + result = await asyncio.wait_for(conn.send(b"\x01"), timeout=2.0) + assert result is False + assert reasons == ["ble_write_timeout"] + + +async def test_ble_stalled_write_does_not_block_later_commands_forever(): + """Head-of-line: queued writes must not inherit an unbounded stall.""" + from meshcore.ble_cx import BLEConnection + + conn = BLEConnection.__new__(BLEConnection) + conn.rx_char = object() + conn._disconnect_callback = None + conn.WRITE_TIMEOUT = 0.05 + + class Stalling: + async def write_gatt_char(self, char, data, response=True): + await asyncio.Event().wait() + + conn.client = Stalling() + + # Five writers behind one stalled holder must all return, not hang. + done = await asyncio.wait_for( + asyncio.gather(*(conn.send(bytes([i])) for i in range(5))), timeout=3.0 + ) + assert done == [False] * 5 + + +async def test_ble_connection_init_provides_the_lock(): + """__init__ must set up the lock slot; the tests must not paper over it.""" + from meshcore.ble_cx import BLEConnection + + conn = BLEConnection.__new__(BLEConnection) + BLEConnection.__init__(conn, address="AA:BB:CC:DD:EE:FF") + assert conn._write_lock_obj is None # lazily created + assert isinstance(conn._write_lock, asyncio.Lock) + assert conn._write_lock is conn._write_lock # stable across reads + + +# ── reply-path bounds ──────────────────────────────────────── + +async def test_encode_reply_path_rejects_unsupported_hash_mode(): + # mode 3 -> 4-byte hops, which Packet::isValidPathLen refuses outright. + assert encode_reply_path(2, "aabbccddeeffaabb", 3) == b"\x00" + + +async def test_encode_reply_path_never_overflows_the_server_buffer(): + """The server memcpys into a fixed 64-byte reply_path with no length check.""" + for mode in (0, 1, 2): + hash_size = mode + 1 + out = encode_reply_path(63, "ab" * 200, mode) + body = out[1:] + assert len(body) <= 64, f"mode {mode} emitted {len(body)} bytes" + assert (out[0] & 63) * hash_size == len(body), "header disagrees with body" + + +async def test_send_anon_req_timeout_tracks_emitted_hops_not_claimed_length(): + """Scaling must follow the hops actually sent, not a length that got clamped.""" + command_handler = CommandHandler() + command_handler.dispatcher = MagicMock() + command_handler._reader = MagicMock() + conn = MagicMock() + conn.send = AsyncMock() + + async def sender(data): + await conn.send(data) + + command_handler._sender_func = sender + # Claims 8 hops but carries only 3 bytes -> 3 hops actually emitted. + command_handler._get_contact_by_prefix = MagicMock(return_value={ + "public_key": VALID_PUBKEY_HEX, + "out_path_len": 8, + "out_path": "aabbcc", + "out_path_hash_mode": 0, + }) + setup_event_response( + command_handler.dispatcher, EventType.MSG_SENT, + {"expected_ack": b"\x01\x02\x03\x04", "suggested_timeout": 1000}, + ) + + result = await command_handler.send_anon_req(VALID_PUBKEY_HEX, MagicMock(value=1)) + + sent = conn.send.await_args.args[0] + assert sent[-4] & 63 == 3 # 3 hops on the wire + assert result.payload["suggested_timeout"] == 4000 # (3 + 1) x 1000, not 9x + + +async def test_encode_reply_path_saturates_rather_than_wrapping(): + """hops is a 6-bit field; masking would wrap 64 to 0. + + A 64-hop path would then be described as zero hops -- a zero-hop reply + request for a distant node -- instead of being clamped and logged. + """ + out = encode_reply_path(64, "ab" * 64, 0) + assert out[0] & 63 == 63 + assert len(out[1:]) == 63