Support multiple types of proxies

safe_eth/eth/proxies/__init__.py

@@ -0,0 +1,5 @@
+# flake8: noqa F401
+from .minimal_proxy import MinimalProxy
+from .proxy import Proxy
+from .safe_proxy import SafeProxy
+from .standard_proxy import StandardProxy

safe_eth/eth/proxies/minimal_proxy.py

@@ -0,0 +1,59 @@
+from functools import cache
+from typing import Optional
+
+from eth_typing import ChecksumAddress
+from hexbytes import HexBytes
+from web3.types import BlockIdentifier
+
+from ..constants import NULL_ADDRESS
+from ..utils import fast_to_checksum_address
+from .proxy import Proxy
+
+
+class MinimalProxy(Proxy):
+    """
+    Minimal proxy implementation, following EIP-1167
+
+    https://eips.ethereum.org/EIPS/eip-1167
+    """
+
+    @staticmethod
+    def get_deployment_data(implementation_address: ChecksumAddress) -> bytes:
+        """
+        :param implementation_address: Contract address the Proxy will point to
+        :return: Deployment data for a minimal proxy pointing to the given `contract_address`
+        """
+        return (
+            HexBytes("0x6c3d82803e903d91602b57fd5bf3600d527f363d3d373d3d3d363d73")
+            + HexBytes(implementation_address)
+            + HexBytes("5af4600052602d6000f3")
+        )
+
+    @staticmethod
+    def get_expected_code(implementation_address: ChecksumAddress) -> bytes:
+        """
+        This method is only relevant to do checks and make sure the code deployed is the one expected
+
+        :param implementation_address:
+        :return: Expected code for a given `contract_address`
+        """
+        return (
+            HexBytes("363d3d373d3d3d363d73")
+            + HexBytes(implementation_address)
+            + HexBytes("5af43d82803e903d91602b57fd5bf3")
+        )
+
+    @cache
+    def get_singleton_address(
+        self, block_identifier: Optional[BlockIdentifier] = "latest"
+    ) -> ChecksumAddress:
+        """
+        Minimal proxies cannot be upgraded, so return value is cached
+
+        :return: Address for the singleton contract the Proxy points to
+        """
+        code = self.get_code()
+        if len(code) != 45:  # Not a minimal proxy implementation
+            return NULL_ADDRESS
+
+        return fast_to_checksum_address(code[10:30])

safe_eth/eth/proxies/proxy.py

@@ -0,0 +1,43 @@
+from abc import ABCMeta
+from functools import cache
+from typing import Optional
+
+from eth_typing import ChecksumAddress
+from web3.types import BlockIdentifier
+
+from ..ethereum_client import EthereumClient
+from ..utils import fast_bytes_to_checksum_address
+
+
+class Proxy(metaclass=ABCMeta):
+    """
+    Generic class for proxy contracts
+    """
+
+    def __init__(self, address: ChecksumAddress, ethereum_client: EthereumClient):
+        """
+        :param address: Proxy address
+        """
+        self.address = address
+        self.ethereum_client = ethereum_client
+        self.w3 = ethereum_client.w3
+
+    def _parse_address_in_storage(self, storage_bytes: bytes) -> ChecksumAddress:
+        """
+        :param storage_slot:
+        :return: A checksummed address in a slot
+        """
+        address = storage_bytes[-20:].rjust(20, b"\0")
+        return fast_bytes_to_checksum_address(address)
+
+    @cache
+    def get_code(self):
+        return self.w3.eth.get_code(self.address)
+
+    def get_singleton_address(
+        self, block_identifier: Optional[BlockIdentifier] = "latest"
+    ) -> ChecksumAddress:
+        """
+        :return: Address for the singleton contract the Proxy points to
+        """
+        raise NotImplementedError

safe_eth/eth/proxies/safe_proxy.py

@@ -0,0 +1,23 @@
+from typing import Optional
+
+from eth_typing import ChecksumAddress
+from web3.types import BlockIdentifier
+
+from .proxy import Proxy
+
+
+class SafeProxy(Proxy):
+    """
+    Proxy implementation from Safe
+    """
+
+    def get_singleton_address(
+        self, block_identifier: Optional[BlockIdentifier] = "latest"
+    ) -> ChecksumAddress:
+        """
+        :return: Address for the singleton contract the Proxy points to
+        """
+        storage_bytes = self.w3.eth.get_storage_at(
+            self.address, 0, block_identifier=block_identifier
+        )
+        return self._parse_address_in_storage(storage_bytes)

safe_eth/eth/proxies/standard_proxy.py

@@ -0,0 +1,60 @@
+from typing import Optional
+
+from eth_typing import ChecksumAddress
+from web3.types import BlockIdentifier
+
+from ..constants import NULL_ADDRESS
+from .proxy import Proxy
+
+
+class StandardProxy(Proxy):
+    """
+    Standard proxy implementation, following EIP-1967
+
+    https://eips.ethereum.org/EIPS/eip-1967
+    """
+
+    # bytes32(uint256(keccak256('eip1967.proxy.implementation')) - 1))
+    LOGIC_CONTRACT_SLOT = (
+        0x360894A13BA1A3210667C828492DB98DCA3E2076CC3735A920A3CA505D382BBC
+    )
+
+    # bytes32(uint256(keccak256('eip1967.proxy.beacon')) - 1)
+    BEACON_CONTRACT_SLOT = (
+        0xA3F0AD74E5423AEBFD80D3EF4346578335A9A72AEAEE59FF6CB3582B35133D50
+    )
+
+    # bytes32(uint256(keccak256('eip1967.proxy.admin')) - 1)
+    ADMIN_CONTRACT_SLOT = (
+        0xB53127684A568B3173AE13B9F8A6016E243E63B6E8EE1178D6A717850B5D6103
+    )
+
+    def get_singleton_address(
+        self, block_identifier: Optional[BlockIdentifier] = "latest"
+    ) -> ChecksumAddress:
+        """
+        :param block_identifier:
+        :return: address of the logic contract that this proxy delegates to or the beacon contract
+                 the proxy relies on (fallback)
+        """
+        for slot in (self.LOGIC_CONTRACT_SLOT, self.BEACON_CONTRACT_SLOT):
+            storage_bytes = self.w3.eth.get_storage_at(
+                self.address, slot, block_identifier=block_identifier
+            )
+            address = self._parse_address_in_storage(storage_bytes)
+            if address != NULL_ADDRESS:
+                return address
+        return NULL_ADDRESS
+
+    def get_admin_address(
+        self, block_identifier: Optional[BlockIdentifier] = "latest"
+    ) -> ChecksumAddress:
+        """
+        :param block_identifier:
+        :return: address that is allowed to upgrade the logic contract address for the proxy (optional)
+        """
+        storage_bytes = self.w3.eth.get_storage_at(
+            self.address, self.ADMIN_CONTRACT_SLOT, block_identifier=block_identifier
+        )
+        address = self._parse_address_in_storage(storage_bytes)
+        return address

safe_eth/eth/tests/proxies/test_minimal_proxy.py

@@ -0,0 +1,25 @@
+from unittest import TestCase
+
+from eth_account import Account
+
+from safe_eth.eth.proxies import MinimalProxy
+from safe_eth.eth.tests.ethereum_test_case import EthereumTestCaseMixin
+
+
+class TestMinimalProxy(EthereumTestCaseMixin, TestCase):
+    def test_get_singleton_address(self):
+        account = self.ethereum_test_account
+        contract_address = Account.create().address
+        deployment_data = MinimalProxy.get_deployment_data(contract_address)
+        expected_code = MinimalProxy.get_expected_code(contract_address)
+
+        tx = {"data": deployment_data}
+
+        tx_hash = self.send_tx(tx, account)
+        tx_receipt = self.w3.eth.wait_for_transaction_receipt(tx_hash)
+        proxy_address = tx_receipt["contractAddress"]
+        code = self.w3.eth.get_code(proxy_address)
+        self.assertEqual(code, expected_code)
+
+        minimal_proxy = MinimalProxy(proxy_address, self.ethereum_client)
+        self.assertEqual(minimal_proxy.get_singleton_address(), contract_address)

safe_eth/eth/tests/proxies/test_safe_proxy.py

@@ -0,0 +1,17 @@
+from unittest import TestCase
+
+from safe_eth.eth.proxies import SafeProxy
+from safe_eth.safe.tests.safe_test_case import SafeTestCaseMixin
+
+
+class TestSafeProxy(SafeTestCaseMixin, TestCase):
+    def test_get_singleton_address(self):
+        safe = self.deploy_test_safe_v1_4_1()
+        self.assertEqual(
+            safe.retrieve_master_copy_address(), self.safe_contract_V1_4_1.address
+        )
+
+        safe_proxy = SafeProxy(safe.address, self.ethereum_client)
+        self.assertEqual(
+            safe_proxy.get_singleton_address(), self.safe_contract_V1_4_1.address
+        )

safe_eth/eth/tests/proxies/test_standard_proxy.py

@@ -0,0 +1,88 @@
+from unittest import TestCase
+
+from eth_typing import ChecksumAddress
+from hexbytes import HexBytes
+
+from safe_eth.eth.proxies import StandardProxy
+from safe_eth.safe import Safe
+from safe_eth.safe.tests.safe_test_case import SafeTestCaseMixin
+
+
+class TestStandardProxy(SafeTestCaseMixin, TestCase):
+    standard_proxy_bytecode = HexBytes(
+        "0x608060405234801561001057600080fd5b506040516106f63803806106f683398181016040528101906100329190610523565b8181610044828261004d60201b60201c565b50505050610607565b61005c826100d260201b60201c565b8173ffffffffffffffffffffffffffffffffffffffff167fbc7cd75a20ee27fd9adebab32041f755214dbc6bffa90cc0225b39da2e5c2d3b60405160405180910390a26000815111156100bf576100b982826101a560201b60201c565b506100ce565b6100cd61022f60201b60201c565b5b5050565b60008173ffffffffffffffffffffffffffffffffffffffff163b0361012e57806040517f4c9c8ce3000000000000000000000000000000000000000000000000000000008152600401610125919061058e565b60405180910390fd5b806101617f360894a13ba1a3210667c828492db98dca3e2076cc3735a920a3ca505d382bbc60001b61026c60201b60201c565b60000160006101000a81548173ffffffffffffffffffffffffffffffffffffffff021916908373ffffffffffffffffffffffffffffffffffffffff16021790555050565b60606000808473ffffffffffffffffffffffffffffffffffffffff16846040516101cf91906105f0565b600060405180830381855af49150503d806000811461020a576040519150601f19603f3d011682016040523d82523d6000602084013e61020f565b606091505b509150915061022585838361027660201b60201c565b9250505092915050565b600034111561026a576040517fb398979f00000000000000000000000000000000000000000000000000000000815260040160405180910390fd5b565b6000819050919050565b6060826102915761028c8261030b60201b60201c565b610303565b600082511480156102b9575060008473ffffffffffffffffffffffffffffffffffffffff163b145b156102fb57836040517f9996b3150000000000000000000000000000000000000000000000000000000081526004016102f2919061058e565b60405180910390fd5b819050610304565b5b9392505050565b60008151111561031e5780518082602001fd5b6040517f1425ea4200000000000000000000000000000000000000000000000000000000815260040160405180910390fd5b6000604051905090565b600080fd5b600080fd5b600073ffffffffffffffffffffffffffffffffffffffff82169050919050565b600061038f82610364565b9050919050565b61039f81610384565b81146103aa57600080fd5b50565b6000815190506103bc81610396565b92915050565b600080fd5b600080fd5b6000601f19601f8301169050919050565b7f4e487b7100000000000000000000000000000000000000000000000000000000600052604160045260246000fd5b610415826103cc565b810181811067ffffffffffffffff82111715610434576104336103dd565b5b80604052505050565b6000610447610350565b9050610453828261040c565b919050565b600067ffffffffffffffff821115610473576104726103dd565b5b61047c826103cc565b9050602081019050919050565b60005b838110156104a757808201518184015260208101905061048c565b60008484015250505050565b60006104c66104c184610458565b61043d565b9050828152602081018484840111156104e2576104e16103c7565b5b6104ed848285610489565b509392505050565b600082601f83011261050a576105096103c2565b5b815161051a8482602086016104b3565b91505092915050565b6000806040838503121561053a5761053961035a565b5b6000610548858286016103ad565b925050602083015167ffffffffffffffff8111156105695761056861035f565b5b610575858286016104f5565b9150509250929050565b61058881610384565b82525050565b60006020820190506105a3600083018461057f565b92915050565b600081519050919050565b600081905092915050565b60006105ca826105a9565b6105d481856105b4565b93506105e4818560208601610489565b80840191505092915050565b60006105fc82846105bf565b915081905092915050565b60e1806106156000396000f3fe6080604052600a600c565b005b60186014601a565b6027565b565b60006022604c565b905090565b3660008037600080366000845af43d6000803e80600081146047573d6000f35b3d6000fd5b600060787f360894a13ba1a3210667c828492db98dca3e2076cc3735a920a3ca505d382bbc60001b60a1565b60000160009054906101000a900473ffffffffffffffffffffffffffffffffffffffff16905090565b600081905091905056fea26469706673582212209a40f50a4ff13192312765b2f472363e3ca8c8f25fbefc2363ea1ad3850c61dc64736f6c634300081b0033"
+    )
+    standard_proxy_abi = [
+        {
+            "inputs": [
+                {
+                    "internalType": "address",
+                    "name": "_implementation",
+                    "type": "address",
+                },
+                {"internalType": "bytes", "name": "_data", "type": "bytes"},
+            ],
+            "stateMutability": "nonpayable",
+            "type": "constructor",
+        },
+        {
+            "inputs": [
+                {"internalType": "address", "name": "target", "type": "address"}
+            ],
+            "name": "AddressEmptyCode",
+            "type": "error",
+        },
+        {
+            "inputs": [
+                {"internalType": "address", "name": "implementation", "type": "address"}
+            ],
+            "name": "ERC1967InvalidImplementation",
+            "type": "error",
+        },
+        {"inputs": [], "name": "ERC1967NonPayable", "type": "error"},
+        {"inputs": [], "name": "FailedInnerCall", "type": "error"},
+        {
+            "anonymous": False,
+            "inputs": [
+                {
+                    "indexed": True,
+                    "internalType": "address",
+                    "name": "implementation",
+                    "type": "address",
+                }
+            ],
+            "name": "Upgraded",
+            "type": "event",
+        },
+        {"stateMutability": "payable", "type": "fallback"},
+    ]
+
+    def deploy_standard_proxy(
+        self, singleton_address: ChecksumAddress
+    ) -> StandardProxy:
+        """
+        Deploy a EIP-1967 proxy
+
+        :param singleton_address: Address the proxy will point to
+        :return: StandardProxy deployed
+        """
+        standard_proxy = self.w3.eth.contract(
+            abi=self.standard_proxy_abi, bytecode=self.standard_proxy_bytecode
+        )
+        tx_hash = standard_proxy.constructor(singleton_address, b"").transact(
+            {"from": self.ethereum_test_account.address}
+        )
+        tx_receipt = self.w3.eth.wait_for_transaction_receipt(tx_hash)
+        proxy_address = tx_receipt["contractAddress"]
+        assert proxy_address is not None
+
+        return StandardProxy(proxy_address, self.ethereum_client)
+
+    def test_get_singleton_address(self):
+        singleton_address = self.safe_contract_V1_4_1.address
+        standard_proxy = self.deploy_standard_proxy(singleton_address)
+        self.assertEqual(standard_proxy.get_singleton_address(), singleton_address)
+
+        # Test Safe class supports the Proxy
+        safe = Safe(standard_proxy.address, self.ethereum_client)
+        self.assertEqual(safe.retrieve_master_copy_address(), singleton_address)

safe_eth/safe/safe.py

@@ -31,14 +31,15 @@
     get_safe_V1_4_1_contract,
     get_simulate_tx_accessor_V1_4_1_contract,
 )
+from safe_eth.eth.proxies import MinimalProxy, SafeProxy, StandardProxy
+from safe_eth.eth.typing import EthereumData
 from safe_eth.eth.utils import (
     fast_bytes_to_checksum_address,
     fast_is_checksum_address,
     fast_keccak,
     get_empty_tx_params,
 )
 
-from ..eth.typing import EthereumData
 from .addresses import SAFE_SIMULATE_TX_ACCESSOR_ADDRESS
 from .enums import SafeOperationEnum
 from .exceptions import CannotEstimateGas, CannotRetrieveSafeInfoException
@@ -660,10 +661,16 @@ def retrieve_guard(
     def retrieve_master_copy_address(
         self, block_identifier: Optional[BlockIdentifier] = "latest"
     ) -> ChecksumAddress:
-        address = self.w3.eth.get_storage_at(
-            self.address, "0x00", block_identifier=block_identifier
-        )[-20:].rjust(20, b"\0")
-        return fast_bytes_to_checksum_address(address)
+        """
+        :param block_identifier:
+        :return: Returns the implementation address. Multiple types of proxies are supported
+        """
+        for ProxyClass in (SafeProxy, StandardProxy, MinimalProxy):
+            proxy = ProxyClass(self.address, self.ethereum_client)
+            address = proxy.get_singleton_address(block_identifier=block_identifier)
+            if address != NULL_ADDRESS:
+                return address
+        return address
 
     def retrieve_modules(
         self,