fix(python): skip unrelated responses when probing transport
What the developer wrote
Authored by Roman Zeyde
fix(python): skip unrelated responses when probing transport
Also, simplify a bit protocol v1 magic handling flow.
Fixes https://github.com/trezor/trezor-firmware/issues/6588.
What changed, and why it matters
This commit fixes a bug in the Python Trezor client library that could cause it to crash or fail to connect when probing a Trezor device. The problem happened when leftover or unrelated data packets were still sitting in the communication channel. Previously, the library would see an unexpected packet, treat it as a protocol error, and give up. Now it skips those unrelated packets and waits for the real response. This is a reliability and minor security improvement for the host-side software, not the hardware wallet firmware itself.
Update the Python trezorlib package to include this fix. If you maintain integrations that probe Trezor devices over USB/UDP/bridge transports, ensure stale packets are handled gracefully. Review whether silently skipping mismagic chunks is acceptable for your threat model, or whether stricter logging/limits are warranted.
Security signals we found
Host-side library could be disrupted by stale or injected packets on the transport during device probing
Unexpected packets previously caused a ProtocolError, potentially enabling denial-of-service against the client connection
Patch adds an ignore-bad-magic mode that silently consumes mismagic chunks, which could also mask malformed or suspicious traffic
No firmware or device-side code is changed; impact is limited to the Python trezorlib client
Evidence from the diff
The patch modifies python/src/trezorlib/protocol_v1.py. It refactors the protocol-v1 chunk reader so the first chunk and continuation chunks use explicit magic-byte expectations (FIRST_MAGIC/NEXT_MAGIC) instead of a boolean flag. It adds an _ignore_bad_magic parameter to read(). When True, chunks whose prefix does not match the expected magic are silently dropped in a loop rather than raising ProtocolError. The probe() function now calls read(transport, _ignore_bad_magic=True) after sending a Cancel message, so stale/unexpected packets left on the transport are discarded while waiting for the Cancel response. The changelog fragment says ‘Skip unrelated device response when probing transport.’
Changed components
python/src/trezorlib/protocol_v1.pytrezorlib transport probing logictrezorlib protocol v1 message readerInspect captured patch +22 / −18
diff --git a/python/.changelog.d/6588.fixed b/python/.changelog.d/6588.fixed
new file mode 100644
index 00000000..548ad393
--- /dev/null
+++ b/python/.changelog.d/6588.fixed
@@ -0,0 +1 @@
+Skip unrelated device response when probing transport.
diff --git a/python/src/trezorlib/protocol_v1.py b/python/src/trezorlib/protocol_v1.py
index 5492e4d9..b733f2d4 100644
--- a/python/src/trezorlib/protocol_v1.py
+++ b/python/src/trezorlib/protocol_v1.py
@@ -59,30 +59,32 @@ def write(transport: Transport, message_type: int, message_data: bytes) -> None:
transport.write_chunk(chunk)
-def read(transport: Transport, timeout: float | None = None) -> tuple[int, bytes]:
+def read(
+ transport: Transport, timeout: float | None = None, _ignore_bad_magic: bool = False
+) -> tuple[int, bytes]:
"""Read out and reassemble protocol-v1 chunked message from transport."""
if timeout is None:
timeout = client._DEFAULT_READ_TIMEOUT
# Chunked transports prefix the first packet with "?##" and all following packets with "?".
# Non-chunked (i.e., bridge) just load all the data in one go.
- use_chunk_magic = transport.CHUNK_SIZE is not None
-
- def read_next_chunk() -> bytes:
- chunk = transport.read_chunk(timeout=timeout)
- if use_chunk_magic and chunk[:1] != b"?":
- raise exceptions.ProtocolError(f"Missing chunk magic: {chunk.hex()}")
- return chunk[1:]
+ if transport.CHUNK_SIZE is not None:
+ FIRST_MAGIC = b"?##"
+ NEXT_MAGIC = b"?"
+ else:
+ FIRST_MAGIC = NEXT_MAGIC = b""
+
+ def read_next_chunk(magic: bytes) -> bytes:
+ while True:
+ chunk = transport.read_chunk(timeout=timeout)
+ if not chunk.startswith(magic):
+ if _ignore_bad_magic:
+ continue
+ raise exceptions.ProtocolError(f"Missing chunk magic: {chunk.hex()}")
+ return chunk[len(magic) :]
# process first chunk
- chunk = read_next_chunk()
- if use_chunk_magic:
- # '?' was stripped in read_next_chunk(), we just detect the "##"
- if chunk[:2] != b"##":
- raise exceptions.ProtocolError(
- f"Unexpected first chunk magic: {chunk.hex()}"
- )
- chunk = chunk[2:]
+ chunk = read_next_chunk(FIRST_MAGIC)
# extract header
header = chunk[:HEADER_LEN]
@@ -91,7 +93,7 @@ def read(transport: Transport, timeout: float | None = None) -> tuple[int, bytes
# read rest of the message
buffer = bytearray(chunk[HEADER_LEN:])
while len(buffer) < datalen:
- buffer.extend(read_next_chunk())
+ buffer.extend(read_next_chunk(NEXT_MAGIC))
return msg_type, bytes(buffer[:datalen])
@@ -351,7 +353,8 @@ def probe(
cancel_msg = messages.Cancel()
cancel_msg_type, cancel_msg_bytes = mapping.encode(cancel_msg)
write(transport, cancel_msg_type, cancel_msg_bytes)
- resp_type, resp_bytes = read(transport)
+ # Ignore previously sent unexpected packets, while waiting for the response.
+ resp_type, resp_bytes = read(transport, _ignore_bad_magic=True)
resp = mapping.decode(resp_type, resp_bytes)
if isinstance(resp, messages.Failure):
if resp.code == messages.FailureType.InvalidProtocol:
Why this scored 38/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.