test(erc7730): Stack 08 embedded replay and evidence contracts

scripts/generate-test-report.py

@@ -437,6 +437,7 @@ def parse_junit(path):
     ('test_msg_ethereum_erc7730_runtime', 'test_interpolated_intent_parts_show_the_same_values_as_fields'),
     ('test_msg_ethereum_erc7730_runtime', 'test_embedded_call_is_shown_under_a_blind_sign_warning'),
     ('test_msg_ethereum_erc7730_runtime', 'test_inner_call_is_clear_signed_with_its_own_definition'),
+    ('test_msg_ethereum_erc7730_runtime', 'test_restoring_outer_definition_rejects_another_valid_definition'),
     ('test_msg_ethereum_erc7730_runtime', 'test_an_inner_definition_the_device_refuses_falls_back_to_blind'),
     # The newest/highest-stakes tx shapes get the full ordered walkthrough too.
     ('test_msg_ethereum_clear_signing', 'test_clearsign_eip7702_setcode_authorization'),
@@ -3438,6 +3439,31 @@ def _arg_shown(a):
           'Nested optional fields remain visible',
           'Groups preserve every field; zero and nonzero optional amounts and fees both appear with the decoded recipient.',
           ['Recipient', 'Amount', 'Fee']),
+         ('EX40', 'test_msg_ethereum_erc7730_runtime',
+          'test_large_inner_call_is_bound_across_transport_chunks',
+          'Large inner calls bind across transport chunks',
+          'A 1,200-byte memo is streamed in 31- and 1,024-byte replies; the inner amount is shown, the ordinary signature matches, and a changed inner replay is refused.',
+          []),
+         ('EX41', 'test_msg_ethereum_erc7730_runtime',
+          'test_outer_bytes_changed_during_inner_review_are_refused',
+          'Inner review binds the entire outer transaction',
+          'Changing only the outer operation word during inner validation is refused before any inner field.',
+          []),
+         ('EX42', 'test_msg_ethereum_erc7730_runtime',
+          'test_restoring_outer_definition_rejects_another_valid_definition',
+          'Restored parent definition keeps its identity',
+          'A separately accepted signed definition for the same outer selector cannot replace the parent after inner review.',
+          ['Inner review before parent restore refusal']),
+         ('EX43', 'test_msg_ethereum_erc7730_runtime',
+          'test_inner_definition_identity_cannot_change_between_chunks',
+          'Inner definition identity is stable across chunks',
+          'Changing the definition ID after the first 31-byte reply is refused before inner action review.',
+          []),
+         ('EX44', 'test_msg_ethereum_erc7730_runtime',
+          'test_cancelled_inner_review_clears_preload_in_the_same_session',
+          'Inner cancellation clears the preload',
+          'Declining inner signer, action or field returns no signature; an immediate ordinary retry without Initialize has no certified screens or definition requests.',
+          []),
      ]),
 
     # Two-character id because all 26 letters were taken. The catalog keys on a

tests/test_msg_ethereum_erc7730_runtime.py

@@ -96,13 +96,16 @@ def _first_pages(self):
                 if code != types.ButtonRequest_Other]
 
     def _walk(self, start, envelope=b"", doc=None, change_pass=None, cancel_button=None,
-              arguments=None, catalog=None, altered=None, cancel_title=None):
+              arguments=None, catalog=None, altered=None, cancel_title=None,
+              chunk_size=None):
         response = self.client.call_raw(start)
         self.definition_requests = 0
         self.button_codes = []
         self.screens = []
+        self.calldata_chunk_sizes = []
         buttons = 0
         calldata_passes = 0
+        calldata_offset = 0
         typed_passes = 0
         for _ in range(1000):
             if isinstance(response, proto.ButtonRequest):
@@ -140,15 +143,25 @@ def _walk(self, start, envelope=b"", doc=None, change_pass=None, cancel_button=N
                          eip712_stream.encode_value(resolved[1], resolved[2]))
                 response = self.client.call_raw(eth.EthereumTypedDataValueAck(value=value))
             elif isinstance(response, eth.EthereumTxRequest) and response.HasField("data_length"):
-                calldata_passes += 1
+                if calldata_offset == 0:
+                    calldata_passes += 1
                 data = arguments
                 if altered and altered[0] == calldata_passes:
                     data = altered[1]
                 if data is None:
                     data = (43 if change_pass == calldata_passes else 42).to_bytes(32, "big")
                     data += (7).to_bytes(32, "big")
-                self.assertEqual(response.data_length, len(data))
-                response = self.client.call_raw(eth.EthereumTxAck(data_chunk=data))
+                self.assertGreater(response.data_length, 0)
+                self.assertLessEqual(response.data_length,
+                                     len(data) - calldata_offset)
+                length = min(response.data_length,
+                             chunk_size or response.data_length)
+                chunk = data[calldata_offset:calldata_offset + length]
+                self.calldata_chunk_sizes.append(length)
+                calldata_offset += length
+                if calldata_offset == len(data):
+                    calldata_offset = 0
+                response = self.client.call_raw(eth.EthereumTxAck(data_chunk=chunk))
             else:
                 return response, buttons, calldata_passes, typed_passes
         self.fail("protocol did not terminate")
@@ -573,7 +586,9 @@ def test_embedded_call_is_shown_under_a_blind_sign_warning(self):
             "uint8 operation)")
     TRANSFER = "transfer(address to,uint256 amount)"
 
-    def _exec_setup(self, amount_path=True, spender="@.to"):
+    def _exec_setup(self, amount_path=True, spender="@.to", memo=None):
+        inner_signature = (self.TRANSFER if memo is None else
+                           "transfer(address to,uint256 amount,bytes memo)")
         params = {"calleePath": "to"}
         if spender:
             params["spenderPath"] = spender
@@ -584,15 +599,15 @@ def _exec_setup(self, amount_path=True, spender="@.to"):
                 {"path": "data", "label": "Transaction", "format": "calldata",
                  "params": params},
                 {"path": "operation", "label": "Operation", "format": "raw"}]}}}}
-        inner_descriptor = {"display": {"formats": {self.TRANSFER: {
+        inner_descriptor = {"display": {"formats": {inner_signature: {
             "intent": "Transfer", "fields": [
                 {"path": "to", "label": "Recipient", "format": "addressName"},
                 {"path": "amount", "label": "Amount", "format": "tokenAmount",
                  "params": {"tokenPath": "@.to"}}]}}}}
         outer = erc7730_compiler.compile_calldata(
             outer_descriptor, self.EXEC, 1, ADDRESS)
         inner = erc7730_compiler.compile_calldata(
-            inner_descriptor, self.TRANSFER, 1, self.USDC)
+            inner_descriptor, inner_signature, 1, self.USDC)
         self._load_signer()
         outer_env = self._envelope(outer)
         inner_def = erc7730.Definition(self._envelope(inner), 1, 1, self.USDC,
@@ -601,6 +616,10 @@ def _exec_setup(self, amount_path=True, spender="@.to"):
         self._exec_outer = (outer, outer_def)
         call = (bytes.fromhex("a9059cbb") + self._word(OTHER_ADDRESS) +
                 self._word(1500000))
+        if memo is not None:
+            call = (inner[38:42] + self._word(OTHER_ADDRESS) +
+                    self._word(1500000) + self._word(96) +
+                    self._word(len(memo)) + memo + bytes(-len(memo) % 32))
         arguments = (self._word(self.USDC) + self._word(0) +
                      self._word(128) + self._word(0) +
                      self._word(len(call)) + call +
@@ -697,6 +716,152 @@ def test_inner_bytes_changed_in_an_inner_pass_are_refused(self):
         self.assertEqual(passes, 5)
         self.assertNotIn("Inner field", [s[0] for s in self.screens])
 
+    def test_large_inner_call_is_bound_across_transport_chunks(self):
+        # The inner bytes exceed both the capture buffer and one transport
+        # reply. A 31-byte reply splits selectors, offsets and ABI words.
+        for chunk_size in (1024, 31):
+            with self.subTest(chunk_size=chunk_size):
+                _, arguments, outer_def, inner_def = self._exec_setup(
+                    memo=b"Z" * 1200)
+                outer = self._exec_outer[0]
+                erc7730.preload(self.client, outer_def)
+                self._drop_setup_screenshots()
+                start = self._audit_start(outer, 4 + len(arguments))
+                reviewed, _, _, _ = self._walk(
+                    start, arguments=arguments, chunk_size=chunk_size,
+                    catalog=erc7730.Catalog((outer_def, inner_def)))
+                self.assertIsInstance(reviewed, eth.EthereumTxRequest)
+                self.assertTrue(reviewed.HasField("signature_r"))
+                reviewed_chunks = list(self.calldata_chunk_sizes)
+                self.assertGreater(len(reviewed_chunks), 1)
+                self.assertLessEqual(max(reviewed_chunks), chunk_size)
+                if chunk_size == 31:
+                    self.assertGreater(len(reviewed_chunks), 40)
+                self.assertIn(("Inner field", "Amount:\n1.5 USDC"),
+                              self._relevant())
+                ordinary, _, _, _ = self._walk(
+                    start, arguments=arguments, chunk_size=chunk_size)
+                self.assertIsInstance(ordinary, eth.EthereumTxRequest)
+                self.assertTrue(ordinary.HasField("signature_r"))
+                self.assertGreater(len(self.calldata_chunk_sizes), 1)
+                self.assertLessEqual(max(self.calldata_chunk_sizes), chunk_size)
+                self.assertEqual(self.definition_requests, 0)
+                self.assertEqual(
+                    (reviewed.signature_r, reviewed.signature_s,
+                     reviewed.signature_v),
+                    (ordinary.signature_r, ordinary.signature_s,
+                     ordinary.signature_v))
+                if chunk_size == 31:
+                    # Change an embedded memo byte on the inner validation
+                    # pass, after the outer review fixed the signed digest.
+                    altered = bytearray(arguments)
+                    altered[-40] ^= 1
+                    erc7730.preload(self.client, outer_def)
+                    self._drop_setup_screenshots()
+                    rejected, _, passes, _ = self._walk(
+                        start, arguments=arguments, altered=(5, bytes(altered)),
+                        chunk_size=chunk_size,
+                        catalog=erc7730.Catalog((outer_def, inner_def)))
+                    assert_failure(self, rejected, types.Failure_SyntaxError,
+                                   "ERC-7730 calldata does not match definition")
+                    self.assertEqual(passes, 5)
+                    self.assertNotIn("Inner field", [s[0] for s in self.screens])
+
+    def test_outer_bytes_changed_during_inner_review_are_refused(self):
+        start, arguments, outer_def, inner_def = self._exec_setup()
+        # The operation word belongs to the outer call, outside the inner
+        # byte range. Inner validation must still bind the whole transaction.
+        altered = bytearray(arguments)
+        altered[127] = 1
+        erc7730.preload(self.client, outer_def)
+        result, _, passes, _ = self._walk(
+            start, arguments=arguments, altered=(5, bytes(altered)),
+            catalog=erc7730.Catalog((outer_def, inner_def)))
+        assert_failure(self, result, types.Failure_SyntaxError,
+                       "ERC-7730 calldata does not match definition")
+        self.assertEqual(passes, 5)
+        self.assertNotIn("Inner field", [s[0] for s in self.screens])
+
+    def test_restoring_outer_definition_rejects_another_valid_definition(self):
+        start, arguments, outer_def, inner_def = self._exec_setup()
+        replacement = erc7730_compiler.compile_calldata(
+            {"display": {"formats": {self.EXEC: {
+                "intent": "Replacement", "fields": []}}}},
+            self.EXEC, 1, ADDRESS)
+        wrong = self._definition(replacement, self._envelope(replacement))
+        # Establish that this is a valid signed definition for the same call;
+        # the failure must come from the frozen outer definition identity.
+        erc7730.preload(self.client, wrong)
+        erc7730.preload(self.client, outer_def)
+        self._drop_setup_screenshots()
+
+        class Substituting(erc7730.Catalog):
+            substituted = False
+
+            def chunk(self, request):
+                if (request.HasField("definition_id") and
+                        not request.HasField("contract_address")):
+                    self.substituted = True
+                    return eth.EthereumClearSignDefinitionChunk(
+                        definition_id=wrong.definition_id, offset=0,
+                        total_length=len(wrong.envelope),
+                        data=wrong.envelope[:request.length])
+                return super().chunk(request)
+
+        catalog = Substituting((outer_def, inner_def))
+        result, _, _, _ = self._walk(start, arguments=arguments,
+                                      catalog=catalog)
+        self.assertTrue(catalog.substituted)
+        self.assertIn(("Inner field", "Amount:\n1.5 USDC"), self._relevant())
+        assert_failure(self, result, types.Failure_SyntaxError,
+                       "Invalid certified ERC-7730 definition")
+        self.assertNotIn(types.ButtonRequest_SignTx, self.button_codes)
+
+    def test_inner_definition_identity_cannot_change_between_chunks(self):
+        start, arguments, outer_def, inner_def = self._exec_setup()
+        erc7730.preload(self.client, outer_def)
+
+        class ChangingIdentity(erc7730.Catalog):
+            changed = False
+
+            def chunk(self, request):
+                response = super().chunk(request)
+                if request.HasField("recursion_depth"):
+                    if request.offset == 0:
+                        response.data = response.data[:31]
+                    else:
+                        self.changed = True
+                        response.definition_id = bytes(32)
+                return response
+
+        catalog = ChangingIdentity((outer_def, inner_def))
+        result, _, _, _ = self._walk(start, arguments=arguments,
+                                      catalog=catalog)
+        self.assertTrue(catalog.changed)
+        assert_failure(self, result, types.Failure_SyntaxError,
+                       "Invalid certified ERC-7730 definition")
+        self.assertNotIn("Inner action", [s[0] for s in self.screens])
+
+    def test_cancelled_inner_review_clears_preload_in_the_same_session(self):
+        for title in ("Inner signer", "Inner action", "Inner field"):
+            with self.subTest(title=title):
+                start, arguments, outer_def, inner_def = self._exec_setup()
+                erc7730.preload(self.client, outer_def)
+                self._drop_setup_screenshots()
+                result, _, _, _ = self._walk(
+                    start, arguments=arguments, cancel_title=title,
+                    catalog=erc7730.Catalog((outer_def, inner_def)))
+                assert_failure(self, result, types.Failure_ActionCancelled,
+                               "Signing cancelled by user")
+                self.assertEqual(self.screens[-1][0], title)
+                self.assertNotIn(types.ButtonRequest_SignTx, self.button_codes)
+                # No Initialize, signer reset or new preload between calls.
+                result, _, _, _ = self._walk(start, arguments=arguments)
+                self.assertIsInstance(result, eth.EthereumTxRequest)
+                self.assertTrue(result.HasField("signature_r"))
+                self.assertEqual(self.definition_requests, 0)
+                self.assertEqual(self._relevant(), [])
+
     # ---- Audit remediation: each of these used to fail mid-review, take a
     # fact from the wrong source, or was untested. ----
 

tests/test_report_variant_validation.py

@@ -99,6 +99,35 @@ def test_complete_release_still_requires_staged_controls(self):
              'test_attestation_does_not_replay_onto_another_transaction',
              'skipped-but-required'), failed)
 
+    def test_stack08_runtime_controls_are_required_on_full_only(self):
+        results = catalog_results_with_solana_lut_skipped('7.15.0')
+        os.environ['KK_RELEASE_MISSING_CAPABILITIES'] = 'solana-lut-attestation'
+        controls = [
+            (module, method) for section, _, _, _, _, tests in REPORT.SECTIONS
+            if section == 'EX' for identifier, module, method, _, _, _ in tests
+            if identifier in ('EX40', 'EX41', 'EX42', 'EX43', 'EX44')]
+        self.assertEqual(5, len(controls))
+        self.assertEqual((True, []),
+                         REPORT.validate_junit('7.15.0', results, 'full'))
+        for module, method in controls:
+            for status in ('missing', 'skip', 'fail'):
+                with self.subTest(method=method, status=status):
+                    changed = dict(results)
+                    key = module + '::' + method
+                    if status == 'missing':
+                        del changed[key]
+                    else:
+                        changed[key] = status
+                    ok, failures = REPORT.validate_junit(
+                        '7.15.0', changed, 'full')
+                    self.assertFalse(ok)
+                    self.assertTrue(any(item[1:3] == (module, method)
+                                        for item in failures))
+            changed = dict(results)
+            changed[module + '::' + method] = 'skip'
+            self.assertEqual((True, []), REPORT.validate_junit(
+                '7.15.0', changed, 'bitcoin-only'))
+
 
 class TestReportVariantEnvironmentIsolation(unittest.TestCase):