9464772673

Committed 11 Jun 2024 11:32AM UTC coverage: 87.16% (+6.5%) from 80.617%

Build # 9464772673

Build Type

push

github

Committed by

web-flow

Commit Message

Update project to use hatch (#404)

* Update project to use hatch

- Update pre-commit, as black previous version was having issues
- Use recommended structure: https://docs.pytest.org/en/stable/explanation/goodpractices.html
- Update CI
- Add run_tests.sh script

* Fix coverage

* Fix version

* Fix module export

* Fix linting

---------

Co-authored-by: Uxio Fuentefria <6909403+Uxio0@users.noreply.github.com>

Run Details

722 of 835 branches covered (86.47%)

Branch coverage included in aggregate %.

3 of 3 new or added lines in 2 files covered. (100.0%)

2326 of 2662 relevant lines covered (87.38%)

3.49 hits per line

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

70.94

/src/safe_cli/operators/safe_tx_service_operator.py

import json
from itertools import chain
from typing import Any, Dict, List, Optional, Sequence, Set, Union

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 tabulate import tabulate

from gnosis.eth.contracts import get_erc20_contract
from gnosis.eth.eip712 import eip712_encode_hash
from gnosis.safe import SafeOperationEnum, SafeTx
from gnosis.safe.api import SafeAPIException
from gnosis.safe.api.transaction_service_api.transaction_service_messages import (
    get_remove_transaction_message,
)
from gnosis.safe.multi_send import MultiSend, MultiSendOperation, MultiSendTx
from gnosis.safe.safe_signature import SafeSignature, SafeSignatureEOA
from gnosis.safe.signatures import signature_to_bytes

from safe_cli.utils import yes_or_no_question

from . import SafeServiceNotAvailable
from .exceptions import AccountNotLoadedException, NonExistingOwnerException
from .hw_wallets.hw_wallet import HwWallet
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,
        eip191_message: Optional[str] = None,
        eip712_message_path: Optional[str] = None,
    ) -> bool:
        if eip712_message_path:
            try:
                message = json.load(open(eip712_message_path, "r"))
                message_hash = eip712_encode_hash(message)
            except ValueError:
                raise ValueError
        else:
            message = eip191_message
            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_signatures: List[SafeSignature] = []
        for eoa_signer in eoa_signers:
            signature_dict = eoa_signer.signHash(safe_message_hash)
            signature = signature_to_bytes(
                signature_dict["v"], signature_dict["r"], signature_dict["s"]
            )
            safe_signatures.append(SafeSignatureEOA(signature, safe_message_hash))

        signatures = SafeSignature.export_signatures(safe_signatures)

        if hw_wallet_signers:
            print_formatted_text(
                HTML(
                    "<ansired>Signing messages is not currently supported by hardware wallets</ansired>"
                )
            )
            return False

        if self.safe_tx_service.post_message(self.address, message, signatures):
            print_formatted_text(
                HTML(
                    "<ansigreen>Message 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 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 = [
            account for account in self.accounts if account.address == signer_address
        ]
        if not signer_account:
            raise AccountNotLoadedException(signer_address)
        elif signer_address not in self.safe_cli_info.owners:
            raise NonExistingOwnerException(signer_address)
        else:
            signer_account = signer_account[0]
            try:
                self.safe_tx_service.add_delegate(
                    self.address, delegate_address, label, signer_account
                )
                return True
            except SafeAPIException:
                return False

    def remove_delegate(self, delegate_address: str, signer_address: str):
        signer_account = [
            account for account in self.accounts if account.address == signer_address
        ]
        if not signer_account:
            raise AccountNotLoadedException(signer_address)
        elif signer_address not in self.safe_cli_info.owners:
            raise NonExistingOwnerException(signer_address)
        else:
            signer_account = signer_account[0]
            try:
                self.safe_tx_service.remove_delegate(
                    self.address, delegate_address, signer_account
                )
                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 {safe_tx_hash.hex()} "
                    f"has already been executed on {tx_hash.hex()}</ansired>"
                )
            )
        else:
            owners = self.get_permitted_signers()
            for account in self.accounts:
                if account.address in owners:
                    safe_tx.sign(account.key)
            # Check if there are ledger signers
            if self.hw_wallet_manager.wallets:
                selected_ledger_accounts = []
                for ledger_account in self.hw_wallet_manager.wallets:
                    if ledger_account.address in owners:
                        selected_ledger_accounts.append(ledger_account)
                if len(selected_ledger_accounts) > 0:
                    safe_tx = self.hw_wallet_manager.sign_safe_tx(
                        safe_tx, selected_ledger_accounts
                    )

            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>"
                )
            )

        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 {safe_tx_hash.hex()} "
                    f"has already been executed on {tx_hash.hex()}</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:
            return self.execute_safe_transaction(safe_tx)

    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={safe_tx.safe_tx_hash.hex()} 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={safe_tx.safe_tx_hash.hex()} 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 search_account(
        self, address: ChecksumAddress
    ) -> Optional[Union[LocalAccount, HwWallet]]:
        """
        Search the provided address between loaded owners

        :param address:
        :return: LocalAccount or HwWallet of the provided address
        """
        for account in chain(self.accounts, self.hw_wallet_manager.wallets):
            if account.address == address:
                return account

    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.signHash(message_hash).signature
            else:
                signature = self.hw_wallet_manager.sign_eip712(eip712_message, [signer])

            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={safe_tx.safe_tx_hash.hex()}"
            ):
                return False

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

safe-global / safe-cli / 9464772673

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