Implement Ledger

pyproject.toml

@@ -38,6 +38,8 @@ dependencies = [
   "eth-abi>=5.0.1; python_version>='3.9'",
   "eth-typing>=5.0.1",
   "jwcrypto==1.5.6",
+  "ledgerblue>=0.1.48",
+  "ledgereth>=0.10",
   "pydantic>=2,<3",
   "pydantic-settings>=2",
   "pynacl==1.5",                           # Needed now as default with _load_account changement

src/aleph/sdk/account.py

@@ -1,20 +1,19 @@
-import asyncio
 import logging
 from pathlib import Path
-from typing import Dict, Optional, Type, TypeVar
+from typing import Dict, Optional, Type, TypeVar, Union
 
 from aleph_message.models import Chain
 
 from aleph.sdk.chains.common import get_fallback_private_key
 from aleph.sdk.chains.ethereum import ETHAccount
 from aleph.sdk.chains.evm import EVMAccount
-from aleph.sdk.chains.remote import RemoteAccount
 from aleph.sdk.chains.solana import SOLAccount
 from aleph.sdk.chains.substrate import DOTAccount
 from aleph.sdk.chains.svm import SVMAccount
-from aleph.sdk.conf import load_main_configuration, settings
+from aleph.sdk.conf import AccountType, load_main_configuration, settings
 from aleph.sdk.evm_utils import get_chains_with_super_token
 from aleph.sdk.types import AccountFromPrivateKey
+from aleph.sdk.wallets.ledger import LedgerETHAccount
 
 logger = logging.getLogger(__name__)
 
@@ -102,7 +101,7 @@ def _load_account(
     private_key_path: Optional[Path] = None,
     account_type: Optional[Type[AccountFromPrivateKey]] = None,
     chain: Optional[Chain] = None,
-) -> AccountFromPrivateKey:
+) -> Union[AccountFromPrivateKey, LedgerETHAccount]:
     """Load an account from a private key string or file, or from the configuration file."""
 
     config = load_main_configuration(settings.CONFIG_FILE)
@@ -134,22 +133,24 @@ def _load_account(
     elif private_key_path and private_key_path.is_file():
         return account_from_file(private_key_path, account_type, chain)
     # For ledger keys
-    elif settings.REMOTE_CRYPTO_HOST:
+    # elif settings.REMOTE_CRYPTO_HOST:
+    #     logger.debug("Using remote account")
+    #     loop = asyncio.get_event_loop()
+    #     return loop.run_until_complete(
+    #         RemoteAccount.from_crypto_host(
+    #             host=settings.REMOTE_CRYPTO_HOST,
+    #             unix_socket=settings.REMOTE_CRYPTO_UNIX_SOCKET,
+    #         )
+    #     )
+    # New Ledger Implementation
+    elif config and config.address and config.type == AccountType.EXTERNAL:
         logger.debug("Using remote account")
-        loop = asyncio.get_event_loop()
-        return loop.run_until_complete(
-            RemoteAccount.from_crypto_host(
-                host=settings.REMOTE_CRYPTO_HOST,
-                unix_socket=settings.REMOTE_CRYPTO_UNIX_SOCKET,
-            )
-        )
+        ledger_account = LedgerETHAccount.from_address(config.address)
+        if ledger_account:
+            return ledger_account
+
     # Fallback: config.path if set, else generate a new private key
-    else:
-        new_private_key = get_fallback_private_key()
-        account = account_from_hex_string(
-            bytes.hex(new_private_key), account_type, chain
-        )
-        logger.info(
-            f"Generated fallback private key with address {account.get_address()}"
-        )
-        return account
+    new_private_key = get_fallback_private_key()
+    account = account_from_hex_string(bytes.hex(new_private_key), account_type, chain)
+    logger.info(f"Generated fallback private key with address {account.get_address()}")
+    return account

src/aleph/sdk/conf.py

@@ -1,6 +1,7 @@
 import json
 import logging
 import os
+from enum import Enum
 from pathlib import Path
 from shutil import which
 from typing import ClassVar, Dict, List, Optional, Union
@@ -280,15 +281,22 @@ class Settings(BaseSettings):
     )
 
 
+class AccountType(str, Enum):
+    INTERNAL: str = "internal"
+    EXTERNAL: str = "external"
+
+
 class MainConfiguration(BaseModel):
     """
     Intern Chain Management with Account.
     """
 
-    path: Path
+    path: Optional[Path] = None
+    type: Optional[AccountType] = AccountType.INTERNAL
     chain: Chain
+    address: Optional[str] = None
 
-    model_config = SettingsConfigDict(use_enum_values=True)
+    # model_config = SettingsConfigDict(use_enum_values=True)
 
 
 # Settings singleton
@@ -322,7 +330,9 @@ class MainConfiguration(BaseModel):
             with open(settings.CONFIG_FILE, "r", encoding="utf-8") as f:
                 config_data = json.load(f)
 
-            if "path" in config_data:
+            if "path" in config_data and (
+                "type" not in config_data or config_data["type"] == AccountType.INTERNAL
+            ):
                 settings.PRIVATE_KEY_FILE = Path(config_data["path"])
         except json.JSONDecodeError:
             pass
@@ -361,8 +371,7 @@ def load_main_configuration(file_path: Path) -> Optional[MainConfiguration]:
     try:
         with file_path.open("rb") as file:
             content = file.read()
-            data = json.loads(content.decode("utf-8"))
-            return MainConfiguration(**data)
+            return MainConfiguration.model_validate_json(content.decode("utf-8"))
     except UnicodeDecodeError as e:
         logger.error(f"Unable to decode {file_path} as UTF-8: {e}")
     except json.JSONDecodeError:

src/aleph/sdk/wallets/ledger/ethereum.py

@@ -2,26 +2,30 @@
 
 from typing import Dict, List, Optional
 
+from aleph_message.models import Chain
 from eth_typing import HexStr
 from ledgerblue.Dongle import Dongle
 from ledgereth import find_account, get_account_by_path, get_accounts
 from ledgereth.comms import init_dongle
 from ledgereth.messages import sign_message
 from ledgereth.objects import LedgerAccount, SignedMessage
 
-from ...chains.common import BaseAccount, get_verification_buffer
+from ...chains.common import get_verification_buffer
+from ...chains.ethereum import ETHAccount
 from ...utils import bytes_from_hex
 
 
-class LedgerETHAccount(BaseAccount):
+class LedgerETHAccount(ETHAccount):
     """Account using the Ethereum app on Ledger hardware wallets."""
 
     CHAIN = "ETH"
     CURVE = "secp256k1"
     _account: LedgerAccount
     _device: Dongle
 
-    def __init__(self, account: LedgerAccount, device: Dongle):
+    def __init__(
+        self, account: LedgerAccount, device: Dongle, chain: Optional[Chain] = None
+    ):
         """Initialize an aleph.im account instance that relies on a LedgerHQ
         device and the Ethereum Ledger application for signatures.
 
@@ -30,6 +34,18 @@ def __init__(self, account: LedgerAccount, device: Dongle):
         """
         self._account = account
         self._device = device
+        self.connect_chain(chain=chain)
+
+    @staticmethod
+    def get_accounts(
+        device: Optional[Dongle] = None, count: int = 5
+    ) -> List[LedgerAccount]:
+        """Initialize an aleph.im account from a LedgerHQ device from
+        a known wallet address.
+        """
+        device = device or init_dongle()
+        accounts: List[LedgerAccount] = get_accounts(dongle=device, count=count)
+        return accounts
 
     @staticmethod
     def from_address(