From 931e39b5410a200913b81051f671f3ebbf931747 Mon Sep 17 00:00:00 2001
From: moisses89 <7888669+moisses89@users.noreply.github.com>
Date: Thu, 14 Dec 2023 14:10:07 +0100
Subject: [PATCH] Add support to execute transactions from ledger
---
safe_cli/operators/hw_wallets/hw_wallet.py | 11 +++
.../operators/hw_wallets/hw_wallet_manager.py | 85 ++++++++++++++++++-
.../operators/hw_wallets/ledger_wallet.py | 25 +++++-
.../operators/hw_wallets/trezor_wallet.py | 11 ++-
safe_cli/operators/safe_operator.py | 47 ++++++++--
5 files changed, 163 insertions(+), 16 deletions(-)
diff --git a/safe_cli/operators/hw_wallets/hw_wallet.py b/safe_cli/operators/hw_wallets/hw_wallet.py
index ab793385..695a04e1 100644
--- a/safe_cli/operators/hw_wallets/hw_wallet.py
+++ b/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}"
diff --git a/safe_cli/operators/hw_wallets/hw_wallet_manager.py b/safe_cli/operators/hw_wallets/hw_wallet_manager.py
index a828859a..9d60a682 100644
--- a/safe_cli/operators/hw_wallets/hw_wallet_manager.py
+++ b/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
diff --git a/safe_cli/operators/hw_wallets/ledger_wallet.py b/safe_cli/operators/hw_wallets/ledger_wallet.py
index 31e01d2d..09505e0f 100644
--- a/safe_cli/operators/hw_wallets/ledger_wallet.py
+++ b/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()
diff --git a/safe_cli/operators/hw_wallets/trezor_wallet.py b/safe_cli/operators/hw_wallets/trezor_wallet.py
index 0ac2ed89..da153db6 100644
--- a/safe_cli/operators/hw_wallets/trezor_wallet.py
+++ b/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
diff --git a/safe_cli/operators/safe_operator.py b/safe_cli/operators/safe_operator.py
index 5ca756b7..88a2ddb0 100644
--- a/safe_cli/operators/safe_operator.py
+++ b/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 {account.address} 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 {address} "
- 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""
)
)
+ elif self.hw_wallet_manager.sender:
+ print_formatted_text(
+ HTML(
+ f"HwDevice sender: {self.hw_wallet_manager.sender}"
+ f""
+ )
+ )
else:
print_formatted_text(
HTML("Not default sender set ")
@@ -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: {call_result}"))
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(