Add support to sign message (#341)

README.md

@@ -124,6 +124,7 @@ Operations on `tx-service` mode, requires a Safe Transaction Service working on
 - `history`: History of multisig transactions (including pending).
 - `execute-tx <safe-tx-hash>`: Execute a pending tx with enough signatures.
 - `sign-tx <safe-tx-hash>`: Sign a tx with the loaded owners for the provided `SafeTxHash`.
+- `sign_message [--eip191_message <str>] [--eip712_path <file-path>]`: sign the provided string message provided by standard input or the `EIP712` provided by file.
 - `batch-txs <safe-nonce> <safe-tx-hash> [ <safe-tx-hash> ... ]`: Batch transactions into one Multisig
 Transaction using the provided `safe-nonce`. **Any safe-tx can be used**: transactions from other Safes, transactions
 already executed, transactions pending for execution... Only limitation is that

requirements.txt

@@ -5,7 +5,7 @@ packaging>=23.1
 prompt_toolkit==3.0.43
 pygments==2.17.2
 requests==2.31.0
-safe-eth-py==6.0.0b13
+safe-eth-py==6.0.0b14
 tabulate==0.9.0
 trezor==0.13.8
 web3==6.14.0

safe_cli/operators/safe_operator.py

@@ -1,7 +1,8 @@
 import dataclasses
+import json
 import os
 from functools import cached_property, wraps
-from typing import List, Optional, Sequence, Set
+from typing import List, Optional, Sequence, Set, Tuple
 
 from ens import ENS
 from eth_account import Account
@@ -27,7 +28,9 @@
     get_erc20_contract,
     get_erc721_contract,
     get_safe_V1_1_1_contract,
+    get_sign_message_lib_contract,
 )
+from gnosis.eth.eip712 import eip712_encode
 from gnosis.eth.utils import get_empty_tx_params
 from gnosis.safe import InvalidInternalTx, Safe, SafeOperation, SafeTx
 from gnosis.safe.api import TransactionServiceApi
@@ -60,12 +63,14 @@
 )
 from safe_cli.safe_addresses import (
     get_default_fallback_handler_address,
+    get_last_sign_message_lib_address,
     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 ..contracts import safe_to_l2_migration
+from .hw_wallets.hw_wallet import HwWallet
 from .hw_wallets.hw_wallet_manager import HwWalletType, get_hw_wallet_manager
 
 
@@ -443,6 +448,43 @@ def approve_hash(self, hash_to_approve: HexBytes, sender: str) -> bool:
                     )
                     return False
 
+    def sign_message(
+        self,
+        eip191_message: Optional[str] = None,
+        eip712_message_path: Optional[str] = None,
+    ) -> bool:
+        if eip712_message_path:
+            try:
+                message = json.load(open(eip712_message_path, "r"))
+                message_bytes = b"".join(eip712_encode(message))
+            except ValueError:
+                raise ValueError
+        else:
+            message = eip191_message
+            message_bytes = eip191_message.encode("UTF-8")
+
+        safe_message_hash = self.safe.get_message_hash(message_bytes)
+
+        sign_message_lib_address = get_last_sign_message_lib_address(
+            self.ethereum_client
+        )
+        contract = get_sign_message_lib_contract(self.ethereum_client.w3, self.address)
+        sign_message_data = HexBytes(
+            contract.functions.signMessage(message_bytes).build_transaction(
+                get_empty_tx_params(),
+            )["data"]
+        )
+        print_formatted_text(HTML(f"Signing message: \n {message}"))
+        if self.prepare_and_execute_safe_transaction(
+            sign_message_lib_address,
+            0,
+            sign_message_data,
+            operation=SafeOperation.DELEGATE_CALL,
+        ):
+            print_formatted_text(
+                HTML(f"Message was signed correctly: {safe_message_hash.hex()}")
+            )
+
     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:
@@ -991,40 +1033,46 @@ def batch_safe_txs(
         else:
             return safe_tx
 
-    # TODO Set sender so we can save gas in that signature
-    def sign_transaction(self, safe_tx: SafeTx) -> SafeTx:
+    def get_signers(self) -> Tuple[List[LocalAccount], List[HwWallet]]:
+        """
+
+        :return: Tuple with eoa signers and hw_wallet signers
+        """
         permitted_signers = self.get_permitted_signers()
         threshold = self.safe_cli_info.threshold
-        selected_accounts: List[
+        eoa_signers: List[
             Account
         ] = []  # Some accounts that are not an owner can be loaded
         for account in self.accounts:
             if account.address in permitted_signers:
-                selected_accounts.append(account)
+                eoa_signers.append(account)
                 threshold -= 1
                 if threshold == 0:
                     break
         # If still pending required signatures continue with ledger owners
-        selected_ledger_accounts = []
+        hw_wallet_signers = []
         if threshold > 0 and self.hw_wallet_manager.wallets:
-            for ledger_account in self.hw_wallet_manager.wallets:
-                if ledger_account.address in permitted_signers:
-                    selected_ledger_accounts.append(ledger_account)
+            for hw_wallet in self.hw_wallet_manager.wallets:
+                if hw_wallet.address in permitted_signers:
+                    hw_wallet_signers.append(hw_wallet)
                     threshold -= 1
                     if threshold == 0:
                         break
 
         if self.require_all_signatures and threshold > 0:
             raise NotEnoughSignatures(threshold)
 
-        for selected_account in selected_accounts:
+        return (eoa_signers, hw_wallet_signers)
+
+    # TODO Set sender so we can save gas in that signature
+    def sign_transaction(self, safe_tx: SafeTx) -> SafeTx:
+        eoa_signers, hw_wallets_signers = self.get_signers()
+        for selected_account in eoa_signers:
             safe_tx.sign(selected_account.key)
 
         # Sign with ledger
-        if len(selected_ledger_accounts) > 0:
-            safe_tx = self.hw_wallet_manager.sign_eip712(
-                safe_tx, selected_ledger_accounts
-            )
+        if len(hw_wallets_signers):
+            safe_tx = self.hw_wallet_manager.sign_eip712(safe_tx, hw_wallets_signers)
 
         return safe_tx
 

safe_cli/operators/safe_tx_service_operator.py

@@ -1,15 +1,20 @@
-from typing import Any, Dict, Optional, Sequence, Set
+import json
+from typing import Any, Dict, List, Optional, Sequence, Set
 
 from colorama import Fore, Style
+from eth_account.messages import defunct_hash_message
 from eth_typing import ChecksumAddress
 from hexbytes import HexBytes
 from prompt_toolkit import HTML, print_formatted_text
 from tabulate import tabulate
 
 from gnosis.eth.contracts import get_erc20_contract
+from gnosis.eth.eip712 import eip712_encode_hash
 from gnosis.safe import SafeOperation, SafeTx
 from gnosis.safe.api import SafeAPIException
 from gnosis.safe.multi_send import MultiSend, MultiSendOperation, MultiSendTx
+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
 
@@ -32,6 +37,51 @@ def __init__(self, address: str, node_url: str):
     def approve_hash(self, hash_to_approve: HexBytes, sender: str) -> bool:
         raise NotImplementedError("Not supported when using tx service")
 
+    def sign_message(
+        self,
+        eip191_message: Optional[str] = None,
+        eip712_message_path: Optional[str] = None,
+    ) -> bool:
+        if eip712_message_path:
+            try:
+                message = json.load(open(eip712_message_path, "r"))
+                message_hash = eip712_encode_hash(message)
+            except ValueError:
+                raise ValueError
+        else:
+            message = eip191_message
+            message_hash = defunct_hash_message(text=message)
+
+        safe_message_hash = self.safe.get_message_hash(message_hash)
+        eoa_signers, hw_wallet_signers = self.get_signers()
+        safe_signatures: List[SafeSignature] = []
+        for eoa_signer in eoa_signers:
+            signature_dict = eoa_signer.signHash(safe_message_hash)
+            signature = signature_to_bytes(
+                signature_dict["v"], signature_dict["r"], signature_dict["s"]
+            )
+            safe_signatures.append(SafeSignatureEOA(signature, safe_message_hash))
+
+        signatures = SafeSignature.export_signatures(safe_signatures)
+
+        if len(hw_wallet_signers):
+            raise NotImplementedError("SignHash by hardware wallet is not implemented")
+
+        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>"
+                )
+            )
+            return True
+        else:
+            print_formatted_text(
+                HTML(
+                    "<ansired>Something went wrong creating message on Safe Transaction Service</ansired>"
+                )
+            )
+            return False
+
     def get_delegates(self):
         delegates = self.safe_tx_service.get_delegates(self.address)
         headers = ["delegate", "delegator", "label"]

safe_cli/prompt_parser.py

@@ -189,6 +189,10 @@ def unload_cli_owners(args):
     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.eip191_message, args.eip712_path)
+
     @safe_exception
     def add_owner(args):
         safe_operator.add_owner(args.address, threshold=args.threshold)
@@ -368,6 +372,13 @@ def remove_delegate(args):
     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", type=str)
+    group.add_argument("--eip712_path", type=str)
+    parser_sign_message.set_defaults(func=sign_message)
+
     # Add owner
     parser_add_owner = subparsers.add_parser("add_owner")
     parser_add_owner.add_argument("address", type=check_ethereum_address)

safe_cli/safe_addresses.py

@@ -99,3 +99,17 @@ def get_last_multisend_call_only_address(
             "0xf220D3b4DFb23C4ade8C88E526C1353AbAcbC38F",  # v1.3.0 zkSync
         ],
     )
+
+
+def get_last_sign_message_lib_address(
+    ethereum_client: EthereumClient,
+) -> ChecksumAddress:
+    return _get_valid_contract(
+        ethereum_client,
+        [
+            "0xd53cd0aB83D845Ac265BE939c57F53AD838012c9",  # v1.4.1
+            "0xA65387F16B013cf2Af4605Ad8aA5ec25a2cbA3a2",  # v1.3.0
+            "0x98FFBBF51bb33A056B08ddf711f289936AafF717",  # v1.3.0
+            "0x357147caf9C0cCa67DfA0CF5369318d8193c8407",  # v1.3.0 zkSync
+        ],
+    )

safe_cli/safe_completer_constants.py

@@ -35,6 +35,7 @@
     "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-tx": "<safe-tx-hash>",
     "unload_cli_owners": "<address> [<address>...]",
     "update": "",
@@ -81,6 +82,9 @@
     "sign-tx": HTML(
         "<b>sign-tx</b> will sign the provided safeTxHash using the owners loaded on the CLI"
     ),
+    "sign_message": HTML(
+        "<b>sign_message</b> sign the provided string message provided by standard input or the EIP712 provided by file"
+    ),
     "info": HTML(
         "<b>info</b> will return all the information available for a Safe, with Gnosis Tx Service and "
         "Etherscan links if the network is supported"

safe_cli/safe_lexer.py

@@ -20,6 +20,7 @@ class SafeLexer(BashLexer):
         "load_cli_owners",
         "unload_cli_owners",
         "approve_hash",
+        "sign_message",
         "add_owner",
         "change_threshold",
         "change_fallback_handler",

setup.py

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