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

88.66

/safe_eth/eth/account_abstraction/user_operation.py

import dataclasses
from functools import cached_property
from typing import Any, Dict, Optional, Union

from eth_abi import encode as abi_encode
from eth_typing import ChecksumAddress, HexStr
from hexbytes import HexBytes

from safe_eth.eth.utils import fast_keccak, fast_to_checksum_address


@dataclasses.dataclass(eq=True, frozen=True)
class UserOperationMetadata:
    transaction_hash: bytes
    block_hash: bytes
    block_number: int


@dataclasses.dataclass(eq=True, frozen=True)
class UserOperation:
    """
    EIP4337 UserOperation for Entrypoint v0.6

    https://github.com/eth-infinitism/account-abstraction/blob/v0.6.0/contracts/interfaces/UserOperation.sol
    """

    user_operation_hash: bytes
    sender: ChecksumAddress
    nonce: int
    init_code: bytes
    call_data: bytes
    call_gas_limit: int
    verification_gas_limit: int
    pre_verification_gas: int
    max_fee_per_gas: int
    max_priority_fee_per_gas: int
    paymaster_and_data: bytes
    signature: bytes
    entry_point: ChecksumAddress
    metadata: Optional[UserOperationMetadata] = None

    @classmethod
    def from_bundler_response(
        cls,
        user_operation_hash: Union[HexStr, bytes],
        user_operation_response: Dict[str, Any],
    ) -> Union["UserOperation", "UserOperationV07"]:
        user_operation = user_operation_response["userOperation"]
        metadata = UserOperationMetadata(
            HexBytes(user_operation_response["transactionHash"]),
            HexBytes(user_operation_response["blockHash"]),
            int(user_operation_response["blockNumber"], 16),
        )
        if "initCode" in user_operation:
            return cls(
                HexBytes(user_operation_hash),
                ChecksumAddress(user_operation["sender"]),
                int(user_operation["nonce"], 16),
                HexBytes(user_operation["initCode"]),
                HexBytes(user_operation["callData"]),
                int(user_operation["callGasLimit"], 16),
                int(user_operation["verificationGasLimit"], 16),
                int(user_operation["preVerificationGas"], 16),
                int(user_operation["maxFeePerGas"], 16),
                int(user_operation["maxPriorityFeePerGas"], 16),
                HexBytes(user_operation["paymasterAndData"]),
                HexBytes(user_operation["signature"]),
                ChecksumAddress(user_operation_response["entryPoint"]),
                metadata=metadata,
            )
        else:
            if paymaster := user_operation.get("paymaster"):
                # Paymaster parameters are optional
                paymaster_verification_gas_limit = int(
                    user_operation["paymasterVerificationGasLimit"], 16
                )
                paymaster_post_op_gas_limit = int(
                    user_operation["paymasterPostOpGasLimit"], 16
                )
                paymaster_data = HexBytes(user_operation["paymasterData"])
            else:
                paymaster_verification_gas_limit = None
                paymaster_post_op_gas_limit = None
                paymaster_data = None

            return UserOperationV07(
                HexBytes(user_operation_hash),
                ChecksumAddress(user_operation["sender"]),
                int(user_operation["nonce"], 16),
                ChecksumAddress(user_operation["factory"]),
                HexBytes(user_operation["factoryData"]),
                HexBytes(user_operation["callData"]),
                int(user_operation["callGasLimit"], 16),
                int(user_operation["verificationGasLimit"], 16),
                int(user_operation["preVerificationGas"], 16),
                int(user_operation["maxPriorityFeePerGas"], 16),
                int(user_operation["maxFeePerGas"], 16),
                HexBytes(user_operation["signature"]),
                ChecksumAddress(user_operation_response["entryPoint"]),
                paymaster_verification_gas_limit,
                paymaster_post_op_gas_limit,
                paymaster,
                paymaster_data,
                metadata=metadata,
            )

    def __str__(self):
        return f"User Operation sender={self.sender} nonce={self.nonce} hash={self.user_operation_hash.hex()}"

    @cached_property
    def paymaster(self) -> Optional[ChecksumAddress]:
        if self.paymaster_and_data and len(self.paymaster_and_data) >= 20:
            return fast_to_checksum_address(self.paymaster_and_data[:20])
        return None

    @cached_property
    def paymaster_data(self) -> Optional[bytes]:
        if self.paymaster_and_data:
            return self.paymaster_and_data[20:]
        return None

    def calculate_user_operation_hash(self, chain_id: int) -> bytes:
        hash_init_code = fast_keccak(self.init_code)
        hash_call_data = fast_keccak(self.call_data)
        hash_paymaster_and_data = fast_keccak(self.paymaster_and_data)
        user_operation_encoded = abi_encode(
            [
                "address",
                "uint256",
                "bytes32",
                "bytes32",
                "uint256",
                "uint256",
                "uint256",
                "uint256",
                "uint256",
                "bytes32",
            ],
            [
                self.sender,
                self.nonce,
                hash_init_code,
                hash_call_data,
                self.call_gas_limit,
                self.verification_gas_limit,
                self.pre_verification_gas,
                self.max_fee_per_gas,
                self.max_priority_fee_per_gas,
                hash_paymaster_and_data,
            ],
        )
        return fast_keccak(
            abi_encode(
                ["bytes32", "address", "uint256"],
                [fast_keccak(user_operation_encoded), self.entry_point, chain_id],
            )
        )


@dataclasses.dataclass(eq=True, frozen=True)
class UserOperationV07:
    """
    EIP4337 UserOperation for Entrypoint v0.7

    https://github.com/eth-infinitism/account-abstraction/blob/v0.7.0/contracts/interfaces/PackedUserOperation.sol
    """

    user_operation_hash: bytes
    sender: ChecksumAddress
    nonce: int
    factory: ChecksumAddress
    factory_data: bytes
    call_data: bytes
    call_gas_limit: int
    verification_gas_limit: int
    pre_verification_gas: int
    max_priority_fee_per_gas: int
    max_fee_per_gas: int
    signature: bytes
    entry_point: ChecksumAddress
    paymaster_verification_gas_limit: Optional[int] = None
    paymaster_post_op_gas_limit: Optional[int] = None
    paymaster: Optional[bytes] = None
    paymaster_data: Optional[bytes] = None
    metadata: Optional[UserOperationMetadata] = None

    @property
    def account_gas_limits(self) -> bytes:
        """
        :return:Account Gas Limits is a `bytes32` in Solidity, first `bytes16` `verification_gas_limit` and then `call_gas_limit`
        """
        return HexBytes(self.verification_gas_limit).rjust(16, b"\x00") + HexBytes(
            self.call_gas_limit
        ).rjust(16, b"\x00")

    @property
    def gas_fees(self) -> bytes:
        """
        :return: Gas Fees is a `bytes32` in Solidity, first `bytes16` `verification_gas_limit` and then `call_gas_limit`
        """
        return HexBytes(self.max_priority_fee_per_gas).rjust(16, b"\x00") + HexBytes(
            self.max_fee_per_gas
        ).rjust(16, b"\x00")

    @property
    def paymaster_and_data(self) -> bytes:
        if (
            not self.paymaster
            or not self.paymaster_verification_gas_limit
            or not self.paymaster_post_op_gas_limit
            or not self.paymaster_data
        ):
            return b""
        return (
            HexBytes(self.paymaster).rjust(20, b"\x00")
            + HexBytes(self.paymaster_verification_gas_limit).rjust(16, b"\x00")
            + HexBytes(self.paymaster_post_op_gas_limit).rjust(16, b"\x00")
            + HexBytes(self.paymaster_data)
        )

    def calculate_user_operation_hash(self, chain_id: int) -> bytes:
        hash_init_code = fast_keccak(HexBytes(self.factory) + self.factory_data)
        hash_call_data = fast_keccak(self.call_data)
        hash_paymaster_and_data = fast_keccak(self.paymaster_and_data)
        user_operation_encoded = abi_encode(
            [
                "address",
                "uint256",
                "bytes32",
                "bytes32",
                "bytes32",
                "uint256",
                "bytes32",
                "bytes32",
            ],
            [
                self.sender,
                self.nonce,
                hash_init_code,
                hash_call_data,
                self.account_gas_limits,
                self.pre_verification_gas,
                self.gas_fees,
                hash_paymaster_and_data,
            ],
        )
        return fast_keccak(
            abi_encode(
                ["bytes32", "address", "uint256"],
                [fast_keccak(user_operation_encoded), self.entry_point, chain_id],
            )
        )

1	import dataclasses	4✔
2	from functools import cached_property	4✔
3	from typing import Any, Dict, Optional, Union	4✔
4
5	from eth_abi import encode as abi_encode	4✔
6	from eth_typing import ChecksumAddress, HexStr	4✔
7	from hexbytes import HexBytes	4✔
8
9	from safe_eth.eth.utils import fast_keccak, fast_to_checksum_address	4✔
10
11
12	@dataclasses.dataclass(eq=True, frozen=True)	4✔
13	class UserOperationMetadata:	4✔
14	transaction_hash: bytes	4✔
15	block_hash: bytes	4✔
16	block_number: int	4✔
17
18
19	@dataclasses.dataclass(eq=True, frozen=True)	4✔
20	class UserOperation:	4✔
21	"""
22	EIP4337 UserOperation for Entrypoint v0.6
23
24	https://github.com/eth-infinitism/account-abstraction/blob/v0.6.0/contracts/interfaces/UserOperation.sol
25	"""
26
27	user_operation_hash: bytes	4✔
28	sender: ChecksumAddress	4✔
29	nonce: int	4✔
30	init_code: bytes	4✔
31	call_data: bytes	4✔
32	call_gas_limit: int	4✔
33	verification_gas_limit: int	4✔
34	pre_verification_gas: int	4✔
35	max_fee_per_gas: int	4✔
36	max_priority_fee_per_gas: int	4✔
37	paymaster_and_data: bytes	4✔
38	signature: bytes	4✔
39	entry_point: ChecksumAddress	4✔
40	metadata: Optional[UserOperationMetadata] = None	4✔
41
42	@classmethod	4✔
43	def from_bundler_response(	4✔
44	cls,
45	user_operation_hash: Union[HexStr, bytes],
46	user_operation_response: Dict[str, Any],
47	) -> Union["UserOperation", "UserOperationV07"]:
48	user_operation = user_operation_response["userOperation"]	4✔
49	metadata = UserOperationMetadata(	4✔
50	HexBytes(user_operation_response["transactionHash"]),
51	HexBytes(user_operation_response["blockHash"]),
52	int(user_operation_response["blockNumber"], 16),
53	)
54	if "initCode" in user_operation:	4✔
55	return cls(	4✔
56	HexBytes(user_operation_hash),
57	ChecksumAddress(user_operation["sender"]),
58	int(user_operation["nonce"], 16),
59	HexBytes(user_operation["initCode"]),
60	HexBytes(user_operation["callData"]),
61	int(user_operation["callGasLimit"], 16),
62	int(user_operation["verificationGasLimit"], 16),
63	int(user_operation["preVerificationGas"], 16),
64	int(user_operation["maxFeePerGas"], 16),
65	int(user_operation["maxPriorityFeePerGas"], 16),
66	HexBytes(user_operation["paymasterAndData"]),
67	HexBytes(user_operation["signature"]),
68	ChecksumAddress(user_operation_response["entryPoint"]),
69	metadata=metadata,
70	)
71	else:
72	if paymaster := user_operation.get("paymaster"):	4✔
73	# Paymaster parameters are optional
74	paymaster_verification_gas_limit = int(	×
75	user_operation["paymasterVerificationGasLimit"], 16
76	)
77	paymaster_post_op_gas_limit = int(	×
78	user_operation["paymasterPostOpGasLimit"], 16
79	)
80	paymaster_data = HexBytes(user_operation["paymasterData"])	×
81	else:
82	paymaster_verification_gas_limit = None	4✔
83	paymaster_post_op_gas_limit = None	4✔
84	paymaster_data = None	4✔
85
86	return UserOperationV07(	4✔
87	HexBytes(user_operation_hash),
88	ChecksumAddress(user_operation["sender"]),
89	int(user_operation["nonce"], 16),
90	ChecksumAddress(user_operation["factory"]),
91	HexBytes(user_operation["factoryData"]),
92	HexBytes(user_operation["callData"]),
93	int(user_operation["callGasLimit"], 16),
94	int(user_operation["verificationGasLimit"], 16),
95	int(user_operation["preVerificationGas"], 16),
96	int(user_operation["maxPriorityFeePerGas"], 16),
97	int(user_operation["maxFeePerGas"], 16),
98	HexBytes(user_operation["signature"]),
99	ChecksumAddress(user_operation_response["entryPoint"]),
100	paymaster_verification_gas_limit,
101	paymaster_post_op_gas_limit,
102	paymaster,
103	paymaster_data,
104	metadata=metadata,
105	)
106
107	def __str__(self):	4✔
108	return f"User Operation sender={self.sender} nonce={self.nonce} hash={self.user_operation_hash.hex()}"	×
109
110	@cached_property	4✔
111	def paymaster(self) -> Optional[ChecksumAddress]:	4✔
112	if self.paymaster_and_data and len(self.paymaster_and_data) >= 20:	×
113	return fast_to_checksum_address(self.paymaster_and_data[:20])	×
114	return None	×
115
116	@cached_property	4✔
117	def paymaster_data(self) -> Optional[bytes]:	4✔
118	if self.paymaster_and_data:	×
119	return self.paymaster_and_data[20:]	×
120	return None	×
121
122	def calculate_user_operation_hash(self, chain_id: int) -> bytes:	4✔
123	hash_init_code = fast_keccak(self.init_code)	4✔
124	hash_call_data = fast_keccak(self.call_data)	4✔
125	hash_paymaster_and_data = fast_keccak(self.paymaster_and_data)	4✔
126	user_operation_encoded = abi_encode(	4✔
127	[
128	"address",
129	"uint256",
130	"bytes32",
131	"bytes32",
132	"uint256",
133	"uint256",
134	"uint256",
135	"uint256",
136	"uint256",
137	"bytes32",
138	],
139	[
140	self.sender,
141	self.nonce,
142	hash_init_code,
143	hash_call_data,
144	self.call_gas_limit,
145	self.verification_gas_limit,
146	self.pre_verification_gas,
147	self.max_fee_per_gas,
148	self.max_priority_fee_per_gas,
149	hash_paymaster_and_data,
150	],
151	)
152	return fast_keccak(	4✔
153	abi_encode(
154	["bytes32", "address", "uint256"],
155	[fast_keccak(user_operation_encoded), self.entry_point, chain_id],
156	)
157	)
158
159
160	@dataclasses.dataclass(eq=True, frozen=True)	4✔
161	class UserOperationV07:	4✔
162	"""
163	EIP4337 UserOperation for Entrypoint v0.7
164
165	https://github.com/eth-infinitism/account-abstraction/blob/v0.7.0/contracts/interfaces/PackedUserOperation.sol
166	"""
167
168	user_operation_hash: bytes	4✔
169	sender: ChecksumAddress	4✔
170	nonce: int	4✔
171	factory: ChecksumAddress	4✔
172	factory_data: bytes	4✔
173	call_data: bytes	4✔
174	call_gas_limit: int	4✔
175	verification_gas_limit: int	4✔
176	pre_verification_gas: int	4✔
177	max_priority_fee_per_gas: int	4✔
178	max_fee_per_gas: int	4✔
179	signature: bytes	4✔
180	entry_point: ChecksumAddress	4✔
181	paymaster_verification_gas_limit: Optional[int] = None	4✔
182	paymaster_post_op_gas_limit: Optional[int] = None	4✔
183	paymaster: Optional[bytes] = None	4✔
184	paymaster_data: Optional[bytes] = None	4✔
185	metadata: Optional[UserOperationMetadata] = None	4✔
186
187	@property	4✔
188	def account_gas_limits(self) -> bytes:	4✔
189	"""
190	:return:Account Gas Limits is a `bytes32` in Solidity, first `bytes16` `verification_gas_limit` and then `call_gas_limit`
191	"""
192	return HexBytes(self.verification_gas_limit).rjust(16, b"\x00") + HexBytes(	4✔
193	self.call_gas_limit
194	).rjust(16, b"\x00")
195
196	@property	4✔
197	def gas_fees(self) -> bytes:	4✔
198	"""
199	:return: Gas Fees is a `bytes32` in Solidity, first `bytes16` `verification_gas_limit` and then `call_gas_limit`
200	"""
201	return HexBytes(self.max_priority_fee_per_gas).rjust(16, b"\x00") + HexBytes(	4✔
202	self.max_fee_per_gas
203	).rjust(16, b"\x00")
204
205	@property	4✔
206	def paymaster_and_data(self) -> bytes:	4✔
207	if (	4✔
208	not self.paymaster
209	or not self.paymaster_verification_gas_limit
210	or not self.paymaster_post_op_gas_limit
211	or not self.paymaster_data
212	):
213	return b""	4✔
214	return (	×
215	HexBytes(self.paymaster).rjust(20, b"\x00")
216	+ HexBytes(self.paymaster_verification_gas_limit).rjust(16, b"\x00")
217	+ HexBytes(self.paymaster_post_op_gas_limit).rjust(16, b"\x00")
218	+ HexBytes(self.paymaster_data)
219	)
220
221	def calculate_user_operation_hash(self, chain_id: int) -> bytes:	4✔
222	hash_init_code = fast_keccak(HexBytes(self.factory) + self.factory_data)	4✔
223	hash_call_data = fast_keccak(self.call_data)	4✔
224	hash_paymaster_and_data = fast_keccak(self.paymaster_and_data)	4✔
225	user_operation_encoded = abi_encode(	4✔
226	[
227	"address",
228	"uint256",
229	"bytes32",
230	"bytes32",
231	"bytes32",
232	"uint256",
233	"bytes32",
234	"bytes32",
235	],
236	[
237	self.sender,
238	self.nonce,
239	hash_init_code,
240	hash_call_data,
241	self.account_gas_limits,
242	self.pre_verification_gas,
243	self.gas_fees,
244	hash_paymaster_and_data,
245	],
246	)
247	return fast_keccak(	4✔
248	abi_encode(
249	["bytes32", "address", "uint256"],
250	[fast_keccak(user_operation_encoded), self.entry_point, chain_id],
251	)
252	)

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