From 4ebde385ddc3558fa761971c694aaf3d99877787 Mon Sep 17 00:00:00 2001 From: agessaman Date: Sun, 26 Jul 2026 19:23:32 -0700 Subject: [PATCH 1/4] fix(anon-req): allow requests to destinations that are not contacts send_anon_req() refused to send whenever the destination pubkey was absent from the client-side contact cache, returning ERROR contact_not_found. The contact is consulted for one thing only: building the reply-path bytes appended to the request. The companion firmware needs no contact of its own -- since FIRMWARE_VER_CODE 13 its CMD_SEND_ANON_REQ handler synthesises a transient anon contact for an unknown pubkey with out_path_len = 0 (zero-hop direct). Those entries live in a reserved slot ring, are hidden from CMD_GET_CONTACTS and are never persisted, so nothing is polluted by them. The client-side refusal therefore blocked a case the device supports, such as asking a freshly discovered neighbour for its regions before it has ever been added as a contact. Fall back to a zero-hop reply path instead. Two smaller fixes in the same function: - out_path_len is now read once into a local rather than re-read from the contact dict after the await. That dict is a live reference other commands mutate in place (send_msg_with_retry's flood fallback, reset_path); if it flipped to -1 mid-send the suggested_timeout multiplier became 4000 * 0 = 0, registering the binary request with a zero timeout so the response was dropped the moment it arrived. - The value is clamped at 0. update_contact() normally reflects the change back onto the dict, but if it fails the dict stays -1 and the unsigned to_bytes raises OverflowError -- which skipped the reset_path at the end of the method and left the contact pinned to zero-hop on the device. Verified against a companion radio on fw ver 13: without the change all five discovered repeaters were refused client-side in 0.0s with no RF sent; with it all three answered with their region scopes in ~1.1s. test_send_anon_req_contact_not_found is replaced -- it codified the removed limitation -- but its original regression (a TypeError on the NoneType subscript) stays covered. --- src/meshcore/commands/base.py | 55 ++++++--- tests/unit/test_error_handling.py | 192 ++++++++++++++++++++++++++++-- 2 files changed, 226 insertions(+), 21 deletions(-) diff --git a/src/meshcore/commands/base.py b/src/meshcore/commands/base.py index 543460a..1c315eb 100644 --- a/src/meshcore/commands/base.py +++ b/src/meshcore/commands/base.py @@ -299,34 +299,61 @@ class CommandHandlerBase: return result async def send_anon_req(self, dst: DestinationType, request_type: AnonReqType, data: Optional[bytes] = None, context={}, timeout=None, min_timeout=0) -> Event: + """Send an anonymous request to *dst*. + + *dst* need not be a known contact. When it is, that contact's out path is + used as the reply path; otherwise a zero-hop direct reply path is + requested (see the comment below). + + Note: *data* is currently ignored -- the request body is the reply path, + which is derived here rather than supplied by the caller. + """ dst_bytes = _validate_destination(dst, prefix_length=32) pubkey_prefix = _validate_destination(dst, prefix_length=6) logger.debug(f"Anon Binary request to {dst_bytes.hex()}") - contact = self._get_contact_by_prefix(dst_bytes.hex()) # need a contact for return path - if contact is None: - logger.error("No contact found") - return Event(EventType.ERROR, {"reason": "contact_not_found"}) + # The contact is consulted only to build the reply path appended to the + # request; it is not required to send one. Companion firmware from + # FIRMWARE_VER_CODE 13 synthesises a transient anon contact for an unknown + # pubkey (out_path_len = 0, zero-hop direct), so an unknown destination is + # reachable as long as we ask it to reply zero-hop. Refusing here would + # block probing any node the client has not already added - for instance + # asking a freshly discovered neighbour for its regions. + contact = self._get_contact_by_prefix(dst_bytes.hex()) zero_hop = False - if contact["out_path_len"] == -1: - logger.info("No path set trying zero hop") - zero_hop = True - await self.change_contact_path(contact, "") + if contact is None: + logger.debug("No contact found, requesting a zero-hop direct reply path") + out_path_len = 0 + out_path = b"" + else: + if contact["out_path_len"] == -1: + logger.info("No path set trying zero hop") + zero_hop = True + await self.change_contact_path(contact, "") + # 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 + # to_bytes below would raise OverflowError -- which would skip the + # reset_path at the end of this method and leave the contact pinned to + # zero-hop on the device. Zero is the right value to send regardless, + # since zero-hop is exactly what we just asked for. + out_path_len = max(contact["out_path_len"], 0) + out_path = bytes.fromhex(contact["out_path"])[::-1] - data = contact["out_path_len"].to_bytes(1, "little") + bytes.fromhex(contact["out_path"])[::-1] + data = out_path_len.to_bytes(1, "little") + out_path data = b"\x39" + dst_bytes + request_type.value.to_bytes(1, "little", signed=False) + (data if data else b"") result = await self.send(data, [EventType.MSG_SENT, EventType.ERROR]) - + # Register the request with the reader if we have both reader and request_type - if (result.type == EventType.MSG_SENT and - self._reader is not None and + if (result.type == EventType.MSG_SENT and + self._reader is not None and request_type is not None): - + 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) * (contact["out_path_len"] + 1) # update timeout from path_len + result.payload["suggested_timeout"] = result.payload.get("suggested_timeout", 4000) * (out_path_len + 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 7f88e28..8be06d1 100644 --- a/tests/unit/test_error_handling.py +++ b/tests/unit/test_error_handling.py @@ -203,21 +203,199 @@ async def test_set_multi_acks_error( assert result.is_error() -# ── send_anon_req returns ERROR on contact not found ───────── +# ── send_anon_req falls back to zero-hop when no contact exists ───────── -async def test_send_anon_req_contact_not_found( - command_handler, mock_dispatcher +async def test_send_anon_req_without_contact_sends_zero_hop( + command_handler, mock_connection, mock_dispatcher ): - """send_anon_req returns ERROR event when contact prefix not found, - instead of raising TypeError on NoneType subscript.""" + """An unknown destination is sent with a zero-hop reply path. + + The contact is only used to build the reply path. Companion firmware from + FIRMWARE_VER_CODE 13 synthesises a transient anon contact for an unknown + pubkey, so refusing to send here would needlessly block probing any node + the client has not already added. Must still not raise TypeError on the + NoneType subscript that this test originally guarded. + """ command_handler._get_contact_by_prefix = MagicMock(return_value=None) + command_handler.change_contact_path = AsyncMock() + 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 result.is_error() - assert result.payload["reason"] == "contact_not_found" + assert not result.is_error() + sent = mock_connection.send.await_args.args[0] + # \x39 | 32-byte pubkey | request type | reply-path-len 0 (no path bytes) + assert sent == b"\x39" + bytes.fromhex(VALID_PUBKEY_HEX) + b"\x01" + b"\x00" + + # No contact to mutate, so no device round-trips for path changes. + command_handler.change_contact_path.assert_not_awaited() + command_handler.reset_path.assert_not_awaited() + + +async def test_send_anon_req_without_contact_does_not_scale_timeout( + command_handler, mock_dispatcher +): + """suggested_timeout must not be multiplied by a path length we don't have.""" + command_handler._get_contact_by_prefix = MagicMock(return_value=None) + reader = MagicMock() + reader.register_binary_request = MagicMock() + command_handler._reader = reader + 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) + ) + + # out_path_len 0 -> (0 + 1) -> unchanged from the device's own estimate. + assert result.payload["suggested_timeout"] == 4000 + reader.register_binary_request.assert_called_once() + + +async def test_send_anon_req_with_contact_still_uses_its_path( + command_handler, mock_connection, mock_dispatcher +): + """A known contact's reply path and timeout scaling are both unchanged. + + _reader must be set: the out_path_len-based timeout scaling is gated behind + it, and that scaling is the only behaviour the no-contact refactor touches on + this path -- without a reader it is never executed. + """ + command_handler._get_contact_by_prefix = MagicMock(return_value={ + "public_key": VALID_PUBKEY_HEX, + "out_path_len": 3, + "out_path": "aabbcc", + }) + command_handler.change_contact_path = AsyncMock() + command_handler.reset_path = AsyncMock() + reader = MagicMock() + command_handler._reader = reader + 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)) + + sent = mock_connection.send.await_args.args[0] + # \x39 | pubkey | request type | reply-path-len 3 | path reversed + assert sent == ( + b"\x39" + bytes.fromhex(VALID_PUBKEY_HEX) + b"\x01" + + b"\x03" + bytes.fromhex("ccbbaa") + ) + # Scaled by out_path_len + 1 = 4. + assert result.payload["suggested_timeout"] == 16000 + reader.register_binary_request.assert_called_once() + assert reader.register_binary_request.call_args.args[3] == 16000 / 800.0 + command_handler.change_contact_path.assert_not_awaited() + command_handler.reset_path.assert_not_awaited() + + +async def test_send_anon_req_timeout_uses_path_len_as_sent( + command_handler, mock_connection, mock_dispatcher +): + """The timeout multiplier must match the path length actually transmitted. + + The contact dict is a live reference that other commands mutate in place, so + re-reading it after the await could scale the timeout by a path length that + was never sent. + """ + contact = {"public_key": VALID_PUBKEY_HEX, "out_path_len": 3, "out_path": "aabbcc"} + command_handler._get_contact_by_prefix = MagicMock(return_value=contact) + command_handler._reader = MagicMock() + + def fake_subscribe(evt_type, handler, attr_filters=None): + sub = MagicMock(spec=Subscription) + sub.unsubscribe = MagicMock() + if evt_type == EventType.MSG_SENT: + # Simulate another command flipping the contact to flood mid-send. + contact["out_path_len"] = -1 + contact["out_path"] = "" + asyncio.get_event_loop().call_soon( + handler, + Event(EventType.MSG_SENT, + {"expected_ack": b"\x01\x02\x03\x04", "suggested_timeout": 4000}), + ) + return sub + + mock_dispatcher.subscribe = MagicMock(side_effect=fake_subscribe) + + result = await command_handler.send_anon_req(VALID_PUBKEY_HEX, MagicMock(value=1)) + + sent = mock_connection.send.await_args.args[0] + assert sent.endswith(b"\x03" + bytes.fromhex("ccbbaa")) + # 4 x 4000, matching the 3-hop path in the frame -- not 0 from the mutated -1. + assert result.payload["suggested_timeout"] == 16000 + + +async def test_send_anon_req_flood_contact_still_forces_and_restores_zero_hop( + command_handler, mock_connection, mock_dispatcher +): + """out_path_len == -1 keeps its existing force-zero-hop-then-restore path.""" + contact = { + "public_key": VALID_PUBKEY_HEX, + "out_path_len": -1, + "out_path": "", + } + + async def fake_change_path(c, path): + # update_contact() reflects the change onto the dict; -1 would otherwise + # raise OverflowError on the unsigned to_bytes below. + c["out_path_len"] = 0 + c["out_path"] = "" + + command_handler._get_contact_by_prefix = MagicMock(return_value=contact) + command_handler.change_contact_path = AsyncMock(side_effect=fake_change_path) + command_handler.reset_path = AsyncMock() + setup_event_response( + mock_dispatcher, EventType.MSG_SENT, + {"expected_ack": b"\x01\x02\x03\x04", "suggested_timeout": 4000}, + ) + + await command_handler.send_anon_req(VALID_PUBKEY_HEX, MagicMock(value=1)) + + command_handler.change_contact_path.assert_awaited_once() + command_handler.reset_path.assert_awaited_once() + sent = mock_connection.send.await_args.args[0] + 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( + command_handler, mock_connection, mock_dispatcher +): + """A failed change_contact_path must not crash and must still restore the path. + + 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. + """ + 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() # ── send_trace handles unknown path_hash_len without NameError ── From 00135bbb9563353054fb946c276b67ff576a27d5 Mon Sep 17 00:00:00 2001 From: agessaman Date: Sun, 26 Jul 2026 19:24:03 -0700 Subject: [PATCH 2/4] fix(ble): serialise writes to the RX characteristic Two overlapping write_gatt_char() calls on the same characteristic drop the BLE link outright. Observed on macOS/CoreBluetooth as "BLE write failed: 19", after which the connection is gone and the pending command never completes. Nothing above the transport guaranteed callers were sequential: schedulers, health checks, periodic status queries and user commands all issue commands independently, so any unlucky overlap could take the radio down. The existing _mesh_request_lock only guards a few binary-request helpers, not the transport. Reproduced on a companion radio over BLE by issuing send_device_query() and send_node_discover_req() concurrently: before: BLE write failed: 19, connected=False, command hung >45s after: both complete in 0.12s, connected=True Issued sequentially the same two commands take 0.09s each and are fine, so it is specifically the overlap. Ruled out as causes beforehand: notification load (three discovers under a 91-packet firehose kept writes at 0.06-0.17s with the link stable) and the discover command itself. The lock is created lazily so it binds to the running loop, and is released on the failure path so one failed write cannot wedge every later command. --- src/meshcore/ble_cx.py | 13 ++++++- tests/unit/test_error_handling.py | 64 +++++++++++++++++++++++++++++++ 2 files changed, 76 insertions(+), 1 deletion(-) diff --git a/src/meshcore/ble_cx.py b/src/meshcore/ble_cx.py index 0a7380f..7cee63f 100644 --- a/src/meshcore/ble_cx.py +++ b/src/meshcore/ble_cx.py @@ -4,6 +4,7 @@ mccli.py : CLI interface to MeschCore BLE companion app import asyncio import logging +from typing import Optional # Make bleak optional - only fail if BLE operations are attempted @@ -52,6 +53,13 @@ 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 def _spawn_background(self, coro) -> asyncio.Task: """Create a tracked background task (prevents GC of fire-and-forget tasks).""" @@ -199,8 +207,11 @@ 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() try: - await self.client.write_gatt_char(self.rx_char, bytes(data), response=True) + async with self._write_lock: + await self.client.write_gatt_char(self.rx_char, bytes(data), response=True) except Exception as exc: logger.warning(f"BLE write failed: {exc}") if self._disconnect_callback: diff --git a/tests/unit/test_error_handling.py b/tests/unit/test_error_handling.py index 8be06d1..2470dfe 100644 --- a/tests/unit/test_error_handling.py +++ b/tests/unit/test_error_handling.py @@ -412,3 +412,67 @@ async def test_send_trace_unknown_path_hash_len( assert result.is_error() assert result.payload["reason"] == "invalid_path_format" + + +# ── BLE transport serialises writes ────────────────────────── + +async def test_ble_send_serialises_concurrent_writes(): + """Overlapping write_gatt_char() calls must not interleave. + + Two concurrent writes to the same characteristic drop the link outright + (observed on macOS/CoreBluetooth: "BLE write failed: 19"). Nothing above the + transport guarantees callers are sequential, so the transport must. + """ + from meshcore.ble_cx import BLEConnection + + 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} + + class FakeClient: + async def write_gatt_char(self, char, data, response=True): + overlap["current"] += 1 + overlap["max"] = max(overlap["max"], overlap["current"]) + await asyncio.sleep(0.01) # a real write is not instantaneous + overlap["current"] -= 1 + + conn.client = FakeClient() + + await asyncio.gather(*(conn.send(b"\x01\x02") for _ in range(8))) + + assert overlap["max"] == 1, f"{overlap['max']} writes overlapped" + + +async def test_ble_send_releases_lock_on_write_failure(): + """A failed write must not wedge the lock for every later command.""" + from meshcore.ble_cx import BLEConnection + + conn = BLEConnection.__new__(BLEConnection) + conn._write_lock = None + conn.rx_char = object() + reasons = [] + + async def capture(reason): + reasons.append(reason) + + conn._disconnect_callback = capture + + class BoomThenOK: + def __init__(self): + self.calls = 0 + + async def write_gatt_char(self, char, data, response=True): + self.calls += 1 + if self.calls == 1: + raise RuntimeError("19") + + conn.client = BoomThenOK() + + await conn.send(b"\x01") + assert reasons == ["ble_write_failed: 19"] + # Second write must still be able to acquire the lock. + await asyncio.wait_for(conn.send(b"\x02"), timeout=1.0) + assert conn.client.calls == 2 From 30446ed093118665e3bc8d06bb707057ee5059d7 Mon Sep 17 00:00:00 2001 From: agessaman Date: Sun, 26 Jul 2026 19:38:33 -0700 Subject: [PATCH 3/4] fix(anon-req): encode the reply path with its hash mode and hop order An anon request tells the server how to route its answer back. The leading byte of that reply path packs two fields, which the server unpacks as: reply_path_len = byte & 63 reply_path_hash_size = (byte >> 6) + 1 Three defects in producing it: 1. The hash mode was never written into the top two bits, so the server always read a hash size of 1 whatever the contact's real mode was. 2. The path was reversed byte-wise (out_path[::-1]) rather than hop-wise. A return path visits the same hops in reverse order with each hop's multi-byte hash intact. 3. reader.py built out_path by stripping every NUL from the fixed 64-byte field. That trims the padding but also eats a legitimate 0x00 inside a hop hash, shortening the path and shifting every hop after it. It now takes out_path_len * hash_size bytes, as PATH_DISCOVERY_RESPONSE already did. Worked example at hash mode 2 (3 bytes per hop), for a contact two hops away via aabbcc then ddeeff: before: lenbyte 0x02, path ffeeddccbbaa -> server reads 2 hops of 1 byte, replies via ['ff', 'ee'] after: lenbyte 0x82, path ddeeffaabbcc -> server reads 2 hops of 3 bytes, replies via ['ddeeff', 'aabbcc'] The old form routes the response to hops that do not exist, so it is dropped and the request times out. At hash mode 0 both encodings are byte-identical -- the mode contributes nothing to the high bits and byte-wise reversal equals hop-wise reversal for single-byte hops -- which is why this stayed latent: mode 0 is the default. Confirmed by the mode-0 and zero-hop tests passing unchanged against the old code while the mode-1/mode-2 tests fail. Scope: only anon requests routed direct to a contact with a known multi-hop path. Flood requests are unaffected (the server answers via createPathReturn and ignores the supplied reply path), as is login (handleLoginReq never sets reply_path_len, so its reply always goes out flood). The neighbors zero-hop probe is unaffected: length 0 makes hash size irrelevant. Encoding is extracted into encode_reply_path() so it can be tested directly. Verified on hardware only for the zero-hop case, which still works; the multi-hop paths are covered by unit tests, as the test radio has no multi-hop contacts to exercise on air. --- src/meshcore/commands/base.py | 43 ++++++++++- src/meshcore/reader.py | 12 ++- tests/unit/test_error_handling.py | 119 ++++++++++++++++++++++++++++++ 3 files changed, 169 insertions(+), 5 deletions(-) diff --git a/src/meshcore/commands/base.py b/src/meshcore/commands/base.py index 1c315eb..2ed4a59 100644 --- a/src/meshcore/commands/base.py +++ b/src/meshcore/commands/base.py @@ -57,6 +57,38 @@ def _validate_destination(dst: DestinationType, prefix_length: int = 6) -> bytes ) +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. + + The leading byte packs two fields, which the server unpacks as: + + reply_path_len = byte & 63 + reply_path_hash_size = (byte >> 6) + 1 + + so the hash mode has to travel in the top two bits. Omitting it makes the + server read a hash size of 1 regardless of the real mode, take the wrong + number of bytes per hop, and route its reply to hops that do not exist. + + 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.) + """ + hash_mode = max(out_path_hash_mode, 0) # -1 means "flood", i.e. no path + hash_size = hash_mode + 1 + hops = max(out_path_len, 0) & 63 + + 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) + + path = b"".join( + raw[i * hash_size:(i + 1) * hash_size] for i in range(hops - 1, -1, -1) + ) + return bytes([hops | (hash_mode << 6)]) + path + + class CommandHandlerBase: """Base class for command handlers. @@ -325,7 +357,7 @@ class CommandHandlerBase: if contact is None: logger.debug("No contact found, requesting a zero-hop direct reply path") out_path_len = 0 - out_path = b"" + reply_path = encode_reply_path(0, "", 0) else: if contact["out_path_len"] == -1: logger.info("No path set trying zero hop") @@ -339,10 +371,13 @@ class CommandHandlerBase: # zero-hop on the device. Zero is the right value to send regardless, # since zero-hop is exactly what we just asked for. out_path_len = max(contact["out_path_len"], 0) - out_path = bytes.fromhex(contact["out_path"])[::-1] + reply_path = encode_reply_path( + out_path_len, + contact["out_path"], + contact.get("out_path_hash_mode", 0), + ) - data = out_path_len.to_bytes(1, "little") + out_path - data = b"\x39" + dst_bytes + request_type.value.to_bytes(1, "little", signed=False) + (data if data else b"") + data = b"\x39" + dst_bytes + request_type.value.to_bytes(1, "little", signed=False) + reply_path result = await self.send(data, [EventType.MSG_SENT, EventType.ERROR]) diff --git a/src/meshcore/reader.py b/src/meshcore/reader.py index ae8067f..d1de1d9 100644 --- a/src/meshcore/reader.py +++ b/src/meshcore/reader.py @@ -112,7 +112,17 @@ class MessageReader: else: c["out_path_hash_mode"] = plen >> 6 c["out_path_len"] = plen & 0x3F # 6 LSB - c["out_path"] = dbuf.read(64).replace(b"\0", b"").hex() + # The field is a fixed 64 bytes, NUL-padded past the real path. + # Take exactly the bytes the path occupies rather than stripping + # NULs: a hop hash may legitimately contain 0x00, and dropping + # those shortens the path and shifts every hop after it. + # (PATH_DISCOVERY_RESPONSE below already reads opl*opl_hlen.) + path_bytes = dbuf.read(64) + if c["out_path_len"] > 0: + used = c["out_path_len"] * (c["out_path_hash_mode"] + 1) + c["out_path"] = path_bytes[:used].hex() + else: + c["out_path"] = "" c["adv_name"] = dbuf.read(32).decode("utf-8", "ignore").replace("\0", "") c["last_advert"] = int.from_bytes(dbuf.read(4), byteorder="little") c["adv_lat"] = ( diff --git a/tests/unit/test_error_handling.py b/tests/unit/test_error_handling.py index 2470dfe..2206895 100644 --- a/tests/unit/test_error_handling.py +++ b/tests/unit/test_error_handling.py @@ -476,3 +476,122 @@ async def test_ble_send_releases_lock_on_write_failure(): # Second write must still be able to acquire the lock. await asyncio.wait_for(conn.send(b"\x02"), timeout=1.0) assert conn.client.calls == 2 + + +# ── reply-path encoding (hash mode + hop-wise reversal) ────── + +from meshcore.commands.base import encode_reply_path # noqa: E402 + + +async def test_encode_reply_path_zero_hop(): + # len 0, mode 0 -> a single 0x00 byte, the zero-hop direct request. + assert encode_reply_path(0, "", 0) == b"\x00" + + +async def test_encode_reply_path_mode0_reverses_hops(): + # hops aa, bb -> reply visits bb then aa. Mode 0 leaves the top bits clear. + assert encode_reply_path(2, "aabb", 0) == b"\x02" + bytes.fromhex("bbaa") + + +async def test_encode_reply_path_carries_hash_mode_in_top_bits(): + """The server reads hash size from bits 6-7; omitting it truncates each hop. + + Mode 2 = 3 bytes per hop. Without the mode the server would read hash_size 1 + and reply to two 1-byte hops that do not exist. + """ + out = encode_reply_path(2, "aabbccddeeff", 2) + assert out[0] == 0x82 # 2 hops | (mode 2 << 6) + assert out[0] & 63 == 2 # server: reply_path_len + assert (out[0] >> 6) + 1 == 3 # server: reply_path_hash_size + # Hop order reversed, each 3-byte hash intact. + assert out[1:] == bytes.fromhex("ddeeff") + bytes.fromhex("aabbcc") + + +async def test_encode_reply_path_mode1_two_byte_hops(): + out = encode_reply_path(3, "aabbccddeeff", 1) + assert out[0] == (3 | (1 << 6)) + assert out[1:] == bytes.fromhex("eeff") + bytes.fromhex("ccdd") + bytes.fromhex("aabb") + + +async def test_encode_reply_path_flood_mode_is_clamped(): + # out_path_hash_mode is -1 for a flood contact; must not produce a negative shift. + assert encode_reply_path(0, "", -1) == b"\x00" + + +async def test_encode_reply_path_ignores_padding_beyond_the_path(): + # Real device fields are NUL-padded to 64 bytes; only the used hops count. + padded = "aabb" + "00" * 60 + assert encode_reply_path(2, padded, 0) == b"\x02" + bytes.fromhex("bbaa") + + +async def test_encode_reply_path_truncated_field_does_not_emit_short_hops(): + # Claims 4 hops of 3 bytes but only 6 bytes present -> emit the 2 it has. + out = encode_reply_path(4, "aabbccddeeff", 2) + assert out[0] & 63 == 2 + assert out[1:] == bytes.fromhex("ddeeff") + bytes.fromhex("aabbcc") + + +async def test_send_anon_req_reply_path_uses_hash_mode( + command_handler, mock_connection, mock_dispatcher +): + """End-to-end: a mode-2 multi-hop contact must get a correct reply path.""" + command_handler._get_contact_by_prefix = MagicMock(return_value={ + "public_key": VALID_PUBKEY_HEX, + "out_path_len": 2, + "out_path": "aabbccddeeff", + "out_path_hash_mode": 2, + }) + command_handler.change_contact_path = AsyncMock() + command_handler.reset_path = AsyncMock() + setup_event_response( + mock_dispatcher, EventType.MSG_SENT, + {"expected_ack": b"\x01\x02\x03\x04", "suggested_timeout": 4000}, + ) + + await command_handler.send_anon_req(VALID_PUBKEY_HEX, MagicMock(value=1)) + + sent = mock_connection.send.await_args.args[0] + assert sent == ( + b"\x39" + bytes.fromhex(VALID_PUBKEY_HEX) + b"\x01" + + b"\x82" + bytes.fromhex("ddeeff") + bytes.fromhex("aabbcc") + ) + + +async def test_reader_contact_out_path_keeps_zero_bytes(): + """A hop hash containing 0x00 must survive parsing by the real reader. + + The 64-byte field is NUL-padded, but stripping every NUL also eats + legitimate hash bytes, shortening the path and shifting every hop after it. + """ + from meshcore.reader import MessageReader + from meshcore.packets import PacketType + + hops = bytes.fromhex("aa00bb") # middle byte is a legitimate 0x00 + frame = ( + bytes([PacketType.CONTACT.value]) + + bytes(32) # public_key + + b"\x02" # type + + b"\x00" # flags + + bytes([1 | (2 << 6)]) # 1 hop, hash mode 2 -> 3 bytes/hop + + hops + bytes(64 - len(hops)) # out_path, NUL-padded to 64 + + b"name".ljust(32, b"\0") # adv_name + + (0).to_bytes(4, "little") # last_advert + + (0).to_bytes(4, "little", signed=True) # adv_lat + + (0).to_bytes(4, "little", signed=True) # adv_lon + + (0).to_bytes(4, "little") # lastmod + ) + + seen = [] + + class Dispatcher: + async def dispatch(self, event): + seen.append(event) + + reader = MessageReader(Dispatcher()) + reader.contacts = {} + await reader.handle_rx(frame) + + contact = next(e.payload for e in seen if e.type == EventType.NEXT_CONTACT) + 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" From 6eeab56779bc2294e4fb2dd426876cbcf009c237 Mon Sep 17 00:00:00 2001 From: agessaman Date: Sun, 26 Jul 2026 19:59:15 -0700 Subject: [PATCH 4/4] 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