Merge 7755051 into 09e8507

README.md

@@ -121,7 +121,20 @@ the information about the Safe using:
 ```
 > refresh
 ```
+## Ledger module
+Ledger module is an optional feature of safe-cli to sign transactions with the help of [ledgereth](https://github.com/mikeshultz/ledger-eth-lib) library based on [ledgerblue](https://github.com/LedgerHQ/blue-loader-python).
 
+To enable, safe-cli must be installed as follows:
+```
+pip install safe-cli[ledger]
+```
+When running on Linux, make sure the following rules have been added to `/etc/udev/rules.d/`:
+```commandline
+SUBSYSTEMS=="usb", ATTRS{idVendor}=="2c97", ATTRS{idProduct}=="0000", MODE="0660", TAG+="uaccess", TAG+="udev-acl" OWNER="<UNIX username>"
+SUBSYSTEMS=="usb", ATTRS{idVendor}=="2c97", ATTRS{idProduct}=="0001", MODE="0660", TAG+="uaccess", TAG+="udev-acl" OWNER="<UNIX username>"
+SUBSYSTEMS=="usb", ATTRS{idVendor}=="2c97", ATTRS{idProduct}=="0004", MODE="0660", TAG+="uaccess", TAG+="udev-acl" OWNER="<UNIX username>"
+```
+**NOTE**: before to sign anything ensure that the sign data proposed on ledger is the same than the safe-cli data.
 ## Creating a new Safe
 Use `safe-creator <node_url> <private_key> --owners <checksummed_address_1> <checksummed_address_2> --threshold <uint> --salt-nonce <uint256>`.
 

requirements.txt

@@ -1,9 +1,10 @@
 art==6.1
 colorama==0.4.6
+ledgereth==0.9.0
 packaging>=23.1
 prompt_toolkit==3.0.39
 pygments==2.16.1
 requests==2.31.0
-safe-eth-py==6.0.0b2
+safe-eth-py==6.0.0b3
 tabulate==0.9.0
 web3==6.10.0

safe_cli/operators/hw_accounts/hw_exceptions.py

@@ -0,0 +1,31 @@
+import functools
+
+from ledgereth.exceptions import (
+    LedgerAppNotOpened,
+    LedgerCancel,
+    LedgerLocked,
+    LedgerNotFound,
+)
+
+from safe_cli.operators.safe_operator import HwDeviceException
+
+
+def hw_account_exception(function):
+    @functools.wraps(function)
+    def wrapper(*args, **kwargs):
+        try:
+            return function(*args, **kwargs)
+        except LedgerNotFound as e:
+            raise HwDeviceException(e.message)
+        except LedgerLocked as e:
+            raise HwDeviceException(e.message)
+        except LedgerAppNotOpened as e:
+            raise HwDeviceException(e.message)
+        except LedgerCancel as e:
+            raise HwDeviceException(e.message)
+        except BaseException as e:
+            if "Error while writing" in e.args:
+                raise HwDeviceException("Ledger error writting, restart safe-cli")
+            raise e
+
+    return wrapper

safe_cli/operators/hw_accounts/ledger_manager.py

@@ -0,0 +1,129 @@
+from typing import List, Optional, Set, Tuple
+
+from eth_typing import ChecksumAddress
+from ledgereth import sign_typed_data_draft
+from ledgereth.accounts import LedgerAccount, get_account_by_path
+from ledgereth.comms import init_dongle
+from ledgereth.exceptions import LedgerNotFound
+from prompt_toolkit import HTML, print_formatted_text
+
+from gnosis.eth.eip712 import eip712_encode
+from gnosis.safe import SafeTx
+from gnosis.safe.signatures import signature_to_bytes
+
+from safe_cli.operators.hw_accounts.hw_exceptions import hw_account_exception
+
+
+class LedgerManager:
+    def __init__(self):
+        self.dongle = None
+        self.accounts: Set[LedgerAccount] = set()
+        self.connect()
+
+    def connect(self) -> bool:
+        """
+        Connect with ledger
+        :return: True if connection was successful or False in other case
+        """
+        try:
+            self.dongle = init_dongle(self.dongle)
+            return True
+        except LedgerNotFound:
+            return False
+
+    @property
+    @hw_account_exception
+    def connected(self) -> bool:
+        """
+        :return: True if ledger is connected or False in other case
+        """
+        return self.connect()
+
+    @hw_account_exception
+    def get_accounts(
+        self, legacy_account: Optional[bool] = False, number_accounts: Optional[int] = 5
+    ) -> List[Tuple[ChecksumAddress, str]]:
+        """
+        Request to ledger device the first n accounts
+
+        :param legacy_account:
+        :param number_accounts: number of accounts requested to ledger
+        :return: a list of tuples with address and derivation path
+        """
+        accounts = []
+        for i in range(number_accounts):
+            if legacy_account:
+                path_string = f"44'/60'/0'/{i}"
+            else:
+                path_string = f"44'/60'/{i}'/0/0"
+
+            account = get_account_by_path(path_string, self.dongle)
+            accounts.append((account.address, account.path))
+        return accounts
+
+    @hw_account_exception
+    def add_account(self, derivation_path: str):
+        """
+        Add an account to ledger manager set
+
+        :param derivation_path:
+        :return:
+        """
+        account = get_account_by_path(derivation_path, self.dongle)
+        self.accounts.add(LedgerAccount(account.path, account.address))
+
+    def delete_accounts(self, addresses: List[ChecksumAddress]) -> Set:
+        """
+        Remove ledger accounts from address
+
+        :param accounts:
+        :return: list with the delete accounts
+        """
+        accounts_to_remove = set()
+        for address in addresses:
+            for account in self.accounts:
+                if account.address == address:
+                    accounts_to_remove.add(account)
+        self.accounts = self.accounts.difference(accounts_to_remove)
+        return accounts_to_remove
+
+    @hw_account_exception
+    def sign_eip712(self, safe_tx: SafeTx, accounts: List[LedgerAccount]) -> SafeTx:
+        """
+        Call ledger ethereum app method to sign eip712 hashes with a ledger account
+
+        :param domain_hash:
+        :param message_hash:
+        :param account: ledger account
+        :return: bytes of signature
+        """
+        encode_hash = eip712_encode(safe_tx.eip712_structured_data)
+        domain_hash = encode_hash[1]
+        message_hash = encode_hash[2]
+        for account in accounts:
+            print_formatted_text(
+                HTML(
+                    "Ensure to compare in your ledger before to sign that domain_hash and message_hash are  both correct"
+                )
+            )
+            print_formatted_text(HTML(f"Domain_hash: <b>{domain_hash.hex()}</b>"))
+            print_formatted_text(HTML(f"Message_hash: <b>{message_hash.hex()}</b>"))
+            signed = sign_typed_data_draft(
+                domain_hash, message_hash, account.path, self.dongle
+            )
+
+            signature = signature_to_bytes(signed.v, signed.r, signed.s)
+            # TODO should be refactored on safe_eth_py function insert_signature_sorted
+            # Insert signature sorted
+            if account.address not in safe_tx.signers:
+                new_owners = safe_tx.signers + [account.address]
+                new_owner_pos = sorted(new_owners, key=lambda x: int(x, 16)).index(
+                    account.address
+                )
+                safe_tx.signatures = (
+                    safe_tx.signatures[: 65 * new_owner_pos]
+                    + signature
+                    + safe_tx.signatures[65 * new_owner_pos :]
+                )
+
+        return safe_tx

safe_cli/operators/safe_operator.py

@@ -39,7 +39,7 @@
     get_safe_contract_address,
     get_safe_l2_contract_address,
 )
-from safe_cli.utils import get_erc_20_list, yes_or_no_question
+from safe_cli.utils import choose_option_question, get_erc_20_list, yes_or_no_question
 
 
 @dataclasses.dataclass
@@ -147,6 +147,10 @@ class SafeServiceNotAvailable(SafeOperatorException):
     pass
 
 
+class HwDeviceException(SafeOperatorException):
+    pass
+
+
 def require_tx_service(f):
     @wraps(f)
     def decorated(self, *args, **kwargs):
@@ -228,6 +232,12 @@ def __init__(self, address: ChecksumAddress, node_url: str):
         self.require_all_signatures = (
             True  # Require all signatures to be present to send a tx
         )
+        try:
+            from safe_cli.operators.hw_accounts.ledger_manager import LedgerManager
+
+            self.ledger_manager = LedgerManager()
+        except (ModuleNotFoundError, IOError):
+            self.ledger_manager = None
 
     @cached_property
     def last_default_fallback_handler_address(self) -> ChecksumAddress:
@@ -325,6 +335,34 @@ def load_cli_owners(self, keys: List[str]):
             except ValueError:
                 print_formatted_text(HTML(f"<ansired>Cannot load key={key}</ansired>"))
 
+    def load_ledger_cli_owners(self, legacy_account: bool = False):
+        if not self.ledger_manager:
+            return None
+
+        ledger_accounts = self.ledger_manager.get_accounts(
+            legacy_account=legacy_account
+        )
+        if len(ledger_accounts) == 0:
+            return None
+
+        for option, ledger_account in enumerate(ledger_accounts):
+            address, _ = ledger_account
+            print_formatted_text(HTML(f"{option} - <b>{address}</b> "))
+
+        option = choose_option_question(
+            "Select the owner address", len(ledger_accounts) - 1
+        )
+        address, derivation_path = ledger_accounts[option]
+        if self.ledger_manager.add_account(derivation_path):
+            balance = self.ethereum_client.get_balance(address)
+            print_formatted_text(
+                HTML(
+                    f"Loaded account <b>{address}</b> "
+                    f'with balance={Web3.from_wei(balance, "ether")} ether'
+                    f"Ledger account cannot be defined as sender"
+                )
+            )
+
     def unload_cli_owners(self, owners: List[str]):
         accounts_to_remove: Set[Account] = set()
         for owner in owners:
@@ -335,6 +373,12 @@ def unload_cli_owners(self, owners: List[str]):
                     accounts_to_remove.add(account)
                     break
         self.accounts = self.accounts.difference(accounts_to_remove)
+        # Check if there are ledger owners
+        if self.ledger_manager and len(accounts_to_remove) < len(owners):
+            accounts_to_remove = (
+                accounts_to_remove | self.ledger_manager.delete_accounts(owners)
+            )
+
         if accounts_to_remove:
             print_formatted_text(
                 HTML("<ansigreen>Accounts have been deleted</ansigreen>")
@@ -343,10 +387,15 @@ def unload_cli_owners(self, owners: List[str]):
             print_formatted_text(HTML("<ansired>No account was deleted</ansired>"))
 
     def show_cli_owners(self):
-        if not self.accounts:
+        accounts = (
+            self.accounts | self.ledger_manager.accounts
+            if self.ledger_manager
+            else self.accounts
+        )
+        if not accounts:
             print_formatted_text(HTML("<ansired>No accounts loaded</ansired>"))
         else:
-            for account in self.accounts:
+            for account in accounts:
                 print_formatted_text(
                     HTML(
                         f"<ansigreen><b>Account</b> {account.address} loaded</ansigreen>"
@@ -677,6 +726,28 @@ def print_info(self):
                 )
             )
 
+        if not self.ledger_manager:
+            print_formatted_text(
+                HTML(
+                    "<b><ansigreen>Ledger</ansigreen></b>="
+                    "<ansired>Disabled </ansired> <b>Optional ledger library is not installed, run pip install safe-cli[ledger] </b>"
+                )
+            )
+        elif self.ledger_manager.connected:
+            print_formatted_text(
+                HTML(
+                    "<b><ansigreen>Ledger</ansigreen></b>="
+                    "<ansiblue>Connected</ansiblue>"
+                )
+            )
+        else:
+            print_formatted_text(
+                HTML(
+                    "<b><ansigreen>Ledger</ansigreen></b>="
+                    "<ansiblue>disconnected</ansiblue>"
+                )
+            )
+
         if not self.is_version_updated():
             print_formatted_text(
                 HTML(
@@ -870,13 +941,26 @@ def sign_transaction(self, safe_tx: SafeTx) -> SafeTx:
                 threshold -= 1
                 if threshold == 0:
                     break
+        # If still pending required signatures continue with ledger owners
+        selected_ledger_accounts = []
+        if threshold > 0 and self.ledger_manager:
+            for ledger_account in self.ledger_manager.accounts:
+                if ledger_account.address in permitted_signers:
+                    selected_ledger_accounts.append(ledger_account)
+                    threshold -= 1
+                    if threshold == 0:
+                        break
 
         if self.require_all_signatures and threshold > 0:
             raise NotEnoughSignatures(threshold)
 
         for selected_account in selected_accounts:
             safe_tx.sign(selected_account.key)
 
+        # Sign with ledger
+        if len(selected_ledger_accounts) > 0:
+            safe_tx = self.ledger_manager.sign_eip712(safe_tx, selected_ledger_accounts)
+
         return safe_tx
 
     @require_tx_service

safe_cli/operators/safe_tx_service_operator.py

@@ -102,6 +102,16 @@ def submit_signatures(self, safe_tx_hash: bytes) -> bool:
             for account in self.accounts:
                 if account.address in owners:
                     safe_tx.sign(account.key)
+            # Check if there are ledger signers
+            if self.ledger_manager:
+                selected_ledger_accounts = []
+                for ledger_account in self.ledger_manager.accounts:
+                    if ledger_account.address in owners:
+                        selected_ledger_accounts.append(ledger_account)
+                if len(selected_ledger_accounts) > 0:
+                    safe_tx = self.ledger_manager.sign_eip712(
+                        safe_tx, selected_ledger_accounts
+                    )
 
             if safe_tx.signers:
                 self.safe_tx_service.post_signatures(safe_tx_hash, safe_tx.signatures)