chore(core): drop trace-level THP logging
What changed, and why it matters
This commit simply turns off very detailed internal debug logging in the Trezor hardware wallet's core firmware to make automated tests run faster. It does not change any security logic, encryption, or how the device protects secrets. The logging was already only present in debug builds and never exposed sensitive data in production firmware.
No security action required. This is a routine test-optimization change. Reviewers may verify that the one remaining non-trace log line still uses an appropriate log level.
Security signals we found
No strong security signals were identified.
Evidence from the diff
The patch introduces a compile-time _TRACE = const(False) flag in three THP (Trezor Host Protocol) modules and gates existing __debug__-only trace log statements behind __debug__ and _TRACE. It also removes one redundant _log("encrypt") call and adds logger=log.warning to an existing non-trace log line about unexpected sequential bits. No cryptographic operations, state machines, memory handling, or protocol behavior are altered.
Changed components
core/src/trezor/wire/thp/channel.pycore/src/trezor/wire/thp/crypto.pycore/src/trezor/wire/thp/interface_context.pyInspect captured patch +21 / −18
diff --git a/core/src/trezor/wire/thp/channel.py b/core/src/trezor/wire/thp/channel.py
index 38477f29..13093a4f 100644
--- a/core/src/trezor/wire/thp/channel.py
+++ b/core/src/trezor/wire/thp/channel.py
@@ -66,6 +66,8 @@ _PREEMPT_TIMEOUT_MS = const(1_000)
EMPTY_ACK_PAYLOAD = memoryview(b"")
+_TRACE = const(False)
+
class Reassembler:
def __init__(self, read_buf: ThpBuffer) -> None:
@@ -169,7 +171,7 @@ class Channel:
self.channel_id: bytes = channel_cache.channel_id
self.iface_ctx: InterfaceContext = ctx
self.read_buf, self.write_buf = buffers
- if __debug__:
+ if __debug__ and _TRACE:
self._log("channel initialization")
self.channel_cache: ChannelCache = channel_cache
@@ -200,7 +202,7 @@ class Channel:
CHANNEL_STATE, default=ChannelState.UNALLOCATED
)
assert isinstance(state, int)
- if __debug__:
+ if __debug__ and _TRACE:
self._log("get_channel_state: ", state_to_str(state))
return state
@@ -211,7 +213,7 @@ class Channel:
def set_channel_state(self, state: ChannelState) -> None:
self.channel_cache.set_int(CHANNEL_STATE, state)
- if __debug__:
+ if __debug__ and _TRACE:
self._log("set_channel_state: ", state_to_str(state))
def replace_old_channels_with_the_same_host_public_key(self) -> None:
@@ -223,7 +225,7 @@ class Channel:
if was_any_replaced:
# In case a channel was replaced, close all running workflows
workflow.close_others()
- if __debug__:
+ if __debug__ and _TRACE:
self._log("Was any channel replaced? ", str(was_any_replaced))
def is_channel_to_replace(self) -> bool:
@@ -293,6 +295,7 @@ class Channel:
if __debug__:
self._log(
"Received message with an unexpected sequential bit",
+ logger=log.warning,
)
await send_ack(self, ack_bit=seq_bit)
continue
@@ -377,16 +380,16 @@ class Channel:
assert key_receive is not None
assert nonce_receive is not None
- if __debug__:
+ if __debug__ and _TRACE:
self._log("Buffer before decryption: ", hexlify_if_bytes(noise_buffer))
is_tag_valid = crypto.dec(noise_buffer, tag, key_receive, nonce_receive)
- if __debug__:
+ if __debug__ and _TRACE:
self._log("Buffer after decryption: ", hexlify_if_bytes(noise_buffer))
self.channel_cache.set_int(CHANNEL_NONCE_RECEIVE, nonce_receive + 1)
- if __debug__:
+ if __debug__ and _TRACE:
self._log("Is decrypted tag valid? ", str(is_tag_valid))
self._log("Received tag: ", hexlify_if_bytes(tag))
self._log("New nonce_receive: ", str((nonce_receive + 1)))
@@ -408,7 +411,7 @@ class Channel:
f"write message: {msg.MESSAGE_NAME}",
logger=log.info,
)
- if utils.EMULATOR:
+ if utils.EMULATOR and _TRACE:
log.debug(
__name__,
"message contents:\n%s",
@@ -462,7 +465,7 @@ class Channel:
This task is spawned concurrently with `_wait_for_ack()` using `loop.race()`,
so it will be cancelled when the expected ACK is received.
"""
- if __debug__:
+ if __debug__ and _TRACE:
self._log(f"Sending {len(payload)} bytes, latency: {ack_latency_ms} ms")
for i in range(_MAX_RETRANSMISSION_COUNT):
@@ -512,9 +515,6 @@ class Channel:
raise Timeout("THP write is blocked")
def _encrypt(self, buffer: AnyBuffer, noise_payload_len: int) -> None:
- if __debug__:
- self._log("encrypt")
-
assert len(buffer) >= noise_payload_len + TAG_LENGTH + CHECKSUM_LENGTH
noise_buffer = memoryview(buffer)[0:noise_payload_len]
@@ -528,7 +528,7 @@ class Channel:
tag = crypto.enc(noise_buffer, key_send, nonce_send)
self.channel_cache.set_int(CHANNEL_NONCE_SEND, nonce_send + 1)
- if __debug__:
+ if __debug__ and _TRACE:
self._log("New nonce_send: ", str((nonce_send + 1)))
buffer[noise_payload_len : noise_payload_len + TAG_LENGTH] = tag
@@ -549,7 +549,7 @@ class Channel:
def send_ack(channel: Channel, ack_bit: int) -> Awaitable[None]:
ctrl_byte = control_byte.add_ack_bit_to_ctrl_byte(ACK_MESSAGE, ack_bit)
header = PacketHeader(ctrl_byte, channel.get_channel_id_int(), CHECKSUM_LENGTH)
- if __debug__:
+ if __debug__ and _TRACE:
log.debug(
__name__,
"Writing ACK message to a channel with cid: %s, ack_bit: %d",
@@ -564,7 +564,7 @@ def handle_ack(ctx: Channel, ack_bit: int) -> None:
if not ABP.is_ack_valid(ctx.channel_cache, ack_bit):
return
# ACK is expected and it has correct sync bit
- if __debug__:
+ if __debug__ and _TRACE:
log.debug(
__name__,
"Received ACK message with correct ack bit",
diff --git a/core/src/trezor/wire/thp/crypto.py b/core/src/trezor/wire/thp/crypto.py
index 6c5eb8a4..1908866a 100644
--- a/core/src/trezor/wire/thp/crypto.py
+++ b/core/src/trezor/wire/thp/crypto.py
@@ -16,6 +16,8 @@ if TYPE_CHECKING:
HARDENED = const(0x8000_0000)
PUBKEY_LENGTH = const(32)
+_TRACE = const(False)
+
if __debug__:
from trezor.utils import hexlify_if_bytes
@@ -25,7 +27,7 @@ def enc(buffer: AnyBuffer, key: bytes, nonce: int, auth_data: bytes = b"") -> by
Encrypts the provided `buffer` with AES-GCM (in place).
Returns a 16-byte long encryption tag.
"""
- if __debug__:
+ if __debug__ and _TRACE:
log.debug(__name__, "enc (key: %s, nonce: %d)", hexlify_if_bytes(key), nonce)
iv = _get_iv_from_nonce(nonce)
aes_ctx = aesgcm(key, iv)
@@ -46,7 +48,7 @@ def dec(
the tag computed in decryption, otherwise it returns `False`.
"""
iv = _get_iv_from_nonce(nonce)
- if __debug__:
+ if __debug__ and _TRACE:
log.debug(__name__, "dec (key: %s, nonce: %d)", hexlify_if_bytes(key), nonce)
aes_ctx = aesgcm(key, iv)
aes_ctx.auth(auth_data)
diff --git a/core/src/trezor/wire/thp/interface_context.py b/core/src/trezor/wire/thp/interface_context.py
index be20f385..07770570 100644
--- a/core/src/trezor/wire/thp/interface_context.py
+++ b/core/src/trezor/wire/thp/interface_context.py
@@ -38,6 +38,7 @@ if TYPE_CHECKING:
from typing import Awaitable, Generator, Iterable, NoReturn
_BROADCAST_PAYLOAD_LENGTH = const(12)
+_TRACE = const(False)
# Uses `yield` instead of `await` to avoid allocations.
@@ -135,7 +136,7 @@ class InterfaceContext:
channel = self._channels[cid] = Channel(cache, self, buffers)
if channel.reassemble(packet):
- if __debug__ and channel.reassembler.message is not None:
+ if __debug__ and _TRACE and channel.reassembler.message is not None:
msg_type = "ACK" if control_byte.is_ack(ctrl_byte) else "message"
msg = channel.reassembler.message
channel._log(f"reassembled valid {msg_type}: {len(msg)} bytes")
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.