19675107724

Committed 25 Nov 2025 03:35PM UTC coverage: 88.372% (-0.04%) from 88.407%

Build # 19675107724

Build Type

push

github

Committed by

Uxio0

Commit Message

Consolidate owner lookup

Run Details

224 of 266 branches covered (84.21%)

Branch coverage included in aggregate %.

15 of 17 new or added lines in 2 files covered. (88.24%)

2 existing lines in 1 file now uncovered.

2892 of 3260 relevant lines covered (88.71%)

3.55 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

66.9

/src/safe_cli/operators/safe_tx_service_operator.py

import json
from typing import Any, Dict, Optional, Sequence, Set

from colorama import Fore, Style
from eth_account.messages import defunct_hash_message
from eth_account.signers.local import LocalAccount
from eth_typing import ChecksumAddress
from hexbytes import HexBytes
from prompt_toolkit import HTML, print_formatted_text
from safe_eth.eth.contracts import get_erc20_contract
from safe_eth.eth.eip712 import eip712_encode_hash
from safe_eth.safe import SafeOperationEnum, SafeTx
from safe_eth.safe.api import SafeAPIException
from safe_eth.safe.api.transaction_service_api.transaction_service_messages import (
    get_remove_transaction_message,
)
from safe_eth.safe.multi_send import MultiSend, MultiSendOperation, MultiSendTx
from safe_eth.safe.safe_signature import SafeSignature
from safe_eth.safe.signatures import signature_to_bytes
from safe_eth.util.util import to_0x_hex_str
from tabulate import tabulate

from ..utils import get_input, yes_or_no_question
from . import SafeServiceNotAvailable
from .exceptions import AccountNotLoadedException, NonExistingOwnerException
from .safe_operator import SafeOperator


class SafeTxServiceOperator(SafeOperator):
    def __init__(self, address: str, node_url: str):
        super().__init__(address, node_url)
        if not self.safe_tx_service:
            raise SafeServiceNotAvailable(
                f"Cannot configure tx service for network {self.network.name}"
            )
        self.require_all_signatures = (
            False  # It doesn't require all signatures to be present to send a tx
        )

    def approve_hash(self, hash_to_approve: HexBytes, sender: str) -> bool:
        raise NotImplementedError("Not supported when using tx service")

    def sign_message(
        self,
        eip712_message_path: Optional[str] = None,
    ) -> bool:
        if eip712_message_path:
            try:
                with open(eip712_message_path, "r") as message_file:
                    message = json.load(message_file)
                message_hash = eip712_encode_hash(message)
            except ValueError:
                raise ValueError
        else:
            print_formatted_text("EIP191 message to sign:")
            message = get_input()
            message_hash = defunct_hash_message(text=message)

        safe_message_hash = self.safe.get_message_hash(message_hash)
        eoa_signers, hw_wallet_signers = self.get_signers()
        # Safe transaction service just accept one signer to create a message
        signature = b""
        if eoa_signers:
            signature_dict = eoa_signers[0].unsafe_sign_hash(safe_message_hash)
            signature = signature_to_bytes(
                signature_dict["v"], signature_dict["r"], signature_dict["s"]
            )

        elif hw_wallet_signers:
            signature = SafeSignature.export_signatures(
                self.hw_wallet_manager.sign_message(
                    safe_message_hash, [hw_wallet_signers[0]]
                )
            )
        else:
            print_formatted_text(
                HTML("<ansired>At least one owner must be loaded</ansired>")
            )

        if self.safe_tx_service.post_message(self.address, message, signature):
            print_formatted_text(
                HTML(
                    f"<ansigreen>Message  with safe-message-hash {to_0x_hex_str(safe_message_hash)} was correctly created on Safe Transaction Service</ansigreen>"
                )
            )
            return True
        else:
            print_formatted_text(
                HTML(
                    "<ansired>Something went wrong creating message on Safe Transaction Service</ansired>"
                )
            )
            return False

    def confirm_message(self, safe_message_hash: bytes, sender: ChecksumAddress):
        # GET message
        try:
            safe_message = self.safe_tx_service.get_message(safe_message_hash)
        except SafeAPIException:
            print_formatted_text(
                HTML(
                    f"<ansired>Message with hash {to_0x_hex_str(safe_message_hash)} does not exist</ansired>"
                )
            )
        if not yes_or_no_question(
            f"Message: {safe_message['message']} \n Do you want to sign the following message?:"
        ):
            return False

        signer = self.search_account(sender)
        if not signer:
            print_formatted_text(
                HTML(f"<ansired>Owner with address {sender} was not loaded</ansired>")
            )

        if isinstance(signer, LocalAccount):
            signature = signer.unsafe_sign_hash(safe_message_hash).signature
        else:
            signature = SafeSignature.export_signatures(
                self.hw_wallet_manager.sign_message(safe_message_hash, [signer])
            )

        try:
            self.safe_tx_service.post_message_signature(safe_message_hash, signature)
        except SafeAPIException as e:
            print_formatted_text(
                HTML(f"<ansired>Message wasn't confirmed due an error: {e}</ansired>")
            )
            return False
        print_formatted_text(
            HTML(
                f"<ansigreen>Message with safe-message-hash {to_0x_hex_str(safe_message_hash)} was correctly confirmed on Safe Transaction Service</ansigreen>"
            )
        )
        return True

    def get_delegates(self):
        delegates = self.safe_tx_service.get_delegates(self.address)
        headers = ["delegate", "delegator", "label"]
        rows = []
        for delegate in delegates:
            row = [delegate["delegate"], delegate["delegator"], delegate["label"]]
            rows.append(row)
        print(tabulate(rows, headers=headers))
        return rows

    def add_delegate(self, delegate_address: str, label: str, signer_address: str):
        signer_account = self.search_account(signer_address)
        if not signer_account or not isinstance(signer_account, LocalAccount):
            raise AccountNotLoadedException(signer_address)
        elif signer_address not in self.safe_cli_info.owners:
            raise NonExistingOwnerException(signer_address)
        else:
            try:
                hash_to_sign = self.safe_tx_service.create_delegate_message_hash(
                    delegate_address
                )
                signature = signer_account.unsafe_sign_hash(hash_to_sign)
                self.safe_tx_service.add_delegate(
                    delegate_address,
                    signer_account.address,
                    label,
                    signature.signature,
                    safe_address=self.address,
                )
                return True
            except SafeAPIException:
                return False

    def remove_delegate(self, delegate_address: str, signer_address: str):
        signer_account = self.search_account(signer_address)
        if not signer_account or not isinstance(signer_account, LocalAccount):
            raise AccountNotLoadedException(signer_address)
        elif signer_address not in self.safe_cli_info.owners:
            raise NonExistingOwnerException(signer_address)
        else:
            try:
                hash_to_sign = self.safe_tx_service.create_delegate_message_hash(
                    delegate_address
                )
                signature = signer_account.unsafe_sign_hash(hash_to_sign)
                self.safe_tx_service.remove_delegate(
                    delegate_address,
                    signer_account.address,
                    signature.signature,
                    safe_address=self.address,
                )
                return True
            except SafeAPIException:
                return False

    def submit_signatures(self, safe_tx_hash: bytes) -> bool:
        """
        Submit signatures to the tx service

        :return:
        """

        safe_tx, tx_hash = self.safe_tx_service.get_safe_transaction(safe_tx_hash)
        safe_tx.signatures = b""  # Don't post again existing signatures
        if tx_hash:
            print_formatted_text(
                HTML(
                    f"<ansired>Tx with safe-tx-hash {to_0x_hex_str(safe_tx_hash)} "
                    f"has already been executed on {to_0x_hex_str(tx_hash)}</ansired>"
                )
            )
        else:
            safe_tx = self.sign_transaction(safe_tx)
            if safe_tx.signers:
                self.safe_tx_service.post_signatures(safe_tx_hash, safe_tx.signatures)
                print_formatted_text(
                    HTML(
                        f"<ansigreen>{len(safe_tx.signers)} signatures were submitted to the tx service</ansigreen>"
                    )
                )
                return True
            else:
                print_formatted_text(
                    HTML(
                        "<ansired>Cannot generate signatures as there were no suitable signers</ansired>"
                    )
                )
        return False

    def batch_txs(self, safe_nonce: int, safe_tx_hashes: Sequence[bytes]) -> bool:
        """
        Submit signatures to the tx service. It's recommended to be on Safe v1.3.0 to prevent issues
        with `safeTxGas` and gas estimation.

        :return:
        """

        try:
            multisend = MultiSend(ethereum_client=self.ethereum_client)
        except ValueError:
            print_formatted_text(
                HTML(
                    "<ansired>Multisend contract is not deployed on this network and it's required for "
                    "batching txs</ansired>"
                )
            )
            return False

        multisend_txs = []
        for safe_tx_hash in safe_tx_hashes:
            safe_tx, _ = self.safe_tx_service.get_safe_transaction(safe_tx_hash)
            # Check if call is already a Multisend call
            inner_txs = MultiSend.from_transaction_data(safe_tx.data)
            if inner_txs:
                multisend_txs.extend(inner_txs)
            else:
                multisend_txs.append(
                    MultiSendTx(
                        MultiSendOperation.CALL, safe_tx.to, safe_tx.value, safe_tx.data
                    )
                )

        if len(multisend_txs) > 1:
            safe_tx = SafeTx(
                self.ethereum_client,
                self.address,
                multisend.address,
                0,
                multisend.build_tx_data(multisend_txs),
                SafeOperationEnum.DELEGATE_CALL.value,
                0,
                0,
                0,
                None,
                None,
                safe_nonce=safe_nonce,
            )
        else:
            safe_tx.safe_tx_gas = 0
            safe_tx.base_gas = 0
            safe_tx.gas_price = 0
            safe_tx.signatures = b""
            safe_tx.safe_nonce = safe_nonce  # Resend single transaction
        safe_tx = self.sign_transaction(safe_tx)
        if not safe_tx.signatures:
            print_formatted_text(
                HTML("<ansired>At least one owner must be loaded</ansired>")
            )
            return False
        else:
            return self.post_transaction_to_tx_service(safe_tx)

    def execute_tx(self, safe_tx_hash: Sequence[bytes]) -> bool:
        """
        Submit transaction on the tx-service to blockchain

        :return:
        """
        safe_tx, tx_hash = self.safe_tx_service.get_safe_transaction(safe_tx_hash)
        if tx_hash:
            print_formatted_text(
                HTML(
                    f"<ansired>Tx with safe-tx-hash {to_0x_hex_str(safe_tx_hash)} "
                    f"has already been executed on {to_0x_hex_str(tx_hash)}</ansired>"
                )
            )
        elif len(safe_tx.signers) < self.safe_cli_info.threshold:
            print_formatted_text(
                HTML(
                    f"<ansired>Number of signatures {len(safe_tx.signers)} "
                    f"must reach the threshold {self.safe_cli_info.threshold}</ansired>"
                )
            )
        else:
            if executed := self.execute_safe_transaction(safe_tx):
                self.refresh_safe_cli_info()
            return executed

    def get_balances(self):
        balances = self.safe_tx_service.get_balances(self.address)
        headers = ["name", "balance", "symbol", "decimals", "tokenAddress"]
        rows = []
        for balance in balances:
            if balance["tokenAddress"]:  # Token
                row = [
                    balance["token"]["name"],
                    f"{int(balance['balance']) / 10 ** int(balance['token']['decimals']):.5f}",
                    balance["token"]["symbol"],
                    balance["token"]["decimals"],
                    balance["tokenAddress"],
                ]
            else:  # Ether
                row = [
                    "ETHER",
                    f"{int(balance['balance']) / 10 ** 18:.5f}",
                    "Ξ",
                    18,
                    "",
                ]
            rows.append(row)
        print(tabulate(rows, headers=headers))
        return rows

    def get_transaction_history(self):
        transactions = self.safe_tx_service.get_transactions(self.address)
        headers = ["nonce", "to", "value", "transactionHash", "safeTxHash"]
        rows = []
        last_executed_tx = False
        for transaction in transactions:
            row = [transaction[header] for header in headers]
            data_decoded: Dict[str, Any] = transaction.get("dataDecoded")
            if data_decoded:
                row.append(self.safe_tx_service.data_decoded_to_text(data_decoded))
            if transaction["transactionHash"]:
                if not transaction["isSuccessful"]:
                    # Transaction failed
                    row[0] = Fore.RED + str(row[0])
                else:
                    row[0] = Fore.GREEN + str(
                        row[0]
                    )  # For executed transactions we use green
                    if not last_executed_tx:
                        row[0] = Style.BRIGHT + row[0]
                        last_executed_tx = True
            else:
                row[0] = Fore.YELLOW + str(
                    row[0]
                )  # For non executed transactions we use yellow

            row[0] = Style.RESET_ALL + row[0]  # Reset all just in case
            rows.append(row)

        headers.append("dataDecoded")
        headers[0] = Style.BRIGHT + headers[0]
        print(tabulate(rows, headers=headers))
        return rows

    def prepare_and_execute_safe_transaction(
        self,
        to: str,
        value: int,
        data: bytes,
        operation: SafeOperationEnum = SafeOperationEnum.CALL,
        safe_nonce: Optional[int] = None,
    ) -> bool:
        safe_tx = self.prepare_safe_transaction(
            to, value, data, operation, safe_nonce=safe_nonce
        )
        return self.post_transaction_to_tx_service(safe_tx)

    def post_transaction_to_tx_service(self, safe_tx: SafeTx) -> bool:
        if not yes_or_no_question(
            f"Do you want to send the tx with safe-tx-hash={to_0x_hex_str(safe_tx.safe_tx_hash)} to Safe Transaction Service (it will not be executed) "
            + str(safe_tx)
        ):
            return False

        self.safe_tx_service.post_transaction(safe_tx)
        print_formatted_text(
            HTML(
                f"<ansigreen>Tx with safe-tx-hash={to_0x_hex_str(safe_tx.safe_tx_hash)} was sent to Safe Transaction service</ansigreen>"
            )
        )
        return True

    def get_permitted_signers(self) -> Set[ChecksumAddress]:
        """
        :return: Owners and delegates, as they also can sign a transaction for the tx service
        """
        owners = super().get_permitted_signers()
        owners.update(
            [
                row["delegate"]
                for row in self.safe_tx_service.get_delegates(self.address)
            ]
        )
        return owners

    # Function that sends all assets to an account (to)
    def drain(self, to: ChecksumAddress):
        balances = self.safe_tx_service.get_balances(self.address)
        safe_txs = []
        safe_tx = None
        for balance in balances:
            amount = int(balance["balance"])
            if balance["tokenAddress"] is None:  # Then is ether
                if amount != 0:
                    safe_tx = self.prepare_safe_transaction(
                        to,
                        amount,
                        b"",
                        SafeOperationEnum.CALL,
                        safe_nonce=None,
                    )
            else:
                transaction = (
                    get_erc20_contract(self.ethereum_client.w3, balance["tokenAddress"])
                    .functions.transfer(to, amount)
                    .build_transaction({"from": self.address, "gas": 0, "gasPrice": 0})
                )
                safe_tx = self.prepare_safe_transaction(
                    balance["tokenAddress"],
                    0,
                    HexBytes(transaction["data"]),
                    SafeOperationEnum.CALL,
                    safe_nonce=None,
                )
            if safe_tx:
                safe_txs.append(safe_tx)
        if len(safe_txs) > 0:
            multisend_tx = self.batch_safe_txs(safe_tx.safe_nonce, safe_txs)
            if multisend_tx is not None:
                self.post_transaction_to_tx_service(multisend_tx)
                print_formatted_text(
                    HTML(
                        "<ansigreen>Transaction to drain account correctly created</ansigreen>"
                    )
                )
        else:
            print_formatted_text(
                HTML("<ansigreen>Safe account is currently empty</ansigreen>")
            )

    def remove_proposed_transaction(self, safe_tx_hash: bytes):
        eip712_message = get_remove_transaction_message(
            self.address, safe_tx_hash, self.ethereum_client.get_chain_id()
        )
        message_hash = eip712_encode_hash(eip712_message)
        try:
            safe_tx, _ = self.safe_tx_service.get_safe_transaction(safe_tx_hash)
            signer = self.search_account(safe_tx.proposer)
            if not signer:
                print_formatted_text(
                    HTML(
                        f"<ansired>The proposer with address: {safe_tx.proposer} was not loaded</ansired>"
                    )
                )
                return False

            if isinstance(signer, LocalAccount):
                signature = signer.unsafe_sign_hash(message_hash).signature
            else:
                signature = self.hw_wallet_manager.sign_eip712(
                    eip712_message, [signer]
                )[0].signature

            if len(safe_tx.signers) >= self.safe.retrieve_threshold():
                print_formatted_text(
                    HTML(
                        "<ansired>The transaction has all the required signatures to be executed!!!\n"
                        "This means that the transaction can be executed by a 3rd party monitoring your Safe even after removal!\n"
                        f"Make sure you execute a transaction with nonce {safe_tx.safe_nonce} to void the current transaction"
                        "</ansired>"
                    )
                )

            if not yes_or_no_question(
                f"Do you want to remove the tx with safe-tx-hash={to_0x_hex_str(safe_tx.safe_tx_hash)}"
            ):
                return False

            self.safe_tx_service.delete_transaction(
                to_0x_hex_str(safe_tx_hash), to_0x_hex_str(signature)
            )
            print_formatted_text(
                HTML(
                    f"<ansigreen>Transaction {to_0x_hex_str(safe_tx_hash)} was removed correctly</ansigreen>"
                )
            )
            return True
        except SafeAPIException as e:
            print_formatted_text(
                HTML(f"<ansired>Transaction wasn't removed due an error: {e}</ansired>")
            )
            return False

1	import json	4✔
2	from typing import Any, Dict, Optional, Sequence, Set	4✔
3
4	from colorama import Fore, Style	4✔
5	from eth_account.messages import defunct_hash_message	4✔
6	from eth_account.signers.local import LocalAccount	4✔
7	from eth_typing import ChecksumAddress	4✔
8	from hexbytes import HexBytes	4✔
9	from prompt_toolkit import HTML, print_formatted_text	4✔
10	from safe_eth.eth.contracts import get_erc20_contract	4✔
11	from safe_eth.eth.eip712 import eip712_encode_hash	4✔
12	from safe_eth.safe import SafeOperationEnum, SafeTx	4✔
13	from safe_eth.safe.api import SafeAPIException	4✔
14	from safe_eth.safe.api.transaction_service_api.transaction_service_messages import (	4✔
15	get_remove_transaction_message,
16	)
17	from safe_eth.safe.multi_send import MultiSend, MultiSendOperation, MultiSendTx	4✔
18	from safe_eth.safe.safe_signature import SafeSignature	4✔
19	from safe_eth.safe.signatures import signature_to_bytes	4✔
20	from safe_eth.util.util import to_0x_hex_str	4✔
21	from tabulate import tabulate	4✔
22
23	from ..utils import get_input, yes_or_no_question	4✔
24	from . import SafeServiceNotAvailable	4✔
25	from .exceptions import AccountNotLoadedException, NonExistingOwnerException	4✔
26	from .safe_operator import SafeOperator	4✔
27
28
29	class SafeTxServiceOperator(SafeOperator):	4✔
30	def __init__(self, address: str, node_url: str):	4✔
31	super().__init__(address, node_url)	4✔
32	if not self.safe_tx_service:	4✔
33	raise SafeServiceNotAvailable(	4✔
34	f"Cannot configure tx service for network {self.network.name}"
35	)
36	self.require_all_signatures = (	4✔
37	False # It doesn't require all signatures to be present to send a tx
38	)
39
40	def approve_hash(self, hash_to_approve: HexBytes, sender: str) -> bool:	4✔
41	raise NotImplementedError("Not supported when using tx service")
42
43	def sign_message(	4✔
44	self,
45	eip712_message_path: Optional[str] = None,
46	) -> bool:
47	if eip712_message_path:	×
48	try:	×
NEW 49	with open(eip712_message_path, "r") as message_file:	×
NEW 50	message = json.load(message_file)	×
51	message_hash = eip712_encode_hash(message)	×
52	except ValueError:	×
53	raise ValueError	×
54	else:
55	print_formatted_text("EIP191 message to sign:")	×
56	message = get_input()	×
57	message_hash = defunct_hash_message(text=message)	×
58
59	safe_message_hash = self.safe.get_message_hash(message_hash)	×
60	eoa_signers, hw_wallet_signers = self.get_signers()	×
61	# Safe transaction service just accept one signer to create a message
62	signature = b""	×
63	if eoa_signers:	×
64	signature_dict = eoa_signers[0].unsafe_sign_hash(safe_message_hash)	×
65	signature = signature_to_bytes(	×
66	signature_dict["v"], signature_dict["r"], signature_dict["s"]
67	)
68
69	elif hw_wallet_signers:	×
70	signature = SafeSignature.export_signatures(	×
71	self.hw_wallet_manager.sign_message(
72	safe_message_hash, [hw_wallet_signers[0]]
73	)
74	)
75	else:
76	print_formatted_text(	×
77	HTML("<ansired>At least one owner must be loaded</ansired>")
78	)
79
80	if self.safe_tx_service.post_message(self.address, message, signature):	×
81	print_formatted_text(	×
82	HTML(
83	f"<ansigreen>Message with safe-message-hash {to_0x_hex_str(safe_message_hash)} was correctly created on Safe Transaction Service</ansigreen>"
84	)
85	)
86	return True	×
87	else:
88	print_formatted_text(	×
89	HTML(
90	"<ansired>Something went wrong creating message on Safe Transaction Service</ansired>"
91	)
92	)
93	return False	×
94
95	def confirm_message(self, safe_message_hash: bytes, sender: ChecksumAddress):	4✔
96	# GET message
97	try:	4✔
98	safe_message = self.safe_tx_service.get_message(safe_message_hash)	4✔
99	except SafeAPIException:	×
100	print_formatted_text(	×
101	HTML(
102	f"<ansired>Message with hash {to_0x_hex_str(safe_message_hash)} does not exist</ansired>"
103	)
104	)
105	if not yes_or_no_question(	4✔
106	f"Message: {safe_message['message']} \n Do you want to sign the following message?:"
107	):
108	return False	×
109
110	signer = self.search_account(sender)	4✔
111	if not signer:	4✔
112	print_formatted_text(	×
113	HTML(f"<ansired>Owner with address {sender} was not loaded</ansired>")
114	)
115
116	if isinstance(signer, LocalAccount):	4✔
117	signature = signer.unsafe_sign_hash(safe_message_hash).signature	4✔
118	else:
119	signature = SafeSignature.export_signatures(	×
120	self.hw_wallet_manager.sign_message(safe_message_hash, [signer])
121	)
122
123	try:	4✔
124	self.safe_tx_service.post_message_signature(safe_message_hash, signature)	4✔
125	except SafeAPIException as e:	4✔
126	print_formatted_text(	4✔
127	HTML(f"<ansired>Message wasn't confirmed due an error: {e}</ansired>")
128	)
129	return False	4✔
130	print_formatted_text(	4✔
131	HTML(
132	f"<ansigreen>Message with safe-message-hash {to_0x_hex_str(safe_message_hash)} was correctly confirmed on Safe Transaction Service</ansigreen>"
133	)
134	)
135	return True	4✔
136
137	def get_delegates(self):	4✔
138	delegates = self.safe_tx_service.get_delegates(self.address)	4✔
139	headers = ["delegate", "delegator", "label"]	4✔
140	rows = []	4✔
141	for delegate in delegates:	4✔
142	row = [delegate["delegate"], delegate["delegator"], delegate["label"]]	4✔
143	rows.append(row)	4✔
144	print(tabulate(rows, headers=headers))	4✔
145	return rows	4✔
146
147	def add_delegate(self, delegate_address: str, label: str, signer_address: str):	4✔
148	signer_account = self.search_account(signer_address)	4✔
149	if not signer_account or not isinstance(signer_account, LocalAccount):	4✔
UNCOV 150	raise AccountNotLoadedException(signer_address)	×
151	elif signer_address not in self.safe_cli_info.owners:	4✔
152	raise NonExistingOwnerException(signer_address)	×
153	else:
154	try:	4✔
155	hash_to_sign = self.safe_tx_service.create_delegate_message_hash(	4✔
156	delegate_address
157	)
158	signature = signer_account.unsafe_sign_hash(hash_to_sign)	4✔
159	self.safe_tx_service.add_delegate(	4✔
160	delegate_address,
161	signer_account.address,
162	label,
163	signature.signature,
164	safe_address=self.address,
165	)
166	return True	4✔
167	except SafeAPIException:	×
168	return False	×
169
170	def remove_delegate(self, delegate_address: str, signer_address: str):	4✔
171	signer_account = self.search_account(signer_address)	4✔
172	if not signer_account or not isinstance(signer_account, LocalAccount):	4✔
UNCOV 173	raise AccountNotLoadedException(signer_address)	×
174	elif signer_address not in self.safe_cli_info.owners:	4✔
175	raise NonExistingOwnerException(signer_address)	×
176	else:
177	try:	4✔
178	hash_to_sign = self.safe_tx_service.create_delegate_message_hash(	4✔
179	delegate_address
180	)
181	signature = signer_account.unsafe_sign_hash(hash_to_sign)	4✔
182	self.safe_tx_service.remove_delegate(	4✔
183	delegate_address,
184	signer_account.address,
185	signature.signature,
186	safe_address=self.address,
187	)
188	return True	4✔
189	except SafeAPIException:	×
190	return False	×
191
192	def submit_signatures(self, safe_tx_hash: bytes) -> bool:	4✔
193	"""
194	Submit signatures to the tx service
195
196	:return:
197	"""
198
199	safe_tx, tx_hash = self.safe_tx_service.get_safe_transaction(safe_tx_hash)	4✔
200	safe_tx.signatures = b"" # Don't post again existing signatures	4✔
201	if tx_hash:	4✔
202	print_formatted_text(	4✔
203	HTML(
204	f"<ansired>Tx with safe-tx-hash {to_0x_hex_str(safe_tx_hash)} "
205	f"has already been executed on {to_0x_hex_str(tx_hash)}</ansired>"
206	)
207	)
208	else:
209	safe_tx = self.sign_transaction(safe_tx)	4✔
210	if safe_tx.signers:	4✔
211	self.safe_tx_service.post_signatures(safe_tx_hash, safe_tx.signatures)	4✔
212	print_formatted_text(	4✔
213	HTML(
214	f"<ansigreen>{len(safe_tx.signers)} signatures were submitted to the tx service</ansigreen>"
215	)
216	)
217	return True	4✔
218	else:
219	print_formatted_text(	4✔
220	HTML(
221	"<ansired>Cannot generate signatures as there were no suitable signers</ansired>"
222	)
223	)
224	return False	4✔
225
226	def batch_txs(self, safe_nonce: int, safe_tx_hashes: Sequence[bytes]) -> bool:	4✔
227	"""
228	Submit signatures to the tx service. It's recommended to be on Safe v1.3.0 to prevent issues
229	with `safeTxGas` and gas estimation.
230
231	:return:
232	"""
233
234	try:	4✔
235	multisend = MultiSend(ethereum_client=self.ethereum_client)	4✔
236	except ValueError:	×
237	print_formatted_text(	×
238	HTML(
239	"<ansired>Multisend contract is not deployed on this network and it's required for "
240	"batching txs</ansired>"
241	)
242	)
243	return False	×
244
245	multisend_txs = []	4✔
246	for safe_tx_hash in safe_tx_hashes:	4✔
247	safe_tx, _ = self.safe_tx_service.get_safe_transaction(safe_tx_hash)	4✔
248	# Check if call is already a Multisend call
249	inner_txs = MultiSend.from_transaction_data(safe_tx.data)	4✔
250	if inner_txs:	4✔
251	multisend_txs.extend(inner_txs)	×
252	else:
253	multisend_txs.append(	4✔
254	MultiSendTx(
255	MultiSendOperation.CALL, safe_tx.to, safe_tx.value, safe_tx.data
256	)
257	)
258
259	if len(multisend_txs) > 1:	4✔
260	safe_tx = SafeTx(	×
261	self.ethereum_client,
262	self.address,
263	multisend.address,
264	0,
265	multisend.build_tx_data(multisend_txs),
266	SafeOperationEnum.DELEGATE_CALL.value,
267	0,
268	0,
269	0,
270	None,
271	None,
272	safe_nonce=safe_nonce,
273	)
274	else:
275	safe_tx.safe_tx_gas = 0	4✔
276	safe_tx.base_gas = 0	4✔
277	safe_tx.gas_price = 0	4✔
278	safe_tx.signatures = b""	4✔
279	safe_tx.safe_nonce = safe_nonce # Resend single transaction	4✔
280	safe_tx = self.sign_transaction(safe_tx)	4✔
281	if not safe_tx.signatures:	4✔
282	print_formatted_text(	×
283	HTML("<ansired>At least one owner must be loaded</ansired>")
284	)
285	return False	×
286	else:
287	return self.post_transaction_to_tx_service(safe_tx)	4✔
288
289	def execute_tx(self, safe_tx_hash: Sequence[bytes]) -> bool:	4✔
290	"""
291	Submit transaction on the tx-service to blockchain
292
293	:return:
294	"""
295	safe_tx, tx_hash = self.safe_tx_service.get_safe_transaction(safe_tx_hash)	×
296	if tx_hash:	×
297	print_formatted_text(	×
298	HTML(
299	f"<ansired>Tx with safe-tx-hash {to_0x_hex_str(safe_tx_hash)} "
300	f"has already been executed on {to_0x_hex_str(tx_hash)}</ansired>"
301	)
302	)
303	elif len(safe_tx.signers) < self.safe_cli_info.threshold:	×
304	print_formatted_text(	×
305	HTML(
306	f"<ansired>Number of signatures {len(safe_tx.signers)} "
307	f"must reach the threshold {self.safe_cli_info.threshold}</ansired>"
308	)
309	)
310	else:
311	if executed := self.execute_safe_transaction(safe_tx):	×
312	self.refresh_safe_cli_info()	×
313	return executed	×
314
315	def get_balances(self):	4✔
316	balances = self.safe_tx_service.get_balances(self.address)	4✔
317	headers = ["name", "balance", "symbol", "decimals", "tokenAddress"]	4✔
318	rows = []	4✔
319	for balance in balances:	4✔
320	if balance["tokenAddress"]: # Token	4✔
321	row = [	4✔
322	balance["token"]["name"],
323	f"{int(balance['balance']) / 10 ** int(balance['token']['decimals']):.5f}",
324	balance["token"]["symbol"],
325	balance["token"]["decimals"],
326	balance["tokenAddress"],
327	]
328	else: # Ether
329	row = [	4✔
330	"ETHER",
331	f"{int(balance['balance']) / 10 ** 18:.5f}",
332	"Ξ",
333	18,
334	"",
335	]
336	rows.append(row)	4✔
337	print(tabulate(rows, headers=headers))	4✔
338	return rows	4✔
339
340	def get_transaction_history(self):	4✔
341	transactions = self.safe_tx_service.get_transactions(self.address)	4✔
342	headers = ["nonce", "to", "value", "transactionHash", "safeTxHash"]	4✔
343	rows = []	4✔
344	last_executed_tx = False	4✔
345	for transaction in transactions:	4✔
346	row = [transaction[header] for header in headers]	4✔
347	data_decoded: Dict[str, Any] = transaction.get("dataDecoded")	4✔
348	if data_decoded:	4✔
349	row.append(self.safe_tx_service.data_decoded_to_text(data_decoded))	4✔
350	if transaction["transactionHash"]:	4✔
351	if not transaction["isSuccessful"]:	4✔
352	# Transaction failed
353	row[0] = Fore.RED + str(row[0])	×
354	else:
355	row[0] = Fore.GREEN + str(	4✔
356	row[0]
357	) # For executed transactions we use green
358	if not last_executed_tx:	4✔
359	row[0] = Style.BRIGHT + row[0]	4✔
360	last_executed_tx = True	4✔
361	else:
362	row[0] = Fore.YELLOW + str(	×
363	row[0]
364	) # For non executed transactions we use yellow
365
366	row[0] = Style.RESET_ALL + row[0] # Reset all just in case	4✔
367	rows.append(row)	4✔
368
369	headers.append("dataDecoded")	4✔
370	headers[0] = Style.BRIGHT + headers[0]	4✔
371	print(tabulate(rows, headers=headers))	4✔
372	return rows	4✔
373
374	def prepare_and_execute_safe_transaction(	4✔
375	self,
376	to: str,
377	value: int,
378	data: bytes,
379	operation: SafeOperationEnum = SafeOperationEnum.CALL,
380	safe_nonce: Optional[int] = None,
381	) -> bool:
382	safe_tx = self.prepare_safe_transaction(	×
383	to, value, data, operation, safe_nonce=safe_nonce
384	)
385	return self.post_transaction_to_tx_service(safe_tx)	×
386
387	def post_transaction_to_tx_service(self, safe_tx: SafeTx) -> bool:	4✔
388	if not yes_or_no_question(	4✔
389	f"Do you want to send the tx with safe-tx-hash={to_0x_hex_str(safe_tx.safe_tx_hash)} to Safe Transaction Service (it will not be executed) "
390	+ str(safe_tx)
391	):
392	return False	×
393
394	self.safe_tx_service.post_transaction(safe_tx)	4✔
395	print_formatted_text(	4✔
396	HTML(
397	f"<ansigreen>Tx with safe-tx-hash={to_0x_hex_str(safe_tx.safe_tx_hash)} was sent to Safe Transaction service</ansigreen>"
398	)
399	)
400	return True	4✔
401
402	def get_permitted_signers(self) -> Set[ChecksumAddress]:	4✔
403	"""
404	:return: Owners and delegates, as they also can sign a transaction for the tx service
405	"""
406	owners = super().get_permitted_signers()	4✔
407	owners.update(	4✔
408	[
409	row["delegate"]
410	for row in self.safe_tx_service.get_delegates(self.address)
411	]
412	)
413	return owners	4✔
414
415	# Function that sends all assets to an account (to)
416	def drain(self, to: ChecksumAddress):	4✔
417	balances = self.safe_tx_service.get_balances(self.address)	×
418	safe_txs = []	×
419	safe_tx = None	×
420	for balance in balances:	×
421	amount = int(balance["balance"])	×
422	if balance["tokenAddress"] is None: # Then is ether	×
423	if amount != 0:	×
424	safe_tx = self.prepare_safe_transaction(	×
425	to,
426	amount,
427	b"",
428	SafeOperationEnum.CALL,
429	safe_nonce=None,
430	)
431	else:
432	transaction = (	×
433	get_erc20_contract(self.ethereum_client.w3, balance["tokenAddress"])
434	.functions.transfer(to, amount)
435	.build_transaction({"from": self.address, "gas": 0, "gasPrice": 0})
436	)
437	safe_tx = self.prepare_safe_transaction(	×
438	balance["tokenAddress"],
439	0,
440	HexBytes(transaction["data"]),
441	SafeOperationEnum.CALL,
442	safe_nonce=None,
443	)
444	if safe_tx:	×
445	safe_txs.append(safe_tx)	×
446	if len(safe_txs) > 0:	×
447	multisend_tx = self.batch_safe_txs(safe_tx.safe_nonce, safe_txs)	×
448	if multisend_tx is not None:	×
449	self.post_transaction_to_tx_service(multisend_tx)	×
450	print_formatted_text(	×
451	HTML(
452	"<ansigreen>Transaction to drain account correctly created</ansigreen>"
453	)
454	)
455	else:
456	print_formatted_text(	×
457	HTML("<ansigreen>Safe account is currently empty</ansigreen>")
458	)
459
460	def remove_proposed_transaction(self, safe_tx_hash: bytes):	4✔
461	eip712_message = get_remove_transaction_message(	4✔
462	self.address, safe_tx_hash, self.ethereum_client.get_chain_id()
463	)
464	message_hash = eip712_encode_hash(eip712_message)	4✔
465	try:	4✔
466	safe_tx, _ = self.safe_tx_service.get_safe_transaction(safe_tx_hash)	4✔
467	signer = self.search_account(safe_tx.proposer)	4✔
468	if not signer:	4✔
469	print_formatted_text(	4✔
470	HTML(
471	f"<ansired>The proposer with address: {safe_tx.proposer} was not loaded</ansired>"
472	)
473	)
474	return False	4✔
475
476	if isinstance(signer, LocalAccount):	4✔
477	signature = signer.unsafe_sign_hash(message_hash).signature	4✔
478	else:
479	signature = self.hw_wallet_manager.sign_eip712(	×
480	eip712_message, [signer]
481	)[0].signature
482
483	if len(safe_tx.signers) >= self.safe.retrieve_threshold():	4✔
484	print_formatted_text(	×
485	HTML(
486	"<ansired>The transaction has all the required signatures to be executed!!!\n"
487	"This means that the transaction can be executed by a 3rd party monitoring your Safe even after removal!\n"
488	f"Make sure you execute a transaction with nonce {safe_tx.safe_nonce} to void the current transaction"
489	"</ansired>"
490	)
491	)
492
493	if not yes_or_no_question(	4✔
494	f"Do you want to remove the tx with safe-tx-hash={to_0x_hex_str(safe_tx.safe_tx_hash)}"
495	):
496	return False	×
497
498	self.safe_tx_service.delete_transaction(	4✔
499	to_0x_hex_str(safe_tx_hash), to_0x_hex_str(signature)
500	)
501	print_formatted_text(	4✔
502	HTML(
503	f"<ansigreen>Transaction {to_0x_hex_str(safe_tx_hash)} was removed correctly</ansigreen>"
504	)
505	)
506	return True	4✔
507	except SafeAPIException as e:	×
508	print_formatted_text(	×
509	HTML(f"<ansired>Transaction wasn't removed due an error: {e}</ansired>")
510	)
511	return False	×

safe-global / safe-cli / 19675107724

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous