Add sign_message

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_V1_3_0_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,6 +63,7 @@
 )
 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,
 )
@@ -430,6 +434,45 @@ def approve_hash(self, hash_to_approve: HexBytes, sender: str) -> bool:
                     )
                     return False
 
+    def sign_message(
+        self,
+        eip151_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 = eip151_message
+            message_bytes = eip151_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_V1_3_0_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:
@@ -978,40 +1021,42 @@ 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]:
         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)
+                    hw_wallet_signers.append(ledger_account)
                     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) > 0:
+            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,19 @@
+import json
 from typing import Any, Dict, 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.signatures import signature_to_bytes
 
 from safe_cli.utils import yes_or_no_question
 
@@ -32,6 +36,58 @@ 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,
+        eip151_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 = eip151_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()
+        signers = []
+        signatures = b""
+        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"]
+            )
+            signers.append(eoa_signer.address)
+            signer_pos = sorted(signers, key=lambda x: int(x, 16)).index(
+                eoa_signer.address
+            )
+            signatures = (
+                signatures[: 65 * signer_pos]
+                + signature
+                + signatures[65 * signer_pos :]
+            )
+
+        if len(hw_wallet_signers) > 0:
+            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.eip151_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("--eip151_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_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",