Skip to content

Commit

Permalink
Implement Safe EIP1271 message signing (#383)
Browse files Browse the repository at this point in the history
* Refactor domain and message typehashes
  • Loading branch information
Uxio0 authored Oct 31, 2022
1 parent 3309d93 commit b398877
Show file tree
Hide file tree
Showing 2 changed files with 106 additions and 3 deletions.
59 changes: 57 additions & 2 deletions gnosis/safe/safe.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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):
"""
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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
Expand All @@ -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:
Expand Down
50 changes: 49 additions & 1 deletion gnosis/safe/tests/test_signatures.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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"),
)

0 comments on commit b398877

Please sign in to comment.