Repository: safe-global/safe-eth-py (language: python)

Issue:
Introduce method `SafeTx.api_post_data`
Currently, in order to post Safe transactions to the web interface (with this project as a dependency), we have to manually construct the `post_data` dictionary out of all the fields already contained in `safe_tx`. Since a `SafeTx` already contains the relevant info to build the post_data, it would be nice if there were some exposed method that would return the required format.

For example, the content of this method (taken from `safe-cli`) which constructs the post data would be nice to include here in this project:

https://github.com/5afe/safe-cli/blob/7a70a18c36b163ee91d5c763208ba1b1c9571317/safe_cli/api/transaction_service_api.py#L198-L222




Discussion / hints:

Fix patch:
diff --git a/gnosis/safe/api/__init__.py b/gnosis/safe/api/__init__.py
new file mode 100644
index 000000000..82e5967bf
--- /dev/null
+++ b/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",
+]
diff --git a/gnosis/safe/api/base_api.py b/gnosis/safe/api/base_api.py
new file mode 100644
index 000000000..9bc4a3871
--- /dev/null
+++ b/gnosis/safe/api/base_api.py
@@ -0,0 +1,58 @@
+from abc import ABC
+from typing import Dict, Optional
+from urllib.parse import urljoin
+
+import requests
+
+from gnosis.eth.ethereum_client import (
+    EthereumClient,
+    EthereumNetwork,
+    EthereumNetworkNotSupported,
+)
+
+
+class SafeAPIException(Exception):
+    pass
+
+
+class SafeBaseAPI(ABC):
+    URL_BY_NETWORK: Dict[EthereumNetwork, str] = {}
+
+    def __init__(
+        self,
+        network: EthereumNetwork,
+        ethereum_client: Optional[EthereumClient] = None,
+        base_url: Optional[str] = None,
+    ):
+        """
+        :param network: Network for the transaction service
+        :param ethereum_client:
+        :param base_url: If a custom transaction service is used
+        :raises: EthereumNetworkNotSupported
+        """
+        self.network = network
+        self.ethereum_client = ethereum_client
+        self.base_url = base_url or self.URL_BY_NETWORK.get(network)
+        if not self.base_url:
+            raise EthereumNetworkNotSupported(network)
+
+    @classmethod
+    def from_ethereum_client(cls, ethereum_client: EthereumClient) -> "SafeBaseAPI":
+        ethereum_network = ethereum_client.get_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"}
+        )
diff --git a/gnosis/safe/api/relay_service_api.py b/gnosis/safe/api/relay_service_api.py
new file mode 100644
index 000000000..6a7329fd4
--- /dev/null
+++ b/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)
diff --git a/gnosis/safe/api/transaction_service_api.py b/gnosis/safe/api/transaction_service_api.py
new file mode 100644
index 000000000..54886922e
--- /dev/null
+++ b/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}")


Test patch:
diff --git a/gnosis/safe/tests/api/__init__.py b/gnosis/safe/tests/api/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/gnosis/safe/tests/api/test_transaction_service_api.py b/gnosis/safe/tests/api/test_transaction_service_api.py
new file mode 100644
index 000000000..1f05ac36f
--- /dev/null
+++ b/gnosis/safe/tests/api/test_transaction_service_api.py
@@ -0,0 +1,104 @@
+from unittest import mock
+
+from django.test import TestCase
+
+from gnosis.eth import EthereumClient, EthereumNetwork, EthereumNetworkNotSupported
+from gnosis.eth.tests.ethereum_test_case import EthereumTestCaseMixin
+
+from ...api.transaction_service_api import TransactionServiceApi
+
+
+class TestTransactionServiceAPI(EthereumTestCaseMixin, TestCase):
+    def setUp(self) -> None:
+        self.transaction_service_api = TransactionServiceApi(
+            EthereumNetwork.RINKEBY, ethereum_client=self.ethereum_client
+        )
+        self.safe_address = "0x7552Ed65a45E27740a15B8D5415E90d8ca64C109"
+
+    def test_constructor(self):
+        ethereum_network = EthereumNetwork.RINKEBY
+        base_url = "https://safe.global"
+        transaction_service_api = TransactionServiceApi(
+            ethereum_network, ethereum_client=None, base_url=base_url
+        )
+        self.assertEqual(transaction_service_api.network, ethereum_network)
+        self.assertIsNone(transaction_service_api.ethereum_client)
+        self.assertEqual(transaction_service_api.base_url, base_url)
+
+    def test_from_ethereum_client(self):
+        with self.assertRaisesMessage(EthereumNetworkNotSupported, "GANACHE"):
+            TransactionServiceApi.from_ethereum_client(self.ethereum_client)
+
+        with mock.patch.object(
+            EthereumClient, "get_network", return_value=EthereumNetwork.RINKEBY
+        ):
+            transaction_service_api = TransactionServiceApi.from_ethereum_client(
+                self.ethereum_client
+            )
+            self.assertEqual(
+                transaction_service_api.ethereum_client, self.ethereum_client
+            )
+            self.assertEqual(transaction_service_api.network, EthereumNetwork.RINKEBY)
+
+    def test_data_decoded_to_text(self):
+        test_data = {
+            "method": "multiSend",
+            "parameters": [
+                {
+                    "name": "transactions",
+                    "type": "bytes",
+                    "value": "0x00c68877b75c3f9b950a798f9c9df4cde121c432ed000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000247de7edef00000000000000000000000034cfac646f301356faa8b21e94227e3583fe3f5f00c68877b75c3f9b950a798f9c9df4cde121c432ed00000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000024f08a0323000000000000000000000000d5d82b6addc9027b22dca772aa68d5d74cdbdf44",
+                    "decodedValue": [
+                        {
+                            "operation": "CALL",
+                            "to": "0xc68877B75c3f9b950a798f9C9dF4cDE121C432eD",
+                            "value": 0,
+                            "data": "0x7de7edef00000000000000000000000034cfac646f301356faa8b21e94227e3583fe3f5f",
+                            "decodedData": {
+                                "method": "changeMasterCopy",
+                                "parameters": [
+                                    {
+                                        "name": "_masterCopy",
+                                        "type": "address",
+                                        "value": "0x34CfAC646f301356fAa8B21e94227e3583Fe3F5F",
+                                    }
+                                ],
+                            },
+                        },
+                        {
+                            "operation": "CALL",
+                            "to": "0xc68877B75c3f9b950a798f9C9dF4cDE121C432eD",
+                            "value": 0,
+                            "data": "0xf08a0323000000000000000000000000d5d82b6addc9027b22dca772aa68d5d74cdbdf44",
+                            "decodedData": {
+                                "method": "setFallbackHandler",
+                                "parameters": [
+                                    {
+                                        "name": "handler",
+                                        "type": "address",
+                                        "value": "0xd5D82B6aDDc9027B22dCA772Aa68D5d74cdBdF44",
+                                    }
+                                ],
+                            },
+                        },
+                    ],
+                }
+            ],
+        }
+        decoded_data_text = self.transaction_service_api.data_decoded_to_text(test_data)
+        self.assertIn(
+            "- changeMasterCopy: 0x34CfAC646f301356fAa8B21e94227e3583Fe3F5F",
+            decoded_data_text,
+        )
+        self.assertIn(
+            "- setFallbackHandler: 0xd5D82B6aDDc9027B22dCA772Aa68D5d74cdBdF44",
+            decoded_data_text,
+        )
+
+    def test_get_balances(self):
+        balances = self.transaction_service_api.get_balances(self.safe_address)
+        self.assertIsInstance(balances, list)
+
+    def test_get_transactions(self):
+        transactions = self.transaction_service_api.get_transactions(self.safe_address)
+        self.assertIsInstance(transactions, list)