Add support for ledger

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 signing anything ensure that the data showing on your ledger is the same as 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.0b5
 tabulate==0.9.0
 web3==6.10.0

safe_cli/operators/__init__.py

@@ -1,4 +1,5 @@
 # flake8: noqa F401
 from .enums import SafeOperatorMode
-from .safe_operator import SafeOperator, SafeServiceNotAvailable
+from .exceptions import SafeServiceNotAvailable
+from .safe_operator import SafeOperator
 from .safe_tx_service_operator import SafeTxServiceOperator

safe_cli/operators/exceptions.py

@@ -0,0 +1,86 @@
+class SafeOperatorException(Exception):
+    pass
+
+
+class ExistingOwnerException(SafeOperatorException):
+    pass
+
+
+class NonExistingOwnerException(SafeOperatorException):
+    pass
+
+
+class HashAlreadyApproved(SafeOperatorException):
+    pass
+
+
+class ThresholdLimitException(SafeOperatorException):
+    pass
+
+
+class SameFallbackHandlerException(SafeOperatorException):
+    pass
+
+
+class InvalidFallbackHandlerException(SafeOperatorException):
+    pass
+
+
+class FallbackHandlerNotSupportedException(SafeOperatorException):
+    pass
+
+
+class SameGuardException(SafeOperatorException):
+    pass
+
+
+class InvalidGuardException(SafeOperatorException):
+    pass
+
+
+class GuardNotSupportedException(SafeOperatorException):
+    pass
+
+
+class SameMasterCopyException(SafeOperatorException):
+    pass
+
+
+class SafeAlreadyUpdatedException(SafeOperatorException):
+    pass
+
+
+class UpdateAddressesNotValid(SafeOperatorException):
+    pass
+
+
+class SenderRequiredException(SafeOperatorException):
+    pass
+
+
+class AccountNotLoadedException(SafeOperatorException):
+    pass
+
+
+class NotEnoughSignatures(SafeOperatorException):
+    pass
+
+
+class InvalidMasterCopyException(SafeOperatorException):
+    pass
+
+
+class NotEnoughEtherToSend(SafeOperatorException):
+    pass
+
+
+class NotEnoughTokenToSend(SafeOperatorException):
+    pass
+
+
+class SafeServiceNotAvailable(SafeOperatorException):
+    pass
+
+
+class HardwareWalletException(SafeOperatorException):
+    pass

safe_cli/operators/hw_accounts/exceptions.py

@@ -0,0 +1,31 @@
+import functools
+
+from ledgereth.exceptions import (
+    LedgerAppNotOpened,
+    LedgerCancel,
+    LedgerLocked,
+    LedgerNotFound,
+)
+
+from safe_cli.operators.exceptions import HardwareWalletException
+
+
+def raise_as_hw_account_exception(function):
+    @functools.wraps(function)
+    def wrapper(*args, **kwargs):
+        try:
+            return function(*args, **kwargs)
+        except LedgerNotFound as e:
+            raise HardwareWalletException(e.message)
+        except LedgerLocked as e:
+            raise HardwareWalletException(e.message)
+        except LedgerAppNotOpened as e:
+            raise HardwareWalletException(e.message)
+        except LedgerCancel as e:
+            raise HardwareWalletException(e.message)
+        except BaseException as e:
+            if "Error while writing" in e.args:
+                raise HardwareWalletException("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.exceptions import raise_as_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
+    @raise_as_hw_account_exception
+    def connected(self) -> bool:
+        """
+        :return: True if ledger is connected or False in other case
+        """
+        return self.connect()
+
+    @raise_as_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
+
+    @raise_as_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
+
+    @raise_as_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(
+                    "<ansired>Make sure in your ledger before signing that domain_hash and message_hash are both correct</ansired>"
+                )
+            )
+            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