Optimize to_checksum_address function

While developing Safe Tx Service we found out that a lot of time was spent
on converting addresses from `bytes` to `ChecksumAddress`. Biggest part
of this time was overhead added on `eth_utils` library to `keccak256`
function.

So we decided to use `pysha3` library (way faster than `pycryptodome`)
directly. That's why we implemented new methods:

- fast_keccak
- fast_keccak_hex
- fast_keccak_hex
- fast_to_checksum_address
- fast_bytes_to_checksum_address
- fast_is_checksum_address

This is also very relevant for the `EthereumAddressV2Field`, as it
stores addresses as `bytes` on database and returns them as
`ChecksumAddress`

Related to:
  - ethereum/eth-utils#95
  - ethereum/eth-hash#35

gnosis/eth/clients/sourcify.py

@@ -2,9 +2,9 @@
 from urllib.parse import urljoin
 
 import requests
-from web3 import Web3
 
 from .. import EthereumNetwork
+from ..utils import fast_is_checksum_address
 from .contract_metadata import ContractMetadata
 
 
@@ -47,7 +47,7 @@ def _do_request(self, url: str) -> Optional[Dict[str, Any]]:
     def get_contract_metadata(
         self, contract_address: str
     ) -> Optional[ContractMetadata]:
-        assert Web3.isChecksumAddress(
+        assert fast_is_checksum_address(
             contract_address
         ), "Expecting a checksummed address"
 

gnosis/eth/django/filters.py

@@ -6,7 +6,8 @@
 
 import django_filters
 from hexbytes import HexBytes
-from web3 import Web3
+
+from ..utils import fast_is_checksum_address
 
 
 class EthereumAddressFieldForm(CharFieldForm):
@@ -21,7 +22,7 @@ def to_python(self, value):
         value = super().to_python(value)
         if value in self.empty_values:
             return None
-        elif not Web3.isChecksumAddress(value):
+        elif not fast_is_checksum_address(value):
             raise ValidationError(self.error_messages["invalid"], code="invalid")
         return value
 

gnosis/eth/django/models.py

@@ -6,9 +6,9 @@
 from django.utils.translation import gettext_lazy as _
 
 from eth_typing import ChecksumAddress
-from eth_utils import to_checksum_address
 from hexbytes import HexBytes
 
+from ..utils import fast_bytes_to_checksum_address, fast_to_checksum_address
 from .filters import EthereumAddressFieldForm, Keccak256FieldForm
 from .validators import validate_checksumed_address
 
@@ -45,7 +45,7 @@ def to_python(self, value):
         value = super().to_python(value)
         if value:
             try:
-                return to_checksum_address(value)
+                return fast_to_checksum_address(value)
             except ValueError:
                 raise exceptions.ValidationError(
                     self.error_messages["invalid"],
@@ -74,7 +74,7 @@ def from_db_value(
         self, value: memoryview, expression, connection
     ) -> Optional[ChecksumAddress]:
         if value:
-            return to_checksum_address(value.hex())
+            return fast_bytes_to_checksum_address(value)
 
     def get_prep_value(self, value: ChecksumAddress) -> Optional[bytes]:
         if value:
@@ -83,7 +83,7 @@ def get_prep_value(self, value: ChecksumAddress) -> Optional[bytes]:
     def to_python(self, value) -> Optional[ChecksumAddress]:
         if value is not None:
             try:
-                return to_checksum_address(value)
+                return fast_to_checksum_address(value)
             except ValueError:
                 raise exceptions.ValidationError(
                     self.error_messages["invalid"],
@@ -101,12 +101,13 @@ def formfield(self, **kwargs):
 
 
 class Uint256Field(models.DecimalField):
-    description = _("Ethereum uint256 number")
     """
     Field to store ethereum uint256 values. Uses Decimal db type without decimals to store
     in the database, but retrieve as `int` instead of `Decimal` (https://docs.python.org/3/library/decimal.html)
     """
 
+    description = _("Ethereum uint256 number")
+
     def __init__(self, *args, **kwargs):
         kwargs["max_digits"] = 79  # 2 ** 256 is 78 digits
         kwargs["decimal_places"] = 0

gnosis/eth/django/serializers.py

@@ -5,7 +5,6 @@
 from hexbytes import HexBytes
 from rest_framework import serializers
 from rest_framework.exceptions import ValidationError
-from web3 import Web3
 
 from ..constants import (
     SIGNATURE_R_MAX_VALUE,
@@ -15,6 +14,7 @@
     SIGNATURE_V_MAX_VALUE,
     SIGNATURE_V_MIN_VALUE,
 )
+from ..utils import fast_is_checksum_address
 
 logger = logging.getLogger(__name__)
 
@@ -44,7 +44,7 @@ def to_representation(self, obj):
     def to_internal_value(self, data):
         # Check if address is valid
         try:
-            if Web3.toChecksumAddress(data) != data:
+            if not fast_is_checksum_address(data):
                 raise ValueError
             elif int(data, 16) == 0 and not self.allow_zero_address:
                 raise ValidationError("0x0 address is not allowed")

gnosis/eth/django/tests/test_forms.py

@@ -16,7 +16,7 @@ class Keccak256Form(forms.Form):
 
 class TestForms(TestCase):
     def test_ethereum_address_field_form(self):
-        form = EthereumAddressForm(data={"value": "not a ethereum address"})
+        form = EthereumAddressForm(data={"value": "not an ethereum address"})
         self.assertFalse(form.is_valid())
         self.assertEqual(
             form.errors["value"], ["Enter a valid checksummed Ethereum Address."]

gnosis/eth/django/tests/test_models.py

@@ -7,6 +7,7 @@
 from web3 import Web3
 
 from ...constants import NULL_ADDRESS, SENTINEL_ADDRESS
+from ...utils import fast_is_checksum_address
 from .models import EthereumAddress, EthereumAddressV2, Keccak256Hash, Sha3Hash, Uint256
 
 faker = Faker()
@@ -17,10 +18,10 @@ def test_ethereum_address_field(self):
         for EthereumAddressModel in (EthereumAddress, EthereumAddressV2):
             with self.subTest(EthereumAddressModel=EthereumAddressModel):
                 address = Account.create().address
-                self.assertTrue(Web3.isChecksumAddress(address))
+                self.assertTrue(fast_is_checksum_address(address))
                 ethereum_address = EthereumAddressModel.objects.create(value=address)
                 ethereum_address.refresh_from_db()
-                self.assertTrue(Web3.isChecksumAddress(ethereum_address.value))
+                self.assertTrue(fast_is_checksum_address(ethereum_address.value))
                 self.assertEqual(address, ethereum_address.value)
 
                 # Test addresses

gnosis/eth/django/validators.py

@@ -1,10 +1,10 @@
 from django.core.exceptions import ValidationError
 
-from web3 import Web3
+from ..utils import fast_is_checksum_address
 
 
 def validate_checksumed_address(address):
-    if not Web3.isChecksumAddress(address):
+    if not fast_is_checksum_address(address):
         raise ValidationError(
             "%(address)s has an invalid checksum",
             params={"address": address},

gnosis/eth/ethereum_client.py

@@ -54,7 +54,11 @@
     Wei,
 )
 
-from gnosis.eth.utils import mk_contract_address
+from gnosis.eth.utils import (
+    fast_is_checksum_address,
+    fast_to_checksum_address,
+    mk_contract_address,
+)
 
 from .constants import (
     ERC20_721_TRANSFER_TOPIC,
@@ -434,7 +438,7 @@ def _decode_transfer_log(
                 try:
                     from_to_data = b"".join(topics[1:])
                     _from, to = (
-                        Web3.toChecksumAddress(address)
+                        fast_to_checksum_address(address)
                         for address in eth_abi.decode_abi(
                             ["address", "address"], from_to_data
                         )
@@ -452,7 +456,9 @@ def _decode_transfer_log(
                 _from, to, token_id = eth_abi.decode_abi(
                     ["address", "address", "uint256"], b"".join(topics[1:])
                 )
-                _from, to = [Web3.toChecksumAddress(address) for address in (_from, to)]
+                _from, to = [
+                    fast_to_checksum_address(address) for address in (_from, to)
+                ]
                 return {"from": _from, "to": to, "tokenId": token_id}
         return None
 
@@ -892,7 +898,7 @@ def _decode_trace_action(self, action: Dict[str, Any]) -> Dict[str, Any]:
 
         # CALL, DELEGATECALL, CREATE or CREATE2
         if "from" in action:
-            decoded["from"] = self.w3.toChecksumAddress(action["from"])
+            decoded["from"] = fast_to_checksum_address(action["from"])
         if "gas" in action:
             decoded["gas"] = int(action["gas"], 16)
         if "value" in action:
@@ -904,21 +910,19 @@ def _decode_trace_action(self, action: Dict[str, Any]) -> Dict[str, Any]:
         if "input" in action:
             decoded["input"] = HexBytes(action["input"])
         if "to" in action:
-            decoded["to"] = self.w3.toChecksumAddress(action["to"])
+            decoded["to"] = fast_to_checksum_address(action["to"])
 
         # CREATE or CREATE2
         if "init" in action:
             decoded["init"] = HexBytes(action["init"])
 
         # SELF-DESTRUCT
         if "address" in action:
-            decoded["address"] = self.w3.toChecksumAddress(action["address"])
+            decoded["address"] = fast_to_checksum_address(action["address"])
         if "balance" in action:
             decoded["balance"] = int(action["balance"], 16)
         if "refundAddress" in action:
-            decoded["refundAddress"] = self.w3.toChecksumAddress(
-                action["refundAddress"]
-            )
+            decoded["refundAddress"] = fast_to_checksum_address(action["refundAddress"])
 
         return decoded
 
@@ -935,7 +939,7 @@ def _decode_trace_result(self, result: Dict[str, Any]) -> Dict[str, Any]:
         if "code" in result:
             decoded["code"] = HexBytes(result["code"])
         if "address" in result:
-            decoded["address"] = self.w3.toChecksumAddress(result["address"])
+            decoded["address"] = fast_to_checksum_address(result["address"])
 
         return decoded
 
@@ -1898,7 +1902,7 @@ def send_eth_to(
         :return: tx_hash
         """
 
-        assert Web3.isChecksumAddress(to)
+        assert fast_is_checksum_address(to)
 
         account = Account.from_key(private_key)
 

gnosis/eth/oracles/oracles.py

@@ -9,7 +9,6 @@
 from eth_abi.packed import encode_abi_packed
 from eth_typing import ChecksumAddress
 from hexbytes import HexBytes
-from web3 import Web3
 from web3.contract import Contract
 from web3.exceptions import BadFunctionCallOutput
 
@@ -23,6 +22,7 @@
     get_uniswap_v2_pair_contract,
     get_uniswap_v2_router_contract,
 )
+from ..utils import fast_bytes_to_checksum_address, fast_keccak
 from .abis.aave_abis import AAVE_ATOKEN_ABI
 from .abis.balancer_abis import balancer_pool_abi
 from .abis.cream_abis import cream_ctoken_abi
@@ -380,16 +380,16 @@ def calculate_pair_address(self, token_address: str, token_address_2: str):
         """
         if token_address.lower() > token_address_2.lower():
             token_address, token_address_2 = token_address_2, token_address
-        salt = Web3.keccak(
+        salt = fast_keccak(
             encode_abi_packed(["address", "address"], [token_address, token_address_2])
         )
-        address = Web3.keccak(
+        address = fast_keccak(
             encode_abi_packed(
                 ["bytes", "address", "bytes", "bytes"],
                 [HexBytes("ff"), self.factory_address, salt, self.pair_init_code],
             )
         )[-20:]
-        return Web3.toChecksumAddress(address)
+        return fast_bytes_to_checksum_address(address)
 
     def get_decimals(self, token_address: str, token_address_2: str) -> Tuple[int, int]:
         if not (

gnosis/eth/tests/test_ethereum_client.py

@@ -22,7 +22,7 @@
     SenderAccountNotFoundInNode,
 )
 from ..exceptions import BatchCallException, InvalidERC20Info
-from ..utils import get_eth_address_with_key
+from ..utils import fast_to_checksum_address, get_eth_address_with_key
 from .ethereum_test_case import EthereumTestCaseMixin
 from .mocks.mock_internal_txs import creation_internal_txs, internal_txs_errored
 from .mocks.mock_log_receipts import invalid_log_receipt, log_receipts
@@ -531,7 +531,7 @@ def test_decode_trace(self):
         self.assertEqual(decoded_traces[0]["result"]["output"], HexBytes(""))
         self.assertEqual(
             decoded_traces[1]["result"]["address"],
-            self.w3.toChecksumAddress(example_traces[1]["result"]["address"]),
+            fast_to_checksum_address(example_traces[1]["result"]["address"]),
         )
         self.assertEqual(
             decoded_traces[1]["result"]["code"],

gnosis/eth/tests/test_utils.py

@@ -1,11 +1,21 @@
+import os
+
 from django.test import TestCase
 
 from eth_abi.packed import encode_abi_packed
 from eth_account import Account
+from eth_utils import to_checksum_address
 from hexbytes import HexBytes
 
 from ..contracts import get_proxy_1_0_0_deployed_bytecode, get_proxy_factory_contract
-from ..utils import compare_byte_code, decode_string_or_bytes32, generate_address_2
+from ..utils import (
+    compare_byte_code,
+    decode_string_or_bytes32,
+    fast_bytes_to_checksum_address,
+    fast_is_checksum_address,
+    fast_to_checksum_address,
+    generate_address_2,
+)
 from .ethereum_test_case import EthereumTestCaseMixin
 
 
@@ -104,3 +114,38 @@ def test_compare_byte_code(self):
         self.assertTrue(
             compare_byte_code(proxy_with_metadata, proxy_with_different_metadata)
         )
+
+    def test_fast_to_checksum_address(self):
+        for _ in range(10):
+            address = os.urandom(20).hex()
+            self.assertEqual(
+                fast_to_checksum_address(address), to_checksum_address(address)
+            )
+
+    def test_fast_bytes_to_checksum_address(self):
+        with self.assertRaises(ValueError):
+            fast_bytes_to_checksum_address(os.urandom(19))
+
+        with self.assertRaises(ValueError):
+            fast_bytes_to_checksum_address(os.urandom(21))
+
+        for _ in range(10):
+            address = os.urandom(20)
+            self.assertEqual(
+                fast_bytes_to_checksum_address(address), to_checksum_address(address)
+            )
+
+    def test_fast_is_checksum_address(self):
+        self.assertFalse(fast_is_checksum_address(None))
+        self.assertFalse(fast_is_checksum_address(""))
+        self.assertFalse(
+            fast_is_checksum_address(0x6ED857DC1DA2C41470A95589BB482152000773E9)
+        )
+        self.assertFalse(fast_is_checksum_address(2))
+        self.assertFalse(fast_is_checksum_address("2"))
+        self.assertFalse(
+            fast_is_checksum_address("0x6ed857dc1da2c41470A95589bB482152000773e9")
+        )
+        self.assertTrue(
+            fast_is_checksum_address("0x6ED857dc1da2c41470A95589bB482152000773e9")
+        )