BLE transport for trezorlib

common/protob/Makefile

@@ -1,4 +1,4 @@
-check: messages.pb messages-bitcoin.pb messages-bootloader.pb messages-cardano.pb messages-common.pb messages-crypto.pb messages-debug.pb messages-ethereum.pb messages-management.pb messages-monero.pb messages-nem.pb messages-ripple.pb messages-stellar.pb messages-tezos.pb messages-eos.pb messages-solana.pb messages-definitions.pb
+check: messages.pb messages-bitcoin.pb messages-ble.pb messages-bootloader.pb messages-cardano.pb messages-common.pb messages-crypto.pb messages-debug.pb messages-ethereum.pb messages-management.pb messages-monero.pb messages-nem.pb messages-ripple.pb messages-stellar.pb messages-tezos.pb messages-eos.pb messages-solana.pb messages-definitions.pb
 
 %.pb: %.proto
 	protoc -I/usr/include -I. $< -o $@

pyproject.toml

@@ -60,6 +60,7 @@ dependencies = [
     "pyelftools>=0.32,<0.33",
     "pytest-retry>=1.7.0,<2",
     "slh-dsa>=0.1.3,<0.2",
+    "bleak>=1.1.0",
 ]
 
 [dependency-groups]

python/.changelog.d/4948.added

@@ -0,0 +1 @@
+Added support for Bluetooth Low Energy transport.

python/pyproject.toml

@@ -40,12 +40,14 @@ ethereum = ["web3>=5"]
 qt-widgets = ["PyQt5"]
 extra = ["Pillow>=10"]
 stellar = ["stellar-sdk>=6"]
+bleak = ["bleak>=1.1.0"]
 full = [
     "hidapi>=0.7.99.post20",
     "web3>=5",
     "PyQt5",
     "Pillow>=10",
     "stellar-sdk>=6",
+    "bleak>=1.1.0",
 ]
 
 [project.urls]

python/src/trezorlib/cli/__init__.py

@@ -139,11 +139,13 @@ def __init__(
         session_id: bytes | None,
         passphrase_on_host: bool,
         script: bool,
+        ble_enabled: bool,
     ) -> None:
         self.path = path
         self.session_id = session_id
         self.passphrase_on_host = passphrase_on_host
         self.script = script
+        self.ble_enabled = ble_enabled
 
     def get_session(
         self,
@@ -208,15 +210,19 @@ def get_transport(self) -> "Transport":
 
         try:
             # look for transport without prefix search
-            _TRANSPORT = transport.get_transport(self.path, prefix_search=False)
+            _TRANSPORT = transport.get_transport(
+                self.path, prefix_search=False, ble_enabled=self.ble_enabled
+            )
         except Exception:
             # most likely not found. try again below.
             pass
 
         # look for transport with prefix search
         # if this fails, we want the exception to bubble up to the caller
         if not _TRANSPORT:
-            _TRANSPORT = transport.get_transport(self.path, prefix_search=True)
+            _TRANSPORT = transport.get_transport(
+                self.path, prefix_search=True, ble_enabled=self.ble_enabled
+            )
 
         _TRANSPORT.open()
         atexit.register(_TRANSPORT.close)

python/src/trezorlib/cli/ble.py

@@ -20,6 +20,7 @@
 import click
 
 from .. import ble, exceptions
+from ..transport.ble import BleProxy
 from . import with_session
 
 if TYPE_CHECKING:
@@ -43,7 +44,7 @@ def unpair(
     session: "Session",
     all: bool,
 ) -> None:
-    """Erase bond of currently connected device, or all devices (on device side)"""
+    """Erase bond of currently connected device, or all devices (on device side)."""
 
     try:
         ble.unpair(session, all)
@@ -53,3 +54,29 @@ def unpair(
     except exceptions.TrezorException as e:
         click.echo(f"Unpair failed: {e}")
         sys.exit(3)
+
+
+@cli.command()
+def connect() -> None:
+    """Connect to the device via BLE. Device has to be disconnected beforehand.
+
+    If the device hasn't been paired you also need to have system bluetooth pairing dialog open.
+    """
+    ble = BleProxy()
+
+    click.echo("Scanning...")
+    devices = ble.scan()
+
+    if len(devices) == 0:
+        click.echo("No BLE devices found")
+        return
+    else:
+        click.echo(f"Found {len(devices)} BLE device(s)")
+
+    for address, name in devices:
+        click.echo(f"Device: {name}, {address}")
+
+    device = devices[0]
+    click.echo(f"Connecting to {device[1]}...")
+    ble.connect(device[0])
+    click.echo("Connected")

python/src/trezorlib/cli/trezorctl.py

@@ -167,6 +167,13 @@ def configure_logging(verbose: int) -> None:
     help="Select device by specific path.",
     default=os.environ.get("TREZOR_PATH"),
 )
+@click.option(
+    "-B",
+    "--ble/--no-ble",
+    help="Enable/disable support for Bluetooth Low Energy.",
+    is_flag=True,
+    default=(os.environ.get("TREZOR_BLE") == "1"),
+)
 @click.option("-v", "--verbose", count=True, help="Show communication messages.")
 @click.option(
     "-j", "--json", "is_json", is_flag=True, help="Print result as JSON object"
@@ -200,6 +207,7 @@ def configure_logging(verbose: int) -> None:
 def cli_main(
     ctx: click.Context,
     path: str,
+    ble: bool,
     verbose: int,
     is_json: bool,
     passphrase_on_host: bool,
@@ -216,7 +224,9 @@ def cli_main(
         except ValueError:
             raise click.ClickException(f"Not a valid session id: {session_id}")
 
-    ctx.obj = TrezorConnection(path, bytes_session_id, passphrase_on_host, script)
+    ctx.obj = TrezorConnection(
+        path, bytes_session_id, passphrase_on_host, script, ble_enabled=ble
+    )
 
     # Optionally record the screen into a specified directory.
     if record:
@@ -284,16 +294,19 @@ def format_device_name(features: messages.Features) -> str:
 
 @cli.command(name="list")
 @click.option("-n", "no_resolve", is_flag=True, help="Do not resolve Trezor names")
-def list_devices(no_resolve: bool) -> Optional[Iterable["Transport"]]:
+@click.pass_obj
+def list_devices(
+    obj: TrezorConnection, no_resolve: bool
+) -> Optional[Iterable["Transport"]]:
     """List connected Trezor devices."""
     if no_resolve:
-        for d in enumerate_devices():
+        for d in enumerate_devices(ble_enabled=obj.ble_enabled):
             click.echo(d.get_path())
         return
 
     from . import get_client
 
-    for transport in enumerate_devices():
+    for transport in enumerate_devices(ble_enabled=obj.ble_enabled):
         try:
             transport.open()
             client = get_client(transport)

python/src/trezorlib/transport/__init__.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import logging
+import os
 import typing as t
 
 from ..exceptions import TrezorException
@@ -95,7 +96,8 @@ def ping(self) -> bool:
     CHUNK_SIZE: t.ClassVar[int | None]
 
 
-def all_transports() -> t.Iterable[t.Type["Transport"]]:
+def all_transports(ble_enabled: bool | None = None) -> t.Iterable[t.Type["Transport"]]:
+    from .ble import BleTransport
     from .bridge import BridgeTransport
     from .hid import HidTransport
     from .udp import UdpTransport
@@ -107,14 +109,19 @@ def all_transports() -> t.Iterable[t.Type["Transport"]]:
         UdpTransport,
         WebUsbTransport,
     )
+    if ble_enabled is None:
+        ble_enabled = os.environ.get("TREZOR_BLE") == "1"
+    if ble_enabled:
+        transports += (BleTransport,)
     return set(t for t in transports if t.ENABLED)
 
 
 def enumerate_devices(
     models: t.Iterable[TrezorModel] | None = None,
+    ble_enabled: bool | None = None,
 ) -> t.Sequence[Transport]:
     devices: t.List[Transport] = []
-    for transport in all_transports():
+    for transport in all_transports(ble_enabled=ble_enabled):
         name = transport.__name__
         try:
             found = list(transport.enumerate(models))
@@ -128,10 +135,14 @@ def enumerate_devices(
     return devices
 
 
-def get_transport(path: str | None = None, prefix_search: bool = False) -> Transport:
+def get_transport(
+    path: str | None = None,
+    prefix_search: bool = False,
+    ble_enabled: bool | None = None,
+) -> Transport:
     if path is None:
         try:
-            return next(iter(enumerate_devices()))
+            return next(iter(enumerate_devices(ble_enabled=ble_enabled)))
         except StopIteration:
             raise TransportException("No Trezor device found") from None
 
@@ -146,7 +157,11 @@ def match_prefix(a: str, b: str) -> bool:
             "prefix" if prefix_search else "full path", path
         )
     )
-    transports = [t for t in all_transports() if match_prefix(path, t.PATH_PREFIX)]
+    transports = [
+        t
+        for t in all_transports(ble_enabled=ble_enabled)
+        if match_prefix(path, t.PATH_PREFIX)
+    ]
     if transports:
         return transports[0].find_by_path(path, prefix_search=prefix_search)