11324900971

Committed 14 Oct 2024 09:37AM UTC coverage: 93.591%. Remained the same

Build # 11324900971

Build Type

push

github

Committed by

Uxio0

Commit Message

Bump faker from 30.1.0 to 30.3.0

Bumps [faker](https://github.com/joke2k/faker) from 30.1.0 to 30.3.0.
- [Release notes](https://github.com/joke2k/faker/releases)
- [Changelog](https://github.com/joke2k/faker/blob/master/CHANGELOG.md)
- [Commits](https://github.com/joke2k/faker/compare/v30.1.0...v30.3.0)

---
updated-dependencies:
- dependency-name: faker
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

Signed-off-by: dependabot[bot] <support@github.com>

Run Details

8923 of 9534 relevant lines covered (93.59%)

3.74 hits per line

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

68.18

/safe_eth/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 eth_utils import to_checksum_address
from hexbytes import HexBytes

from safe_eth.eth import EthereumClient, EthereumNetwork
from safe_eth.eth.eip712 import eip712_encode_hash
from safe_eth.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(
                        [
                            decoded_text
                            for decoded_value in parameter.get("decodedValue", {})
                            if (
                                decoded_text := cls.data_decoded_to_text(
                                    decoded_value.get("decodedData", {})
                                )
                            )
                            is not None
                        ]
                    )
                    + "\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 HexBytes(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"
                ]
            )
        return None

    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(
            to_checksum_address(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() if isinstance(safe_tx_hash, bytes) else safe_tx_hash} "
                f"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!r}")
        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!r}"
            )

        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, HexBytes(safe_tx.tx_hash) if safe_tx.tx_hash else None

    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!r}")

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

safe-global / safe-eth-py / 11324900971

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