feat(python): skip THP ACKs on `TrezorClientThp._call()`
What the developer wrote
Authored by Roman Zeyde
feat(python): skip THP ACKs on `TrezorClientThp._call()`
Due to event loop restart, the last response from the device will be explicitly ACKed.
It will allow piggyback `ButtonRequest` THP ACKs using corresponding `ButtonAck` messages.
What changed, and why it matters
This commit changes how the Trezor Python library sends low-level acknowledgment (ACK) packets during a newer USB protocol called THP. Instead of always sending a standalone ACK, it lets the next normal message (for example a button confirmation) carry the ACK 'for free.' This is described as a performance/correctness optimization, not a security fix. There is no direct evidence in the commit that it repairs a vulnerability.
Treat as a normal feature/optimization commit. Review THP protocol state-machine invariants to ensure suppressed ACKs cannot cause replay, desynchronization, or denial-of-service during error paths; verify the fallback ACK in `_send_ack(None)` uses the correct sync bit and channel id. No immediate security response is indicated by the available evidence.
Security signals we found
Protocol-layer ACK behavior change in THP transport
New context manager suppresses explicit ACKs during a workflow
ACK generation now falls back to current sync bit when no prior message is supplied
No mention of vulnerability, CVE, bug bounty, or security advisory in commit or title
Evidence from the diff
The patch introduces a context manager piggyback_acks() in channel.py and overrides _call() in TrezorClientThp to wrap every call with it. While active, explicit ACKs are suppressed; on exit, a final explicit ACK is sent for the last received message. _send_ack() is modified to build an ACK from the last received message or, when piggybacking, from the current receive sync bit and channel id. The stated reason is to allow piggybacking ButtonAck onto ButtonRequest ACKs and because the device event loop restart already ACKs the last response.
Changed components
python/src/trezorlib/thp/channel.pypython/src/trezorlib/thp/client.pyTrezor Python client library THP transportInspect captured patch +43 / −4
diff --git a/python/src/trezorlib/thp/channel.py b/python/src/trezorlib/thp/channel.py
index 2d0317ac..00ef7e50 100644
--- a/python/src/trezorlib/thp/channel.py
+++ b/python/src/trezorlib/thp/channel.py
@@ -300,7 +300,8 @@ class Channel:
if e.code == exceptions.ThpErrorCode.DEVICE_LOCKED:
raise DeviceLockedError from e
raise
- self._send_ack(message)
+ if not self.is_ack_piggybacking_allowed:
+ self._send_ack(message)
if not message.is_handshake_init_response():
raise ProtocolError(f"Not a valid handshake init response: {message}")
@@ -408,9 +409,17 @@ class Channel:
continue
raise
- def _send_ack(self, acked_message: Message) -> None:
- ack = control_byte.make_ack_for(acked_message.ctrl_byte)
- ack_message = Message(ack, acked_message.cid, b"")
+ def _send_ack(self, acked_message: Message | None) -> None:
+ if self.is_ack_piggybacking_allowed and self._active_workflow is not None:
+ return
+
+ if acked_message is not None:
+ ack = control_byte.make_ack_for(acked_message.ctrl_byte)
+ ack_message = Message(ack, acked_message.cid, b"")
+ else:
+ ack = control_byte.make_ack(not self.sync_bit_receive)
+ ack_message = Message(ack, self.channel_id, b"")
+
thp_io.write_payload_to_wire(self.transport, ack_message)
def _read_ack(self, message: Message) -> None:
@@ -432,6 +441,23 @@ class Channel:
f"Failed to read ACK in {retries} retries for message: {message}"
)
+ @contextmanager
+ def piggyback_acks(self, marker: object) -> t.Generator[None, None, None]:
+ # Make sure the previous workflow is over.
+ assert self._active_workflow is None
+ self._active_workflow = marker
+ # Skip explicit ACKs during this workflow
+ try:
+ yield
+ finally:
+ active = self._active_workflow
+ self._active_workflow = None
+ assert active is marker
+ if self.is_ack_piggybacking_allowed:
+ # Explicitly ACK the latest received message. The device may restart
+ # the event loop, so the next request will be sent in a separate message.
+ self._send_ack(None)
+
def write_chunk(self, data: bytes, /) -> None:
self._assert_handshake_done()
encrypted_data = self.noise.encrypt(data)
diff --git a/python/src/trezorlib/thp/client.py b/python/src/trezorlib/thp/client.py
index 8c240223..778b0e60 100644
--- a/python/src/trezorlib/thp/client.py
+++ b/python/src/trezorlib/thp/client.py
@@ -188,6 +188,19 @@ class TrezorClientThp(client.TrezorClient[ThpSession]):
else:
self._session_message_queue[session_id].append(msg)
+ def _call(
+ self,
+ session: ThpSession,
+ msg: client.MessageType,
+ *,
+ expect: type[client.MT] = client.MessageType,
+ timeout: float | None = None,
+ ) -> client.MT:
+ with self.channel.piggyback_acks(msg):
+ return super()._call(
+ session=session, msg=msg, expect=expect, timeout=timeout
+ )
+
@staticmethod
def detect_model(props: messages.ThpDeviceProperties) -> models.TrezorModel:
internal_model = props.internal_model
Why this scored 20/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.