refactor(python): better management of open-depth for transport
What the developer wrote
Authored by matejcik
refactor(python): better management of open-depth for transport
this way a high-level "reopen" functionality is exposed to callers that
want to cycle the connection, without needing to fragilly mess with open
count
[no changelog]
What changed, and why it matters
This is a Python-only internal refactoring of how Trezor's test/debugging library keeps track of open connections. It replaces fragile manual open/close bookkeeping with a cleaner 'reopen' flag and context-manager handling. There is no direct evidence this fixes an exploitable security bug; it appears to be code-quality work to prevent connection-state mistakes during testing.
Treat as a normal maintenance/refactoring patch. Reviewers may optionally verify that the new `is_open()` implementations correctly reflect actual connection state for each transport, since incorrect state reporting could lead to connection-handling bugs, but no immediate security action is indicated.
Security signals we found
Refactors resource-lifecycle management, which can indirectly affect reliability and availability
Removes manual exception-swallowing close/open patterns in debug/test code
Adds explicit warning logs when open/close calls are mismatched (already-open open(), context-managed close())
No direct memory-safety, cryptographic, or authorization changes visible in diff
Evidence from the diff
The commit refactors the Transport base class in trezorlib to manage an ‘open-depth’ counter more robustly. It adds an open(reopen=False) parameter that lets callers request a connection reset without manually saving and restoring _opened. It also fixes __enter__ so nested context managers increment the counter correctly, and adds an abstract is_open() method with concrete implementations across BLE, Bridge, HID, UDP, and WebUSB transports. is_ready() now defaults to is_open(). Callers such as DebugLink, TrezorTestContext.reset_instance(), EmuBleTransport.wait_until_ready(), and UdpTransport.wait_until_ready() are simplified to use open(reopen=True) or with self: instead of manual try/finally close logic.
Changed components
python/src/trezorlib/transport/__init__.pypython/src/trezorlib/transport/ble.pypython/src/trezorlib/transport/bridge.pypython/src/trezorlib/transport/hid.pypython/src/trezorlib/transport/udp.pypython/src/trezorlib/transport/webusb.pypython/src/trezorlib/debuglink.pypython/src/trezorlib/_internal/emu_ble.pyInspect captured patch +54 / −33
diff --git a/python/src/trezorlib/_internal/emu_ble.py b/python/src/trezorlib/_internal/emu_ble.py
index 590780b9..ef8b1bb9 100644
--- a/python/src/trezorlib/_internal/emu_ble.py
+++ b/python/src/trezorlib/_internal/emu_ble.py
@@ -227,8 +227,7 @@ class EmuBleTransport(Transport):
return UdpTransport(f"{host}:{port - 3}")
def wait_until_ready(self, timeout: float = 10) -> None:
- try:
- self.open()
+ with self:
start = time.monotonic()
while True:
if self.ping():
@@ -238,8 +237,6 @@ class EmuBleTransport(Transport):
raise Timeout("Timed out waiting for connection.")
time.sleep(0.05)
- finally:
- self.close()
def ping(self) -> bool:
"""Test if the device is listening."""
diff --git a/python/src/trezorlib/debuglink.py b/python/src/trezorlib/debuglink.py
index 9161f1f7..baaf80ca 100644
--- a/python/src/trezorlib/debuglink.py
+++ b/python/src/trezorlib/debuglink.py
@@ -516,11 +516,7 @@ def _make_input_func(
class DebugLink:
def __init__(self, transport: "Transport", auto_interact: bool = True) -> None:
- try:
- self.transport.close()
- except Exception:
- pass
- transport.open()
+ transport.open(reopen=True)
self.transport = transport
self.allow_interactions = auto_interact
@@ -1354,13 +1350,7 @@ class TrezorTestContext:
return client
def reset_instance(self) -> None:
- try:
- height = self.transport._opened
- self.transport.close()
- except Exception:
- pass
- self.transport.open()
- self.transport._opened = height
+ self.transport.open(reopen=True)
# debug-specific initialization
self.ui: DebugUI = DebugUI(self.debug)
diff --git a/python/src/trezorlib/transport/__init__.py b/python/src/trezorlib/transport/__init__.py
index 5ae74add..48a0cf10 100644
--- a/python/src/trezorlib/transport/__init__.py
+++ b/python/src/trezorlib/transport/__init__.py
@@ -89,20 +89,40 @@ class Transport(metaclass=ABCMeta):
def find_debug(self) -> Transport:
raise NotImplementedError
- def open(self) -> None:
- LOG.info(f"Opening transport: {self}")
- self._open()
- self._opened = max(self._opened, 1)
+ def open(self, reopen: bool = False) -> None:
+ if self._opened == 0:
+ # the natural case: open a closed transport
+ LOG.info(f"Opening transport: {self}")
+ self._open()
+ self._opened = 1
+ return
+
+ if reopen:
+ # transport is already open, we want to close and reestablish
+ # the connection at the same open-height
+ LOG.info(f"Closing transport and reopening: {self}")
+ self._close()
+ self._open()
+ return
+
+ # finally, someone's calling open() when they're already open
+ # via a context manager.
+ LOG.warning(f"Transport {self} is already open")
def close(self) -> None:
+ if self._opened > 1:
+ LOG.warning(
+ f"Transport {self} is open via a context manager. Closing unconditionally."
+ )
LOG.info(f"Closing transport: {self}")
self._close()
self._opened = 0
def __enter__(self) -> Transport:
- self._opened += 1
- if self._opened == 1: # means it previously was 0
- self.open()
+ if self._opened == 0:
+ self.open() # resets self._opened to 1
+ else:
+ self._opened += 1
return self
def __exit__(
@@ -116,6 +136,10 @@ class Transport(metaclass=ABCMeta):
if self._opened == 0:
self.close()
+ @abstractmethod
+ def is_open(self) -> bool:
+ raise NotImplementedError
+
@abstractmethod
def _open(self) -> None:
raise NotImplementedError
@@ -132,9 +156,8 @@ class Transport(metaclass=ABCMeta):
def read_chunk(self, *, timeout: float | None = None) -> bytes:
raise NotImplementedError
- @abstractmethod
def is_ready(self) -> bool:
- raise NotImplementedError
+ return self.is_open()
def all_transports() -> t.Iterable[type[Transport]]:
diff --git a/python/src/trezorlib/transport/ble.py b/python/src/trezorlib/transport/ble.py
index e5e7164c..558229d2 100644
--- a/python/src/trezorlib/transport/ble.py
+++ b/python/src/trezorlib/transport/ble.py
@@ -106,6 +106,9 @@ class BleTransport(Transport):
# instead we rely on atexit handler to avoid reconnecting
pass
+ def is_open(self) -> bool:
+ return self.ble_proxy().is_connected(self.device)
+
def write_chunk(self, chunk: bytes) -> None:
LOG.log(DUMP_PACKETS, f"sending packet: {chunk.hex()}")
self.ble_proxy().write(self.device, chunk)
@@ -303,6 +306,12 @@ class BleAsync:
periph.queue = queue
LOG.info(f"Connected to {client.address}")
+ async def is_connected(self, address: str) -> bool:
+ periph = self.devices.get(address)
+ if not periph:
+ return False
+ return bool(periph.client)
+
async def disconnect(self, address: str) -> None:
periph = self.devices.get(address)
if not periph or not periph.client:
diff --git a/python/src/trezorlib/transport/bridge.py b/python/src/trezorlib/transport/bridge.py
index e1471f37..bf944f70 100644
--- a/python/src/trezorlib/transport/bridge.py
+++ b/python/src/trezorlib/transport/bridge.py
@@ -195,5 +195,5 @@ class BridgeTransport(Transport):
def read_chunk(self, *, timeout: float | None = None) -> bytes:
return self.handle.read_buf(timeout=timeout)
- def is_ready(self) -> bool:
+ def is_open(self) -> bool:
return self.session is not None
diff --git a/python/src/trezorlib/transport/hid.py b/python/src/trezorlib/transport/hid.py
index 40f18438..82f750b8 100644
--- a/python/src/trezorlib/transport/hid.py
+++ b/python/src/trezorlib/transport/hid.py
@@ -156,7 +156,7 @@ class HidTransport(Transport):
return 1
raise TransportException("Unknown HID version")
- def is_ready(self) -> bool:
+ def is_open(self) -> bool:
return self.handle is not None
diff --git a/python/src/trezorlib/transport/udp.py b/python/src/trezorlib/transport/udp.py
index 3698d71d..258d5644 100644
--- a/python/src/trezorlib/transport/udp.py
+++ b/python/src/trezorlib/transport/udp.py
@@ -132,8 +132,7 @@ class UdpTransport(Transport):
return UdpTransport(f"{host}:{port + 1}")
def wait_until_ready(self, timeout: float = 10) -> None:
- try:
- self.open()
+ with self:
start = time.monotonic()
while True:
if self.is_ready():
@@ -143,12 +142,15 @@ class UdpTransport(Transport):
raise Timeout("Timed out waiting for connection.")
time.sleep(0.05)
- finally:
- self.close()
+
+ def is_open(self) -> bool:
+ return self.socket is not None
def is_ready(self) -> bool:
"""Test if the device is listening."""
- assert self.socket is not None
+ if not self.is_open():
+ return False
+ assert self.socket is not None # pyright fails otherwise
try:
LOG.log(DUMP_PACKETS, f"PINGing {self.device}")
self.socket.sendall(b"PINGPING")
diff --git a/python/src/trezorlib/transport/webusb.py b/python/src/trezorlib/transport/webusb.py
index 6dcdd4df..b6ae6af7 100644
--- a/python/src/trezorlib/transport/webusb.py
+++ b/python/src/trezorlib/transport/webusb.py
@@ -181,7 +181,7 @@ class WebUsbTransport(Transport):
# For v1 protocol, find debug USB interface for the same serial number
return self.__class__(self.device, debug=True)
- def is_ready(self) -> bool:
+ def is_open(self) -> bool:
return self.handle is not None
Why this scored 17/100
Community notes
Notes can correct, qualify, or add evidence to the AI analysis. Every note shown here has been validated by a human moderator.
The AI analysis stands alone for now. Submit a note if you can add evidence or important context.