test(core): wait for `Cancel` response before sending `Ping`
What the developer wrote
Authored by Roman Zeyde
test(core): wait for `Cancel` response before sending `Ping`
Otherwise, we may get stuck when using USB transport:
if the device is stuck sending, and not reading new messages from the
host, the first write may get stuck - and the test will deadlock.
[no changelog]
What changed, and why it matters
This is a test-only change to the Python Trezor client library. It fixes a potential deadlock in test synchronization code by waiting for a response to a Cancel message before sending a Ping message. There is no change to device firmware or any security-sensitive behavior, and no security vulnerability is present in the commit.
No security action needed. Treat as a normal test reliability improvement.
Security signals we found
No strong security signals were identified.
Evidence from the diff
The diff refactors sync_responses() in python/src/trezorlib/protocol_v1.py. Previously it sent Cancel then immediately sent Ping, then read responses. The new code wraps send+read into a _call() helper and waits for the expected Failure response to Cancel before sending Ping. This prevents a USB transport deadlock where a write could block because the device is still sending and not reading. The change is purely in test infrastructure and does not alter protocol security.
Changed components
python/src/trezorlib/protocol_v1.pytest synchronization helper `sync_responses()`Inspect captured patch +15 / −12
diff --git a/python/src/trezorlib/protocol_v1.py b/python/src/trezorlib/protocol_v1.py
index 692f20eb..9bd1c705 100644
--- a/python/src/trezorlib/protocol_v1.py
+++ b/python/src/trezorlib/protocol_v1.py
@@ -381,19 +381,22 @@ def sync_responses(
retries: int = 10,
) -> None:
"""Sync responses from the transport."""
+
+ def _call(msg: MessageType, is_expected: t.Callable[[MessageType], bool]) -> None:
+ write(transport, *mapping.encode(msg))
+ for _ in range(retries):
+ resp_type, resp_bytes = read(transport, _ignore_bad_magic=True)
+ resp = mapping.decode(resp_type, resp_bytes)
+ if is_expected(resp):
+ return
+ raise exceptions.ProtocolError("Failed to sync responses")
+
# cancel anything on screen -- on T1B1 this is the only way to exit e.g. a PIN prompt.
- cancel_msg = mapping.encode(messages.Cancel())
- write(transport, *cancel_msg)
+ _call(messages.Cancel(), lambda msg: isinstance(msg, messages.Failure))
# prepare an unique message to wait for
sync_string = "SYNC" + secrets.token_hex(8)
- ping_msg = mapping.encode(messages.Ping(message=sync_string))
- # prepare
- write(transport, *ping_msg)
-
- for _ in range(retries):
- resp_type, resp_bytes = read(transport, _ignore_bad_magic=True)
- resp = mapping.decode(resp_type, resp_bytes)
- if isinstance(resp, messages.Success) and resp.message == sync_string:
- return
- raise exceptions.ProtocolError("Failed to sync responses")
+ _call(
+ messages.Ping(message=sync_string),
+ lambda msg: isinstance(msg, messages.Success) and msg.message == sync_string,
+ )
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.