Allow deleting multisig transactions not executed

safe_transaction_service/history/helpers.py

@@ -2,14 +2,16 @@
 import time
 from typing import List
 
-from eth_typing import ChecksumAddress
+from eth_typing import ChecksumAddress, Hash32, HexStr
 from eth_utils import keccak
 
+from gnosis.eth.eip712 import eip712_encode_hash
+
 from safe_transaction_service.history.models import TransferDict
 from safe_transaction_service.tokens.models import Token
 
 
-class DelegateSignatureHelper:
+class TemporarySignatureHelper:
     @classmethod
     def calculate_totp(
         cls, totp_tx: int = 3600, totp_t0: int = 0, previous: bool = False
@@ -28,6 +30,57 @@ def calculate_totp(
 
         return int((time.time() - totp_t0) // totp_tx)
 
+
+class DeleteMultisigTxSignatureHelper(TemporarySignatureHelper):
+    @classmethod
+    def calculate_hash(
+        cls,
+        safe_address: ChecksumAddress,
+        safe_tx_hash: HexStr,
+        chain_id: int,
+        previous_totp: bool = False,
+    ) -> Hash32:
+        """
+        Builds a EIP712 object and calculates its hash
+
+        :param safe_address:
+        :param safe_tx_hash:
+        :param chain_id:
+        :param previous_totp:
+        :return: Hash for the EIP712 generated object from the provided parameters
+        """
+        totp = cls.calculate_totp(previous=previous_totp)
+
+        payload = {
+            "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": totp,
+            },
+        }
+
+        return eip712_encode_hash(payload)
+
+
+class DelegateSignatureHelper(TemporarySignatureHelper):
     @classmethod
     def calculate_hash(
         cls,

safe_transaction_service/history/serializers.py

@@ -2,12 +2,15 @@
 from enum import Enum
 from typing import Any, Dict, List, Optional
 
+from django.http import Http404
+
 from drf_yasg.utils import swagger_serializer_method
 from eth_typing import ChecksumAddress, HexStr
 from rest_framework import serializers
 from rest_framework.exceptions import NotFound, ValidationError
 
 from gnosis.eth import EthereumClient, EthereumClientProvider
+from gnosis.eth.constants import NULL_ADDRESS
 from gnosis.eth.django.models import EthereumAddressV2Field as EthereumAddressDbField
 from gnosis.eth.django.models import Keccak256Field as Keccak256DbField
 from gnosis.eth.django.serializers import (
@@ -27,7 +30,7 @@
 from safe_transaction_service.utils.serializers import get_safe_owners, get_safe_version
 
 from .exceptions import NodeConnectionException
-from .helpers import DelegateSignatureHelper
+from .helpers import DelegateSignatureHelper, DeleteMultisigTxSignatureHelper
 from .models import (
     EthereumTx,
     ModuleTransaction,
@@ -352,7 +355,7 @@ def check_delegate_signature(
         delegator: ChecksumAddress,
     ) -> bool:
         """
-        Checks signature and returns a valid owner if found, None otherwise
+        Verifies signature to check if it matches the delegator
 
         :param ethereum_client:
         :param signature:
@@ -472,6 +475,51 @@ def validate(self, attrs):
         )
 
 
+class SafeMultisigTransactionDeleteSerializer(serializers.Serializer):
+    safe_tx_hash = Sha3HashField()
+    signature = HexadecimalField(min_length=65)
+
+    def validate(self, attrs):
+        super().validate(attrs)
+        safe_tx_hash = attrs["safe_tx_hash"]
+        signature = attrs["signature"]
+
+        try:
+            multisig_tx = MultisigTransaction.objects.select_related("ethereum_tx").get(
+                safe_tx_hash=safe_tx_hash
+            )
+        except MultisigTransaction.DoesNotExist:
+            raise Http404("Multisig transaction not found")
+
+        if multisig_tx.executed:
+            raise ValidationError("Executed transactions cannot be deleted")
+
+        proposer = multisig_tx.proposer
+        if not proposer or proposer == NULL_ADDRESS:
+            raise ValidationError("Old transactions without proposer cannot be deleted")
+
+        ethereum_client = EthereumClientProvider()
+        chain_id = ethereum_client.get_chain_id()
+        safe_address = multisig_tx.safe
+        # Accept a message with the current topt and the previous totp (to prevent replay attacks)
+        for previous_totp in (True, False):
+            message_hash = DeleteMultisigTxSignatureHelper.calculate_hash(
+                safe_address, safe_tx_hash, chain_id, previous_totp=previous_totp
+            )
+            safe_signatures = SafeSignature.parse_signature(signature, message_hash)
+            if len(safe_signatures) != 1:
+                raise ValidationError(
+                    f"1 owner signature was expected, {len(safe_signatures)} received"
+                )
+            safe_signature = safe_signatures[0]
+            if safe_signature.signature_type != SafeSignatureType.EOA:
+                raise ValidationError("Only EOA signatures are supported")
+            if safe_signature.owner == proposer:
+                return attrs
+
+        raise ValidationError("Provided owner is not the proposer of the transaction")
+
+
 class DataDecoderSerializer(serializers.Serializer):
     data = HexadecimalField(allow_null=False, allow_blank=False, min_length=4)
     to = EthereumAddressField(allow_null=True, required=False)

safe_transaction_service/history/tests/test_views.py

@@ -16,6 +16,7 @@
 from hexbytes import HexBytes
 from requests import ReadTimeout
 from rest_framework import status
+from rest_framework.exceptions import ErrorDetail
 from rest_framework.test import APIRequestFactory, APITestCase, force_authenticate
 from web3 import Web3
 
@@ -34,7 +35,7 @@
 from safe_transaction_service.tokens.tests.factories import TokenFactory
 
 from ...utils.redis import get_redis
-from ..helpers import DelegateSignatureHelper
+from ..helpers import DelegateSignatureHelper, DeleteMultisigTxSignatureHelper
 from ..models import (
     IndexingStatus,
     MultisigConfirmation,
@@ -868,6 +869,131 @@ def test_get_multisig_transaction(self):
         )
         self.assertEqual(response.data["proposer"], proposer)
 
+    def test_delete_multisig_transaction(self):
+        owner_account = Account.create()
+        safe_tx_hash = Web3.keccak(text="random-tx").hex()
+        url = reverse("v1:history:multisig-transaction", args=(safe_tx_hash,))
+        data = {"signature": "0x" + "1" * (130 * 2)}  # 2 signatures of 65 bytes
+        response = self.client.delete(url, format="json", data=data)
+        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
+        # Add our test MultisigTransaction to the database
+        multisig_transaction = MultisigTransactionFactory(safe_tx_hash=safe_tx_hash)
+
+        # Add other MultisigTransactions to the database to make sure they are not deleted
+        MultisigTransactionFactory()
+        MultisigTransactionFactory()
+
+        response = self.client.delete(url, format="json", data=data)
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertDictEqual(
+            response.data,
+            {
+                "non_field_errors": [
+                    ErrorDetail(
+                        string="Executed transactions cannot be deleted", code="invalid"
+                    )
+                ]
+            },
+        )
+
+        multisig_transaction.ethereum_tx = None
+        multisig_transaction.save(update_fields=["ethereum_tx"])
+        response = self.client.delete(url, format="json", data=data)
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertDictEqual(
+            response.data,
+            {
+                "non_field_errors": [
+                    ErrorDetail(
+                        string="Old transactions without proposer cannot be deleted",
+                        code="invalid",
+                    )
+                ]
+            },
+        )
+
+        # Set a random proposer for the transaction
+        multisig_transaction.proposer = Account.create().address
+        multisig_transaction.save(update_fields=["proposer"])
+        response = self.client.delete(url, format="json", data=data)
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertDictEqual(
+            response.data,
+            {
+                "non_field_errors": [
+                    ErrorDetail(
+                        string="1 owner signature was expected, 2 received",
+                        code="invalid",
+                    )
+                ]
+            },
+        )
+
+        # Use a contract signature
+        data = {"signature": "0x" + "0" * 130}  # 1 signature of 65 bytes
+        response = self.client.delete(url, format="json", data=data)
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertDictEqual(
+            response.data,
+            {
+                "non_field_errors": [
+                    ErrorDetail(
+                        string="Only EOA signatures are supported", code="invalid"
+                    )
+                ]
+            },
+        )
+
+        # Use a real not valid signature and set the right proposer
+        multisig_transaction.proposer = owner_account.address
+        multisig_transaction.save(update_fields=["proposer"])
+        data = {
+            "signature": owner_account.signHash(safe_tx_hash)[
+                "signature"
+            ].hex()  # Random signature
+        }
+        response = self.client.delete(url, format="json", data=data)
+        self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+        self.assertDictEqual(
+            response.data,
+            {
+                "non_field_errors": [
+                    ErrorDetail(
+                        string="Provided owner is not the proposer of the transaction",
+                        code="invalid",
+                    )
+                ]
+            },
+        )
+
+        # Use a proper signature
+        message_hash = DeleteMultisigTxSignatureHelper.calculate_hash(
+            multisig_transaction.safe,
+            safe_tx_hash,
+            self.ethereum_client.get_chain_id(),
+            previous_totp=False,
+        )
+        data = {
+            "signature": owner_account.signHash(message_hash)[
+                "signature"
+            ].hex()  # Random signature
+        }
+        self.assertEqual(MultisigTransaction.objects.count(), 3)
+        self.assertTrue(
+            MultisigTransaction.objects.filter(safe_tx_hash=safe_tx_hash).exists()
+        )
+        response = self.client.delete(url, format="json", data=data)
+        self.assertEqual(response.status_code, status.HTTP_204_NO_CONTENT)
+        self.assertEqual(MultisigTransaction.objects.count(), 2)
+        self.assertFalse(
+            MultisigTransaction.objects.filter(safe_tx_hash=safe_tx_hash).exists()
+        )
+
+        # Trying to do the query again should raise a 404
+        response = self.client.delete(url, format="json", data=data)
+        self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND)
+
     def test_get_multisig_transactions(self):
         safe_address = Account.create().address
         proposer = Account.create().address

safe_transaction_service/history/views.py

@@ -586,6 +586,60 @@ def get_queryset(self):
             .select_related("ethereum_tx__block")
         )
 
+    @swagger_auto_schema(
+        request_body=serializers.SafeMultisigTransactionDeleteSerializer(),
+        responses={
+            204: "Deleted",
+            404: "Transaction not found",
+            400: "Error processing data",
+        },
+    )
+    def delete(self, request, safe_tx_hash: HexStr):
+        """
+        Delete a queued but not executed multisig transaction. Only the proposer can delete the transaction.
+        Delegates are not valid, if the transaction was proposed by a delegator the owner who delegated to
+        the delegate must be used.
+        An EOA is required to sign the following EIP712 data:
+
+        ```python
+         {
+            "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": totp,
+            },
+        }
+        ```
+
+        `totp` parameter is calculated with `T0=0` and `Tx=3600`. `totp` is calculated by taking the
+        Unix UTC epoch time (no milliseconds) and dividing by 3600 (natural division, no decimals)
+        """
+        request.data["safe_tx_hash"] = safe_tx_hash
+        serializer = serializers.SafeMultisigTransactionDeleteSerializer(
+            data=request.data
+        )
+        serializer.is_valid(raise_exception=True)
+        MultisigTransaction.objects.filter(safe_tx_hash=safe_tx_hash).delete()
+        return Response(status=status.HTTP_204_NO_CONTENT)
+
 
 class SafeMultisigTransactionDeprecatedDetailView(SafeMultisigTransactionDetailView):
     @swagger_auto_schema(