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,24 +1,26 @@
-import asyncio
 import logging
 from pathlib import Path
-from typing import Dict, Optional, Type, TypeVar
+from typing import Dict, Literal, Optional, Type, TypeVar, Union, overload
 
 from aleph_message.models import Chain
+from ledgereth.exceptions import LedgerError
+from typing_extensions import TypeAlias
 
 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.types import AccountFromPrivateKey, HardwareAccount
+from aleph.sdk.wallets.ledger import LedgerETHAccount
 
 logger = logging.getLogger(__name__)
 
 T = TypeVar("T", bound=AccountFromPrivateKey)
+AccountTypes: TypeAlias = Union["AccountFromPrivateKey", "HardwareAccount"]
 
 chain_account_map: Dict[Chain, Type[T]] = {  # type: ignore
     Chain.ARBITRUM: EVMAccount,
@@ -56,7 +58,7 @@ def load_chain_account_type(chain: Chain) -> Type[AccountFromPrivateKey]:
 
 def account_from_hex_string(
     private_key_str: str,
-    account_type: Optional[Type[T]],
+    account_type: Optional[Type[AccountFromPrivateKey]],
     chain: Optional[Chain] = None,
 ) -> AccountFromPrivateKey:
     if private_key_str.startswith("0x"):
@@ -78,7 +80,7 @@ def account_from_hex_string(
 
 def account_from_file(
     private_key_path: Path,
-    account_type: Optional[Type[T]],
+    account_type: Optional[Type[AccountFromPrivateKey]],
     chain: Optional[Chain] = None,
 ) -> AccountFromPrivateKey:
     private_key = private_key_path.read_bytes()
@@ -97,13 +99,60 @@ def account_from_file(
     return account
 
 
+@overload
+def _load_account(
+    private_key_str: str,
+    private_key_path: None = None,
+    account_type: Type[AccountFromPrivateKey] = ...,
+    chain: Optional[Chain] = None,
+) -> AccountFromPrivateKey: ...
+
+
+@overload
+def _load_account(
+    private_key_str: Literal[None],
+    private_key_path: Path,
+    account_type: Type[AccountFromPrivateKey] = ...,
+    chain: Optional[Chain] = None,
+) -> AccountFromPrivateKey: ...
+
+
+@overload
+def _load_account(
+    private_key_str: Literal[None],
+    private_key_path: Literal[None],
+    account_type: Type[HardwareAccount],
+    chain: Optional[Chain] = None,
+) -> HardwareAccount: ...
+
+
+@overload
 def _load_account(
     private_key_str: Optional[str] = None,
     private_key_path: Optional[Path] = None,
-    account_type: Optional[Type[AccountFromPrivateKey]] = None,
+    account_type: Optional[Type[AccountTypes]] = None,
     chain: Optional[Chain] = None,
-) -> AccountFromPrivateKey:
-    """Load an account from a private key string or file, or from the configuration file."""
+) -> AccountTypes: ...
+
+
+def _load_account(
+    private_key_str: Optional[str] = None,
+    private_key_path: Optional[Path] = None,
+    account_type: Optional[Type[AccountTypes]] = None,
+    chain: Optional[Chain] = None,
+) -> AccountTypes:
+    """Load an account from a private key string or file, or from the configuration file.
+
+    This function can return different types of accounts based on the input:
+    - AccountFromPrivateKey: When a private key is provided (string or file)
+    - HardwareAccount: When config has AccountType.HARDWARE and a Ledger device is connected
+
+    The function will attempt to load an account in the following order:
+    1. From provided private key string
+    2. From provided private key file
+    3. From Ledger device (if config.type is HARDWARE)
+    4. Generate a fallback private key
+    """
 
     config = load_main_configuration(settings.CONFIG_FILE)
     default_chain = settings.DEFAULT_CHAIN
@@ -129,27 +178,31 @@ def _load_account(
 
     # Loads private key from a string
     if private_key_str:
-        return account_from_hex_string(private_key_str, account_type, chain)
+        return account_from_hex_string(private_key_str, None, chain)
+
     # Loads private key from a file
     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:
-        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,
-            )
-        )
+        return account_from_file(private_key_path, account_type, chain)  # type: ignore
+    elif config and config.address and config.type == AccountType.HARDWARE:
+        logger.debug("Using ledger account")
+        try:
+            ledger_account = LedgerETHAccount.from_address(config.address)
+            if ledger_account:
+                # Connect provider to the chain
+                # Only valid for EVM chain sign we sign TX using device
+                # and then use Superfluid logic to publish it to BASE / AVAX
+                if chain:
+                    ledger_account.connect_chain(chain)
+                return ledger_account
+        except LedgerError as e:
+            logger.warning(f"Ledger Error : {e.message}")
+            raise e
+        except OSError as e:
+            logger.warning("Please ensure Udev rules are set to use Ledger")
+            raise e
+
     # 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), None, chain)
+    logger.info(f"Generated fallback private key with address {account.get_address()}")
+    return account

src/aleph/sdk/chains/ethereum.py

@@ -1,5 +1,6 @@
 import asyncio
 import base64
+from abc import abstractmethod
 from decimal import Decimal
 from pathlib import Path
 from typing import Awaitable, Dict, Optional, Union
@@ -36,65 +37,30 @@
 from .common import BaseAccount, get_fallback_private_key, get_public_key
 
 
-class ETHAccount(BaseAccount):
-    """Interact with an Ethereum address or key pair on EVM blockchains"""
+class BaseEthAccount(BaseAccount):
+    """Base logic to interact with EVM blockchains"""
 
     CHAIN = "ETH"
     CURVE = "secp256k1"
-    _account: LocalAccount
+
     _provider: Optional[Web3]
     chain: Optional[Chain]
     chain_id: Optional[int]
     rpc: Optional[str]
     superfluid_connector: Optional[Superfluid]
 
-    def __init__(
-        self,
-        private_key: bytes,
-        chain: Optional[Chain] = None,
-    ):
-        self.private_key = private_key
-        self._account: LocalAccount = Account.from_key(self.private_key)
+    def __init__(self, chain: Optional[Chain] = None):
+        self.chain = chain
         self.connect_chain(chain=chain)
 
-    @staticmethod
-    def from_mnemonic(mnemonic: str, chain: Optional[Chain] = None) -> "ETHAccount":
-        Account.enable_unaudited_hdwallet_features()
-        return ETHAccount(
-            private_key=Account.from_mnemonic(mnemonic=mnemonic).key, chain=chain
-        )
-
-    def export_private_key(self) -> str:
-        """Export the private key using standard format."""
-        return f"0x{base64.b16encode(self.private_key).decode().lower()}"
-
-    def get_address(self) -> str:
-        return self._account.address
-
-    def get_public_key(self) -> str:
-        return "0x" + get_public_key(private_key=self._account.key).hex()
-
-    async def sign_raw(self, buffer: bytes) -> bytes:
-        """Sign a raw buffer."""
-        msghash = encode_defunct(text=buffer.decode("utf-8"))
-        sig = self._account.sign_message(msghash)
-        return sig["signature"]
-
-    async def sign_message(self, message: Dict) -> Dict:
+    @abstractmethod
+    async def _sign_and_send_transaction(self, tx_params: TxParams) -> str:
         """
-        Returns a signed message from an aleph.im message.
-        Args:
-            message: Message to sign
-        Returns:
-            Dict: Signed message
+        Sign and broadcast a transaction using the provided ETHAccount
+        @param tx_params - Transaction parameters
+        @returns - str - Transaction hash
         """
-        signed_message = await super().sign_message(message)
-
-        # Apply that fix as seems that sometimes the .hex() method doesn't add the 0x str at the beginning
-        if not str(signed_message["signature"]).startswith("0x"):
-            signed_message["signature"] = "0x" + signed_message["signature"]
-
-        return signed_message
+        raise NotImplementedError
 
     def connect_chain(self, chain: Optional[Chain] = None):
         self.chain = chain
@@ -150,36 +116,13 @@ def can_transact(self, tx: TxParams, block=True) -> bool:
             )
         return valid
 
-    async def _sign_and_send_transaction(self, tx_params: TxParams) -> str:
-        """
-        Sign and broadcast a transaction using the provided ETHAccount
-        @param tx_params - Transaction parameters
-        @returns - str - Transaction hash
-        """
-
-        def sign_and_send() -> TxReceipt:
-            if self._provider is None:
-                raise ValueError("Provider not connected")
-            signed_tx = self._provider.eth.account.sign_transaction(
-                tx_params, self._account.key
-            )
-
-            tx_hash = self._provider.eth.send_raw_transaction(signed_tx.raw_transaction)
-            tx_receipt = self._provider.eth.wait_for_transaction_receipt(
-                tx_hash, settings.TX_TIMEOUT
+    def get_eth_balance(self) -> Decimal:
+        if not self._provider:
+            raise ValueError(
+                "Provider not set. Please configure a provider before checking balance."
             )
-            return tx_receipt
 
-        loop = asyncio.get_running_loop()
-        tx_receipt = await loop.run_in_executor(None, sign_and_send)
-        return tx_receipt["transactionHash"].hex()
-
-    def get_eth_balance(self) -> Decimal:
-        return Decimal(
-            self._provider.eth.get_balance(self._account.address)
-            if self._provider
-            else 0
-        )
+        return Decimal(self._provider.eth.get_balance(self.get_address()))
 
     def get_token_balance(self) -> Decimal:
         if self.chain and self._provider:
@@ -247,6 +190,84 @@ def manage_flow(
         )
 
 
+class ETHAccount(BaseEthAccount):
+    """Interact with an Ethereum address or key pair on EVM blockchains"""
+
+    _account: LocalAccount
+
+    def __init__(
+        self,
+        private_key: bytes,
+        chain: Optional[Chain] = None,
+    ):
+        self.private_key = private_key
+        self._account = Account.from_key(self.private_key)
+        super().__init__(chain=chain)
+
+    @staticmethod
+    def from_mnemonic(mnemonic: str, chain: Optional[Chain] = None) -> "ETHAccount":
+        Account.enable_unaudited_hdwallet_features()
+        return ETHAccount(
+            private_key=Account.from_mnemonic(mnemonic=mnemonic).key, chain=chain
+        )
+
+    def export_private_key(self) -> str:
+        """Export the private key using standard format."""
+        return f"0x{base64.b16encode(self.private_key).decode().lower()}"
+
+    def get_address(self) -> str:
+        return self._account.address
+
+    def get_public_key(self) -> str:
+        return "0x" + get_public_key(private_key=self._account.key).hex()
+
+    async def sign_raw(self, buffer: bytes) -> bytes:
+        """Sign a raw buffer."""
+        msghash = encode_defunct(text=buffer.decode("utf-8"))
+        sig = self._account.sign_message(msghash)
+        return sig["signature"]
+
+    async def sign_message(self, message: Dict) -> Dict:
+        """
+        Returns a signed message from an aleph Cloud message.
+        Args:
+            message: Message to sign
+        Returns:
+            Dict: Signed message
+        """
+        signed_message = await super().sign_message(message)
+
+        # Apply that fix as seems that sometimes the .hex() method doesn't add the 0x str at the beginning
+        if not str(signed_message["signature"]).startswith("0x"):
+            signed_message["signature"] = "0x" + signed_message["signature"]
+
+        return signed_message
+
+    async def _sign_and_send_transaction(self, tx_params: TxParams) -> str:
+        """
+        Sign and broadcast a transaction using the provided ETHAccount
+        @param tx_params - Transaction parameters
+        @returns - str - Transaction hash
+        """
+
+        def sign_and_send() -> TxReceipt:
+            if self._provider is None:
+                raise ValueError("Provider not connected")
+            signed_tx = self._provider.eth.account.sign_transaction(
+                tx_params, self._account.key
+            )
+
+            tx_hash = self._provider.eth.send_raw_transaction(signed_tx.raw_transaction)
+            tx_receipt = self._provider.eth.wait_for_transaction_receipt(
+                tx_hash, settings.TX_TIMEOUT
+            )
+            return tx_receipt
+
+        loop = asyncio.get_running_loop()
+        tx_receipt = await loop.run_in_executor(None, sign_and_send)
+        return tx_receipt["transactionHash"].hex()
+
+
 def get_fallback_account(
     path: Optional[Path] = None, chain: Optional[Chain] = None
 ) -> ETHAccount: