From b39887773385aa530affd36b8d39688dfaa33347 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ux=C3=ADo?= Date: Mon, 31 Oct 2022 10:13:26 +0100 Subject: [PATCH] Implement Safe EIP1271 message signing (#383) * Refactor domain and message typehashes --- gnosis/safe/safe.py | 59 +++++++++++++++++++++++++++- gnosis/safe/tests/test_signatures.py | 50 ++++++++++++++++++++++- 2 files changed, 106 insertions(+), 3 deletions(-) diff --git a/gnosis/safe/safe.py b/gnosis/safe/safe.py index bbc2e0a6e..d4ba20e1f 100644 --- a/gnosis/safe/safe.py +++ b/gnosis/safe/safe.py @@ -4,9 +4,11 @@ from logging import getLogger from typing import Callable, List, NamedTuple, Optional, Union +from eth_abi import encode_abi +from eth_abi.packed import encode_abi_packed from eth_account import Account from eth_account.signers.local import LocalAccount -from eth_typing import ChecksumAddress +from eth_typing import ChecksumAddress, Hash32 from hexbytes import HexBytes from web3 import Web3 from web3.contract import Contract @@ -27,6 +29,7 @@ from gnosis.eth.utils import ( fast_bytes_to_checksum_address, fast_is_checksum_address, + fast_keccak, get_eth_address_with_key, ) from gnosis.safe.proxy_factory import ProxyFactory @@ -76,12 +79,22 @@ class Safe: Class to manage a Gnosis Safe """ + # keccak256("fallback_manager.handler.address") FALLBACK_HANDLER_STORAGE_SLOT = ( 0x6C9A6C4A39284E37ED1CF53D337577D14212A4870FB976A4366C693B939918D5 ) + # keccak256("guard_manager.guard.address") GUARD_STORAGE_SLOT = ( 0x4A204F620C8C5CCDCA3FD54D003BADD85BA500436A431F0CBDA4F558C93C34C8 ) + # keccak256("EIP712Domain(uint256 chainId,address verifyingContract)"); + DOMAIN_TYPEHASH = bytes.fromhex( + "47e79534a245952e8b16893a336b85a3d9ea9fa8c573f3d803afb92a79469218" + ) + # keccak256("SafeMessage(bytes message)"); + SAFE_MESSAGE_TYPEHASH = bytes.fromhex( + "60b3cbf8b4a223d68d641b3b6ddf9a298e7f33710cf3d3a9d1146b5a6150fbca" + ) def __init__(self, address: ChecksumAddress, ethereum_client: EthereumClient): """ @@ -504,6 +517,19 @@ def contract(self) -> Contract: else: return get_safe_V1_1_1_contract(self.w3, address=self.address) + @cached_property + def domain_separator(self): + return fast_keccak( + encode_abi( + ["bytes32", "uint256", "address"], + [ + self.DOMAIN_TYPEHASH, + self.ethereum_client.get_chain_id(), + self.address, + ], + ) + ) + def check_funds_for_tx_gas( self, safe_tx_gas: int, base_gas: int, gas_price: int, gas_token: str ) -> bool: @@ -805,7 +831,7 @@ def estimate_tx_gas(self, to: str, value: int, data: bytes, operation: int) -> i + WEB3_ESTIMATION_OFFSET ) - def estimate_tx_operational_gas(self, data_bytes_length: int): + def estimate_tx_operational_gas(self, data_bytes_length: int) -> int: """ DEPRECATED. `estimate_tx_base_gas` already includes this. Estimates the gas for the verification of the signatures and other safe related tasks @@ -822,6 +848,35 @@ def estimate_tx_operational_gas(self, data_bytes_length: int): threshold = self.retrieve_threshold() return 15000 + data_bytes_length // 32 * 100 + 5000 * threshold + def get_message_hash(self, message: Union[str, Hash32]) -> Hash32: + """ + Return hash of a message that can be signed by owners. + + :param message: Message that should be hashed + :return: Message hash + """ + + if isinstance(message, str): + message = message.encode() + message_hash = fast_keccak(message) + + safe_message_hash = Web3.keccak( + encode_abi( + ["bytes32", "bytes32"], [self.SAFE_MESSAGE_TYPEHASH, message_hash] + ) + ) + return Web3.keccak( + encode_abi_packed( + ["bytes1", "bytes1", "bytes32", "bytes32"], + [ + bytes.fromhex("19"), + bytes.fromhex("01"), + self.domain_separator, + safe_message_hash, + ], + ) + ) + def retrieve_all_info( self, block_identifier: Optional[BlockIdentifier] = "latest" ) -> SafeInfo: diff --git a/gnosis/safe/tests/test_signatures.py b/gnosis/safe/tests/test_signatures.py index 5aa98b248..f54fd24fe 100644 --- a/gnosis/safe/tests/test_signatures.py +++ b/gnosis/safe/tests/test_signatures.py @@ -4,11 +4,13 @@ from web3 import Web3 from gnosis.eth.constants import NULL_ADDRESS +from gnosis.eth.contracts import get_compatibility_fallback_handler_V1_3_0_contract from ..signatures import get_signing_address +from .safe_test_case import SafeTestCaseMixin -class TestSafeSignature(TestCase): +class TestSafeSignature(SafeTestCaseMixin, TestCase): def test_get_signing_address(self): account = Account.create() # Random hash @@ -24,3 +26,49 @@ def test_get_signing_address(self): get_signing_address(random_hash, signature.v - 8, signature.r, signature.s), NULL_ADDRESS, ) + + def test_eip1271_signing(self): + owner = Account.create() + message = "luar_na_lubre" + safe = self.deploy_test_safe(threshold=1, owners=[owner.address]) + + self.assertEqual( + safe.contract.functions.domainSeparator().call(), safe.domain_separator + ) + + compatibility_contract = get_compatibility_fallback_handler_V1_3_0_contract( + self.w3, safe.address + ) + safe_message_hash = safe.get_message_hash(message) + self.assertEqual( + compatibility_contract.functions.getMessageHash(message.encode()).call(), + safe_message_hash, + ) + + # Use deprecated isValidSignature method (receives bytes) + signature = owner.signHash(safe_message_hash) + is_valid_bytes_fn = compatibility_contract.get_function_by_signature( + "isValidSignature(bytes,bytes)" + ) + self.assertEqual( + is_valid_bytes_fn(message.encode(), signature.signature).call(), + bytes.fromhex("20c13b0b"), + ) + + # Use new isValidSignature method (receives bytes32 == hash of the message) + # Message needs to be hashed first + message_hash = Web3.keccak(text=message) + safe_message_hash = safe.get_message_hash(message_hash) + self.assertEqual( + compatibility_contract.functions.getMessageHash(message_hash).call(), + safe_message_hash, + ) + + signature = owner.signHash(safe_message_hash) + is_valid_bytes_fn = compatibility_contract.get_function_by_signature( + "isValidSignature(bytes32,bytes)" + ) + self.assertEqual( + is_valid_bytes_fn(message_hash, signature.signature).call(), + bytes.fromhex("1626ba7e"), + )