10527345804

Committed 23 Aug 2024 02:20PM UTC coverage: 93.892% (-0.01%) from 93.903%

Build # 10527345804

Build Type

push

github

Committed by

web-flow

Commit Message

Add addresses 1.4.1 for chain BEVM_TESTNET (#1297)

* Add new master copy address 0x41675C099F32341bf84BFc5382aF534df5C7461a

* Add new master copy address 0x29fcB43b46531BcA003ddC8FCB67FFE91900C762

* Add new proxy address 0x4e1DCf7AD4e460CfD30791CCC4F9c8a4f820ec67

* Apply linter fixes

---------

Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>

Run Details

8670 of 9234 relevant lines covered (93.89%)

3.76 hits per line

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

68.42

/gnosis/safe/api/transaction_service_api/transaction_service_api.py

import logging
import os
from typing import Any, Dict, List, Optional, Tuple, Union
from urllib.parse import urlencode

from eth_typing import ChecksumAddress, Hash32, HexStr
from hexbytes import HexBytes

from gnosis.eth import EthereumClient, EthereumNetwork
from gnosis.eth.eip712 import eip712_encode_hash
from gnosis.safe import SafeTx

from ..base_api import SafeAPIException, SafeBaseAPI
from .entities import Balance, DataDecoded, DelegateUser, Message, Transaction
from .transaction_service_messages import get_delegate_message
from .transaction_service_tx import TransactionServiceTx

logger = logging.getLogger(__name__)


class ApiSafeTxHashNotMatchingException(SafeAPIException):
    pass


class TransactionServiceApi(SafeBaseAPI):
    URL_BY_NETWORK = {
        EthereumNetwork.ARBITRUM_ONE: "https://safe-transaction-arbitrum.safe.global",
        EthereumNetwork.AURORA_MAINNET: "https://safe-transaction-aurora.safe.global",
        EthereumNetwork.AVALANCHE_C_CHAIN: "https://safe-transaction-avalanche.safe.global",
        EthereumNetwork.BASE_GOERLI_TESTNET: "https://safe-transaction-base-testnet.safe.global",
        EthereumNetwork.BASE: "https://safe-transaction-base.safe.global",
        EthereumNetwork.BNB_SMART_CHAIN_MAINNET: "https://safe-transaction-bsc.safe.global",
        EthereumNetwork.CELO_MAINNET: "https://safe-transaction-celo.safe.global",
        EthereumNetwork.GNOSIS: "https://safe-transaction-gnosis-chain.safe.global",
        EthereumNetwork.GOERLI: "https://safe-transaction-goerli.safe.global",
        EthereumNetwork.MAINNET: "https://safe-transaction-mainnet.safe.global",
        EthereumNetwork.OPTIMISM: "https://safe-transaction-optimism.safe.global",
        EthereumNetwork.POLYGON: "https://safe-transaction-polygon.safe.global",
        EthereumNetwork.POLYGON_ZKEVM: "https://safe-transaction-zkevm.safe.global",
        EthereumNetwork.SEPOLIA: "https://safe-transaction-sepolia.safe.global",
        EthereumNetwork.ZKSYNC_MAINNET: "https://safe-transaction-zksync.safe.global",
    }

    def __init__(
        self,
        network: EthereumNetwork,
        ethereum_client: Optional[EthereumClient] = None,
        base_url: Optional[str] = None,
        request_timeout: int = int(
            os.environ.get("SAFE_TRANSACTION_SERVICE_REQUEST_TIMEOUT", 10)
        ),
    ):
        super().__init__(network, ethereum_client, base_url, request_timeout)

    @classmethod
    def data_decoded_to_text(cls, data_decoded: Dict[str, Any]) -> Optional[str]:
        """
        Decoded data decoded to text
        :param data_decoded:
        :return:
        """
        if not data_decoded:
            return None

        method = data_decoded["method"]
        parameters = data_decoded.get("parameters", [])
        text = ""
        for (
            parameter
        ) in parameters:  # Multisend or executeTransaction from another Safe
            if "decodedValue" in parameter:
                text += (
                    method
                    + ":\n - "
                    + "\n - ".join(
                        [
                            cls.data_decoded_to_text(
                                decoded_value.get("decodedData", {})
                            )
                            for decoded_value in parameter.get("decodedValue", {})
                        ]
                    )
                    + "\n"
                )
        if text:
            return text.strip()
        else:
            return (
                method
                + ": "
                + ",".join([str(parameter["value"]) for parameter in parameters])
            )

    @classmethod
    def parse_signatures(cls, raw_tx: Transaction) -> Optional[bytes]:
        """
        Parse signatures in `confirmations` list to build a valid signature (owners must be sorted lexicographically)

        :param raw_tx:
        :return: Valid signature with signatures sorted lexicographically
        """
        if raw_tx["signatures"]:
            # Tx was executed and signatures field is populated
            return raw_tx["signatures"]
        elif raw_tx["confirmations"]:
            # Parse offchain transactions
            return b"".join(
                [
                    HexBytes(confirmation["signature"])
                    for confirmation in sorted(
                        raw_tx["confirmations"], key=lambda x: int(x["owner"], 16)
                    )
                    if confirmation["signatureType"] == "EOA"
                ]
            )

    def create_delegate_message_hash(self, delegate_address: ChecksumAddress) -> Hash32:
        return eip712_encode_hash(
            get_delegate_message(delegate_address, self.network.value)
        )

    def _build_transaction_service_tx(
        self, safe_tx_hash: Union[bytes, HexStr], tx_raw: Transaction
    ) -> TransactionServiceTx:
        signatures = self.parse_signatures(tx_raw)
        safe_tx = TransactionServiceTx(
            tx_raw["proposer"],
            self.ethereum_client,
            tx_raw["safe"],
            tx_raw["to"],
            int(tx_raw["value"]),
            HexBytes(tx_raw["data"]) if tx_raw["data"] else b"",
            int(tx_raw["operation"]),
            int(tx_raw["safeTxGas"]),
            int(tx_raw["baseGas"]),
            int(tx_raw["gasPrice"]),
            tx_raw["gasToken"],
            tx_raw["refundReceiver"],
            signatures=signatures if signatures else b"",
            safe_nonce=int(tx_raw["nonce"]),
            chain_id=self.network.value,
        )
        safe_tx.tx_hash = (
            HexBytes(tx_raw["transactionHash"]) if tx_raw["transactionHash"] else None
        )

        if safe_tx.safe_tx_hash != safe_tx_hash:
            raise ApiSafeTxHashNotMatchingException(
                f"API safe-tx-hash: {safe_tx_hash.hex()} doesn't match the calculated safe-tx-hash: {safe_tx.safe_tx_hash.hex()}"
            )

        return safe_tx

    def get_balances(self, safe_address: str) -> List[Balance]:
        """

        :param safe_address:
        :return: a list of balances for provided Safe
        """
        response = self._get_request(f"/api/v1/safes/{safe_address}/balances/")
        if not response.ok:
            raise SafeAPIException(f"Cannot get balances: {response.content}")
        return response.json()

    def get_safe_transaction(
        self, safe_tx_hash: Union[bytes, HexStr]
    ) -> Tuple[TransactionServiceTx, Optional[HexBytes]]:
        """
        :param safe_tx_hash:
        :return: SafeTx and `tx-hash` if transaction was executed
        """
        safe_tx_hash_str = HexBytes(safe_tx_hash).hex()
        response = self._get_request(
            f"/api/v1/multisig-transactions/{safe_tx_hash_str}/"
        )
        if not response.ok:
            raise SafeAPIException(
                f"Cannot get transaction with safe-tx-hash={safe_tx_hash_str}: {response.content}"
            )

        if not self.ethereum_client:
            logger.warning(
                "EthereumClient should be defined to get a executable SafeTx"
            )

        result = response.json()
        safe_tx = self._build_transaction_service_tx(safe_tx_hash, result)

        return safe_tx, safe_tx.tx_hash

    def get_transactions(
        self, safe_address: ChecksumAddress, **kwargs: Dict[str, Union[str, int, bool]]
    ) -> List[Transaction]:
        """

        :param safe_address:
        :return: a list of transactions for provided Safe
        """
        url = f"/api/v1/safes/{safe_address}/multisig-transactions/"

        if kwargs:
            query_string = urlencode(
                {key: str(value) for key, value in kwargs.items() if value is not None}
            )
            url += "?" + query_string

        response = self._get_request(url)
        if not response.ok:
            raise SafeAPIException(f"Cannot get transactions: {response.content}")

        transactions = response.json().get("results", [])

        if safe_tx_hash_arg := kwargs.get("safe_tx_hash", None):
            # Validation that the calculated safe_tx_hash is the same as the safe_tx_hash provided for filter.
            safe_tx_hash = HexBytes(safe_tx_hash_arg)
            [
                self._build_transaction_service_tx(safe_tx_hash, tx)
                for tx in transactions
            ]

        return transactions

    def get_delegates(self, safe_address: ChecksumAddress) -> List[DelegateUser]:
        """

        :param safe_address:
        :return: a list of delegates for provided Safe
        """
        response = self._get_request(f"/api/v2/delegates/?safe={safe_address}")
        if not response.ok:
            raise SafeAPIException(f"Cannot get delegates: {response.content}")
        return response.json().get("results", [])

    def get_safes_for_owner(
        self, owner_address: ChecksumAddress
    ) -> List[ChecksumAddress]:
        """

        :param owner_address:
        :return: a List of Safe addresses which the owner_address is an owner
        """
        response = self._get_request(f"/api/v1/owners/{owner_address}/safes/")
        if not response.ok:
            raise SafeAPIException(f"Cannot get delegates: {response.content}")
        return response.json().get("safes", [])

    def post_signatures(self, safe_tx_hash: bytes, signatures: bytes) -> bool:
        """
        Create a new confirmation with provided signature for the given safe_tx_hash
        :param safe_tx_hash:
        :param signatures:
        :return: True if new confirmation was created
        """
        safe_tx_hash = HexBytes(safe_tx_hash).hex()
        response = self._post_request(
            f"/api/v1/multisig-transactions/{safe_tx_hash}/confirmations/",
            payload={"signature": HexBytes(signatures).hex()},
        )
        if not response.ok:
            raise SafeAPIException(
                f"Cannot post signatures for tx with safe-tx-hash={safe_tx_hash}: {response.content}"
            )
        return True

    def add_delegate(
        self,
        delegate_address: ChecksumAddress,
        delegator_address: ChecksumAddress,
        label: str,
        signature: bytes,
        safe_address: Optional[ChecksumAddress] = None,
    ) -> bool:
        add_payload = {
            "delegate": delegate_address,
            "delegator": delegator_address,
            "signature": HexBytes(signature).hex(),
            "label": label,
        }
        if safe_address:
            add_payload["safe"] = safe_address
        response = self._post_request("/api/v2/delegates/", add_payload)
        if not response.ok:
            raise SafeAPIException(f"Cannot add delegate: {response.content}")
        return True

    def remove_delegate(
        self,
        delegate_address: ChecksumAddress,
        delegator_address: ChecksumAddress,
        signature: bytes,
        safe_address: Optional[ChecksumAddress] = None,
    ) -> bool:
        """
        Deletes a delegated user

        :param delegator_address:
        :param delegate_address:
        :param signature: Signature of a hash of an eip712 message.
        :param safe_address: If specified, a delegate is removed for a delegator for the specific safe.
            Otherwise, the delegate is deleted in a global form.
        :return:
        """
        remove_payload = {
            "delegator": delegator_address,
            "signature": HexBytes(signature).hex(),
        }
        if safe_address:
            remove_payload["safe"] = safe_address
        response = self._delete_request(
            f"/api/v2/delegates/{delegate_address}/",
            remove_payload,
        )
        if not response.ok:
            raise SafeAPIException(f"Cannot remove delegate: {response.content}")
        return True

    def post_transaction(self, safe_tx: SafeTx) -> bool:
        random_sender = "0x0000000000000000000000000000000000000002"
        sender = safe_tx.sorted_signers[0] if safe_tx.sorted_signers else random_sender
        data = {
            "to": safe_tx.to,
            "value": safe_tx.value,
            "data": safe_tx.data.hex() if safe_tx.data else None,
            "operation": safe_tx.operation,
            "gasToken": safe_tx.gas_token,
            "safeTxGas": safe_tx.safe_tx_gas,
            "baseGas": safe_tx.base_gas,
            "gasPrice": safe_tx.gas_price,
            "refundReceiver": safe_tx.refund_receiver,
            "nonce": safe_tx.safe_nonce,
            "contractTransactionHash": safe_tx.safe_tx_hash.hex(),
            "sender": sender,
            "signature": safe_tx.signatures.hex() if safe_tx.signatures else None,
            "origin": "Safe-CLI",
        }
        response = self._post_request(
            f"/api/v1/safes/{safe_tx.safe_address}/multisig-transactions/", data
        )
        if not response.ok:
            raise SafeAPIException(f"Error posting transaction: {response.content}")
        return True

    def delete_transaction(self, safe_tx_hash: str, signature: str) -> bool:
        """

        :param safe_tx_hash: hash of eip712 see in transaction_service_messages.py generate_remove_transaction_message function
        :param signature: signature of safe_tx_hash by transaction proposer
        :return:
        """
        payload = {"safeTxHash": safe_tx_hash, "signature": signature}
        response = self._delete_request(
            f"/api/v1/multisig-transactions/{safe_tx_hash}/", payload
        )
        if not response.ok:
            raise SafeAPIException(f"Error deleting transaction: {response.content}")
        return True

    def post_message(
        self,
        safe_address: ChecksumAddress,
        message: Union[str, Dict],
        signature: bytes,
        safe_app_id: Optional[int] = 0,
    ) -> bool:
        """
        Create safe message on transaction service for provided Safe address

        :param safe_address:
        :param message: If str it will be encoded using EIP191, and if it's a dictionary it will be encoded using EIP712
        :param signature:
        :return:
        """
        payload = {
            "message": message,
            "safeAppId": safe_app_id,
            "signature": HexBytes(signature).hex(),
        }
        response = self._post_request(
            f"/api/v1/safes/{safe_address}/messages/", payload
        )
        if not response.ok:
            raise SafeAPIException(f"Error posting message: {response.content}")
        return True

    def get_message(self, safe_message_hash: bytes) -> Message:
        """

        :param safe_message_hash:
        :return: Safe message for provided Safe message hash
        """
        response = self._get_request(
            f"/api/v1/messages/{HexBytes(safe_message_hash).hex()}/"
        )
        if not response.ok:
            raise SafeAPIException(f"Cannot get messages: {response.content}")
        return response.json()

    def get_messages(self, safe_address: ChecksumAddress) -> List[Message]:
        """

        :param safe_address:
        :return: list of messages for provided Safe address
        """
        response = self._get_request(f"/api/v1/safes/{safe_address}/messages/")
        if not response.ok:
            raise SafeAPIException(f"Cannot get messages: {response.content}")
        return response.json().get("results", [])

    def post_message_signature(
        self, safe_message_hash: bytes, signature: bytes
    ) -> bool:
        """
        Add a new confirmation for provided Safe message hash

        :param safe_message_hash:
        :param signature:
        :return:
        """
        payload = {"signature": HexBytes(signature).hex()}
        response = self._post_request(
            f"/api/v1/messages/{HexBytes(safe_message_hash).hex()}/signatures/", payload
        )
        if not response.ok:
            raise SafeAPIException(
                f"Error posting message signature: {response.content}"
            )
        return True

    def decode_data(
        self, data: Union[bytes, HexStr], to_address: Optional[ChecksumAddress] = None
    ) -> DataDecoded:
        """
        Retrieve decoded information using tx service internal ABI information given the tx data.

        :param data: tx data as a 0x prefixed hexadecimal string.
        :param to_address: address of the contract. This will be used in case of more than one function identifiers matching.
        :return:
        """
        payload = {"data": HexBytes(data).hex()}
        if to_address:
            payload["to"] = to_address
        response = self._post_request("/api/v1/data-decoder/", payload)
        if not response.ok:
            raise SafeAPIException(f"Cannot decode tx data: {response.content}")
        return response.json()

1	import logging	4✔
2	import os	4✔
3	from typing import Any, Dict, List, Optional, Tuple, Union	4✔
4	from urllib.parse import urlencode	4✔
5
6	from eth_typing import ChecksumAddress, Hash32, HexStr	4✔
7	from hexbytes import HexBytes	4✔
8
9	from gnosis.eth import EthereumClient, EthereumNetwork	4✔
10	from gnosis.eth.eip712 import eip712_encode_hash	4✔
11	from gnosis.safe import SafeTx	4✔
12
13	from ..base_api import SafeAPIException, SafeBaseAPI	4✔
14	from .entities import Balance, DataDecoded, DelegateUser, Message, Transaction	4✔
15	from .transaction_service_messages import get_delegate_message	4✔
16	from .transaction_service_tx import TransactionServiceTx	4✔
17
18	logger = logging.getLogger(__name__)	4✔
19
20
21	class ApiSafeTxHashNotMatchingException(SafeAPIException):	4✔
22	pass	4✔
23
24
25	class TransactionServiceApi(SafeBaseAPI):	4✔
26	URL_BY_NETWORK = {	4✔
27	EthereumNetwork.ARBITRUM_ONE: "https://safe-transaction-arbitrum.safe.global",
28	EthereumNetwork.AURORA_MAINNET: "https://safe-transaction-aurora.safe.global",
29	EthereumNetwork.AVALANCHE_C_CHAIN: "https://safe-transaction-avalanche.safe.global",
30	EthereumNetwork.BASE_GOERLI_TESTNET: "https://safe-transaction-base-testnet.safe.global",
31	EthereumNetwork.BASE: "https://safe-transaction-base.safe.global",
32	EthereumNetwork.BNB_SMART_CHAIN_MAINNET: "https://safe-transaction-bsc.safe.global",
33	EthereumNetwork.CELO_MAINNET: "https://safe-transaction-celo.safe.global",
34	EthereumNetwork.GNOSIS: "https://safe-transaction-gnosis-chain.safe.global",
35	EthereumNetwork.GOERLI: "https://safe-transaction-goerli.safe.global",
36	EthereumNetwork.MAINNET: "https://safe-transaction-mainnet.safe.global",
37	EthereumNetwork.OPTIMISM: "https://safe-transaction-optimism.safe.global",
38	EthereumNetwork.POLYGON: "https://safe-transaction-polygon.safe.global",
39	EthereumNetwork.POLYGON_ZKEVM: "https://safe-transaction-zkevm.safe.global",
40	EthereumNetwork.SEPOLIA: "https://safe-transaction-sepolia.safe.global",
41	EthereumNetwork.ZKSYNC_MAINNET: "https://safe-transaction-zksync.safe.global",
42	}
43
44	def __init__(	4✔
45	self,
46	network: EthereumNetwork,
47	ethereum_client: Optional[EthereumClient] = None,
48	base_url: Optional[str] = None,
49	request_timeout: int = int(
50	os.environ.get("SAFE_TRANSACTION_SERVICE_REQUEST_TIMEOUT", 10)
51	),
52	):
53	super().__init__(network, ethereum_client, base_url, request_timeout)	4✔
54
55	@classmethod	4✔
56	def data_decoded_to_text(cls, data_decoded: Dict[str, Any]) -> Optional[str]:	4✔
57	"""
58	Decoded data decoded to text
59	:param data_decoded:
60	:return:
61	"""
62	if not data_decoded:	4✔
63	return None	×
64
65	method = data_decoded["method"]	4✔
66	parameters = data_decoded.get("parameters", [])	4✔
67	text = ""	4✔
68	for (	4✔
69	parameter
70	) in parameters: # Multisend or executeTransaction from another Safe
71	if "decodedValue" in parameter:	4✔
72	text += (	4✔
73	method
74	+ ":\n - "
75	+ "\n - ".join(
76	[
77	cls.data_decoded_to_text(
78	decoded_value.get("decodedData", {})
79	)
80	for decoded_value in parameter.get("decodedValue", {})
81	]
82	)
83	+ "\n"
84	)
85	if text:	4✔
86	return text.strip()	4✔
87	else:
88	return (	4✔
89	method
90	+ ": "
91	+ ",".join([str(parameter["value"]) for parameter in parameters])
92	)
93
94	@classmethod	4✔
95	def parse_signatures(cls, raw_tx: Transaction) -> Optional[bytes]:	4✔
96	"""
97	Parse signatures in `confirmations` list to build a valid signature (owners must be sorted lexicographically)
98
99	:param raw_tx:
100	:return: Valid signature with signatures sorted lexicographically
101	"""
102	if raw_tx["signatures"]:	4✔
103	# Tx was executed and signatures field is populated
104	return raw_tx["signatures"]	4✔
105	elif raw_tx["confirmations"]:	×
106	# Parse offchain transactions
107	return b"".join(	×
108	[
109	HexBytes(confirmation["signature"])
110	for confirmation in sorted(
111	raw_tx["confirmations"], key=lambda x: int(x["owner"], 16)
112	)
113	if confirmation["signatureType"] == "EOA"
114	]
115	)
116
117	def create_delegate_message_hash(self, delegate_address: ChecksumAddress) -> Hash32:	4✔
118	return eip712_encode_hash(	4✔
119	get_delegate_message(delegate_address, self.network.value)
120	)
121
122	def _build_transaction_service_tx(	4✔
123	self, safe_tx_hash: Union[bytes, HexStr], tx_raw: Transaction
124	) -> TransactionServiceTx:
125	signatures = self.parse_signatures(tx_raw)	4✔
126	safe_tx = TransactionServiceTx(	4✔
127	tx_raw["proposer"],
128	self.ethereum_client,
129	tx_raw["safe"],
130	tx_raw["to"],
131	int(tx_raw["value"]),
132	HexBytes(tx_raw["data"]) if tx_raw["data"] else b"",
133	int(tx_raw["operation"]),
134	int(tx_raw["safeTxGas"]),
135	int(tx_raw["baseGas"]),
136	int(tx_raw["gasPrice"]),
137	tx_raw["gasToken"],
138	tx_raw["refundReceiver"],
139	signatures=signatures if signatures else b"",
140	safe_nonce=int(tx_raw["nonce"]),
141	chain_id=self.network.value,
142	)
143	safe_tx.tx_hash = (	4✔
144	HexBytes(tx_raw["transactionHash"]) if tx_raw["transactionHash"] else None
145	)
146
147	if safe_tx.safe_tx_hash != safe_tx_hash:	4✔
148	raise ApiSafeTxHashNotMatchingException(	4✔
149	f"API safe-tx-hash: {safe_tx_hash.hex()} doesn't match the calculated safe-tx-hash: {safe_tx.safe_tx_hash.hex()}"
150	)
151
152	return safe_tx	4✔
153
154	def get_balances(self, safe_address: str) -> List[Balance]:	4✔
155	"""
156
157	:param safe_address:
158	:return: a list of balances for provided Safe
159	"""
160	response = self._get_request(f"/api/v1/safes/{safe_address}/balances/")	4✔
161	if not response.ok:	4✔
162	raise SafeAPIException(f"Cannot get balances: {response.content}")	×
163	return response.json()	4✔
164
165	def get_safe_transaction(	4✔
166	self, safe_tx_hash: Union[bytes, HexStr]
167	) -> Tuple[TransactionServiceTx, Optional[HexBytes]]:
168	"""
169	:param safe_tx_hash:
170	:return: SafeTx and `tx-hash` if transaction was executed
171	"""
172	safe_tx_hash_str = HexBytes(safe_tx_hash).hex()	4✔
173	response = self._get_request(	4✔
174	f"/api/v1/multisig-transactions/{safe_tx_hash_str}/"
175	)
176	if not response.ok:	4✔
177	raise SafeAPIException(	4✔
178	f"Cannot get transaction with safe-tx-hash={safe_tx_hash_str}: {response.content}"
179	)
180
181	if not self.ethereum_client:	4✔
182	logger.warning(	×
183	"EthereumClient should be defined to get a executable SafeTx"
184	)
185
186	result = response.json()	4✔
187	safe_tx = self._build_transaction_service_tx(safe_tx_hash, result)	4✔
188
189	return safe_tx, safe_tx.tx_hash	4✔
190
191	def get_transactions(	4✔
192	self, safe_address: ChecksumAddress, **kwargs: Dict[str, Union[str, int, bool]]
193	) -> List[Transaction]:
194	"""
195
196	:param safe_address:
197	:return: a list of transactions for provided Safe
198	"""
199	url = f"/api/v1/safes/{safe_address}/multisig-transactions/"	4✔
200
201	if kwargs:	4✔
202	query_string = urlencode(	4✔
203	{key: str(value) for key, value in kwargs.items() if value is not None}
204	)
205	url += "?" + query_string	4✔
206
207	response = self._get_request(url)	4✔
208	if not response.ok:	4✔
209	raise SafeAPIException(f"Cannot get transactions: {response.content}")	×
210
211	transactions = response.json().get("results", [])	4✔
212
213	if safe_tx_hash_arg := kwargs.get("safe_tx_hash", None):	4✔
214	# Validation that the calculated safe_tx_hash is the same as the safe_tx_hash provided for filter.
215	safe_tx_hash = HexBytes(safe_tx_hash_arg)	4✔
216	[	4✔
217	self._build_transaction_service_tx(safe_tx_hash, tx)
218	for tx in transactions
219	]
220
221	return transactions	4✔
222
223	def get_delegates(self, safe_address: ChecksumAddress) -> List[DelegateUser]:	4✔
224	"""
225
226	:param safe_address:
227	:return: a list of delegates for provided Safe
228	"""
229	response = self._get_request(f"/api/v2/delegates/?safe={safe_address}")	×
230	if not response.ok:	×
231	raise SafeAPIException(f"Cannot get delegates: {response.content}")	×
232	return response.json().get("results", [])	×
233
234	def get_safes_for_owner(	4✔
235	self, owner_address: ChecksumAddress
236	) -> List[ChecksumAddress]:
237	"""
238
239	:param owner_address:
240	:return: a List of Safe addresses which the owner_address is an owner
241	"""
242	response = self._get_request(f"/api/v1/owners/{owner_address}/safes/")	4✔
243	if not response.ok:	4✔
244	raise SafeAPIException(f"Cannot get delegates: {response.content}")	×
245	return response.json().get("safes", [])	4✔
246
247	def post_signatures(self, safe_tx_hash: bytes, signatures: bytes) -> bool:	4✔
248	"""
249	Create a new confirmation with provided signature for the given safe_tx_hash
250	:param safe_tx_hash:
251	:param signatures:
252	:return: True if new confirmation was created
253	"""
254	safe_tx_hash = HexBytes(safe_tx_hash).hex()	×
255	response = self._post_request(	×
256	f"/api/v1/multisig-transactions/{safe_tx_hash}/confirmations/",
257	payload={"signature": HexBytes(signatures).hex()},
258	)
259	if not response.ok:	×
260	raise SafeAPIException(	×
261	f"Cannot post signatures for tx with safe-tx-hash={safe_tx_hash}: {response.content}"
262	)
263	return True	×
264
265	def add_delegate(	4✔
266	self,
267	delegate_address: ChecksumAddress,
268	delegator_address: ChecksumAddress,
269	label: str,
270	signature: bytes,
271	safe_address: Optional[ChecksumAddress] = None,
272	) -> bool:
273	add_payload = {	4✔
274	"delegate": delegate_address,
275	"delegator": delegator_address,
276	"signature": HexBytes(signature).hex(),
277	"label": label,
278	}
279	if safe_address:	4✔
280	add_payload["safe"] = safe_address	4✔
281	response = self._post_request("/api/v2/delegates/", add_payload)	4✔
282	if not response.ok:	4✔
283	raise SafeAPIException(f"Cannot add delegate: {response.content}")	×
284	return True	4✔
285
286	def remove_delegate(	4✔
287	self,
288	delegate_address: ChecksumAddress,
289	delegator_address: ChecksumAddress,
290	signature: bytes,
291	safe_address: Optional[ChecksumAddress] = None,
292	) -> bool:
293	"""
294	Deletes a delegated user
295
296	:param delegator_address:
297	:param delegate_address:
298	:param signature: Signature of a hash of an eip712 message.
299	:param safe_address: If specified, a delegate is removed for a delegator for the specific safe.
300	Otherwise, the delegate is deleted in a global form.
301	:return:
302	"""
303	remove_payload = {	4✔
304	"delegator": delegator_address,
305	"signature": HexBytes(signature).hex(),
306	}
307	if safe_address:	4✔
308	remove_payload["safe"] = safe_address	4✔
309	response = self._delete_request(	4✔
310	f"/api/v2/delegates/{delegate_address}/",
311	remove_payload,
312	)
313	if not response.ok:	4✔
314	raise SafeAPIException(f"Cannot remove delegate: {response.content}")	×
315	return True	4✔
316
317	def post_transaction(self, safe_tx: SafeTx) -> bool:	4✔
318	random_sender = "0x0000000000000000000000000000000000000002"	×
319	sender = safe_tx.sorted_signers[0] if safe_tx.sorted_signers else random_sender	×
320	data = {	×
321	"to": safe_tx.to,
322	"value": safe_tx.value,
323	"data": safe_tx.data.hex() if safe_tx.data else None,
324	"operation": safe_tx.operation,
325	"gasToken": safe_tx.gas_token,
326	"safeTxGas": safe_tx.safe_tx_gas,
327	"baseGas": safe_tx.base_gas,
328	"gasPrice": safe_tx.gas_price,
329	"refundReceiver": safe_tx.refund_receiver,
330	"nonce": safe_tx.safe_nonce,
331	"contractTransactionHash": safe_tx.safe_tx_hash.hex(),
332	"sender": sender,
333	"signature": safe_tx.signatures.hex() if safe_tx.signatures else None,
334	"origin": "Safe-CLI",
335	}
336	response = self._post_request(	×
337	f"/api/v1/safes/{safe_tx.safe_address}/multisig-transactions/", data
338	)
339	if not response.ok:	×
340	raise SafeAPIException(f"Error posting transaction: {response.content}")	×
341	return True	×
342
343	def delete_transaction(self, safe_tx_hash: str, signature: str) -> bool:	4✔
344	"""
345
346	:param safe_tx_hash: hash of eip712 see in transaction_service_messages.py generate_remove_transaction_message function
347	:param signature: signature of safe_tx_hash by transaction proposer
348	:return:
349	"""
350	payload = {"safeTxHash": safe_tx_hash, "signature": signature}	×
351	response = self._delete_request(	×
352	f"/api/v1/multisig-transactions/{safe_tx_hash}/", payload
353	)
354	if not response.ok:	×
355	raise SafeAPIException(f"Error deleting transaction: {response.content}")	×
356	return True	×
357
358	def post_message(	4✔
359	self,
360	safe_address: ChecksumAddress,
361	message: Union[str, Dict],
362	signature: bytes,
363	safe_app_id: Optional[int] = 0,
364	) -> bool:
365	"""
366	Create safe message on transaction service for provided Safe address
367
368	:param safe_address:
369	:param message: If str it will be encoded using EIP191, and if it's a dictionary it will be encoded using EIP712
370	:param signature:
371	:return:
372	"""
373	payload = {	×
374	"message": message,
375	"safeAppId": safe_app_id,
376	"signature": HexBytes(signature).hex(),
377	}
378	response = self._post_request(	×
379	f"/api/v1/safes/{safe_address}/messages/", payload
380	)
381	if not response.ok:	×
382	raise SafeAPIException(f"Error posting message: {response.content}")	×
383	return True	×
384
385	def get_message(self, safe_message_hash: bytes) -> Message:	4✔
386	"""
387
388	:param safe_message_hash:
389	:return: Safe message for provided Safe message hash
390	"""
391	response = self._get_request(	×
392	f"/api/v1/messages/{HexBytes(safe_message_hash).hex()}/"
393	)
394	if not response.ok:	×
395	raise SafeAPIException(f"Cannot get messages: {response.content}")	×
396	return response.json()	×
397
398	def get_messages(self, safe_address: ChecksumAddress) -> List[Message]:	4✔
399	"""
400
401	:param safe_address:
402	:return: list of messages for provided Safe address
403	"""
404	response = self._get_request(f"/api/v1/safes/{safe_address}/messages/")	×
405	if not response.ok:	×
406	raise SafeAPIException(f"Cannot get messages: {response.content}")	×
407	return response.json().get("results", [])	×
408
409	def post_message_signature(	4✔
410	self, safe_message_hash: bytes, signature: bytes
411	) -> bool:
412	"""
413	Add a new confirmation for provided Safe message hash
414
415	:param safe_message_hash:
416	:param signature:
417	:return:
418	"""
419	payload = {"signature": HexBytes(signature).hex()}	×
420	response = self._post_request(	×
421	f"/api/v1/messages/{HexBytes(safe_message_hash).hex()}/signatures/", payload
422	)
423	if not response.ok:	×
424	raise SafeAPIException(	×
425	f"Error posting message signature: {response.content}"
426	)
427	return True	×
428
429	def decode_data(	4✔
430	self, data: Union[bytes, HexStr], to_address: Optional[ChecksumAddress] = None
431	) -> DataDecoded:
432	"""
433	Retrieve decoded information using tx service internal ABI information given the tx data.
434
435	:param data: tx data as a 0x prefixed hexadecimal string.
436	:param to_address: address of the contract. This will be used in case of more than one function identifiers matching.
437	:return:
438	"""
439	payload = {"data": HexBytes(data).hex()}	4✔
440	if to_address:	4✔
441	payload["to"] = to_address	4✔
442	response = self._post_request("/api/v1/data-decoder/", payload)	4✔
443	if not response.ok:	4✔
444	raise SafeAPIException(f"Cannot decode tx data: {response.content}")	4✔
445	return response.json()	4✔

safe-global / safe-eth-py / 10527345804

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