From a93c45b59b0d5f10846aabb3d825003c480b1de2 Mon Sep 17 00:00:00 2001 From: Roman Zeyde Date: Wed, 20 Aug 2025 14:39:47 +0300 Subject: [PATCH 1/6] refactor(python): refactor THP-related exception types [no changelog] --- python/src/trezorlib/cli/__init__.py | 2 +- python/src/trezorlib/debuglink.py | 16 ++------- python/src/trezorlib/exceptions.py | 26 ++++++++++++-- .../trezorlib/transport/thp/protocol_v2.py | 35 ++++++------------- tests/device_tests/thp/test_multiple_hosts.py | 5 ++- tests/device_tests/thp/test_pairing.py | 6 ++-- 6 files changed, 41 insertions(+), 49 deletions(-) diff --git a/python/src/trezorlib/cli/__init__.py b/python/src/trezorlib/cli/__init__.py index c1f0b5d58bd..4509e863743 100644 --- a/python/src/trezorlib/cli/__init__.py +++ b/python/src/trezorlib/cli/__init__.py @@ -293,7 +293,7 @@ def session_context( empty_passphrase=empty_passphrase, must_resume=must_resume, ) - except exceptions.DeviceLockedException: + except exceptions.DeviceLocked: click.echo( "Device is locked, enter a pin on the device.", err=True, diff --git a/python/src/trezorlib/debuglink.py b/python/src/trezorlib/debuglink.py index 1a47f7ddb26..af92bb5c627 100644 --- a/python/src/trezorlib/debuglink.py +++ b/python/src/trezorlib/debuglink.py @@ -34,12 +34,7 @@ from . import btc, mapping, messages, models, protobuf from .client import ProtocolVersion, TrezorClient -from .exceptions import ( - Cancelled, - DeviceLockedException, - TrezorFailure, - UnexpectedMessageError, -) +from .exceptions import Cancelled, TrezorFailure, UnexpectedMessageError from .log import DUMP_BYTES from .messages import DebugTouchEventType, DebugWaitType from .tools import parse_path @@ -1317,14 +1312,7 @@ def get_pin(_msg: messages.PinMatrixRequest) -> str: self.pin_callback = get_pin self.button_callback = self.ui.button_request - try: - super().__init__(transport) - except DeviceLockedException: - LOG.debug("Locked device handling") - self.debug.input("") - self.debug.input(self.debug.encode_pin("1234")) - super().__init__(transport) - + super().__init__(transport) self.sync_responses() # So that we can choose right screenshotting logic (T1 vs TT) diff --git a/python/src/trezorlib/exceptions.py b/python/src/trezorlib/exceptions.py index 95d0cd30f2c..0f8c0d1f275 100644 --- a/python/src/trezorlib/exceptions.py +++ b/python/src/trezorlib/exceptions.py @@ -111,13 +111,33 @@ class DerivationOnUninitaizedDeviceError(TrezorException): To communicate with uninitialized device, use seedless session instead.""" -class DeviceLockedException(TrezorException): +class UnexpectedCodeEntryTagException(TrezorException): pass -class UnexpectedCodeEntryTagException(TrezorException): +class ThpError(TrezorException): pass -class ThpError(TrezorException): +class TransportBusy(ThpError): + pass + + +class UnallocatedChannel(ThpError): + pass + + +class DecryptionFailed(ThpError): + pass + + +class InvalidData(ThpError): + pass + + +class DeviceLocked(ThpError): + pass + + +class ThpUnknownError(ThpError): pass diff --git a/python/src/trezorlib/transport/thp/protocol_v2.py b/python/src/trezorlib/transport/thp/protocol_v2.py index c805e249f2a..7a56a57812c 100644 --- a/python/src/trezorlib/transport/thp/protocol_v2.py +++ b/python/src/trezorlib/transport/thp/protocol_v2.py @@ -231,11 +231,6 @@ def _send_handshake_init_request(self) -> None: def _read_handshake_init_response(self) -> bytes: header, payload = self._read_until_valid_crc_check() - if control_byte.is_error(header.ctrl_byte): - if payload == b"\x05": - raise exceptions.DeviceLockedException() - else: - raise exceptions.ThpError(_get_error_from_int(payload[0])) if not header.is_handshake_init_response(): LOG.error("Received message is not a valid handshake init response message") @@ -278,8 +273,6 @@ def _read_handshake_completion_response(self) -> int: header, data = self._read_until_valid_crc_check() if not header.is_handshake_comp_response(): LOG.error("Received message is not a valid handshake completion response") - if control_byte.is_error(header.ctrl_byte): - raise exceptions.ThpError(_get_error_from_int(data[0])) trezor_state = self._noise.decrypt(bytes(data)) assert trezor_state == b"\x00" or trezor_state == b"\x01" self._send_ack_bit(bit=1) @@ -289,8 +282,6 @@ def _read_ack(self): header, payload = self._read_until_valid_crc_check() if not header.is_ack() or len(payload) > 0: LOG.error("Received message is not a valid ACK") - if control_byte.is_error(header.ctrl_byte): - raise exceptions.ThpError(_get_error_from_int(payload[0])) def _send_ack_bit(self, bit: int): if bit not in (0, 1): @@ -336,8 +327,6 @@ def read_and_decrypt( continue if control_byte.is_ack(header.ctrl_byte): continue - if control_byte.is_error(header.ctrl_byte): - raise exceptions.ThpError(_get_error_from_int(raw_payload[0])) if not header.is_encrypted_transport(): LOG.error( "Trying to decrypt not encrypted message! (" @@ -392,6 +381,10 @@ def _read_until_valid_crc_check( self.sync_bit_receive = 1 - self.sync_bit_receive + if control_byte.is_error(header.ctrl_byte): + code = payload[0] + raise _ERRORS_MAP.get(code) or exceptions.ThpUnknownError(code) + return header, payload def _is_valid_channel_allocation_response( @@ -420,16 +413,10 @@ def _is_valid_pong( return True -def _get_error_from_int(error_code: int) -> str: - # TODO FIXME improve this (ThpErrorType) - if error_code == 1: - return "TRANSPORT BUSY" - if error_code == 2: - return "UNALLOCATED CHANNEL" - if error_code == 3: - return "DECRYPTION FAILED" - if error_code == 4: - return "INVALID DATA" - if error_code == 5: - return "DEVICE LOCKED" - raise Exception("Not Implemented error case") +_ERRORS_MAP = { + 1: exceptions.TransportBusy, + 2: exceptions.UnallocatedChannel, + 3: exceptions.DecryptionFailed, + 4: exceptions.InvalidData, + 5: exceptions.DeviceLocked, +} diff --git a/tests/device_tests/thp/test_multiple_hosts.py b/tests/device_tests/thp/test_multiple_hosts.py index 4e145ecc7bf..d424576ece6 100644 --- a/tests/device_tests/thp/test_multiple_hosts.py +++ b/tests/device_tests/thp/test_multiple_hosts.py @@ -31,10 +31,9 @@ def test_concurrent_handshakes(client: Client) -> None: # The second host starts handshake protocol_2._send_handshake_init_request() - # The second host should not be able to interrupt the first host's handshake - with pytest.raises(exceptions.ThpError) as e: + # The second host should not be able to interrupt the first host's handshake immediately + with pytest.raises(exceptions.TransportBusy): protocol_2._read_ack() - assert e.value.args[0] == "TRANSPORT BUSY" # The first host can complete handshake protocol_1._send_handshake_completion_request() diff --git a/tests/device_tests/thp/test_pairing.py b/tests/device_tests/thp/test_pairing.py index 5885762572f..63186abc598 100644 --- a/tests/device_tests/thp/test_pairing.py +++ b/tests/device_tests/thp/test_pairing.py @@ -406,9 +406,8 @@ def test_credential_phase(client: Client) -> None: protocol._noise.noise_protocol.cipher_state_encrypt.n = 250 protocol._send_message(ButtonAck()) - with pytest.raises(exceptions.ThpError) as e: + with pytest.raises(exceptions.DecryptionFailed): protocol.read(1) - assert e.value.args[0] == "DECRYPTION FAILED" # Connect using credential with confirmation and ask for autoconnect credential. protocol = prepare_protocol_for_handshake(client) @@ -457,9 +456,8 @@ def test_credential_phase(client: Client) -> None: protocol._noise.noise_protocol.cipher_state_encrypt.n = 100 protocol._send_message(ButtonAck()) - with pytest.raises(exceptions.ThpError) as e: + with pytest.raises(exceptions.DecryptionFailed): protocol.read(1) - assert e.value.args[0] == "DECRYPTION FAILED" # Connect using autoconnect credential - should work the same as above protocol = prepare_protocol_for_handshake(client) From 1d77d848530e60fb89739f171e72505840f7fe4a Mon Sep 17 00:00:00 2001 From: Roman Zeyde Date: Thu, 21 Aug 2025 12:40:46 +0300 Subject: [PATCH 2/6] fix(core): correct `ticks_diff` arguments' order [no changelog] --- core/mocks/utime.pyi | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/mocks/utime.pyi b/core/mocks/utime.pyi index c2ff9bb82e2..3bbcae6884a 100644 --- a/core/mocks/utime.pyi +++ b/core/mocks/utime.pyi @@ -5,5 +5,5 @@ def ticks_ms() -> int: ... def ticks_us() -> int: ... def ticks_cpu() -> int: ... def ticks_add(ticks_in: int, delta_in: int) -> int: ... -def ticks_diff(old: int, new: int) -> int: ... +def ticks_diff(new: int, old: int) -> int: ... def gmtime2000(timestamp: int) -> tuple: ... From 6a33e2d2d473462443725a43af9deb1f959f9086 Mon Sep 17 00:00:00 2001 From: Roman Zeyde Date: Thu, 21 Aug 2025 12:41:51 +0300 Subject: [PATCH 3/6] chore(core): exclude logging on non-debug builds [no changelog] --- core/src/trezor/wire/thp/session_context.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/core/src/trezor/wire/thp/session_context.py b/core/src/trezor/wire/thp/session_context.py index 6253a011713..13a13177755 100644 --- a/core/src/trezor/wire/thp/session_context.py +++ b/core/src/trezor/wire/thp/session_context.py @@ -50,7 +50,8 @@ async def handle(self, message: Message | None = None) -> None: if message is None: message = await self._read_next_message() await handle_single_message(self, message) - self.channel._log("session loop is over") + if __debug__: + self.channel._log("session loop is over") return except protocol_common.WireError as e: if __debug__: @@ -71,9 +72,10 @@ async def _read_next_message(self) -> Message: session_id, message = await self.channel.decrypt_message() if session_id == self.session_id: return message - self.channel._log( - "Ignored message for unexpected session", logger=log.warning - ) + if __debug__: + self.channel._log( + "Ignored message for unexpected session", logger=log.warning + ) async def read( self, From 79c9fef1eb9a07a38ead4d8502d175373d72aa27 Mon Sep 17 00:00:00 2001 From: Roman Zeyde Date: Sun, 24 Aug 2025 21:15:16 +0300 Subject: [PATCH 4/6] refactor(core): inline THP `PairingContext` confirmation methods [no changelog] --- core/src/apps/thp/pairing.py | 14 +++++++++++--- core/src/trezor/wire/thp/pairing_context.py | 8 -------- 2 files changed, 11 insertions(+), 11 deletions(-) diff --git a/core/src/apps/thp/pairing.py b/core/src/apps/thp/pairing.py index 61a37bf27f1..53970a55ec1 100644 --- a/core/src/apps/thp/pairing.py +++ b/core/src/apps/thp/pairing.py @@ -35,7 +35,13 @@ SilentError, UnexpectedMessage, ) -from trezor.wire.thp import ChannelState, ThpError, crypto, get_enabled_pairing_methods +from trezor.wire.thp import ( + ChannelState, + ThpError, + crypto, + get_enabled_pairing_methods, + ui, +) from trezor.wire.thp.pairing_context import PairingContext from .credential_manager import is_credential_autoconnect, issue_credential @@ -195,7 +201,7 @@ async def handle_credential_phase( raise DataError("Missing host/app name in credential") if show_connection_dialog and not autoconnect: - await ctx.show_connection_dialog() + await ui.show_connection_dialog(ctx.host_name, ctx.app_name) while ThpCredentialRequest.is_type_of(message): message = await _handle_credential_request(ctx, message) @@ -425,7 +431,9 @@ async def _handle_credential_request( "Cannot ask for autoconnect credential without a valid credential!" ) - await ctx.show_autoconnect_credential_confirmation_screen() # TODO add device name + await ui.show_autoconnect_credential_confirmation_screen( + host_name=ctx.host_name, app_name=ctx.app_name + ) trezor_static_public_key = crypto.get_trezor_static_public_key() credential_metadata = ThpCredentialMetadata( diff --git a/core/src/trezor/wire/thp/pairing_context.py b/core/src/trezor/wire/thp/pairing_context.py index 8520f08c2c5..7d97b935b99 100644 --- a/core/src/trezor/wire/thp/pairing_context.py +++ b/core/src/trezor/wire/thp/pairing_context.py @@ -143,14 +143,6 @@ async def show_pairing_dialog(self) -> None: action=action_string, ) - async def show_connection_dialog(self) -> None: - await ui.show_connection_dialog(self.host_name, self.app_name) - - async def show_autoconnect_credential_confirmation_screen(self) -> None: - await ui.show_autoconnect_credential_confirmation_screen( - self.host_name, self.app_name - ) - async def show_pairing_method_screen( self, selected_method: ThpPairingMethod | None = None ) -> UiResult: From c9ec733768e2bc2397cb5b185929d61351f0deb2 Mon Sep 17 00:00:00 2001 From: Roman Zeyde Date: Sun, 24 Aug 2025 21:36:30 +0300 Subject: [PATCH 5/6] chore(core): remove `thp.pairing._skip_pairing_dialog()` It is currently unused, since `should_show_pairing_dialog` is always `True`. [no changelog] --- core/src/apps/thp/pairing.py | 24 ++---------------------- core/src/trezor/wire/thp/channel.py | 3 --- 2 files changed, 2 insertions(+), 25 deletions(-) diff --git a/core/src/apps/thp/pairing.py b/core/src/apps/thp/pairing.py index 53970a55ec1..8c5dfbd7d8c 100644 --- a/core/src/apps/thp/pairing.py +++ b/core/src/apps/thp/pairing.py @@ -118,11 +118,8 @@ async def handle_pairing_request( ctx.host_name = message.host_name ctx.app_name = message.app_name - if __debug__ and not ctx.channel_ctx.should_show_pairing_dialog: - await _skip_pairing_dialog(ctx) - else: - await ctx.show_pairing_dialog() - await ctx.write(ThpPairingRequestApproved()) + await ctx.show_pairing_dialog() + await ctx.write(ThpPairingRequestApproved()) assert ThpSelectMethod.MESSAGE_WIRE_TYPE is not None select_method_msg = await ctx.read( [ @@ -484,20 +481,3 @@ def _check_method_is_allowed(ctx: PairingContext, method: ThpPairingMethod) -> N def _check_method_is_selected(ctx: PairingContext, method: ThpPairingMethod) -> None: if method is not ctx.selected_method: raise ThpError("Not selected pairing method") - - -if __debug__: - - async def _skip_pairing_dialog(ctx: PairingContext) -> None: - from trezor.enums import ButtonRequestType - from trezor.messages import ButtonAck, ButtonRequest, ThpPairingRequestApproved - from trezor.wire.errors import ActionCancelled - - resp = await ctx.call( - ButtonRequest(code=ButtonRequestType.Other, name="thp_pairing_request"), - expected_type=ButtonAck, - ) - if isinstance(resp, ButtonAck): - await ctx.write(ThpPairingRequestApproved()) - else: - raise ActionCancelled diff --git a/core/src/trezor/wire/thp/channel.py b/core/src/trezor/wire/thp/channel.py index d6ce8a28813..4ac08a87a41 100644 --- a/core/src/trezor/wire/thp/channel.py +++ b/core/src/trezor/wire/thp/channel.py @@ -158,9 +158,6 @@ def __init__( self.credential: ThpPairingCredential | None = None self.connection_context: PairingContext | None = None - if __debug__: - self.should_show_pairing_dialog: bool = True - @property def iface(self) -> WireInterface: return self.ctx._iface From 0900245614740fb706afb37a190f299cdabc0442 Mon Sep 17 00:00:00 2001 From: Roman Zeyde Date: Mon, 25 Aug 2025 12:26:24 +0300 Subject: [PATCH 6/6] chore(core): move `show_pairing_dialog()` to `trezor.wire.thp.ui` module [no changelog] --- core/src/apps/thp/pairing.py | 2 +- core/src/trezor/wire/thp/pairing_context.py | 11 ----------- core/src/trezor/wire/thp/ui.py | 12 ++++++++++++ 3 files changed, 13 insertions(+), 12 deletions(-) diff --git a/core/src/apps/thp/pairing.py b/core/src/apps/thp/pairing.py index 8c5dfbd7d8c..af1c7d327e4 100644 --- a/core/src/apps/thp/pairing.py +++ b/core/src/apps/thp/pairing.py @@ -118,7 +118,7 @@ async def handle_pairing_request( ctx.host_name = message.host_name ctx.app_name = message.app_name - await ctx.show_pairing_dialog() + await ui.show_pairing_dialog(ctx.host_name, ctx.app_name) await ctx.write(ThpPairingRequestApproved()) assert ThpSelectMethod.MESSAGE_WIRE_TYPE is not None select_method_msg = await ctx.read( diff --git a/core/src/trezor/wire/thp/pairing_context.py b/core/src/trezor/wire/thp/pairing_context.py index 7d97b935b99..bb3476d93a3 100644 --- a/core/src/trezor/wire/thp/pairing_context.py +++ b/core/src/trezor/wire/thp/pairing_context.py @@ -132,17 +132,6 @@ def set_selected_method(self, selected_method: ThpPairingMethod) -> None: raise DataError("Selected pairing method is not supported") self.selected_method = selected_method - async def show_pairing_dialog(self) -> None: - from trezor.ui.layouts import confirm_action - - subject = ui._app_on_host(self.app_name, self.host_name) - action_string = f"Allow {subject} to pair with this Trezor?" - await confirm_action( - br_name="thp_pairing_request", - title="Before you continue", - action=action_string, - ) - async def show_pairing_method_screen( self, selected_method: ThpPairingMethod | None = None ) -> UiResult: diff --git a/core/src/trezor/wire/thp/ui.py b/core/src/trezor/wire/thp/ui.py index 31a93c32402..47f41ce7743 100644 --- a/core/src/trezor/wire/thp/ui.py +++ b/core/src/trezor/wire/thp/ui.py @@ -29,6 +29,18 @@ async def show_autoconnect_credential_confirmation_screen( ) +async def show_pairing_dialog(host_name: str | None, app_name: str | None) -> None: + from trezor.ui.layouts import confirm_action + + subject = _app_on_host(app_name, host_name) + action_string = f"Allow {subject} to pair with this Trezor?" + await confirm_action( + br_name="thp_pairing_request", + title="Before you continue", + action=action_string, + ) + + async def show_connection_dialog(host_name: str | None, app_name: str | None) -> None: from trezor.ui.layouts import confirm_action