Add option to load a Safe from owner

README.md

@@ -33,11 +33,25 @@ to run the actual **safe-cli**
 pip3 install -U safe-cli
 ```
 
-## Using
+## Usage
+
+```bash
+safe-cli [-h] [--history] [--is-owner] address node_url
+
+positional arguments:
+  address     The address of the Safe, or an owner address if --is-owner is specified.
+  node_url    Ethereum node url
+
+options:
+  -h, --help  show this help message and exit
+  --history   Enable history. By default it's disabled due to security reasons
+  ----get-safes-from-owner  Indicates that address is an owner (Safe Transaction Service is required for this feature)
+```
+### Quick Load Command:
+To load a Safe, use the following command:
 ```bash
 safe-cli <checksummed_safe_address> <ethereum_node_url>
 ```
-
 Then you should be on the prompt and see information about the Safe, like the owners, version, etc.
 Next step would be loading some owners for the Safe. At least `threshold` owners need to be loaded to do operations
 on the Safe and at least one of them should have funds for sending transactions.

requirements.txt

@@ -5,6 +5,6 @@ packaging>=23.1
 prompt_toolkit==3.0.40
 pygments==2.16.1
 requests==2.31.0
-safe-eth-py==6.0.0b6
+safe-eth-py==6.0.0b8
 tabulate==0.9.0
 web3==6.11.3

safe_cli/main.py

@@ -20,6 +20,7 @@
 from safe_cli.prompt_parser import PromptParser
 from safe_cli.safe_completer import SafeCompleter
 from safe_cli.safe_lexer import SafeLexer
+from safe_cli.utils import get_safe_from_owner
 
 from .version import version
 
@@ -118,11 +119,11 @@ def loop(self):
                 pass
 
 
-def build_safe_cli():
+def build_safe_cli() -> Optional[SafeCli]:
     parser = argparse.ArgumentParser()
     parser.add_argument(
-        "safe_address",
-        help="Address of Safe to use",
+        "address",
+        help="The address of the Safe, or an owner address if --get-safes-from-owner is specified.",
         type=check_ethereum_address,
     )
     parser.add_argument("node_url", help="Ethereum node url")
@@ -132,10 +133,21 @@ def build_safe_cli():
         help="Enable history. By default it's disabled due to security reasons",
         default=False,
     )
+    parser.add_argument(
+        "--get-safes-from-owner",
+        action="store_true",
+        help="Indicates that address is an owner (Safe Transaction Service is required for this feature)",
+        default=False,
+    )
 
     args = parser.parse_args()
-
-    return SafeCli(args.safe_address, args.node_url, args.history)
+    if args.get_safes_from_owner:
+        if (
+            safe_address := get_safe_from_owner(args.address, args.node_url)
+        ) is not None:
+            return SafeCli(safe_address, args.node_url, args.history)
+    else:
+        return SafeCli(args.address, args.node_url, args.history)
 
 
 def main(*args, **kwargs):

safe_cli/operators/safe_operator.py

@@ -63,7 +63,7 @@
     get_safe_contract_address,
     get_safe_l2_contract_address,
 )
-from safe_cli.utils import choose_option_question, get_erc_20_list, yes_or_no_question
+from safe_cli.utils import choose_option_from_list, get_erc_20_list, yes_or_no_question
 
 from ..contracts import safe_to_l2_migration
 
@@ -294,12 +294,8 @@ def load_ledger_cli_owners(
             if len(ledger_accounts) == 0:
                 return None
 
-            for option, ledger_account in enumerate(ledger_accounts):
-                address, _ = ledger_account
-                print_formatted_text(HTML(f"{option} - <b>{address}</b> "))
-
-            option = choose_option_question(
-                "Select the owner address", len(ledger_accounts)
+            option = choose_option_from_list(
+                "Select the owner address", ledger_accounts
             )
             if option is None:
                 return None

safe_cli/utils.py

@@ -1,20 +1,29 @@
 import os
-from typing import Optional
+from typing import List, Optional
 
+from eth_typing import ChecksumAddress
 from prompt_toolkit import HTML, print_formatted_text
 
 from gnosis.eth import EthereumClient
+from gnosis.safe.api import TransactionServiceApi
 
 
-# Return a list of address of ERC20 tokens related with the safe_address
-# block_step is the number of blocks retrieved for each get until get all blocks between from_block until to_block
 def get_erc_20_list(
     ethereum_client: EthereumClient,
     safe_address: str,
     from_block: int,
     to_block: int,
     block_step: int = 500000,
 ) -> list:
+    """
+
+    :param ethereum_client:
+    :param safe_address:
+    :param from_block:
+    :param to_block:
+    :param block_step: is the number of blocks retrieved for each get until get all blocks between from_block until to_block
+    :return: a list of address of ERC20 tokens related with the safe_address
+    """
     addresses = set()
     for i in range(from_block, to_block + 1, block_step):
         events = ethereum_client.erc20.get_total_transfer_history(
@@ -46,11 +55,14 @@ def yes_or_no_question(question: str, default_no: bool = True) -> bool:
         return False if default_no else True
 
 
-def choose_option_question(
-    question: str, number_options: int, default_option: int = 0
+def choose_option_from_list(
+    question: str, options: List, default_option: int = 0
 ) -> Optional[int]:
     if "PYTEST_CURRENT_TEST" in os.environ:
         return default_option  # Ignore confirmations when running tests
+    number_options = len(options)
+    for number_option, option in enumerate(options):
+        print_formatted_text(HTML(f"{number_option} - <b>{option}</b> "))
     choices = f" [0-{number_options-1}] default {default_option}: "
     reply = str(get_input(question + choices)).lower().strip() or str(default_option)
     try:
@@ -61,8 +73,30 @@ def choose_option_question(
 
     if option not in range(0, number_options):
         print_formatted_text(
-            HTML(f"<ansired> {option} is not between [0-{number_options}}} </ansired>")
+            HTML(f"<ansired> {option} is not between [0-{number_options-1}] </ansired>")
         )
         return None
 
     return option
+
+
+def get_safe_from_owner(
+    owner: ChecksumAddress, node_url: str
+) -> Optional[ChecksumAddress]:
+    """
+    Show a list of Safe to chose between them and return the selected one.
+    :param owner:
+    :param node_url:
+    :return: Safe address of a selected Safe
+    """
+    ethereum_client = EthereumClient(node_url)
+    safe_tx_service = TransactionServiceApi.from_ethereum_client(ethereum_client)
+    safes = safe_tx_service.get_safes_for_owner(owner)
+    if safes:
+        option = choose_option_from_list(
+            "Select the Safe to initialize the safe-cli", safes
+        )
+        if option is not None:
+            return safes[option]
+    else:
+        raise ValueError(f"No safe was found for the specified owner {owner}")

setup.py

@@ -24,7 +24,7 @@
         "prompt_toolkit>=3",
         "pygments>=2",
         "requests>=2",
-        "safe-eth-py==6.0.0b5",
+        "safe-eth-py==6.0.0b8",
         "tabulate>=0.8",
     ],
     extras_require={"ledger": ["ledgereth==0.9.1"]},

tests/test_entrypoint.py

@@ -7,7 +7,9 @@
 from prompt_toolkit import HTML
 
 from gnosis.eth.constants import NULL_ADDRESS
+from gnosis.eth.ethereum_client import EthereumClient
 from gnosis.safe import Safe
+from gnosis.safe.api import TransactionServiceApi
 from gnosis.safe.safe import SafeInfo
 
 from safe_cli.main import SafeCli, build_safe_cli
@@ -22,9 +24,20 @@ class SafeCliEntrypointTestCase(SafeCliTestCaseMixin, unittest.TestCase):
     @mock.patch("argparse.ArgumentParser.parse_args")
     def build_test_safe_cli(self, mock_parse_args: MagicMock):
         mock_parse_args.return_value = argparse.Namespace(
-            safe_address=self.random_safe_address,
+            address=self.random_safe_address,
             node_url=self.ethereum_node_url,
             history=True,
+            get_safes_from_owner=False,
+        )
+        return build_safe_cli()
+
+    @mock.patch("argparse.ArgumentParser.parse_args")
+    def build_test_safe_cli_for_owner(self, mock_parse_args: MagicMock):
+        mock_parse_args.return_value = argparse.Namespace(
+            address=self.random_safe_address,
+            node_url=self.ethereum_node_url,
+            history=True,
+            get_safes_from_owner=True,
         )
         return build_safe_cli()
 
@@ -48,6 +61,37 @@ def test_build_safe_cli(self, retrieve_all_info_mock: MagicMock):
         self.assertIsInstance(safe_cli.get_prompt_text(), HTML)
         self.assertIsInstance(safe_cli.get_bottom_toolbar(), HTML)
 
+    @mock.patch.object(EthereumClient, "get_chain_id", return_value=5)
+    @mock.patch.object(TransactionServiceApi, "get_safes_for_owner")
+    @mock.patch.object(Safe, "retrieve_all_info")
+    def test_build_safe_cli_for_owner(
+        self,
+        retrieve_all_info_mock: MagicMock,
+        get_safes_for_owner_mock: MagicMock,
+        get_chain_id_mock: MagicMock,
+    ):
+        retrieve_all_info_mock.return_value = SafeInfo(
+            self.random_safe_address,
+            "0xfd0732Dc9E303f09fCEf3a7388Ad10A83459Ec99",
+            NULL_ADDRESS,
+            "0x29fcB43b46531BcA003ddC8FCB67FFE91900C762",
+            [],
+            0,
+            [Account.create().address],
+            1,
+            "1.4.1",
+        )
+        get_safes_for_owner_mock.return_value = []
+        with self.assertRaises(ValueError):
+            self.build_test_safe_cli_for_owner()
+        get_safes_for_owner_mock.return_value = [self.random_safe_address]
+        safe_cli = self.build_test_safe_cli_for_owner()
+        self.assertIsNotNone(safe_cli)
+        with mock.patch.object(SafeOperator, "is_version_updated", return_value=True):
+            self.assertIsNone(safe_cli.print_startup_info())
+        self.assertIsInstance(safe_cli.get_prompt_text(), HTML)
+        self.assertIsInstance(safe_cli.get_bottom_toolbar(), HTML)
+
     def test_parse_operator_mode(self):
         safe_cli = self.build_test_safe_cli()
         self.assertIsNone(safe_cli.parse_operator_mode("tx-service"))

tests/test_utils.py

@@ -3,7 +3,9 @@
 from unittest import mock
 from unittest.mock import MagicMock
 
-from safe_cli.utils import choose_option_question, yes_or_no_question
+from eth_account import Account
+
+from safe_cli.utils import choose_option_from_list, yes_or_no_question
 
 
 class TestUtils(unittest.TestCase):
@@ -35,19 +37,20 @@ def test_yes_or_no_question(self, input_mock: MagicMock):
         os.environ["PYTEST_CURRENT_TEST"] = pytest_current_test
 
     @mock.patch("safe_cli.utils.get_input")
-    def test_choose_option_question(self, input_mock: MagicMock):
+    def test_choose_option_from_list(self, input_mock: MagicMock):
         pytest_current_test = os.environ.pop("PYTEST_CURRENT_TEST")
-
+        address = Account.create().address
+        options = [address for i in range(0, 5)]
         input_mock.return_value = ""
-        self.assertEqual(choose_option_question("", 1), 0)
+        self.assertEqual(choose_option_from_list("", options), 0)
         input_mock.return_value = ""
-        self.assertEqual(choose_option_question("", 5, 4), 4)
+        self.assertEqual(choose_option_from_list("", options, 4), 4)
         input_mock.return_value = "m"
-        self.assertIsNone(choose_option_question("", 1))
+        self.assertIsNone(choose_option_from_list("", options))
         input_mock.return_value = "10"
-        self.assertIsNone(choose_option_question("", 1))
+        self.assertIsNone(choose_option_from_list("", options))
         input_mock.return_value = "1"
-        self.assertEqual(choose_option_question("", 2), 1)
+        self.assertEqual(choose_option_from_list("", options), 1)
 
         os.environ["PYTEST_CURRENT_TEST"] = pytest_current_test