Support Safe Singleton Factory

gnosis/eth/constants.py

@@ -6,6 +6,13 @@
     HexAddress(HexStr("0x" + "0" * 39 + "1"))
 )
 
+SAFE_SINGLETON_FACTORY_DEPLOYER_ADDRESS: ChecksumAddress = ChecksumAddress(
+    HexAddress(HexStr("0xE1CB04A0fA36DdD16a06ea828007E35e1a3cBC37"))
+)
+SAFE_SINGLETON_FACTORY_ADDRESS: ChecksumAddress = ChecksumAddress(
+    HexAddress(HexStr("0x914d7Fec6aaC8cd542e72Bca78B30650d45643d7"))
+)
+
 # keccak('Transfer(address,address,uint256)')
 ERC20_721_TRANSFER_TOPIC: HexStr = HexStr(
     "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"

gnosis/eth/ethereum_client.py

@@ -57,6 +57,7 @@
     fast_is_checksum_address,
     fast_to_checksum_address,
     mk_contract_address,
+    mk_contract_address_2,
 )
 from gnosis.util import cache, chunks
 
@@ -66,12 +67,14 @@
     GAS_CALL_DATA_BYTE,
     GAS_CALL_DATA_ZERO_BYTE,
     NULL_ADDRESS,
+    SAFE_SINGLETON_FACTORY_ADDRESS,
 )
 from .contracts import get_erc20_contract, get_erc721_contract
 from .ethereum_network import EthereumNetwork, EthereumNetworkNotSupported
 from .exceptions import (
     BatchCallFunctionFailed,
     ChainIdIsRequired,
+    ContractAlreadyDeployed,
     FromAddressNotFound,
     GasLimitExceeded,
     InsufficientFunds,
@@ -1325,6 +1328,23 @@ def get_network(self) -> EthereumNetwork:
         """
         return EthereumNetwork(self.get_chain_id())
 
+    @cache
+    def get_singleton_factory_address(self) -> Optional[ChecksumAddress]:
+        """
+        Get singleton factory address if available. Try the singleton managed by Safe by default unless
+        SAFE_SINGLETON_FACTORY_ADDRESS environment variable is defined.
+
+        More info: https://github.com/safe-global/safe-singleton-factory
+
+        :return: Get singleton factory address if available
+        """
+        address = os.environ.get(
+            "SAFE_SINGLETON_FACTORY_ADDRESS", SAFE_SINGLETON_FACTORY_ADDRESS
+        )
+        if self.is_contract(address):
+            return address
+        return None
+
     @cache
     def is_eip1559_supported(self) -> bool:
         """
@@ -1430,22 +1450,54 @@ def batch_call_same_function(
     def deploy_and_initialize_contract(
         self,
         deployer_account: LocalAccount,
-        constructor_data: bytes,
-        initializer_data: bytes = b"",
+        constructor_data: Union[bytes, HexStr],
+        initializer_data: Optional[Union[bytes, HexStr]] = None,
         check_receipt: bool = True,
+        deterministic: bool = True,
     ) -> EthereumTxSent:
+        """
+
+        :param deployer_account:
+        :param constructor_data:
+        :param initializer_data:
+        :param check_receipt:
+        :param deterministic: Use Safe singleton factory for CREATE2 deterministic deployment
+        :return:
+        """
         contract_address: Optional[ChecksumAddress] = None
         for data in (constructor_data, initializer_data):
             # Because initializer_data is not mandatory
             if data:
+                data = HexBytes(data)
                 tx: TxParams = {
                     "from": deployer_account.address,
                     "data": data,
                     "gasPrice": self.w3.eth.gas_price,
                     "value": Wei(0),
                     "to": contract_address if contract_address else "",
                     "chainId": self.get_chain_id(),
+                    "nonce": self.get_nonce_for_account(deployer_account.address),
                 }
+                if not contract_address:
+                    if deterministic and (
+                        singleton_factory_address := self.get_singleton_factory_address()
+                    ):
+                        salt = HexBytes("0" * 64)
+                        tx["data"] = (
+                            salt + data
+                        )  # Add 32 bytes salt for singleton factory
+                        tx["to"] = singleton_factory_address
+                        contract_address = mk_contract_address_2(
+                            singleton_factory_address, salt, data
+                        )
+                        if self.is_contract(contract_address):
+                            raise ContractAlreadyDeployed(
+                                f"Contract {contract_address} already deployed",
+                                contract_address,
+                            )
+                    else:
+                        contract_address = mk_contract_address(tx["from"], tx["nonce"])
+
                 tx["gas"] = self.w3.eth.estimate_gas(tx)
                 tx_hash = self.send_unsigned_transaction(
                     tx, private_key=deployer_account.key
@@ -1457,11 +1509,6 @@ def deploy_and_initialize_contract(
                     assert tx_receipt
                     assert tx_receipt["status"]
 
-                if not contract_address:
-                    contract_address = ChecksumAddress(
-                        mk_contract_address(tx["from"], tx["nonce"])
-                    )
-
         return EthereumTxSent(tx_hash, tx, contract_address)
 
     def get_nonce_for_account(

gnosis/eth/exceptions.py

@@ -1,3 +1,6 @@
+from eth_typing import ChecksumAddress
+
+
 class EthereumClientException(ValueError):
     pass
 
@@ -54,6 +57,12 @@ class TransactionGasPriceTooLow(EthereumClientException):
     pass
 
 
+class ContractAlreadyDeployed(EthereumClientException):
+    def __init__(self, message: str, address: ChecksumAddress):
+        super().__init__(message)
+        self.address = address
+
+
 class InvalidERC20Info(EthereumClientException):
     pass
 

gnosis/eth/multicall.py

@@ -20,7 +20,7 @@
 from . import EthereumClient, EthereumNetwork, EthereumNetworkNotSupported
 from .contracts import ContractBase, get_multicall_v3_contract
 from .ethereum_client import EthereumTxSent
-from .exceptions import BatchCallFunctionFailed
+from .exceptions import BatchCallFunctionFailed, ContractAlreadyDeployed
 from .utils import get_empty_tx_params
 
 logger = logging.getLogger(__name__)
@@ -92,33 +92,37 @@ def get_contract_fn(self) -> Callable[[Web3, Optional[ChecksumAddress]], Contrac
     @classmethod
     def deploy_contract(
         cls, ethereum_client: EthereumClient, deployer_account: LocalAccount
-    ) -> EthereumTxSent:
+    ) -> Optional[EthereumTxSent]:
         """
         Deploy contract
 
         :param ethereum_client:
         :param deployer_account: Ethereum Account
-        :return: ``EthereumTxSent`` with the deployed contract address
+        :return: ``EthereumTxSent`` with the deployed contract address, ``None`` if already deployed
         """
         contract_fn = cls.get_contract_fn(cls)
         contract = contract_fn(ethereum_client.w3)
         constructor_data = contract.constructor().build_transaction(
             get_empty_tx_params()
         )["data"]
 
-        ethereum_tx_sent = ethereum_client.deploy_and_initialize_contract(
-            deployer_account, constructor_data
-        )
+        try:
+            ethereum_tx_sent = ethereum_client.deploy_and_initialize_contract(
+                deployer_account, constructor_data
+            )
 
-        contract_address = ethereum_tx_sent.contract_address
-        logger.info(
-            "Deployed Multicall V2 Contract %s by %s",
-            contract_address,
-            deployer_account.address,
-        )
-        # Add address to addresses dictionary
-        cls.ADDRESSES[ethereum_client.get_network()] = contract_address
-        return ethereum_tx_sent
+            contract_address = ethereum_tx_sent.contract_address
+            logger.info(
+                "Deployed Multicall V2 Contract %s by %s",
+                contract_address,
+                deployer_account.address,
+            )
+            # Add address to addresses dictionary
+            cls.ADDRESSES[ethereum_client.get_network()] = contract_address
+            return ethereum_tx_sent
+        except ContractAlreadyDeployed as e:
+            cls.ADDRESSES[ethereum_client.get_network()] = e.address
+            return None
 
     @staticmethod
     def _build_payload(

gnosis/eth/tests/ethereum_test_case.py

@@ -30,7 +30,7 @@ class EthereumTestCaseMixin:
     def setUpClass(cls):
         super().setUpClass()
 
-        cls.ethereum_test_account = cls.get_ethereum_test_account(cls)
+        cls.ethereum_test_account = cls.get_ethereum_test_account()
         # Caching ethereum_client to prevent initializing again
         cls.ethereum_client = _cached_data["ethereum_client"]
 
@@ -45,7 +45,8 @@ def setUpClass(cls):
         cls.multicall = cls.ethereum_client.multicall
         assert cls.multicall, "Multicall must be defined"
 
-    def get_ethereum_test_account(self) -> LocalAccount:
+    @classmethod
+    def get_ethereum_test_account(cls):
         try:
             from django.conf import settings