7087242091

Committed 04 Dec 2023 01:23PM UTC coverage: 93.371%. Remained the same

Build # 7087242091

Build Type

Pull #320

github

Committed by

web-flow

Commit Message

Merge 489f75539 into 443cd5f78

Pull Request Pull Request #320: Bump web3 from 6.11.3 to 6.11.4

Run Details

1803 of 1931 relevant lines covered (93.37%)

3.69 hits per line

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

86.8

/safe_cli/operators/safe_tx_service_operator.py

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

from colorama import Fore, Style
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.safe import SafeOperation, SafeTx
from gnosis.safe.api import SafeAPIException
from gnosis.safe.multi_send import MultiSend, MultiSendOperation, MultiSendTx

from safe_cli.utils import 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 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.ledger_manager:
                selected_ledger_accounts = []
                for ledger_account in self.ledger_manager.accounts:
                    if ledger_account.address in owners:
                        selected_ledger_accounts.append(ledger_account)
                if len(selected_ledger_accounts) > 0:
                    safe_tx = self.ledger_manager.sign_eip712(
                        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),
                SafeOperation.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: SafeOperation = SafeOperation.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"",
                        SafeOperation.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"]),
                    SafeOperation.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>")
            )

1	from typing import Any, Dict, Optional, Sequence, Set	4✔
2
3	from colorama import Fore, Style	4✔
4	from eth_typing import ChecksumAddress	4✔
5	from hexbytes import HexBytes	4✔
6	from prompt_toolkit import HTML, print_formatted_text	4✔
7	from tabulate import tabulate	4✔
8
9	from gnosis.eth.contracts import get_erc20_contract	4✔
10	from gnosis.safe import SafeOperation, SafeTx	4✔
11	from gnosis.safe.api import SafeAPIException	4✔
12	from gnosis.safe.multi_send import MultiSend, MultiSendOperation, MultiSendTx	4✔
13
14	from safe_cli.utils import yes_or_no_question	4✔
15
16	from . import SafeServiceNotAvailable	4✔
17	from .exceptions import AccountNotLoadedException, NonExistingOwnerException	4✔
18	from .safe_operator import SafeOperator	4✔
19
20
21	class SafeTxServiceOperator(SafeOperator):	4✔
22	def __init__(self, address: str, node_url: str):	4✔
23	super().__init__(address, node_url)	4✔
24	if not self.safe_tx_service:	4✔
25	raise SafeServiceNotAvailable(	4✔
26	f"Cannot configure tx service for network {self.network.name}"	4✔
27	)
28	self.require_all_signatures = (	4✔
29	False # It doesn't require all signatures to be present to send a tx	4✔
30	)
31
32	def approve_hash(self, hash_to_approve: HexBytes, sender: str) -> bool:	4✔
33	raise NotImplementedError("Not supported when using tx service")	4✔
34
35	def get_delegates(self):	4✔
36	delegates = self.safe_tx_service.get_delegates(self.address)	4✔
37	headers = ["delegate", "delegator", "label"]	4✔
38	rows = []	4✔
39	for delegate in delegates:	4✔
40	row = [delegate["delegate"], delegate["delegator"], delegate["label"]]	4✔
41	rows.append(row)	4✔
42	print(tabulate(rows, headers=headers))	4✔
43	return rows	4✔
44
45	def add_delegate(self, delegate_address: str, label: str, signer_address: str):	4✔
46	signer_account = [	4✔
47	account for account in self.accounts if account.address == signer_address	4✔
48	]
49	if not signer_account:	4✔
50	raise AccountNotLoadedException(signer_address)	×
51	elif signer_address not in self.safe_cli_info.owners:	4✔
52	raise NonExistingOwnerException(signer_address)	×
53	else:	×
54	signer_account = signer_account[0]	4✔
55	try:	4✔
56	self.safe_tx_service.add_delegate(	4✔
57	self.address, delegate_address, label, signer_account	4✔
58	)
59	return True	4✔
60	except SafeAPIException:	×
61	return False	×
62
63	def remove_delegate(self, delegate_address: str, signer_address: str):	4✔
64	signer_account = [	4✔
65	account for account in self.accounts if account.address == signer_address	4✔
66	]
67	if not signer_account:	4✔
68	raise AccountNotLoadedException(signer_address)	×
69	elif signer_address not in self.safe_cli_info.owners:	4✔
70	raise NonExistingOwnerException(signer_address)	×
71	else:	×
72	signer_account = signer_account[0]	4✔
73	try:	4✔
74	self.safe_tx_service.remove_delegate(	4✔
75	self.address, delegate_address, signer_account	4✔
76	)
77	return True	4✔
78	except SafeAPIException:	×
79	return False	×
80
81	def submit_signatures(self, safe_tx_hash: bytes) -> bool:	4✔
82	"""
83	Submit signatures to the tx service
84
85	:return:
86	"""
87
88	safe_tx, tx_hash = self.safe_tx_service.get_safe_transaction(safe_tx_hash)	4✔
89	safe_tx.signatures = b"" # Don't post again existing signatures	4✔
90	if tx_hash:	4✔
91	print_formatted_text(	4✔
92	HTML(	4✔
93	f"<ansired>Tx with safe-tx-hash {safe_tx_hash.hex()} "	4✔
94	f"has already been executed on {tx_hash.hex()}</ansired>"	3✔
95	)
96	)
97	else:	×
98	owners = self.get_permitted_signers()	4✔
99	for account in self.accounts:	4✔
100	if account.address in owners:	4✔
101	safe_tx.sign(account.key)	4✔
102	# Check if there are ledger signers
103	if self.ledger_manager:	4✔
104	selected_ledger_accounts = []	4✔
105	for ledger_account in self.ledger_manager.accounts:	4✔
106	if ledger_account.address in owners:	4✔
107	selected_ledger_accounts.append(ledger_account)	4✔
108	if len(selected_ledger_accounts) > 0:	4✔
109	safe_tx = self.ledger_manager.sign_eip712(	4✔
110	safe_tx, selected_ledger_accounts	4✔
111	)
112
113	if safe_tx.signers:	4✔
114	self.safe_tx_service.post_signatures(safe_tx_hash, safe_tx.signatures)	4✔
115	print_formatted_text(	4✔
116	HTML(	4✔
117	f"<ansigreen>{len(safe_tx.signers)} signatures were submitted to the tx service</ansigreen>"	4✔
118	)
119	)
120	return True	4✔
121	else:	×
122	print_formatted_text(	4✔
123	HTML(	4✔
124	"<ansired>Cannot generate signatures as there were no suitable signers</ansired>"	4✔
125	)
126	)
127	return False	4✔
128
129	def batch_txs(self, safe_nonce: int, safe_tx_hashes: Sequence[bytes]) -> bool:	4✔
130	"""
131	Submit signatures to the tx service. It's recommended to be on Safe v1.3.0 to prevent issues
132	with `safeTxGas` and gas estimation.
133
134	:return:
135	"""
136
137	try:	4✔
138	multisend = MultiSend(ethereum_client=self.ethereum_client)	4✔
139	except ValueError:	×
140	print_formatted_text(
141	HTML(
142	"<ansired>Multisend contract is not deployed on this network and it's required for "
143	"batching txs</ansired>"
144	)
145	)
146
147	multisend_txs = []	4✔
148	for safe_tx_hash in safe_tx_hashes:	4✔
149	safe_tx, _ = self.safe_tx_service.get_safe_transaction(safe_tx_hash)	4✔
150	# Check if call is already a Multisend call
151	inner_txs = MultiSend.from_transaction_data(safe_tx.data)	4✔
152	if inner_txs:	4✔
153	multisend_txs.extend(inner_txs)	×
154	else:	×
155	multisend_txs.append(	4✔
156	MultiSendTx(	4✔
157	MultiSendOperation.CALL, safe_tx.to, safe_tx.value, safe_tx.data	4✔
158	)
159	)
160
161	if len(multisend_txs) > 1:	4✔
162	safe_tx = SafeTx(
163	self.ethereum_client,
164	self.address,
165	multisend.address,
166	0,
167	multisend.build_tx_data(multisend_txs),
168	SafeOperation.DELEGATE_CALL.value,
169	0,
170	0,
171	0,
172	None,
173	None,
174	safe_nonce=safe_nonce,
175	)
176	else:	×
177	safe_tx.safe_tx_gas = 0	4✔
178	safe_tx.base_gas = 0	4✔
179	safe_tx.gas_price = 0	4✔
180	safe_tx.signatures = b""	4✔
181	safe_tx.safe_nonce = safe_nonce # Resend single transaction	4✔
182	safe_tx = self.sign_transaction(safe_tx)	4✔
183	if not safe_tx.signatures:	4✔
184	print_formatted_text(
185	HTML("<ansired>At least one owner must be loaded</ansired>")
186	)
187	return False	×
188	else:	×
189	return self.post_transaction_to_tx_service(safe_tx)	4✔
190
191	def execute_tx(self, safe_tx_hash: Sequence[bytes]) -> bool:	4✔
192	"""
193	Submit transaction on the tx-service to blockchain
194
195	:return:
196	"""
197	safe_tx, tx_hash = self.safe_tx_service.get_safe_transaction(safe_tx_hash)	×
198	if tx_hash:	×
199	print_formatted_text(
200	HTML(
201	f"<ansired>Tx with safe-tx-hash {safe_tx_hash.hex()} "
202	f"has already been executed on {tx_hash.hex()}</ansired>"
203	)
204	)
205	elif len(safe_tx.signers) < self.safe_cli_info.threshold:	×
206	print_formatted_text(
207	HTML(
208	f"<ansired>Number of signatures {len(safe_tx.signers)} "
209	f"must reach the threshold {self.safe_cli_info.threshold}</ansired>"
210	)
211	)
212	else:	×
213	return self.execute_safe_transaction(safe_tx)	×
214
215	def get_balances(self):	4✔
216	balances = self.safe_tx_service.get_balances(self.address)	4✔
217	headers = ["name", "balance", "symbol", "decimals", "tokenAddress"]	4✔
218	rows = []	4✔
219	for balance in balances:	4✔
220	if balance["tokenAddress"]: # Token	4✔
221	row = [	4✔
222	balance["token"]["name"],	4✔
223	f"{int(balance['balance']) / 10 ** int(balance['token']['decimals']):.5f}",	4✔
224	balance["token"]["symbol"],	4✔
225	balance["token"]["decimals"],	4✔
226	balance["tokenAddress"],	4✔
227	]
228	else: # Ether	×
229	row = [	4✔
230	"ETHER",	4✔
231	f"{int(balance['balance']) / 10 ** 18:.5f}",	4✔
232	"Ξ",	4✔
233	18,	4✔
234	"",	4✔
235	]
236	rows.append(row)	4✔
237	print(tabulate(rows, headers=headers))	4✔
238	return rows	4✔
239
240	def get_transaction_history(self):	4✔
241	transactions = self.safe_tx_service.get_transactions(self.address)	4✔
242	headers = ["nonce", "to", "value", "transactionHash", "safeTxHash"]	4✔
243	rows = []	4✔
244	last_executed_tx = False	4✔
245	for transaction in transactions:	4✔
246	row = [transaction[header] for header in headers]	4✔
247	data_decoded: Dict[str, Any] = transaction.get("dataDecoded")	4✔
248	if data_decoded:	4✔
249	row.append(self.safe_tx_service.data_decoded_to_text(data_decoded))	4✔
250	if transaction["transactionHash"]:	4✔
251	if not transaction["isSuccessful"]:	4✔
252	# Transaction failed
253	row[0] = Fore.RED + str(row[0])	×
254	else:	×
255	row[0] = Fore.GREEN + str(	4✔
256	row[0]	4✔
257	) # For executed transactions we use green
258	if not last_executed_tx:	4✔
259	row[0] = Style.BRIGHT + row[0]	4✔
260	last_executed_tx = True	4✔
261	else:
262	row[0] = Fore.YELLOW + str(
263	row[0]
264	) # For non executed transactions we use yellow
265
266	row[0] = Style.RESET_ALL + row[0] # Reset all just in case	4✔
267	rows.append(row)	4✔
268
269	headers.append("dataDecoded")	4✔
270	headers[0] = Style.BRIGHT + headers[0]	4✔
271	print(tabulate(rows, headers=headers))	4✔
272	return rows	4✔
273
274	def prepare_and_execute_safe_transaction(	4✔
275	self,
276	to: str,	4✔
277	value: int,	4✔
278	data: bytes,	4✔
279	operation: SafeOperation = SafeOperation.CALL,	4✔
280	safe_nonce: Optional[int] = None,	4✔
281	) -> bool:	4✔
282	safe_tx = self.prepare_safe_transaction(
283	to, value, data, operation, safe_nonce=safe_nonce
284	)
285	return self.post_transaction_to_tx_service(safe_tx)
286
287	def post_transaction_to_tx_service(self, safe_tx: SafeTx) -> bool:	4✔
288	if not yes_or_no_question(	4✔
289	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) "	4✔
290	+ str(safe_tx)	4✔
291	):
292	return False
293
294	self.safe_tx_service.post_transaction(safe_tx)	4✔
295	print_formatted_text(	4✔
296	HTML(	4✔
297	f"<ansigreen>Tx with safe-tx-hash={safe_tx.safe_tx_hash.hex()} was sent to Safe Transaction service</ansigreen>"	4✔
298	)
299	)
300	return True	4✔
301
302	def get_permitted_signers(self) -> Set[ChecksumAddress]:	4✔
303	"""
304	:return: Owners and delegates, as they also can sign a transaction for the tx service
305	"""
306	owners = super().get_permitted_signers()	4✔
307	owners.update(	4✔
308	[	4✔
309	row["delegate"]
310	for row in self.safe_tx_service.get_delegates(self.address)	4✔
311	]
312	)
313	return owners	4✔
314
315	# Function that sends all assets to an account (to)
316	def drain(self, to: ChecksumAddress):	4✔
317	balances = self.safe_tx_service.get_balances(self.address)
318	safe_txs = []
319	safe_tx = None
320	for balance in balances:
321	amount = int(balance["balance"])
322	if balance["tokenAddress"] is None: # Then is ether
323	if amount != 0:
324	safe_tx = self.prepare_safe_transaction(
325	to,
326	amount,
327	b"",
328	SafeOperation.CALL,
329	safe_nonce=None,
330	)
331	else:
332	transaction = (
333	get_erc20_contract(self.ethereum_client.w3, balance["tokenAddress"])
334	.functions.transfer(to, amount)
335	.build_transaction({"from": self.address, "gas": 0, "gasPrice": 0})
336	)
337	safe_tx = self.prepare_safe_transaction(
338	balance["tokenAddress"],
339	0,
340	HexBytes(transaction["data"]),
341	SafeOperation.CALL,
342	safe_nonce=None,
343	)
344	if safe_tx:
345	safe_txs.append(safe_tx)
346	if len(safe_txs) > 0:
347	multisend_tx = self.batch_safe_txs(safe_tx.safe_nonce, safe_txs)
348	if multisend_tx is not None:
349	self.post_transaction_to_tx_service(multisend_tx)
350	print_formatted_text(
351	HTML(
352	"<ansigreen>Transaction to drain account correctly created</ansigreen>"
353	)
354	)
355	else:
356	print_formatted_text(
357	HTML("<ansigreen>Safe account is currently empty</ansigreen>")
358	)

safe-global / safe-cli / 7087242091

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