6508017815

Committed 13 Oct 2023 12:09PM UTC coverage: 83.849% (-0.4%) from 84.269%

Build # 6508017815

Build Type

Pull #277

github

Committed by

web-flow

Commit Message

Merge f49030687 into 688bc1673

Pull Request Pull Request #277: Remove relaying

Run Details

867 of 1034 relevant lines covered (83.85%)

3.35 hits per line

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

94.74

/safe_cli/api/transaction_service_api.py

import time
from typing import Any, Dict, List, Optional, Tuple
from urllib.parse import urljoin

import requests
from eth_account.signers.local import LocalAccount
from hexbytes import HexBytes
from web3 import Web3

from gnosis.eth.ethereum_client import EthereumNetwork
from gnosis.safe import SafeTx

from .base_api import BaseAPI, BaseAPIException


class TransactionServiceApi(BaseAPI):
    URL_BY_NETWORK = {
        EthereumNetwork.MAINNET: "https://safe-transaction-mainnet.safe.global",
        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.BINANCE_SMART_CHAIN_MAINNET: "https://safe-transaction-bsc.safe.global",
        EthereumNetwork.ENERGY_WEB_CHAIN: "https://safe-transaction-ewc.safe.global",
        EthereumNetwork.GOERLI: "https://safe-transaction-goerli.safe.global",
        EthereumNetwork.POLYGON: "https://safe-transaction-polygon.safe.global",
        EthereumNetwork.OPTIMISM: "https://safe-transaction-optimism.safe.global",
        EthereumNetwork.ENERGY_WEB_VOLTA_TESTNET: "https://safe-transaction-volta.safe.global",
        EthereumNetwork.GNOSIS: "https://safe-transaction-gnosis-chain.safe.global",
    }

    @classmethod
    def create_delegate_message_hash(cls, delegate_address: str) -> str:
        totp = int(time.time()) // 3600
        hash_to_sign = Web3.keccak(text=delegate_address + str(totp))
        return hash_to_sign

    def data_decoded_to_text(self, 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(
                        [
                            self.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])
            )

    def get_balances(self, safe_address: str) -> List[Dict[str, Any]]:
        response = self._get_request(f"/api/v1/safes/{safe_address}/balances/")
        if not response.ok:
            raise BaseAPIException(f"Cannot get balances: {response.content}")
        else:
            return response.json()

    def get_safe_transaction(
        self, safe_tx_hash: bytes
    ) -> Tuple[SafeTx, Optional[HexBytes]]:
        """
        :param safe_tx_hash:
        :return: SafeTx and `tx-hash` if transaction was executed
        """
        safe_tx_hash = HexBytes(safe_tx_hash).hex()
        response = self._get_request(f"/api/v1/multisig-transactions/{safe_tx_hash}/")
        if not response.ok:
            raise BaseAPIException(
                f"Cannot get transaction with safe-tx-hash={safe_tx_hash}: {response.content}"
            )
        else:
            result = response.json()
            # TODO return tx-hash if executed
            signatures = self.parse_signatures(result)
            return (
                SafeTx(
                    self.ethereum_client,
                    result["safe"],
                    result["to"],
                    int(result["value"]),
                    HexBytes(result["data"]) if result["data"] else b"",
                    int(result["operation"]),
                    int(result["safeTxGas"]),
                    int(result["baseGas"]),
                    int(result["gasPrice"]),
                    result["gasToken"],
                    result["refundReceiver"],
                    signatures=signatures if signatures else b"",
                    safe_nonce=int(result["nonce"]),
                ),
                HexBytes(result["transactionHash"])
                if result["transactionHash"]
                else None,
            )

    def parse_signatures(self, raw_tx: Dict[str, Any]) -> Optional[HexBytes]:
        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 get_transactions(self, safe_address: str) -> List[Dict[str, Any]]:
        response = self._get_request(
            f"/api/v1/safes/{safe_address}/multisig-transactions/"
        )
        if not response.ok:
            raise BaseAPIException(f"Cannot get transactions: {response.content}")
        else:
            return response.json().get("results", [])

    def get_delegates(self, safe_address: str) -> List[Dict[str, Any]]:
        # 200 delegates should be enough so we don't paginate
        response = self._get_request(
            f"/api/v1/delegates/?safe={safe_address}&limit=200"
        )
        if not response.ok:
            raise BaseAPIException(f"Cannot get delegates: {response.content}")
        else:
            return response.json().get("results", [])

    def post_signatures(self, safe_tx_hash: bytes, signatures: bytes) -> None:
        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 BaseAPIException(
                f"Cannot post signatures for tx with safe-tx-hash={safe_tx_hash}: {response.content}"
            )

    def add_delegate(
        self,
        safe_address: str,
        delegate_address: str,
        label: str,
        signer_account: LocalAccount,
    ):
        hash_to_sign = self.create_delegate_message_hash(delegate_address)
        signature = signer_account.signHash(hash_to_sign)
        add_payload = {
            "safe": safe_address,
            "delegate": delegate_address,
            "signature": signature.signature.hex(),
            "label": label,
        }
        response = self._post_request(
            f"/api/v1/safes/{safe_address}/delegates/", add_payload
        )
        if not response.ok:
            raise BaseAPIException(f"Cannot add delegate: {response.content}")

    def remove_delegate(
        self, safe_address: str, delegate_address: str, signer_account: LocalAccount
    ):
        hash_to_sign = self.create_delegate_message_hash(delegate_address)
        signature = signer_account.signHash(hash_to_sign)
        remove_payload = {"signature": signature.signature.hex()}
        response = self._delete_request(
            f"/api/v1/safes/{safe_address}/delegates/{delegate_address}/",
            remove_payload,
        )
        if not response.ok:
            raise BaseAPIException(f"Cannot remove delegate: {response.content}")

    def post_transaction(self, safe_address: str, safe_tx: SafeTx):
        url = urljoin(
            self.base_url, f"/api/v1/safes/{safe_address}/multisig-transactions/"
        )
        random_account = "0x1b95E981F808192Dc5cdCF92ef589f9CBe6891C4"
        sender = safe_tx.sorted_signers[0] if safe_tx.sorted_signers else random_account
        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 = requests.post(url, json=data)
        if not response.ok:
            raise BaseAPIException(f"Error posting transaction: {response.content}")

1	import time	4✔
2	from typing import Any, Dict, List, Optional, Tuple	4✔
3	from urllib.parse import urljoin	4✔
4
5	import requests	4✔
6	from eth_account.signers.local import LocalAccount	4✔
7	from hexbytes import HexBytes	4✔
8	from web3 import Web3	4✔
9
10	from gnosis.eth.ethereum_client import EthereumNetwork	4✔
11	from gnosis.safe import SafeTx	4✔
12
13	from .base_api import BaseAPI, BaseAPIException	4✔
14
15
16	class TransactionServiceApi(BaseAPI):	4✔
17	URL_BY_NETWORK = {	4✔
18	EthereumNetwork.MAINNET: "https://safe-transaction-mainnet.safe.global",	4✔
19	EthereumNetwork.ARBITRUM_ONE: "https://safe-transaction-arbitrum.safe.global",	4✔
20	EthereumNetwork.AURORA_MAINNET: "https://safe-transaction-aurora.safe.global",	4✔
21	EthereumNetwork.AVALANCHE_C_CHAIN: "https://safe-transaction-avalanche.safe.global",	4✔
22	EthereumNetwork.BINANCE_SMART_CHAIN_MAINNET: "https://safe-transaction-bsc.safe.global",	4✔
23	EthereumNetwork.ENERGY_WEB_CHAIN: "https://safe-transaction-ewc.safe.global",	4✔
24	EthereumNetwork.GOERLI: "https://safe-transaction-goerli.safe.global",	4✔
25	EthereumNetwork.POLYGON: "https://safe-transaction-polygon.safe.global",	4✔
26	EthereumNetwork.OPTIMISM: "https://safe-transaction-optimism.safe.global",	4✔
27	EthereumNetwork.ENERGY_WEB_VOLTA_TESTNET: "https://safe-transaction-volta.safe.global",	4✔
28	EthereumNetwork.GNOSIS: "https://safe-transaction-gnosis-chain.safe.global",	4✔
29	}
30
31	@classmethod	4✔
32	def create_delegate_message_hash(cls, delegate_address: str) -> str:	4✔
33	totp = int(time.time()) // 3600	×
34	hash_to_sign = Web3.keccak(text=delegate_address + str(totp))	×
35	return hash_to_sign	×
36
37	def data_decoded_to_text(self, data_decoded: Dict[str, Any]) -> Optional[str]:	4✔
38	"""
39	Decoded data decoded to text
40	:param data_decoded:
41	:return:
42	"""
43	if not data_decoded:	4✔
44	return None	×
45
46	method = data_decoded["method"]	4✔
47	parameters = data_decoded.get("parameters", [])	4✔
48	text = ""	4✔
49	for (	4✔
50	parameter	4✔
51	) in parameters: # Multisend or executeTransaction from another Safe	4✔
52	if "decodedValue" in parameter:	4✔
53	text += (	4✔
54	method	4✔
55	+ ":\n - "	4✔
56	+ "\n - ".join(	4✔
57	[	4✔
58	self.data_decoded_to_text(	4✔
59	decoded_value.get("decodedData", {})	4✔
60	)
61	for decoded_value in parameter.get("decodedValue", {})	4✔
62	]
63	)
64	+ "\n"	4✔
65	)
66	if text:	4✔
67	return text.strip()	4✔
68	else:
69	return (	4✔
70	method	4✔
71	+ ": "	4✔
72	+ ",".join([str(parameter["value"]) for parameter in parameters])	4✔
73	)
74
75	def get_balances(self, safe_address: str) -> List[Dict[str, Any]]:	4✔
76	response = self._get_request(f"/api/v1/safes/{safe_address}/balances/")	4✔
77	if not response.ok:	4✔
78	raise BaseAPIException(f"Cannot get balances: {response.content}")
79	else:
80	return response.json()	4✔
81
82	def get_safe_transaction(	4✔
83	self, safe_tx_hash: bytes	4✔
84	) -> Tuple[SafeTx, Optional[HexBytes]]:	4✔
85	"""
86	:param safe_tx_hash:
87	:return: SafeTx and `tx-hash` if transaction was executed
88	"""
89	safe_tx_hash = HexBytes(safe_tx_hash).hex()
90	response = self._get_request(f"/api/v1/multisig-transactions/{safe_tx_hash}/")
91	if not response.ok:
92	raise BaseAPIException(
93	f"Cannot get transaction with safe-tx-hash={safe_tx_hash}: {response.content}"
94	)
95	else:
96	result = response.json()
97	# TODO return tx-hash if executed
98	signatures = self.parse_signatures(result)
99	return (
100	SafeTx(
101	self.ethereum_client,
102	result["safe"],
103	result["to"],
104	int(result["value"]),
105	HexBytes(result["data"]) if result["data"] else b"",
106	int(result["operation"]),
107	int(result["safeTxGas"]),
108	int(result["baseGas"]),
109	int(result["gasPrice"]),
110	result["gasToken"],
111	result["refundReceiver"],
112	signatures=signatures if signatures else b"",
113	safe_nonce=int(result["nonce"]),
114	),
115	HexBytes(result["transactionHash"])
116	if result["transactionHash"]
117	else None,
118	)
119
120	def parse_signatures(self, raw_tx: Dict[str, Any]) -> Optional[HexBytes]:	4✔
121	if raw_tx["signatures"]:
122	# Tx was executed and signatures field is populated
123	return raw_tx["signatures"]
124	elif raw_tx["confirmations"]:
125	# Parse offchain transactions
126	return b"".join(
127	[
128	HexBytes(confirmation["signature"])
129	for confirmation in sorted(
130	raw_tx["confirmations"], key=lambda x: int(x["owner"], 16)
131	)
132	if confirmation["signatureType"] == "EOA"
133	]
134	)
135
136	def get_transactions(self, safe_address: str) -> List[Dict[str, Any]]:	4✔
137	response = self._get_request(	4✔
138	f"/api/v1/safes/{safe_address}/multisig-transactions/"	4✔
139	)
140	if not response.ok:	4✔
141	raise BaseAPIException(f"Cannot get transactions: {response.content}")
142	else:
143	return response.json().get("results", [])	4✔
144
145	def get_delegates(self, safe_address: str) -> List[Dict[str, Any]]:	4✔
146	# 200 delegates should be enough so we don't paginate
147	response = self._get_request(
148	f"/api/v1/delegates/?safe={safe_address}&limit=200"
149	)
150	if not response.ok:
151	raise BaseAPIException(f"Cannot get delegates: {response.content}")
152	else:
153	return response.json().get("results", [])
154
155	def post_signatures(self, safe_tx_hash: bytes, signatures: bytes) -> None:	4✔
156	safe_tx_hash = HexBytes(safe_tx_hash).hex()
157	response = self._post_request(
158	f"/api/v1/multisig-transactions/{safe_tx_hash}/confirmations/",
159	payload={"signature": HexBytes(signatures).hex()},
160	)
161	if not response.ok:
162	raise BaseAPIException(
163	f"Cannot post signatures for tx with safe-tx-hash={safe_tx_hash}: {response.content}"
164	)
165
166	def add_delegate(	4✔
167	self,
168	safe_address: str,	4✔
169	delegate_address: str,	4✔
170	label: str,	4✔
171	signer_account: LocalAccount,	4✔
172	):
173	hash_to_sign = self.create_delegate_message_hash(delegate_address)
174	signature = signer_account.signHash(hash_to_sign)
175	add_payload = {
176	"safe": safe_address,
177	"delegate": delegate_address,
178	"signature": signature.signature.hex(),
179	"label": label,
180	}
181	response = self._post_request(
182	f"/api/v1/safes/{safe_address}/delegates/", add_payload
183	)
184	if not response.ok:
185	raise BaseAPIException(f"Cannot add delegate: {response.content}")
186
187	def remove_delegate(	4✔
188	self, safe_address: str, delegate_address: str, signer_account: LocalAccount	4✔
189	):
190	hash_to_sign = self.create_delegate_message_hash(delegate_address)
191	signature = signer_account.signHash(hash_to_sign)
192	remove_payload = {"signature": signature.signature.hex()}
193	response = self._delete_request(
194	f"/api/v1/safes/{safe_address}/delegates/{delegate_address}/",
195	remove_payload,
196	)
197	if not response.ok:
198	raise BaseAPIException(f"Cannot remove delegate: {response.content}")
199
200	def post_transaction(self, safe_address: str, safe_tx: SafeTx):	4✔
201	url = urljoin(
202	self.base_url, f"/api/v1/safes/{safe_address}/multisig-transactions/"
203	)
204	random_account = "0x1b95E981F808192Dc5cdCF92ef589f9CBe6891C4"
205	sender = safe_tx.sorted_signers[0] if safe_tx.sorted_signers else random_account
206	data = {
207	"to": safe_tx.to,
208	"value": safe_tx.value,
209	"data": safe_tx.data.hex() if safe_tx.data else None,
210	"operation": safe_tx.operation,
211	"gasToken": safe_tx.gas_token,
212	"safeTxGas": safe_tx.safe_tx_gas,
213	"baseGas": safe_tx.base_gas,
214	"gasPrice": safe_tx.gas_price,
215	"refundReceiver": safe_tx.refund_receiver,
216	"nonce": safe_tx.safe_nonce,
217	"contractTransactionHash": safe_tx.safe_tx_hash.hex(),
218	"sender": sender,
219	"signature": safe_tx.signatures.hex() if safe_tx.signatures else None,
220	"origin": "Safe-CLI",
221	}
222	response = requests.post(url, json=data)
223	if not response.ok:
224	raise BaseAPIException(f"Error posting transaction: {response.content}")

safe-global / safe-cli / 6508017815

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