diff --git a/pyproject.toml b/pyproject.toml index c148831c..62ea5fc5 100644 --- a/pyproject.toml +++ b/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 diff --git a/src/aleph/sdk/account.py b/src/aleph/sdk/account.py index 6af5e32c..eb6f4092 100644 --- a/src/aleph/sdk/account.py +++ b/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 diff --git a/src/aleph/sdk/conf.py b/src/aleph/sdk/conf.py index fc852417..19d000b3 100644 --- a/src/aleph/sdk/conf.py +++ b/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: diff --git a/src/aleph/sdk/wallets/ledger/ethereum.py b/src/aleph/sdk/wallets/ledger/ethereum.py index 5dc40f03..370f142e 100644 --- a/src/aleph/sdk/wallets/ledger/ethereum.py +++ b/src/aleph/sdk/wallets/ledger/ethereum.py @@ -2,6 +2,7 @@ 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 @@ -9,11 +10,12 @@ 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" @@ -21,7 +23,9 @@ class LedgerETHAccount(BaseAccount): _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(