fix(core): make sure to increment THP `seq_bit`
What the developer wrote
Authored by Roman Zeyde
fix(core): make sure to increment THP `seq_bit`
Make it compatible with Suite behaviour in case the last THP ACK is lost.
What changed, and why it matters
This commit fixes a bug in the Trezor hardware wallet's THP (Trezor Host Protocol) communication code. Previously, if the final acknowledgment (ACK) from the computer/app was lost, the device would not properly advance its internal sequence bit. This could cause the next message to use the wrong sequence number, potentially confusing the host software (Trezor Suite) and disrupting communication. The fix ensures the sequence bit is always advanced after a write attempt, even if the ACK is lost or the write is blocked.
Treat as a low-to-moderate reliability/protocol-integrity fix. Review whether the changed 'write blocked' behavior (now raising Timeout instead of breaking the loop) affects any error-handling assumptions in host software. No immediate exploit mitigation is evident, but firmware updates should include this fix to prevent THP state desync.
Security signals we found
Protocol state desynchronization between device and host after lost final ACK
Alternating-bit sequence number not advanced on error/timeout paths
Potential communication stall or confusion with Trezor Suite on reconnection
Fix explicitly references compatibility with Suite behavior for lost ACKs
Evidence from the diff
In core/src/trezor/wire/thp/channel.py, write_encrypted_payload() previously only flipped the alternating-bit-protocol (ABP) send_seq_bit when a valid ACK was received (ABP.set_send_seq_bit_to_opposite() inside the success path). If the last ACK was lost, all retransmissions failed, or the USB peer stopped reading, the function raised Timeout without flipping the bit. The patch wraps the send/ACK loop in a local _write_loop() and moves the sequence-bit flip into a finally block, guaranteeing it runs on every exit path. A new helper _write_payload_once() also changes the ‘write blocked’ case from silently breaking the loop to raising Timeout. Tests confirm the sequence bit advances after both total timeout and blocked-write scenarios.
Changed components
core/src/trezor/wire/thp/channel.pyTrezor Host Protocol (THP) writerABP (alternating bit protocol) sequence-bit stateTrezor Suite host communication compatibilityInspect captured patch +92 / −35
diff --git a/core/.changelog.d/6138.fixed b/core/.changelog.d/6138.fixed
new file mode 100644
index 000000000..b28eee33e
--- /dev/null
+++ b/core/.changelog.d/6138.fixed
@@ -0,0 +1 @@
+Make sure to increment THP `seq_bit`.
diff --git a/core/src/trezor/wire/thp/channel.py b/core/src/trezor/wire/thp/channel.py
index 52d02e9ce..dbc2d0172 100644
--- a/core/src/trezor/wire/thp/channel.py
+++ b/core/src/trezor/wire/thp/channel.py
@@ -408,54 +408,70 @@ class Channel:
return await self.write_encrypted_payload(ENCRYPTED, buffer[:payload_length])
async def write_encrypted_payload(self, ctrl_byte: int, payload: AnyBytes) -> None:
- ack_latency_ms = self.channel_cache.get_int(CHANNEL_ACK_LATENCY_MS) or 0
- if __debug__:
- self._log(f"Sending {len(payload)} bytes, latency: {ack_latency_ms} ms")
-
assert ABP.is_sending_allowed(self.channel_cache)
+ # Construct THP header
payload_len = len(payload) + CHECKSUM_LENGTH
sync_bit = ABP.get_send_seq_bit(self.channel_cache)
ctrl_byte = control_byte.add_seq_bit_to_ctrl_byte(ctrl_byte, sync_bit)
header = PacketHeader(ctrl_byte, self.get_channel_id_int(), payload_len)
- # ACK is needed before sending more data
- ABP.set_sending_allowed(self.channel_cache, False)
+ async def _write_loop() -> None:
+ """Send the payload and wait for an ACK with retransmissions."""
- # allows preempting this channel, if another channel becomes active
- self.last_write_ms = utime.ticks_ms()
+ ack_latency_ms = self.channel_cache.get_int(CHANNEL_ACK_LATENCY_MS) or 0
+ if __debug__:
+ self._log(f"Sending {len(payload)} bytes, latency: {ack_latency_ms} ms")
- for i in range(_MAX_RETRANSMISSION_COUNT):
- result = await race(
- self.iface_ctx.write_payload(header, payload), _WRITE_TIMEOUT
- )
- if isinstance(result, int):
- if __debug__:
- log.error(__name__, "Sending is stuck for %d ms", _WRITE_TIMEOUT_MS)
- break
+ # ACK is needed before sending more data
+ ABP.set_sending_allowed(self.channel_cache, False)
- # Channel's estimated latency + a variable delay (from 200ms till ~3.52s)
- timeout_ms = ack_latency_ms + round(10300 - 1010000 / (100 + i))
- try:
- # wait and return after receiving an ACK, or raise in case of an unexpected message.
- await self.recv_payload(expected_ctrl_byte=None, timeout_ms=timeout_ms)
- except Timeout:
- if __debug__:
- log.warning(__name__, "Retransmit after %d ms", timeout_ms)
- continue
+ # allows preempting this channel, if another channel becomes active
+ self.last_write_ms = utime.ticks_ms()
- ack_latency_ms = utime.ticks_diff(utime.ticks_ms(), self.last_write_ms)
- # Limit estimated latency to avoid integer overflows and too long delays
- ack_latency_ms = max(0, min(800, ack_latency_ms))
- self.channel_cache.set_int(CHANNEL_ACK_LATENCY_MS, ack_latency_ms)
+ for i in range(_MAX_RETRANSMISSION_COUNT):
+ await self._write_payload_once(header, payload)
- # `ABP.set_sending_allowed()` will be called after a valid ACK
- if ABP.is_sending_allowed(self.channel_cache):
- ABP.set_send_seq_bit_to_opposite(self.channel_cache)
- return
+ # Channel's estimated latency + a variable delay (from 200ms till ~3.52s)
+ timeout_ms = ack_latency_ms + round(10300 - 1010000 / (100 + i))
+ try:
+ # wait and return after receiving an ACK, or raise in case of an unexpected message.
+ await self.recv_payload(
+ expected_ctrl_byte=None, timeout_ms=timeout_ms
+ )
+ except Timeout:
+ if __debug__:
+ log.warning(__name__, "Retransmit after %d ms", timeout_ms)
+ continue
+
+ ack_latency_ms = utime.ticks_diff(utime.ticks_ms(), self.last_write_ms)
+ # Limit estimated latency to avoid integer overflows and too long delays
+ ack_latency_ms = max(0, min(800, ack_latency_ms))
+ self.channel_cache.set_int(CHANNEL_ACK_LATENCY_MS, ack_latency_ms)
+
+ # `ABP.set_sending_allowed()` will be called after a valid ACK
+ if ABP.is_sending_allowed(self.channel_cache):
+ return
- # restart event loop due to unresponsive channel
- raise Timeout("THP retransmission timeout")
+ # restart event loop due to unresponsive channel
+ raise Timeout("THP retransmission timeout")
+
+ try:
+ return await _write_loop()
+ finally:
+ # Make sure to use the next `seq_bit` for the next payload
+ ABP.set_send_seq_bit_to_opposite(self.channel_cache)
+
+ async def _write_payload_once(
+ self, header: PacketHeader, payload: AnyBytes
+ ) -> None:
+ """Write the payload and raise if the interface is blocked."""
+ result = await race(
+ self.iface_ctx.write_payload(header, payload), _WRITE_TIMEOUT
+ )
+ if isinstance(result, int):
+ # Can happen when the USB peer is not reading.
+ raise Timeout("THP write is blocked")
def _encrypt(self, buffer: AnyBuffer, noise_payload_len: int) -> None:
if __debug__:
diff --git a/core/tests/test_trezor.wire.thp.writer.py b/core/tests/test_trezor.wire.thp.writer.py
index b26159bc0..dc9824eed 100644
--- a/core/tests/test_trezor.wire.thp.writer.py
+++ b/core/tests/test_trezor.wire.thp.writer.py
@@ -6,7 +6,10 @@ from typing import Any, Awaitable
if utils.USE_THP:
import thp_common
from mock_wire_interface import MockHID
+ from trezor.loop import Timeout
from trezor.wire.thp import ENCRYPTED, PacketHeader
+ from trezor.wire.thp import alternating_bit_protocol as ABP
+ from trezor.wire.thp.channel import _MAX_RETRANSMISSION_COUNT
from trezor.wire.thp.interface_context import ThpContext
@@ -131,6 +134,43 @@ class TestTrezorHostProtocolWriter(unittest.TestCase):
self.longer_payload_with_checksum_expected[i],
)
+ def test_write_timeout(self):
+ channel = thp_common.get_new_channel(self.interface)
+ seq_bit = ABP.get_send_seq_bit(channel.channel_cache)
+
+ task = channel.write_encrypted_payload(ENCRYPTED, b"PAYLOAD")
+ task.send(None) # start the generator
+
+ for _ in range(_MAX_RETRANSMISSION_COUNT - 1):
+ task.send(None) # complete write
+ task.throw(Timeout()) # no ACK is received
+
+ task.send(None) # complete write last time
+ with self.assertRaises(Timeout):
+ task.throw(Timeout()) # no ACK is received
+
+ # next write should use the next `seq_bit` (see #6138)
+ self.assertNotEqual(ABP.get_send_seq_bit(channel.channel_cache), seq_bit)
+
+ def test_write_blocked(self):
+ channel = thp_common.get_new_channel(self.interface)
+ seq_bit = ABP.get_send_seq_bit(channel.channel_cache)
+
+ task = channel.write_encrypted_payload(ENCRYPTED, b"PAYLOAD")
+ task.send(None) # start the generator
+
+ # Re-transmit a few times
+ for _ in range(3):
+ task.send(None) # complete write
+ task.throw(Timeout()) # no ACK is received
+
+ with self.assertRaises(Timeout):
+ # timeout write (as if `loop.sleep` has completed) using dummy "ticks" integer value
+ task.send(12345)
+
+ # next write should use the next `seq_bit` (see #6138)
+ self.assertNotEqual(ABP.get_send_seq_bit(channel.channel_cache), seq_bit)
+
if __name__ == "__main__":
unittest.main()
Why this scored 42/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.