ble://: use a paired link

cli/src/mpftp/ble.py

@@ -92,12 +92,22 @@ def parse_device(device: str) -> str:
     return target
 
 
-def check_password(password: Optional[str], device: str) -> str:
+_NO_PASSWORD = (
+    "no BLE REPL password. Set MPFTP_BLE_PASSWORD, or blePassword in "
+    "~/.mpftp/config.json, or save one with `mpftp wifi password ble://NAME`"
+)
+
+#: ATT errors that mean "pair first" (insufficient authentication, encryption).
+_NEEDS_PAIRING = (0x05, 0x0F)
+
+
+def check_password(password: Optional[str], device: str) -> Optional[str]:
+    """The password, checked; ``None`` when there's none, which is fine for a
+    board that pairing unlocks (``bledev.repl.start(pairing="passkey",
+    password=False)``). One that asks for a password without one is refused
+    at login."""
     if not password:
-        raise BleAuthError(
-            f"{device}: no BLE REPL password. Set MPFTP_BLE_PASSWORD, or blePassword in "
-            "~/.mpftp/config.json, or save one with `mpftp wifi password ble://NAME`"
-        )
+        return None
     if not MIN_PASSWORD_LEN <= len(password) <= MAX_PASSWORD_LEN or "\r" in password or "\n" in password:
         raise BleAuthError(
             f"{device}: a bledev.repl password is {MIN_PASSWORD_LEN} to {MAX_PASSWORD_LEN} "
@@ -106,6 +116,24 @@ def check_password(password: Optional[str], device: str) -> str:
     return password
 
 
+def _refused_for_pairing(e: BaseException) -> bool:
+    """Whether a failed write was the board asking for pairing (ATT 0x05/0x0F)."""
+    cause: Any = e
+    while cause is not None:
+        try:
+            from bleak.exc import BleakGATTProtocolError
+
+            if isinstance(cause, BleakGATTProtocolError) and int(cause.args[0]) in _NEEDS_PAIRING:
+                return True
+        except ImportError:
+            pass
+        text = str(cause).lower()
+        if "insufficient authentication" in text or "insufficient encryption" in text:
+            return True
+        cause = cause.__cause__
+    return False
+
+
 def _need_bleak() -> Any:
     try:
         import bleak
@@ -364,8 +392,17 @@ async def send() -> None:
                 raise BleError(f"{self.port}: BLE write failed: {e}") from e
 
     def _login(self) -> None:
-        self.write(b"\r")  # asks for the prompt again if we subscribed after it went out
-        self._read_until_any((_PASSWORD_PROMPT,), self.login_timeout, "the password prompt")
+        # Asks for the prompt again if we subscribed after it went out. A
+        # board started with pairing refuses it until this host has paired.
+        self._first_write()
+        seen = self._read_until_any((_PASSWORD_PROMPT, _CONNECTED), self.login_timeout, "the password prompt")
+        if _CONNECTED in seen:
+            # Pairing was the lock: no password asked.
+            self._read_until_any((b">>> ",), self.login_timeout, "the prompt after login")
+            return
+        if self._password is None:
+            self._closed_reason = "no password"
+            raise BleAuthError(f"{self.port}: the board asks for a password: {_NO_PASSWORD}")
         self.write(self._password.encode("utf-8") + b"\r")
         seen = self._read_until_any((_CONNECTED, _DENIED), self.login_timeout, "the login reply")
         if _DENIED in seen:
@@ -377,6 +414,68 @@ def _login(self) -> None:
         # The REPL login unlocks files on the same connection too.
         self._read_until_any((b">>> ",), self.login_timeout, "the prompt after login")
 
+    def _first_write(self) -> None:
+        # With a response: a write without one to a characteristic the board
+        # protects is dropped without a word, so an unpaired computer would
+        # just wait for a prompt that never comes.
+        try:
+            self._write_char(NUS_RX, b"\r", response=True)
+            return
+        except BleError as e:
+            if self._closed_reason or not _refused_for_pairing(e):
+                raise self._explain_drop(e) from e
+        # The board wants pairing. "Just works" needs nobody: a board without
+        # a display. One that shows a passkey needs a person, once.
+        try:
+            self._pair()
+        except Exception as e:
+            raise BleAuthError(
+                f"{self.port}: the board wants this computer paired and just works wasn't enough: "
+                "it shows a passkey. Pair once with `python -m bledev.bleak pair NAME` (or Windows "
+                f"Settings > Bluetooth > Add device), then try again. ({e})"
+            ) from e
+        try:
+            self._write_char(NUS_RX, b"\r", response=True)
+        except BleError as e:
+            if _refused_for_pairing(e):
+                # Just works was too weak for this board. Don't leave that
+                # pairing behind: it would only get in the way of the real one.
+                try:
+                    self._unpair()
+                except Exception:
+                    pass
+                raise BleAuthError(
+                    f"{self.port}: the board wants a passkey pairing, which needs a person once: "
+                    "pair with `python -m bledev.bleak pair NAME` (or Windows Settings > Bluetooth "
+                    "> Add device), typing in the passkey the board shows, then try again."
+                ) from e
+            raise self._explain_drop(e) from e
+
+    def _pair(self) -> None:
+        """Pair through the OS: bleak's, which on Windows answers just works."""
+        self._run(self._client.pair(), 40)
+
+    def _unpair(self) -> None:
+        self._run(self._client.unpair(), 20)
+
+    def _explain_drop(self, e: BleError) -> BleError:
+        """A link that falls at the first write, on a computer paired with the
+        board, is the board refusing the computer's keys: it lost its bond."""
+        if self._closed_reason and self._windows_paired():
+            return BleError(
+                f"{self.port}: the board hung up on this computer's stored keys; it has probably lost "
+                "its bond (a chip erase). Unpair it (`python -m bledev.bleak unpair NAME`, or Windows "
+                f"Settings > Bluetooth > Remove device) and pair again. Port is closed. ({e})"
+            )
+        return e
+
+    def _windows_paired(self) -> bool:
+        requester = getattr(getattr(self._client, "_backend", None), "_requester", None)
+        try:
+            return bool(requester.device_information.pairing.is_paired)
+        except Exception:
+            return False
+
     def _read_until_any(self, needles: tuple[bytes, ...], limit: float, what: str) -> bytes:
         deadline = time.monotonic() + limit
         with self._cond:
@@ -391,7 +490,7 @@ def _read_until_any(self, needles: tuple[bytes, ...], limit: float, what: str) -
                     raise BleError(f"{self.port}: the board closed the link before {what} (it said {data[-80:]!r})")
                 left = deadline - time.monotonic()
                 if left <= 0:
-                    raise BleError(f"{self.port}: no {what} within {limit:.0f} s (got {data[-80:]!r})")
+                    raise BleError(f"{self.port}: {what} didn't come within {limit:.0f} s (got {data[-80:]!r})")
                 self._cond.wait(min(left, 0.5))
 
     def _closed_message(self) -> str:

cli/tests/test_ble.py

@@ -92,7 +92,7 @@ def serve(self) -> None:
                 return
 
 
-def link(password: str = PASSWORD) -> ble.BleSerial:
+def link(password: str | None = PASSWORD) -> ble.BleSerial:
     serial = ble.BleSerial("ble://rack", password)
     serial.mtu = 247
     serial.has_files = True
@@ -115,8 +115,9 @@ def test_bad_addresses(self) -> None:
             ble.parse_device("ble://pw@rack")
 
     def test_password_rules(self) -> None:
-        with self.assertRaisesRegex(ble.BleAuthError, "MPFTP_BLE_PASSWORD"):
-            ble.check_password(None, "ble://rack")
+        # No password is fine: a board that pairing unlocks never asks.
+        self.assertIsNone(ble.check_password(None, "ble://rack"))
+        self.assertIsNone(ble.check_password("", "ble://rack"))
         with self.assertRaises(ble.BleAuthError):
             ble.check_password("abc", "ble://rack")
         with self.assertRaises(ble.BleAuthError):
@@ -175,6 +176,85 @@ def test_wrong_password(self) -> None:
             serial._login()
 
 
+class PairedBoard(FakeBoard):
+    """bledev.repl started with pairing: RX refuses an unpaired host (ATT
+    0x05); with ``password=None`` the first line opens the REPL."""
+
+    def __init__(self, link: ble.BleSerial, *, password=None, justworks_enough: bool = True) -> None:
+        FakeBoard.__init__(self, link, password=password or "unused")
+        self.board_password = password
+        self.paired = False
+        self.pairings = 0
+        self.justworks_enough = justworks_enough
+        link._pair = self.pair  # type: ignore[method-assign]
+        link._unpair = self.unpair  # type: ignore[method-assign]
+
+    def unpair(self) -> None:
+        self.paired = False
+
+    def pair(self) -> None:
+        self.pairings += 1
+        self.paired = True
+
+    def receive(self, uuid: str, data: bytes, response: bool = False) -> None:
+        if not (self.paired and self.justworks_enough):
+            try:
+                raise RuntimeError("Protocol Error 0x05: Insufficient Authentication")
+            except RuntimeError as e:
+                raise ble.BleError("ble://rack: BLE write failed") from e
+        if uuid == ble.NUS_RX and self.board_password is None:
+            self.link._on_repl(None, bytearray(b"\r\nbledev REPL connected\r\n>>> "))
+            return
+        FakeBoard.receive(self, uuid, data, response)
+
+
+class Pairing(unittest.TestCase):
+    def test_pairs_just_works_when_refused_then_logs_in_without_a_password(self) -> None:
+        serial = link(password=None)
+        board = PairedBoard(serial)
+        serial._login()
+        self.assertEqual(board.pairings, 1)
+        self.assertTrue(serial.is_open)
+
+    def test_already_paired_needs_no_pairing(self) -> None:
+        serial = link(password=None)
+        board = PairedBoard(serial)
+        board.paired = True
+        serial._login()
+        self.assertEqual(board.pairings, 0)
+
+    def test_pairing_plus_password(self) -> None:
+        serial = link()
+        board = PairedBoard(serial, password=PASSWORD)
+        serial._login()
+        self.assertEqual(board.pairings, 1)
+
+    def test_a_board_that_asks_for_a_password_needs_one(self) -> None:
+        serial = link(password=None)
+        PairedBoard(serial, password=PASSWORD)
+        with self.assertRaisesRegex(ble.BleAuthError, "asks for a password"):
+            serial._login()
+
+    def test_a_passkey_board_says_how_to_pair(self) -> None:
+        serial = link(password=None)
+        board = PairedBoard(serial, justworks_enough=False)
+        with self.assertRaisesRegex(ble.BleAuthError, "bledev.bleak pair"):
+            serial._login()
+        self.assertFalse(board.paired, "the useless just-works pairing was left behind")
+
+    def test_other_write_failures_are_not_pairing(self) -> None:
+        serial = link(password=None)
+        board = PairedBoard(serial)
+
+        def broken(uuid: str, data: bytes, response: bool = False) -> None:
+            raise ble.BleError("ble://rack: BLE write failed: the radio is off")
+
+        serial._write_char = broken  # type: ignore[method-assign]
+        with self.assertRaisesRegex(ble.BleError, "radio is off"):
+            serial._login()
+        self.assertEqual(board.pairings, 0)
+
+
 class Files(unittest.TestCase):
     def test_round_trip_in_windows(self) -> None:
         serial = link()

docs/agent-guide.md

@@ -157,6 +157,13 @@ export MPFTP_BLE_PASSWORD=...     # the board's bledev password; never print it
   serial-only.
 - The board takes one client at a time. The laptop's radio may be shared with
   other agents, so give the board a name nobody else advertises.
+- **Paired boards** (`bledev` started with `pairing=`): mpftp pairs "just
+  works" by itself when the board asks. A passkey board needs a person once
+  (`python -m bledev.bleak pair NAME`, typing the passkey from its screen);
+  mpftp then uses Windows' bond, and `MPFTP_BLE_PASSWORD` can be unset if the
+  board has no password. If the board lost its keys (a chip erase), the link
+  drops at the first write: unpair (`python -m bledev.bleak unpair NAME`) and
+  pair again. Unpair every board you paired before you finish.
 - The sidecar's Windows Python needs bleak (`python.exe -m pip install --user
   bleak`).
 

docs/user-guide.md

@@ -76,8 +76,12 @@ A board running pydevices' `bledev` can be reached over BLE, with no cable to
 this machine: any CLI board command takes `-d ble://NAME`, where NAME is the
 name the board advertises. The board serves it from `main.py`, e.g.
 `bledev.filetransfer.start(password="...", name="rack")`, and the password
-comes from `MPFTP_BLE_PASSWORD` (or `mpftp wifi password ble://NAME`). The
-extension's Connect list doesn't offer BLE boards yet. Details and speeds:
+comes from `MPFTP_BLE_PASSWORD` (or `mpftp wifi password ble://NAME`). A
+board started with `pairing="passkey"` needs this computer paired once: run
+`python -m bledev.bleak pair NAME` and type in the passkey the board shows
+(or use Windows Settings, Add device); after that mpftp uses the bond, with no
+password if the board has none. A board with `pairing="justworks"` is paired
+by mpftp itself. The extension's Connect list doesn't offer BLE boards yet. Details and speeds:
 [mpftp over Bluetooth](plans/ble.md).
 
 ### Soft reset and packages