From 77390860dae180340830006dccf8e440a28efdf2 Mon Sep 17 00:00:00 2001 From: Jean-Pierre Martineau Date: Thu, 9 Jul 2026 19:26:31 -0400 Subject: [PATCH] Handle PIN-protected Bluetooth companions gracefully MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A BLE companion that requires pairing accepts the link-layer connection but rejects the GATT subscribe with "Insufficient Authentication", which bleak raised as a bare BleakGATTProtocolError — escaping connect() as an unhandled traceback. Translate that (and related "Insufficient Encryption"/"not paired" rejections) into a clean DeviceCommandError that names the fix: pass --ble-pin, and pair once in the OS Bluetooth settings on Windows. It's classified as its own actionable case, never as a dropped link, so the session offers a PIN rather than a pointless reconnect. The startup probe now lets this actionable error through instead of collapsing every failure to a bare "didn't answer", so the picker shows the real remedy in the splash. Co-Authored-By: Claude Opus 4.8 --- meshterm/core/connection.py | 88 +++++++++++++++++++++++++++++++++++- meshterm/ui/device_picker.py | 42 ++++++++++++----- tests/test_disconnect.py | 45 ++++++++++++++++++ 3 files changed, 162 insertions(+), 13 deletions(-) diff --git a/meshterm/core/connection.py b/meshterm/core/connection.py index 2379e72..14ebf01 100644 --- a/meshterm/core/connection.py +++ b/meshterm/core/connection.py @@ -153,6 +153,42 @@ def is_connection_lost(exc: BaseException) -> bool: return False +#: Message fragments a BLE stack uses when a characteristic can't be accessed without a bond — +#: i.e. the companion is PIN-protected and we're unpaired (or gave the wrong PIN). The GATT +#: subscribe fails with one of these rather than a dropped link, so they're handled as a +#: distinct, actionable "needs a PIN" case (see :func:`_is_ble_auth_error`) and never as +#: connection loss. Matched by text so ``bleak`` need not be imported here. +_BLE_AUTH_HINTS = ( + "insufficient authentication", + "insufficient authorization", + "insufficient encryption", + "not paired", +) + + +def _is_ble_auth_error(exc: BaseException) -> bool: + """Return whether ``exc`` (or any it was raised from) is a BLE authentication rejection. + + Walks the whole ``__cause__``/``__context__`` chain matching :data:`_BLE_AUTH_HINTS`, so a + ``BleakGATTProtocolError`` wrapped by the meshcore transport is still recognized. + + Args: + exc: The exception raised while opening the Bluetooth connection. + + Returns: + ``True`` if the failure is a missing/rejected pairing rather than a dropped link. + """ + seen: set[int] = set() + current: Optional[BaseException] = exc + while current is not None and id(current) not in seen: + seen.add(id(current)) + text = str(current).lower() + if any(hint in text for hint in _BLE_AUTH_HINTS): + return True + current = current.__cause__ or current.__context__ + return False + + def serial_port_present(port: str) -> bool: """Return whether a serial port named ``port`` is currently enumerated by the OS. @@ -652,6 +688,37 @@ class MeshCoreDevice(Device): "Run `pip install -e .` (or `pip install bleak`), use a USB device, or " "run with --mock." ) from exc + except Exception as exc: # noqa: BLE001 - translate auth failures; re-raise the rest + # A PIN-protected companion accepts the link-layer connection but rejects the + # GATT subscribe with an authentication error ("Insufficient Authentication" / + # "Insufficient Encryption" / "not paired"). That's not a dropped link — it's a + # missing bond — so surface a clean, actionable message instead of a raw traceback + # (which is what a bare BleakGATTProtocolError would produce). Anything else + # propagates unchanged so genuine link-loss still flows to is_connection_lost. + if not _is_ble_auth_error(exc): + raise + raise DeviceCommandError(self._ble_auth_message()) from exc + + def _ble_auth_message(self) -> str: + """A clean, actionable error for a Bluetooth companion that requires a PIN/bond. + + Distinguishes "you gave the wrong PIN" from "you gave none at all", and points at + both fixes: MeshTerm's ``--ble-pin`` and the one-time OS pairing that Windows needs + before an authenticated characteristic can be subscribed. + """ + where = self._address or "the selected Bluetooth device" + if self._pin: + return ( + f"{where} rejected the Bluetooth PIN — it needs pairing and the PIN provided " + "wasn't accepted. Double-check the 6-digit code shown on the device (or in the " + "MeshCore app) and pass it with --ble-pin, then try again. On Windows you may " + "also need to remove and re-pair the device in Settings > Bluetooth." + ) + return ( + f"{where} requires a Bluetooth pairing PIN. Pass it with --ble-pin (the " + "6-digit code shown on the device or in the MeshCore app). On Windows you may also " + "need to pair the device once in Settings > Bluetooth before it will connect." + ) def _no_response_message(self) -> str: """A clean, recoverable error for an endpoint that didn't answer as a companion.""" @@ -1967,6 +2034,11 @@ async def probe_device( ``(device, self_info)`` with a connected :class:`MeshCoreDevice` on success (the caller owns and must eventually close it), or ``None`` if it is not a reachable MeshCore companion. + + Raises: + DeviceCommandError: On an actionable failure the user can fix — e.g. a Bluetooth + companion that requires a pairing PIN — so the caller can show the remedy rather + than an unhelpful "didn't answer". """ if device.is_ble: timeout = _PROBE_TIMEOUT_BLE_S @@ -2012,7 +2084,14 @@ async def _probe( timeout: Handshake window bounding both the connect and the identity read. Returns: - ``(device, self_info)`` with the connection left open, or ``None`` on any failure. + ``(device, self_info)`` with the connection left open, or ``None`` if the endpoint + simply isn't a reachable MeshCore companion. + + Raises: + DeviceCommandError: On an *actionable* failure the user can fix — e.g. a Bluetooth + companion that needs a pairing PIN. This is deliberately distinct from ``None`` + (an unremarkable "not a companion" miss) so the caller can show the real remedy + instead of a generic "didn't answer". """ # ``timeout`` is the handshake window handed to the client, so a non-MeshCore endpoint is # rejected in ~``timeout`` seconds and the client cleans up its own connection. The outer @@ -2022,7 +2101,12 @@ async def _probe( try: await asyncio.wait_for(device.connect(), timeout + 4.0) info = await asyncio.wait_for(device.get_self_info(), timeout) - except Exception: # noqa: BLE001 - any failure just means "not confirmed" + except DeviceCommandError: + # A clean, actionable failure (the device needs a PIN, say): close the probe and let it + # through so the picker surfaces the remedy rather than hiding it behind "didn't answer". + await _safe_disconnect(device) + raise + except Exception: # noqa: BLE001 - any other failure just means "not confirmed" await _safe_disconnect(device) return None if not info: diff --git a/meshterm/ui/device_picker.py b/meshterm/ui/device_picker.py index 574676d..c0b9002 100644 --- a/meshterm/ui/device_picker.py +++ b/meshterm/ui/device_picker.py @@ -3,9 +3,10 @@ Shown once at the start of the interactive menu when no port was given explicitly. Unlike the in-menu prompts, it is drawn as a chromeless splash — the MeshTerm wordmark centered above a content-sized box, with no header/footer status bars. It lists the discovered devices in -aligned columns — name, connection target, a TYPE glyph (wired serial vs Bluetooth), and the -hardware vendor — tags the ones already confirmed as MeshCore companions, marks the remembered -"last known good" one, and preselects it as the default. +aligned columns — name, connection target, a TYPE glyph (wired serial vs Bluetooth), and a +HARDWARE column (the confirmed device's firmware model, else the USB vendor) — tags the ones +already confirmed as MeshCore companions, marks the remembered "last known good" one, and +preselects it as the default. Confirmed companions are sorted to the top, most-recently-used first, and shown by the mesh node name we learned when we last talked to them (in white, so they stand out from ports we've @@ -24,6 +25,7 @@ from typing import TYPE_CHECKING, Awaitable, Callable, Optional from rich.cells import cell_len from rich.text import Text +from ..core.connection import DeviceCommandError from ..core.device_store import DeviceStore, RememberedDevice from ..core.discovery import DiscoveredDevice from .logo import load_logo @@ -34,7 +36,9 @@ if TYPE_CHECKING: #: A smoke test: probe a chosen device and return its self-info dict if it is a MeshCore #: companion, else ``None``. Supplied by the caller so this UI module stays free of the -#: connection layer. +#: connection *logic*. It may raise :class:`DeviceCommandError` for an actionable failure the +#: user can fix (e.g. a Bluetooth device that needs a PIN); the picker shows that message +#: verbatim instead of the generic "didn't answer". Verify = Callable[[DiscoveredDevice], Awaitable[Optional[dict]]] #: The Quit row's value. Selecting it — like pressing Esc — leaves the splash without a @@ -220,13 +224,29 @@ async def prompt_device( # Smoke-test the choice in place: the splash keeps its wordmark and box, only the box # contents swap for an animated spinner while we talk to the device. BLE connect and # service discovery take a few seconds, so the spinner matters most here. - info = await ui.busy_startup( - f"Talking to {name} {where}…", - verify(chosen), - title="Checking companion", - banner=load_logo(), - footnote=footnote, - ) + try: + info = await ui.busy_startup( + f"Talking to {name} {where}…", + verify(chosen), + title="Checking companion", + banner=load_logo(), + footnote=footnote, + ) + except DeviceCommandError as exc: + # The probe reached the device but it told us how it must be reached (e.g. it needs + # a BLE PIN). Show that remedy verbatim — not the generic "didn't answer" below, + # which would leave the user with no idea a PIN was the missing piece. Built as + # plain styled text so the message's own punctuation is never parsed as markup. + notice = Text() + notice.append(str(exc), style="warn") + notice.append("\nChoose another device.") + await ui.notify_startup( + notice, + title="Can't connect yet", + banner=load_logo(), + footnote=footnote, + ) + continue if info is None: reason = ( diff --git a/tests/test_disconnect.py b/tests/test_disconnect.py index ba5c71f..fe44180 100644 --- a/tests/test_disconnect.py +++ b/tests/test_disconnect.py @@ -93,6 +93,51 @@ def test_is_connection_lost_matches_ble_errors() -> None: assert is_connection_lost(RuntimeError("ble_transport_lost")) +class BleakGATTProtocolError(Exception): + """Stand-in for bleak's GATT auth rejection (classifier matches its message, not import).""" + + +def test_ble_auth_error_is_recognized_and_is_not_a_lost_link() -> None: + """A PIN/pairing rejection is its own actionable case — never mistaken for a dropped link.""" + exc = BleakGATTProtocolError("(5, 'GATT Protocol Error: Insufficient Authentication')") + assert connection._is_ble_auth_error(exc) + assert not is_connection_lost(exc) # so the session offers a PIN, not a reconnect + + +def test_ble_auth_error_walks_the_exception_chain() -> None: + """An auth rejection wrapped by the meshcore transport is still recognized via its cause.""" + try: + try: + raise BleakGATTProtocolError("Insufficient Encryption") + except Exception as cause: + raise RuntimeError("connect failed") from cause + except Exception as exc: + assert connection._is_ble_auth_error(exc) + + +async def test_create_ble_translates_auth_error_to_pin_guidance() -> None: + """A raw GATT auth rejection becomes a clean DeviceCommandError that names the PIN fix.""" + + class _FakeMeshCore: + @staticmethod + async def create_ble(**kwargs): + raise BleakGATTProtocolError("Insufficient Authentication") + + # No PIN supplied → tell the user to pass one. + dev = connection.MeshCoreDevice(transport="ble", address="00:11:22:33:44:55") + with pytest.raises(DeviceCommandError) as excinfo: + await dev._create_ble(_FakeMeshCore) + assert "--ble-pin" in str(excinfo.value) + + # PIN supplied but rejected → say it was wrong, not that none was given. + dev_pin = connection.MeshCoreDevice( + transport="ble", address="00:11:22:33:44:55", pin="123456" + ) + with pytest.raises(DeviceCommandError) as excinfo_pin: + await dev_pin._create_ble(_FakeMeshCore) + assert "rejected" in str(excinfo_pin.value).lower() + + def _make_ctx(tmp_path: Path) -> AppContext: """Build a real, mock-backed application context for reconnect tests.""" settings = Settings(config_dir=tmp_path, db_path=tmp_path / "disc.db")