arno/trezor-firmware
1
0
Fork 0
mirror of https://github.com/trezor/trezor-firmware.git synced 2025-03-27 13:35:44 +00:00
Code Issues Releases Wiki Activity

fix(python/debuglink): time out after waiting for empty-state response

Browse Source 
...avoiding a problem where, if the timing is unfortunate, the reply may
get lost when a workflow is ending

(cherry picked from commit 3056b4934e)

[no changelog]
This commit is contained in:
matejcik 2025-02-21 12:21:09 +01:00 committed by Roman Zeyde
parent 49c9ad04cd
commit 2e72a8ce1b
1 changed files with 12 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
python/src/trezorlib/debuglink.py
View File

@ -47,6 +47,7 @@ from .client import TrezorClient
from .exceptions import TrezorFailure, UnexpectedMessageError from .exceptions import TrezorFailure, UnexpectedMessageError
from .log import DUMP_BYTES from .log import DUMP_BYTES
from .messages import DebugWaitType from .messages import DebugWaitType
from .transport import Timeout
if TYPE_CHECKING: if TYPE_CHECKING:
from typing_extensions import Protocol from typing_extensions import Protocol
@ -502,8 +503,8 @@ class DebugLink:
) )
self.transport.write(msg_type, msg_bytes) self.transport.write(msg_type, msg_bytes)
def _read(self) -> protobuf.MessageType: def _read(self, timeout: float | None = None) -> protobuf.MessageType:
ret_type, ret_bytes = self.transport.read() ret_type, ret_bytes = self.transport.read(timeout=timeout)
LOG.log( LOG.log(
DUMP_BYTES, DUMP_BYTES,
f"received type {ret_type} ({len(ret_bytes)} bytes): {ret_bytes.hex()}", f"received type {ret_type} ({len(ret_bytes)} bytes): {ret_bytes.hex()}",
@ -522,9 +523,9 @@ class DebugLink:
) )
return msg return msg
def _call(self, msg: protobuf.MessageType) -> Any: def _call(self, msg: protobuf.MessageType, timeout: float | None = None) -> Any:
self._write(msg) self._write(msg)
return self._read() return self._read(timeout=timeout)
def state(self, wait_type: DebugWaitType | None = None) -> messages.DebugLinkState: def state(self, wait_type: DebugWaitType | None = None) -> messages.DebugLinkState:
if wait_type is None: if wait_type is None:
@ -649,7 +650,13 @@ class DebugLink:
if self.model is models.T1B1: if self.model is models.T1B1:
return return
# When the call below returns, we know that `decision` has been processed in Core. # When the call below returns, we know that `decision` has been processed in Core.
self._call(messages.DebugLinkGetState(return_empty_state=True)) # XXX Due to a bug, the reply may get lost at the end of a workflow.
# We assume that no single input event takes more than 5 seconds to process,
# and give up waiting after that.
try:
self._call(messages.DebugLinkGetState(return_empty_state=True), timeout=5)
except Timeout as e:
LOG.warning("timeout waiting for DebugLinkState: %s", e)
press_yes = _make_input_func(button=messages.DebugButton.YES) press_yes = _make_input_func(button=messages.DebugButton.YES)
"""Confirm current layout. See `_decision` for more details.""" """Confirm current layout. See `_decision` for more details."""