diff --git a/src/safe_cli/ethereum_hd_wallet.py b/src/safe_cli/ethereum_hd_wallet.py index c8dc1577..7463529d 100644 --- a/src/safe_cli/ethereum_hd_wallet.py +++ b/src/safe_cli/ethereum_hd_wallet.py @@ -17,7 +17,9 @@ def get_account_from_words( :raises: eth_utils.ValidationError """ Account.enable_unaudited_hdwallet_features() - if index: + # Derive from the base path by index only when the caller did not + # supply a custom hd_path; otherwise honor the provided hd_path + if index and hd_path == ETHEREUM_DEFAULT_PATH: hd_path = f"{ETHEREUM_BASE_PATH}/{index}" return Account.from_mnemonic(words, account_path=hd_path) diff --git a/src/safe_cli/main.py b/src/safe_cli/main.py index dc46461e..84cbeb54 100644 --- a/src/safe_cli/main.py +++ b/src/safe_cli/main.py @@ -17,6 +17,7 @@ from .argparse_validators import check_hex_str from .operators import SafeOperator from .safe_cli import SafeCli +from .tx_builder.exceptions import SoliditySyntaxError, TxBuilderEncodingError from .tx_builder.tx_builder_file_decoder import convert_to_proposed_transactions from .typer_validators import ( ChecksumAddressParser, @@ -46,7 +47,7 @@ def _check_interactive_mode(interactive_mode: bool) -> bool: # --non-interactive arg > env var. env_var = os.getenv("SAFE_CLI_INTERACTIVE") - if env_var: + if env_var is not None: return env_var.lower() in ("true", "1", "yes") return True @@ -289,11 +290,19 @@ def tx_builder( safe_operator = _build_safe_operator_and_load_keys( safe_address, node_url, private_key, interactive ) - data = json.loads(file_path.read_text()) - safe_txs = [ - safe_operator.prepare_safe_transaction(tx.to, tx.value, tx.data) - for tx in convert_to_proposed_transactions(data) - ] + try: + data = json.loads(file_path.read_text()) + safe_txs = [ + safe_operator.prepare_safe_transaction(tx.to, tx.value, tx.data) + for tx in convert_to_proposed_transactions(data) + ] + except ( + json.JSONDecodeError, + KeyError, + SoliditySyntaxError, + TxBuilderEncodingError, + ) as e: + raise typer.BadParameter(f"Invalid tx-builder file: {e}") from e if len(safe_txs) == 0: raise typer.BadParameter("No transactions found.") @@ -385,7 +394,7 @@ def _is_safe_cli_default_command(arguments: list[str]) -> bool: # Only added if is not a valid command, and it is an address. safe-cli 0xaddress http://url if arguments[1] not in [ get_command_name(key) for key in get_command(app).commands.keys() - ] and Web3.is_checksum_address(arguments[1]): + ] and Web3.is_address(arguments[1]): return True return False diff --git a/src/safe_cli/operators/hw_wallets/ledger_exceptions.py b/src/safe_cli/operators/hw_wallets/ledger_exceptions.py index 5ae41280..9fa406e1 100644 --- a/src/safe_cli/operators/hw_wallets/ledger_exceptions.py +++ b/src/safe_cli/operators/hw_wallets/ledger_exceptions.py @@ -27,7 +27,7 @@ def wrapper(*args, **kwargs): except InvalidDerivationPath as e: raise HardwareWalletException(e.message) from e except BaseException as e: - if "Error while writing" in e.args: + if "Error while writing" in str(e): raise HardwareWalletException( "Ledger error writing, restart safe-cli" ) from e diff --git a/src/safe_cli/operators/hw_wallets/trezor_wallet.py b/src/safe_cli/operators/hw_wallets/trezor_wallet.py index bba100bc..cb328d65 100644 --- a/src/safe_cli/operators/hw_wallets/trezor_wallet.py +++ b/src/safe_cli/operators/hw_wallets/trezor_wallet.py @@ -116,7 +116,7 @@ def get_signed_raw_transaction( gas_limit=tx_parameters["gas"], to=tx_parameters["to"], value=tx_parameters["value"], - data=HexBytes(tx_parameters.get("data")), + data=HexBytes(tx_parameters["data"]), chain_id=chain_id, ) diff --git a/src/safe_cli/operators/safe_operator.py b/src/safe_cli/operators/safe_operator.py index 10b693cf..0c94779b 100644 --- a/src/safe_cli/operators/safe_operator.py +++ b/src/safe_cli/operators/safe_operator.py @@ -259,7 +259,15 @@ def is_version_updated(self) -> bool: def load_cli_owners_from_words(self, words: list[str]): if len(words) == 1: # Reading seed from Environment Variable words = os.environ.get(words[0], default="").strip().split(" ") - parsed_words = " ".join(words) + parsed_words = " ".join(words).strip() + if not parsed_words: + print_formatted_text( + HTML( + "Cannot load owners from words: empty seed " + "(check the mnemonic or the environment variable)" + ) + ) + return try: accounts = [] for index in range(100): # Try first 100 accounts of seed phrase @@ -600,7 +608,9 @@ def change_fallback_handler(self, new_fallback_handler: str) -> bool: elif semantic_version.parse( self.safe_cli_info.version ) < semantic_version.parse("1.1.0"): - raise FallbackHandlerNotSupportedException() + raise FallbackHandlerNotSupportedException( + "Fallback handler is not supported for Safes with version < 1.1.0" + ) elif ( new_fallback_handler != NULL_ADDRESS and not self.ethereum_client.is_contract(new_fallback_handler) @@ -616,6 +626,7 @@ def change_fallback_handler(self, new_fallback_handler: str) -> bool: self.safe_cli_info.fallback_handler = new_fallback_handler self.safe_cli_info.version = self.safe.retrieve_version() return True + return False def change_guard(self, guard: str) -> bool: if guard == self.safe_cli_info.guard: @@ -623,7 +634,9 @@ def change_guard(self, guard: str) -> bool: elif semantic_version.parse( self.safe_cli_info.version ) < semantic_version.parse("1.3.0"): - raise GuardNotSupportedException() + raise GuardNotSupportedException( + "Guard is not supported for Safes with version < 1.3.0" + ) elif guard != NULL_ADDRESS and not self.ethereum_client.is_contract(guard): raise InvalidGuardException(f"{guard} address is not a contract") else: @@ -634,6 +647,7 @@ def change_guard(self, guard: str) -> bool: self.safe_cli_info.guard = guard self.safe_cli_info.version = self.safe.retrieve_version() return True + return False def change_module_guard(self, module_guard: str) -> bool: if module_guard == self.safe_cli_info.module_guard: @@ -661,6 +675,7 @@ def change_module_guard(self, module_guard: str) -> bool: self.safe_cli_info.module_guard = module_guard self.safe_cli_info.version = self.safe.retrieve_version() return True + return False def change_master_copy(self, new_master_copy: str) -> bool: if new_master_copy == self.safe_cli_info.master_copy: @@ -684,6 +699,7 @@ def change_master_copy(self, new_master_copy: str) -> bool: self.safe_cli_info.master_copy = new_master_copy self.safe_cli_info.version = self.safe.retrieve_version() return True + return False def update_version(self) -> bool | None: """ @@ -739,6 +755,8 @@ def update_version(self) -> bool | None: self.last_default_fallback_handler_address ) self.safe_cli_info.version = self.safe.retrieve_version() + return True + return False def update_version_to_l2( self, migration_contract_address: ChecksumAddress @@ -803,12 +821,18 @@ def update_version_to_l2( self.safe_cli_info.master_copy = safe_l2_singleton self.safe_cli_info.fallback_handler = fallback_handler self.safe_cli_info.version = self.safe.retrieve_version() + return True + return False def change_threshold(self, threshold: int): if threshold == self.safe_cli_info.threshold: print_formatted_text( HTML(f"Threshold is already {threshold}") ) + elif threshold < 1: + print_formatted_text( + HTML("Threshold must be at least 1") + ) elif threshold > len(self.safe_cli_info.owners): print_formatted_text( HTML( @@ -1044,7 +1068,7 @@ def execute_safe_transaction(self, safe_tx: SafeTx): f"deducted={fees}" ) ) - self.safe_cli_info.nonce += 1 + self.safe_cli_info.nonce = safe_tx.safe_nonce + 1 return True else: print_formatted_text( diff --git a/src/safe_cli/operators/safe_tx_service_operator.py b/src/safe_cli/operators/safe_tx_service_operator.py index d1c88c5f..680e45e0 100644 --- a/src/safe_cli/operators/safe_tx_service_operator.py +++ b/src/safe_cli/operators/safe_tx_service_operator.py @@ -77,6 +77,7 @@ def sign_message( print_formatted_text( HTML("At least one owner must be loaded") ) + return False if self.safe_tx_service.post_message(self.address, message, signature): print_formatted_text( @@ -114,6 +115,7 @@ def confirm_message(self, safe_message_hash: bytes, sender: ChecksumAddress): print_formatted_text( HTML(f"Owner with address {sender} was not loaded") ) + return False if isinstance(signer, LocalAccount): signature = signer.unsafe_sign_hash(safe_message_hash).signature diff --git a/src/safe_cli/prompt_parser.py b/src/safe_cli/prompt_parser.py index 0666273a..3e09108f 100644 --- a/src/safe_cli/prompt_parser.py +++ b/src/safe_cli/prompt_parser.py @@ -1,5 +1,6 @@ import argparse import functools +import shlex from prompt_toolkit import HTML, print_formatted_text from prompt_toolkit.formatted_text import html @@ -134,7 +135,8 @@ def wrapper(*args, **kwargs): HTML(f"HwDevice exception: {e.args[0]}") ) except SafeOperatorException as e: - print_formatted_text(HTML(f"{e.args[0]}")) + message = e.args[0] if e.args else type(e).__name__ + print_formatted_text(HTML(f"{message}")) return wrapper @@ -146,7 +148,14 @@ def __init__(self, safe_operator: SafeOperator): self.prompt_parser = build_prompt_parser(safe_operator) def process_command(self, command: str): - args = self.prompt_parser.parse_args(command.split()) + try: + split_command = shlex.split(command) + except ValueError as e: + print_formatted_text( + HTML(f"Invalid command syntax: {e}") + ) + return + args = self.prompt_parser.parse_args(split_command) return args.func(args) diff --git a/src/safe_cli/safe_completer.py b/src/safe_cli/safe_completer.py index 2db06acc..74272b3a 100644 --- a/src/safe_cli/safe_completer.py +++ b/src/safe_cli/safe_completer.py @@ -3,8 +3,9 @@ from prompt_toolkit.document import Document from .safe_completer_constants import ( + SAFE_ARGUMENT_COLOR, + SAFE_EMPTY_ARGUMENT_COLOR, meta, - safe_color_arguments, safe_commands, safe_commands_arguments, ) @@ -32,8 +33,10 @@ def get_completions( if command.startswith(word.lower()): if command in safe_commands_arguments: safe_command = safe_commands_arguments[command] - safe_argument_color = safe_color_arguments.get( - safe_command, "default" + safe_argument_color = ( + SAFE_ARGUMENT_COLOR + if safe_command + else SAFE_EMPTY_ARGUMENT_COLOR ) display = HTML( " > %s <" diff --git a/src/safe_cli/safe_completer_constants.py b/src/safe_cli/safe_completer_constants.py index e700b340..9c0e8842 100644 --- a/src/safe_cli/safe_completer_constants.py +++ b/src/safe_cli/safe_completer_constants.py @@ -14,7 +14,7 @@ "change_guard": "
", "change_module_guard": "
", "change_master_copy": "
", - "change_threshold": "
", + "change_threshold": "", "disable_module": "
", "enable_module": "
", "execute-tx": "", @@ -52,18 +52,6 @@ safe_commands = list(safe_commands_arguments.keys()) -safe_color_arguments = { - "(read-only)": SAFE_ARGUMENT_COLOR, - "": SAFE_ARGUMENT_COLOR, - "
": SAFE_ARGUMENT_COLOR, - "": SAFE_ARGUMENT_COLOR, - "": SAFE_ARGUMENT_COLOR, - "": SAFE_ARGUMENT_COLOR, - "": SAFE_ARGUMENT_COLOR, - "": SAFE_ARGUMENT_COLOR, - "": SAFE_ARGUMENT_COLOR, -} - meta = { "approve_hash": HTML( "approve_hash will approve a safe-tx-hash for the provided sender address. " @@ -105,10 +93,6 @@ "get_delegates": HTML( "Command get_delegates will return information about the current delegates." ), - "change_owner": HTML( - "Command change_owner will change an old account <address> for the new " - "check-summed <address> account." - ), "add_owner": HTML( "Command add_owner will add a check-summed <address> owner account." ), diff --git a/src/safe_cli/safe_lexer.py b/src/safe_cli/safe_lexer.py index 5309129a..06c98dca 100644 --- a/src/safe_cli/safe_lexer.py +++ b/src/safe_cli/safe_lexer.py @@ -8,7 +8,7 @@ class SafeLexer(BashLexer): name = "SafeLexer" aliases = ["safe_lexer"] - ADDRESS = r"^0x[aA-zZ,0-9]{40}$|^0x[aA-zZ,0-9]{62}$" + ADDRESS = r"^0x[a-fA-F0-9]{40}$|^0x[a-fA-F0-9]{64}$" EXTRA_KEYWORDS = { "refresh", "get_nonce", diff --git a/src/safe_cli/tx_builder/tx_builder_file_decoder.py b/src/safe_cli/tx_builder/tx_builder_file_decoder.py index cfb37b81..aaf80fee 100644 --- a/src/safe_cli/tx_builder/tx_builder_file_decoder.py +++ b/src/safe_cli/tx_builder/tx_builder_file_decoder.py @@ -21,7 +21,10 @@ def _parse_types_to_encoding_types(contract_fields: list[dict[str, Any]]) -> lis component_types = ",".join( component["type"] for component in field["components"] ) - types.append(f"({component_types})") + # Preserve any array suffix (e.g. tuple[], tuple[2]) so the + # function selector and ABI encoding target the right signature + array_suffix = field["type"].removeprefix("tuple") + types.append(f"({component_types}){array_suffix}") else: types.append(field["type"]) @@ -42,14 +45,16 @@ def encode_contract_method_to_hex_data( if not is_valid_contract_method: return None + contract_fields_values = contract_fields_values or {} try: encoding_types = _parse_types_to_encoding_types(contract_fields) - values = [ - parse_input_value( - field["type"], contract_fields_values.get(field["name"], "") + values = [] + for field in contract_fields: + if field["name"] not in contract_fields_values: + raise SoliditySyntaxError(f"Missing value for input '{field['name']}'") + values.append( + parse_input_value(field["type"], contract_fields_values[field["name"]]) ) - for field in contract_fields - ] function_signature = f"{contract_method_name}({','.join(encoding_types)})" function_selector = Web3.keccak(text=function_signature)[:4] @@ -81,9 +86,7 @@ def parse_int_value(value: str) -> int: if trimmed_value == "": raise SoliditySyntaxError("Invalid empty strings for integers") try: - if not trimmed_value.isdigit() and bool( - re.fullmatch(r"0[xX][0-9a-fA-F]+|[0-9a-fA-F]+$", trimmed_value) - ): + if re.fullmatch(r"0[xX][0-9a-fA-F]+", trimmed_value): return int(trimmed_value, 16) return int(trimmed_value) @@ -176,18 +179,26 @@ def is_multi_dimensional_array_field_type(field_type: str) -> bool: return field_type.count("[") > 1 +def is_any_array_field_type(field_type: str) -> bool: + return is_array_field_type(field_type) or is_multi_dimensional_array_field_type( + field_type + ) + + def parse_input_value(field_type: str, value: str) -> Any: trimmed_value = value.strip() if isinstance(value, str) else value if is_tuple_field_type(field_type): - return tuple(json.loads(trimmed_value)) + parsed = json.loads(trimmed_value) + if is_any_array_field_type(field_type): + # Array of tuples (e.g. tuple[]): a list of tuples, not one tuple + return [tuple(item) for item in parsed] + return tuple(parsed) if is_array_of_strings_field_type(field_type): return json.loads(trimmed_value) - if is_array_field_type(field_type) or is_multi_dimensional_array_field_type( - field_type - ): + if is_any_array_field_type(field_type): return parse_array_of_values(trimmed_value, field_type) if is_boolean_field_type(field_type): @@ -228,11 +239,14 @@ def convert_to_proposed_transactions( to_0x_hex_str(encoded_data) if encoded_data is not None else "0x" ) + raw_value = transaction.get("value") + value = parse_int_value(str(raw_value)) if raw_value not in (None, "") else 0 + proposed_transactions.append( SafeProposedTx( id=index, to=transaction.get("to"), - value=transaction.get("value"), + value=value, data=data_value, ) )