diff --git a/safe_cli/operators/safe_operator.py b/safe_cli/operators/safe_operator.py
index 2c470361..008c1723 100644
--- a/safe_cli/operators/safe_operator.py
+++ b/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,25 +1021,24 @@ 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
@@ -1004,14 +1046,17 @@ def sign_transaction(self, safe_tx: SafeTx) -> SafeTx:
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
diff --git a/safe_cli/operators/safe_tx_service_operator.py b/safe_cli/operators/safe_tx_service_operator.py
index 1fad9454..529e8698 100644
--- a/safe_cli/operators/safe_tx_service_operator.py
+++ b/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(
+ "Message was correctly created on Safe Transaction Service"
+ )
+ )
+ return True
+ else:
+ print_formatted_text(
+ HTML(
+ "Something went wrong creating message on Safe Transaction Service"
+ )
+ )
+ return False
+
def get_delegates(self):
delegates = self.safe_tx_service.get_delegates(self.address)
headers = ["delegate", "delegator", "label"]
diff --git a/safe_cli/prompt_parser.py b/safe_cli/prompt_parser.py
index 2ecdd43d..a3d87607 100644
--- a/safe_cli/prompt_parser.py
+++ b/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)
diff --git a/safe_cli/safe_addresses.py b/safe_cli/safe_addresses.py
index 249e0ebd..64267f5b 100644
--- a/safe_cli/safe_addresses.py
+++ b/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
+ ],
+ )
diff --git a/safe_cli/safe_lexer.py b/safe_cli/safe_lexer.py
index bdd743cf..1020c81b 100644
--- a/safe_cli/safe_lexer.py
+++ b/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",