10793540350

Committed 10 Sep 2024 01:31PM UTC coverage: 93.551% (-0.3%) from 93.892%

Build # 10793540350

Build Type

push

github

Committed by

falvaradorodriguez

Commit Message

Fix cowswap test

Run Details

8777 of 9382 relevant lines covered (93.55%)

3.74 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

89.58

/safe_eth/eth/utils.py

import os
from functools import lru_cache
from typing import Any, Union

import eth_abi
from eth._utils.address import generate_contract_address
from eth_account import Account
from eth_typing import Address, AnyAddress, ChecksumAddress, Hash32, HexAddress, HexStr
from eth_utils import to_normalized_address
from hexbytes import HexBytes
from sha3 import keccak_256
from web3.types import TxParams, Wei


def get_empty_tx_params() -> TxParams:
    """
    :return: Empty tx params, so calls like `build_transaction` don't call the RPC trying to get information
    """
    return {
        "gas": Wei(1),
        "gasPrice": Wei(1),
    }


@lru_cache(maxsize=int(os.getenv("CACHE_KECCAK", 512)))
def _keccak_256(value: bytes) -> keccak_256:
    return keccak_256(value)


def fast_keccak(value: bytes) -> Hash32:
    """
    Calculates ethereum keccak256 using fast library `pysha3`

    :param value:
    :return: Keccak256 used by ethereum as `HexBytes`
    """
    return Hash32(HexBytes(_keccak_256(value).digest()))


def fast_keccak_text(value: str) -> Hash32:
    """
    Calculates ethereum keccak256 using fast library `pysha3`

    :param value:
    :return: Keccak256 used by ethereum as `HexBytes`
    """
    return fast_keccak(value.encode())


def fast_keccak_hex(value: bytes) -> HexStr:
    """
    Same as `fast_keccak`, but it's a little more optimal calling `hexdigest()`
    than calling `digest()` and then `hex()`

    :param value:
    :return: Keccak256 used by ethereum as a hex string (not 0x prefixed)
    """
    return HexStr(_keccak_256(value).hexdigest())


def _build_checksum_address(
    norm_address: HexStr, address_hash: HexStr
) -> ChecksumAddress:
    """
    https://github.com/ethereum/EIPs/blob/master/EIPS/eip-55.md

    :param norm_address: address in lowercase (not 0x prefixed)
    :param address_hash: keccak256 of `norm_address` (not 0x prefixed)
    :return:
    """
    return ChecksumAddress(
        HexAddress(
            HexStr(
                "0x"
                + (
                    "".join(
                        (
                            norm_address[i].upper()
                            if int(address_hash[i], 16) > 7
                            else norm_address[i]
                        )
                        for i in range(0, 40)
                    )
                )
            )
        )
    )


@lru_cache(maxsize=int(os.getenv("CACHE_CHECKSUM_ADDRESS", 1_000_000_000)))
def _fast_to_checksum_address(address: HexAddress):
    address_hash = fast_keccak_hex(address.encode())
    return _build_checksum_address(address, address_hash)


def fast_to_checksum_address(value: Union[AnyAddress, str, bytes]) -> ChecksumAddress:
    """
    Converts to checksum_address. Uses more optimal `pysha3` instead of `eth_utils` for keccak256 calculation

    :param value:
    :return:
    """
    if isinstance(value, bytes):
        if len(value) != 20:
            raise ValueError(
                "Cannot convert %s to a checksum address, 20 bytes were expected"
            )

    norm_address = HexAddress(HexStr(to_normalized_address(value)[2:]))
    return _fast_to_checksum_address(norm_address)


def fast_bytes_to_checksum_address(value: bytes) -> ChecksumAddress:
    """
    Converts to checksum_address. Uses more optimal `pysha3` instead of `eth_utils` for keccak256 calculation.
    As input is already in bytes, some checks and conversions can be skipped, providing a speedup of ~50%

    :param value:
    :return:
    """
    if len(value) != 20:
        raise ValueError(
            "Cannot convert %s to a checksum address, 20 bytes were expected"
        )
    norm_address = HexAddress(HexStr(bytes(value).hex()))
    return _fast_to_checksum_address(norm_address)


def fast_is_checksum_address(value: Union[AnyAddress, str, bytes]) -> bool:
    """
    Fast version to check if an address is a checksum_address

    :param value:
    :return: `True` if checksummed, `False` otherwise
    """
    if not isinstance(value, str) or len(value) != 42 or not value.startswith("0x"):
        return False
    try:
        return fast_to_checksum_address(value) == value
    except ValueError:
        return False


def get_eth_address_with_invalid_checksum() -> str:
    address = Account.create().address
    return "0x" + "".join(
        [c.lower() if c.isupper() else c.upper() for c in address[2:]]
    )


def decode_string_or_bytes32(data: bytes) -> str:
    try:
        return eth_abi.decode(["string"], data)[0]
    except (OverflowError, eth_abi.exceptions.DecodingError):
        name = eth_abi.decode(["bytes32"], data)[0]
        end_position = name.find(b"\x00")
        if end_position == -1:
            return name.decode()
        else:
            return name[:end_position].decode()


def remove_swarm_metadata(code: bytes) -> bytes:
    """
    Remove swarm metadata from Solidity bytecode

    :param code:
    :return: Code without metadata
    """
    swarm = b"\xa1\x65bzzr0"
    position = code.rfind(swarm)
    if position == -1:
        raise ValueError("Swarm metadata not found in code %s" % code.hex())
    return code[:position]


def compare_byte_code(code_1: bytes, code_2: bytes) -> bool:
    """
    Compare code, removing swarm metadata if necessary

    :param code_1:
    :param code_2:
    :return: True if same code, False otherwise
    """
    if code_1 == code_2:
        return True
    else:
        codes = []
        for code in (code_1, code_2):
            try:
                codes.append(remove_swarm_metadata(code))
            except ValueError:
                codes.append(code)

        return codes[0] == codes[1]


def mk_contract_address(address: Union[str, bytes], nonce: int) -> ChecksumAddress:
    """
    Generate expected contract address when using EVM CREATE

    :param address:
    :param nonce:
    :return:
    """
    return fast_to_checksum_address(
        generate_contract_address(Address(HexBytes(address)), nonce)
    )


def mk_contract_address_2(
    from_: Union[ChecksumAddress, bytes],
    salt: Union[HexStr, bytes],
    init_code: Union[HexStr, bytes],
) -> ChecksumAddress:
    """
    Generate expected contract address when using EVM CREATE2.

    :param from_: The address which is creating this new address (need to be 20 bytes)
    :param salt: A salt (32 bytes)
    :param init_code: A init code of the contract being created
    :return: Address of the new contract
    """

    from_ = HexBytes(from_)
    salt = HexBytes(salt)
    init_code = HexBytes(init_code)

    assert len(from_) == 20, f"Address {from_.hex()} is not valid. Must be 20 bytes"
    assert len(salt) == 32, f"Salt {salt.hex()} is not valid. Must be 32 bytes"
    assert len(init_code) > 0, f"Init code {init_code.hex()} is not valid"

    init_code_hash = fast_keccak(init_code)
    contract_address = fast_keccak(HexBytes("ff") + from_ + salt + init_code_hash)
    return fast_bytes_to_checksum_address(contract_address[12:])


def bytes_to_float(value: Any) -> float:
    """
    Convert a value of type Any to float.

    :param value: The value to convert.
    :return: The converted float value.
    :raises ValueError: If the value cannot be converted to float.
    """
    assert value is not None, "Cannot convert None to float"
    if isinstance(value, (int, float)):
        return float(value)
    elif isinstance(value, bytes):
        try:
            return float(int.from_bytes(value, "big"))
        except (ValueError, OverflowError) as e:
            raise ValueError(f"Cannot convert bytes to float: {e}")
    else:
        raise ValueError(f"Unsupported type for conversion to float: {type(value)}")

1	import os	4✔
2	from functools import lru_cache	4✔
3	from typing import Any, Union	4✔
4
5	import eth_abi	4✔
6	from eth._utils.address import generate_contract_address	4✔
7	from eth_account import Account	4✔
8	from eth_typing import Address, AnyAddress, ChecksumAddress, Hash32, HexAddress, HexStr	4✔
9	from eth_utils import to_normalized_address	4✔
10	from hexbytes import HexBytes	4✔
11	from sha3 import keccak_256	4✔
12	from web3.types import TxParams, Wei	4✔
13
14
15	def get_empty_tx_params() -> TxParams:	4✔
16	"""
17	:return: Empty tx params, so calls like `build_transaction` don't call the RPC trying to get information
18	"""
19	return {	4✔
20	"gas": Wei(1),
21	"gasPrice": Wei(1),
22	}
23
24
25	@lru_cache(maxsize=int(os.getenv("CACHE_KECCAK", 512)))	4✔
26	def _keccak_256(value: bytes) -> keccak_256:	4✔
27	return keccak_256(value)	4✔
28
29
30	def fast_keccak(value: bytes) -> Hash32:	4✔
31	"""
32	Calculates ethereum keccak256 using fast library `pysha3`
33
34	:param value:
35	:return: Keccak256 used by ethereum as `HexBytes`
36	"""
37	return Hash32(HexBytes(_keccak_256(value).digest()))	4✔
38
39
40	def fast_keccak_text(value: str) -> Hash32:	4✔
41	"""
42	Calculates ethereum keccak256 using fast library `pysha3`
43
44	:param value:
45	:return: Keccak256 used by ethereum as `HexBytes`
46	"""
47	return fast_keccak(value.encode())	4✔
48
49
50	def fast_keccak_hex(value: bytes) -> HexStr:	4✔
51	"""
52	Same as `fast_keccak`, but it's a little more optimal calling `hexdigest()`
53	than calling `digest()` and then `hex()`
54
55	:param value:
56	:return: Keccak256 used by ethereum as a hex string (not 0x prefixed)
57	"""
58	return HexStr(_keccak_256(value).hexdigest())	4✔
59
60
61	def _build_checksum_address(	4✔
62	norm_address: HexStr, address_hash: HexStr
63	) -> ChecksumAddress:
64	"""
65	https://github.com/ethereum/EIPs/blob/master/EIPS/eip-55.md
66
67	:param norm_address: address in lowercase (not 0x prefixed)
68	:param address_hash: keccak256 of `norm_address` (not 0x prefixed)
69	:return:
70	"""
71	return ChecksumAddress(	4✔
72	HexAddress(
73	HexStr(
74	"0x"
75	+ (
76	"".join(
77	(
78	norm_address[i].upper()
79	if int(address_hash[i], 16) > 7
80	else norm_address[i]
81	)
82	for i in range(0, 40)
83	)
84	)
85	)
86	)
87	)
88
89
90	@lru_cache(maxsize=int(os.getenv("CACHE_CHECKSUM_ADDRESS", 1_000_000_000)))	4✔
91	def _fast_to_checksum_address(address: HexAddress):	4✔
92	address_hash = fast_keccak_hex(address.encode())	4✔
93	return _build_checksum_address(address, address_hash)	4✔
94
95
96	def fast_to_checksum_address(value: Union[AnyAddress, str, bytes]) -> ChecksumAddress:	4✔
97	"""
98	Converts to checksum_address. Uses more optimal `pysha3` instead of `eth_utils` for keccak256 calculation
99
100	:param value:
101	:return:
102	"""
103	if isinstance(value, bytes):	4✔
104	if len(value) != 20:	4✔
105	raise ValueError(	×
106	"Cannot convert %s to a checksum address, 20 bytes were expected"
107	)
108
109	norm_address = HexAddress(HexStr(to_normalized_address(value)[2:]))	4✔
110	return _fast_to_checksum_address(norm_address)	4✔
111
112
113	def fast_bytes_to_checksum_address(value: bytes) -> ChecksumAddress:	4✔
114	"""
115	Converts to checksum_address. Uses more optimal `pysha3` instead of `eth_utils` for keccak256 calculation.
116	As input is already in bytes, some checks and conversions can be skipped, providing a speedup of ~50%
117
118	:param value:
119	:return:
120	"""
121	if len(value) != 20:	4✔
122	raise ValueError(	4✔
123	"Cannot convert %s to a checksum address, 20 bytes were expected"
124	)
125	norm_address = HexAddress(HexStr(bytes(value).hex()))	4✔
126	return _fast_to_checksum_address(norm_address)	4✔
127
128
129	def fast_is_checksum_address(value: Union[AnyAddress, str, bytes]) -> bool:	4✔
130	"""
131	Fast version to check if an address is a checksum_address
132
133	:param value:
134	:return: `True` if checksummed, `False` otherwise
135	"""
136	if not isinstance(value, str) or len(value) != 42 or not value.startswith("0x"):	4✔
137	return False	4✔
138	try:	4✔
139	return fast_to_checksum_address(value) == value	4✔
140	except ValueError:	×
141	return False	×
142
143
144	def get_eth_address_with_invalid_checksum() -> str:	4✔
145	address = Account.create().address	4✔
146	return "0x" + "".join(	4✔
147	[c.lower() if c.isupper() else c.upper() for c in address[2:]]
148	)
149
150
151	def decode_string_or_bytes32(data: bytes) -> str:	4✔
152	try:	4✔
153	return eth_abi.decode(["string"], data)[0]	4✔
154	except (OverflowError, eth_abi.exceptions.DecodingError):	4✔
155	name = eth_abi.decode(["bytes32"], data)[0]	4✔
156	end_position = name.find(b"\x00")	4✔
157	if end_position == -1:	4✔
158	return name.decode()	×
159	else:
160	return name[:end_position].decode()	4✔
161
162
163	def remove_swarm_metadata(code: bytes) -> bytes:	4✔
164	"""
165	Remove swarm metadata from Solidity bytecode
166
167	:param code:
168	:return: Code without metadata
169	"""
170	swarm = b"\xa1\x65bzzr0"	4✔
171	position = code.rfind(swarm)	4✔
172	if position == -1:	4✔
173	raise ValueError("Swarm metadata not found in code %s" % code.hex())	4✔
174	return code[:position]	4✔
175
176
177	def compare_byte_code(code_1: bytes, code_2: bytes) -> bool:	4✔
178	"""
179	Compare code, removing swarm metadata if necessary
180
181	:param code_1:
182	:param code_2:
183	:return: True if same code, False otherwise
184	"""
185	if code_1 == code_2:	4✔
186	return True	4✔
187	else:
188	codes = []	4✔
189	for code in (code_1, code_2):	4✔
190	try:	4✔
191	codes.append(remove_swarm_metadata(code))	4✔
192	except ValueError:	4✔
193	codes.append(code)	4✔
194
195	return codes[0] == codes[1]	4✔
196
197
198	def mk_contract_address(address: Union[str, bytes], nonce: int) -> ChecksumAddress:	4✔
199	"""
200	Generate expected contract address when using EVM CREATE
201
202	:param address:
203	:param nonce:
204	:return:
205	"""
206	return fast_to_checksum_address(	4✔
207	generate_contract_address(Address(HexBytes(address)), nonce)
208	)
209
210
211	def mk_contract_address_2(	4✔
212	from_: Union[ChecksumAddress, bytes],
213	salt: Union[HexStr, bytes],
214	init_code: Union[HexStr, bytes],
215	) -> ChecksumAddress:
216	"""
217	Generate expected contract address when using EVM CREATE2.
218
219	:param from_: The address which is creating this new address (need to be 20 bytes)
220	:param salt: A salt (32 bytes)
221	:param init_code: A init code of the contract being created
222	:return: Address of the new contract
223	"""
224
225	from_ = HexBytes(from_)	4✔
226	salt = HexBytes(salt)	4✔
227	init_code = HexBytes(init_code)	4✔
228
229	assert len(from_) == 20, f"Address {from_.hex()} is not valid. Must be 20 bytes"	4✔
230	assert len(salt) == 32, f"Salt {salt.hex()} is not valid. Must be 32 bytes"	4✔
231	assert len(init_code) > 0, f"Init code {init_code.hex()} is not valid"	4✔
232
233	init_code_hash = fast_keccak(init_code)	4✔
234	contract_address = fast_keccak(HexBytes("ff") + from_ + salt + init_code_hash)	4✔
235	return fast_bytes_to_checksum_address(contract_address[12:])	4✔
236
237
238	def bytes_to_float(value: Any) -> float:	4✔
239	"""
240	Convert a value of type Any to float.
241
242	:param value: The value to convert.
243	:return: The converted float value.
244	:raises ValueError: If the value cannot be converted to float.
245	"""
246	assert value is not None, "Cannot convert None to float"	4✔
247	if isinstance(value, (int, float)):	4✔
248	return float(value)	4✔
249	elif isinstance(value, bytes):	×
250	try:	×
251	return float(int.from_bytes(value, "big"))	×
252	except (ValueError, OverflowError) as e:	×
253	raise ValueError(f"Cannot convert bytes to float: {e}")	×
254	else:
255	raise ValueError(f"Unsupported type for conversion to float: {type(value)}")	×

safe-global / safe-eth-py / 10793540350

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous