Add message confirmation

src/safe_cli/operators/safe_operator.py

@@ -67,7 +67,12 @@
  get_safe_contract_address,
  get_safe_l2_contract_address,
 )
-from safe_cli.utils import choose_option_from_list, get_erc_20_list, yes_or_no_question
+from safe_cli.utils import (
+ choose_option_from_list,
+ get_erc_20_list,
+ get_input,
+ yes_or_no_question,
+)
 
 from ..contracts import safe_to_l2_migration
 from .hw_wallets.hw_wallet import HwWallet
@@ -451,7 +456,6 @@ def approve_hash(self, hash_to_approve: HexBytes, sender: str) -> bool:
 
  def sign_message(
  self,
- eip191_message: Optional[str] = None,
  eip712_message_path: Optional[str] = None,
  ) -> bool:
  if eip712_message_path:
@@ -461,8 +465,9 @@ def sign_message(
  except ValueError:
  raise ValueError
  else:
- message = eip191_message
- message_bytes = eip191_message.encode("UTF-8")
+ print_formatted_text("EIP191 message to sign:")
+ message = get_input()
+ message_bytes = message.encode("UTF-8")
 
  safe_message_hash = self.safe.get_message_hash(message_bytes)
 
@@ -486,6 +491,9 @@ def sign_message(
  HTML(f"Message was signed correctly: {safe_message_hash.hex()}")
  )
 
+ def confirm_message(self, safe_message_hash: bytes, sender: ChecksumAddress):
+ return self._require_tx_service_mode()
+
  def add_owner(self, new_owner: str, threshold: Optional[int] = None) -> bool:
  threshold = threshold if threshold is not None else self.safe_cli_info.threshold
  if new_owner in self.safe_cli_info.owners:

src/safe_cli/operators/safe_tx_service_operator.py

@@ -21,8 +21,7 @@
 from gnosis.safe.safe_signature import SafeSignature, SafeSignatureEOA
 from gnosis.safe.signatures import signature_to_bytes
 
-from safe_cli.utils import yes_or_no_question
-
+from ..utils import get_input, yes_or_no_question
 from . import SafeServiceNotAvailable
 from .exceptions import AccountNotLoadedException, NonExistingOwnerException
 from .hw_wallets.hw_wallet import HwWallet
@@ -45,7 +44,6 @@ def approve_hash(self, hash_to_approve: HexBytes, sender: str) -> bool:
 
  def sign_message(
  self,
- eip191_message: Optional[str] = None,
  eip712_message_path: Optional[str] = None,
  ) -> bool:
  if eip712_message_path:
@@ -55,7 +53,8 @@ def sign_message(
  except ValueError:
  raise ValueError
  else:
- message = eip191_message
+ print_formatted_text("EIP191 message to sign:")
+ message = get_input()
  message_hash = defunct_hash_message(text=message)
 
  safe_message_hash = self.safe.get_message_hash(message_hash)
@@ -81,7 +80,7 @@ def sign_message(
  if self.safe_tx_service.post_message(self.address, message, signatures):
  print_formatted_text(
  HTML(
- "<ansigreen>Message was correctly created on Safe Transaction Service</ansigreen>"
+ f"<ansigreen>Message with safe-message-hash {safe_message_hash} was correctly created on Safe Transaction Service</ansigreen>"
  )
  )
  return True
@@ -93,6 +92,51 @@ def sign_message(
  )
  return False
 
+ def confirm_message(self, safe_message_hash: bytes, sender: ChecksumAddress):
+ # GET message
+ try:
+ safe_message = self.safe_tx_service.get_message(safe_message_hash)
+ except SafeAPIException:
+ print_formatted_text(
+ HTML(
+ f"<ansired>Message with hash {safe_message_hash} does not exist</ansired>"
+ )
+ )
+ if not yes_or_no_question(
+ f"Message: {safe_message['message']} \n Do you want to sign the following message?:"
+ ):
+ return False
+
+ signer = self.search_account(sender)
+ if not signer:
+ print_formatted_text(
+ HTML(f"<ansired>Owner with address {sender} was not loaded</ansired>")
+ )
+
+ if isinstance(signer, LocalAccount):
+ signature = signer.signHash(safe_message_hash).signature
+ else:
+ print_formatted_text(
+ HTML(
+ "<ansired>Signing messages is not currently supported by hardware wallets</ansired>"
+ )
+ )
+ return False
+
+ try:
+ self.safe_tx_service.post_message_signature(safe_message_hash, signature)
+ except SafeAPIException as e:
+ print_formatted_text(
+ HTML(f"<ansired>Message wasn't confirmed due an error: {e}</ansired>")
+ )
+ return False
+ print_formatted_text(
+ HTML(
+ f"<ansigreen>Message with safe-message-hash {safe_message_hash} was correctly confirmed on Safe Transaction Service</ansigreen>"
+ )
+ )
+ return True
+
  def get_delegates(self):
  delegates = self.safe_tx_service.get_delegates(self.address)
  headers = ["delegate", "delegator", "label"]
@@ -156,21 +200,7 @@ def submit_signatures(self, safe_tx_hash: bytes) -> bool:
  )
  )
  else:
- owners = self.get_permitted_signers()
- for account in self.accounts:
- if account.address in owners:
- safe_tx.sign(account.key)
- # Check if there are ledger signers
- if self.hw_wallet_manager.wallets:
- selected_ledger_accounts = []
- for ledger_account in self.hw_wallet_manager.wallets:
- if ledger_account.address in owners:
- selected_ledger_accounts.append(ledger_account)
- if len(selected_ledger_accounts) > 0:
- safe_tx = self.hw_wallet_manager.sign_safe_tx(
- safe_tx, selected_ledger_accounts
- )
-
+ safe_tx = self.sign_transaction(safe_tx)
  if safe_tx.signers:
  self.safe_tx_service.post_signatures(safe_tx_hash, safe_tx.signatures)
  print_formatted_text(

src/safe_cli/prompt_parser.py

@@ -194,7 +194,11 @@ def approve_hash(args):
 
  @safe_exception
  def sign_message(args):
- safe_operator.sign_message(args.eip191_message, args.eip712_path)
+ 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):
@@ -400,10 +404,16 @@ def terminate_cli(args):
  # 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", type=str)
+ 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)

src/safe_cli/safe_completer_constants.py

@@ -36,7 +36,8 @@
  "send_erc721": "<address> <token-address> <token-id> [--safe-nonce <int>]",
  "send_ether": "<address> <value-wei> [--safe-nonce <int>]",
  "show_cli_owners": "(read-only)",
- "sign_message": "[--eip191_message <str>] [--eip712_path <file-path>]",
+ "sign_message": "[--eip191_message] [--eip712_path <file-path>]",
+ "confirm_message": "<safe-message-hash> <signer-address>",
  "sign-tx": "<safe-tx-hash>",
  "unload_cli_owners": "<address> [<address>...]",
  "update": "",

tests/test_safe_operator.py

@@ -185,7 +185,9 @@ def test_sign_message(self, mock_sign_message_lib_address):
  safe_operator.accounts.add(self.ethereum_test_account)
  safe_operator.default_sender = self.ethereum_test_account
  message_hash = safe_operator.safe.get_message_hash(bytes(message, "utf-8"))
- safe_operator.sign_message(eip191_message=message)
+ with mock.patch("builtins.input", return_value=message):
+ safe_operator.sign_message()
+
  self.assertTrue(safe_operator.safe.retrieve_is_message_signed(message_hash))
  eip712_path = "tests/mocks/mock_eip712.json"
  message = json.load(open(eip712_path, "r"))

tests/test_safe_tx_service_operator.py

@@ -9,7 +9,7 @@
 
 from gnosis.eth import EthereumClient
 from gnosis.safe import SafeTx
-from gnosis.safe.api import TransactionServiceApi
+from gnosis.safe.api import SafeAPIException, TransactionServiceApi
 
 from safe_cli.operators import SafeOperatorMode, SafeTxServiceOperator
 
@@ -551,6 +551,37 @@ def test_data_decoded_to_text(self):
  decoded_data_text,
  )
 
+ @mock.patch.object(
+ TransactionServiceApi, "post_message_signature", return_value=True
+ )
+ @mock.patch.object(TransactionServiceApi, "get_message", return_value=None)
+ def test_confirm_message(
+ self, get_message: MagicMock, post_message_signature: MagicMock
+ ):
+ safe_operator = self.setup_operator(
+ number_owners=1, mode=SafeOperatorMode.TX_SERVICE
+ )
+ # confirm_message just need the raw message to print
+ get_message.return_value = {"message": "We just need a message"}
+ safe_message_hash = HexBytes(
+ "0xfbf5d6f7faaec8fd7770cb532f87e38fb82ecb47f58474cbeb29a174f1afd829"
+ )
+ expected_signature = HexBytes(
+ "0x82484cd1f17430915f940d7c589e1687253de7f4062748b6807d29c67af52e3957b90427a9202ba4bc2104a129183387d994be2e4a8623a1f1ba2bad75d3546d1b"
+ )
+ sender = list(safe_operator.accounts)[0]
+ self.assertTrue(
+ safe_operator.confirm_message(safe_message_hash, sender.address)
+ )
+ post_message_signature.assert_called_with(safe_message_hash, expected_signature)
+
+ post_message_signature.side_effect = SafeAPIException(
+ "Error posting message signature"
+ )
+ self.assertFalse(
+ safe_operator.confirm_message(safe_message_hash, sender.address)
+ )
+
  def test_drain(self):
  # TODO Drain is a complex to mock
  pass