fix(core): unallocated THP session lookup
What the developer wrote
Authored by Martin Milata
fix(core): unallocated THP session lookup
[no changelog]
What changed, and why it matters
This commit fixes a bug in how the Trezor hardware wallet looks up unused secure sessions. Previously, the code checked whether a session slot was 'empty' by comparing the whole slot object to a placeholder. After a code change, that comparison no longer worked correctly, so a slot could be mistaken for empty when it actually contained another user's session data. The fix makes the code explicitly read the session-state field to decide if a slot is truly unallocated. The risk is that a new session could overwrite an active session, potentially mixing up or leaking cached secrets between sessions. The commit message does not call this a security fix, and no public advisory was supplied.
Treat as a defensive correctness fix. Review whether any firmware release shipped with the buggy identity-comparison version and, if so, assess whether active sessions could be overwritten under memory pressure or malicious session creation. Add regression tests specifically for unallocated-slot detection after cache eviction. Consider whether _UNALLOCATED_STATE should be an enum or constant that cannot collide with a cleared object's default state.
Security signals we found
Incorrect identity comparison used for 'unallocated' slot detection
Session cache slot reuse/collision risk if empty slots are misidentified
Fix touches THP (Trezor Host Protocol) session management, which protects seed-derived secrets
No changelog entry and no explicit security framing by vendor
Evidence from the diff
In core/src/storage/cache_thp.py, _get_unallocated_session_index() previously tested _SESSIONS[i] is _UNALLOCATED_STATE. Because _SESSIONS is now a list of SessionThpCache/DataCache objects rather than sentinel placeholders, identity comparison against _UNALLOCATED_STATE would always be False, so no slot would ever be reported as unallocated. create_or_replace_session() was changed from assigning a fresh SessionThpCache() to calling .clear() on the chosen index, which only safely works if the chosen index is actually unallocated or intended for replacement. The new code reads the SESSION_STATE integer and compares it to _UNALLOCATED_STATE, correctly identifying free slots. The test changes confirm the lookup semantics changed and that session data is now compared via .data rather than object identity. There is no direct evidence in the diff of an exploitable secret leak, but the bug class (stale/incorrect empty-slot detection) can lead to session-cache collision or reuse.
Changed components
core/src/storage/cache_thp.pyTrezor Model T / Core THP session cache_get_unallocated_session_index()create_or_replace_session()Inspect captured patch +34 / −28
diff --git a/core/src/storage/cache_thp.py b/core/src/storage/cache_thp.py
index 40bc9236..1ec2e104 100644
--- a/core/src/storage/cache_thp.py
+++ b/core/src/storage/cache_thp.py
@@ -142,7 +142,7 @@ def create_or_replace_session(channel_id: bytes, session_id: bytes) -> SessionTh
if index is None:
index = _get_next_session_index()
- _SESSIONS[index] = SessionThpCache()
+ _SESSIONS[index].clear()
_SESSIONS[index].set(CHANNEL_ID, channel_id)
_SESSIONS[index].set(SESSION_ID, session_id)
_SESSIONS[index].set_int(LAST_USAGE, _get_usage_counter_and_increment())
@@ -173,7 +173,10 @@ def _get_next_session_index() -> int:
def _get_unallocated_session_index() -> int | None:
for i in range(_MAX_SESSIONS_COUNT):
- if (_SESSIONS[i]) is _UNALLOCATED_STATE:
+ if (
+ _SESSIONS[i].get_int(SESSION_STATE, _UNALLOCATED_STATE)
+ == _UNALLOCATED_STATE
+ ):
return i
return None
diff --git a/core/tests/test_storage.cache.py b/core/tests/test_storage.cache.py
index a85845b4..8b7ba89a 100644
--- a/core/tests/test_storage.cache.py
+++ b/core/tests/test_storage.cache.py
@@ -7,13 +7,16 @@ if utils.USE_THP:
import thp_common
from mock_wire_interface import MockHID
from storage import cache, cache_thp, cache_thp_keys
- from storage.cache_common import SESSION_STATE
+ from storage.cache_common import SESSION_STATE, DataCache
from trezor.wire.thp.session_context import SessionContext
_PROTOCOL_CACHE = cache_thp
KEY = cache_thp_keys.APP_COMMON_SEED
+ def _copy_data(dc: DataCache) -> list[bytearray]:
+ return [bytearray(f) for f in dc.data]
+
else:
from mock_storage import mock_storage
from storage import cache, cache_codec, cache_codec_keys
@@ -114,48 +117,43 @@ class TestStorageCache(TestCaseWithContext):
cid = []
sid = []
for i in range(3):
- sesions_A.append(
- cache_thp.create_or_replace_session(
- channel_A.channel_id_bytes(), (i + 1).to_bytes(1, "big")
- )
+ s = cache_thp.create_or_replace_session(
+ channel_A.channel_id_bytes(), (i + 1).to_bytes(1, "big")
)
- cid.append(sesions_A[i].channel_id)
- sid.append(sesions_A[i].session_id)
+ sesions_A.append(_copy_data(s))
+ cid.append(s.channel_id)
+ sid.append(s.session_id)
sessions_B = []
for i in range(cache_thp._MAX_SESSIONS_COUNT - 3):
- sessions_B.append(
- cache_thp.create_or_replace_session(
- channel_B.channel_id_bytes(), (i + 10).to_bytes(1, "big")
- )
+ s = cache_thp.create_or_replace_session(
+ channel_B.channel_id_bytes(), (i + 10).to_bytes(1, "big")
)
+ sessions_B.append(_copy_data(s))
for i in range(3):
- self.assertEqual(sesions_A[i], cache_thp._SESSIONS[i])
- self.assertEqual(cid[i], cache_thp._SESSIONS[i].channel_id)
- self.assertEqual(sid[i], cache_thp._SESSIONS[i].session_id)
+ self.assertEqual(sesions_A[i], cache_thp._SESSIONS[i].data)
for i in range(3, cache_thp._MAX_SESSIONS_COUNT):
- self.assertEqual(sessions_B[i - 3], cache_thp._SESSIONS[i])
+ self.assertEqual(sessions_B[i - 3], cache_thp._SESSIONS[i].data)
# Assert that new session replaces the oldest (least used) one (_SESSIONS[0])
new_session = cache_thp.create_or_replace_session(
channel_B.channel_id_bytes(), b"\xab"
)
self.assertEqual(new_session, cache_thp._SESSIONS[0])
- self.assertNotEqual(new_session.channel_id, cid[0])
- self.assertNotEqual(new_session.session_id, sid[0])
+ self.assertNotEqual(new_session.data, sesions_A[0])
# Assert that creating a new session on channel B shifts the "last usage" again
# and that _SESSIONS[1] was not replaced, but that _SESSIONS[2] was replaced
- cache_thp.update_session_last_used(
- channel_A.channel_id_bytes(), sesions_A[1].session_id
- )
- new_new_session = cache_thp.create_or_replace_session(
- channel_B.channel_id_bytes(), b"\xaa"
+ cache_thp.update_session_last_used(channel_A.channel_id_bytes(), sid[1])
+ new_new_session = _copy_data(
+ cache_thp.create_or_replace_session(
+ channel_B.channel_id_bytes(), b"\xaa"
+ )
)
- self.assertEqualExceptLastUsage(sesions_A[1], cache_thp._SESSIONS[1])
- self.assertNotEqual(sesions_A[2], cache_thp._SESSIONS[2])
- self.assertEqual(new_new_session, cache_thp._SESSIONS[2])
+ self.assertEqualExceptLastUsage(sesions_A[1], cache_thp._SESSIONS[1].data)
+ self.assertNotEqual(sesions_A[2], cache_thp._SESSIONS[2].data)
+ self.assertEqual(new_new_session, cache_thp._SESSIONS[2].data)
def test_clear(self):
channel_A = thp_common.get_new_channel(self.interface)
@@ -246,6 +244,7 @@ class TestStorageCache(TestCaseWithContext):
session_1.set_bool(KEY_BOOL, True)
# Change length of first session field to 0 so that the length check passes
+ orig_fields = session_1.fields
session_1.fields = (0,) + session_1.fields[1:]
# with self.assertRaises(AssertionError) as e:
@@ -255,7 +254,7 @@ class TestStorageCache(TestCaseWithContext):
session_2 = cache_thp.create_or_replace_session(
channel.channel_id_bytes(), b"\x02"
)
- session_2.fields = session_2.fields = (0,) + session_2.fields[1:]
+ session_2.fields = (0,) + session_2.fields[1:]
session_2.set_bool(KEY_BOOL, False)
self.assertEqual(session_2.get_bool(KEY_BOOL), False)
@@ -267,6 +266,10 @@ class TestStorageCache(TestCaseWithContext):
self.assertFalse(session_1.get_bool(KEY_BOOL))
self.assertFalse(session_2.get_bool(KEY_BOOL))
+ # Restore fields for next testcases
+ session_1.fields = orig_fields
+ session_2.fields = orig_fields
+
def test_delete(self):
channel = thp_common.get_new_channel(self.interface)
session_1 = cache_thp.create_or_replace_session(
Why this scored 45/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.