Add method to get L1 message hash

starknet_py/hash/utils.py

@@ -25,6 +25,12 @@ def _starknet_keccak(data: bytes) -> int:
     return int_from_bytes(k.digest()) & MASK_250
 
 
+def keccak256(data: bytes) -> int:
+    k = keccak.new(digest_bits=256)
+    k.update(data)
+    return int_from_bytes(k.digest())
+
+
 def pedersen_hash(left: int, right: int) -> int:
     """
     One of two hash functions (along with _starknet_keccak) used throughout Starknet.
@@ -70,3 +76,11 @@ def private_to_stark_key(priv_key: int) -> int:
     Deduces the public key given a private key.
     """
     return cpp_get_public_key(priv_key)
+
+
+def encode_uint(value: int, bytes_length: int = 32) -> bytes:
+    return value.to_bytes(bytes_length, byteorder="big")
+
+
+def encode_uint_list(data: List[int]) -> bytes:
+    return b"".join(encode_uint(x) for x in data)

starknet_py/hash/utils_test.py

@@ -2,7 +2,13 @@
 # fmt: off
 import pytest
 
-from starknet_py.hash.utils import compute_hash_on_elements, pedersen_hash
+from starknet_py.hash.utils import (
+    compute_hash_on_elements,
+    encode_uint,
+    encode_uint_list,
+    keccak256,
+    pedersen_hash,
+)
 
 
 @pytest.mark.parametrize(
@@ -33,3 +39,49 @@ def test_compute_hash_on_elements(data, calculated_hash):
 )
 def test_pedersen_hash(first, second, hash_):
     assert pedersen_hash(first, second) == hash_
+
+
+@pytest.mark.parametrize(
+    "value, expected_encoded",
+    [
+        (0, b"\x00" * 32),
+        (1, b"\x00" * 31 + b"\x01"),
+        (123456789, b"\x00" * 28 + b"\x07\x5b\xcd\x15")
+    ]
+)
+def test_encode_uint(value, expected_encoded):
+    assert encode_uint(value) == expected_encoded
+
+
+@pytest.mark.parametrize(
+    "value, expected_encoded",
+    [
+        ([], b""),
+        ([1, 2, 3], b"\x00" * 31 + b"\x01" + b"\x00" * 31 + b"\x02" + b"\x00" * 31 + b"\x03"),
+    ]
+)
+def test_encode_uint_list(value, expected_encoded):
+    assert encode_uint_list(value) == expected_encoded
+
+
+@pytest.mark.parametrize(
+    "string, expected_hash",
+    [
+        ("", 0xc5d2460186f7233c927e7db2dcc703c0e500b653ca82273b7bfad8045d85a470),
+        ("test", 0x9c22ff5f21f0b81b113e63f7db6da94fedef11b2119b4088b89664fb9a3cb658),
+        ("longer test string", 0x47bed17bfbbc08d6b5a0f603eff1b3e932c37c10b865847a7bc73d55b260f32a)
+    ]
+)
+def test_keccak256_strings(string, expected_hash):
+    assert keccak256(string.encode("utf-8")) == expected_hash
+
+
+@pytest.mark.parametrize(
+    "value, expected_hash",
+    [
+        (4, 0x8a35acfbc15ff81a39ae7d344fd709f28e8600b4aa8c65c6b64bfe7fe36bd19b),
+        (5, 0x036b6384b5eca791c62761152d0c79bb0604c104a5fb6f4eb0703f3154bb3db0)
+    ]
+)
+def test_keccak256_ints(value, expected_hash):
+    assert keccak256(encode_uint(value)) == expected_hash

starknet_py/net/client_utils.py

@@ -2,7 +2,8 @@
 
 from typing_extensions import get_args
 
-from starknet_py.net.client_models import Hash, Tag
+from starknet_py.hash.utils import encode_uint, encode_uint_list
+from starknet_py.net.client_models import Hash, L1HandlerTransaction, Tag
 
 
 def hash_to_felt(value: Hash) -> str:
@@ -17,3 +18,21 @@ def hash_to_felt(value: Hash) -> str:
 
 def is_block_identifier(value: Union[int, Hash, Tag]) -> bool:
     return isinstance(value, str) and value in get_args(Tag)
+
+
+def encode_l1_message(tx: L1HandlerTransaction) -> bytes:
+    # TODO (#1047): remove this assert once GatewayClient is deprecated and nonce is always required
+    assert tx.nonce is not None
+
+    from_address = tx.calldata[0]
+    # Pop first element to have in calldata the actual payload
+    tx.calldata.pop(0)
+
+    return (
+        encode_uint(from_address)
+        + encode_uint(tx.contract_address)
+        + encode_uint(tx.nonce)
+        + encode_uint(tx.entry_point_selector)
+        + encode_uint(len(tx.calldata))
+        + encode_uint_list(tx.calldata)
+    )

starknet_py/net/full_node_client.py

@@ -6,6 +6,7 @@
 from marshmallow import EXCLUDE
 
 from starknet_py.constants import RPC_CONTRACT_ERROR
+from starknet_py.hash.utils import keccak256
 from starknet_py.net.client import Client
 from starknet_py.net.client_errors import ClientError
 from starknet_py.net.client_models import (
@@ -19,6 +20,7 @@
     EstimatedFee,
     EventsChunk,
     Hash,
+    L1HandlerTransaction,
     PendingBlockStateUpdate,
     PendingStarknetBlock,
     PendingStarknetBlockWithTxHashes,
@@ -36,6 +38,7 @@
     TransactionTrace,
     TransactionType,
 )
+from starknet_py.net.client_utils import encode_l1_message
 from starknet_py.net.http_client import RpcHttpClient
 from starknet_py.net.models.transaction import (
     AccountTransaction,
@@ -322,6 +325,16 @@ async def get_transaction(
             raise TransactionNotReceivedError() from ex
         return cast(Transaction, TypesOfTransactionsSchema().load(res, unknown=EXCLUDE))
 
+    async def get_l1_message_hash(self, tx_hash: Hash) -> Hash:
+        tx = await self.get_transaction(tx_hash)
+        if not isinstance(tx, L1HandlerTransaction):
+            raise TypeError(
+                f"Transaction {tx_hash} is not a result of L1->L2 interaction."
+            )
+
+        encoded_message = encode_l1_message(tx)
+        return keccak256(encoded_message)
+
     async def get_transaction_receipt(self, tx_hash: Hash) -> TransactionReceipt:
         res = await self._client.call(
             method_name="getTransactionReceipt",

starknet_py/tests/e2e/tests_on_networks/client_test.py

@@ -402,6 +402,35 @@ async def test_get_block(full_node_client_integration):
         assert tx.hash is not None
 
 
+@pytest.mark.skipif(
+    condition="--client=gateway" in sys.argv,
+    reason="Method get_l1_message_hash not implemented for Gateway client.",
+)
+@pytest.mark.asyncio
+async def test_get_l1_message_hash(full_node_client_integration):
+    tx_hash = "0x0060bd50c38082211e6aedb21838fe7402a67216d559d9a4848e6c5e9670c90e"
+    l1_message_hash = await full_node_client_integration.get_l1_message_hash(tx_hash)
+    assert (
+        hex(l1_message_hash)
+        == "0x140185c79e5a04c7c3fae513001f358beb66653dcee75be38f05bd30adba85dd"
+    )
+
+
+@pytest.mark.skipif(
+    condition="--client=gateway" in sys.argv,
+    reason="Method get_l1_message_hash not implemented for Gateway client.",
+)
+@pytest.mark.asyncio
+async def test_get_l1_message_hash_raises_on_incorrect_transaction_type(
+    full_node_client_integration,
+):
+    tx_hash = "0x06d11fa74255c1f86aace54cbf382ab8c89e2b90fb0801f751834ca52bf2a2a2"
+    with pytest.raises(
+        TypeError, match=f"Transaction {tx_hash} is not a result of L1->L2 interaction."
+    ):
+        await full_node_client_integration.get_l1_message_hash(tx_hash)
+
+
 @pytest.mark.asyncio
 async def test_get_public_key(gateway_client_integration):
     current_public_key = (