diff --git a/src/meshcore/ble_cx.py b/src/meshcore/ble_cx.py index 0a7380f..b0f7db2 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 @@ -27,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. @@ -52,6 +58,27 @@ class BLEConnection: self.rx_char = None self._disconnect_callback = None self._background_tasks: set[asyncio.Task] = set() + 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).""" @@ -190,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") @@ -199,8 +230,24 @@ class BLEConnection: if not self.rx_char: logger.error("RX characteristic not found") return False + # 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: - 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 543460a..5e4216a 100644 --- a/src/meshcore/commands/base.py +++ b/src/meshcore/commands/base.py @@ -57,6 +57,64 @@ 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. + + 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. (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 + # 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) + ) + return bytes([hops | (hash_mode << 6)]) + path + + class CommandHandlerBase: """Base class for command handlers. @@ -299,34 +357,73 @@ 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 + reply_path = encode_reply_path(0, "", 0) + else: + if contact["out_path_len"] == -1: + logger.info("No path set trying zero hop") + zero_hop = True + 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 + # 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) + reply_path = encode_reply_path( + out_path_len, + contact["out_path"], + contact.get("out_path_hash_mode", 0), + ) - data = contact["out_path_len"].to_bytes(1, "little") + bytes.fromhex(contact["out_path"])[::-1] - 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]) - + # 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 + 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/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 7f88e28..3153795 100644 --- a/tests/unit/test_error_handling.py +++ b/tests/unit/test_error_handling.py @@ -203,21 +203,195 @@ 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 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_aborts_when_change_contact_path_fails( + command_handler, mock_connection, mock_dispatcher +): + """A failed zero-hop switch must abort, not send an unanswerable request. + + 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) + command_handler.change_contact_path = AsyncMock( + return_value=Event(EventType.ERROR, {"reason": "device_query_failed"}) + ) + command_handler.reset_path = AsyncMock() + + 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 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 ── @@ -234,3 +408,309 @@ 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.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.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 + + +# ── 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" + + +# ── 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