11553543762

Committed 28 Oct 2024 12:06PM CUT coverage: 88.612%. Remained the same

Build # 11553543762

Build Type

push

github

Committed by

Uxio0

Commit Message

Set version 1.4.0

Run Details

221 of 262 branches covered (84.35%)

Branch coverage included in aggregate %.

2868 of 3224 relevant lines covered (88.96%)

2.67 hits per line

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

76.23

/src/safe_cli/prompt_parser.py

import argparse
import functools

from prompt_toolkit import HTML, print_formatted_text
from prompt_toolkit.formatted_text import html
from safe_eth.safe.api import SafeAPIException

from .argparse_validators import (
    check_ethereum_address,
    check_hex_str,
    check_keccak256_hash,
)
from .operators import SafeServiceNotAvailable
from .operators.exceptions import (
    AccountNotLoadedException,
    ExistingOwnerException,
    FallbackHandlerNotSupportedException,
    HardwareWalletException,
    HashAlreadyApproved,
    InvalidMasterCopyException,
    InvalidMigrationContractException,
    InvalidNonceException,
    NonExistingOwnerException,
    NotEnoughEtherToSend,
    NotEnoughSignatures,
    NotEnoughTokenToSend,
    SafeAlreadyUpdatedException,
    SafeCliTerminationException,
    SafeOperatorException,
    SafeVersionNotSupportedException,
    SameFallbackHandlerException,
    SameMasterCopyException,
    SenderRequiredException,
    ThresholdLimitException,
)
from .operators.safe_operator import SafeOperator
from .safe_completer_constants import meta, safe_commands_arguments


def safe_exception(function):
    @functools.wraps(function)
    def wrapper(*args, **kwargs):
        try:
            return function(*args, **kwargs)
        except SafeAPIException as e:
            if e.args:
                print_formatted_text(HTML(f"<b><ansired>{e.args[0]}</ansired></b>"))
        except AccountNotLoadedException as e:
            print_formatted_text(
                HTML(f"<ansired>Account {e.args[0]} is not loaded</ansired>")
            )
        except NotEnoughSignatures as e:
            print_formatted_text(
                HTML(
                    f"<ansired>Cannot find enough owners to sign. {e.args[0]} missing</ansired>"
                )
            )
        except SenderRequiredException:
            print_formatted_text(
                HTML("<ansired>Please load a default sender</ansired>")
            )
        except ExistingOwnerException as e:
            print_formatted_text(
                HTML(
                    f"<ansired>Owner {e.args[0]} is already an owner of the Safe"
                    f"</ansired>"
                )
            )
        except NonExistingOwnerException as e:
            print_formatted_text(
                HTML(
                    f"<ansired>Owner {e.args[0]} is not an owner of the Safe"
                    f"</ansired>"
                )
            )
        except HashAlreadyApproved as e:
            print_formatted_text(
                HTML(
                    f"<ansired>Transaction with safe-tx-hash {e.args[0].hex()} has already been approved by "
                    f"owner {e.args[1]}</ansired>"
                )
            )
        except ThresholdLimitException:
            print_formatted_text(
                HTML(
                    "<ansired>Having less owners than threshold is not allowed"
                    "</ansired>"
                )
            )
        except SameFallbackHandlerException as e:
            print_formatted_text(
                HTML(
                    f"<ansired>Fallback handler {e.args[0]} is the current one</ansired>"
                )
            )
        except FallbackHandlerNotSupportedException:
            print_formatted_text(
                HTML(
                    "<ansired>Fallback handler is not supported for your Safe, "
                    "you need to <b>update</b> first</ansired>"
                )
            )
        except SameMasterCopyException as e:
            print_formatted_text(
                HTML(f"<ansired>Master Copy {e.args[0]} is the current one</ansired>")
            )
        except InvalidMasterCopyException as e:
            print_formatted_text(
                HTML(f"<ansired>Master Copy {e.args[0]} is not valid</ansired>")
            )
        except InvalidMigrationContractException as e:
            print_formatted_text(HTML(f"<ansired>{e.args[0]}</ansired>"))
        except InvalidNonceException as e:
            print_formatted_text(HTML(f"<ansired>{e.args[0]}</ansired>"))
        except SafeAlreadyUpdatedException:
            print_formatted_text(HTML("<ansired>Safe is already updated</ansired>"))
        except SafeVersionNotSupportedException as e:
            print_formatted_text(HTML(f"<ansired>{e.args[0]}</ansired>"))
        except (NotEnoughEtherToSend, NotEnoughTokenToSend) as e:
            print_formatted_text(
                HTML(
                    f"<ansired>Cannot find enough to send. Current balance is {e.args[0]}"
                    f"</ansired>"
                )
            )
        except SafeServiceNotAvailable as e:
            print_formatted_text(
                HTML(
                    f"<ansired>Service not available for network {e.args[0]}</ansired>"
                )
            )
        except HardwareWalletException as e:
            print_formatted_text(
                HTML(f"<ansired>HwDevice exception: {e.args[0]}</ansired>")
            )
        except SafeOperatorException as e:
            print_formatted_text(HTML(f"<ansired>{e.args[0]}</ansired>"))

    return wrapper


class PromptParser:
    def __init__(self, safe_operator: SafeOperator):
        self.mode_parser = argparse.ArgumentParser(prog="")
        self.safe_operator = safe_operator
        self.prompt_parser = build_prompt_parser(safe_operator)

    def process_command(self, command: str):
        args = self.prompt_parser.parse_args(command.split())
        return args.func(args)


def build_prompt_parser(safe_operator: SafeOperator) -> argparse.ArgumentParser:
    """
    Returns an ArgParse capable of decoding and executing the Safe commands
    :param safe_operator:
    :return:
    """
    prompt_parser = argparse.ArgumentParser(prog="")
    subparsers = prompt_parser.add_subparsers()

    @safe_exception
    def show_cli_owners(args):
        safe_operator.show_cli_owners()

    @safe_exception
    def load_cli_owners_from_words(args):
        safe_operator.load_cli_owners_from_words(args.words)

    @safe_exception
    def load_cli_owners(args):
        safe_operator.load_cli_owners(args.keys)

    @safe_exception
    def load_ledger_cli_owners(args):
        safe_operator.load_ledger_cli_owners(
            derivation_path=args.derivation_path, legacy_account=args.legacy_accounts
        )

    @safe_exception
    def load_trezor_cli_owners(args):
        safe_operator.load_trezor_cli_owners(
            derivation_path=args.derivation_path, legacy_account=args.legacy_accounts
        )

    @safe_exception
    def unload_cli_owners(args):
        safe_operator.unload_cli_owners(args.addresses)

    @safe_exception
    def approve_hash(args):
        safe_operator.approve_hash(args.hash_to_approve, args.sender)

    @safe_exception
    def sign_message(args):
        safe_operator.sign_message(args.eip712_path)

    @safe_exception
    def confirm_message(args):
        safe_operator.confirm_message(args.safe_message_hash, args.sender)

    @safe_exception
    def add_owner(args):
        safe_operator.add_owner(args.address, threshold=args.threshold)

    @safe_exception
    def remove_owner(args):
        safe_operator.remove_owner(args.address, threshold=args.threshold)

    @safe_exception
    def change_fallback_handler(args):
        safe_operator.change_fallback_handler(args.address)

    @safe_exception
    def change_guard(args):
        safe_operator.change_guard(args.address)

    @safe_exception
    def change_master_copy(args):
        safe_operator.change_master_copy(args.address)

    @safe_exception
    def change_threshold(args):
        safe_operator.change_threshold(args.threshold)

    @safe_exception
    def send_custom(args):
        safe_operator.send_custom(
            args.to,
            args.value,
            args.data,
            safe_nonce=args.safe_nonce,
            delegate_call=args.delegate,
        )

    @safe_exception
    def send_ether(args):
        safe_operator.send_ether(args.to, args.value, safe_nonce=args.safe_nonce)

    @safe_exception
    def send_erc20(args):
        safe_operator.send_erc20(
            args.to, args.token_address, args.amount, safe_nonce=args.safe_nonce
        )

    @safe_exception
    def send_erc721(args):
        safe_operator.send_erc721(
            args.to, args.token_address, args.token_id, safe_nonce=args.safe_nonce
        )

    @safe_exception
    def drain(args):
        safe_operator.drain(args.to)

    @safe_exception
    def get_threshold(args):
        safe_operator.get_threshold()

    @safe_exception
    def get_nonce(args):
        safe_operator.get_nonce()

    @safe_exception
    def get_owners(args):
        safe_operator.get_owners()

    @safe_exception
    def enable_module(args):
        safe_operator.enable_module(args.address)

    @safe_exception
    def disable_module(args):
        safe_operator.disable_module(args.address)

    @safe_exception
    def update_version(args):
        safe_operator.update_version()

    @safe_exception
    def update_version_to_l2(args):
        safe_operator.update_version_to_l2(args.migration_contract)

    @safe_exception
    def get_info(args):
        safe_operator.print_info()

    @safe_exception
    def get_refresh(args):
        safe_operator.refresh_safe_cli_info()

    @safe_exception
    def get_balances(args):
        safe_operator.get_balances()

    @safe_exception
    def get_history(args):
        safe_operator.get_transaction_history()

    @safe_exception
    def sign_tx(args):
        safe_operator.submit_signatures(args.safe_tx_hash)

    @safe_exception
    def batch_txs(args):
        safe_operator.batch_txs(args.safe_nonce, args.safe_tx_hashes)

    @safe_exception
    def execute_tx(args):
        safe_operator.execute_tx(args.safe_tx_hash)

    @safe_exception
    def get_delegates(args):
        safe_operator.get_delegates()

    @safe_exception
    def add_delegate(args):
        safe_operator.add_delegate(args.address, args.label, args.signer)

    @safe_exception
    def remove_delegate(args):
        safe_operator.remove_delegate(args.address, args.signer)

    @safe_exception
    def remove_proposed_transaction(args):
        safe_operator.remove_proposed_transaction(args.safe_tx_hash)

    def list_commands(args):
        print_formatted_text(
            HTML("<b>The following commands can be used:</b>"), end="\n\n"
        )
        for key in sorted(safe_commands_arguments.keys()):
            print_formatted_text(
                HTML(
                    f"<b>{key} {html.html_escape(safe_commands_arguments.get(key))}</b>"
                )
            )
            print_formatted_text(HTML("&#9;"), meta.get(key, ""))
        print_formatted_text(
            HTML("\nUse the <b>tab key</b> to show options in interactive mode.")
        )

    def terminate_cli(args):
        raise SafeCliTerminationException()

    # Cli owners
    parser_show_cli_owners = subparsers.add_parser("show_cli_owners")
    parser_show_cli_owners.set_defaults(func=show_cli_owners)

    parser_load_cli_owners_from_words = subparsers.add_parser(
        "load_cli_owners_from_words"
    )
    parser_load_cli_owners_from_words.add_argument("words", type=str, nargs="+")
    parser_load_cli_owners_from_words.set_defaults(func=load_cli_owners_from_words)

    parser_load_cli_owners = subparsers.add_parser("load_cli_owners")
    parser_load_cli_owners.add_argument("keys", type=str, nargs="+")
    parser_load_cli_owners.set_defaults(func=load_cli_owners)

    parser_load_ledger_cli_owners = subparsers.add_parser("load_ledger_cli_owners")
    parser_load_ledger_cli_owners.add_argument(
        "--derivation-path",
        type=str,
        help="Load address for the provided derivation path",
    )
    parser_load_ledger_cli_owners.add_argument(
        "--legacy-accounts",
        action="store_true",
        help="Search for legacy accounts",
    )
    parser_load_ledger_cli_owners.set_defaults(func=load_ledger_cli_owners)

    parser_load_trezor_cli_owners = subparsers.add_parser("load_trezor_cli_owners")
    parser_load_trezor_cli_owners.add_argument(
        "--derivation-path",
        type=str,
        help="Load address for the provided derivation path",
    )
    parser_load_trezor_cli_owners.add_argument(
        "--legacy-accounts",
        action="store_true",
        help="Search for legacy accounts",
    )
    parser_load_trezor_cli_owners.set_defaults(func=load_trezor_cli_owners)

    parser_unload_cli_owners = subparsers.add_parser("unload_cli_owners")
    parser_unload_cli_owners.add_argument(
        "addresses", type=check_ethereum_address, nargs="+"
    )
    parser_unload_cli_owners.set_defaults(func=unload_cli_owners)

    # Change threshold
    parser_change_threshold = subparsers.add_parser("change_threshold")
    parser_change_threshold.add_argument("threshold", type=int)
    parser_change_threshold.set_defaults(func=change_threshold)

    # Approve hash
    parser_approve_hash = subparsers.add_parser("approve_hash")
    parser_approve_hash.add_argument("hash_to_approve", type=check_keccak256_hash)
    parser_approve_hash.add_argument("sender", type=check_ethereum_address)
    parser_approve_hash.set_defaults(func=approve_hash)

    # Sign message
    parser_sign_message = subparsers.add_parser("sign_message")
    group = parser_sign_message.add_mutually_exclusive_group(required=True)
    group.add_argument("--eip191_message", action="store_true")
    group.add_argument("--eip712_path", type=str)
    parser_sign_message.set_defaults(func=sign_message)

    # Confirm message
    parser_confirm_message = subparsers.add_parser("confirm_message")
    parser_confirm_message.add_argument("safe_message_hash", type=check_keccak256_hash)
    parser_confirm_message.add_argument("sender", type=check_ethereum_address)
    parser_confirm_message.set_defaults(func=confirm_message)

    # Add owner
    parser_add_owner = subparsers.add_parser("add_owner")
    parser_add_owner.add_argument("address", type=check_ethereum_address)
    parser_add_owner.add_argument("--threshold", type=int, default=None)
    parser_add_owner.set_defaults(func=add_owner)

    # Remove owner
    parser_remove_owner = subparsers.add_parser("remove_owner")
    parser_remove_owner.add_argument("address", type=check_ethereum_address)
    parser_remove_owner.add_argument("--threshold", type=int, default=None)
    parser_remove_owner.set_defaults(func=remove_owner)

    # Change FallbackHandler
    parser_change_fallback_handler = subparsers.add_parser("change_fallback_handler")
    parser_change_fallback_handler.add_argument("address", type=check_ethereum_address)
    parser_change_fallback_handler.set_defaults(func=change_fallback_handler)

    # Change Guard
    parser_change_guard = subparsers.add_parser("change_guard")
    parser_change_guard.add_argument("address", type=check_ethereum_address)
    parser_change_guard.set_defaults(func=change_guard)

    # Change MasterCopy
    parser_change_master_copy = subparsers.add_parser("change_master_copy")
    parser_change_master_copy.add_argument("address", type=check_ethereum_address)
    parser_change_master_copy.set_defaults(func=change_master_copy)

    # Update Safe to last version
    parser_update_version = subparsers.add_parser("update")
    parser_update_version.set_defaults(func=update_version)

    # Update non L2 Safe to L2 Safe
    parser_update_version_to_l2 = subparsers.add_parser("update_version_to_l2")
    parser_update_version_to_l2.add_argument(
        "migration_contract", type=check_ethereum_address
    )
    parser_update_version_to_l2.set_defaults(func=update_version_to_l2)

    # Send custom/ether/erc20/erc721
    parser_send_custom = subparsers.add_parser("send_custom")
    parser_send_ether = subparsers.add_parser("send_ether")
    parser_send_erc20 = subparsers.add_parser("send_erc20")
    parser_send_erc721 = subparsers.add_parser("send_erc721")
    parser_send_custom.set_defaults(func=send_custom)
    parser_send_ether.set_defaults(func=send_ether)
    parser_send_erc20.set_defaults(func=send_erc20)
    parser_send_erc721.set_defaults(func=send_erc721)
    # They have some common arguments
    for parser in (
        parser_send_custom,
        parser_send_ether,
        parser_send_erc20,
        parser_send_erc721,
    ):
        parser.add_argument(
            "--safe-nonce",
            type=int,
            help="Use custom safe nonce instead of "
            "the one for last executed SafeTx + 1",
        )

    # To/value is common for send custom and send ether
    for parser in (parser_send_custom, parser_send_ether):
        parser.add_argument("to", type=check_ethereum_address)
        parser.add_argument("value", type=int)

    parser_send_custom.add_argument("data", type=check_hex_str)
    parser_send_custom.add_argument(
        "--delegate", action="store_true", help="Use DELEGATE_CALL. By default use CALL"
    )

    # Send erc20/721 have common arguments
    for parser in (parser_send_erc20, parser_send_erc721):
        parser.add_argument("to", type=check_ethereum_address)
        parser.add_argument("token_address", type=check_ethereum_address)

    parser_send_erc20.add_argument("amount", type=int)
    parser_send_erc721.add_argument("token_id", type=int)

    # Drain only needs receiver account
    parser_drain = subparsers.add_parser("drain")
    parser_drain.set_defaults(func=drain)
    parser_drain.add_argument("to", type=check_ethereum_address)

    # Retrieve threshold, nonce or owners
    parser_get_threshold = subparsers.add_parser("get_threshold")
    parser_get_threshold.set_defaults(func=get_threshold)

    parser_get_nonce = subparsers.add_parser("get_nonce")
    parser_get_nonce.set_defaults(func=get_nonce)

    parser_get_owners = subparsers.add_parser("get_owners")
    parser_get_owners.set_defaults(func=get_owners)

    # Enable and disable modules
    parser_enable_module = subparsers.add_parser("enable_module")
    parser_enable_module.add_argument("address", type=check_ethereum_address)
    parser_enable_module.set_defaults(func=enable_module)

    parser_disable_module = subparsers.add_parser("disable_module")
    parser_disable_module.add_argument("address", type=check_ethereum_address)
    parser_disable_module.set_defaults(func=disable_module)

    # Info and refresh
    parser_info = subparsers.add_parser("info")
    parser_info.set_defaults(func=get_info)

    parser_refresh = subparsers.add_parser("refresh")
    parser_refresh.set_defaults(func=get_refresh)

    # Tx-Service
    # TODO Use subcommands
    parser_tx_service = subparsers.add_parser("balances")
    parser_tx_service.set_defaults(func=get_balances)

    parser_tx_service = subparsers.add_parser("history")
    parser_tx_service.set_defaults(func=get_history)

    parser_tx_service = subparsers.add_parser("sign-tx")
    parser_tx_service.set_defaults(func=sign_tx)
    parser_tx_service.add_argument("safe_tx_hash", type=check_keccak256_hash)

    parser_tx_service = subparsers.add_parser("batch-txs")
    parser_tx_service.set_defaults(func=batch_txs)
    parser_tx_service.add_argument("safe_nonce", type=int)
    parser_tx_service.add_argument(
        "safe_tx_hashes", type=check_keccak256_hash, nargs="+"
    )

    parser_tx_service = subparsers.add_parser("execute-tx")
    parser_tx_service.set_defaults(func=execute_tx)
    parser_tx_service.add_argument("safe_tx_hash", type=check_keccak256_hash)

    # List delegates
    parser_delegates = subparsers.add_parser("get_delegates")
    parser_delegates.set_defaults(func=get_delegates)

    # Add delegate
    parser_add_delegate = subparsers.add_parser("add_delegate")
    parser_add_delegate.set_defaults(func=add_delegate)
    parser_add_delegate.add_argument("address", type=check_ethereum_address)
    parser_add_delegate.add_argument("label", type=str)
    parser_add_delegate.add_argument("signer", type=check_ethereum_address)

    # Remove delegate
    parser_remove_delegate = subparsers.add_parser("remove_delegate")
    parser_remove_delegate.set_defaults(func=remove_delegate)
    parser_remove_delegate.add_argument("address", type=check_ethereum_address)
    parser_remove_delegate.add_argument("signer", type=check_ethereum_address)

    # Remove not executed proposed transaction
    parser_remove_proposed_transaction = subparsers.add_parser(
        "remove_proposed_transaction"
    )
    parser_remove_proposed_transaction.set_defaults(func=remove_proposed_transaction)
    parser_remove_proposed_transaction.add_argument(
        "safe_tx_hash", type=check_keccak256_hash
    )

    # List all command options
    parser_help = subparsers.add_parser("help")
    parser_help.set_defaults(func=list_commands)

    # Terminate safe cli
    parser_exit = subparsers.add_parser("exit")
    parser_exit.set_defaults(func=terminate_cli)

    return prompt_parser

1	import argparse	3✔
2	import functools	3✔
3
4	from prompt_toolkit import HTML, print_formatted_text	3✔
5	from prompt_toolkit.formatted_text import html	3✔
6	from safe_eth.safe.api import SafeAPIException	3✔
7
8	from .argparse_validators import (	3✔
9	check_ethereum_address,
10	check_hex_str,
11	check_keccak256_hash,
12	)
13	from .operators import SafeServiceNotAvailable	3✔
14	from .operators.exceptions import (	3✔
15	AccountNotLoadedException,
16	ExistingOwnerException,
17	FallbackHandlerNotSupportedException,
18	HardwareWalletException,
19	HashAlreadyApproved,
20	InvalidMasterCopyException,
21	InvalidMigrationContractException,
22	InvalidNonceException,
23	NonExistingOwnerException,
24	NotEnoughEtherToSend,
25	NotEnoughSignatures,
26	NotEnoughTokenToSend,
27	SafeAlreadyUpdatedException,
28	SafeCliTerminationException,
29	SafeOperatorException,
30	SafeVersionNotSupportedException,
31	SameFallbackHandlerException,
32	SameMasterCopyException,
33	SenderRequiredException,
34	ThresholdLimitException,
35	)
36	from .operators.safe_operator import SafeOperator	3✔
37	from .safe_completer_constants import meta, safe_commands_arguments	3✔
38
39
40	def safe_exception(function):	3✔
41	@functools.wraps(function)	3✔
42	def wrapper(args, *kwargs):	3✔
43	try:	3✔
44	return function(args, *kwargs)	3✔
45	except SafeAPIException as e:	3✔
46	if e.args:	×
47	print_formatted_text(HTML(f"<b><ansired>{e.args[0]}</ansired></b>"))	×
48	except AccountNotLoadedException as e:	3✔
49	print_formatted_text(	×
50	HTML(f"<ansired>Account {e.args[0]} is not loaded</ansired>")
51	)
52	except NotEnoughSignatures as e:	3✔
53	print_formatted_text(	3✔
54	HTML(
55	f"<ansired>Cannot find enough owners to sign. {e.args[0]} missing</ansired>"
56	)
57	)
58	except SenderRequiredException:	×
59	print_formatted_text(	×
60	HTML("<ansired>Please load a default sender</ansired>")
61	)
62	except ExistingOwnerException as e:	×
63	print_formatted_text(	×
64	HTML(
65	f"<ansired>Owner {e.args[0]} is already an owner of the Safe"
66	f"</ansired>"
67	)
68	)
69	except NonExistingOwnerException as e:	×
70	print_formatted_text(	×
71	HTML(
72	f"<ansired>Owner {e.args[0]} is not an owner of the Safe"
73	f"</ansired>"
74	)
75	)
76	except HashAlreadyApproved as e:	×
77	print_formatted_text(	×
78	HTML(
79	f"<ansired>Transaction with safe-tx-hash {e.args[0].hex()} has already been approved by "
80	f"owner {e.args[1]}</ansired>"
81	)
82	)
83	except ThresholdLimitException:	×
84	print_formatted_text(	×
85	HTML(
86	"<ansired>Having less owners than threshold is not allowed"
87	"</ansired>"
88	)
89	)
90	except SameFallbackHandlerException as e:	×
91	print_formatted_text(	×
92	HTML(
93	f"<ansired>Fallback handler {e.args[0]} is the current one</ansired>"
94	)
95	)
96	except FallbackHandlerNotSupportedException:	×
97	print_formatted_text(	×
98	HTML(
99	"<ansired>Fallback handler is not supported for your Safe, "
100	"you need to <b>update</b> first</ansired>"
101	)
102	)
103	except SameMasterCopyException as e:	×
104	print_formatted_text(	×
105	HTML(f"<ansired>Master Copy {e.args[0]} is the current one</ansired>")
106	)
107	except InvalidMasterCopyException as e:	×
108	print_formatted_text(	×
109	HTML(f"<ansired>Master Copy {e.args[0]} is not valid</ansired>")
110	)
111	except InvalidMigrationContractException as e:	×
112	print_formatted_text(HTML(f"<ansired>{e.args[0]}</ansired>"))	×
113	except InvalidNonceException as e:	×
114	print_formatted_text(HTML(f"<ansired>{e.args[0]}</ansired>"))	×
115	except SafeAlreadyUpdatedException:	×
116	print_formatted_text(HTML("<ansired>Safe is already updated</ansired>"))	×
117	except SafeVersionNotSupportedException as e:	×
118	print_formatted_text(HTML(f"<ansired>{e.args[0]}</ansired>"))	×
119	except (NotEnoughEtherToSend, NotEnoughTokenToSend) as e:	×
120	print_formatted_text(	×
121	HTML(
122	f"<ansired>Cannot find enough to send. Current balance is {e.args[0]}"
123	f"</ansired>"
124	)
125	)
126	except SafeServiceNotAvailable as e:	×
127	print_formatted_text(	×
128	HTML(
129	f"<ansired>Service not available for network {e.args[0]}</ansired>"
130	)
131	)
132	except HardwareWalletException as e:	×
133	print_formatted_text(	×
134	HTML(f"<ansired>HwDevice exception: {e.args[0]}</ansired>")
135	)
136	except SafeOperatorException as e:	×
137	print_formatted_text(HTML(f"<ansired>{e.args[0]}</ansired>"))	×
138
139	return wrapper	3✔
140
141
142	class PromptParser:	3✔
143	def __init__(self, safe_operator: SafeOperator):	3✔
144	self.mode_parser = argparse.ArgumentParser(prog="")	3✔
145	self.safe_operator = safe_operator	3✔
146	self.prompt_parser = build_prompt_parser(safe_operator)	3✔
147
148	def process_command(self, command: str):	3✔
149	args = self.prompt_parser.parse_args(command.split())	3✔
150	return args.func(args)	3✔
151
152
153	def build_prompt_parser(safe_operator: SafeOperator) -> argparse.ArgumentParser:	3✔
154	"""
155	Returns an ArgParse capable of decoding and executing the Safe commands
156	:param safe_operator:
157	:return:
158	"""
159	prompt_parser = argparse.ArgumentParser(prog="")	3✔
160	subparsers = prompt_parser.add_subparsers()	3✔
161
162	@safe_exception	3✔
163	def show_cli_owners(args):	3✔
164	safe_operator.show_cli_owners()	×
165
166	@safe_exception	3✔
167	def load_cli_owners_from_words(args):	3✔
168	safe_operator.load_cli_owners_from_words(args.words)	×
169
170	@safe_exception	3✔
171	def load_cli_owners(args):	3✔
172	safe_operator.load_cli_owners(args.keys)	3✔
173
174	@safe_exception	3✔
175	def load_ledger_cli_owners(args):	3✔
176	safe_operator.load_ledger_cli_owners(	×
177	derivation_path=args.derivation_path, legacy_account=args.legacy_accounts
178	)
179
180	@safe_exception	3✔
181	def load_trezor_cli_owners(args):	3✔
182	safe_operator.load_trezor_cli_owners(	×
183	derivation_path=args.derivation_path, legacy_account=args.legacy_accounts
184	)
185
186	@safe_exception	3✔
187	def unload_cli_owners(args):	3✔
188	safe_operator.unload_cli_owners(args.addresses)	×
189
190	@safe_exception	3✔
191	def approve_hash(args):	3✔
192	safe_operator.approve_hash(args.hash_to_approve, args.sender)	3✔
193
194	@safe_exception	3✔
195	def sign_message(args):	3✔
196	safe_operator.sign_message(args.eip712_path)	×
197
198	@safe_exception	3✔
199	def confirm_message(args):	3✔
200	safe_operator.confirm_message(args.safe_message_hash, args.sender)	×
201
202	@safe_exception	3✔
203	def add_owner(args):	3✔
204	safe_operator.add_owner(args.address, threshold=args.threshold)	×
205
206	@safe_exception	3✔
207	def remove_owner(args):	3✔
208	safe_operator.remove_owner(args.address, threshold=args.threshold)	3✔
209
210	@safe_exception	3✔
211	def change_fallback_handler(args):	3✔
212	safe_operator.change_fallback_handler(args.address)	×
213
214	@safe_exception	3✔
215	def change_guard(args):	3✔
216	safe_operator.change_guard(args.address)	×
217
218	@safe_exception	3✔
219	def change_master_copy(args):	3✔
220	safe_operator.change_master_copy(args.address)	×
221
222	@safe_exception	3✔
223	def change_threshold(args):	3✔
224	safe_operator.change_threshold(args.threshold)	3✔
225
226	@safe_exception	3✔
227	def send_custom(args):	3✔
228	safe_operator.send_custom(	×
229	args.to,
230	args.value,
231	args.data,
232	safe_nonce=args.safe_nonce,
233	delegate_call=args.delegate,
234	)
235
236	@safe_exception	3✔
237	def send_ether(args):	3✔
238	safe_operator.send_ether(args.to, args.value, safe_nonce=args.safe_nonce)	3✔
239
240	@safe_exception	3✔
241	def send_erc20(args):	3✔
242	safe_operator.send_erc20(	×
243	args.to, args.token_address, args.amount, safe_nonce=args.safe_nonce
244	)
245
246	@safe_exception	3✔
247	def send_erc721(args):	3✔
248	safe_operator.send_erc721(	×
249	args.to, args.token_address, args.token_id, safe_nonce=args.safe_nonce
250	)
251
252	@safe_exception	3✔
253	def drain(args):	3✔
254	safe_operator.drain(args.to)	×
255
256	@safe_exception	3✔
257	def get_threshold(args):	3✔
258	safe_operator.get_threshold()	×
259
260	@safe_exception	3✔
261	def get_nonce(args):	3✔
262	safe_operator.get_nonce()	×
263
264	@safe_exception	3✔
265	def get_owners(args):	3✔
266	safe_operator.get_owners()	×
267
268	@safe_exception	3✔
269	def enable_module(args):	3✔
270	safe_operator.enable_module(args.address)	×
271
272	@safe_exception	3✔
273	def disable_module(args):	3✔
274	safe_operator.disable_module(args.address)	×
275
276	@safe_exception	3✔
277	def update_version(args):	3✔
278	safe_operator.update_version()	×
279
280	@safe_exception	3✔
281	def update_version_to_l2(args):	3✔
282	safe_operator.update_version_to_l2(args.migration_contract)	×
283
284	@safe_exception	3✔
285	def get_info(args):	3✔
286	safe_operator.print_info()	×
287
288	@safe_exception	3✔
289	def get_refresh(args):	3✔
290	safe_operator.refresh_safe_cli_info()	×
291
292	@safe_exception	3✔
293	def get_balances(args):	3✔
294	safe_operator.get_balances()	×
295
296	@safe_exception	3✔
297	def get_history(args):	3✔
298	safe_operator.get_transaction_history()	×
299
300	@safe_exception	3✔
301	def sign_tx(args):	3✔
302	safe_operator.submit_signatures(args.safe_tx_hash)	×
303
304	@safe_exception	3✔
305	def batch_txs(args):	3✔
306	safe_operator.batch_txs(args.safe_nonce, args.safe_tx_hashes)	×
307
308	@safe_exception	3✔
309	def execute_tx(args):	3✔
310	safe_operator.execute_tx(args.safe_tx_hash)	×
311
312	@safe_exception	3✔
313	def get_delegates(args):	3✔
314	safe_operator.get_delegates()	×
315
316	@safe_exception	3✔
317	def add_delegate(args):	3✔
318	safe_operator.add_delegate(args.address, args.label, args.signer)	×
319
320	@safe_exception	3✔
321	def remove_delegate(args):	3✔
322	safe_operator.remove_delegate(args.address, args.signer)	×
323
324	@safe_exception	3✔
325	def remove_proposed_transaction(args):	3✔
326	safe_operator.remove_proposed_transaction(args.safe_tx_hash)	×
327
328	def list_commands(args):	3✔
329	print_formatted_text(	×
330	HTML("<b>The following commands can be used:</b>"), end="\n\n"
331	)
332	for key in sorted(safe_commands_arguments.keys()):	×
333	print_formatted_text(	×
334	HTML(
335	f"<b>{key} {html.html_escape(safe_commands_arguments.get(key))}</b>"
336	)
337	)
338	print_formatted_text(HTML(" "), meta.get(key, ""))	×
339	print_formatted_text(	×
340	HTML("\nUse the <b>tab key</b> to show options in interactive mode.")
341	)
342
343	def terminate_cli(args):	3✔
344	raise SafeCliTerminationException()	3✔
345
346	# Cli owners
347	parser_show_cli_owners = subparsers.add_parser("show_cli_owners")	3✔
348	parser_show_cli_owners.set_defaults(func=show_cli_owners)	3✔
349
350	parser_load_cli_owners_from_words = subparsers.add_parser(	3✔
351	"load_cli_owners_from_words"
352	)
353	parser_load_cli_owners_from_words.add_argument("words", type=str, nargs="+")	3✔
354	parser_load_cli_owners_from_words.set_defaults(func=load_cli_owners_from_words)	3✔
355
356	parser_load_cli_owners = subparsers.add_parser("load_cli_owners")	3✔
357	parser_load_cli_owners.add_argument("keys", type=str, nargs="+")	3✔
358	parser_load_cli_owners.set_defaults(func=load_cli_owners)	3✔
359
360	parser_load_ledger_cli_owners = subparsers.add_parser("load_ledger_cli_owners")	3✔
361	parser_load_ledger_cli_owners.add_argument(	3✔
362	"--derivation-path",
363	type=str,
364	help="Load address for the provided derivation path",
365	)
366	parser_load_ledger_cli_owners.add_argument(	3✔
367	"--legacy-accounts",
368	action="store_true",
369	help="Search for legacy accounts",
370	)
371	parser_load_ledger_cli_owners.set_defaults(func=load_ledger_cli_owners)	3✔
372
373	parser_load_trezor_cli_owners = subparsers.add_parser("load_trezor_cli_owners")	3✔
374	parser_load_trezor_cli_owners.add_argument(	3✔
375	"--derivation-path",
376	type=str,
377	help="Load address for the provided derivation path",
378	)
379	parser_load_trezor_cli_owners.add_argument(	3✔
380	"--legacy-accounts",
381	action="store_true",
382	help="Search for legacy accounts",
383	)
384	parser_load_trezor_cli_owners.set_defaults(func=load_trezor_cli_owners)	3✔
385
386	parser_unload_cli_owners = subparsers.add_parser("unload_cli_owners")	3✔
387	parser_unload_cli_owners.add_argument(	3✔
388	"addresses", type=check_ethereum_address, nargs="+"
389	)
390	parser_unload_cli_owners.set_defaults(func=unload_cli_owners)	3✔
391
392	# Change threshold
393	parser_change_threshold = subparsers.add_parser("change_threshold")	3✔
394	parser_change_threshold.add_argument("threshold", type=int)	3✔
395	parser_change_threshold.set_defaults(func=change_threshold)	3✔
396
397	# Approve hash
398	parser_approve_hash = subparsers.add_parser("approve_hash")	3✔
399	parser_approve_hash.add_argument("hash_to_approve", type=check_keccak256_hash)	3✔
400	parser_approve_hash.add_argument("sender", type=check_ethereum_address)	3✔
401	parser_approve_hash.set_defaults(func=approve_hash)	3✔
402
403	# Sign message
404	parser_sign_message = subparsers.add_parser("sign_message")	3✔
405	group = parser_sign_message.add_mutually_exclusive_group(required=True)	3✔
406	group.add_argument("--eip191_message", action="store_true")	3✔
407	group.add_argument("--eip712_path", type=str)	3✔
408	parser_sign_message.set_defaults(func=sign_message)	3✔
409
410	# Confirm message
411	parser_confirm_message = subparsers.add_parser("confirm_message")	3✔
412	parser_confirm_message.add_argument("safe_message_hash", type=check_keccak256_hash)	3✔
413	parser_confirm_message.add_argument("sender", type=check_ethereum_address)	3✔
414	parser_confirm_message.set_defaults(func=confirm_message)	3✔
415
416	# Add owner
417	parser_add_owner = subparsers.add_parser("add_owner")	3✔
418	parser_add_owner.add_argument("address", type=check_ethereum_address)	3✔
419	parser_add_owner.add_argument("--threshold", type=int, default=None)	3✔
420	parser_add_owner.set_defaults(func=add_owner)	3✔
421
422	# Remove owner
423	parser_remove_owner = subparsers.add_parser("remove_owner")	3✔
424	parser_remove_owner.add_argument("address", type=check_ethereum_address)	3✔
425	parser_remove_owner.add_argument("--threshold", type=int, default=None)	3✔
426	parser_remove_owner.set_defaults(func=remove_owner)	3✔
427
428	# Change FallbackHandler
429	parser_change_fallback_handler = subparsers.add_parser("change_fallback_handler")	3✔
430	parser_change_fallback_handler.add_argument("address", type=check_ethereum_address)	3✔
431	parser_change_fallback_handler.set_defaults(func=change_fallback_handler)	3✔
432
433	# Change Guard
434	parser_change_guard = subparsers.add_parser("change_guard")	3✔
435	parser_change_guard.add_argument("address", type=check_ethereum_address)	3✔
436	parser_change_guard.set_defaults(func=change_guard)	3✔
437
438	# Change MasterCopy
439	parser_change_master_copy = subparsers.add_parser("change_master_copy")	3✔
440	parser_change_master_copy.add_argument("address", type=check_ethereum_address)	3✔
441	parser_change_master_copy.set_defaults(func=change_master_copy)	3✔
442
443	# Update Safe to last version
444	parser_update_version = subparsers.add_parser("update")	3✔
445	parser_update_version.set_defaults(func=update_version)	3✔
446
447	# Update non L2 Safe to L2 Safe
448	parser_update_version_to_l2 = subparsers.add_parser("update_version_to_l2")	3✔
449	parser_update_version_to_l2.add_argument(	3✔
450	"migration_contract", type=check_ethereum_address
451	)
452	parser_update_version_to_l2.set_defaults(func=update_version_to_l2)	3✔
453
454	# Send custom/ether/erc20/erc721
455	parser_send_custom = subparsers.add_parser("send_custom")	3✔
456	parser_send_ether = subparsers.add_parser("send_ether")	3✔
457	parser_send_erc20 = subparsers.add_parser("send_erc20")	3✔
458	parser_send_erc721 = subparsers.add_parser("send_erc721")	3✔
459	parser_send_custom.set_defaults(func=send_custom)	3✔
460	parser_send_ether.set_defaults(func=send_ether)	3✔
461	parser_send_erc20.set_defaults(func=send_erc20)	3✔
462	parser_send_erc721.set_defaults(func=send_erc721)	3✔
463	# They have some common arguments
464	for parser in (	3✔
465	parser_send_custom,
466	parser_send_ether,
467	parser_send_erc20,
468	parser_send_erc721,
469	):
470	parser.add_argument(	3✔
471	"--safe-nonce",
472	type=int,
473	help="Use custom safe nonce instead of "
474	"the one for last executed SafeTx + 1",
475	)
476
477	# To/value is common for send custom and send ether
478	for parser in (parser_send_custom, parser_send_ether):	3✔
479	parser.add_argument("to", type=check_ethereum_address)	3✔
480	parser.add_argument("value", type=int)	3✔
481
482	parser_send_custom.add_argument("data", type=check_hex_str)	3✔
483	parser_send_custom.add_argument(	3✔
484	"--delegate", action="store_true", help="Use DELEGATE_CALL. By default use CALL"
485	)
486
487	# Send erc20/721 have common arguments
488	for parser in (parser_send_erc20, parser_send_erc721):	3✔
489	parser.add_argument("to", type=check_ethereum_address)	3✔
490	parser.add_argument("token_address", type=check_ethereum_address)	3✔
491
492	parser_send_erc20.add_argument("amount", type=int)	3✔
493	parser_send_erc721.add_argument("token_id", type=int)	3✔
494
495	# Drain only needs receiver account
496	parser_drain = subparsers.add_parser("drain")	3✔
497	parser_drain.set_defaults(func=drain)	3✔
498	parser_drain.add_argument("to", type=check_ethereum_address)	3✔
499
500	# Retrieve threshold, nonce or owners
501	parser_get_threshold = subparsers.add_parser("get_threshold")	3✔
502	parser_get_threshold.set_defaults(func=get_threshold)	3✔
503
504	parser_get_nonce = subparsers.add_parser("get_nonce")	3✔
505	parser_get_nonce.set_defaults(func=get_nonce)	3✔
506
507	parser_get_owners = subparsers.add_parser("get_owners")	3✔
508	parser_get_owners.set_defaults(func=get_owners)	3✔
509
510	# Enable and disable modules
511	parser_enable_module = subparsers.add_parser("enable_module")	3✔
512	parser_enable_module.add_argument("address", type=check_ethereum_address)	3✔
513	parser_enable_module.set_defaults(func=enable_module)	3✔
514
515	parser_disable_module = subparsers.add_parser("disable_module")	3✔
516	parser_disable_module.add_argument("address", type=check_ethereum_address)	3✔
517	parser_disable_module.set_defaults(func=disable_module)	3✔
518
519	# Info and refresh
520	parser_info = subparsers.add_parser("info")	3✔
521	parser_info.set_defaults(func=get_info)	3✔
522
523	parser_refresh = subparsers.add_parser("refresh")	3✔
524	parser_refresh.set_defaults(func=get_refresh)	3✔
525
526	# Tx-Service
527	# TODO Use subcommands
528	parser_tx_service = subparsers.add_parser("balances")	3✔
529	parser_tx_service.set_defaults(func=get_balances)	3✔
530
531	parser_tx_service = subparsers.add_parser("history")	3✔
532	parser_tx_service.set_defaults(func=get_history)	3✔
533
534	parser_tx_service = subparsers.add_parser("sign-tx")	3✔
535	parser_tx_service.set_defaults(func=sign_tx)	3✔
536	parser_tx_service.add_argument("safe_tx_hash", type=check_keccak256_hash)	3✔
537
538	parser_tx_service = subparsers.add_parser("batch-txs")	3✔
539	parser_tx_service.set_defaults(func=batch_txs)	3✔
540	parser_tx_service.add_argument("safe_nonce", type=int)	3✔
541	parser_tx_service.add_argument(	3✔
542	"safe_tx_hashes", type=check_keccak256_hash, nargs="+"
543	)
544
545	parser_tx_service = subparsers.add_parser("execute-tx")	3✔
546	parser_tx_service.set_defaults(func=execute_tx)	3✔
547	parser_tx_service.add_argument("safe_tx_hash", type=check_keccak256_hash)	3✔
548
549	# List delegates
550	parser_delegates = subparsers.add_parser("get_delegates")	3✔
551	parser_delegates.set_defaults(func=get_delegates)	3✔
552
553	# Add delegate
554	parser_add_delegate = subparsers.add_parser("add_delegate")	3✔
555	parser_add_delegate.set_defaults(func=add_delegate)	3✔
556	parser_add_delegate.add_argument("address", type=check_ethereum_address)	3✔
557	parser_add_delegate.add_argument("label", type=str)	3✔
558	parser_add_delegate.add_argument("signer", type=check_ethereum_address)	3✔
559
560	# Remove delegate
561	parser_remove_delegate = subparsers.add_parser("remove_delegate")	3✔
562	parser_remove_delegate.set_defaults(func=remove_delegate)	3✔
563	parser_remove_delegate.add_argument("address", type=check_ethereum_address)	3✔
564	parser_remove_delegate.add_argument("signer", type=check_ethereum_address)	3✔
565
566	# Remove not executed proposed transaction
567	parser_remove_proposed_transaction = subparsers.add_parser(	3✔
568	"remove_proposed_transaction"
569	)
570	parser_remove_proposed_transaction.set_defaults(func=remove_proposed_transaction)	3✔
571	parser_remove_proposed_transaction.add_argument(	3✔
572	"safe_tx_hash", type=check_keccak256_hash
573	)
574
575	# List all command options
576	parser_help = subparsers.add_parser("help")	3✔
577	parser_help.set_defaults(func=list_commands)	3✔
578
579	# Terminate safe cli
580	parser_exit = subparsers.add_parser("exit")	3✔
581	parser_exit.set_defaults(func=terminate_cli)	3✔
582
583	return prompt_parser	3✔

safe-global / safe-cli / 11553543762

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