Support multiple types of proxies

- Until now, Safe only supported Safe Proxies
- Now ERC-1167 and ERC-1967 proxies are also supported

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,