Add support to execute transactions from ledger

safe_cli/operators/hw_wallets/hw_wallet.py

@@ -1,6 +1,9 @@
 import re
 from abc import ABC, abstractmethod
 
+from eth_typing import HexStr
+from web3.types import TxParams
+
 from .constants import BIP32_ETH_PATTERN, BIP32_LEGACY_LEDGER_PATTERN
 from .exceptions import InvalidDerivationPath
 
@@ -44,6 +47,14 @@ def sign_typed_hash(self, domain_hash: bytes, message_hash: bytes) -> bytes:
         :return: signature bytes
         """
 
+    @abstractmethod
+    def get_signed_transaction(self, tx_parameters: TxParams) -> HexStr:
+        """
+
+        :param tx_parameters:
+        :return:
+        """
+
     def __str__(self):
         return f"{self.__class__.__name__} device with address {self.address}"
 

safe_cli/operators/hw_wallets/hw_wallet_manager.py

@@ -3,8 +3,11 @@
 from typing import Dict, List, Optional, Set, Tuple
 
 from eth_typing import ChecksumAddress
+from hexbytes import HexBytes
 from prompt_toolkit import HTML, print_formatted_text
+from web3.types import TxParams, Wei
 
+from gnosis.eth import TxSpeed
 from gnosis.eth.eip712 import eip712_encode
 from gnosis.safe import SafeTx
 
@@ -25,6 +28,7 @@ class HwWalletManager:
     def __init__(self):
         self.wallets: Set[HwWallet] = set()
         self.supported_hw_wallet_types: Dict[str, HwWallet] = {}
+        self.sender: Optional[HwWallet] = None
         try:
             from .ledger_wallet import LedgerWallet
 
@@ -71,7 +75,10 @@ def get_accounts(
         return accounts
 
     def add_account(
-        self, hw_wallet_type: HwWalletType, derivation_path: str
+        self,
+        hw_wallet_type: HwWalletType,
+        derivation_path: str,
+        set_as_sender: Optional[bool] = False,
     ) -> ChecksumAddress:
         """
         Add an account to ledger manager set and return the added address
@@ -82,9 +89,21 @@ def add_account(
 
         hw_wallet = self.get_hw_wallet(hw_wallet_type)
 
-        address = hw_wallet(derivation_path).address
-        self.wallets.add(hw_wallet(derivation_path))
-        return address
+        wallet = hw_wallet(derivation_path)
+        self.wallets.add(wallet)
+        if set_as_sender:
+            self.sender = wallet
+        return wallet.address
+
+    def set_sender(self, hw_wallet_type: HwWalletType, derivation_path: str):
+        """
+        Set a harware wallet as a sender to enable execute transaction from it.
+        :param hw_wallet_type:
+        :param derivation_path:
+        :return:
+        """
+        hw_wallet = self.get_hw_wallet(hw_wallet_type)
+        self.sender = hw_wallet(derivation_path)
 
     def delete_accounts(self, addresses: List[ChecksumAddress]) -> Set:
         """
@@ -136,3 +155,61 @@ def sign_eip712(self, safe_tx: SafeTx, wallets: List[HwWallet]) -> SafeTx:
                 )
 
         return safe_tx
+
+    def execute(
+        self,
+        safe_tx: SafeTx,
+        tx_gas: Optional[int] = None,
+        tx_gas_price: Optional[int] = None,
+        tx_nonce: Optional[int] = None,
+        eip1559_speed: Optional[TxSpeed] = None,
+    ) -> Tuple[HexBytes, TxParams]:
+        """
+        Send multisig tx to the Safe
+
+        :param tx_sender_private_key: Sender private key
+        :param tx_gas: Gas for the external tx. If not, `(safe_tx_gas + base_gas) * 2` will be used
+        :param tx_gas_price: Gas price of the external tx. If not, `gas_price` will be used
+        :param tx_nonce: Force nonce for `tx_sender`
+        :param block_identifier: `latest` or `pending`
+        :param eip1559_speed: If provided, use EIP1559 transaction
+        :return: Tuple(tx_hash, tx)
+        :raises: InvalidMultisigTx: If user tx cannot go through the Safe
+        """
+
+        if eip1559_speed and safe_tx.ethereum_client.is_eip1559_supported():
+            tx_parameters = safe_tx.ethereum_client.set_eip1559_fees(
+                {
+                    "from": self.sender.address,
+                },
+                tx_speed=eip1559_speed,
+            )
+        else:
+            tx_parameters = {
+                "from": self.sender.address,
+                "gasPrice": tx_gas_price or safe_tx.w3.eth.gas_price,
+            }
+
+        if tx_gas:
+            tx_parameters["gas"] = tx_gas
+
+        if tx_nonce is not None:
+            tx_parameters["nonce"] = tx_nonce
+        else:
+            tx_parameters["nonce"] = safe_tx.ethereum_client.get_nonce_for_account(
+                self.sender.address, block_identifier="latest"
+            )
+        safe_tx.tx = safe_tx.w3_tx.build_transaction(tx_parameters)
+        safe_tx.tx["gas"] = Wei(
+            tx_gas or (max(safe_tx.tx["gas"] + 75000, safe_tx.recommended_gas()))
+        )
+        signed_raw_transaction = self.sender.get_signed_transaction(
+            safe_tx.tx
+        )  # sign with ledger
+        safe_tx.tx_hash = safe_tx.ethereum_client.w3.eth.send_raw_transaction(
+            signed_raw_transaction
+        )
+        # Set signatures empty after executing the tx. `Nonce` is increased even if it fails,
+        # so signatures are not valid anymore
+        safe_tx.signatures = b""
+        return safe_tx.tx_hash, safe_tx.tx

safe_cli/operators/hw_wallets/ledger_wallet.py

@@ -1,10 +1,11 @@
 from typing import Optional
 
-from eth_typing import ChecksumAddress
+from eth_typing import ChecksumAddress, HexStr
 from ledgerblue.Dongle import Dongle
-from ledgereth import sign_typed_data_draft
+from ledgereth import create_transaction, sign_typed_data_draft
 from ledgereth.accounts import get_account_by_path
 from ledgereth.comms import init_dongle
+from web3.types import TxParams
 
 from gnosis.safe.signatures import signature_to_bytes
 
@@ -49,3 +50,23 @@ def sign_typed_hash(self, domain_hash: bytes, message_hash: bytes) -> bytes:
         )
 
         return signature_to_bytes(signed.v, signed.r, signed.s)
+
+    def get_signed_transaction(self, tx_parameters: TxParams) -> HexStr:
+        """
+
+        :param tx_parameters:
+        :return:
+        """
+        signed_transaction = create_transaction(
+            destination=tx_parameters["to"],
+            amount=tx_parameters["value"],
+            gas=tx_parameters["gas"],
+            nonce=tx_parameters["nonce"],
+            data=tx_parameters["data"],
+            max_priority_fee_per_gas=tx_parameters["maxPriorityFeePerGas"],
+            max_fee_per_gas=tx_parameters["maxPriorityFeePerGas"],
+            chain_id=5,
+            sender_path=self.derivation_path,
+            dongle=self.dongle,
+        )
+        return signed_transaction.raw_transaction()

safe_cli/operators/hw_wallets/trezor_wallet.py

@@ -1,10 +1,11 @@
 from functools import lru_cache
 
-from eth_typing import ChecksumAddress
+from eth_typing import ChecksumAddress, HexStr
 from trezorlib import tools
 from trezorlib.client import TrezorClient, get_default_client
 from trezorlib.ethereum import get_address, sign_typed_data_hash
 from trezorlib.ui import ClickUI
+from web3.types import TxParams
 
 from .hw_wallet import HwWallet
 from .trezor_exceptions import raise_trezor_exception_as_hw_wallet_exception
@@ -49,3 +50,11 @@ def sign_typed_hash(self, domain_hash: bytes, message_hash: bytes) -> bytes:
             self.client, n=address_n, domain_hash=domain_hash, message_hash=message_hash
         )
         return signed.signature
+
+    def get_signed_transaction(self, tx_parameters: TxParams) -> HexStr:
+        """
+
+        :param tx_parameters:
+        :return:
+        """
+        raise NotImplementedError

safe_cli/operators/safe_operator.py

@@ -116,7 +116,7 @@ def require_default_sender(f):
 
     @wraps(f)
     def decorated(self, *args, **kwargs):
-        if not self.default_sender:
+        if not self.default_sender and not self.hw_wallet_manager.sender:
             raise SenderRequiredException()
         else:
             return f(self, *args, **kwargs)
@@ -260,7 +260,11 @@ def load_cli_owners(self, keys: List[str]):
                         f'with balance={Web3.from_wei(balance, "ether")} ether'
                     )
                 )
-                if not self.default_sender and balance > 0:
+                if (
+                    not self.default_sender
+                    and not self.hw_wallet_manager.sender
+                    and balance > 0
+                ):
                     print_formatted_text(
                         HTML(
                             f"Set account <b>{account.address}</b> as default sender of txs"
@@ -288,17 +292,27 @@ def load_hw_wallet(
             if option is None:
                 return None
             _, derivation_path = ledger_accounts[option]
-
         address = self.hw_wallet_manager.add_account(hw_wallet_type, 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.\n'
-                f"Ledger account cannot be defined as sender"
+                f'with balance={Web3.from_wei(balance, "ether")} ether.'
             )
         )
 
+        if (
+            not self.default_sender
+            and not self.hw_wallet_manager.sender
+            and hw_wallet_type == HwWalletType.LEDGER
+            and balance > 0
+        ):
+            self.hw_wallet_manager.set_sender(hw_wallet_type, derivation_path)
+            print_formatted_text(HTML(f"HwDevice {address} added as sender"))
+        else:
+            print_formatted_text(HTML(f"HwDevice {address} wasn't added as sender"))
+
     def load_ledger_cli_owners(
         self, derivation_path: str = None, legacy_account: bool = False
     ):
@@ -350,6 +364,13 @@ def show_cli_owners(self):
                         f"</ansigreen>"
                     )
                 )
+            elif self.hw_wallet_manager.sender:
+                print_formatted_text(
+                    HTML(
+                        f"<ansigreen><b>HwDevice sender:</b> {self.hw_wallet_manager.sender}"
+                        f"</ansigreen>"
+                    )
+                )
             else:
                 print_formatted_text(
                     HTML("<ansigreen>Not default sender set </ansigreen>")
@@ -838,12 +859,20 @@ def prepare_and_execute_safe_transaction(
     @require_default_sender  # Throws Exception if default sender not found
     def execute_safe_transaction(self, safe_tx: SafeTx):
         try:
-            call_result = safe_tx.call(self.default_sender.address)
+            if self.default_sender:
+                call_result = safe_tx.call(self.default_sender.address)
+            else:
+                call_result = safe_tx.call(self.hw_wallet_manager.sender.address)
             print_formatted_text(HTML(f"Result: <ansigreen>{call_result}</ansigreen>"))
             if yes_or_no_question("Do you want to execute tx " + str(safe_tx)):
-                tx_hash, tx = safe_tx.execute(
-                    self.default_sender.key, eip1559_speed=TxSpeed.NORMAL
-                )
+                if self.default_sender:
+                    tx_hash, tx = safe_tx.execute(
+                        self.default_sender.key, eip1559_speed=TxSpeed.NORMAL
+                    )
+                else:
+                    tx_hash, tx = self.hw_wallet_manager.execute(
+                        safe_tx, eip1559_speed=TxSpeed.NORMAL
+                    )
                 self.executed_transactions.append(tx_hash.hex())
                 print_formatted_text(
                     HTML(