Merge pull request #527 from LedgerHQ/feat/apa/eip712_address_substitution

EIP-712 address substitution

client/CHANGELOG.md

@@ -5,6 +5,16 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.3.0] - 2024-02-13
+
+### Added
+
+- New `provide_token_metadata` function
+
+### Fixed
+
+- Increased the delay between `autonext` callback calls for EIP-712 on Stax
+
 ## [0.2.1] - 2023-12-01
 
 ### Fixed

client/src/ledger_app_clients/ethereum/client.py

@@ -231,3 +231,25 @@ class EthAppClient:
             with self._send(chunk):
                 pass
         return self._send(chunks[-1])
+
+    def provide_token_metadata(self,
+                               ticker: str,
+                               addr: bytes,
+                               decimals: int,
+                               chain_id: int,
+                               sig: Optional[bytes] = None):
+        if sig is None:
+            # Temporarily get a command with an empty signature to extract the payload and
+            # compute the signature on it
+            tmp = self._cmd_builder.provide_erc20_token_information(ticker,
+                                                                    addr,
+                                                                    decimals,
+                                                                    chain_id,
+                                                                    bytes())
+            # skip APDU header & empty sig
+            sig = sign_data(Key.CAL, tmp[6:])
+        return self._send(self._cmd_builder.provide_erc20_token_information(ticker,
+                                                                            addr,
+                                                                            decimals,
+                                                                            chain_id,
+                                                                            sig))

client/src/ledger_app_clients/ethereum/command_builder.py

@@ -13,6 +13,7 @@ class InsType(IntEnum):
     GET_PUBLIC_ADDR = 0x02
     SIGN = 0x04
     PERSONAL_SIGN = 0x08
+    PROVIDE_ERC20_TOKEN_INFORMATION = 0x0a
     PROVIDE_NFT_INFORMATION = 0x14
     SET_PLUGIN = 0x16
     EIP712_SEND_STRUCT_DEF = 0x1a
@@ -310,3 +311,21 @@ class CommandBuilder:
             payload = payload[chunk_size:]
             p1 = P1Type.SIGN_SUBSQT_CHUNK
         return chunks
+
+    def provide_erc20_token_information(self,
+                                        ticker: str,
+                                        addr: bytes,
+                                        decimals: int,
+                                        chain_id: int,
+                                        sig: bytes) -> bytes:
+        payload = bytearray()
+        payload.append(len(ticker))
+        payload += ticker.encode()
+        payload += addr
+        payload += struct.pack(">I", decimals)
+        payload += struct.pack(">I", chain_id)
+        payload += sig
+        return self._serialize(InsType.PROVIDE_ERC20_TOKEN_INFORMATION,
+                               0x00,
+                               0x00,
+                               payload)

client/src/ledger_app_clients/ethereum/eip712/InputData.py

@@ -337,12 +337,11 @@ def next_timeout(_signum: int, _frame):
 
 
 def enable_autonext():
-    seconds = 1/4
     if app_client._client.firmware.device == 'stax':  # Stax Speculos is slow
-        interval = seconds * 3
+        delay = 1.5
     else:
-        interval = seconds
-    signal.setitimer(signal.ITIMER_REAL, seconds, interval)
+        delay = 1/4
+    signal.setitimer(signal.ITIMER_REAL, delay, delay)
 
 
 def disable_autonext():

src_features/signMessageEIP712/ui_logic.c

@@ -16,6 +16,7 @@
 #include "typed_data.h"
 #include "commands_712.h"
 #include "common_ui.h"
+#include "domain_name.h"
 
 static t_ui_context *ui_ctx = NULL;
 
@@ -184,6 +185,46 @@ static void ui_712_format_str(const uint8_t *const data, uint8_t length) {
     }
 }
 
+/**
+ * Find a substitute token ticker for a given address
+ *
+ * @param[in] addr the given address
+ * @return the ticker name if found, \ref NULL otherwise
+ */
+static const char *get_address_token_ticker(const uint8_t *addr) {
+    tokenDefinition_t *token;
+
+    // Loop over the received token informations
+    for (uint8_t token_idx = 0; token_idx < MAX_ITEMS; ++token_idx) {
+        if (tmpCtx.transactionContext.tokenSet[token_idx] == 1) {
+            token = &tmpCtx.transactionContext.extraInfo[token_idx].token;
+            if (memcmp(token->address, addr, ADDRESS_LENGTH) == 0) {
+                return token->ticker;
+            }
+        }
+    }
+    return NULL;
+}
+
+/**
+ * Find a substitute (token ticker or domain name) for a given address
+ *
+ * @param[in] addr the given address
+ * @return the substitute if found, \ref NULL otherwise
+ */
+static const char *get_address_substitute(const uint8_t *addr) {
+    const char *str = NULL;
+
+    str = get_address_token_ticker(addr);
+    if (str == NULL) {
+        if (has_domain_name(&eip712_context->chain_id, addr)) {
+            // No handling of the verbose domains setting
+            str = g_domain_name;
+        }
+    }
+    return str;
+}
+
 /**
  * Format a given data as a string representation of an address
  *
@@ -196,13 +237,20 @@ static bool ui_712_format_addr(const uint8_t *const data, uint8_t length) {
         apdu_response_code = APDU_RESPONSE_INVALID_DATA;
         return false;
     }
+
     if (ui_712_field_shown()) {
-        if (!getEthDisplayableAddress((uint8_t *) data,
-                                      strings.tmp.tmp,
-                                      sizeof(strings.tmp.tmp),
-                                      &global_sha3,
-                                      chainConfig->chainId)) {
-            THROW(APDU_RESPONSE_ERROR_NO_INFO);
+        const char *sub;
+
+        if (!N_storage.verbose_eip712 && ((sub = get_address_substitute(data)) != NULL)) {
+            ui_712_set_value(sub, strlen(sub));
+        } else {
+            if (!getEthDisplayableAddress((uint8_t *) data,
+                                          strings.tmp.tmp,
+                                          sizeof(strings.tmp.tmp),
+                                          &global_sha3,
+                                          chainConfig->chainId)) {
+                THROW(APDU_RESPONSE_ERROR_NO_INFO);
+            }
         }
     }
     return true;

tests/ragger/constants.py

@@ -0,0 +1,3 @@
+from pathlib import Path
+
+ROOT_SNAPSHOT_PATH = Path(__file__).parent

tests/ragger/eip712_input_files/address_substitution.json

@@ -0,0 +1,29 @@
+{
+    "domain": {
+        "chainId": 1,
+        "name": "Token test",
+        "verifyingContract": "0xCcCCccccCCCCcCCCCCCcCcCccCcCCCcCcccccccC",
+        "version": "1"
+    },
+    "message": {
+        "from": "0xaAaAaAaaAaAaAaaAaAAAAAAAAaaaAaAaAaaAaaAa",
+        "to": "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045",
+        "amount": "117",
+        "token": "0x6B175474E89094C44Da98b954EedeAC495271d0F"
+    },
+    "primaryType": "Transfer",
+    "types": {
+        "EIP712Domain": [
+            { "name": "name", "type": "string" },
+            { "name": "version", "type": "string" },
+            { "name": "chainId", "type": "uint256" },
+            { "name": "verifyingContract", "type": "address" }
+        ],
+        "Transfer": [
+            { "name": "from", "type": "address" },
+            { "name": "to", "type": "address" },
+            { "name": "amount", "type": "uint256" },
+            { "name": "token", "type": "address" }
+        ]
+    }
+}

tests/ragger/test_domain_name.py

@@ -1,9 +1,9 @@
 import pytest
-from pathlib import Path
 from ragger.backend import BackendInterface
 from ragger.firmware import Firmware
 from ragger.error import ExceptionRAPDU
 from ragger.navigator import Navigator, NavInsID
+from constants import ROOT_SNAPSHOT_PATH
 
 import ledger_app_clients.ethereum.response_parser as ResponseParser
 from ledger_app_clients.ethereum.client import EthAppClient, StatusWord
@@ -12,8 +12,6 @@ from ledger_app_clients.ethereum.settings import SettingID, settings_toggle
 from web3 import Web3
 
 
-ROOT_SCREENSHOT_PATH = Path(__file__).parent
-
 # Values used across all tests
 CHAIN_ID = 1
 NAME = "ledger.eth"
@@ -74,7 +72,7 @@ def test_send_fund(firmware: Firmware,
             if verbose:
                 moves += [NavInsID.USE_CASE_REVIEW_TAP]
             moves += [NavInsID.USE_CASE_REVIEW_CONFIRM]
-        navigator.navigate_and_compare(ROOT_SCREENSHOT_PATH,
+        navigator.navigate_and_compare(ROOT_SNAPSHOT_PATH,
                                        "domain_name_verbose_" + str(verbose),
                                        moves)
 
@@ -123,7 +121,7 @@ def test_send_fund_wrong_addr(firmware: Firmware,
         else:
             moves += [NavInsID.USE_CASE_REVIEW_TAP] * 2
             moves += [NavInsID.USE_CASE_REVIEW_CONFIRM]
-        navigator.navigate_and_compare(ROOT_SCREENSHOT_PATH,
+        navigator.navigate_and_compare(ROOT_SNAPSHOT_PATH,
                                        "domain_name_wrong_addr",
                                        moves)
 
@@ -154,7 +152,7 @@ def test_send_fund_non_mainnet(firmware: Firmware,
         else:
             moves += [NavInsID.USE_CASE_REVIEW_TAP] * 2
             moves += [NavInsID.USE_CASE_REVIEW_CONFIRM]
-        navigator.navigate_and_compare(ROOT_SCREENSHOT_PATH,
+        navigator.navigate_and_compare(ROOT_SNAPSHOT_PATH,
                                        "domain_name_non_mainnet",
                                        moves)
 
@@ -185,7 +183,7 @@ def test_send_fund_unknown_chain(firmware: Firmware,
         else:
             moves += [NavInsID.USE_CASE_REVIEW_TAP] * 3
             moves += [NavInsID.USE_CASE_REVIEW_CONFIRM]
-        navigator.navigate_and_compare(ROOT_SCREENSHOT_PATH,
+        navigator.navigate_and_compare(ROOT_SNAPSHOT_PATH,
                                        "domain_name_unknown_chain",
                                        moves)
 

tests/ragger/test_eip712.py

@@ -9,6 +9,8 @@ from ragger.backend import BackendInterface
 from ragger.firmware import Firmware
 from ragger.navigator import Navigator, NavInsID
 import json
+from typing import Optional
+from constants import ROOT_SNAPSHOT_PATH
 
 import ledger_app_clients.ethereum.response_parser as ResponseParser
 from ledger_app_clients.ethereum.client import EthAppClient
@@ -16,12 +18,26 @@ from ledger_app_clients.ethereum.eip712 import InputData
 from ledger_app_clients.ethereum.settings import SettingID, settings_toggle
 
 
+class SnapshotsConfig:
+    test_name: str
+    idx: int
+
+    def __init__(self, test_name: str, idx: int = 0):
+        self.test_name = test_name
+        self.idx = idx
+
+
 BIP32_PATH = "m/44'/60'/0'/0/0"
+snaps_config: Optional[SnapshotsConfig] = None
+
+
+def eip712_json_path() -> str:
+    return "%s/eip712_input_files" % (os.path.dirname(__file__))
 
 
 def input_files() -> list[str]:
     files = []
-    for file in os.scandir("%s/eip712_input_files" % (os.path.dirname(__file__))):
+    for file in os.scandir(eip712_json_path()):
         if fnmatch.fnmatch(file, "*-data.json"):
             files.append(file.path)
     return sorted(files)
@@ -77,7 +93,52 @@ def autonext(fw: Firmware, nav: Navigator):
         moves = [NavInsID.RIGHT_CLICK]
     else:
         moves = [NavInsID.USE_CASE_REVIEW_TAP]
-    nav.navigate(moves, screen_change_before_first_instruction=False, screen_change_after_last_instruction=False)
+    if snaps_config is not None:
+        nav.navigate_and_compare(ROOT_SNAPSHOT_PATH,
+                                 snaps_config.test_name,
+                                 moves,
+                                 screen_change_before_first_instruction=False,
+                                 screen_change_after_last_instruction=False,
+                                 snap_start_idx=snaps_config.idx)
+        snaps_config.idx += 1
+    else:
+        nav.navigate(moves,
+                     screen_change_before_first_instruction=False,
+                     screen_change_after_last_instruction=False)
+
+
+def eip712_new_common(fw: Firmware,
+                      nav: Navigator,
+                      app_client: EthAppClient,
+                      json_data: dict,
+                      filters: Optional[dict],
+                      verbose: bool):
+    assert InputData.process_data(app_client,
+                                  json_data,
+                                  filters,
+                                  partial(autonext, fw, nav))
+    with app_client.eip712_sign_new(BIP32_PATH):
+        moves = list()
+        if fw.device.startswith("nano"):
+            # need to skip the message hash
+            if not verbose and filters is None:
+                moves = [NavInsID.RIGHT_CLICK] * 2
+            moves += [NavInsID.BOTH_CLICK]
+        else:
+            time.sleep(1.5)
+            # need to skip the message hash
+            if not verbose and filters is None:
+                moves += [NavInsID.USE_CASE_REVIEW_TAP]
+            moves += [NavInsID.USE_CASE_REVIEW_CONFIRM]
+        if snaps_config is not None:
+            nav.navigate_and_compare(ROOT_SNAPSHOT_PATH,
+                                     snaps_config.test_name,
+                                     moves,
+                                     snap_start_idx=snaps_config.idx)
+            snaps_config.idx += 1
+        else:
+            nav.navigate(moves)
+    return ResponseParser.signature(app_client.response().data)
 
 
 def test_eip712_new(firmware: Firmware,
@@ -114,26 +175,69 @@ def test_eip712_new(firmware: Firmware,
             settings_toggle(firmware, navigator, [SettingID.VERBOSE_EIP712])
 
         with open(input_file) as file:
-            assert InputData.process_data(app_client,
-                                          json.load(file),
-                                          filters,
-                                          partial(autonext, firmware, navigator))
-            with app_client.eip712_sign_new(BIP32_PATH):
-                # tight on timing, needed by the CI otherwise might fail sometimes
-                time.sleep(0.5)
+            v, r, s = eip712_new_common(firmware,
+                                        navigator,
+                                        app_client,
+                                        json.load(file),
+                                        filters,
+                                        verbose)
 
-                moves = list()
-                if firmware.device.startswith("nano"):
-                    if not verbose and not filtering:  # need to skip the message hash
-                        moves = [NavInsID.RIGHT_CLICK] * 2
-                    moves += [NavInsID.BOTH_CLICK]
-                else:
-                    if not verbose and not filtering:  # need to skip the message hash
-                        moves += [NavInsID.USE_CASE_REVIEW_TAP]
-                    moves += [NavInsID.USE_CASE_REVIEW_CONFIRM]
-                navigator.navigate(moves)
-            v, r, s = ResponseParser.signature(app_client.response().data)
+    assert v == bytes.fromhex(config["signature"]["v"])
+    assert r == bytes.fromhex(config["signature"]["r"])
+    assert s == bytes.fromhex(config["signature"]["s"])
 
-        assert v == bytes.fromhex(config["signature"]["v"])
-        assert r == bytes.fromhex(config["signature"]["r"])
-        assert s == bytes.fromhex(config["signature"]["s"])
+
+def test_eip712_address_substitution(firmware: Firmware,
+                                     backend: BackendInterface,
+                                     navigator: Navigator,
+                                     verbose: bool):
+    global snaps_config
+
+    app_client = EthAppClient(backend)
+    if firmware.device == "nanos":
+        pytest.skip("Not supported on LNS")
+    else:
+        test_name = "eip712_address_substitution"
+        if verbose:
+            test_name += "_verbose"
+        snaps_config = SnapshotsConfig(test_name)
+        with open("%s/address_substitution.json" % (eip712_json_path())) as file:
+            data = json.load(file)
+
+            with app_client.provide_token_metadata("DAI",
+                                                   bytes.fromhex(data["message"]["token"][2:]),
+                                                   18,
+                                                   1):
+                pass
+
+            with app_client.get_challenge():
+                pass
+            challenge = ResponseParser.challenge(app_client.response().data)
+            with app_client.provide_domain_name(challenge,
+                                                "vitalik.eth",
+                                                bytes.fromhex(data["message"]["to"][2:])):
+                pass
+
+            if verbose:
+                settings_toggle(firmware, navigator, [SettingID.VERBOSE_EIP712])
+                filters = None
+            else:
+                filters = {
+                    "name": "Token test",
+                    "fields": {
+                        "amount": "Amount",
+                        "token": "Token",
+                        "to": "To",
+                    }
+                }
+
+            v, r, s = eip712_new_common(firmware,
+                                        navigator,
+                                        app_client,
+                                        data,
+                                        filters,
+                                        verbose)
+
+    assert v == bytes.fromhex("1b")
+    assert r == bytes.fromhex("d4a0e058251cdc3845aaa5eb8409d8a189ac668db7c55a64eb3121b0db7fd8c0")
+    assert s == bytes.fromhex("3221800e4f45272c6fa8fafda5e94c848d1a4b90c442aa62afa8e8d6a9af0f00")

tests/ragger/test_get_address.py

@@ -1,6 +1,5 @@
 import pytest
 from typing import Optional
-from pathlib import Path
 from ragger.error import ExceptionRAPDU
 from ragger.firmware import Firmware
 from ragger.backend import BackendInterface
@@ -8,8 +7,7 @@ from ragger.navigator import Navigator, NavInsID
 from ledger_app_clients.ethereum.client import EthAppClient, StatusWord
 import ledger_app_clients.ethereum.response_parser as ResponseParser
 from ragger.bip import calculate_public_key_and_chaincode, CurveChoice
-
-ROOT_SCREENSHOT_PATH = Path(__file__).parent
+from constants import ROOT_SNAPSHOT_PATH
 
 
 @pytest.fixture(params=[True, False])
@@ -56,7 +54,7 @@ def test_get_pk_rejected(firmware: Firmware,
 
     try:
         with app_client.get_public_addr():
-            navigator.navigate_and_compare(ROOT_SCREENSHOT_PATH,
+            navigator.navigate_and_compare(ROOT_SNAPSHOT_PATH,
                                            "get_pk_rejected",
                                            get_moves(firmware, navigator, reject=True))
     except ExceptionRAPDU as e:
@@ -73,7 +71,7 @@ def test_get_pk(firmware: Firmware,
     app_client = EthAppClient(backend)
 
     with app_client.get_public_addr(chaincode=with_chaincode, chain_id=chain):
-        navigator.navigate_and_compare(ROOT_SCREENSHOT_PATH,
+        navigator.navigate_and_compare(ROOT_SNAPSHOT_PATH,
                                        "get_pk_%s" % (chain),
                                        get_moves(firmware, navigator, chain=chain))
     pk, addr, chaincode = ResponseParser.pk_addr(app_client.response().data, with_chaincode)

tests/ragger/test_nft.py

@@ -12,9 +12,9 @@ from ledger_app_clients.ethereum.utils import get_selector_from_data, recover_tr
 from web3 import Web3
 import json
 import os
+from constants import ROOT_SNAPSHOT_PATH
 
 
-ROOT_SCREENSHOT_PATH = Path(__file__).parent
 ABIS_FOLDER = "%s/abis" % (os.path.dirname(__file__))
 
 BIP32_PATH = "m/44'/60'/0'/0/0"
@@ -116,7 +116,7 @@ def common_test_nft(fw: Firmware,
         "data": data,
     }
     with app_client.sign(BIP32_PATH, tx_params):
-        nav.navigate_and_compare(ROOT_SCREENSHOT_PATH,
+        nav.navigate_and_compare(ROOT_SNAPSHOT_PATH,
                                  snapshot_test_name(plugin_name.lower(),
                                                     action.fn_name,
                                                     collec.chain_id,