Add hw wallet sign message (#420)

* Add support for sign_message HwWallet

* Add Hw wallet support on SafeOperator tx-service mode

src/safe_cli/operators/hw_wallets/hw_wallet.py

@@ -58,6 +58,15 @@ def get_signed_raw_transaction(
         :return: raw transaction signed
         """
 
+    @abstractmethod
+    def sign_message(self, message: bytes) -> bytes:
+        """
+        Call sign message of hw wallet
+
+        :param message:
+        :return: bytes signature
+        """
+
     def __str__(self):
         return f"{self.__class__.__name__} device with address {self.address}"
 

src/safe_cli/operators/hw_wallets/hw_wallet_manager.py

@@ -10,7 +10,11 @@
 from gnosis.eth import TxSpeed
 from gnosis.eth.eip712 import eip712_encode, eip712_encode_hash
 from gnosis.safe import SafeTx
-from gnosis.safe.safe_signature import SafeSignature, SafeSignatureEOA
+from gnosis.safe.safe_signature import (
+    SafeSignature,
+    SafeSignatureEOA,
+    SafeSignatureEthSign,
+)
 
 from .hw_wallet import HwWallet
 
@@ -213,3 +217,26 @@ def execute_safe_tx(
         # so signatures are not valid anymore
         safe_tx.signatures = b""
         return safe_tx.tx_hash, safe_tx.tx
+
+    def sign_message(
+        self, message: bytes, wallets: List[HwWallet]
+    ) -> List[SafeSignature]:
+        """
+        Sign a message for all the provided wallets
+
+        :param message:
+        :param wallets:
+        :return:
+        """
+        signatures: List[SafeSignature] = []
+        for wallet in wallets:
+            print_formatted_text(
+                HTML(
+                    f"<ansired>Make sure before signing in your {wallet} that the message is correct</ansired>"
+                )
+            )
+            signatures.append(
+                SafeSignatureEthSign(wallet.sign_message(message), message)
+            )
+
+        return signatures

src/safe_cli/operators/hw_wallets/ledger_wallet.py

@@ -3,7 +3,7 @@
 from eth_typing import ChecksumAddress
 from hexbytes import HexBytes
 from ledgerblue.Dongle import Dongle
-from ledgereth import create_transaction, sign_typed_data_draft
+from ledgereth import create_transaction, sign_message, sign_typed_data_draft
 from ledgereth.accounts import get_account_by_path
 from ledgereth.comms import init_dongle
 from web3.types import TxParams
@@ -75,3 +75,15 @@ def get_signed_raw_transaction(
             dongle=self.dongle,
         )
         return HexBytes(signed_transaction.raw_transaction())
+
+    @raise_ledger_exception_as_hw_wallet_exception
+    def sign_message(self, message: bytes) -> bytes:
+        """
+        Call sign message of Ledger wallet
+
+        :param message:
+        :return: bytes signature
+        """
+        signed = sign_message(message, self.derivation_path, self.dongle)
+        # V field must be greater than 30 for signed messages. https://github.com/safe-global/safe-smart-account/blob/main/contracts/Safe.sol#L309
+        return signature_to_bytes(signed.v + 4, signed.r, signed.s)

src/safe_cli/operators/hw_wallets/trezor_wallet.py

@@ -7,13 +7,16 @@
 from trezorlib.client import TrezorClient, get_default_client
 from trezorlib.ethereum import (
     get_address,
+    sign_message,
     sign_tx,
     sign_tx_eip1559,
     sign_typed_data_hash,
 )
 from trezorlib.ui import ClickUI
 from web3.types import TxParams
 
+from gnosis.safe.signatures import signature_split, signature_to_bytes
+
 from .hw_wallet import HwWallet
 from .trezor_exceptions import raise_trezor_exception_as_hw_wallet_exception
 
@@ -34,15 +37,15 @@ def get_trezor_client() -> TrezorClient:
 class TrezorWallet(HwWallet):
     def __init__(self, derivation_path: str):
         self.client: TrezorClient = get_trezor_client()
+        self.address_n = tools.parse_path(derivation_path)
         super().__init__(derivation_path)
 
     @raise_trezor_exception_as_hw_wallet_exception
     def get_address(self) -> ChecksumAddress:
         """
         :return: public address for derivation_path
         """
-        address_n = tools.parse_path(self.derivation_path)
-        return get_address(client=self.client, n=address_n)
+        return get_address(client=self.client, n=self.address_n)
 
     @raise_trezor_exception_as_hw_wallet_exception
     def sign_typed_hash(self, domain_hash: bytes, message_hash: bytes) -> bytes:
@@ -52,9 +55,11 @@ def sign_typed_hash(self, domain_hash: bytes, message_hash: bytes) -> bytes:
         :param message_hash:
         :return: signature bytes
         """
-        address_n = tools.parse_path(self.derivation_path)
         signed = sign_typed_data_hash(
-            self.client, n=address_n, domain_hash=domain_hash, message_hash=message_hash
+            self.client,
+            n=self.address_n,
+            domain_hash=domain_hash,
+            message_hash=message_hash,
         )
         return signed.signature
 
@@ -68,12 +73,11 @@ def get_signed_raw_transaction(
         :param tx_parameters:
         :return: raw transaction signed
         """
-        address_n = tools.parse_path(self.derivation_path)
         if tx_parameters.get("maxPriorityFeePerGas"):
             # EIP1559
             v, r, s = sign_tx_eip1559(
                 self.client,
-                n=address_n,
+                n=self.address_n,
                 nonce=tx_parameters["nonce"],
                 gas_limit=tx_parameters["gas"],
                 to=tx_parameters["to"],
@@ -107,7 +111,7 @@ def get_signed_raw_transaction(
             # Legacy transaction
             v, r, s = sign_tx(
                 self.client,
-                n=address_n,
+                n=self.address_n,
                 nonce=tx_parameters["nonce"],
                 gas_price=tx_parameters["gasPrice"],
                 gas_limit=tx_parameters["gas"],
@@ -132,3 +136,16 @@ def get_signed_raw_transaction(
             ).hex()
 
         return HexBytes(encoded_transaction)
+
+    @raise_trezor_exception_as_hw_wallet_exception
+    def sign_message(self, message: bytes) -> bytes:
+        """
+        Call sign message of Trezor wallet
+
+        :param message:
+        :return: bytes signature
+        """
+        signed = sign_message(self.client, self.address_n, message)
+        # V field must be greater than 30 for signed messages. https://github.com/safe-global/safe-smart-account/blob/main/contracts/Safe.sol#L309
+        v, r, s = signature_split(signed.signature)
+        return signature_to_bytes(v + 4, r, s)

src/safe_cli/operators/safe_operator.py

@@ -1065,6 +1065,7 @@ def batch_safe_txs(
 
     def get_signers(self) -> Tuple[List[LocalAccount], List[HwWallet]]:
         """
+        Get the signers necessary to sign a transaction, raise an exception if was not uploaded enough signers.
 
         :return: Tuple with eoa signers and hw_wallet signers
         """

src/safe_cli/operators/safe_tx_service_operator.py

@@ -1,6 +1,6 @@
 import json
 from itertools import chain
-from typing import Any, Dict, List, Optional, Sequence, Set, Union
+from typing import Any, Dict, Optional, Sequence, Set, Union
 
 from colorama import Fore, Style
 from eth_account.messages import defunct_hash_message
@@ -18,7 +18,7 @@
     get_remove_transaction_message,
 )
 from gnosis.safe.multi_send import MultiSend, MultiSendOperation, MultiSendTx
-from gnosis.safe.safe_signature import SafeSignature, SafeSignatureEOA
+from gnosis.safe.safe_signature import SafeSignature
 from gnosis.safe.signatures import signature_to_bytes
 
 from ..utils import get_input, yes_or_no_question
@@ -59,25 +59,26 @@ def sign_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)
+        # Safe transaction service just accept one signer to create a message
+        signature = b""
+        if eoa_signers:
+            signature_dict = eoa_signers[0].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 hw_wallet_signers:
-            print_formatted_text(
-                HTML(
-                    "<ansired>Signing messages is not currently supported by hardware wallets</ansired>"
+        elif hw_wallet_signers:
+            signature = SafeSignature.export_signatures(
+                self.hw_wallet_manager.sign_message(
+                    safe_message_hash, [hw_wallet_signers[0]]
                 )
             )
-            return False
+        else:
+            print_formatted_text(
+                HTML("<ansired>At least one owner must be loaded</ansired>")
+            )
 
-        if self.safe_tx_service.post_message(self.address, message, signatures):
+        if self.safe_tx_service.post_message(self.address, message, signature):
             print_formatted_text(
                 HTML(
                     f"<ansigreen>Message  with safe-message-hash {safe_message_hash.hex()} was correctly created on Safe Transaction Service</ansigreen>"
@@ -116,12 +117,9 @@ def confirm_message(self, safe_message_hash: bytes, sender: ChecksumAddress):
         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>"
-                )
+            signature = SafeSignature.export_signatures(
+                self.hw_wallet_manager.sign_message(safe_message_hash, [signer])
             )
-            return False
 
         try:
             self.safe_tx_service.post_message_signature(safe_message_hash, signature)

tests/test_ledger_wallet.py

@@ -12,7 +12,7 @@
     LedgerLocked,
     LedgerNotFound,
 )
-from ledgereth.objects import LedgerAccount, SignedTransaction
+from ledgereth.objects import LedgerAccount, SignedMessage, SignedTransaction
 
 from gnosis.eth.eip712 import eip712_encode
 from gnosis.safe import SafeTx
@@ -244,3 +244,44 @@ def test_get_signed_raw_transaction(
             safe_tx.tx, safe_tx.ethereum_client.get_chain_id()
         )  # return raw signed transaction
         self.assertEqual(signed_fields.rawTransaction, HexBytes(raw_signed_tx))
+
+    @mock.patch(
+        "safe_cli.operators.hw_wallets.ledger_wallet.sign_message",
+        autospec=True,
+        return_value=Dongle(),
+    )
+    @mock.patch(
+        "safe_cli.operators.hw_wallets.ledger_wallet.get_account_by_path",
+        autospec=True,
+    )
+    @mock.patch(
+        "safe_cli.operators.hw_wallets.ledger_wallet.init_dongle",
+        autospec=True,
+        return_value=Dongle(),
+    )
+    def test_sign_message(
+        self,
+        mock_init_dongle: MagicMock,
+        mock_get_account_by_path: MagicMock,
+        mock_ledger_sign_message: MagicMock,
+    ):
+        owner = Account.create()
+        derivation_path = "44'/60'/0'/0"
+        mock_get_account_by_path.return_value = LedgerAccount(
+            derivation_path, owner.address
+        )
+        ledger_wallet = LedgerWallet(derivation_path)
+        expected_signature = HexBytes(
+            "0xbc941061f14cfbf055332537a282834dd66f4e944b3b4608aea062e203c7fd505b5e74c0a984d62ec088cd1d82c00d7c6f5f71076d6bc536fcc02be463d9128820"
+        )
+        safe_message_hash = HexBytes(
+            "0x08a1b4472ed4f7f71ac2a8ec9978da670476b3675720b8c4e11fe71a75b56f38"
+        )
+        v, r, s = signature_split(expected_signature)
+        # Checking that v is incremented by sign_message by 4
+        mock_ledger_sign_message.return_value = SignedMessage(
+            safe_message_hash, v - 4, r, s
+        )
+
+        signature = ledger_wallet.sign_message(safe_message_hash)
+        self.assertEqual(signature, expected_signature)

tests/test_trezor_wallet.py

@@ -13,6 +13,7 @@
 
 from gnosis.eth.eip712 import eip712_encode
 from gnosis.safe import SafeTx
+from gnosis.safe.signatures import signature_split, signature_to_bytes
 from gnosis.safe.tests.safe_test_case import SafeTestCaseMixin
 
 from safe_cli.operators.exceptions import HardwareWalletException
@@ -251,3 +252,43 @@ def test_get_signed_raw_transaction(
         )  # return raw signed transaction
         mock_sign_tx_eip1559.assert_called_once()
         self.assertEqual(signed_fields.rawTransaction, HexBytes(raw_signed_tx))
+
+    @mock.patch(
+        "safe_cli.operators.hw_wallets.trezor_wallet.sign_message",
+        autospec=True,
+    )
+    @mock.patch(
+        "safe_cli.operators.hw_wallets.trezor_wallet.get_address",
+        autospec=True,
+    )
+    @mock.patch(
+        "safe_cli.operators.hw_wallets.trezor_wallet.get_trezor_client",
+        autospec=True,
+    )
+    def test_get_sign_message(
+        self,
+        mock_trezor_client: MagicMock,
+        mock_get_address: MagicMock,
+        mock_sign_message: MagicMock,
+    ):
+        owner = Account.create()
+        transport_mock = MagicMock(auto_spec=True)
+        mock_trezor_client.return_value = TrezorClient(
+            transport_mock, ui=ClickUI(), _init_device=False
+        )
+        mock_trezor_client.return_value.is_outdated = MagicMock(return_value=False)
+        mock_get_address.return_value = owner.address
+        trezor_wallet = TrezorWallet("44'/60'/0'/0")
+        expected_signature = HexBytes(
+            "0xbc941061f14cfbf055332537a282834dd66f4e944b3b4608aea062e203c7fd505b5e74c0a984d62ec088cd1d82c00d7c6f5f71076d6bc536fcc02be463d9128820"
+        )
+        safe_message_hash = HexBytes(
+            "0x08a1b4472ed4f7f71ac2a8ec9978da670476b3675720b8c4e11fe71a75b56f38"
+        )
+        v, r, s = signature_split(expected_signature)
+        mock_trezor_signed = MagicMock()
+        # Checking that v is incremented by sign_message by 4
+        mock_trezor_signed.signature = signature_to_bytes(v - 4, r, s)
+        mock_sign_message.return_value = mock_trezor_signed
+        signature = trezor_wallet.sign_message(safe_message_hash)
+        self.assertEqual(HexBytes(signature), expected_signature)