• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

safe-global / safe-cli / 19670041703

25 Nov 2025 12:47PM UTC coverage: 88.407% (-0.2%) from 88.573%
19670041703

push

github

Uxio0
Fix CLI defaults and Safe interactions

- Detect default/attended mode from provided argv instead of sys.argv
- Use the sign-message-lib address for message signing
- Return values from threshold/nonce/owners getters and fix prompt output
- Fix tx-service batching when Multisend is not deployed
- Improve seed-derived owner loading feedback and fix approve-hash owner lookup
- Handle missing signer in approve_hash

224 of 266 branches covered (84.21%)

Branch coverage included in aggregate %.

23 of 39 new or added lines in 3 files covered. (58.97%)

2 existing lines in 1 file now uncovered.

2895 of 3262 relevant lines covered (88.75%)

3.55 hits per line

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

68.26
/src/safe_cli/operators/safe_tx_service_operator.py
1
import json
4✔
2
from itertools import chain
4✔
3
from typing import Any, Dict, 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 safe_eth.eth.contracts import get_erc20_contract
4✔
12
from safe_eth.eth.eip712 import eip712_encode_hash
4✔
13
from safe_eth.safe import SafeOperationEnum, SafeTx
4✔
14
from safe_eth.safe.api import SafeAPIException
4✔
15
from safe_eth.safe.api.transaction_service_api.transaction_service_messages import (
4✔
16
    get_remove_transaction_message,
17
)
18
from safe_eth.safe.multi_send import MultiSend, MultiSendOperation, MultiSendTx
4✔
19
from safe_eth.safe.safe_signature import SafeSignature
4✔
20
from safe_eth.safe.signatures import signature_to_bytes
4✔
21
from safe_eth.util.util import to_0x_hex_str
4✔
22
from tabulate import tabulate
4✔
23

24
from ..utils import get_input, yes_or_no_question
4✔
25
from . import SafeServiceNotAvailable
4✔
26
from .exceptions import AccountNotLoadedException, NonExistingOwnerException
4✔
27
from .hw_wallets.hw_wallet import HwWallet
4✔
28
from .safe_operator import SafeOperator
4✔
29

30

31
class SafeTxServiceOperator(SafeOperator):
4✔
32
    def __init__(self, address: str, node_url: str):
4✔
33
        super().__init__(address, node_url)
4✔
34
        if not self.safe_tx_service:
4✔
35
            raise SafeServiceNotAvailable(
4✔
36
                f"Cannot configure tx service for network {self.network.name}"
37
            )
38
        self.require_all_signatures = (
4✔
39
            False  # It doesn't require all signatures to be present to send a tx
40
        )
41

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

45
    def sign_message(
4✔
46
        self,
47
        eip712_message_path: Optional[str] = None,
48
    ) -> bool:
49
        if eip712_message_path:
×
50
            try:
×
51
                message = json.load(open(eip712_message_path, "r"))
×
52
                message_hash = eip712_encode_hash(message)
×
53
            except ValueError:
×
54
                raise ValueError
×
55
        else:
56
            print_formatted_text("EIP191 message to sign:")
×
57
            message = get_input()
×
58
            message_hash = defunct_hash_message(text=message)
×
59

60
        safe_message_hash = self.safe.get_message_hash(message_hash)
×
61
        eoa_signers, hw_wallet_signers = self.get_signers()
×
62
        # Safe transaction service just accept one signer to create a message
63
        signature = b""
×
64
        if eoa_signers:
×
65
            signature_dict = eoa_signers[0].unsafe_sign_hash(safe_message_hash)
×
66
            signature = signature_to_bytes(
×
67
                signature_dict["v"], signature_dict["r"], signature_dict["s"]
68
            )
69

70
        elif hw_wallet_signers:
×
71
            signature = SafeSignature.export_signatures(
×
72
                self.hw_wallet_manager.sign_message(
73
                    safe_message_hash, [hw_wallet_signers[0]]
74
                )
75
            )
76
        else:
77
            print_formatted_text(
×
78
                HTML("<ansired>At least one owner must be loaded</ansired>")
79
            )
80

81
        if self.safe_tx_service.post_message(self.address, message, signature):
×
82
            print_formatted_text(
×
83
                HTML(
84
                    f"<ansigreen>Message  with safe-message-hash {to_0x_hex_str(safe_message_hash)} 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 confirm_message(self, safe_message_hash: bytes, sender: ChecksumAddress):
4✔
97
        # GET message
98
        try:
4✔
99
            safe_message = self.safe_tx_service.get_message(safe_message_hash)
4✔
100
        except SafeAPIException:
×
101
            print_formatted_text(
×
102
                HTML(
103
                    f"<ansired>Message with hash {to_0x_hex_str(safe_message_hash)} does not exist</ansired>"
104
                )
105
            )
106
        if not yes_or_no_question(
4✔
107
            f"Message: {safe_message['message']} \n Do you want to sign the following message?:"
108
        ):
109
            return False
×
110

111
        signer = self.search_account(sender)
4✔
112
        if not signer:
4✔
113
            print_formatted_text(
×
114
                HTML(f"<ansired>Owner with address {sender} was not loaded</ansired>")
115
            )
116

117
        if isinstance(signer, LocalAccount):
4✔
118
            signature = signer.unsafe_sign_hash(safe_message_hash).signature
4✔
119
        else:
120
            signature = SafeSignature.export_signatures(
×
121
                self.hw_wallet_manager.sign_message(safe_message_hash, [signer])
122
            )
123

124
        try:
4✔
125
            self.safe_tx_service.post_message_signature(safe_message_hash, signature)
4✔
126
        except SafeAPIException as e:
4✔
127
            print_formatted_text(
4✔
128
                HTML(f"<ansired>Message wasn't confirmed due an error: {e}</ansired>")
129
            )
130
            return False
4✔
131
        print_formatted_text(
4✔
132
            HTML(
133
                f"<ansigreen>Message with safe-message-hash {to_0x_hex_str(safe_message_hash)} was correctly confirmed on Safe Transaction Service</ansigreen>"
134
            )
135
        )
136
        return True
4✔
137

138
    def get_delegates(self):
4✔
139
        delegates = self.safe_tx_service.get_delegates(self.address)
4✔
140
        headers = ["delegate", "delegator", "label"]
4✔
141
        rows = []
4✔
142
        for delegate in delegates:
4✔
143
            row = [delegate["delegate"], delegate["delegator"], delegate["label"]]
4✔
144
            rows.append(row)
4✔
145
        print(tabulate(rows, headers=headers))
4✔
146
        return rows
4✔
147

148
    def add_delegate(self, delegate_address: str, label: str, signer_address: str):
4✔
149
        signer_account = [
4✔
150
            account for account in self.accounts if account.address == signer_address
151
        ]
152
        if not signer_account:
4✔
153
            raise AccountNotLoadedException(signer_address)
×
154
        elif signer_address not in self.safe_cli_info.owners:
4✔
155
            raise NonExistingOwnerException(signer_address)
×
156
        else:
157
            signer_account = signer_account[0]
4✔
158
            try:
4✔
159
                hash_to_sign = self.safe_tx_service.create_delegate_message_hash(
4✔
160
                    delegate_address
161
                )
162
                signature = signer_account.unsafe_sign_hash(hash_to_sign)
4✔
163
                self.safe_tx_service.add_delegate(
4✔
164
                    delegate_address,
165
                    signer_account.address,
166
                    label,
167
                    signature.signature,
168
                    safe_address=self.address,
169
                )
170
                return True
4✔
171
            except SafeAPIException:
×
172
                return False
×
173

174
    def remove_delegate(self, delegate_address: str, signer_address: str):
4✔
175
        signer_account = [
4✔
176
            account for account in self.accounts if account.address == signer_address
177
        ]
178
        if not signer_account:
4✔
179
            raise AccountNotLoadedException(signer_address)
×
180
        elif signer_address not in self.safe_cli_info.owners:
4✔
181
            raise NonExistingOwnerException(signer_address)
×
182
        else:
183
            signer_account = signer_account[0]
4✔
184
            try:
4✔
185
                hash_to_sign = self.safe_tx_service.create_delegate_message_hash(
4✔
186
                    delegate_address
187
                )
188
                signature = signer_account.unsafe_sign_hash(hash_to_sign)
4✔
189
                self.safe_tx_service.remove_delegate(
4✔
190
                    delegate_address,
191
                    signer_account.address,
192
                    signature.signature,
193
                    safe_address=self.address,
194
                )
195
                return True
4✔
196
            except SafeAPIException:
×
197
                return False
×
198

199
    def submit_signatures(self, safe_tx_hash: bytes) -> bool:
4✔
200
        """
201
        Submit signatures to the tx service
202

203
        :return:
204
        """
205

206
        safe_tx, tx_hash = self.safe_tx_service.get_safe_transaction(safe_tx_hash)
4✔
207
        safe_tx.signatures = b""  # Don't post again existing signatures
4✔
208
        if tx_hash:
4✔
209
            print_formatted_text(
4✔
210
                HTML(
211
                    f"<ansired>Tx with safe-tx-hash {to_0x_hex_str(safe_tx_hash)} "
212
                    f"has already been executed on {to_0x_hex_str(tx_hash)}</ansired>"
213
                )
214
            )
215
        else:
216
            safe_tx = self.sign_transaction(safe_tx)
4✔
217
            if safe_tx.signers:
4✔
218
                self.safe_tx_service.post_signatures(safe_tx_hash, safe_tx.signatures)
4✔
219
                print_formatted_text(
4✔
220
                    HTML(
221
                        f"<ansigreen>{len(safe_tx.signers)} signatures were submitted to the tx service</ansigreen>"
222
                    )
223
                )
224
                return True
4✔
225
            else:
226
                print_formatted_text(
4✔
227
                    HTML(
228
                        "<ansired>Cannot generate signatures as there were no suitable signers</ansired>"
229
                    )
230
                )
231
        return False
4✔
232

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

238
        :return:
239
        """
240

241
        try:
4✔
242
            multisend = MultiSend(ethereum_client=self.ethereum_client)
4✔
243
        except ValueError:
×
244
            print_formatted_text(
×
245
                HTML(
246
                    "<ansired>Multisend contract is not deployed on this network and it's required for "
247
                    "batching txs</ansired>"
248
                )
249
            )
NEW
250
            return False
×
251

252
        multisend_txs = []
4✔
253
        for safe_tx_hash in safe_tx_hashes:
4✔
254
            safe_tx, _ = self.safe_tx_service.get_safe_transaction(safe_tx_hash)
4✔
255
            # Check if call is already a Multisend call
256
            inner_txs = MultiSend.from_transaction_data(safe_tx.data)
4✔
257
            if inner_txs:
4✔
258
                multisend_txs.extend(inner_txs)
×
259
            else:
260
                multisend_txs.append(
4✔
261
                    MultiSendTx(
262
                        MultiSendOperation.CALL, safe_tx.to, safe_tx.value, safe_tx.data
263
                    )
264
                )
265

266
        if len(multisend_txs) > 1:
4✔
267
            safe_tx = SafeTx(
×
268
                self.ethereum_client,
269
                self.address,
270
                multisend.address,
271
                0,
272
                multisend.build_tx_data(multisend_txs),
273
                SafeOperationEnum.DELEGATE_CALL.value,
274
                0,
275
                0,
276
                0,
277
                None,
278
                None,
279
                safe_nonce=safe_nonce,
280
            )
281
        else:
282
            safe_tx.safe_tx_gas = 0
4✔
283
            safe_tx.base_gas = 0
4✔
284
            safe_tx.gas_price = 0
4✔
285
            safe_tx.signatures = b""
4✔
286
            safe_tx.safe_nonce = safe_nonce  # Resend single transaction
4✔
287
        safe_tx = self.sign_transaction(safe_tx)
4✔
288
        if not safe_tx.signatures:
4✔
289
            print_formatted_text(
×
290
                HTML("<ansired>At least one owner must be loaded</ansired>")
291
            )
292
            return False
×
293
        else:
294
            return self.post_transaction_to_tx_service(safe_tx)
4✔
295

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

300
        :return:
301
        """
302
        safe_tx, tx_hash = self.safe_tx_service.get_safe_transaction(safe_tx_hash)
×
303
        if tx_hash:
×
304
            print_formatted_text(
×
305
                HTML(
306
                    f"<ansired>Tx with safe-tx-hash {to_0x_hex_str(safe_tx_hash)} "
307
                    f"has already been executed on {to_0x_hex_str(tx_hash)}</ansired>"
308
                )
309
            )
310
        elif len(safe_tx.signers) < self.safe_cli_info.threshold:
×
311
            print_formatted_text(
×
312
                HTML(
313
                    f"<ansired>Number of signatures {len(safe_tx.signers)} "
314
                    f"must reach the threshold {self.safe_cli_info.threshold}</ansired>"
315
                )
316
            )
317
        else:
318
            if executed := self.execute_safe_transaction(safe_tx):
×
319
                self.refresh_safe_cli_info()
×
320
            return executed
×
321

322
    def get_balances(self):
4✔
323
        balances = self.safe_tx_service.get_balances(self.address)
4✔
324
        headers = ["name", "balance", "symbol", "decimals", "tokenAddress"]
4✔
325
        rows = []
4✔
326
        for balance in balances:
4✔
327
            if balance["tokenAddress"]:  # Token
4✔
328
                row = [
4✔
329
                    balance["token"]["name"],
330
                    f"{int(balance['balance']) / 10 ** int(balance['token']['decimals']):.5f}",
331
                    balance["token"]["symbol"],
332
                    balance["token"]["decimals"],
333
                    balance["tokenAddress"],
334
                ]
335
            else:  # Ether
336
                row = [
4✔
337
                    "ETHER",
338
                    f"{int(balance['balance']) / 10 ** 18:.5f}",
339
                    "Ξ",
340
                    18,
341
                    "",
342
                ]
343
            rows.append(row)
4✔
344
        print(tabulate(rows, headers=headers))
4✔
345
        return rows
4✔
346

347
    def get_transaction_history(self):
4✔
348
        transactions = self.safe_tx_service.get_transactions(self.address)
4✔
349
        headers = ["nonce", "to", "value", "transactionHash", "safeTxHash"]
4✔
350
        rows = []
4✔
351
        last_executed_tx = False
4✔
352
        for transaction in transactions:
4✔
353
            row = [transaction[header] for header in headers]
4✔
354
            data_decoded: Dict[str, Any] = transaction.get("dataDecoded")
4✔
355
            if data_decoded:
4✔
356
                row.append(self.safe_tx_service.data_decoded_to_text(data_decoded))
4✔
357
            if transaction["transactionHash"]:
4✔
358
                if not transaction["isSuccessful"]:
4✔
359
                    # Transaction failed
360
                    row[0] = Fore.RED + str(row[0])
×
361
                else:
362
                    row[0] = Fore.GREEN + str(
4✔
363
                        row[0]
364
                    )  # For executed transactions we use green
365
                    if not last_executed_tx:
4✔
366
                        row[0] = Style.BRIGHT + row[0]
4✔
367
                        last_executed_tx = True
4✔
368
            else:
369
                row[0] = Fore.YELLOW + str(
×
370
                    row[0]
371
                )  # For non executed transactions we use yellow
372

373
            row[0] = Style.RESET_ALL + row[0]  # Reset all just in case
4✔
374
            rows.append(row)
4✔
375

376
        headers.append("dataDecoded")
4✔
377
        headers[0] = Style.BRIGHT + headers[0]
4✔
378
        print(tabulate(rows, headers=headers))
4✔
379
        return rows
4✔
380

381
    def prepare_and_execute_safe_transaction(
4✔
382
        self,
383
        to: str,
384
        value: int,
385
        data: bytes,
386
        operation: SafeOperationEnum = SafeOperationEnum.CALL,
387
        safe_nonce: Optional[int] = None,
388
    ) -> bool:
389
        safe_tx = self.prepare_safe_transaction(
×
390
            to, value, data, operation, safe_nonce=safe_nonce
391
        )
392
        return self.post_transaction_to_tx_service(safe_tx)
×
393

394
    def post_transaction_to_tx_service(self, safe_tx: SafeTx) -> bool:
4✔
395
        if not yes_or_no_question(
4✔
396
            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) "
397
            + str(safe_tx)
398
        ):
399
            return False
×
400

401
        self.safe_tx_service.post_transaction(safe_tx)
4✔
402
        print_formatted_text(
4✔
403
            HTML(
404
                f"<ansigreen>Tx with safe-tx-hash={to_0x_hex_str(safe_tx.safe_tx_hash)} was sent to Safe Transaction service</ansigreen>"
405
            )
406
        )
407
        return True
4✔
408

409
    def get_permitted_signers(self) -> Set[ChecksumAddress]:
4✔
410
        """
411
        :return: Owners and delegates, as they also can sign a transaction for the tx service
412
        """
413
        owners = super().get_permitted_signers()
4✔
414
        owners.update(
4✔
415
            [
416
                row["delegate"]
417
                for row in self.safe_tx_service.get_delegates(self.address)
418
            ]
419
        )
420
        return owners
4✔
421

422
    # Function that sends all assets to an account (to)
423
    def drain(self, to: ChecksumAddress):
4✔
424
        balances = self.safe_tx_service.get_balances(self.address)
×
425
        safe_txs = []
×
426
        safe_tx = None
×
427
        for balance in balances:
×
428
            amount = int(balance["balance"])
×
429
            if balance["tokenAddress"] is None:  # Then is ether
×
430
                if amount != 0:
×
431
                    safe_tx = self.prepare_safe_transaction(
×
432
                        to,
433
                        amount,
434
                        b"",
435
                        SafeOperationEnum.CALL,
436
                        safe_nonce=None,
437
                    )
438
            else:
439
                transaction = (
×
440
                    get_erc20_contract(self.ethereum_client.w3, balance["tokenAddress"])
441
                    .functions.transfer(to, amount)
442
                    .build_transaction({"from": self.address, "gas": 0, "gasPrice": 0})
443
                )
444
                safe_tx = self.prepare_safe_transaction(
×
445
                    balance["tokenAddress"],
446
                    0,
447
                    HexBytes(transaction["data"]),
448
                    SafeOperationEnum.CALL,
449
                    safe_nonce=None,
450
                )
451
            if safe_tx:
×
452
                safe_txs.append(safe_tx)
×
453
        if len(safe_txs) > 0:
×
454
            multisend_tx = self.batch_safe_txs(safe_tx.safe_nonce, safe_txs)
×
455
            if multisend_tx is not None:
×
456
                self.post_transaction_to_tx_service(multisend_tx)
×
457
                print_formatted_text(
×
458
                    HTML(
459
                        "<ansigreen>Transaction to drain account correctly created</ansigreen>"
460
                    )
461
                )
462
        else:
463
            print_formatted_text(
×
464
                HTML("<ansigreen>Safe account is currently empty</ansigreen>")
465
            )
466

467
    def search_account(
4✔
468
        self, address: ChecksumAddress
469
    ) -> Optional[Union[LocalAccount, HwWallet]]:
470
        """
471
        Search the provided address between loaded owners
472

473
        :param address:
474
        :return: LocalAccount or HwWallet of the provided address
475
        """
476
        for account in chain(self.accounts, self.hw_wallet_manager.wallets):
4✔
477
            if account.address == address:
4✔
478
                return account
4✔
479

480
    def remove_proposed_transaction(self, safe_tx_hash: bytes):
4✔
481
        eip712_message = get_remove_transaction_message(
4✔
482
            self.address, safe_tx_hash, self.ethereum_client.get_chain_id()
483
        )
484
        message_hash = eip712_encode_hash(eip712_message)
4✔
485
        try:
4✔
486
            safe_tx, _ = self.safe_tx_service.get_safe_transaction(safe_tx_hash)
4✔
487
            signer = self.search_account(safe_tx.proposer)
4✔
488
            if not signer:
4✔
489
                print_formatted_text(
4✔
490
                    HTML(
491
                        f"<ansired>The proposer with address: {safe_tx.proposer} was not loaded</ansired>"
492
                    )
493
                )
494
                return False
4✔
495

496
            if isinstance(signer, LocalAccount):
4✔
497
                signature = signer.unsafe_sign_hash(message_hash).signature
4✔
498
            else:
499
                signature = self.hw_wallet_manager.sign_eip712(
×
500
                    eip712_message, [signer]
501
                )[0].signature
502

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

513
            if not yes_or_no_question(
4✔
514
                f"Do you want to remove the tx with safe-tx-hash={to_0x_hex_str(safe_tx.safe_tx_hash)}"
515
            ):
516
                return False
×
517

518
            self.safe_tx_service.delete_transaction(
4✔
519
                to_0x_hex_str(safe_tx_hash), to_0x_hex_str(signature)
520
            )
521
            print_formatted_text(
4✔
522
                HTML(
523
                    f"<ansigreen>Transaction {to_0x_hex_str(safe_tx_hash)} was removed correctly</ansigreen>"
524
                )
525
            )
526
            return True
4✔
527
        except SafeAPIException as e:
×
528
            print_formatted_text(
×
529
                HTML(f"<ansired>Transaction wasn't removed due an error: {e}</ansired>")
530
            )
531
            return False
×
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc