feat(core): allow caching recent THP paired hostnames and MAC addresses
What the developer wrote
Authored by Roman Zeyde
feat(core): allow caching recent THP paired hostnames and MAC addresses
Only entries with bonded MAC addresses will be stored.
[no changelog]
What changed, and why it matters
This commit adds a small cache that remembers recently paired Bluetooth-like device names and MAC addresses for Trezor's experimental Trezor Host Protocol (THP). It only stores entries whose MAC address is already bonded in the device's secure BLE storage. The change is purely additive and gated behind the USE_THP build flag. There is no direct evidence in the commit of a security vulnerability; it appears to be a feature implementation with a filtering safeguard.
No immediate action required. Treat as a normal feature commit. If auditing THP, verify that (1) the BLE bond store itself cannot be manipulated to add attacker-controlled MACs, (2) host_name values are sanitized before UI display, and (3) the protobuf message size limit is enforced at the message-definition layer, not only by the 400-byte test assertion.
Security signals we found
New flash storage namespace _THP_PAIRED_CACHE (0x22) added for serialized paired-host cache
Store path filters entries by bonded MAC address, preventing arbitrary unpaired host data from persisting
Dependency on external BLE bond list (trezorble.get_bonds) for authorization decision
Experimental protobuf decoding disabled for this message type
No changelog entry; feature is additive and behind USE_THP flag
Evidence from the diff
The patch introduces apps.thp.paired_cache and storage helpers set_thp_paired_cache/get_thp_paired_cache. On store(), entries are filtered against the set of bonded MAC addresses from trezorble.get_bonds() before protobuf serialization and flash storage. The test suite verifies that unbonded entries are dropped, bonded entries round-trip, and the serialized blob stays under 400 bytes. No input validation beyond the bond-set check is visible, and the protobuf decode uses _ENABLE_EXPERIMENTAL = False. The feature is conditionally compiled only when utils.USE_THP is true.
Changed components
core/src/apps/thp/paired_cache.pycore/src/storage/device.pycore/embed/upymod/qstrdefsport.hcore/tests/test_apps.thp.paired_cache.pyInspect captured patch +147 / −0
diff --git a/core/embed/upymod/qstrdefsport.h b/core/embed/upymod/qstrdefsport.h
index d246c9273..5ed364803 100644
--- a/core/embed/upymod/qstrdefsport.h
+++ b/core/embed/upymod/qstrdefsport.h
@@ -405,6 +405,7 @@ Q(ThpPairingMethod)
Q(alternating_bit_protocol)
Q(apps.thp)
Q(apps.thp.credential_manager)
+Q(apps.thp.paired_cache)
Q(apps.thp.pairing)
Q(cache_thp)
Q(cache_thp_keys)
@@ -417,6 +418,7 @@ Q(credential_manager)
Q(crypto)
Q(interface_context)
Q(memory_manager)
+Q(paired_cache)
Q(pairing)
Q(pairing_context)
Q(received_message_handler)
diff --git a/core/src/apps/thp/paired_cache.py b/core/src/apps/thp/paired_cache.py
new file mode 100644
index 000000000..26da46e58
--- /dev/null
+++ b/core/src/apps/thp/paired_cache.py
@@ -0,0 +1,34 @@
+from micropython import const
+
+from trezor.messages import ThpPairedCache, ThpPairedCacheEntry
+
+_ENABLE_EXPERIMENTAL = const(False)
+
+
+def load() -> list[ThpPairedCacheEntry]:
+ """Load THP paired entries from flash."""
+ from storage.device import get_thp_paired_cache
+ from trezor.protobuf import decode
+
+ if (blob := get_thp_paired_cache()) is None:
+ return [] # an empty cache
+
+ cache = decode(blob, ThpPairedCache, _ENABLE_EXPERIMENTAL)
+ return cache.entries
+
+
+def store(entries: list[ThpPairedCacheEntry], _bonds: set[bytes] | None = None) -> None:
+ """Store THP paired entries to flash."""
+ from storage.device import set_thp_paired_cache
+ from trezor.protobuf import dump_message_buffer
+
+ if _bonds is None:
+ from trezorble import get_bonds
+
+ _bonds = set(get_bonds())
+
+ # Remove entries with unbonded MAC addresses
+ entries = [e for e in entries if e.mac_addr in _bonds]
+
+ cache = ThpPairedCache(entries=entries)
+ set_thp_paired_cache(dump_message_buffer(cache))
diff --git a/core/src/storage/device.py b/core/src/storage/device.py
index 668d6c243..1c4d8c48e 100644
--- a/core/src/storage/device.py
+++ b/core/src/storage/device.py
@@ -42,6 +42,8 @@ if utils.USE_THP:
# _BRIGHTNESS = const(0x19) # int
_DISABLE_HAPTIC_FEEDBACK = const(0x20) # bool (0x01 or empty)
_DISABLE_RGB_LED = const(0x21) # bool (0x01 or empty)
+if utils.USE_THP:
+ _THP_PAIRED_CACHE = const(0x22) # bytes
SAFETY_CHECK_LEVEL_STRICT : Literal[0] = const(0)
@@ -408,3 +410,18 @@ def get_rgb_led() -> bool:
Get RGB LED enable, default to true if not set.
"""
return not common.get_bool(_NAMESPACE, _DISABLE_RGB_LED, True)
+
+
+if utils.USE_THP:
+
+ def set_thp_paired_cache(blob: bytes) -> None:
+ """
+ Set THP paired entries' cache (using protobuf serialization).
+ """
+ common.set(_NAMESPACE, _THP_PAIRED_CACHE, blob)
+
+ def get_thp_paired_cache() -> bytes | None:
+ """
+ Get THP paired entries' cache (using protobuf serialization).
+ """
+ return common.get(_NAMESPACE, _THP_PAIRED_CACHE)
diff --git a/core/tests/test_apps.thp.paired_cache.py b/core/tests/test_apps.thp.paired_cache.py
new file mode 100644
index 000000000..a9fcc368e
--- /dev/null
+++ b/core/tests/test_apps.thp.paired_cache.py
@@ -0,0 +1,94 @@
+# flake8: noqa: F403,F405
+from common import * # isort: skip
+from trezor import config, utils
+
+if utils.USE_THP:
+ from storage.device import get_thp_paired_cache
+ from trezor.messages import ThpPairedCacheEntry
+
+ from apps.thp import paired_cache
+
+ ALL_ENTRIES = [
+ ThpPairedCacheEntry(mac_addr=b"\x01\x02\x03\x04\x05\x06", host_name="First"),
+ ThpPairedCacheEntry(mac_addr=b"\x11\x12\x13\x14\x15\x16", host_name="Second"),
+ ThpPairedCacheEntry(mac_addr=b"\x21\x22\x23\x24\x25\x26", host_name="Third"),
+ ThpPairedCacheEntry(mac_addr=b"\x31\x32\x33\x34\x35\x36", host_name="Fourth"),
+ ThpPairedCacheEntry(mac_addr=b"\x41\x42\x43\x44\x45\x46", host_name="Fifth"),
+ ThpPairedCacheEntry(mac_addr=b"\x51\x52\x53\x54\x55\x56", host_name="Sixth"),
+ ThpPairedCacheEntry(mac_addr=b"\x61\x62\x63\x64\x65\x66", host_name="Seventh"),
+ ThpPairedCacheEntry(mac_addr=b"\x71\x72\x73\x74\x75\x76", host_name="Eighth"),
+ ]
+
+
+@unittest.skipUnless(utils.USE_THP, "only needed for THP")
+class TestTrezorHostProtocolPairedCache(unittest.TestCase):
+ def setUp(self):
+ config.init()
+ config.wipe()
+
+ def test_empty(self):
+ self.assertEqual(paired_cache.load(), [])
+ paired_cache.store(entries=[], _bonds=[])
+ self.assertEqual(paired_cache.load(), [])
+
+ def test_store_and_load(self):
+ for i in range(len(ALL_ENTRIES)):
+ entries = ALL_ENTRIES[: i + 1]
+ bonds = {e.mac_addr for e in entries}
+ paired_cache.store(entries=entries, _bonds=bonds)
+ self.assertListEqual(paired_cache.load(), entries)
+
+ paired_cache.store(entries=entries, _bonds=set())
+ self.assertListEqual(paired_cache.load(), [])
+
+ def test_store_no_bonds(self):
+ for i in range(len(ALL_ENTRIES)):
+ entries = ALL_ENTRIES[: i + 1]
+ paired_cache.store(entries=entries, _bonds=[])
+ self.assertListEqual(paired_cache.load(), [])
+
+ paired_cache.store(entries=entries, _bonds=set())
+ self.assertListEqual(paired_cache.load(), [])
+
+ def test_store_less_bonds(self):
+ for i in range(len(ALL_ENTRIES)):
+ entries = ALL_ENTRIES[: i + 1]
+ # last entry has no matching bond
+ bonds = {e.mac_addr for e in entries[:-1]}
+ paired_cache.store(entries=entries, _bonds=bonds)
+ self.assertListEqual(paired_cache.load(), entries[:-1])
+
+ paired_cache.store(entries=entries, _bonds=set())
+ self.assertListEqual(paired_cache.load(), [])
+
+ def test_store_more_bonds(self):
+ for i in range(len(ALL_ENTRIES)):
+ entries = ALL_ENTRIES[:i]
+ # last bond have no matching entry
+ bonds = {e.mac_addr for e in ALL_ENTRIES[: i + 1]}
+ paired_cache.store(entries=entries, _bonds=bonds)
+ self.assertListEqual(paired_cache.load(), entries)
+
+ paired_cache.store(entries=entries, _bonds=set())
+ self.assertListEqual(paired_cache.load(), [])
+
+ def test_max_size(self):
+ self.assertIsNone(get_thp_paired_cache())
+ # serialize longest `host_name`` and maximal number of bonds
+ entries = [
+ ThpPairedCacheEntry(mac_addr=bytes([i] * 6), host_name=f"{i}" * 32)
+ for i in range(8)
+ ]
+ bonds = {e.mac_addr for e in entries}
+ paired_cache.store(entries=entries, _bonds=bonds)
+ self.assertListEqual(paired_cache.load(), entries)
+
+ cache_blob = get_thp_paired_cache()
+ self.assertIsNotNone(cache_blob)
+ # Check that serialized size is not too large:
+ # 8 entries x (32 bytes [name] + 6 bytes [addr]) = 304 bytes
+ assert len(cache_blob) <= 400
+
+
+if __name__ == "__main__":
+ unittest.main()
Why this scored 20/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.