From f872faa68666cc6c2f2a2a917ba12464d0201861 Mon Sep 17 00:00:00 2001 From: Brad Barnett <127794626+bdbarnett@users.noreply.github.com> Date: Thu, 24 Sep 2026 20:21:46 -0500 Subject: [PATCH 1/2] bledev.mpble: a board as central types in the passkey even when the stack never asks MicroPython's NimBLE bindings raise _IRQ_PASSKEY_ACTION only on connections the board accepted. peripheral_gap_event_cb, which handles the ones it made, drops BLE_GAP_EVENT_PASSKEY_ACTION, so a board as central is never asked for the passkey its peer shows and pairing stalls until the stack's 30 s timeout (micropython/micropython#19470 moves the event). Seen on the P4 against the LCD-7: the LCD-7 showed its passkey, the P4 got no IRQ. pair(passkey=...) now waits 1.5 s for the stack to ask; if it hasn't, it asks the provider itself and offers the passkey with gap_passkey() until NimBLE takes it (EINVAL until the pairing reaches its confirm step). A stack that does ask answers from the same value instead of asking twice. --- lib/bledev/mpble.py | 40 +++++++++++++++++++++++++++++++++++++++- lib/bledev/security.py | 8 +++++++- 2 files changed, 46 insertions(+), 2 deletions(-) diff --git a/lib/bledev/mpble.py b/lib/bledev/mpble.py index dfd3b9ba..6e701ad0 100644 --- a/lib/bledev/mpble.py +++ b/lib/bledev/mpble.py @@ -69,6 +69,11 @@ _IRQ_GATTS_WRITE = 3 _IRQ_PERIPHERAL_DISCONNECT = 8 _IRQ_MTU_EXCHANGED = 21 +_PASSKEY_INPUT = 2 # gap_passkey's action for typing in the peer's passkey +_EINVAL = 22 +# As central with a passkey to type: how long to wait for the stack to ask +# before asking anyway (see _MPConnection._type_passkey). +_ASK_WITHIN_MS = 1500 _FLAG_READ_ENCRYPTED = 0x0200 _FLAG_WRITE_ENCRYPTED = 0x1000 @@ -575,7 +580,9 @@ def on_change(conn, encrypted, authenticated, bonded, key_size): try: with _Translate("pair"): _aioble_core.ble.gap_pair(handle) - deadline = time.ticks_add(time.ticks_ms(), timeout_ms) + started = time.ticks_ms() + deadline = time.ticks_add(started, timeout_ms) + typed = passkey is None # The stack reports the outcome as an encryption update, success # or not, so a wrong passkey fails here at once, not at a timeout. while not updates: @@ -583,13 +590,44 @@ def on_change(conn, encrypted, authenticated, bonded, key_size): raise PairingError("pair: the peer hung up") if time.ticks_diff(deadline, time.ticks_ms()) <= 0: raise BLETimeoutError("pair: timed out") + if not typed and handle not in security._asked and time.ticks_diff(time.ticks_ms(), started) > _ASK_WITHIN_MS: + typed = True + await self._type_passkey(handle, deadline, updates) await sleep_ms(20) finally: security.remove_listener(on_change) security.expect_passkey(handle, None) + security._asked.discard(handle) if not security.state(handle)[0]: raise PairingError("pair: refused (wrong passkey, or the peer lost its keys)") + async def _type_passkey(self, handle, deadline, updates): + # MicroPython's NimBLE bindings raise _IRQ_PASSKEY_ACTION only on + # connections the board accepted, never on one it made + # (peripheral_gap_event_cb in extmod/nimble/modbluetooth_nimble.c + # doesn't handle it; micropython/micropython#19470 moves it), so a + # board as central is never asked for the passkey and pairing stalls + # until the stack's 30 s timeout. When the stack hasn't asked by now, + # ask for the passkey here and hand it to the stack ourselves: NimBLE + # takes it once its pairing reaches the confirm step and answers + # EINVAL before that, so keep offering it until it's taken. + from . import security + + given = security._inputs.get(handle) + if callable(given): + given = given() + given = int(given) + # If the stack does ask after all, it answers from this, not by asking twice. + security._inputs[handle] = given + while not updates and handle not in security._asked and self.is_connected(): + try: + _aioble_core.ble.gap_passkey(handle, _PASSKEY_INPUT, given) + return + except OSError as e: + if e.errno != _EINVAL or time.ticks_diff(deadline, time.ticks_ms()) <= 0: + return # taken already (EALREADY), or the pairing is over + await sleep_ms(50) + async def unpair(self): from . import security diff --git a/lib/bledev/security.py b/lib/bledev/security.py index 49d8fbc3..8a67b61e 100644 --- a/lib/bledev/security.py +++ b/lib/bledev/security.py @@ -60,6 +60,9 @@ def const(x): # conn_handle -> what to type in when this board, as central, is asked for the # peer's passkey: an int, or a callable returning one. _inputs = {} +# Connections the stack has asked, as central, for the peer's passkey. Stock +# MicroPython's NimBLE never asks on a connection it made (see mpble's pair()). +_asked = set() # The server's side: what to do when the stack wants a passkey shown or a # number confirmed. Set by configure(). _show = None @@ -438,7 +441,9 @@ def _irq(event, data): core.ble.gap_passkey(conn, action, passkey) if _show is not None: _defer(_show, passkey) - elif action == _ACTION_INPUT and not callable(_inputs.get(conn)): + if action == _ACTION_INPUT: + _asked.add(conn) + if action == _ACTION_INPUT and not callable(_inputs.get(conn)): given = _inputs.get(conn) # No passkey to give: answer with one that can't be relied on to # match, so the pairing fails now rather than at a timeout. @@ -451,6 +456,7 @@ def _irq(event, data): conn = data[0] had = _links.pop(conn, None) _inputs.pop(conn, None) + _asked.discard(conn) if had is None and _hide is not None: _defer(_call_hide, None) return None From 9af1d019a6cc7b334b37fa70a22a4a4b661bd0a6 Mon Sep 17 00:00:00 2001 From: Brad Barnett <127794626+bdbarnett@users.noreply.github.com> Date: Thu, 24 Sep 2026 20:21:46 -0500 Subject: [PATCH 2/2] tests/bledev_board: UART rigs for the P4 HID gate, connect counts and a typed passkey uart_rig.py drives boards' friendly REPLs over their UARTs from Windows Python, several at once, so ESP-IDF's log stays on the console and the P4 can be hard-reset between runs (micropython-pydevices#13). p4_hid_gate.py repeats the HID gate with an S3 as the device and the P4 as host, counting alloc_acl_from_ll drops; --mode connect runs connect_loop_server/client.py to count connect timeouts. passkey_gate.py pairs two boards with a passkey carried the way a person would: read off the serving board's console, typed into the other board's. files_pair_client.run(passkey="console") is that other board's side. --- tests/bledev_board/connect_loop_client.py | 61 +++++++++ tests/bledev_board/connect_loop_server.py | 31 +++++ tests/bledev_board/files_pair_client.py | 33 ++++- tests/bledev_board/p4_hid_gate.py | 146 +++++++++++++++++++++ tests/bledev_board/passkey_gate.py | 153 ++++++++++++++++++++++ tests/bledev_board/uart_rig.py | 103 +++++++++++++++ 6 files changed, 523 insertions(+), 4 deletions(-) create mode 100644 tests/bledev_board/connect_loop_client.py create mode 100644 tests/bledev_board/connect_loop_server.py create mode 100644 tests/bledev_board/p4_hid_gate.py create mode 100644 tests/bledev_board/passkey_gate.py create mode 100644 tests/bledev_board/uart_rig.py diff --git a/tests/bledev_board/connect_loop_client.py b/tests/bledev_board/connect_loop_client.py new file mode 100644 index 00000000..1aa49701 --- /dev/null +++ b/tests/bledev_board/connect_loop_client.py @@ -0,0 +1,61 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""Connect reliability, the central: find ``connect_loop_server.py`` and +connect to it ``N`` times, counting connects that time out. + +Each attempt scans for the board (up to ``FIND_MS``), then connects with a +``CONNECT_MS`` timeout at the default connection interval (``FAST`` asks +for 7.5-15 ms), holds the link 300 ms and hangs up. Ends with a ``CONNECTS`` +line: attempts, connected, timed out, other errors, not found, and the +connect times in ms. +""" +import asyncio +import time + +import bledev +import bledev.mpble + +NAME = "bledev-conn-loop" +N = 10 +FIND_MS = 20000 +CONNECT_MS = 10000 +FAST = False + + +async def main(): + ble = bledev.mpble.get() + conn_kw = {"min_conn_interval_us": 7500, "max_conn_interval_us": 15000} if FAST else {} + ok = timeouts = other = missing = 0 + times = [] + for i in range(N): + try: + device = await ble.find(name=NAME, timeout_ms=FIND_MS) + except bledev.BLETimeoutError: + missing += 1 + print("attempt", i + 1, "not found") + continue + t = time.ticks_ms() + try: + conn = await device.connect(timeout_ms=CONNECT_MS, **conn_kw) + except bledev.BLETimeoutError as e: + timeouts += 1 + print("attempt", i + 1, "connect timed out after", time.ticks_diff(time.ticks_ms(), t), "ms:", e) + await asyncio.sleep_ms(500) + continue + except bledev.BLEError as e: + other += 1 + print("attempt", i + 1, "connect failed:", repr(e)) + await asyncio.sleep_ms(500) + continue + dt = time.ticks_diff(time.ticks_ms(), t) + ok += 1 + times.append(dt) + print("attempt", i + 1, "connected in", dt, "ms") + await asyncio.sleep_ms(300) + await conn.disconnect() + await asyncio.sleep_ms(1000) + print("CONNECTS", N, "ok", ok, "timeouts", timeouts, "other", other, "not found", missing, "times", sorted(times)) + + +asyncio.run(main()) diff --git a/tests/bledev_board/connect_loop_server.py b/tests/bledev_board/connect_loop_server.py new file mode 100644 index 00000000..dd005d90 --- /dev/null +++ b/tests/bledev_board/connect_loop_server.py @@ -0,0 +1,31 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""Connect reliability, the peripheral: advertise, take a connection, wait for +it to end, and advertise again, forever. Pair with ``connect_loop_client.py``. + +Prints ``accepted N`` for each connection. +""" +import asyncio + +import bledev +import bledev.mpble + +NAME = "bledev-conn-loop" + + +async def main(): + ble = bledev.mpble.get() + n = 0 + while True: + conn = await ble.advertise(name=NAME) + n += 1 + print("accepted", n) + try: + await conn.disconnected(30000) + except bledev.BLEError as e: + print("still connected after 30 s:", e) + await conn.disconnect() + + +asyncio.run(main()) diff --git a/tests/bledev_board/files_pair_client.py b/tests/bledev_board/files_pair_client.py index c38087ad..e08709d0 100644 --- a/tests/bledev_board/files_pair_client.py +++ b/tests/bledev_board/files_pair_client.py @@ -22,6 +22,13 @@ ``run(plant="flip")`` flips one bit in one notification the client receives, and ``run(plant="nopair")`` never pairs; each must fail. Ends with ``RESULT PASS`` or ``RESULT FAIL`` (also written to /files_pair_result.txt). + +``run(passkey="console")`` is for a serving board started with +``pair_server.start("passkey")``: when the other board shows its passkey, this +board prints ``PASSKEY?`` and reads the digits from its own console, the way a +person would type what they see. The link must then also be authenticated, and +the reconnect must not ask again. ``passkey_gate.py`` carries the digits from +one board's console to the other's. """ import asyncio import hashlib @@ -61,7 +68,17 @@ def pattern(n): return bytes(out) -async def _main(plant): +asked = [] + + +def _typed(): + # Called when the peer shows a passkey: read it off this board's console. + asked.append(time.ticks_ms()) + print("PASSKEY?") + return int(input().strip()) + + +async def _main(plant, passkey): ble = bledev.mpble.get() ble.enable_bonding() log("store", security.store().kind, "bonds before", security.bonds()) @@ -79,14 +96,18 @@ def feed(self, data): ft.Client._feed = feed t = time.ticks_ms() try: - client = await ft.connect(ble, PASSWORD, name=NAME, pair=False if plant == "nopair" else None, timeout_ms=15000) + client = await ft.connect(ble, PASSWORD, name=NAME, pair=False if plant == "nopair" else None, timeout_ms=15000, + passkey=_typed if passkey == "console" else None) except bledev.BLEError as e: check(False, "connect and pair", repr(e)) + log("passkey asked", len(asked), "bonds after", security.bonds()) return conn = client.connection log("connected and logged in in", time.ticks_diff(time.ticks_ms(), t), "ms, mtu", conn.mtu) enc = security.state(conn._aconn._conn_handle) check(enc[0] and enc[2], "the link is encrypted and bonded", enc) + if passkey: + check(enc[1] and len(asked) == 1, "the link is authenticated by the passkey", (enc, len(asked))) data = pattern(20 * 1024) t = time.ticks_ms() await client.write(PATH, data) @@ -107,17 +128,21 @@ def feed(self, data): log("reconnected from the bond in", time.ticks_diff(time.ticks_ms(), t), "ms") enc = security.state(conn._aconn._conn_handle) check(enc[0] and enc[2], "the second link is encrypted from the bond", enc) + if passkey: + check(enc[1] and len(asked) == 1, "the second link is authenticated with no passkey asked", (enc, len(asked))) entries = await client.listdir("/") check(any(e[0] == PATH[1:] and e[1] == len(data) for e in entries), "the file is on the board, 20 KB") await client.close() + log("bonds after", security.bonds()) -def run(plant=None): +def run(plant=None, passkey=None): _lines.clear() _failures.clear() + asked.clear() log("PLANTED", plant) if plant else None try: - asyncio.run(_main(plant)) + asyncio.run(_main(plant, passkey)) except Exception as e: check(False, "no exception", repr(e)) log("RESULT", "FAIL " + "; ".join(_failures) if _failures else "PASS") diff --git a/tests/bledev_board/p4_hid_gate.py b/tests/bledev_board/p4_hid_gate.py new file mode 100644 index 00000000..0275478e --- /dev/null +++ b/tests/bledev_board/p4_hid_gate.py @@ -0,0 +1,146 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""The HID gate, repeated, with a board as the device and the P4 as the host. + +Runs on Windows Python with pyserial, driving both boards' friendly REPLs over +their UARTs (``uart_rig.py``). Set up once with mpftp: + +* the device board (an S3): ``/hidp.py`` = ``hid_peripheral.py``, and + ``/hidp_plant.py`` = the same with ``PLANT = True``; +* the P4: ``/hidc.py`` = ``hid_central.py``, and ``/hidc_fast.py`` = the same + with ``FAST = True`` (a 7.5-15 ms interval); +* ``hid_script.mpy`` in both boards' ``/lib``. + +Then:: + + python.exe p4_hid_gate.py --p4 COM4 --dev COM17 --mode default --runs 10 --log hid.log + +Each run soft-resets the device and starts it, hard-resets the P4 (BLE can't +start twice after a soft reset there, micropython-pydevices#13), then starts +the host at the friendly prompt, where ESP-IDF's log still reaches the UART: +every ``alloc_acl_from_ll failed`` (a packet ESP-Hosted dropped) is counted. +``--mode plant`` runs the planted device, which must FAIL. A run whose host +never connects is counted as a connect failure and not as a gate result. + +``--mode connect`` measures connects instead: the device runs +``connect_loop_server.py`` and the P4 ``connect_loop_client.py`` (both at +``/``), ten connects per hard reset. ``--wifi-off`` turns the P4's Wi-Fi +station off after each boot, for a board whose boot.py joins a network. +""" +import re +import sys +import time + +from uart_rig import Board, wait + + +WIFI_OFF = "--wifi-off" in sys.argv # turn the P4's station off after boot.py joins it + + +def arg(name, default=None): + return sys.argv[sys.argv.index(name) + 1] if name in sys.argv else default + + +def connects(p4, dev): + r = {"mode": "connect"} + if not dev.soft_reset(): + r["error"] = "device: no prompt after soft reset" + return r + dev.mark() + dev.line("import connect_loop_server") + time.sleep(2) + if not p4.hard_reset(): + r["error"] = "P4: no prompt after hard reset" + return r + if WIFI_OFF: + p4.say("import network; network.WLAN(network.STA_IF).active(False)") + p4.mark() + p4.line("import connect_loop_client") + wait([p4, dev], lambda: (b"CONNECTS" in p4.buf or b"Traceback" in p4.buf) and p4.buf.rstrip().endswith(b">>>"), 600) + host = p4.text() + r["drops"] = host.count("alloc_acl_from_ll failed") + m = re.search(r"CONNECTS (\d+) ok (\d+) timeouts (\d+) other (\d+) not found (\d+) times (.*)", host) + if m: + r.update(attempts=int(m.group(1)), ok=int(m.group(2)), timeouts=int(m.group(3)), other=int(m.group(4)), + missing=int(m.group(5)), times=m.group(6).strip()) + else: + r["error"] = "no CONNECTS line" + r["accepted"] = dev.text().count("accepted") + return r + + +def one(p4, dev, mode): + if mode == "connect": + return connects(p4, dev) + r = {"mode": mode} + if not dev.soft_reset(): + r["error"] = "device: no prompt after soft reset" + return r + dev.mark() + dev.line("import hidp_plant" if mode == "plant" else "import hidp") + if not wait([dev], lambda: b"advertising as" in dev.buf, 20): + r["error"] = "device never advertised" + return r + if not p4.hard_reset(): + r["error"] = "P4: no prompt after hard reset" + return r + if WIFI_OFF: + p4.say("import network; network.WLAN(network.STA_IF).active(False)") + p4.mark() + p4.line("import hidc_fast" if mode == "fast" else "import hidc") + finished = lambda: (b"RESULT " in p4.buf or b"Traceback" in p4.buf) and p4.buf.rstrip().endswith(b">>>") + ok = wait([p4, dev], finished, 180) + wait([p4, dev], lambda: b"DONE" in dev.buf, 10 if b"RESULT" in p4.buf else 0) + host, device = p4.text(), dev.text() + r["drops"] = host.count("alloc_acl_from_ll failed") + m = re.search(r"connected and started in (\d+) ms", host) + r["connect_ms"] = int(m.group(1)) if m else None + m = re.search(r"events (\d+) want (\d+)", host) + r["events"] = "{}/{}".format(m.group(1), m.group(2)) if m else None + m = re.search(r"MISMATCH (.*)", host) + r["mismatch"] = m.group(1)[:100] if m else None + m = re.search(r"RESULT (\w+)", host) + r["result"] = m.group(1) if m else None + m = re.search(r"latency phase stalled after (\d+)", host) + r["stalled_after"] = int(m.group(1)) if m else None + m = re.search(r"round trip, press to the host's LED write: (.*)", device) + r["rtt"] = m.group(1).strip() if m else None + if "Traceback" in host: + tb = host[host.index("Traceback"):].strip().split("\n") + r["exception"] = tb[-2] if tb[-1].startswith(">>>") else tb[-1] + r["connect_timeout"] = r["connect_ms"] is None and "Timeout" in r["exception"] and "no device matching" not in r["exception"] + if not ok: + r["error"] = "host didn't finish in 180 s" + return r + + +def main(): + mode = arg("--mode", "default") + runs = int(arg("--runs", "1")) + log = open(arg("--log", "p4_hid_gate.log"), "a", encoding="utf-8") + p4 = Board(arg("--p4", "COM4"), "P4", log) + dev = Board(arg("--dev", "COM17"), "DEV", log) + results = [] + for i in range(runs): + log.write("==== run {} mode {} {}\n".format(i + 1, mode, time.strftime("%H:%M:%S"))) + r = one(p4, dev, mode) + results.append(r) + print("run", i + 1, r, flush=True) + dev.prompt() + p4.hard_reset() + if mode == "connect": + print("SUMMARY connect sessions", len(results), "attempts", sum(r.get("attempts", 0) for r in results), + "ok", sum(r.get("ok", 0) for r in results), "timeouts", sum(r.get("timeouts", 0) for r in results), + "other", sum(r.get("other", 0) for r in results), "not found", sum(r.get("missing", 0) for r in results), + "drops", sum(r.get("drops") or 0 for r in results)) + return + done = [r for r in results if r.get("result")] + print("SUMMARY mode", mode, "runs", len(results), "completed", len(done), + "PASS", sum(r["result"] == "PASS" for r in done), + "with drops", sum(bool(r.get("drops")) for r in results), + "drops total", sum(r.get("drops") or 0 for r in results), + "connect timeouts", sum(bool(r.get("connect_timeout")) for r in results)) + + +main() diff --git a/tests/bledev_board/passkey_gate.py b/tests/bledev_board/passkey_gate.py new file mode 100644 index 00000000..1afa68e7 --- /dev/null +++ b/tests/bledev_board/passkey_gate.py @@ -0,0 +1,153 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""The pairing gate, board to board with a passkey: one board shows it, the other types it. + +Runs on Windows Python with pyserial (``uart_rig.py``). The serving board has +a display and ``/pair_server.py``; the typing board has +``/files_pair_client.py``. Both have bledev in ``/lib``:: + + python.exe passkey_gate.py --server COM17 --client COM4 right wrong --log gate.log + +The passkey travels the way a person would carry it. The serving board draws +it on its screen and prints it on its console (``bledev: Bluetooth passkey +123 456``); this script reads that line off the serving board's UART. The +typing board asks with ``PASSKEY?`` on its own console, and this script types +the digits into that board's UART, where ``input()`` reads them. Nothing else +passes between the two. + +Steps, each starting from hard resets and no bonds on either board: + +``right`` type what the screen shows. The link must be encrypted, + authenticated and bonded on both boards, 20 KB must cross both ways + byte for byte (the serving board's copy hashed too), and a second + connection must encrypt from the bond without asking again. After + hard resets both boards must still hold one bond. +``wrong`` type the passkey one off. Pairing must be refused, and after hard + resets neither board may hold a bond. + +``--plant justworks`` starts the server with pair_server's ``justworks`` +plant (it says passkey but pairs without one); ``wrong`` must then FAIL. +Ends with ``GATE PASS`` or ``GATE FAIL``. +""" +import re +import sys +import time + +from uart_rig import Board, wait + +BONDS = "from bledev import security; security.use_store(); print('BONDS', security.bonds())" + + +def arg(name, default=None): + return sys.argv[sys.argv.index(name) + 1] if name in sys.argv else default + + +def bonds(board): + out = board.say(BONDS) + m = re.search(r"BONDS (\d+)", out) + return int(m.group(1)) if m else None + + +def fresh(server, client): + for b in (server, client): + b.hard_reset() + b.say("from bledev import security; security.use_store(); security.forget()") + server.say("import os; [os.remove(f) for f in os.listdir('/') if f == 'pair_20k.bin']") + + +def step(server, client, kind, plant, say): + failures = [] + + def check(ok, what): + say(("PASS " if ok else "FAIL ") + what) + if not ok: + failures.append(what) + + fresh(server, client) + check(bonds(server) == 0 and bonds(client) == 0, "both boards start with no bond") + for b in (server, client): + b.hard_reset() + started = server.say("import pair_server; pair_server.start('passkey'{})".format( + ", plant='justworks'" if plant else ""), 20) + say("server:", started) + client.mark() + server.mark() + client.line("import files_pair_client as c; c.run(passkey='console')") + t0 = time.monotonic() + shown = asked = None + finished = lambda: b"RESULT" in client.buf and client.buf.rstrip().endswith(b">>>") + + def ready(): + nonlocal shown, asked + m = re.search(rb"Bluetooth passkey (\d{3}) (\d{3})", server.buf) + if m and shown is None: + shown = int(m.group(1) + m.group(2)) + say("the server's console showed {:06d} after {:.2f} s".format(shown, time.monotonic() - t0)) + if b"PASSKEY?" in client.buf and asked is None: + asked = time.monotonic() - t0 + say("the client asked for it after {:.2f} s".format(asked)) + return (shown is not None and asked is not None) or finished() + + wait([server, client], ready, 60) + if shown is not None and asked is not None: + typed = shown if kind == "right" else (shown + 1) % 1000000 + say("typing {:06d} into the client's console ({})".format(typed, "as shown" if typed == shown else "one off")) + client.write(b"%06d\r" % typed) + else: + say("no passkey exchange: shown", shown, "asked", asked) + wait([server, client], finished, 120) + out = client.text() + for line in out.split("\n"): + if line.startswith(("PASS", "FAIL", "RESULT", "connected", "write", "reconnected", "sha256", "passkey asked", + "bonds after", "store")): + say("client:", line.strip()) + status = server.say("pair_server.status()") + say("server:", status) + result = re.search(r"RESULT (.*)", out) + result = result.group(1).strip() if result else None + if kind == "right": + check(result == "PASS", "the client's checks pass (paired, authenticated, 20 KB, reconnect from the bond)") + check(shown is not None and asked is not None, "the passkey was shown and asked for") + sha = re.search(r"sha256 ([0-9a-f]{64})", out) + theirs = server.say("import hashlib, binascii; print(binascii.hexlify(hashlib.sha256(" + "open('/pair_20k.bin','rb').read()).digest()).decode())") + check(bool(sha) and sha.group(1) in theirs, "the serving board's copy has the same SHA-256") + m = re.search(r"links \{(.*?)\}", status) + say("server links:", m.group(1) if m else None) + else: + check(result is not None and result.startswith("FAIL") and "connect and pair" in result, + "pairing is refused") + for b in (server, client): + b.hard_reset() + n_server, n_client = bonds(server), bonds(client) + say("after hard resets: server bonds", n_server, "client bonds", n_client) + want = 1 if kind == "right" else 0 + check(n_server == want and n_client == want, "each board holds {} bond(s) after resets".format(want)) + return failures + + +def main(): + log = open(arg("--log", "passkey_gate.log"), "a", encoding="utf-8") + server = Board(arg("--server", "COM17"), "SERVER", log) + client = Board(arg("--client", "COM4"), "CLIENT", log) + plant = arg("--plant") + steps = [a for a in sys.argv[1:] if a in ("right", "wrong")] + + def say(*parts): + line = " ".join(str(p) for p in parts) + print(line, flush=True) + log.write("#### " + line + "\n") + log.flush() + + failed = [] + for kind in steps: + say("==== step", kind, "plant", plant, time.strftime("%H:%M:%S")) + failed += ["{}: {}".format(kind, f) for f in step(server, client, kind, plant, say)] + for b in (server, client): + b.say("from bledev import security; security.use_store(); security.forget()") + b.hard_reset() + say("GATE", "FAIL " + "; ".join(failed) if failed else "PASS") + + +main() diff --git a/tests/bledev_board/uart_rig.py b/tests/bledev_board/uart_rig.py new file mode 100644 index 00000000..b81fdfa6 --- /dev/null +++ b/tests/bledev_board/uart_rig.py @@ -0,0 +1,103 @@ +# SPDX-FileCopyrightText: 2026 Brad Barnett +# +# SPDX-License-Identifier: MIT +"""Drive boards' friendly REPLs over their UARTs, several at once, from Windows Python. + +The board-to-board gates that need what mpftp can't give use this: a run +started at the friendly prompt keeps ESP-IDF's log on the console (the raw +REPL mutes it, so the P4's ``alloc_acl_from_ll failed`` only shows here), a +hard reset between runs (the P4 can't turn BLE on twice after a soft reset, +micropython-pydevices#13), and one board's console read while another board +is typed at, which is how ``passkey_gate.py`` carries a passkey from one +screen to another board's keyboard. + +Ports open with pyserial's defaults (DTR and RTS asserted), as mpremote opens +them: on the LCD-7's CH343, opening with both released resets the board. +Everything each board prints goes to the log, prefixed with its name. +""" +import time + +import serial + + +class Board: + def __init__(self, port, name, log=None): + self.s = serial.Serial(port, 115200, timeout=0) + self.name = name + self.log = log + self.buf = b"" # since the last mark() + self._partial = b"" + + def pump(self): + data = self.s.read(8192) + if data: + self.buf += data + if self.log is not None: + self._partial += data + *lines, self._partial = self._partial.split(b"\n") + for line in lines: + self.log.write("{:9.3f} {} | {}\n".format( + time.monotonic(), self.name, line.decode("utf-8", "replace").rstrip("\r"))) + self.log.flush() + return data + + def mark(self): + self.pump() + self.buf = b"" + + def text(self): + return self.buf.decode("utf-8", "replace").replace("\r", "") + + def write(self, data): + self.s.write(data) + + def line(self, text): + self.s.write(text.encode() + b"\r") + + def prompt(self, timeout=5): + """Stop whatever runs and get to the friendly prompt.""" + self.mark() + self.write(b"\x03\x03") + time.sleep(0.1) + self.write(b"\r") + return wait([self], lambda: self.buf.rstrip().endswith(b">>>"), timeout) + + def soft_reset(self, timeout=20): + self.prompt() + self.mark() + self.write(b"\x04") + return wait([self], lambda: b"soft reboot" in self.buf and self.buf.rstrip().endswith(b">>>"), timeout) + + def hard_reset(self, timeout=40): + """machine.reset() from the prompt, then wait for the next prompt.""" + self.prompt() + self.mark() + self.line("import machine; machine.reset()") + return wait([self], lambda: b"MicroPython v" in self.buf and self.buf.rstrip().endswith(b">>>"), timeout) + + def say(self, text, timeout=10): + """Type one line at the prompt and return what it printed.""" + self.mark() + self.line(text) + wait([self], lambda: b"\n" in self.buf and self.buf.split(b"\n", 1)[1].rstrip().endswith(b">>>"), timeout) + body = self.text().split("\n", 1)[1] if "\n" in self.text() else "" + return body.rstrip().removesuffix(">>>").rstrip() + + def close(self): + self.s.close() + + +def wait(boards, done, timeout): + """Pump every board until ``done()`` or the timeout. True if done.""" + end = time.monotonic() + timeout + while True: + got = False + for b in boards: + if b.pump(): + got = True + if done(): + return True + if time.monotonic() > end: + return False + if not got: + time.sleep(0.01)