Fix correctness bugs and clean up dead code across CLI

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)
 

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

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

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

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(
+                    "<ansired>Cannot load owners from words: empty seed "
+                    "(check the mnemonic or the environment variable)</ansired>"
+                )
+            )
+            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,14 +626,17 @@ 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:
             raise SameGuardException(guard)
         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"<ansired>Threshold is already {threshold}</ansired>")
             )
+        elif threshold < 1:
+            print_formatted_text(
+                HTML("<ansired>Threshold must be at least 1</ansired>")
+            )
         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}</ansigreen>"
                         )
                     )
-                    self.safe_cli_info.nonce += 1
+                    self.safe_cli_info.nonce = safe_tx.safe_nonce + 1
                     return True
                 else:
                     print_formatted_text(

src/safe_cli/operators/safe_tx_service_operator.py

@@ -77,6 +77,7 @@ def sign_message(
             print_formatted_text(
                 HTML("<ansired>At least one owner must be loaded</ansired>")
             )
+            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"<ansired>Owner with address {sender} was not loaded</ansired>")
             )
+            return False
 
         if isinstance(signer, LocalAccount):
             signature = signer.unsafe_sign_hash(safe_message_hash).signature

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"<ansired>HwDevice exception: {e.args[0]}</ansired>")
             )
         except SafeOperatorException as e:
-            print_formatted_text(HTML(f"<ansired>{e.args[0]}</ansired>"))
+            message = e.args[0] if e.args else type(e).__name__
+            print_formatted_text(HTML(f"<ansired>{message}</ansired>"))
 
     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"<ansired>Invalid command syntax: {e}</ansired>")
+            )
+            return
+        args = self.prompt_parser.parse_args(split_command)
         return args.func(args)
 
 

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(
                             "<b><ansired> &gt; </ansired>%s</b> <"

src/safe_cli/safe_completer_constants.py

@@ -14,7 +14,7 @@
     "change_guard": "<address>",
     "change_module_guard": "<address>",
     "change_master_copy": "<address>",
-    "change_threshold": "<address>",
+    "change_threshold": "<integer>",
     "disable_module": "<address>",
     "enable_module": "<address>",
     "execute-tx": "<safe-tx-hash>",
@@ -52,18 +52,6 @@
 
 safe_commands = list(safe_commands_arguments.keys())
 
-safe_color_arguments = {
-    "(read-only)": SAFE_ARGUMENT_COLOR,
-    "<account-private-key>": SAFE_ARGUMENT_COLOR,
-    "<address>": SAFE_ARGUMENT_COLOR,
-    "<hex-str>": SAFE_ARGUMENT_COLOR,
-    "<integer>": SAFE_ARGUMENT_COLOR,
-    "<safe-tx-hash>": SAFE_ARGUMENT_COLOR,
-    "<token-address>": SAFE_ARGUMENT_COLOR,
-    "<token-id>": SAFE_ARGUMENT_COLOR,
-    "<value-wei>": SAFE_ARGUMENT_COLOR,
-}
-
 meta = {
     "approve_hash": HTML(
         "<b>approve_hash</b> will approve a safe-tx-hash for the provided sender address. "
@@ -105,10 +93,6 @@
     "get_delegates": HTML(
         "Command <b>get_delegates</b> will return information about the current delegates."
     ),
-    "change_owner": HTML(
-        "Command <b>change_owner</b> will change an old account <u>&lt;address&gt;</u> for the new "
-        "check-summed <u>&lt;address&gt;</u> account."
-    ),
     "add_owner": HTML(
         "Command <b>add_owner</b> will add a check-summed <u>&lt;address&gt;</u> owner account."
     ),

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