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 <noreply@anthropic.com>

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.
 

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")