fix: transaction signing issues [APE-1254] (#39)

ape_ledger/_cli.py

@@ -1,10 +1,11 @@
-from typing import List
+from typing import List, Tuple, Union
 
 import click
 from ape import accounts
 from ape.cli import (
     ape_cli_context,
     existing_alias_argument,
+    network_option,
     non_existing_alias_argument,
     skip_confirmation_option,
 )
@@ -13,9 +14,13 @@
 
 from ape_ledger.accounts import LedgerAccount
 from ape_ledger.choices import AddressPromptChoice
-from ape_ledger.client import connect_to_ethereum_app
 from ape_ledger.exceptions import LedgerSigningError
-from ape_ledger.hdpath import HDBasePath
+from ape_ledger.hdpath import HDAccountPath, HDBasePath
+
+
+def _select_account(hd_path: Union[HDBasePath, str]) -> Tuple[str, HDAccountPath]:
+    choices = AddressPromptChoice(hd_path)
+    return choices.get_user_selected_account()
 
 
 @click.group(short_help="Manage Ledger accounts")
@@ -36,7 +41,7 @@ def _list(cli_ctx):
         cli_ctx.logger.warning("No accounts found.")
         return
 
-    num_accounts = len(accounts)
+    num_accounts = len(ledger_accounts)
     header = f"Found {num_accounts} account"
     header += "s:" if num_accounts > 1 else ":"
     click.echo(header)
@@ -66,9 +71,7 @@ def _get_ledger_accounts() -> List[LedgerAccount]:
 def add(cli_ctx, alias, hd_path):
     """Add an account from your Ledger hardware wallet"""
 
-    app = connect_to_ethereum_app(hd_path)
-    choices = AddressPromptChoice(app)
-    address, account_hd_path = choices.get_user_selected_account()
+    address, account_hd_path = _select_account(hd_path)
     container = accounts.containers.get("ledger")
     container.save_account(alias, address, str(account_hd_path))
     cli_ctx.logger.success(f"Account '{address}' successfully added with alias '{alias}'.")
@@ -111,9 +114,23 @@ def delete_all(cli_ctx, skip_confirmation):
 @ape_cli_context()
 @click.argument("alias")
 @click.argument("message", default="Hello World!")
-def sign_message(cli_ctx, alias, message):
+@network_option()
+def sign_message(cli_ctx, alias, message, network):
     """Sign a message using a Ledger account"""
 
+    ctx = None
+    if network:
+        ctx = cli_ctx.network_manager.parse_network_choice(network)
+        ctx.__enter__()
+
+    try:
+        _sign_message(cli_ctx, alias, message)
+    finally:
+        if network and ctx and ctx._provider and ctx._provider.is_connected:
+            ctx.__exit__()
+
+
+def _sign_message(cli_ctx, alias, message):
     if alias not in accounts.aliases:
         cli_ctx.abort(f"Account with alias '{alias}' does not exist.")
 

ape_ledger/accounts.py

@@ -1,19 +1,31 @@
 import json
 from pathlib import Path
-from typing import Iterator, Optional, Union
+from typing import Dict, Iterator, Optional, Union
 
 import click
-import rlp  # type: ignore
 from ape.api import AccountAPI, AccountContainerAPI, TransactionAPI
 from ape.types import AddressType, MessageSignature, TransactionSignature
-from ape_ethereum.transactions import TransactionType
+from ape_ethereum.transactions import DynamicFeeTransaction, StaticFeeTransaction
 from eth_account.messages import SignableMessage
+from eth_utils import is_0x_prefixed, to_bytes
 from hexbytes import HexBytes
 
-from ape_ledger.client import LedgerEthereumAccountClient, connect_to_ethereum_account
+from ape_ledger.client import LedgerDeviceClient, get_device
 from ape_ledger.exceptions import LedgerSigningError
 from ape_ledger.hdpath import HDAccountPath
-from ape_ledger.objects import DynamicFeeTransaction, StaticFeeTransaction
+
+
+def _to_bytes(val):
+    if val is None:
+        return b""
+    elif isinstance(val, str) and is_0x_prefixed(val):
+        return to_bytes(hexstr=val)
+    elif isinstance(val, str):
+        return to_bytes(text=val)
+    elif isinstance(val, HexBytes):
+        return bytes(val)
+    else:
+        return to_bytes(val)
 
 
 class AccountContainer(AccountContainerAPI):
@@ -62,13 +74,14 @@ def _echo_object_to_sign(obj: Union[TransactionAPI, SignableMessage]):
 class LedgerAccount(AccountAPI):
     account_file_path: Path
 
-    # Optional because it's lazily loaded
-    account_client: Optional[LedgerEthereumAccountClient] = None
-
     @property
     def alias(self) -> str:
         return self.account_file_path.stem
 
+    @property
+    def _client(self) -> LedgerDeviceClient:
+        return get_device(self.hdpath)
+
     @property
     def address(self) -> AddressType:
         ecosystem = self.network_manager.get_ecosystem("ethereum")
@@ -83,46 +96,51 @@ def hdpath(self) -> HDAccountPath:
     def account_file(self) -> dict:
         return json.loads(self.account_file_path.read_text())
 
-    @property
-    def _client(self) -> LedgerEthereumAccountClient:
-        if self.account_client is None:
-            self.account_client = connect_to_ethereum_account(self.address, self.hdpath)
-        return self.account_client
-
     def sign_message(self, msg: SignableMessage) -> Optional[MessageSignature]:
         version = msg.version
         if version == b"E":
             _echo_object_to_sign(msg)
-            signed_msg = self._client.sign_personal_message(msg.body)
+            signed_msg = self._client.sign_message(msg.body)
         elif version == b"\x01":
             _echo_object_to_sign(msg)
-            signed_msg = self._client.sign_typed_data(msg.header, msg.body)
+            header = _to_bytes(msg.header)
+            body = _to_bytes(msg.body)
+            signed_msg = self._client.sign_typed_data(header, body)
         else:
             raise LedgerSigningError(
                 f"Unsupported message-signing specification, (version={version!r})."
             )
 
         v, r, s = signed_msg
-        return MessageSignature(v, r, s)
+        return MessageSignature(v=v, r=HexBytes(r), s=HexBytes(s))
 
     def sign_transaction(self, txn: TransactionAPI, **kwargs) -> Optional[TransactionAPI]:
-        txn_type = TransactionType(txn.type)  # In case it is not enum
-        if txn_type == TransactionType.STATIC:
-            serializable_txn = StaticFeeTransaction(**txn.dict())
-            txn_bytes = rlp.encode(serializable_txn, StaticFeeTransaction)
+        txn.chain_id = 1
+        txn_dict: Dict = {
+            "nonce": txn.nonce,
+            "gas": txn.gas_limit,
+            "amount": txn.value,
+            "data": _to_bytes(txn.data.hex()),
+            "destination": _to_bytes(txn.receiver),
+            "chain_id": txn.chain_id,
+        }
+        if isinstance(txn, StaticFeeTransaction):
+            txn_dict["gas_price"] = txn.gas_price
+
+        elif isinstance(txn, DynamicFeeTransaction):
+            txn_dict["max_fee_per_gas"] = txn.max_fee
+            txn_dict["max_priority_fee_per_gas"] = txn.max_priority_fee
+            if txn.access_list:
+                txn_dict["access_list"] = [[ls.address, ls.storage_keys] for ls in txn.access_list]
+
         else:
-            serializable_txn = DynamicFeeTransaction(**txn.dict())
-            version_byte = bytes(HexBytes(TransactionType.DYNAMIC.value))
-            txn_bytes = version_byte + rlp.encode(serializable_txn, DynamicFeeTransaction)
+            raise TypeError(type(txn))
 
         _echo_object_to_sign(txn)
-        v, r, s = self._client.sign_transaction(txn_bytes)
-
-        chain_id = txn.chain_id
-        # NOTE: EIP-1559 transactions don't pack 'chain_id' with 'v'.
-        if txn_type != TransactionType.DYNAMIC and (chain_id * 2 + 35) + 1 > 255:
-            ecc_parity = v - ((chain_id * 2 + 35) % 256)
-            v = (chain_id * 2 + 35) + ecc_parity
-
-        txn.signature = TransactionSignature(v, r, s)
+        v, r, s = self._client.sign_transaction(txn_dict)
+        txn.signature = TransactionSignature(
+            v=v,
+            r=HexBytes(r),
+            s=HexBytes(s),
+        )
         return txn

ape_ledger/choices.py

@@ -1,10 +1,10 @@
-from typing import Any, Optional, Tuple
+from typing import Any, Optional, Tuple, Union
 
 import click
 from ape.cli import PromptChoice
 from click import Context, Parameter
 
-from ape_ledger.client import LedgerEthereumAppClient
+from ape_ledger.client import get_device
 from ape_ledger.hdpath import HDAccountPath, HDBasePath
 
 
@@ -17,23 +17,21 @@ class AddressPromptChoice(PromptChoice):
 
     def __init__(
         self,
-        app: LedgerEthereumAppClient,
+        hd_path: Union[HDBasePath, str],
         index_offset: int = 0,
         page_size: int = DEFAULT_PAGE_SIZE,
     ):
-        self._app = app
+        if isinstance(hd_path, str):
+            hd_path = HDBasePath(base_path=hd_path)
+
+        self._hd_root_path = hd_path
         self._index_offset = index_offset
         self._page_size = page_size
         self._choice_index = None
 
         # Must call ``_load_choices()`` to set address choices
         super().__init__([])
 
-    @property
-    def _hd_root_path(self) -> HDBasePath:
-        """The base HD path of the Ethereum wallet."""
-        return self._app.hd_root_path
-
     @property
     def _is_incremented(self) -> bool:
         """Returns ``True`` if the user has paged past the first page."""
@@ -99,7 +97,9 @@ def _load_choices(self):
         self.choices = [self._get_address(i) for i in index_range]
 
     def _get_address(self, account_id: int) -> str:
-        return str(self._app.load_account(account_id))
+        path = self._hd_root_path.get_account_path(account_id)
+        device = get_device(path)
+        return device.get_address()
 
 
-__all__ = ["AddressPromptChoice", "PromptChoice"]
+__all__ = ["AddressPromptChoice"]