Add decode data method in Tx Service Api

gnosis/safe/api/transaction_service_api/entities.py

@@ -0,0 +1,97 @@
+from typing import Any, List, Optional, TypedDict
+
+from eth_typing import AnyAddress, HexStr
+
+
+class ParameterDecoded(TypedDict):
+    name: str
+    type: str
+    value: Any
+
+
+class DataDecoded(TypedDict):
+    method: str
+    parameters: List[ParameterDecoded]
+
+
+class Erc20Info(TypedDict):
+    name: str
+    symbol: str
+    decimals: int
+    logo_uri: str
+
+
+class Balance(TypedDict):
+    token_address: Optional[AnyAddress]
+    token: Optional[Erc20Info]
+    balance: int
+
+
+class DelegateUser(TypedDict):
+    safe: Optional[AnyAddress]
+    delegate: AnyAddress
+    delegator: AnyAddress
+    label: str
+
+
+class MessageConfirmation(TypedDict):
+    created: str
+    modified: str
+    owner: AnyAddress
+    signature: HexStr
+    signatureType: str
+
+
+class Message(TypedDict):
+    created: str
+    modified: str
+    safe: AnyAddress
+    messageHash: HexStr
+    message: Any
+    proposedBy: AnyAddress
+    safeAppId: int
+    confirmations: Optional[List[MessageConfirmation]]
+    preparedSignature: Optional[HexStr]
+
+
+class TransactionConfirmation(TypedDict):
+    owner: AnyAddress
+    submissionDate: str
+    transactionHash: HexStr
+    signature: HexStr
+    signatureType: str
+
+
+class Transaction(TypedDict):
+    safe: AnyAddress
+    to: AnyAddress
+    value: str
+    data: Optional[HexStr]
+    operation: int
+    gasToken: Optional[AnyAddress]
+    safeTxGas: int
+    baseGas: int
+    gasPrice: str
+    refundReceiver: Optional[AnyAddress]
+    nonce: int
+    execution_date: str
+    submission_date: str
+    modified: str
+    blockNumber: Optional[int]
+    transactionHash: HexStr
+    safeTxHash: HexStr
+    proposer: AnyAddress
+    executor: Optional[AnyAddress]
+    isExecuted: bool
+    isSuccessful: Optional[bool]
+    ethGasPrice: Optional[str]
+    maxFeePerGas: Optional[str]
+    maxPriorityFeePerGas: Optional[str]
+    gasUsed: Optional[int]
+    fee: Optional[int]
+    origin: Optional[str]
+    dataDecoded: Optional[List[DataDecoded]]
+    confirmationsRequired: int
+    confirmations: Optional[List[TransactionConfirmation]]
+    trusted: bool
+    signatures: Optional[HexStr]

gnosis/safe/api/transaction_service_api/transaction_service_api.py

@@ -12,6 +12,7 @@
 from gnosis.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
 
@@ -92,7 +93,7 @@ def data_decoded_to_text(cls, data_decoded: Dict[str, Any]) -> Optional[str]:
             )
 
     @classmethod
-    def parse_signatures(cls, raw_tx: Dict[str, Any]) -> Optional[bytes]:
+    def parse_signatures(cls, raw_tx: Transaction) -> Optional[bytes]:
         """
         Parse signatures in `confirmations` list to build a valid signature (owners must be sorted lexicographically)
 
@@ -120,7 +121,7 @@ def create_delegate_message_hash(self, delegate_address: ChecksumAddress) -> Has
         )
 
     def _build_transaction_service_tx(
-        self, safe_tx_hash: Union[bytes, HexStr], tx_raw: Dict[str, Any]
+        self, safe_tx_hash: Union[bytes, HexStr], tx_raw: Transaction
     ) -> TransactionServiceTx:
         signatures = self.parse_signatures(tx_raw)
         safe_tx = TransactionServiceTx(
@@ -151,7 +152,7 @@ def _build_transaction_service_tx(
 
         return safe_tx
 
-    def get_balances(self, safe_address: str) -> List[Dict[str, Any]]:
+    def get_balances(self, safe_address: str) -> List[Balance]:
         """
 
         :param safe_address:
@@ -190,7 +191,7 @@ def get_safe_transaction(
 
     def get_transactions(
         self, safe_address: ChecksumAddress, **kwargs: Dict[str, Union[str, int, bool]]
-    ) -> List[Dict[str, Any]]:
+    ) -> List[Transaction]:
         """
 
         :param safe_address:
@@ -220,7 +221,7 @@ def get_transactions(
 
         return transactions
 
-    def get_delegates(self, safe_address: ChecksumAddress) -> List[Dict[str, Any]]:
+    def get_delegates(self, safe_address: ChecksumAddress) -> List[DelegateUser]:
         """
 
         :param safe_address:
@@ -373,7 +374,7 @@ def post_message(
             raise SafeAPIException(f"Error posting message: {response.content}")
         return True
 
-    def get_message(self, safe_message_hash: bytes) -> Dict[str, Any]:
+    def get_message(self, safe_message_hash: bytes) -> Message:
         """
 
         :param safe_message_hash:
@@ -386,7 +387,7 @@ def get_message(self, safe_message_hash: bytes) -> Dict[str, Any]:
             raise SafeAPIException(f"Cannot get messages: {response.content}")
         return response.json()
 
-    def get_messages(self, safe_address: ChecksumAddress) -> List[Dict[str, Any]]:
+    def get_messages(self, safe_address: ChecksumAddress) -> List[Message]:
         """
 
         :param safe_address:
@@ -416,3 +417,21 @@ def post_message_signature(
                 f"Error posting message signature: {response.content}"
             )
         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}")
+        return response.json()

gnosis/safe/tests/api/test_transaction_service_api.py

@@ -4,6 +4,7 @@
 
 from django.test import TestCase
 
+from eth_typing import HexStr
 from hexbytes import HexBytes
 
 from gnosis.eth import EthereumClient, EthereumNetwork, EthereumNetworkNotSupported
@@ -15,7 +16,11 @@
 )
 
 from ...api import SafeAPIException
-from ..mocks.mock_transactions import transaction_data_mock, transaction_mock
+from ..mocks.mock_transactions import (
+    transaction_data_decoded_mock,
+    transaction_data_mock,
+    transaction_mock,
+)
 
 
 class TestTransactionServiceAPI(EthereumTestCaseMixin, TestCase):
@@ -183,3 +188,30 @@ def test_get_safe_transaction(self):
                 f"Cannot get transaction with safe-tx-hash={safe_tx_hash.hex()}:",
                 str(context.exception),
             )
+
+    def test_decode_data(self):
+        with patch.object(TransactionServiceApi, "_post_request") as mock_post_request:
+            mock_post_request.return_value.ok = True
+            mock_post_request.return_value.json = MagicMock(
+                return_value=transaction_data_decoded_mock
+            )
+            data = HexStr(
+                "0x095ea7b3000000000000000000000000e6fc577e87f7c977c4393300417dcc592d90acf8ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff"
+            )
+            to_address = "0x4127839cdf4F73d9fC9a2C2861d8d1799e9DF40C"
+            decoded_data = self.transaction_service_api.decode_data(data, to_address)
+            expected_url = "/api/v1/data-decoder/"
+            expected_payload = {"data": HexBytes(data).hex(), "to": to_address}
+            mock_post_request.assert_called_once_with(expected_url, expected_payload)
+            self.assertEqual(decoded_data.get("method"), "approve")
+            self.assertEqual(decoded_data.get("parameters")[0].get("name"), "spender")
+            self.assertEqual(decoded_data.get("parameters")[1].get("name"), "value")
+
+            # Test response not ok
+            mock_post_request.return_value.ok = False
+            with self.assertRaises(SafeAPIException) as context:
+                self.transaction_service_api.decode_data(data, to_address)
+            self.assertIn(
+                "Cannot decode tx data:",
+                str(context.exception),
+            )

gnosis/safe/tests/mocks/mock_transactions.py

@@ -104,3 +104,19 @@
     "trusted": True,
     "signatures": "0x000000000000000000000000c6b82ba149cfa113f8f48d5e3b1f78e933e16dfd000000000000000000000000000000000000000000000000000000000000000001",
 }
+
+transaction_data_decoded_mock = {
+    "method": "approve",
+    "parameters": [
+        {
+            "name": "spender",
+            "type": "address",
+            "value": "0xe6fC577E87F7c977c4393300417dCC592D90acF8",
+        },
+        {
+            "name": "value",
+            "type": "uint256",
+            "value": "115792089237316195423570985008687907853269984665640564039457584007913129639935",
+        },
+    ],
+}