BitHighlander · BitHighlander · Sep 19, 2026 · Sep 18, 2026 · Aug 27, 2026 · Sep 18, 2026
diff --git a/keepkeylib/messages_ethereum_pb2.py b/keepkeylib/messages_ethereum_pb2.py
diff --git a/keepkeylib/messages_pb2.py b/keepkeylib/messages_pb2.py
diff --git a/scripts/generate-test-report.py b/scripts/generate-test-report.py
@@ -3401,6 +3401,17 @@ def screenshot_filter(fw_version):
     return ' or '.join(terms)
 
 
+def screenshot_test_list(fw_version):
+    """Return exact module::method selectors consumed by conftest.py."""
+    active = [x for x in SECTIONS if ver_ge(fw_version, x[2])]
+    pairs = set()
+    for _letter, _title, _mf, _bg, _fl, tests in active:
+        for _tid, mod, meth, _ttl, _ctx, screens in tests:
+            if screens:
+                pairs.add('%s::%s' % (mod, meth))
+    return '\n'.join(sorted(pairs))
+
+
 # Modules whose tests must actually RUN once the firmware is new enough to be
 # catalogued for them -- a skip is a failure, not a waiver.
 #
@@ -3515,6 +3526,8 @@ def main():
                    help='JUnit XML for --screenshot-audit, so skipped tests are not counted missing')
     p.add_argument('--screenshot-filter', action='store_true',
                    help='Print pytest -k expression for tests needing screenshots, then exit')
+    p.add_argument('--screenshot-test-list', action='store_true',
+                   help='Print exact module::method screenshot selectors, then exit')
     p.add_argument('--validate-junit', action='store_true',
                    help='Validate JUnit results against SECTIONS, exit non-zero on failures')
     p.add_argument('--build-variant', choices=('full', 'bitcoin-only'), default='full',
@@ -3550,6 +3563,9 @@ def main():
     if args.screenshot_filter:
         print(screenshot_filter(fw))
         sys.exit(0)
+    if args.screenshot_test_list:
+        print(screenshot_test_list(fw))
+        sys.exit(0)
 
     if args.validate_junit:
         if not args.junit:

diff --git a/setup.py b/setup.py
@@ -13,7 +13,7 @@
     test_suite='tests/**/test_*.py',
     install_requires=[
         'ecdsa>=0.9',
-        'protobuf>=3.20.0',
+        'protobuf>=3.17,<4',
         'mnemonic>=0.8',
         'hidapi>=0.7.99.post15',
         'libusb1>=1.6'

diff --git a/tests/common.py b/tests/common.py
@@ -127,6 +127,13 @@ def assertEqual(self, lhs, rhs):
     def assertEndsWith(self, s, suffix):
         self.assertTrue(s.endswith(suffix), "'{}'.endswith('{}')".format(s, suffix))
 
+    def firmware_at_least(self, ver_required):
+        """Return whether the connected firmware includes a versioned feature."""
+        self.client.init_device()
+        features = self.client.features
+        version = "%s.%s.%s" % (features.major_version, features.minor_version, features.patch_version)
+        return semver.VersionInfo.parse(version) >= semver.VersionInfo.parse(ver_required)
+
     def requires_firmware(self, ver_required):
         self.client.init_device()
         features = self.client.features
@@ -239,6 +246,16 @@ def requires_fullFeature(self):
             self.client.features.firmware_variant == "EmulatorBTC":
         self.skipTest("Full feature firmware required to run this test")
 
+    def requires_dice_modes(self):
+        """Skip unless the firmware reports the verifiable dice modes.
+
+        A capability, not a version: firmware without the unit skips the
+        unknown ResetDevice.dice_only field and runs the older ceremony.
+        """
+        self.client.init_device()
+        if not getattr(self.client.features, 'supports_dice_modes', False):
+            self.skipTest("Firmware does not report supports_dice_modes")
+
     def requires_bitcoinOnly(self):
       """Inverse of requires_fullFeature(): skip unless this IS the
       bitcoin-only product.

diff --git a/tests/conftest.py b/tests/conftest.py
@@ -19,6 +19,43 @@
 
 import requests
 
+
+def pytest_collection_modifyitems(config, items):
+    """Select exact report test IDs for the screenshot-only pytest phase.
+
+    Firmware CI's python-keepkey-tests.sh passes the report's selector list in
+    KEEPKEY_SCREENSHOT_TESTS. Without this hook the screenshot phase ran the
+    whole suite and the job hit its 30-minute limit.
+    """
+    if os.environ.get('KEEPKEY_SCREENSHOT') != '1':
+        return
+    encoded = os.environ.get('KEEPKEY_SCREENSHOT_TESTS', '')
+    if not encoded:
+        raise pytest.UsageError(
+            'KEEPKEY_SCREENSHOT_TESTS must list exact module::method pairs')
+    selected_pairs = set()
+    for line in encoded.splitlines():
+        if not line:
+            continue
+        parts = line.split('::')
+        if len(parts) != 2 or not all(parts):
+            raise pytest.UsageError(
+                'invalid KEEPKEY_SCREENSHOT_TESTS entry %r' % line)
+        selected_pairs.add(tuple(parts))
+    selected = []
+    deselected = []
+    for item in items:
+        module = os.path.splitext(os.path.basename(item.location[0]))[0]
+        method = getattr(item, 'originalname', None) or item.name.split('[', 1)[0]
+        if (module, method) in selected_pairs:
+            selected.append(item)
+        else:
+            deselected.append(item)
+    if deselected:
+        config.hook.pytest_deselected(items=deselected)
+    items[:] = selected
+
+
 if os.environ.get('KEEPKEY_SCREENSHOT') == '1':
     import common
 
@@ -56,6 +93,33 @@ def _patched_setUp(self):
     common.KeepKeyTest.setUp = _patched_setUp
 
 
+def _configured_emulator_endpoints(getaddrinfo):
+    """Return the exact emulator names and addresses the harness configured.
+
+    In the firmware CI compose network the emulator is the service `kkemu`
+    (KK_TRANSPORT_MAIN=kkemu:11044), which is not loopback. Allow exactly the
+    two configured transports, as the audit line does, and nothing else.
+    """
+    names = set()
+    addresses = set()
+    for variable, default in (
+            ('KK_TRANSPORT_MAIN', '127.0.0.1:11044'),
+            ('KK_TRANSPORT_DEBUG', '127.0.0.1:11045')):
+        endpoint = os.environ.get(variable, default)
+        try:
+            host, port_text = endpoint.rsplit(':', 1)
+            port = int(port_text)
+        except (AttributeError, TypeError, ValueError):
+            raise RuntimeError(
+                '%s must be a host:port emulator endpoint, got %r' %
+                (variable, endpoint))
+        names.add((host, port))
+        for result in getaddrinfo(host, port, type=socket.SOCK_DGRAM):
+            sockaddr = result[4]
+            addresses.add((sockaddr[0], sockaddr[1]))
+    return names, addresses
+
+
 def _is_loopback_address(address):
     """Allow emulator traffic while rejecting every external destination."""
     if not isinstance(address, tuple):
@@ -81,30 +145,40 @@ def deny_external_network(monkeypatch, request):
     original_connect_ex = socket.socket.connect_ex
     original_sendto = socket.socket.sendto
     original_request = requests.sessions.Session.request
+    emulator_names, emulator_addresses = _configured_emulator_endpoints(
+        original_getaddrinfo)
+
+    def allowed(address):
+        if isinstance(address, tuple) and len(address) >= 2:
+            endpoint = (address[0], address[1])
+            if endpoint in emulator_names or endpoint in emulator_addresses:
+                return True
+        return _is_loopback_address(address)
 
     def denied(destination):
         raise AssertionError(
             'authoritative test attempted external network access: '
             'test=%s destination=%r' % (nodeid, destination))
 
     def guarded_getaddrinfo(host, *args, **kwargs):
-        if not _is_loopback_address((host, 0)):
+        port = args[0] if args else kwargs.get('port')
+        if (host, port) not in emulator_names and not _is_loopback_address((host, 0)):
             denied(host)
         return original_getaddrinfo(host, *args, **kwargs)
 
     def guarded_connect(sock, address):
-        if not _is_loopback_address(address):
+        if not allowed(address):
             denied(address)
         return original_connect(sock, address)
 
     def guarded_connect_ex(sock, address):
-        if not _is_loopback_address(address):
+        if not allowed(address):
             denied(address)
         return original_connect_ex(sock, address)
 
     def guarded_sendto(sock, data, *args):
         address = args[-1]
-        if not _is_loopback_address(address):
+        if not allowed(address):
             denied(address)
         return original_sendto(sock, data, *args)
 

diff --git a/tests/test_msg_ethereum_clear_signing.py b/tests/test_msg_ethereum_clear_signing.py
@@ -1018,15 +1018,26 @@ def test_binding_happy_path_signs_and_recovers(self):
     def _clearsign_flow(self, flow, chain_id=1):
         """Run one catalog flow END-TO-END with AdvancedMode ON: real tx,
         per-tx-bound metadata, who/what/why annotation plus the ordinary raw
-        review (auto-acked), sign, and assert the signature recovers to the
-        device signer over this exact digest."""
+        review (auto-acked), then either sign and recover the exact digest or
+        assert the release policy's explicit fail-closed rejection."""
         n = parse_path(DEVICE_PATH)
         tx_hash = flow_tx_hash(flow, chain_id)
         resp = self.client.ethereum_send_tx_metadata(
             signed_payload=flow_blob(flow, chain_id),
             metadata_version=1, key_id=TEST_KEY_ID)
         self.assertEqual(resp.classification, CLASSIFICATION_VERIFIED)
 
+        if flow['key'] == 'erc20-approve-unlimited':
+            with self.assertRaises(CallException) as ctx:
+                self.client.ethereum_sign_tx(
+                    n=n, nonce=FLOW_NONCE, gas_price=FLOW_GAS_PRICE,
+                    gas_limit=FLOW_GAS_LIMIT, to=flow['to'],
+                    value=flow['value'], data=flow['data'],
+                    chain_id=chain_id)
+            self.assertIn('Unlimited ERC20 approval is disabled',
+                          str(ctx.exception))
+            return
+
         sig_v, sig_r, sig_s = self.client.ethereum_sign_tx(
             n=n, nonce=FLOW_NONCE, gas_price=FLOW_GAS_PRICE,
             gas_limit=FLOW_GAS_LIMIT, to=flow['to'], value=flow['value'],
@@ -1089,11 +1100,8 @@ def test_replay_rejected_when_digest_differs(self):
     def test_advanced_mode_gate(self):
         """AdvancedMode OFF + unknown contract + no metadata → hard reject;
         ON → raw-data confirm path signs; recognized ERC-20 transfer unaffected."""
-        # RC18 predates the rule that loading a runtime signer itself requires
-        # AdvancedMode. The first released firmware line carrying that complete
-        # gate is 7.16; the older blind-transaction gate remains covered by
-        # test_msg_ethereum_signtx on RC18.
-        self.requires_firmware("7.16.0")
+        # Canonical 7.15 requires AdvancedMode before runtime signer loading.
+        self.requires_firmware("7.15.0")
         n = parse_path(DEVICE_PATH)
         data = aave_supply_calldata(1000000000000000000)
 

diff --git a/tests/test_msg_ethereum_erc20_uniswap_liquidity.py b/tests/test_msg_ethereum_erc20_uniswap_liquidity.py
@@ -27,36 +27,47 @@
 
 class TestMsgEthereumUniswaptxERC20(common.KeepKeyTest):
 
+    def setUp(self):
+        super(TestMsgEthereumUniswaptxERC20, self).setUp()
+        # Canonical 7.15 routes unknown token contracts through explicitly
+        # authorized raw review. Exercise that path instead of an emulator skip.
+        self.requires_firmware("7.15.0")
+
     def test_sign_uni_approve_liquidity_ETH(self):
         self.requires_fullFeature()
         self.requires_firmware("7.1.0")
         self.setup_mnemonic_nopin_nopassphrase()
+        self.client.apply_policy("AdvancedMode", 1)
 
-        # Approval tx for the ETH/FOX pool
-        sig_v, sig_r, sig_s = self.client.ethereum_sign_tx(
-            n=[2147483692,2147483708,2147483648,0,0],
-            nonce=0xf,
-            gas_price=0x2980872680,
-            gas_limit=0xbd0e,
-            value=0x0,
-            to=binascii.unhexlify('470e8de2ebaef52014a47cb5e6af86884947f08c'),     # fox pool
-            address_type=0,
-            chain_id=1,
-            # The data below is generally broken into 32-byte chunks except for the function selector (4 bytes_ and 
-            # keccak signatures (4 bytes)
-            data=binascii.unhexlify('095ea7b3' +                                      # approve
-                '0000000000000000000000007a250d5630b4cf539739df2c5dacb4c659f2488d' +  # uniswap v2: router 2 contract address
-                'ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff')   # approve amount
+        # Unlimited approval is deliberately disabled on canonical 7.15.
+        # This legacy vector must be refused, not skipped or signed.
+        with self.assertRaises(CallException) as caught:
+            self.client.ethereum_sign_tx(
+                n=[2147483692,2147483708,2147483648,0,0],
+                nonce=0xf,
+                gas_price=0x2980872680,
+                gas_limit=0xbd0e,
+                value=0x0,
+                to=binascii.unhexlify('470e8de2ebaef52014a47cb5e6af86884947f08c'),     # fox pool
+                address_type=0,
+                chain_id=1,
+                # The data below is generally broken into 32-byte chunks except for the function selector (4 bytes_ and
+                # keccak signatures (4 bytes)
+                data=binascii.unhexlify('095ea7b3' +                                      # approve
+                    '0000000000000000000000007a250d5630b4cf539739df2c5dacb4c659f2488d' +  # uniswap v2: router 2 contract address
+                    'ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff')   # approve amount
+
+            )
+        self.assertEqual(caught.exception.args[0],
+                         proto_types.Failure_ActionCancelled)
+        self.assertIn("Unlimited ERC20 approval is disabled",
+                      str(caught.exception))
 
-        )        
-        self.assertEqual(sig_v, 38)
-        self.assertEqual(binascii.hexlify(sig_r), '7f7a5ce501371a01ead394d2186385742d5fbdc3d85da98249d2a05043ac6d5a')
-        self.assertEqual(binascii.hexlify(sig_s), '329954b284ed1df9a6242820e793b9719c0c6c21cae5f90190ce61c7f73c731e')
-
     def test_sign_uni_add_liquidity_ETH(self):
         self.requires_fullFeature()
         self.requires_firmware("7.1.0")
         self.setup_mnemonic_nopin_nopassphrase()
+        self.client.apply_policy("AdvancedMode", 1)
 
         # Add liquidity to ETH/FOX pool
         sig_v, sig_r, sig_s = self.client.ethereum_sign_tx(
@@ -68,7 +79,7 @@ def test_sign_uni_add_liquidity_ETH(self):
             to=binascii.unhexlify('7a250d5630B4cF539739dF2C5dAcb4c659F2488D'),     # UNISWAP router
             address_type=0,
             chain_id=1,
-            # The data below is generally broken into 32-byte chunks except for the function selector (4 bytes_ and 
+            # The data below is generally broken into 32-byte chunks except for the function selector (4 bytes_ and
             # keccak signatures (4 bytes)
             data=binascii.unhexlify('f305d719' +                                      # addLiquidityETH
                 '000000000000000000000000c770eefad204b5180df6a14ee197d99d808ee52d' +  # FOX token
@@ -77,17 +88,16 @@ def test_sign_uni_add_liquidity_ETH(self):
                 '0000000000000000000000000000000000000000000000000000fb98b65aba40' +  # min amount of eth token
                 '0000000000000000000000003f2329C9ADFbcCd9A84f52c906E936A42dA18CB8' +  # eth address (self)
                 '00000000000000000000000000000000000000000000000000000178a9380e5f')   # deadline
-        )        
+        )
         self.assertEqual(sig_v, 37)
-        self.assertEqual(binascii.hexlify(sig_r), '8547542bc74c0dcc6ca8b02a79e0dccd336856d8c48376289a2a697d864a5892')
-        self.assertEqual(binascii.hexlify(sig_s), '0a8eec6856aef8caa234240b06862976f8e238e8b24f5c989279507dd7e51ccd')
+        self.assertEqual(binascii.hexlify(sig_r).decode("ascii"), '8547542bc74c0dcc6ca8b02a79e0dccd336856d8c48376289a2a697d864a5892')
+        self.assertEqual(binascii.hexlify(sig_s).decode("ascii"), '0a8eec6856aef8caa234240b06862976f8e238e8b24f5c989279507dd7e51ccd')
 
     def test_sign_uni_remove_liquidity_ETH(self):
         self.requires_fullFeature()
-        # Sending the withdrawn assets to a third-party recipient was refused
-        # by RC18. The reviewed external-recipient flow lands on the 7.16 line.
-        self.requires_firmware("7.16.0")
+        self.requires_firmware("7.1.0")
         self.setup_mnemonic_nopin_nopassphrase()
+        self.client.apply_policy("AdvancedMode", 1)
 
         # remove liquidity from the ETH/FOX pool
         sig_v, sig_r, sig_s = self.client.ethereum_sign_tx(
@@ -99,7 +109,7 @@ def test_sign_uni_remove_liquidity_ETH(self):
             to=binascii.unhexlify('7a250d5630B4cF539739dF2C5dAcb4c659F2488D'),     # UNISWAP router
             address_type=0,
             chain_id=1,
-            # The data below is generally broken into 32-byte chunks except for the function selector (4 bytes_ and 
+            # The data below is generally broken into 32-byte chunks except for the function selector (4 bytes_ and
             # keccak signatures (4 bytes)
             data=binascii.unhexlify('02751cec' +                                      # addLiquidityETH
                 '000000000000000000000000c770eefad204b5180df6a14ee197d99d808ee52d' +  # FOX token
@@ -108,10 +118,10 @@ def test_sign_uni_remove_liquidity_ETH(self):
                 '0000000000000000000000000000000000000000000000000000fb04c77f3e94' +  # min amount of eth token
                 '0000000000000000000000005028d647b74f12903e6d5f3969f8f624e6a9a93d' +  # to address (not self)
                 '00000000000000000000000000000000000000000000000000000178b2062f3d')   # deadline
-        )        
+        )
         self.assertEqual(sig_v, 37)
-        self.assertEqual(binascii.hexlify(sig_r), '7143f0d8e5505a8cfb1df55e9c5d7433eba33a61959137c08cc5c088ec12ab5d')
-        self.assertEqual(binascii.hexlify(sig_s), '20b456d6c13295f5abb6109d7ade2c5d5fc395963b1e45d92e6dc8c33749c517')
+        self.assertEqual(binascii.hexlify(sig_r).decode("ascii"), '7143f0d8e5505a8cfb1df55e9c5d7433eba33a61959137c08cc5c088ec12ab5d')
+        self.assertEqual(binascii.hexlify(sig_s).decode("ascii"), '20b456d6c13295f5abb6109d7ade2c5d5fc395963b1e45d92e6dc8c33749c517')
 
 if __name__ == '__main__':
     unittest.main()