refactor(python): simplify THP pairing state handling
What changed, and why it matters
This is a straightforward internal code cleanup in Trezor's Python library. It replaces two separate flags that tracked whether a device was paired with a single simpler flag, and adds extra debug logging. There is no indication this fixes or introduces a security vulnerability.
No security action required. Treat as normal refactoring.
Security signals we found
No strong security signals were identified.
Evidence from the diff
The commit refactors THP (Trezor Host Protocol) pairing state handling in trezorlib. It removes the TrezorState enum and consolidates _has_valid_channel and trezor_state into a single _is_paired boolean. The handshake completion response is now used to set _is_paired directly, and several debug log statements were added. Test files were updated to use the new flag name. No functional security change is evident from the diff.
Changed components
python/src/trezorlib/client.pypython/src/trezorlib/transport/thp/protocol_v2.pytests/device_tests/thp/test_pairing.pyInspect captured patch +22 / −29
diff --git a/python/src/trezorlib/client.py b/python/src/trezorlib/client.py
index 85158da9..d45d1679 100644
--- a/python/src/trezorlib/client.py
+++ b/python/src/trezorlib/client.py
@@ -29,7 +29,7 @@ from .transport import Transport, get_transport
from .transport.thp.channel import Channel
from .transport.thp.cpace import Cpace
from .transport.thp.protocol_v1 import ProtocolV1Channel, UnexpectedMagicError
-from .transport.thp.protocol_v2 import ProtocolV2Channel, TrezorState
+from .transport.thp.protocol_v2 import ProtocolV2Channel
if t.TYPE_CHECKING:
from .transport.session import Session, SessionV1, SessionV2
@@ -120,6 +120,7 @@ class TrezorClient:
raise RuntimeError(
"Connected Trezor does not support any trezorlib-compatible pairing method."
)
+ LOG.debug("Starting pairing: %r", pairing_method)
session = SessionV2.seedless(self)
session.call(
messages.ThpPairingRequest(host_name="Trezorlib"),
@@ -142,7 +143,7 @@ class TrezorClient:
skip_firmware_version_check=True,
)
assert isinstance(self.protocol, ProtocolV2Channel)
- self.protocol._has_valid_channel = True
+ self.protocol._is_paired = True
def _handle_code_entry(self, session: SessionV2) -> None:
from .cli import get_code_entry_code
@@ -209,7 +210,7 @@ class TrezorClient:
)
assert isinstance(self.protocol, ProtocolV2Channel)
- self.protocol._has_valid_channel = True
+ self.protocol._is_paired = True
def get_session(
self,
@@ -240,7 +241,7 @@ class TrezorClient:
if isinstance(self.protocol, ProtocolV2Channel):
from .transport.session import SessionV2
- if self.protocol.trezor_state is TrezorState.UNPAIRED:
+ if not self.protocol._is_paired:
self.do_pairing()
if passphrase is SEEDLESS:
@@ -273,10 +274,7 @@ class TrezorClient:
def _get_features(self) -> messages.Features:
if isinstance(self.protocol, ProtocolV2Channel):
- if (
- self.protocol.trezor_state is TrezorState.UNPAIRED
- or not self.protocol._has_valid_channel
- ):
+ if not self.protocol._is_paired:
self.do_pairing()
return self.protocol.get_features()
diff --git a/python/src/trezorlib/transport/thp/protocol_v2.py b/python/src/trezorlib/transport/thp/protocol_v2.py
index 7a56a578..2d85b6c6 100644
--- a/python/src/trezorlib/transport/thp/protocol_v2.py
+++ b/python/src/trezorlib/transport/thp/protocol_v2.py
@@ -20,7 +20,6 @@ import logging
import os
import typing as t
from binascii import hexlify
-from enum import IntEnum
from noise.connection import Keypair, NoiseConnection
@@ -44,11 +43,6 @@ if t.TYPE_CHECKING:
MT = t.TypeVar("MT", bound=protobuf.MessageType)
-class TrezorState(IntEnum):
- UNPAIRED = 0x00
- PAIRED = 0x01
-
-
class ProtocolV2Channel(Channel):
channel_id: int
sync_bit_send: int
@@ -56,9 +50,8 @@ class ProtocolV2Channel(Channel):
handshake_hash: bytes
device_properties: bytes
- _has_valid_channel: bool = False
_features: messages.Features | None = None
- trezor_state: int = TrezorState.UNPAIRED
+ _is_paired: bool = False
def __init__(
self,
@@ -72,11 +65,13 @@ class ProtocolV2Channel(Channel):
if prepare_channel_without_pairing:
# allow skipping unrelated response packets (e.g. in case of retransmissions)
self._do_channel_allocation(retries=MAX_RETRANSMISSION_COUNT)
- self.trezor_state = self._do_handshake(credential=credential)
+ LOG.debug("THP channel allocated: %04x", self.channel_id)
+ self._do_handshake(credential=credential)
+ LOG.debug("THP handshake done: is_paired=%s", self._is_paired)
def get_channel(self) -> ProtocolV2Channel:
- if not self._has_valid_channel:
- raise RuntimeError("Channel is invalidated")
+ if not self._is_paired:
+ raise RuntimeError("Channel is not paired")
return self
def read(self, session_id: int, timeout: float | None = None) -> t.Any:
@@ -92,8 +87,8 @@ class ProtocolV2Channel(Channel):
self._encrypt_and_write(session_id, msg_type, msg_data)
def get_features(self) -> messages.Features:
- if not self._has_valid_channel:
- raise RuntimeError("Channel is invalidated")
+ if not self._is_paired:
+ raise RuntimeError("Channel is not paired")
if self._features is None:
self.update_features()
assert self._features is not None
@@ -205,7 +200,7 @@ class ProtocolV2Channel(Channel):
credential: bytes | None = None,
host_static_randomness: bytes | None = None,
host_ephemeral_randomness: bytes | None = None,
- ) -> int:
+ ) -> None:
randomness_static = host_static_randomness or os.urandom(32)
if host_ephemeral_randomness is not None:
@@ -268,15 +263,15 @@ class ProtocolV2Channel(Channel):
)
self.handshake_hash = self._noise.get_handshake_hash()
- def _read_handshake_completion_response(self) -> int:
- # Read handshake completion response, ignore payload as we do not care about the state
+ def _read_handshake_completion_response(self) -> None:
+ # Read handshake completion response
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")
trezor_state = self._noise.decrypt(bytes(data))
assert trezor_state == b"\x00" or trezor_state == b"\x01"
self._send_ack_bit(bit=1)
- return int.from_bytes(trezor_state, "big")
+ self._is_paired = bool(int.from_bytes(trezor_state, "big"))
def _read_ack(self):
header, payload = self._read_until_valid_crc_check()
diff --git a/tests/device_tests/thp/test_pairing.py b/tests/device_tests/thp/test_pairing.py
index 63186abc..12b35eec 100644
--- a/tests/device_tests/thp/test_pairing.py
+++ b/tests/device_tests/thp/test_pairing.py
@@ -101,7 +101,7 @@ def test_pairing_qr_code(client: Client) -> None:
protocol._send_message(ThpEndRequest())
protocol._read_message(ThpEndResponse)
- protocol._has_valid_channel = True
+ protocol._is_paired = True
@pytest.mark.filterwarnings(
@@ -172,7 +172,7 @@ def test_pairing_code_entry(
protocol._send_message(ThpEndRequest())
protocol._read_message(ThpEndResponse)
- protocol._has_valid_channel = True
+ protocol._is_paired = True
@pytest.mark.filterwarnings(
@@ -241,7 +241,7 @@ def test_pairing_nfc(client: Client) -> None:
protocol._send_message(ThpEndRequest())
protocol._read_message(ThpEndResponse)
- protocol._has_valid_channel = True
+ protocol._is_paired = True
def _nfc_pairing(client: Client, protocol: ProtocolV2Channel) -> None:
@@ -486,9 +486,9 @@ def test_credential_request_in_encrypted_transport_phase(client: Client) -> None
credential = credential_response.credential
protocol._send_message(ThpEndRequest())
protocol._read_message(ThpEndResponse)
+ protocol._is_paired = True # pairing has been done above
session = client.get_seedless_session()
-
session.call(
ThpCredentialRequest(
host_static_public_key=host_static_public_key,
Why this scored 15/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.