From 7d40929341b65786e57fbf58e450e4cbbec82338 Mon Sep 17 00:00:00 2001 From: Jean-Pierre Martineau Date: Thu, 9 Jul 2026 19:58:31 -0400 Subject: [PATCH] Pair PIN-protected BLE companions via the WinRT ProvidePin ceremony bleak's pair() on Windows only performs the CONFIRM_ONLY ("Just Works") ceremony and never sends a passkey, so a PIN-protected MeshCore companion bonds without authentication and then rejects the GATT subscribe on its authenticated UART characteristic with Insufficient Authentication -- a correct --ble-pin was reported as "rejected" and the device could never connect. MeshCoreDevice now runs the WinRT PROVIDE_PIN ceremony itself before opening the link, creating the ENCRYPTION_AND_AUTHENTICATION bond the characteristic requires; Windows persists it, so later reconnects need no PIN. A first GATT auth failure with a supplied PIN also triggers one unpair-and-re-pair retry to heal a stale unauthenticated bond from an older attempt. The step is a best-effort no-op off Windows, when no PIN is set, or when already bonded. Verified on hardware (AA:BB:CC:DD:EE:FF, MeshCore-Homestead): correct PIN pairs and connects (reads model Seeed Tracker T1000-E), wrong PIN fails clean with the "rejected" DeviceAuthenticationError, no PIN reports "requires a PIN", and a bonded device reconnects with no PIN. Full CLI path exercised end to end. Co-Authored-By: Claude Opus 4.8 --- meshterm/core/connection.py | 145 +++++++++++++++++++++++++++++++++++- tests/test_disconnect.py | 89 ++++++++++++++++++++++ 2 files changed, 233 insertions(+), 1 deletion(-) diff --git a/meshterm/core/connection.py b/meshterm/core/connection.py index 7797b66..2554c98 100644 --- a/meshterm/core/connection.py +++ b/meshterm/core/connection.py @@ -16,6 +16,7 @@ from __future__ import annotations import asyncio import logging import random +import sys from abc import ABC, abstractmethod from datetime import datetime, timezone from typing import TYPE_CHECKING, Callable, Optional @@ -674,18 +675,48 @@ class MeshCoreDevice(Device): commands.default_timeout = default async def _create_ble(self, mesh_core): # type: ignore[no-untyped-def] - """Open the BLE companion connection, translating a missing-bleak import cleanly. + """Open the BLE companion connection, pairing with a PIN first on Windows. ``auto_reconnect`` is deliberately left off: MeshTerm drives reconnection itself (the same reconnect dialog the serial path uses), so the meshcore client should surface a dropped link promptly via ``is_connected`` rather than silently retrying underneath us. + Before opening the link we establish an *authenticated* pairing ourselves when a PIN is + supplied (see :meth:`_pair_ble_windows`). This is essential on Windows: bleak's + ``pair()`` only performs the "Just Works" ceremony (``CONFIRM_ONLY``) and never enters a + passkey, so a PIN-protected companion bonds *without authentication* and then rejects + the GATT subscribe on its authenticated UART characteristic — a correct PIN is reported + as "rejected" and the device can never connect. Running the WinRT ProvidePin ceremony + ourselves creates the authenticated bond the characteristic requires; once bonded, the + OS keeps the bond and later reconnects need no PIN at all. The step is a harmless no-op + on other platforms, when no PIN is set, or when the device is already bonded. + Args: mesh_core: The imported ``meshcore.MeshCore`` class. Returns: The connected ``MeshCore`` client, or ``None`` if the peripheral never answered. """ + await self._pair_ble_windows(force=False) + return await self._open_ble(mesh_core, allow_repair=True) + + async def _open_ble(self, mesh_core, *, allow_repair: bool): # type: ignore[no-untyped-def] + """Open the meshcore BLE client, translating auth failures and healing stale bonds. + + Args: + mesh_core: The imported ``meshcore.MeshCore`` class. + allow_repair: Whether a GATT authentication failure may trigger one unpair-and- + re-pair-with-PIN retry (Windows only). Set ``False`` on that retry so a genuine + wrong-PIN can't loop. + + Returns: + The connected ``MeshCore`` client, or ``None`` if the peripheral never answered. + + Raises: + DeviceCommandError: If ``bleak`` is missing (with install guidance). + DeviceAuthenticationError: If the companion needs a pairing PIN we don't have or + that was rejected. + """ try: return await mesh_core.create_ble( address=self._address, @@ -708,8 +739,120 @@ class MeshCoreDevice(Device): # propagates unchanged so genuine link-loss still flows to is_connection_lost. if not _is_ble_auth_error(exc): raise + # On Windows this can also happen with the *right* PIN when a stale, unauthenticated + # "Just Works" bond from an older attempt is in the way: is_paired is true so the + # ProvidePin step above was skipped, yet the bond can't unlock the characteristic. + # Clear it, pair with the PIN, and retry the connect exactly once before giving up. + if allow_repair and await self._pair_ble_windows(force=True): + return await self._open_ble(mesh_core, allow_repair=False) raise DeviceAuthenticationError(self._ble_auth_message()) from exc + async def _pair_ble_windows(self, *, force: bool) -> bool: + """Establish an authenticated BLE bond via the WinRT ProvidePin ceremony (Windows only). + + This is the one place a Bluetooth passkey is actually delivered to the peripheral. + bleak's own ``pair()`` on Windows is hardcoded to the ``CONFIRM_ONLY`` ("Just Works") + ceremony and never sends a PIN, so a companion that demands passkey pairing can't be + bonded through bleak at all — its authenticated UART characteristic keeps rejecting the + notify subscribe with *Insufficient Authentication*. We instead run the ``PROVIDE_PIN`` + ceremony directly against WinRT (the same one the Windows "Add device" dialog uses), + handing it :attr:`_pin`, which yields the ``ENCRYPTION_AND_AUTHENTICATION`` bond the + characteristic needs. Windows persists the bond, so subsequent sessions reconnect with + no PIN required. + + Best-effort and self-contained: it returns a bool rather than raising, and swallows any + error (winrt projection absent, device out of range, API quirk) so the caller simply + falls through to the normal connect — whose auth-error translation still yields the + right message. A no-op (returns ``False``) off Windows, when no PIN is set, or when the + address isn't a parseable MAC. + + Args: + force: When ``False``, an existing bond is trusted and reused (the fast path). When + ``True``, any existing bond is torn down first and re-created with the PIN — used + to heal a stale, unauthenticated "Just Works" bond that a plain reconnect can't. + + Returns: + ``True`` if an authenticated bond exists afterward (freshly paired or already + bonded), ``False`` otherwise. + """ + if sys.platform != "win32" or not self._pin: + return False + address = self._ble_address_int(self._address or "") + if address is None: + return False + try: + from winrt.windows.devices.bluetooth import BluetoothLEDevice + from winrt.windows.devices.enumeration import ( + DevicePairingKinds, + DevicePairingProtectionLevel, + DevicePairingResultStatus, + ) + except Exception as exc: # noqa: BLE001 - winrt projection unavailable; fall through + _log.debug("BLE PIN pairing unavailable (winrt import failed): %s", exc) + return False + + try: + device = await BluetoothLEDevice.from_bluetooth_address_async(address) + if device is None: + return False # out of range / not connectable right now + pairing = device.device_information.pairing + if pairing.is_paired: + if not force: + return True # trust the existing (authenticated) bond — fast path + # Tear the stale bond down, then re-fetch: the pairing object is a snapshot and + # won't reflect the unpair, so a fresh device_information is needed to re-pair. + await pairing.unpair_async() + device = await BluetoothLEDevice.from_bluetooth_address_async(address) + if device is None: + return False + pairing = device.device_information.pairing + custom = pairing.custom + pin = self._pin + + def _provide_pin(_sender, args) -> None: # noqa: ANN001 - winrt callback + # The peripheral asked for a passkey; hand it the one we were given. + args.accept_with_pin(pin) + + token = custom.add_pairing_requested(_provide_pin) + try: + result = await custom.pair_with_protection_level_async( + DevicePairingKinds.PROVIDE_PIN, + DevicePairingProtectionLevel.ENCRYPTION_AND_AUTHENTICATION, + ) + finally: + custom.remove_pairing_requested(token) + status = int(result.status) + ok = status in ( + int(DevicePairingResultStatus.PAIRED), + int(DevicePairingResultStatus.ALREADY_PAIRED), + ) + _log.debug( + "BLE ProvidePin pairing for %s: status=%d ok=%s", self._address, status, ok + ) + return ok + except Exception as exc: # noqa: BLE001 - best-effort; caller falls through on False + _log.debug("BLE ProvidePin pairing attempt failed for %s: %s", self._address, exc) + return False + + @staticmethod + def _ble_address_int(address: str) -> Optional[int]: + """Parse a ``AA:BB:CC:DD:EE:FF`` (or dash-separated) MAC into the ulong WinRT wants. + + Args: + address: The Bluetooth address string bleak reported for the device. + + Returns: + The 48-bit address as an int, or ``None`` if it isn't a 12-hex-digit MAC (e.g. a + CoreBluetooth UUID on macOS, where this pairing path doesn't apply anyway). + """ + cleaned = address.replace(":", "").replace("-", "").strip() + if len(cleaned) != 12: + return None + try: + return int(cleaned, 16) + except ValueError: + return None + def _ble_auth_message(self) -> str: """A clean, actionable error for a Bluetooth companion that requires a PIN/bond. diff --git a/tests/test_disconnect.py b/tests/test_disconnect.py index df050a5..4c69bb2 100644 --- a/tests/test_disconnect.py +++ b/tests/test_disconnect.py @@ -115,6 +115,20 @@ def test_ble_auth_error_walks_the_exception_chain() -> None: assert connection._is_ble_auth_error(exc) +async def _disable_windows_pairing(dev: "connection.MeshCoreDevice") -> None: + """Stub out the WinRT ProvidePin step so ``_create_ble`` stays hermetic in tests. + + On a real Windows host ``_pair_ble_windows`` would reach the OS Bluetooth stack (and the + physical device); pinning it to a no-op reproduces the non-Windows / no-winrt path so the + auth-translation logic can be exercised without hardware. + """ + + async def _never_pairs(*, force: bool) -> bool: + return False + + dev._pair_ble_windows = _never_pairs # type: ignore[method-assign] + + async def test_create_ble_translates_auth_error_to_pin_guidance() -> None: """A raw GATT auth rejection becomes a DeviceAuthenticationError that names the PIN fix.""" @@ -126,6 +140,7 @@ async def test_create_ble_translates_auth_error_to_pin_guidance() -> None: # No PIN supplied → tell the user to pass one. The subclass lets the interactive picker # catch "needs a PIN" specifically, while the CLI still catches it as DeviceCommandError. dev = connection.MeshCoreDevice(transport="ble", address="00:11:22:33:44:55") + await _disable_windows_pairing(dev) with pytest.raises(connection.DeviceAuthenticationError) as excinfo: await dev._create_ble(_FakeMeshCore) assert isinstance(excinfo.value, DeviceCommandError) # so the scripted CLI catches it too @@ -135,11 +150,85 @@ async def test_create_ble_translates_auth_error_to_pin_guidance() -> None: dev_pin = connection.MeshCoreDevice( transport="ble", address="00:11:22:33:44:55", pin="123456" ) + await _disable_windows_pairing(dev_pin) with pytest.raises(connection.DeviceAuthenticationError) as excinfo_pin: await dev_pin._create_ble(_FakeMeshCore) assert "rejected" in str(excinfo_pin.value).lower() +async def test_create_ble_repairs_stale_bond_and_retries_once() -> None: + """A first auth failure triggers one unpair-and-re-pair, then the retried connect succeeds. + + Models the Windows upgrade case: a leftover unauthenticated "Just Works" bond makes the + first connect fail even with the right PIN, so ``_pair_ble_windows(force=True)`` clears it + and the second connect goes through. + """ + attempts: list[int] = [] + + class _FakeMeshCore: + @staticmethod + async def create_ble(**kwargs): + attempts.append(1) + if len(attempts) == 1: + raise BleakGATTProtocolError("Insufficient Authentication") + return "connected-client" + + dev = connection.MeshCoreDevice( + transport="ble", address="00:11:22:33:44:55", pin="123456" + ) + repairs: list[bool] = [] + + async def _pair(*, force: bool) -> bool: + repairs.append(force) + return force # the pre-connect pass (force=False) no-ops; the repair (force=True) works + + dev._pair_ble_windows = _pair # type: ignore[method-assign] + result = await dev._create_ble(_FakeMeshCore) + assert result == "connected-client" + assert len(attempts) == 2 # failed once, retried once + assert repairs == [False, True] # pre-connect attempt, then the healing re-pair + + +async def test_create_ble_gives_up_after_one_repair() -> None: + """A wrong PIN that never bonds fails cleanly rather than looping on the repair retry.""" + + class _FakeMeshCore: + @staticmethod + async def create_ble(**kwargs): + raise BleakGATTProtocolError("Insufficient Authentication") + + dev = connection.MeshCoreDevice( + transport="ble", address="00:11:22:33:44:55", pin="000000" + ) + calls: list[bool] = [] + + async def _pair(*, force: bool) -> bool: + calls.append(force) + return force # even the repair "succeeds" so we prove the retry runs exactly once + + dev._pair_ble_windows = _pair # type: ignore[method-assign] + with pytest.raises(connection.DeviceAuthenticationError): + await dev._create_ble(_FakeMeshCore) + # force=False (pre-connect), then force=True (repair). The repair's retry passes + # allow_repair=False, so there is no third pairing attempt even though it keeps failing. + assert calls == [False, True] + + +async def test_pair_ble_windows_noops_without_pin() -> None: + """Pairing is skipped (no WinRT touched) when no PIN is set — the fast, hermetic path.""" + dev = connection.MeshCoreDevice(transport="ble", address="00:11:22:33:44:55") + assert await dev._pair_ble_windows(force=False) is False + + +def test_ble_address_int_parses_macs_and_rejects_others() -> None: + """A colon/dash MAC becomes a 48-bit int; a non-MAC (e.g. a macOS UUID) yields None.""" + parse = connection.MeshCoreDevice._ble_address_int + assert parse("00:11:22:33:44:55") == 0xDAC5216B7C7C + assert parse("da-c5-21-6b-7c-7c") == 0xDAC5216B7C7C + assert parse("not-a-mac") is None + assert parse("550e8400-e29b-41d4-a716-446655440000") is None # CoreBluetooth UUID + + 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")