10629724120

Committed 30 Aug 2024 08:49AM UTC coverage: 93.903%. Remained the same

Build # 10629724120

Build Type

Pull #1309

github

Committed by

web-flow

Commit Message

Merge 175bdc70e into d0479e88e

Pull Request Pull Request #1309: Add reference to main in safe-eth-py dependency in addresses actions

Run Details

8671 of 9234 relevant lines covered (93.9%)

3.76 hits per line

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

91.61

/safe_eth/safe/safe_tx.py

from functools import cached_property
from typing import Any, Dict, List, NoReturn, Optional, Tuple, Type

from eth_account import Account
from eth_typing import ChecksumAddress
from hexbytes import HexBytes
from packaging.version import Version
from web3.contract.contract import ContractFunction
from web3.exceptions import Web3Exception
from web3.types import BlockIdentifier, TxParams, Wei

from safe_eth.eth import EthereumClient
from safe_eth.eth.constants import NULL_ADDRESS
from safe_eth.eth.contracts import get_safe_contract
from safe_eth.eth.eip712 import eip712_encode
from safe_eth.eth.ethereum_client import TxSpeed

from ..eth.utils import fast_keccak
from .exceptions import (
    CouldNotFinishInitialization,
    CouldNotPayGasWithEther,
    CouldNotPayGasWithToken,
    HashHasNotBeenApproved,
    InvalidContractSignatureLocation,
    InvalidInternalTx,
    InvalidMultisigTx,
    InvalidOwnerProvided,
    InvalidSignaturesProvided,
    MethodCanOnlyBeCalledFromThisContract,
    ModuleManagerException,
    NotEnoughSafeTransactionGas,
    OnlyOwnersCanApproveAHash,
    OwnerManagerException,
    SafeTransactionFailedWhenGasPriceAndSafeTxGasEmpty,
    SignatureNotProvidedByOwner,
    SignaturesDataTooShort,
    ThresholdNeedsToBeDefined,
)
from .safe_signature import SafeSignature
from .signatures import signature_to_bytes


class SafeTx:
    def __init__(
        self,
        ethereum_client: EthereumClient,
        safe_address: ChecksumAddress,
        to: Optional[ChecksumAddress],
        value: int,
        data: bytes,
        operation: int,
        safe_tx_gas: int,
        base_gas: int,
        gas_price: int,
        gas_token: Optional[ChecksumAddress],
        refund_receiver: Optional[ChecksumAddress],
        signatures: Optional[bytes] = None,
        safe_nonce: Optional[int] = None,
        safe_version: Optional[str] = None,
        chain_id: Optional[int] = None,
    ):
        """
        :param ethereum_client:
        :param safe_address:
        :param to:
        :param value:
        :param data:
        :param operation:
        :param safe_tx_gas:
        :param base_gas:
        :param gas_price:
        :param gas_token:
        :param refund_receiver:
        :param signatures:
        :param safe_nonce: Current nonce of the Safe. If not provided, it will be retrieved from network
        :param safe_version: Safe version 1.0.0 renamed `baseGas` to `dataGas`. Safe version 1.3.0 added `chainId` to
        the `domainSeparator`. If not provided, it will be retrieved from network
        :param chain_id: Ethereum network chain_id is used in hash calculation for Safes >= 1.3.0. If not provided,
        it will be retrieved from the provided ethereum_client
        """

        self.ethereum_client = ethereum_client
        self.safe_address = safe_address
        self.to = to or NULL_ADDRESS
        self.value = int(value)
        self.data = HexBytes(data) if data else b""
        self.operation = int(operation)
        self.safe_tx_gas = int(safe_tx_gas)
        self.base_gas = int(base_gas)
        self.gas_price = int(gas_price)
        self.gas_token = gas_token or NULL_ADDRESS
        self.refund_receiver = refund_receiver or NULL_ADDRESS
        self.signatures = signatures or b""
        self._safe_nonce = safe_nonce and int(safe_nonce)
        self._safe_version = safe_version
        self._chain_id = chain_id and int(chain_id)

        self.tx: Optional[TxParams] = None  # If executed, `tx` is set
        self.tx_hash: Optional[bytes] = None  # If executed, `tx_hash` is set

    def __str__(self):
        return (
            f"SafeTx - safe={self.safe_address} - to={self.to} - value={self.value} - data={self.data.hex()} - "
            f"operation={self.operation} - safe-tx-gas={self.safe_tx_gas} - base-gas={self.base_gas} - "
            f"gas-price={self.gas_price} - gas-token={self.gas_token} - refund-receiver={self.refund_receiver} - "
            f"signers = {self.signers}"
        )

    @property
    def w3(self):
        return self.ethereum_client.w3

    @cached_property
    def contract(self):
        return get_safe_contract(self.w3, address=self.safe_address)

    @cached_property
    def chain_id(self) -> int:
        if self._chain_id is not None:
            return self._chain_id
        else:
            return self.ethereum_client.get_chain_id()

    @cached_property
    def safe_nonce(self) -> int:
        if self._safe_nonce is not None:
            return self._safe_nonce
        else:
            return self.contract.functions.nonce().call()

    @cached_property
    def safe_version(self) -> str:
        if self._safe_version is not None:
            return self._safe_version
        else:
            return self.contract.functions.VERSION().call()

    @property
    def eip712_structured_data(self) -> Dict[str, Any]:
        safe_version = Version(self.safe_version)

        # Safes >= 1.0.0 Renamed `baseGas` to `dataGas`
        base_gas_key = "baseGas" if safe_version >= Version("1.0.0") else "dataGas"

        types = {
            "EIP712Domain": [{"name": "verifyingContract", "type": "address"}],
            "SafeTx": [
                {"name": "to", "type": "address"},
                {"name": "value", "type": "uint256"},
                {"name": "data", "type": "bytes"},
                {"name": "operation", "type": "uint8"},
                {"name": "safeTxGas", "type": "uint256"},
                {"name": base_gas_key, "type": "uint256"},
                {"name": "gasPrice", "type": "uint256"},
                {"name": "gasToken", "type": "address"},
                {"name": "refundReceiver", "type": "address"},
                {"name": "nonce", "type": "uint256"},
            ],
        }
        message = {
            "to": self.to,
            "value": self.value,
            "data": self.data,
            "operation": self.operation,
            "safeTxGas": self.safe_tx_gas,
            base_gas_key: self.base_gas,
            "dataGas": self.base_gas,
            "gasPrice": self.gas_price,
            "gasToken": self.gas_token,
            "refundReceiver": self.refund_receiver,
            "nonce": self.safe_nonce,
        }

        payload = {
            "types": types,
            "primaryType": "SafeTx",
            "domain": {"verifyingContract": self.safe_address},
            "message": message,
        }

        # Enable chainId from v1.3.0 onwards
        if safe_version >= Version("1.3.0"):
            payload["domain"]["chainId"] = self.chain_id
            types["EIP712Domain"].insert(0, {"name": "chainId", "type": "uint256"})

        return payload

    @property
    def safe_tx_hash_preimage(self) -> HexBytes:
        return HexBytes(b"".join(eip712_encode(self.eip712_structured_data)))

    @property
    def safe_tx_hash(self) -> HexBytes:
        return HexBytes(fast_keccak(self.safe_tx_hash_preimage))

    @property
    def signers(self) -> List[str]:
        if not self.signatures:
            return []
        else:
            return [
                safe_signature.owner
                for safe_signature in SafeSignature.parse_signature(
                    self.signatures, self.safe_tx_hash
                )
            ]

    @property
    def sorted_signers(self):
        return sorted(self.signers, key=lambda x: int(x, 16))

    @property
    def w3_tx(self) -> ContractFunction:
        """
        :return: Web3 contract tx prepared for `call`, `transact` or `build_transaction`
        """
        return self.contract.functions.execTransaction(
            self.to,
            self.value,
            self.data,
            self.operation,
            self.safe_tx_gas,
            self.base_gas,
            self.gas_price,
            self.gas_token,
            self.refund_receiver,
            self.signatures,
        )

    def _raise_safe_vm_exception(self, message: str) -> NoReturn:
        error_with_exception: Dict[str, Type[InvalidMultisigTx]] = {
            # https://github.com/safe-global/safe-contracts/blob/v1.3.0/docs/error_codes.md
            "GS000": CouldNotFinishInitialization,
            "GS001": ThresholdNeedsToBeDefined,
            "Could not pay gas costs with ether": CouldNotPayGasWithEther,
            "GS011": CouldNotPayGasWithEther,
            "Could not pay gas costs with token": CouldNotPayGasWithToken,
            "GS012": CouldNotPayGasWithToken,
            "GS013": SafeTransactionFailedWhenGasPriceAndSafeTxGasEmpty,
            "Hash has not been approved": HashHasNotBeenApproved,
            "Hash not approved": HashHasNotBeenApproved,
            "GS025": HashHasNotBeenApproved,
            "Invalid contract signature location: data not complete": InvalidContractSignatureLocation,
            "GS023": InvalidContractSignatureLocation,
            "Invalid contract signature location: inside static part": InvalidContractSignatureLocation,
            "GS021": InvalidContractSignatureLocation,
            "Invalid contract signature location: length not present": InvalidContractSignatureLocation,
            "GS022": InvalidContractSignatureLocation,
            "Invalid contract signature provided": InvalidContractSignatureLocation,
            "GS024": InvalidContractSignatureLocation,
            "Invalid owner provided": InvalidOwnerProvided,
            "Invalid owner address provided": InvalidOwnerProvided,
            "GS026": InvalidOwnerProvided,
            "Invalid signatures provided": InvalidSignaturesProvided,
            "Not enough gas to execute safe transaction": NotEnoughSafeTransactionGas,
            "GS010": NotEnoughSafeTransactionGas,
            "Only owners can approve a hash": OnlyOwnersCanApproveAHash,
            "GS030": OnlyOwnersCanApproveAHash,
            "GS031": MethodCanOnlyBeCalledFromThisContract,
            "Signature not provided by owner": SignatureNotProvidedByOwner,
            "Signatures data too short": SignaturesDataTooShort,
            "GS020": SignaturesDataTooShort,
            # ModuleManager
            "GS100": ModuleManagerException,
            "Invalid module address provided": ModuleManagerException,
            "GS101": ModuleManagerException,
            "GS102": ModuleManagerException,
            "Invalid prevModule, module pair provided": ModuleManagerException,
            "GS103": ModuleManagerException,
            "Method can only be called from an enabled module": ModuleManagerException,
            "GS104": ModuleManagerException,
            "Module has already been added": ModuleManagerException,
            # OwnerManager
            "Address is already an owner": OwnerManagerException,
            "GS200": OwnerManagerException,  # Owners have already been setup
            "GS201": OwnerManagerException,  # Threshold cannot exceed owner count
            "GS202": OwnerManagerException,  # Invalid owner address provided
            "GS203": OwnerManagerException,  # Invalid ower address provided
            "GS204": OwnerManagerException,  # Address is already an owner
            "GS205": OwnerManagerException,  # Invalid prevOwner, owner pair provided
            "Invalid prevOwner, owner pair provided": OwnerManagerException,
            "New owner count needs to be larger than new threshold": OwnerManagerException,
            "Threshold cannot exceed owner count": OwnerManagerException,
            "Threshold needs to be greater than 0": OwnerManagerException,
        }

        for reason, custom_exception in error_with_exception.items():
            if reason in message:
                raise custom_exception(message)
        raise InvalidMultisigTx(message)

    def call(
        self,
        tx_sender_address: Optional[str] = None,
        tx_gas: Optional[int] = None,
        block_identifier: Optional[BlockIdentifier] = "latest",
    ) -> int:
        """
        :param tx_sender_address:
        :param tx_gas: Force a gas limit
        :param block_identifier:
        :return: `1` if everything ok
        """
        parameters: Dict[str, Any] = {
            "from": tx_sender_address if tx_sender_address else self.safe_address
        }

        if tx_gas:
            parameters["gas"] = tx_gas
        try:
            success = self.w3_tx.call(parameters, block_identifier=block_identifier)

            if not success:
                raise InvalidInternalTx(
                    "Success bit is %d, should be equal to 1" % success
                )
            return success
        except (Web3Exception, ValueError) as exc:
            # e.g. web3.exceptions.ContractLogicError: execution reverted: Invalid owner provided
            return self._raise_safe_vm_exception(str(exc))
        except ValueError as exc:  # Parity
            """
            Parity throws a ValueError, e.g.
            {'code': -32015,
             'message': 'VM execution error.',
             'data': 'Reverted 0x08c379a0000000000000000000000000000000000000000000000000000000000000020000000000000000
                      000000000000000000000000000000000000000000000001b496e76616c6964207369676e6174757265732070726f7669
                      6465640000000000'
            }
            """
            error_dict = exc.args[0]
            data = error_dict.get("data")
            if data and isinstance(data, str) and "Reverted " in data:
                # Parity
                result = HexBytes(data.replace("Reverted ", ""))
                return self._raise_safe_vm_exception(str(result))
            else:
                raise exc

    def recommended_gas(self) -> Wei:
        """
        :return: Recommended gas to use on the ethereum_tx
        """
        return Wei(self.base_gas + self.safe_tx_gas + 75000)

    def execute(
        self,
        tx_sender_private_key: str,
        tx_gas: Optional[int] = None,
        tx_gas_price: Optional[int] = None,
        tx_nonce: Optional[int] = None,
        block_identifier: Optional[BlockIdentifier] = "latest",
        eip1559_speed: Optional[TxSpeed] = None,
    ) -> Tuple[HexBytes, TxParams]:
        """
        Send multisig tx to the Safe

        :param tx_sender_private_key: Sender private key
        :param tx_gas: Gas for the external tx. If not, `(safe_tx_gas + base_gas) * 2` will be used
        :param tx_gas_price: Gas price of the external tx. If not, `gas_price` will be used
        :param tx_nonce: Force nonce for `tx_sender`
        :param block_identifier: `latest` or `pending`
        :param eip1559_speed: If provided, use EIP1559 transaction
        :return: Tuple(tx_hash, tx)
        :raises: InvalidMultisigTx: If user tx cannot go through the Safe
        """

        sender_account = Account.from_key(tx_sender_private_key)
        if eip1559_speed and self.ethereum_client.is_eip1559_supported():
            tx_parameters = self.ethereum_client.set_eip1559_fees(
                {
                    "from": sender_account.address,
                },
                tx_speed=eip1559_speed,
            )
        else:
            tx_parameters = {
                "from": sender_account.address,
                "gasPrice": tx_gas_price or self.w3.eth.gas_price,
            }

        if tx_gas:
            tx_parameters["gas"] = tx_gas
        if tx_nonce is not None:
            tx_parameters["nonce"] = tx_nonce

        self.tx: TxParams = self.w3_tx.build_transaction(tx_parameters)
        self.tx["gas"] = Wei(
            tx_gas or (max(self.tx["gas"] + 75000, self.recommended_gas()))
        )

        self.tx_hash = self.ethereum_client.send_unsigned_transaction(
            self.tx,
            private_key=sender_account.key,
            retry=False if tx_nonce is not None else True,
            block_identifier=block_identifier,
        )

        # Set signatures empty after executing the tx. `Nonce` is increased even if it fails,
        # so signatures are not valid anymore
        self.signatures = b""
        return self.tx_hash, self.tx

    def sign(self, private_key: str) -> bytes:
        """
        {bytes32 r}{bytes32 s}{uint8 v}
        :param private_key:
        :return: Signature
        """
        account = Account.from_key(private_key)
        signature_dict = account.signHash(self.safe_tx_hash)
        signature = signature_to_bytes(
            signature_dict["v"], signature_dict["r"], signature_dict["s"]
        )

        # Insert signature sorted
        if account.address not in self.signers:
            unsorted_signatures = SafeSignature.parse_signature(
                self.signatures + signature, self.safe_tx_hash
            )
            self.signatures = SafeSignature.export_signatures(unsorted_signatures)

        return signature

    def unsign(self, address: str) -> bool:
        current_tx_signatures = SafeSignature.parse_signature(
            self.signatures, self.safe_tx_hash
        )
        filtered_tx_signatures = list(
            filter(lambda x: x.owner != address, current_tx_signatures)
        )
        if current_tx_signatures != filtered_tx_signatures:
            self.signatures = SafeSignature.export_signatures(filtered_tx_signatures)
            return True
        return False

1	from functools import cached_property	4✔
2	from typing import Any, Dict, List, NoReturn, Optional, Tuple, Type	4✔
3
4	from eth_account import Account	4✔
5	from eth_typing import ChecksumAddress	4✔
6	from hexbytes import HexBytes	4✔
7	from packaging.version import Version	4✔
8	from web3.contract.contract import ContractFunction	4✔
9	from web3.exceptions import Web3Exception	4✔
10	from web3.types import BlockIdentifier, TxParams, Wei	4✔
11
12	from safe_eth.eth import EthereumClient	4✔
13	from safe_eth.eth.constants import NULL_ADDRESS	4✔
14	from safe_eth.eth.contracts import get_safe_contract	4✔
15	from safe_eth.eth.eip712 import eip712_encode	4✔
16	from safe_eth.eth.ethereum_client import TxSpeed	4✔
17
18	from ..eth.utils import fast_keccak	4✔
19	from .exceptions import (	4✔
20	CouldNotFinishInitialization,
21	CouldNotPayGasWithEther,
22	CouldNotPayGasWithToken,
23	HashHasNotBeenApproved,
24	InvalidContractSignatureLocation,
25	InvalidInternalTx,
26	InvalidMultisigTx,
27	InvalidOwnerProvided,
28	InvalidSignaturesProvided,
29	MethodCanOnlyBeCalledFromThisContract,
30	ModuleManagerException,
31	NotEnoughSafeTransactionGas,
32	OnlyOwnersCanApproveAHash,
33	OwnerManagerException,
34	SafeTransactionFailedWhenGasPriceAndSafeTxGasEmpty,
35	SignatureNotProvidedByOwner,
36	SignaturesDataTooShort,
37	ThresholdNeedsToBeDefined,
38	)
39	from .safe_signature import SafeSignature	4✔
40	from .signatures import signature_to_bytes	4✔
41
42
43	class SafeTx:	4✔
44	def __init__(	4✔
45	self,
46	ethereum_client: EthereumClient,
47	safe_address: ChecksumAddress,
48	to: Optional[ChecksumAddress],
49	value: int,
50	data: bytes,
51	operation: int,
52	safe_tx_gas: int,
53	base_gas: int,
54	gas_price: int,
55	gas_token: Optional[ChecksumAddress],
56	refund_receiver: Optional[ChecksumAddress],
57	signatures: Optional[bytes] = None,
58	safe_nonce: Optional[int] = None,
59	safe_version: Optional[str] = None,
60	chain_id: Optional[int] = None,
61	):
62	"""
63	:param ethereum_client:
64	:param safe_address:
65	:param to:
66	:param value:
67	:param data:
68	:param operation:
69	:param safe_tx_gas:
70	:param base_gas:
71	:param gas_price:
72	:param gas_token:
73	:param refund_receiver:
74	:param signatures:
75	:param safe_nonce: Current nonce of the Safe. If not provided, it will be retrieved from network
76	:param safe_version: Safe version 1.0.0 renamed `baseGas` to `dataGas`. Safe version 1.3.0 added `chainId` to
77	the `domainSeparator`. If not provided, it will be retrieved from network
78	:param chain_id: Ethereum network chain_id is used in hash calculation for Safes >= 1.3.0. If not provided,
79	it will be retrieved from the provided ethereum_client
80	"""
81
82	self.ethereum_client = ethereum_client	4✔
83	self.safe_address = safe_address	4✔
84	self.to = to or NULL_ADDRESS	4✔
85	self.value = int(value)	4✔
86	self.data = HexBytes(data) if data else b""	4✔
87	self.operation = int(operation)	4✔
88	self.safe_tx_gas = int(safe_tx_gas)	4✔
89	self.base_gas = int(base_gas)	4✔
90	self.gas_price = int(gas_price)	4✔
91	self.gas_token = gas_token or NULL_ADDRESS	4✔
92	self.refund_receiver = refund_receiver or NULL_ADDRESS	4✔
93	self.signatures = signatures or b""	4✔
94	self._safe_nonce = safe_nonce and int(safe_nonce)	4✔
95	self._safe_version = safe_version	4✔
96	self._chain_id = chain_id and int(chain_id)	4✔
97
98	self.tx: Optional[TxParams] = None # If executed, `tx` is set	4✔
99	self.tx_hash: Optional[bytes] = None # If executed, `tx_hash` is set	4✔
100
101	def __str__(self):	4✔
102	return (	×
103	f"SafeTx - safe={self.safe_address} - to={self.to} - value={self.value} - data={self.data.hex()} - "
104	f"operation={self.operation} - safe-tx-gas={self.safe_tx_gas} - base-gas={self.base_gas} - "
105	f"gas-price={self.gas_price} - gas-token={self.gas_token} - refund-receiver={self.refund_receiver} - "
106	f"signers = {self.signers}"
107	)
108
109	@property	4✔
110	def w3(self):	4✔
111	return self.ethereum_client.w3	4✔
112
113	@cached_property	4✔
114	def contract(self):	4✔
115	return get_safe_contract(self.w3, address=self.safe_address)	4✔
116
117	@cached_property	4✔
118	def chain_id(self) -> int:	4✔
119	if self._chain_id is not None:	4✔
120	return self._chain_id	4✔
121	else:
122	return self.ethereum_client.get_chain_id()	4✔
123
124	@cached_property	4✔
125	def safe_nonce(self) -> int:	4✔
126	if self._safe_nonce is not None:	4✔
127	return self._safe_nonce	4✔
128	else:
129	return self.contract.functions.nonce().call()	×
130
131	@cached_property	4✔
132	def safe_version(self) -> str:	4✔
133	if self._safe_version is not None:	4✔
134	return self._safe_version	4✔
135	else:
136	return self.contract.functions.VERSION().call()	4✔
137
138	@property	4✔
139	def eip712_structured_data(self) -> Dict[str, Any]:	4✔
140	safe_version = Version(self.safe_version)	4✔
141
142	# Safes >= 1.0.0 Renamed `baseGas` to `dataGas`
143	base_gas_key = "baseGas" if safe_version >= Version("1.0.0") else "dataGas"	4✔
144
145	types = {	4✔
146	"EIP712Domain": [{"name": "verifyingContract", "type": "address"}],
147	"SafeTx": [
148	{"name": "to", "type": "address"},
149	{"name": "value", "type": "uint256"},
150	{"name": "data", "type": "bytes"},
151	{"name": "operation", "type": "uint8"},
152	{"name": "safeTxGas", "type": "uint256"},
153	{"name": base_gas_key, "type": "uint256"},
154	{"name": "gasPrice", "type": "uint256"},
155	{"name": "gasToken", "type": "address"},
156	{"name": "refundReceiver", "type": "address"},
157	{"name": "nonce", "type": "uint256"},
158	],
159	}
160	message = {	4✔
161	"to": self.to,
162	"value": self.value,
163	"data": self.data,
164	"operation": self.operation,
165	"safeTxGas": self.safe_tx_gas,
166	base_gas_key: self.base_gas,
167	"dataGas": self.base_gas,
168	"gasPrice": self.gas_price,
169	"gasToken": self.gas_token,
170	"refundReceiver": self.refund_receiver,
171	"nonce": self.safe_nonce,
172	}
173
174	payload = {	4✔
175	"types": types,
176	"primaryType": "SafeTx",
177	"domain": {"verifyingContract": self.safe_address},
178	"message": message,
179	}
180
181	# Enable chainId from v1.3.0 onwards
182	if safe_version >= Version("1.3.0"):	4✔
183	payload["domain"]["chainId"] = self.chain_id	4✔
184	types["EIP712Domain"].insert(0, {"name": "chainId", "type": "uint256"})	4✔
185
186	return payload	4✔
187
188	@property	4✔
189	def safe_tx_hash_preimage(self) -> HexBytes:	4✔
190	return HexBytes(b"".join(eip712_encode(self.eip712_structured_data)))	4✔
191
192	@property	4✔
193	def safe_tx_hash(self) -> HexBytes:	4✔
194	return HexBytes(fast_keccak(self.safe_tx_hash_preimage))	4✔
195
196	@property	4✔
197	def signers(self) -> List[str]:	4✔
198	if not self.signatures:	4✔
199	return []	4✔
200	else:
201	return [	4✔
202	safe_signature.owner
203	for safe_signature in SafeSignature.parse_signature(
204	self.signatures, self.safe_tx_hash
205	)
206	]
207
208	@property	4✔
209	def sorted_signers(self):	4✔
210	return sorted(self.signers, key=lambda x: int(x, 16))	4✔
211
212	@property	4✔
213	def w3_tx(self) -> ContractFunction:	4✔
214	"""
215	:return: Web3 contract tx prepared for `call`, `transact` or `build_transaction`
216	"""
217	return self.contract.functions.execTransaction(	4✔
218	self.to,
219	self.value,
220	self.data,
221	self.operation,
222	self.safe_tx_gas,
223	self.base_gas,
224	self.gas_price,
225	self.gas_token,
226	self.refund_receiver,
227	self.signatures,
228	)
229
230	def _raise_safe_vm_exception(self, message: str) -> NoReturn:	4✔
231	error_with_exception: Dict[str, Type[InvalidMultisigTx]] = {	4✔
232	# https://github.com/safe-global/safe-contracts/blob/v1.3.0/docs/error_codes.md
233	"GS000": CouldNotFinishInitialization,
234	"GS001": ThresholdNeedsToBeDefined,
235	"Could not pay gas costs with ether": CouldNotPayGasWithEther,
236	"GS011": CouldNotPayGasWithEther,
237	"Could not pay gas costs with token": CouldNotPayGasWithToken,
238	"GS012": CouldNotPayGasWithToken,
239	"GS013": SafeTransactionFailedWhenGasPriceAndSafeTxGasEmpty,
240	"Hash has not been approved": HashHasNotBeenApproved,
241	"Hash not approved": HashHasNotBeenApproved,
242	"GS025": HashHasNotBeenApproved,
243	"Invalid contract signature location: data not complete": InvalidContractSignatureLocation,
244	"GS023": InvalidContractSignatureLocation,
245	"Invalid contract signature location: inside static part": InvalidContractSignatureLocation,
246	"GS021": InvalidContractSignatureLocation,
247	"Invalid contract signature location: length not present": InvalidContractSignatureLocation,
248	"GS022": InvalidContractSignatureLocation,
249	"Invalid contract signature provided": InvalidContractSignatureLocation,
250	"GS024": InvalidContractSignatureLocation,
251	"Invalid owner provided": InvalidOwnerProvided,
252	"Invalid owner address provided": InvalidOwnerProvided,
253	"GS026": InvalidOwnerProvided,
254	"Invalid signatures provided": InvalidSignaturesProvided,
255	"Not enough gas to execute safe transaction": NotEnoughSafeTransactionGas,
256	"GS010": NotEnoughSafeTransactionGas,
257	"Only owners can approve a hash": OnlyOwnersCanApproveAHash,
258	"GS030": OnlyOwnersCanApproveAHash,
259	"GS031": MethodCanOnlyBeCalledFromThisContract,
260	"Signature not provided by owner": SignatureNotProvidedByOwner,
261	"Signatures data too short": SignaturesDataTooShort,
262	"GS020": SignaturesDataTooShort,
263	# ModuleManager
264	"GS100": ModuleManagerException,
265	"Invalid module address provided": ModuleManagerException,
266	"GS101": ModuleManagerException,
267	"GS102": ModuleManagerException,
268	"Invalid prevModule, module pair provided": ModuleManagerException,
269	"GS103": ModuleManagerException,
270	"Method can only be called from an enabled module": ModuleManagerException,
271	"GS104": ModuleManagerException,
272	"Module has already been added": ModuleManagerException,
273	# OwnerManager
274	"Address is already an owner": OwnerManagerException,
275	"GS200": OwnerManagerException, # Owners have already been setup
276	"GS201": OwnerManagerException, # Threshold cannot exceed owner count
277	"GS202": OwnerManagerException, # Invalid owner address provided
278	"GS203": OwnerManagerException, # Invalid ower address provided
279	"GS204": OwnerManagerException, # Address is already an owner
280	"GS205": OwnerManagerException, # Invalid prevOwner, owner pair provided
281	"Invalid prevOwner, owner pair provided": OwnerManagerException,
282	"New owner count needs to be larger than new threshold": OwnerManagerException,
283	"Threshold cannot exceed owner count": OwnerManagerException,
284	"Threshold needs to be greater than 0": OwnerManagerException,
285	}
286
287	for reason, custom_exception in error_with_exception.items():	4✔
288	if reason in message:	4✔
289	raise custom_exception(message)	4✔
290	raise InvalidMultisigTx(message)	×
291
292	def call(	4✔
293	self,
294	tx_sender_address: Optional[str] = None,
295	tx_gas: Optional[int] = None,
296	block_identifier: Optional[BlockIdentifier] = "latest",
297	) -> int:
298	"""
299	:param tx_sender_address:
300	:param tx_gas: Force a gas limit
301	:param block_identifier:
302	:return: `1` if everything ok
303	"""
304	parameters: Dict[str, Any] = {	4✔
305	"from": tx_sender_address if tx_sender_address else self.safe_address
306	}
307
308	if tx_gas:	4✔
309	parameters["gas"] = tx_gas	4✔
310	try:	4✔
311	success = self.w3_tx.call(parameters, block_identifier=block_identifier)	4✔
312
313	if not success:	4✔
314	raise InvalidInternalTx(	4✔
315	"Success bit is %d, should be equal to 1" % success
316	)
317	return success	4✔
318	except (Web3Exception, ValueError) as exc:	4✔
319	# e.g. web3.exceptions.ContractLogicError: execution reverted: Invalid owner provided
320	return self._raise_safe_vm_exception(str(exc))	4✔
321	except ValueError as exc: # Parity	4✔
322	"""
323	Parity throws a ValueError, e.g.
324	{'code': -32015,
325	'message': 'VM execution error.',
326	'data': 'Reverted 0x08c379a0000000000000000000000000000000000000000000000000000000000000020000000000000000
327	000000000000000000000000000000000000000000000001b496e76616c6964207369676e6174757265732070726f7669
328	6465640000000000'
329	}
330	"""
331	error_dict = exc.args[0]	×
332	data = error_dict.get("data")	×
333	if data and isinstance(data, str) and "Reverted " in data:	×
334	# Parity
335	result = HexBytes(data.replace("Reverted ", ""))	×
336	return self._raise_safe_vm_exception(str(result))	×
337	else:
338	raise exc	×
339
340	def recommended_gas(self) -> Wei:	4✔
341	"""
342	:return: Recommended gas to use on the ethereum_tx
343	"""
344	return Wei(self.base_gas + self.safe_tx_gas + 75000)	4✔
345
346	def execute(	4✔
347	self,
348	tx_sender_private_key: str,
349	tx_gas: Optional[int] = None,
350	tx_gas_price: Optional[int] = None,
351	tx_nonce: Optional[int] = None,
352	block_identifier: Optional[BlockIdentifier] = "latest",
353	eip1559_speed: Optional[TxSpeed] = None,
354	) -> Tuple[HexBytes, TxParams]:
355	"""
356	Send multisig tx to the Safe
357
358	:param tx_sender_private_key: Sender private key
359	:param tx_gas: Gas for the external tx. If not, `(safe_tx_gas + base_gas) * 2` will be used
360	:param tx_gas_price: Gas price of the external tx. If not, `gas_price` will be used
361	:param tx_nonce: Force nonce for `tx_sender`
362	:param block_identifier: `latest` or `pending`
363	:param eip1559_speed: If provided, use EIP1559 transaction
364	:return: Tuple(tx_hash, tx)
365	:raises: InvalidMultisigTx: If user tx cannot go through the Safe
366	"""
367
368	sender_account = Account.from_key(tx_sender_private_key)	4✔
369	if eip1559_speed and self.ethereum_client.is_eip1559_supported():	4✔
370	tx_parameters = self.ethereum_client.set_eip1559_fees(	×
371	{
372	"from": sender_account.address,
373	},
374	tx_speed=eip1559_speed,
375	)
376	else:
377	tx_parameters = {	4✔
378	"from": sender_account.address,
379	"gasPrice": tx_gas_price or self.w3.eth.gas_price,
380	}
381
382	if tx_gas:	4✔
383	tx_parameters["gas"] = tx_gas	×
384	if tx_nonce is not None:	4✔
385	tx_parameters["nonce"] = tx_nonce	×
386
387	self.tx: TxParams = self.w3_tx.build_transaction(tx_parameters)	4✔
388	self.tx["gas"] = Wei(	4✔
389	tx_gas or (max(self.tx["gas"] + 75000, self.recommended_gas()))
390	)
391
392	self.tx_hash = self.ethereum_client.send_unsigned_transaction(	4✔
393	self.tx,
394	private_key=sender_account.key,
395	retry=False if tx_nonce is not None else True,
396	block_identifier=block_identifier,
397	)
398
399	# Set signatures empty after executing the tx. `Nonce` is increased even if it fails,
400	# so signatures are not valid anymore
401	self.signatures = b""	4✔
402	return self.tx_hash, self.tx	4✔
403
404	def sign(self, private_key: str) -> bytes:	4✔
405	"""
406	{bytes32 r}{bytes32 s}{uint8 v}
407	:param private_key:
408	:return: Signature
409	"""
410	account = Account.from_key(private_key)	4✔
411	signature_dict = account.signHash(self.safe_tx_hash)	4✔
412	signature = signature_to_bytes(	4✔
413	signature_dict["v"], signature_dict["r"], signature_dict["s"]
414	)
415
416	# Insert signature sorted
417	if account.address not in self.signers:	4✔
418	unsorted_signatures = SafeSignature.parse_signature(	4✔
419	self.signatures + signature, self.safe_tx_hash
420	)
421	self.signatures = SafeSignature.export_signatures(unsorted_signatures)	4✔
422
423	return signature	4✔
424
425	def unsign(self, address: str) -> bool:	4✔
426	current_tx_signatures = SafeSignature.parse_signature(	4✔
427	self.signatures, self.safe_tx_hash
428	)
429	filtered_tx_signatures = list(	4✔
430	filter(lambda x: x.owner != address, current_tx_signatures)
431	)
432	if current_tx_signatures != filtered_tx_signatures:	4✔
433	self.signatures = SafeSignature.export_signatures(filtered_tx_signatures)	4✔
434	return True	4✔
435	return False	4✔

safe-global / safe-eth-py / 10629724120

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