fix(python): support receiving piggybacked ACKs
What the developer wrote
Authored by Martin Milata
fix(python): support receiving piggybacked ACKs
[no changelog]
What changed, and why it matters
This commit fixes the Trezor Python host library so it correctly handles a protocol optimization called 'piggybacked ACKs.' In the Trezor THP (Trezor Host Protocol), a data message can also carry an acknowledgment (ACK) bit, meaning one side can confirm it received the previous message while simultaneously sending new data. Previously, the Python library apparently treated such combined messages as invalid ACKs and would keep waiting or retry, which could cause communication failures or timeouts. The change lets the library accept the ACK part of such a message, save the data part for the next read, and continue normally.
Treat as a normal bugfix. Review whether the previous behavior could cause denial-of-service or state desynchronization during device communication, but there is no direct evidence of exploitable vulnerability. Update the Python library if using THP piggybacked ACKs.
Security signals we found
Protocol state machine fix in host-side transport layer
Changes ACK validation to accept piggybacked ACKs on data messages
Adds queueing of next message to avoid dropping piggybacked payload
No changelog entry suggests routine bugfix, not advertised security fix
No explicit security keywords in commit title or message
Evidence from the diff
The patch modifies python/src/trezorlib/thp/channel.py and python/src/trezorlib/thp/control_byte.py. It introduces _next_message to store a data message whose ACK bit was consumed as an ACK. _read_ack() now distinguishes three cases: a clean standalone ACK, an ACK with unexpected payload (still treated as invalid), and a data message with the correct ACK bit when piggybacking is allowed (the ACK is accepted and the data message is queued). _read() returns the queued message before reading from the transport. control_byte.py is updated so get_ack_bit() returns a value for both ACK and DATA messages, because data messages can carry an ACK bit. This is a protocol correctness fix rather than a memory-safety or cryptographic bug.
Changed components
python/src/trezorlib/thp/channel.pypython/src/trezorlib/thp/control_byte.pyTrezor Python host library THP transport implementationInspect captured patch +21 / −3
diff --git a/python/src/trezorlib/thp/channel.py b/python/src/trezorlib/thp/channel.py
index 48e883f4..b5000025 100644
--- a/python/src/trezorlib/thp/channel.py
+++ b/python/src/trezorlib/thp/channel.py
@@ -172,6 +172,8 @@ class Channel:
self.state = channel_state
self.trezor_public_keys: TrezorPublicKeys | None = None
self._active_contexts: list[AbstractContextManager] = []
+ # Message processed as ACK but not yet its content (piggybacking only).
+ self._next_message: Message | None = None
@functools.cached_property
def is_ack_piggybacking_allowed(self) -> bool:
@@ -456,16 +458,27 @@ class Channel:
def _read_ack(self, message: Message) -> None:
expected_seq_bit = message.seq_bit
+ assert expected_seq_bit is not None
retries = MAX_RETRANSMISSION_COUNT
time_start = time.monotonic()
for _ in range(1 + retries):
time_elapsed = time.monotonic() - time_start
message = self._read(timeout=ACK_TIMEOUT - time_elapsed)
- if not message.is_ack() or len(message.data) > 0:
+ ack_bit_ok = message.ack_bit == expected_seq_bit
+ if message.is_ack() and len(message.data) == 0:
+ pass # standalone ACK message
+ elif message.is_ack():
+ LOG.warning("Received ACK with non-empty data: %s", message)
+ continue
+ elif self.is_ack_piggybacking_allowed and ack_bit_ok:
+ assert self._next_message is None
+ # process ACK bit, return message in next _read
+ self._next_message = message
+ else:
LOG.error("Received message is not a valid ACK: %s", message)
# data messages and their acks should have been handled by _read()
continue
- if message.ack_bit != expected_seq_bit:
+ if not ack_bit_ok:
LOG.warning("Received ACK with unexpected sequence bit: %s", message)
continue
return
@@ -501,6 +514,11 @@ class Channel:
if timeout is None:
timeout = client._DEFAULT_READ_TIMEOUT
+ if self._next_message:
+ message = self._next_message
+ self._next_message = None
+ return message
+
while True:
message = thp_io.read(self.transport, timeout)
if message.seq_bit is not None:
diff --git a/python/src/trezorlib/thp/control_byte.py b/python/src/trezorlib/thp/control_byte.py
index b560c4dd..e5fa87f0 100644
--- a/python/src/trezorlib/thp/control_byte.py
+++ b/python/src/trezorlib/thp/control_byte.py
@@ -107,7 +107,7 @@ def get_seq_bit(ctrl_byte: int) -> bool | None:
def get_ack_bit(ctrl_byte: int) -> bool | None:
- if not is_ack(ctrl_byte):
+ if not is_ack(ctrl_byte) and not is_data(ctrl_byte):
return None
return bool(ctrl_byte & ACK_SEQ_BIT)
Why this scored 32/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.