Add get_create_address()

docs/api.rst

@@ -209,6 +209,8 @@ Utility classes
 Utility methods
 ---------------
 
+.. autofunction:: pons.get_create_address
+
 .. autofunction:: pons.get_create2_address
 
 

docs/changelog.rst

@@ -40,6 +40,7 @@ Added
 - Support for a custom block number in gas estimation methods. (PR_70_)
 - ``LocalProvider`` accepts an ``evm_version`` parameter. (PR_78_)
 - ``get_create2_address()``. (PR_80_)
+- ``get_create_address()``. (PR_80_)
 
 
 Fixed

pons/__init__.py

@@ -49,7 +49,7 @@
 from ._local_provider import LocalProvider, SnapshotID
 from ._provider import HTTPProvider, ProtocolError, Provider, Unreachable
 from ._signer import AccountSigner, Signer
-from ._utils import get_create2_address
+from ._utils import get_create2_address, get_create_address
 
 __all__ = [
     "ABIDecodingError",
@@ -99,5 +99,6 @@
     "Unreachable",
     "abi",
     "compile_contract_file",
+    "get_create_address",
     "get_create2_address",
 ]

pons/_utils.py

@@ -1,6 +1,51 @@
 from ethereum_rpc import Address, keccak
 
 
+def _rlp_encode(value: int | bytes | list[int | bytes]) -> bytes:
+    """
+    A limited subset of RLP encoding, so that we don't have to carry a whole `rlp` dependency
+    just for contract address calculation.
+    """
+    list_size_limit = 55
+    list_prefix = 0xC0
+    string_size_limit = 55
+    string_prefix = 0x80
+    small_int_limit = 0x7F
+
+    if isinstance(value, list):
+        items = [_rlp_encode(item) for item in value]
+        list_bytes = b"".join(items)
+        assert len(list_bytes) <= list_size_limit  # noqa: S101
+        return (list_prefix + len(list_bytes)).to_bytes(1, byteorder="big") + list_bytes
+
+    if isinstance(value, int):
+        # Note that there is an error in the official docs.
+        # It says "For a single byte whose value is in the [0x00, 0x7f] (decimal [0, 127]) range,
+        # that byte is its own RLP encoding."
+        # But the encoding of `0` is `0x80`, not `0x00`
+        # (that is, the encoding of a 0-length string).
+        if 0 < value <= small_int_limit:
+            return value.to_bytes(1, byteorder="big")
+        value_len = (value.bit_length() + 7) // 8
+        return _rlp_encode(value.to_bytes(value_len, byteorder="big"))
+
+    assert len(value) <= string_size_limit  # noqa: S101
+    return (string_prefix + len(value)).to_bytes(1, byteorder="big") + value
+
+
+def get_create_address(deployer: Address, nonce: int) -> Address:
+    """
+    Returns the deterministic deployed contract address as produced by ``CREATE`` opcode.
+    Here `deployer` is the contract address invoking ``CREATE`` (if initiated in a contract),
+    or the transaction initiator, if a contract is created via an RPC transaction.
+    ``init_code`` is the deployment code (see :py:attr:`~pons.BoundConstructorCall.data_bytes`).
+    """
+    # This will not hit the length limits since the nonce length is 32 bytes,
+    # and the address length is 20 bytes, which, with length specifiers, is 54 bytes in total.
+    contract_address = keccak(_rlp_encode([bytes(deployer), nonce]))[-20:]
+    return Address(contract_address)
+
+
 def get_create2_address(deployer: Address, init_code: bytes, salt: bytes) -> Address:
     """
     Returns the deterministic deployed contract address as produced by ``CREATE2`` opcode.

tests/test_utils.py

@@ -4,6 +4,7 @@
 import pytest
 
 from pons import EVMVersion, compile_contract_file, get_create2_address
+from pons._utils import get_create_address
 
 
 @pytest.fixture
@@ -12,6 +13,27 @@ def compiled_contracts():
     return compile_contract_file(path, evm_version=EVMVersion.CANCUN)
 
 
+async def test_create(session, root_signer, compiled_contracts):
+    compiled_to_deploy = compiled_contracts["ToDeploy"]
+
+    # Try deploying the contract with different nonces
+
+    nonce = await session.eth_get_transaction_count(root_signer.address)
+    assert nonce == 0
+    to_deploy = await session.deploy(root_signer, compiled_to_deploy.constructor(123))
+    assert to_deploy.address == get_create_address(root_signer.address, nonce)
+
+    nonce = await session.eth_get_transaction_count(root_signer.address)
+    assert nonce == 1
+    to_deploy = await session.deploy(root_signer, compiled_to_deploy.constructor(123))
+    assert to_deploy.address == get_create_address(root_signer.address, nonce)
+
+    nonce = await session.eth_get_transaction_count(root_signer.address)
+    assert nonce == 2
+    to_deploy = await session.deploy(root_signer, compiled_to_deploy.constructor(123))
+    assert to_deploy.address == get_create_address(root_signer.address, nonce)
+
+
 async def test_create2(session, root_signer, compiled_contracts):
     compiled_deployer = compiled_contracts["Create2Deployer"]
     compiled_to_deploy = compiled_contracts["ToDeploy"]