Merge pull request #775 from LedgerHQ/feat/apa/web3_v7

Update client to use Web3.py v7

Author: apaillier-ledger

client/pyproject.toml

@@ -16,19 +16,20 @@ readme = { file = "README.md", content-type = "text/markdown" }
 # license = { file = "LICENSE" }
 classifiers = [
     "License :: OSI Approved :: Apache Software License",
-    "Programming Language :: Python :: 3.7",
-    "Programming Language :: Python :: 3.8",
     "Programming Language :: Python :: 3.9",
     "Programming Language :: Python :: 3.10",
+    "Programming Language :: Python :: 3.11",
+    "Programming Language :: Python :: 3.12",
+    "Programming Language :: Python :: 3.13",
     "Operating System :: POSIX :: Linux",
     "Operating System :: Microsoft :: Windows",
     "Operating System :: MacOS :: MacOS X",
 ]
 dynamic = [ "version" ]
-requires-python = ">=3.7"
+requires-python = ">=3.9"
 dependencies = [
     "ragger[speculos]",
-    "web3~=6.0",
+    "web3~=7.0",
 ]
 
 [tools.setuptools]

client/src/ledger_app_clients/ethereum/client.py

@@ -1,6 +1,6 @@
 import struct
 from enum import IntEnum
-from typing import Optional, Tuple
+from typing import Optional
 from hashlib import sha256
 import rlp
 from web3 import Web3
@@ -213,13 +213,10 @@ def eip712_filtering_trusted_name(self,
     def eip712_filtering_raw(self, name: str, sig: bytes, discarded: bool):
         return self._exchange_async(self._cmd_builder.eip712_filtering_raw(name, sig, discarded))
 
-    def serialize_tx(self, tx_params: dict, tx_raw: Optional[bytes] = None) -> Tuple[bytes, bytes]:
+    def serialize_tx(self, tx_params: dict) -> tuple[bytes, bytes]:
         """Computes the serialized TX and its hash"""
 
-        if tx_raw is not None:
-            tx = tx_raw
-        else:
-            tx = Web3().eth.account.create().sign_transaction(tx_params).rawTransaction
+        tx = Web3().eth.account.create().sign_transaction(tx_params).raw_transaction
         prefix = bytes()
         suffix = []
         if tx[0] in [0x01, 0x02, 0x04]:
@@ -228,20 +225,16 @@ def serialize_tx(self, tx_params: dict, tx_raw: Optional[bytes] = None) -> Tuple
         else:  # legacy
             if "chainId" in tx_params:
                 suffix = [int(tx_params["chainId"]), bytes(), bytes()]
-        if tx_raw is None:
-            decoded_tx = rlp.decode(tx)[:-3]  # remove already computed signature
-        else:
-            decoded_tx = rlp.decode(tx)
+        decoded_tx = rlp.decode(tx)[:-3]  # remove already computed signature
         encoded_tx = prefix + rlp.encode(decoded_tx + suffix)
         tx_hash = keccak(encoded_tx)
         return encoded_tx, tx_hash
 
     def sign(self,
              bip32_path: str,
              tx_params: dict,
-             tx_raw: Optional[bytes] = None,  # Introduced for 7702 until web3.py supports authorization lists
              mode: SignMode = SignMode.BASIC):
-        tx, _ = self.serialize_tx(tx_params, tx_raw)
+        tx, _ = self.serialize_tx(tx_params)
         chunks = self._cmd_builder.sign(bip32_path, tx, mode)
         for chunk in chunks[:-1]:
             self._exchange(chunk)
@@ -338,7 +331,7 @@ def provide_trusted_name_v2(self,
                                 chain_id: int,
                                 nft_id: Optional[int] = None,
                                 challenge: Optional[int] = None,
-                                not_valid_after: Optional[Tuple[int]] = None) -> RAPDU:
+                                not_valid_after: Optional[tuple[int, int, int]] = None) -> RAPDU:
         payload = format_tlv(FieldTag.STRUCT_VERSION, 2)
         payload += format_tlv(FieldTag.TRUSTED_NAME, name)
         payload += format_tlv(FieldTag.ADDRESS, addr)

client/src/ledger_app_clients/ethereum/response_parser.py

@@ -1,11 +1,11 @@
-def signature(data: bytes) -> tuple[bytes, bytes, bytes]:
+def signature(data: bytes) -> tuple[int, int, int]:
     assert len(data) == (1 + 32 + 32)
 
-    v = data[0:1]
+    v = int.from_bytes(data[0:1], "big")
     data = data[1:]
-    r = data[0:32]
+    r = int.from_bytes(data[0:32], "big")
     data = data[32:]
-    s = data[0:32]
+    s = int.from_bytes(data[0:32], "big")
 
     return v, r, s
 

client/src/ledger_app_clients/ethereum/utils.py

@@ -1,36 +1,27 @@
-from typing import Optional
 from eth_account import Account
 from eth_account.messages import encode_defunct, encode_typed_data
+from eth_account.datastructures import SignedSetCodeAuthorization
+from eth_account.typed_transactions.set_code_transaction import Authorization
+from eth_keys.datatypes import Signature
 import rlp
 
 
-# eth_account requires it for some reason
-def normalize_vrs(vrs: tuple) -> tuple:
-    vrs_l = []
-    for elem in vrs:
-        vrs_l.append(elem.lstrip(b'\x00'))
-    return tuple(vrs_l)
-
-
 def get_selector_from_data(data: str) -> bytes:
     raw_data = bytes.fromhex(data[2:])
     return raw_data[:4]
 
 
-def recover_message(msg, vrs: tuple) -> bytes:
+def recover_message(msg, vrs: tuple[int, int, int]) -> bytes:
     if isinstance(msg, dict):  # EIP-712
         smsg = encode_typed_data(full_message=msg)
     else:  # EIP-191
         smsg = encode_defunct(primitive=msg)
-    addr = Account.recover_message(smsg, normalize_vrs(vrs))
+    addr = Account.recover_message(smsg, vrs)
     return bytes.fromhex(addr[2:])
 
 
-def recover_transaction(tx_params, vrs: tuple, raw_tx_param: Optional[bytes] = None) -> bytes:
-    if raw_tx_param is None:
-        raw_tx = Account.create().sign_transaction(tx_params).rawTransaction
-    else:
-        raw_tx = raw_tx_param
+def recover_transaction(tx_params, vrs: tuple[int, int, int]) -> bytes:
+    raw_tx = Account.create().sign_transaction(tx_params).raw_transaction
     prefix = bytes()
     if raw_tx[0] in [0x01, 0x02, 0x04]:
         prefix = raw_tx[:1]
@@ -44,7 +35,7 @@ def recover_transaction(tx_params, vrs: tuple, raw_tx_param: Optional[bytes] = N
                 trunc_chain_id >>= 8
 
             trunc_target = trunc_chain_id * 2 + 35
-            trunc_v = int.from_bytes(vrs[0], "big")
+            trunc_v = vrs[0]
 
             if (trunc_target & 0xff) == trunc_v:
                 parity = 0
@@ -56,15 +47,35 @@ def recover_transaction(tx_params, vrs: tuple, raw_tx_param: Optional[bytes] = N
 
             # https://github.com/ethereum/EIPs/blob/master/EIPS/eip-155.md
             full_v = parity + tx_params["chainId"] * 2 + 35
-            # 9 bytes would be big enough even for the biggest chain ID
-            vrs = (int(full_v).to_bytes(9, "big"), vrs[1], vrs[2])
+            vrs = (full_v, vrs[1], vrs[2])
         else:
             # Pre EIP-155 TX
             assert False
     decoded = rlp.decode(raw_tx)
-    if raw_tx_param is None:
-        reencoded = rlp.encode(decoded[:-3] + list(normalize_vrs(vrs)))
-    else:
-        reencoded = rlp.encode(decoded + list(normalize_vrs(vrs)))
+    reencoded = rlp.encode(decoded[:-3] + list(vrs))
     addr = Account.recover_transaction(prefix + reencoded)
     return bytes.fromhex(addr[2:])
+
+
+# Code inspired by :
+# https://github.com/ethereum/eth-account/blob/a1ba20c9a112d3534ac3296f21f51e2f5127bf9b/eth_account/account.py#L1057
+def get_authorization_obj(chain_id: int,
+                          nonce: int,
+                          address: bytes,
+                          vrs: tuple[int, int, int]) -> SignedSetCodeAuthorization:
+    unsigned_authorization = Authorization(chain_id, address, nonce)
+    sig = Signature(vrs=vrs)
+    return SignedSetCodeAuthorization(
+        chain_id=chain_id,
+        address=address,
+        nonce=nonce,
+        y_parity=sig.v,
+        r=sig.r,
+        s=sig.s,
+        signature=sig,
+        authorization_hash=unsigned_authorization.hash(),
+    )
+
+
+def recover_authorization(chain_id: int, nonce: int, address: bytes, vrs: tuple[int, int, int]) -> bytes:
+    return get_authorization_obj(chain_id, nonce, address, vrs).authority

tests/ragger/requirements.txt

@@ -1,5 +1,5 @@
 pytest
 ecdsa
-web3~=6.0
+web3~=7.0
 ragger[speculos]
 py_ecc

tests/ragger/test_eip7702.py

@@ -1,5 +1,6 @@
 import pytest
 
+from typing import Optional
 from ragger.error import ExceptionRAPDU
 from ragger.backend import BackendInterface
 from ragger.firmware import Firmware
@@ -11,6 +12,8 @@
 import client.response_parser as ResponseParser
 from client.tx_auth_7702 import TxAuth7702
 
+from client.utils import recover_authorization
+
 BIP32_PATH = "m/44'/60'/0'/0/0"
 TEST_ADDRESS_0 = bytes.fromhex("00" * 20)
 TEST_ADDRESS_1 = bytes.fromhex("01" * 20)
@@ -24,6 +27,8 @@
 NONCE = 1337
 NONCE_MAX = 0xFFFFFFFFFFFFFFFF
 
+DEVICE_ADDR: Optional[bytes] = None
+
 # Test vectors computed with
 # cast wallet sign-auth $ADDRESS --mnemonic $MNEMONIC --mnemonic-derivation-path "m/44'/60'/0'/0/0" --nonce $NONCE --chain $CHAINID
 # Decoded by https://codechain-io.github.io/rlp-debugger/
@@ -35,23 +40,25 @@ def common(firmware: Firmware,
            test_name: str,
            delegate: bytes,
            nonce: int,
-           chain_id: int,
-           v: bytes = None,
-           r: bytes = None,
-           s: bytes = None):
+           chain_id: int):
 
+    global DEVICE_ADDR
     app_client = EthAppClient(backend)
 
     if firmware.is_nano:
         end_text = "Accept"
     else:
         end_text = ".*Sign.*"
 
+    if DEVICE_ADDR is None:
+        with app_client.get_public_addr(bip32_path=BIP32_PATH, display=False):
+            pass
+        _, DEVICE_ADDR, _ = ResponseParser.pk_addr(app_client.response().data)
     auth_params = TxAuth7702(delegate, nonce, chain_id)
     with app_client.sign_eip7702_authorization(BIP32_PATH, auth_params):
         scenario.review_approve(test_name=test_name, custom_screen_text=end_text)
     vrs = ResponseParser.signature(app_client.response().data)
-    assert vrs == (v, r, s)
+    assert recover_authorization(chain_id, nonce, delegate, vrs) == DEVICE_ADDR
 
 
 def common_rejected(firmware: Firmware,
@@ -95,10 +102,7 @@ def test_eip7702_in_whitelist(firmware: Firmware,
            test_name,
            TEST_ADDRESS_1,
            NONCE,
-           CHAIN_ID_1,
-           bytes.fromhex("00"),
-           bytes.fromhex("f82e 50a7 55fa 989f 4bb9 b36b 15af b442 4ce9 cda6 9752 fd17 a7eb 1473 7d96 3e62"),
-           bytes.fromhex("07c9 c91d 6140 b45e a52f 29de 7a5e ffb9 dd34 0607 a26e 225c 4027 8e91 c405 4492"))
+           CHAIN_ID_1)
 
 
 def test_eip7702_in_whitelist_all_chain_whitelisted(firmware: Firmware,
@@ -114,10 +118,7 @@ def test_eip7702_in_whitelist_all_chain_whitelisted(firmware: Firmware,
            test_name,
            TEST_ADDRESS_0,
            NONCE,
-           CHAIN_ID_2,
-           bytes.fromhex("00"),
-           bytes.fromhex("0378 f7ac 482e c728 b65d 19d0 3943 bbb3 fe73 07c7 2c64 6e7d 2d0c 11be e81e b2b9"),
-           bytes.fromhex("3322 66ec 3ef9 96bf 835c 50a8 3300 6b4c 8039 8d59 7d0e 6846 19db 4d51 a384 a38d"))
+           CHAIN_ID_2)
 
 
 def test_eip7702_in_whitelist_all_chain_param(firmware: Firmware,
@@ -133,10 +134,7 @@ def test_eip7702_in_whitelist_all_chain_param(firmware: Firmware,
            test_name,
            TEST_ADDRESS_2,
            NONCE,
-           CHAIN_ID_0,
-           bytes.fromhex("01"),
-           bytes.fromhex("a24f 35ca fc6b 408c e325 39d4 bd89 a67e dd4d 6303 fc67 6dfd df93 b984 05b7 ee5e"),
-           bytes.fromhex("1594 56ba be65 6692 959c a3d8 29ca 269e 8f82 387c 91e4 0a33 633d 190d da7a 3c5c"))
+           CHAIN_ID_0)
 
 
 def test_eip7702_in_whitelist_max(firmware: Firmware,
@@ -152,10 +150,7 @@ def test_eip7702_in_whitelist_max(firmware: Firmware,
            test_name,
            TEST_ADDRESS_MAX,
            NONCE_MAX,
-           CHAIN_ID_MAX,
-           bytes.fromhex("00"),
-           bytes.fromhex("a07c 6808 9449 8c21 e80f ebb0 11ae 5d62 ede6 645d 77d7 c902 db06 5d5a 082d d220"),
-           bytes.fromhex("6b5c 6222 5cf6 5a62 40c2 583d acc1 641c 8d69 2ed3 bd00 473c c3e2 8fe1 a4f5 2294"))
+           CHAIN_ID_MAX)
 
 
 def test_eip7702_in_whitelist_wrong_chain(firmware: Firmware,