Refactor transaction service delegate users api methods to use v2 (#973)

gnosis/safe/api/transaction_service_api/transaction_service_api.py

@@ -4,11 +4,11 @@
 from urllib.parse import urlencode
 
 from eth_account.signers.local import LocalAccount
-from eth_typing import ChecksumAddress, HexStr
+from eth_typing import ChecksumAddress, Hash32, HexStr
 from hexbytes import HexBytes
 
 from gnosis.eth import EthereumClient, EthereumNetwork
-from gnosis.eth.utils import fast_keccak_text
+from gnosis.eth.eip712 import eip712_encode_hash
 from gnosis.safe import SafeTx
 
 from ..base_api import SafeAPIException, SafeBaseAPI
@@ -52,10 +52,6 @@ def __init__(
     ):
         super().__init__(network, ethereum_client, base_url, request_timeout)
 
-    @classmethod
-    def create_delegate_message_hash(cls, delegate_address: ChecksumAddress) -> str:
-        return fast_keccak_text(get_delegate_message(delegate_address))
-
     @classmethod
     def data_decoded_to_text(cls, data_decoded: Dict[str, Any]) -> Optional[str]:
         """
@@ -118,6 +114,11 @@ def parse_signatures(cls, raw_tx: Dict[str, Any]) -> Optional[bytes]:
                 ]
             )
 
+    def create_delegate_message_hash(self, delegate_address: ChecksumAddress) -> Hash32:
+        return eip712_encode_hash(
+            get_delegate_message(delegate_address, self.network.value)
+        )
+
     def _build_transaction_service_tx(
         self, safe_tx_hash: Union[bytes, HexStr], tx_raw: Dict[str, Any]
     ) -> TransactionServiceTx:
@@ -225,7 +226,7 @@ def get_delegates(self, safe_address: ChecksumAddress) -> List[Dict[str, Any]]:
         :param safe_address:
         :return: a list of delegates for provided Safe
         """
-        response = self._get_request(f"/api/v1/delegates/?safe={safe_address}")
+        response = self._get_request(f"/api/v2/delegates/?safe={safe_address}")
         if not response.ok:
             raise SafeAPIException(f"Cannot get delegates: {response.content}")
         return response.json().get("results", [])
@@ -278,7 +279,7 @@ def add_delegate(
             "signature": signature.signature.hex(),
             "label": label,
         }
-        response = self._post_request("/api/v1/delegates/", add_payload)
+        response = self._post_request("/api/v2/delegates/", add_payload)
         if not response.ok:
             raise SafeAPIException(f"Cannot add delegate: {response.content}")
         return True
@@ -291,9 +292,13 @@ def remove_delegate(
     ) -> bool:
         hash_to_sign = self.create_delegate_message_hash(delegate_address)
         signature = signer_account.signHash(hash_to_sign)
-        remove_payload = {"signature": signature.signature.hex()}
+        remove_payload = {
+            "safe": safe_address,
+            "delegator": signer_account.address,
+            "signature": signature.signature.hex(),
+        }
         response = self._delete_request(
-            f"/api/v1/safes/{safe_address}/delegates/{delegate_address}/",
+            f"/api/v2/delegates/{delegate_address}/",
             remove_payload,
         )
         if not response.ok:

gnosis/safe/api/transaction_service_api/transaction_service_messages.py

@@ -2,7 +2,7 @@
 Utilities to build EIP712 messages for Safe
 """
 import time
-from typing import Dict
+from typing import Any, Dict
 
 from eth_typing import ChecksumAddress
 
@@ -15,20 +15,47 @@ def get_totp() -> int:
     return int(time.time()) // 3600
 
 
-def get_delegate_message(delegate_address: ChecksumAddress) -> str:
+def get_delegate_message(
+    delegate_address: ChecksumAddress, chain_id: int
+) -> Dict[str, Any]:
     """
     Retrieves the required message for creating or removing a delegate on Safe Transaction Service.
 
     :param delegate_address:
-    :return: generated str message
+    :param chain_id:
+    :return: generated EIP712 message
     """
-    totp = get_totp()
-    return delegate_address + str(totp)
+
+    delegate_message = {
+        "types": {
+            "EIP712Domain": [
+                {"name": "name", "type": "string"},
+                {"name": "version", "type": "string"},
+                {"name": "chainId", "type": "uint256"},
+            ],
+            "Delegate": [
+                {"name": "delegateAddress", "type": "bytes32"},
+                {"name": "totp", "type": "uint256"},
+            ],
+        },
+        "primaryType": "Delegate",
+        "domain": {
+            "name": "Safe Transaction Service",
+            "version": "1.0",
+            "chainId": chain_id,
+        },
+        "message": {
+            "delegateAddress": delegate_address,
+            "totp": get_totp(),
+        },
+    }
+
+    return delegate_message
 
 
 def get_remove_transaction_message(
     safe_address: ChecksumAddress, safe_tx_hash: bytes, chain_id: int
-) -> Dict:
+) -> Dict[str, Any]:
     """
     Retrieves the required message for removing a not executed transaction on Safe Transaction Service.
 

gnosis/safe/tests/api/test_transaction_service_messages.py

@@ -0,0 +1,104 @@
+from unittest import TestCase
+from unittest.mock import patch
+
+from eth_typing import ChecksumAddress
+from hexbytes import HexBytes
+
+from gnosis.safe.api.transaction_service_api.transaction_service_messages import (
+    get_delegate_message,
+    get_remove_transaction_message,
+    get_totp,
+)
+
+
+class TestTransactionServiceMessages(TestCase):
+    def test_get_totp(self):
+        mocked_time = 1615974000
+        with patch("time.time", return_value=mocked_time):
+            totp = get_totp()
+
+        expected_totp = mocked_time // 3600
+        self.assertEqual(totp, expected_totp)
+
+    def test_get_delegate_message(self):
+        chain_id = 1
+        mocked_totp = 123
+        delegate_address = ChecksumAddress("0x1234567890123456789012345678901234567890")
+
+        expected_message = {
+            "types": {
+                "EIP712Domain": [
+                    {"name": "name", "type": "string"},
+                    {"name": "version", "type": "string"},
+                    {"name": "chainId", "type": "uint256"},
+                ],
+                "Delegate": [
+                    {"name": "delegateAddress", "type": "bytes32"},
+                    {"name": "totp", "type": "uint256"},
+                ],
+            },
+            "primaryType": "Delegate",
+            "domain": {
+                "name": "Safe Transaction Service",
+                "version": "1.0",
+                "chainId": chain_id,
+            },
+            "message": {
+                "delegateAddress": delegate_address,
+                "totp": mocked_totp,
+            },
+        }
+
+        with patch(
+            "gnosis.safe.api.transaction_service_api.transaction_service_messages.get_totp",
+            return_value=mocked_totp,
+        ):
+            message = get_delegate_message(delegate_address, chain_id)
+
+        self.assertEqual(message, expected_message)
+
+    def test_get_remove_transaction_message(self):
+        chain_id = 1
+        mocked_totp = 123
+        delegate_address = ChecksumAddress("0x1234567890123456789012345678901234567890")
+        safe_address = ChecksumAddress("0x1234567890123456789012345678901234567890")
+
+        safe_tx_hash = HexBytes(
+            "0x4c9577d1b1b8dec52329a983ae26238b65f74b7dd9fb28d74ad9548e92aaf196"
+        )
+
+        expected_message = {
+            "types": {
+                "EIP712Domain": [
+                    {"name": "name", "type": "string"},
+                    {"name": "version", "type": "string"},
+                    {"name": "chainId", "type": "uint256"},
+                    {"name": "verifyingContract", "type": "address"},
+                ],
+                "DeleteRequest": [
+                    {"name": "safeTxHash", "type": "bytes32"},
+                    {"name": "totp", "type": "uint256"},
+                ],
+            },
+            "primaryType": "DeleteRequest",
+            "domain": {
+                "name": "Safe Transaction Service",
+                "version": "1.0",
+                "chainId": chain_id,
+                "verifyingContract": safe_address,
+            },
+            "message": {
+                "safeTxHash": safe_tx_hash,
+                "totp": mocked_totp,
+            },
+        }
+
+        with patch(
+            "gnosis.safe.api.transaction_service_api.transaction_service_messages.get_totp",
+            return_value=mocked_totp,
+        ):
+            message = get_remove_transaction_message(
+                delegate_address, safe_tx_hash, chain_id
+            )
+
+        self.assertEqual(message, expected_message)