Merge 804908b into 3f01e6f

gnosis/safe/api/__init__.py

@@ -0,0 +1,12 @@
+# flake8: noqa F401
+from .base_api import SafeAPIException
+from .relay_service_api import RelayEstimation, RelaySentTransaction, RelayServiceApi
+from .transaction_service_api import TransactionServiceApi
+
+__all__ = [
+    "SafeAPIException",
+    "RelayServiceApi",
+    "RelayEstimation",
+    "RelaySentTransaction",
+    "TransactionServiceApi",
+]

gnosis/safe/api/base_api.py

@@ -0,0 +1,46 @@
+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 SafeAPIException(Exception):
+    pass
+
+
+class SafeBaseAPI(ABC):
+    URL_BY_NETWORK: Dict[EthereumNetwork, str] = {}
+
+    def __init__(
+        self, network: EthereumNetwork, ethereum_client: Optional[EthereumClient] = None
+    ):
+        self.network = network
+        self.ethereum_client = ethereum_client
+        self.base_url = self.URL_BY_NETWORK[network]
+
+    @classmethod
+    def from_ethereum_client(
+        cls, ethereum_client: EthereumClient
+    ) -> Optional["SafeBaseAPI"]:
+        ethereum_network = ethereum_client.get_network()
+        if ethereum_network in cls.URL_BY_NETWORK:
+            return cls(ethereum_network, ethereum_client=ethereum_client)
+
+    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"}
+        )

gnosis/safe/api/relay_service_api.py

@@ -0,0 +1,97 @@
+from urllib.parse import urljoin
+
+import requests
+from eth_typing import ChecksumAddress, HexStr
+
+from gnosis.eth.ethereum_client import EthereumNetwork
+
+from .. import SafeTx
+from ..signatures import signature_split
+from .base_api import SafeAPIException, SafeBaseAPI
+
+try:
+    from typing import TypedDict
+except ImportError:
+    from typing_extensions import TypedDict
+
+
+class RelayEstimation(TypedDict):
+    safeTxGas: int
+    baseGas: int
+    gasPrice: int
+    lastUsedNonce: int
+    gasToken: ChecksumAddress
+    refundReceiver: ChecksumAddress
+
+
+class RelaySentTransaction(TypedDict):
+    safeTxHash: HexStr
+    txHash: HexStr
+
+
+class RelayServiceApi(SafeBaseAPI):
+    URL_BY_NETWORK = {
+        EthereumNetwork.GOERLI: "https://safe-relay.goerli.gnosis.io/",
+        EthereumNetwork.MAINNET: "https://safe-relay.gnosis.io",
+        EthereumNetwork.RINKEBY: "https://safe-relay.rinkeby.gnosis.io",
+    }
+
+    def send_transaction(
+        self, safe_address: str, safe_tx: SafeTx
+    ) -> RelaySentTransaction:
+        url = urljoin(self.base_url, f"/api/v1/safes/{safe_address}/transactions/")
+        signatures = []
+        for i in range(len(safe_tx.signatures) // 65):
+            v, r, s = signature_split(safe_tx.signatures, i)
+            signatures.append(
+                {
+                    "v": v,
+                    "r": r,
+                    "s": s,
+                }
+            )
+
+        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,
+            "dataGas": safe_tx.base_gas,
+            "gasPrice": safe_tx.gas_price,
+            "refundReceiver": safe_tx.refund_receiver,
+            "nonce": safe_tx.safe_nonce,
+            "signatures": signatures,
+        }
+        response = requests.post(url, json=data)
+        if not response.ok:
+            raise SafeAPIException(f"Error posting transaction: {response.content}")
+        else:
+            return RelaySentTransaction(response.json())
+
+    def get_estimation(self, safe_address: str, safe_tx: SafeTx) -> RelayEstimation:
+        """
+        :param safe_address:
+        :param safe_tx:
+        :return: RelayEstimation
+        """
+        url = urljoin(
+            self.base_url, f"/api/v2/safes/{safe_address}/transactions/estimate/"
+        )
+        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,
+        }
+        response = requests.post(url, json=data)
+        if not response.ok:
+            raise SafeAPIException(f"Error posting transaction: {response.content}")
+        else:
+            response_json = response.json()
+            # Convert values to int
+            for key in ("safeTxGas", "baseGas", "gasPrice"):
+                response_json[key] = int(response_json[key])
+            return RelayEstimation(response_json)

gnosis/safe/api/transaction_service_api.py

@@ -0,0 +1,233 @@
+import logging
+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 import EthereumNetwork
+from gnosis.safe import SafeTx
+
+from .base_api import SafeAPIException, SafeBaseAPI
+
+logger = logging.getLogger(__name__)
+
+
+class TransactionServiceApi(SafeBaseAPI):
+    URL_BY_NETWORK = {
+        EthereumNetwork.ARBITRUM: "https://safe-transaction.arbitrum.gnosis.io",
+        EthereumNetwork.AURORA: "https://safe-transaction.aurora.gnosis.io/",
+        EthereumNetwork.AVALANCHE: "https://safe-transaction.avalanche.gnosis.io",
+        EthereumNetwork.BINANCE: "https://safe-transaction.bsc.gnosis.io",
+        EthereumNetwork.ENERGY_WEB_CHAIN: "https://safe-transaction.ewc.gnosis.io",
+        EthereumNetwork.GOERLI: "https://safe-transaction.goerli.gnosis.io",
+        EthereumNetwork.MAINNET: "https://safe-transaction.mainnet.gnosis.io",
+        EthereumNetwork.MATIC: "https://safe-transaction.polygon.gnosis.io",
+        EthereumNetwork.OPTIMISTIC: "https://safe-transaction.optimism.gnosis.io/",
+        EthereumNetwork.RINKEBY: "https://safe-transaction.rinkeby.gnosis.io",
+        EthereumNetwork.VOLTA: "https://safe-transaction.volta.gnosis.io",
+        EthereumNetwork.XDAI: "https://safe-transaction.xdai.gnosis.io",
+    }
+
+    @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
+
+    @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: 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_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 SafeAPIException(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 SafeAPIException(
+                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)
+            if not self.ethereum_client:
+                logger.warning(
+                    "EthereumClient should be defined to get a executable SafeTx"
+                )
+            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"]),
+                    chain_id=self.network.value,
+                ),
+                HexBytes(result["transactionHash"])
+                if result["transactionHash"]
+                else None,
+            )
+
+    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 SafeAPIException(f"Cannot get transactions: {response.content}")
+        else:
+            return response.json().get("results", [])
+
+    def get_delegates(self, safe_address: str) -> List[Dict[str, Any]]:
+        response = self._get_request(f"/api/v1/safes/{safe_address}/delegates/")
+        if not response.ok:
+            raise SafeAPIException(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 SafeAPIException(
+                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 SafeAPIException(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 SafeAPIException(f"Cannot remove delegate: {response.content}")
+
+    def post_transaction(self, safe_tx: SafeTx):
+        url = urljoin(
+            self.base_url,
+            f"/api/v1/safes/{safe_tx.safe_address}/multisig-transactions/",
+        )
+        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 = requests.post(url, json=data)
+        if not response.ok:
+            raise SafeAPIException(f"Error posting transaction: {response.content}")