From f16aa081070189f61c31064cc8e6eac0925eab24 Mon Sep 17 00:00:00 2001 From: Uxio Fuentefria Date: Wed, 11 Oct 2023 16:39:21 +0200 Subject: [PATCH] Increase coverage - Use TransactionServiceApi from safe-eth-py --- safe_cli/api/__init__.py | 0 safe_cli/api/base_api.py | 44 ---- safe_cli/api/transaction_service_api.py | 224 ------------------ safe_cli/operators/safe_operator.py | 39 ++- .../operators/safe_tx_service_operator.py | 6 +- safe_cli/prompt_parser.py | 5 +- tests/api/test_gnosis_transaction.py | 3 +- 7 files changed, 40 insertions(+), 281 deletions(-) delete mode 100644 safe_cli/api/__init__.py delete mode 100644 safe_cli/api/base_api.py delete mode 100644 safe_cli/api/transaction_service_api.py diff --git a/safe_cli/api/__init__.py b/safe_cli/api/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/safe_cli/api/base_api.py b/safe_cli/api/base_api.py deleted file mode 100644 index 4cfa89f4..00000000 --- a/safe_cli/api/base_api.py +++ /dev/null @@ -1,44 +0,0 @@ -from abc import ABC -from typing import Dict, Optional -from urllib.parse import urljoin - -import requests - -from gnosis.eth.ethereum_client import EthereumClient, EthereumNetwork - - -class BaseAPIException(Exception): - pass - - -class BaseAPI(ABC): - URL_BY_NETWORK: Dict[EthereumNetwork, str] = {} - - def __init__(self, ethereum_client: EthereumClient, network: EthereumNetwork): - self.ethereum_client = ethereum_client - self.network = network - self.base_url = self.URL_BY_NETWORK[network] - - @classmethod - def from_ethereum_client( - cls, ethereum_client: EthereumClient - ) -> Optional["BaseAPI"]: - ethereum_network = ethereum_client.get_network() - if ethereum_network in cls.URL_BY_NETWORK: - return cls(ethereum_client, ethereum_network) - - def _get_request(self, url: str) -> requests.Response: - full_url = urljoin(self.base_url, url) - return requests.get(full_url) - - def _post_request(self, url: str, payload: Dict) -> requests.Response: - full_url = urljoin(self.base_url, url) - return requests.post( - full_url, json=payload, headers={"Content-type": "application/json"} - ) - - def _delete_request(self, url: str, payload: Dict) -> requests.Response: - full_url = urljoin(self.base_url, url) - return requests.delete( - full_url, json=payload, headers={"Content-type": "application/json"} - ) diff --git a/safe_cli/api/transaction_service_api.py b/safe_cli/api/transaction_service_api.py deleted file mode 100644 index c22fce80..00000000 --- a/safe_cli/api/transaction_service_api.py +++ /dev/null @@ -1,224 +0,0 @@ -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}") diff --git a/safe_cli/operators/safe_operator.py b/safe_cli/operators/safe_operator.py index fc05b0be..b8167c17 100644 --- a/safe_cli/operators/safe_operator.py +++ b/safe_cli/operators/safe_operator.py @@ -12,9 +12,15 @@ from packaging import version as semantic_version from prompt_toolkit import HTML, print_formatted_text from web3 import Web3 +from web3.contract import Contract from web3.exceptions import BadFunctionCallOutput -from gnosis.eth import EthereumClient, EthereumNetwork, TxSpeed +from gnosis.eth import ( + EthereumClient, + EthereumNetwork, + EthereumNetworkNotSupported, + TxSpeed, +) from gnosis.eth.clients import EtherscanClient, EtherscanClientConfigurationProblem from gnosis.eth.constants import NULL_ADDRESS, SENTINEL_ADDRESS from gnosis.eth.contracts import ( @@ -24,9 +30,9 @@ get_safe_V1_1_1_contract, ) from gnosis.safe import InvalidInternalTx, Safe, SafeOperation, SafeTx +from gnosis.safe.api import TransactionServiceApi from gnosis.safe.multi_send import MultiSend, MultiSendOperation, MultiSendTx -from safe_cli.api.transaction_service_api import TransactionServiceApi from safe_cli.ethereum_hd_wallet import get_account_from_words from safe_cli.safe_addresses import ( get_default_fallback_handler_address, @@ -176,7 +182,23 @@ def decorated(self, *args, **kwargs): class SafeOperator: - def __init__(self, address: str, node_url: str): + address: ChecksumAddress + node_url: str + ethereum_client: EthereumClient + ens: ENS + network: EthereumNetwork + etherscan: Optional[EtherscanClient] + safe_tx_service: Optional[TransactionServiceApi] + safe: Safe + safe_contract: Contract + safe_contract_1_1_0: Contract + accounts: Set[LocalAccount] = set() + default_sender: Optional[LocalAccount] + executed_transactions: List[str] + _safe_cli_info: Optional[SafeCliInfo] + require_all_signatures: bool + + def __init__(self, address: ChecksumAddress, node_url: str): self.address = address self.node_url = node_url self.ethereum_client = EthereumClient(self.node_url) @@ -186,9 +208,14 @@ def __init__(self, address: str, node_url: str): self.etherscan = EtherscanClient(self.network) except EtherscanClientConfigurationProblem: self.etherscan = None - self.safe_tx_service = TransactionServiceApi.from_ethereum_client( - self.ethereum_client - ) + + try: + self.safe_tx_service = TransactionServiceApi.from_ethereum_client( + self.ethereum_client + ) + except EthereumNetworkNotSupported: + self.safe_tx_service = None + self.safe = Safe(address, self.ethereum_client) self.safe_contract = self.safe.contract self.safe_contract_1_1_0 = get_safe_V1_1_1_contract( diff --git a/safe_cli/operators/safe_tx_service_operator.py b/safe_cli/operators/safe_tx_service_operator.py index 7ffa4eec..804508fb 100644 --- a/safe_cli/operators/safe_tx_service_operator.py +++ b/safe_cli/operators/safe_tx_service_operator.py @@ -7,9 +7,9 @@ from gnosis.eth.contracts import get_erc20_contract from gnosis.safe import SafeOperation, SafeTx +from gnosis.safe.api import SafeAPIException from gnosis.safe.multi_send import MultiSend, MultiSendOperation, MultiSendTx -from safe_cli.api.base_api import BaseAPIException from safe_cli.utils import yes_or_no_question from .safe_operator import ( @@ -58,7 +58,7 @@ def add_delegate(self, delegate_address: str, label: str, signer_address: str): self.address, delegate_address, label, signer_account ) return True - except BaseAPIException: + except SafeAPIException: return False def remove_delegate(self, delegate_address: str, signer_address: str): @@ -76,7 +76,7 @@ def remove_delegate(self, delegate_address: str, signer_address: str): self.address, delegate_address, signer_account ) return True - except BaseAPIException: + except SafeAPIException: return False def submit_signatures(self, safe_tx_hash: bytes) -> bool: diff --git a/safe_cli/prompt_parser.py b/safe_cli/prompt_parser.py index b6a372a6..64fcfaab 100644 --- a/safe_cli/prompt_parser.py +++ b/safe_cli/prompt_parser.py @@ -5,7 +5,8 @@ from prompt_toolkit import HTML, print_formatted_text from web3 import Web3 -from .api.base_api import BaseAPIException +from gnosis.safe.api import SafeAPIException + from .operators.safe_operator import ( AccountNotLoadedException, ExistingOwnerException, @@ -77,7 +78,7 @@ def safe_exception(function): def wrapper(*args, **kwargs): try: return function(*args, **kwargs) - except BaseAPIException as e: + except SafeAPIException as e: if e.args: print_formatted_text(HTML(f"{e.args[0]}")) except AccountNotLoadedException as e: diff --git a/tests/api/test_gnosis_transaction.py b/tests/api/test_gnosis_transaction.py index 559ba1dd..09e66b0e 100644 --- a/tests/api/test_gnosis_transaction.py +++ b/tests/api/test_gnosis_transaction.py @@ -2,8 +2,7 @@ from unittest import mock from gnosis.eth import EthereumClient, EthereumNetwork - -from safe_cli.api.transaction_service_api import TransactionServiceApi +from gnosis.safe.api import TransactionServiceApi class TestTransactionService(unittest.TestCase):