fix: wip

ape_ledger/_cli.py

@@ -36,7 +36,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)

ape_ledger/accounts.py

@@ -10,7 +10,7 @@
 from eth_utils import is_0x_prefixed, to_bytes
 from hexbytes import HexBytes
 
-from ape_ledger.device import get_device
+from ape_ledger.client import LedgerDeviceClient, get_device
 from ape_ledger.exceptions import LedgerSigningError
 from ape_ledger.hdpath import HDAccountPath
 
@@ -76,6 +76,10 @@ class LedgerAccount(AccountAPI):
     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")
@@ -101,16 +105,15 @@ def sign_message(self, msg: SignableMessage) -> Optional[MessageSignature]:
                 f"Unsupported message-signing specification, (version={version!r})."
             )
 
-        device = get_device(self.hdpath)
-
         try:
-            v, r, s = device.sign_message(msg.body)
+            v, r, s = self._client.sign_message(msg.body)
         except Exception as err:
             raise LedgerSigningError(str(err)) from err
 
         return MessageSignature(v=v, r=HexBytes(r), s=HexBytes(s))
 
     def sign_transaction(self, txn: TransactionAPI, **kwargs) -> Optional[TransactionAPI]:
+        txn.chain_id = 1
         txn_dict: Dict = {
             "nonce": txn.nonce,
             "gas": txn.gas_limit,
@@ -131,9 +134,7 @@ def sign_transaction(self, txn: TransactionAPI, **kwargs) -> Optional[Transactio
         else:
             raise TypeError(type(txn))
 
-        device = get_device(self.hdpath)
-
-        v, r, s = device.sign_transaction(txn_dict)
+        v, r, s = self._client.sign_transaction(txn_dict)
         txn.signature = TransactionSignature(
             v=v,
             r=HexBytes(r),

ape_ledger/choices.py

@@ -4,7 +4,7 @@
 from ape.cli import PromptChoice
 from click import Context, Parameter
 
-from ape_ledger.device import get_device
+from ape_ledger.client import get_device
 from ape_ledger.hdpath import HDAccountPath, HDBasePath
 
 

ape_ledger/client.py

@@ -0,0 +1,76 @@
+import atexit
+from functools import cached_property
+from typing import Dict, Tuple
+
+import hid  # type: ignore
+from ape.logging import LogLevel, logger
+from ledgerblue.comm import HIDDongleHIDAPI, getDongle  # type: ignore
+from ledgereth.accounts import get_account_by_path
+from ledgereth.messages import sign_message
+from ledgereth.transactions import SignedType2Transaction, create_transaction
+
+from ape_ledger.hdpath import HDAccountPath
+
+
+class DeviceFactory:
+    device_map: Dict[str, "LedgerDeviceClient"] = {}
+
+    def create_device(self, account: HDAccountPath):
+        if account.path in self.device_map:
+            return self.device_map[account.path]
+
+        device = LedgerDeviceClient(account)
+        self.device_map[account.path] = device
+        return device
+
+
+def get_dongle(debug: bool = False, reopen_on_fail: bool = True) -> HIDDongleHIDAPI:
+    try:
+        return getDongle(debug=debug)
+    except (OSError, RuntimeError) as err:
+        if str(err).lower().strip() in ("open failed", "already open") and reopen_on_fail:
+            # Device was not closed properly.
+            device = hid.device()
+            device.close()
+            return get_dongle(debug=debug, reopen_on_fail=False)
+
+        raise  # the OSError
+
+
+class LedgerDeviceClient:
+    def __init__(self, account: HDAccountPath):
+        self._account = account.path.lstrip("m/")
+
+    @cached_property
+    def dongle(self):
+        debug = logger.level <= LogLevel.DEBUG
+        device = get_dongle(debug=debug)
+
+        def close():
+            logger.info("Closing device.")
+            device.close()
+
+        atexit.register(close)
+        return device
+
+    def get_address(self) -> str:
+        return get_account_by_path(self._account, dongle=self.dongle).address
+
+    def sign_message(self, text: bytes) -> Tuple[int, int, int]:
+        signed_msg = sign_message(text, sender_path=self._account)
+        return signed_msg.v, signed_msg.r, signed_msg.s
+
+    def sign_transaction(self, txn: Dict) -> Tuple[int, int, int]:
+        kwargs = {**txn, "sender_path": self._account, "dongle": self.dongle}
+        signed_tx = create_transaction(**kwargs)
+        if isinstance(signed_tx, SignedType2Transaction):
+            return signed_tx.y_parity, signed_tx.sender_r, signed_tx.sender_s
+        else:
+            return signed_tx.v, signed_tx.r, signed_tx.s
+
+
+_device_factory = DeviceFactory()
+
+
+def get_device(account: HDAccountPath):
+    return _device_factory.create_device(account)

tests/conftest.py

@@ -1,30 +1,90 @@
 import json
 
 import pytest
+from ape import accounts, networks
 from ape.api.accounts import AccountContainerAPI
 from click.testing import CliRunner
-from eth_typing import HexAddress, HexStr
-
-TEST_ADDRESSES = [
-    HexAddress(HexStr("0x0A78AAAAA2122100000b9046f0A085AB2E111113")),
-    HexAddress(HexStr("0x1A78AAAAA2122100000b9046f0A085AB2E111113")),
-    HexAddress(HexStr("0x2A78AAAAA2122100000b9046f0A085AB2E111113")),
-    HexAddress(HexStr("0x3A78AAAAA2122100000b9046f0A085AB2E111113")),
-    HexAddress(HexStr("0x4A78AAAAA2122100000b9046f0A085AB2E111113")),
-    HexAddress(HexStr("0x5A78AAAAA2122100000b9046f0A085AB2E111113")),
-    HexAddress(HexStr("0x6A78AAAAA2122100000b9046f0A085AB2E111113")),
-    HexAddress(HexStr("0x7A78AAAAA2122100000b9046f0A085AB2E111113")),
-    HexAddress(HexStr("0x8A78AAAAA2122100000b9046f0A085AB2E111113")),
-    HexAddress(HexStr("0x9A78AAAAA2122100000b9046f0A085AB2E111113")),
-]
-TEST_ADDRESS = TEST_ADDRESSES[0]
+from eth_account.messages import encode_defunct
+from ethpm_types import HexBytes
+
+from ape_ledger.client import LedgerDeviceClient
+
 TEST_ALIAS = "TestAlias"
 TEST_HD_PATH = "m/44'/60'/0'/0/0"
 
 
 @pytest.fixture
-def mock_apdu(mocker):
-    return mocker.MagicMock()
+def hd_path():
+    return TEST_HD_PATH
+
+
+@pytest.fixture
+def alias():
+    return TEST_ALIAS
+
+
+@pytest.fixture
+def test_accounts():
+    return accounts.test_accounts
+
+
+@pytest.fixture
+def account_addresses(test_accounts):
+    return [a.address for a in test_accounts]
+
+
+@pytest.fixture
+def account_0(test_accounts):
+    return test_accounts[0]
+
+
+@pytest.fixture
+def account_1(test_accounts):
+    return test_accounts[0]
+
+
+@pytest.fixture
+def address(account_addresses):
+    return account_addresses[0]
+
+
+@pytest.fixture(autouse=True)
+def connection():
+    with networks.ethereum.local.use_provider("test") as provider:
+        yield provider
+
+
+@pytest.fixture
+def msg_signature(account_0):
+    msg = encode_defunct(text="__TEST_MESSAGE__")
+    sig = account_0.sign_message(msg)
+    return (
+        sig.v,
+        int(HexBytes(sig.r).hex(), 16),
+        int(HexBytes(sig.s).hex(), 16),
+    )
+
+
+@pytest.fixture
+def tx_signature(account_0, account_1):
+    txn = account_0.transfer(account_1, "1 gwei")
+    return (
+        txn.signature.v,
+        int(HexBytes(txn.signature.r).hex(), 16),
+        int(HexBytes(txn.signature.s).hex(), 16),
+    )
+
+
+@pytest.fixture(autouse=True)
+def mock_device(mocker, hd_path, account_addresses, msg_signature, tx_signature):
+    device = mocker.MagicMock(spec=LedgerDeviceClient)
+    patch = mocker.patch("ape_ledger.client.get_device")
+    patch.return_value = device
+    device._account = hd_path
+    device.get_address.side_effect = lambda a: account_addresses[a]
+    device.sign_message.side_effect = lambda *args, **kwargs: msg_signature
+    device.sign_transaction.side_effect = lambda *args, **kwargs: tx_signature
+    return device
 
 
 @pytest.fixture
@@ -37,8 +97,13 @@ def runner():
     return CliRunner()
 
 
-def assert_account(account_path, expected_address=TEST_ADDRESS, expected_hdpath="m/44'/60'/0'/0/0"):
-    with open(account_path) as account_file:
-        account_data = json.load(account_file)
-        assert account_data["address"] == expected_address
-        assert account_data["hdpath"] == expected_hdpath
+@pytest.fixture
+def assert_account(address):
+    def fn(account_path, expected_address=None, expected_hdpath="m/44'/60'/0'/0/0"):
+        expected_address = expected_address or address
+        with open(account_path) as account_file:
+            account_data = json.load(account_file)
+            assert account_data["address"] == expected_address
+            assert account_data["hdpath"] == expected_hdpath
+
+    return fn