Allow deleting multisig transactions not executed (#1802)

- Closes #1757

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/models.py

@@ -55,6 +55,8 @@
 
 logger = getLogger(__name__)
 
+MAX_SIGNATURE_LENGTH = 5_000
+
 
 class ConfirmationType(Enum):
     CONFIRMATION = 0
@@ -1522,7 +1524,7 @@ class MultisigConfirmation(TimeStampedModel):
     )  # Use this while we don't have a `multisig_transaction`
     owner = EthereumAddressV2Field()
 
-    signature = HexField(null=True, default=None, max_length=5000)
+    signature = HexField(null=True, default=None, max_length=MAX_SIGNATURE_LENGTH)
     signature_type = models.PositiveSmallIntegerField(
         choices=[(tag.value, tag.name) for tag in SafeSignatureType], db_index=True
     )

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,8 +30,9 @@
 from safe_transaction_service.utils.serializers import get_safe_owners
 
 from .exceptions import NodeConnectionException
-from .helpers import DelegateSignatureHelper
+from .helpers import DelegateSignatureHelper, DeleteMultisigTxSignatureHelper
 from .models import (
+    MAX_SIGNATURE_LENGTH,
     EthereumTx,
     ModuleTransaction,
     MultisigConfirmation,
@@ -60,7 +64,7 @@ class GnosisBaseModelSerializer(serializers.ModelSerializer):
 #            Request Serializers
 # ================================================ #
 class SafeMultisigConfirmationSerializer(serializers.Serializer):
-    signature = HexadecimalField(min_length=65)  # Signatures must be at least 65 bytes
+    signature = HexadecimalField(min_length=65, max_length=MAX_SIGNATURE_LENGTH)
 
     def validate_signature(self, signature: bytes):
         safe_tx_hash = self.context["safe_tx_hash"]
@@ -370,7 +374,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:
@@ -403,7 +407,7 @@ class DelegateSerializer(DelegateSignatureCheckerMixin, serializers.Serializer):
     safe = EthereumAddressField(allow_null=True, required=False, default=None)
     delegate = EthereumAddressField()
     delegator = EthereumAddressField()
-    signature = HexadecimalField(min_length=65)
+    signature = HexadecimalField(min_length=65, max_length=MAX_SIGNATURE_LENGTH)
     label = serializers.CharField(max_length=50)
 
     def validate(self, attrs):
@@ -465,7 +469,7 @@ def save(self, **kwargs):
 class DelegateDeleteSerializer(DelegateSignatureCheckerMixin, serializers.Serializer):
     delegate = EthereumAddressField()
     delegator = EthereumAddressField()
-    signature = HexadecimalField(min_length=65)
+    signature = HexadecimalField(min_length=65, max_length=MAX_SIGNATURE_LENGTH)
 
     def validate(self, attrs):
         super().validate(attrs)
@@ -490,6 +494,57 @@ def validate(self, attrs):
         )
 
 
+class SafeMultisigTransactionDeleteSerializer(serializers.Serializer):
+    safe_tx_hash = Sha3HashField()
+    signature = HexadecimalField(min_length=65, max_length=MAX_SIGNATURE_LENGTH)
+
+    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]
+            # Currently almost all the transactions are proposed using EOAs. Adding support for EIP1271, for example,
+            # would require to use the EIP712 domain of the Safe and a blockchain check. For starting
+            # with this feature we will try to keep it simple and only support EOA signatures.
+            if safe_signature.signature_type not in (
+                SafeSignatureType.EOA,
+                SafeSignatureType.ETH_SIGN,
+            ):
+                raise ValidationError("Only EOA and ETH_SIGN 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)
@@ -931,7 +986,7 @@ class SafeDelegateDeleteSerializer(serializers.Serializer):
 
     safe = EthereumAddressField()
     delegate = EthereumAddressField()
-    signature = HexadecimalField(min_length=65)
+    signature = HexadecimalField(min_length=65, max_length=MAX_SIGNATURE_LENGTH)
 
     def get_valid_delegators(
         self,

safe_transaction_service/history/tests/test_views.py

@@ -17,6 +17,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
 
@@ -35,7 +36,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,
@@ -869,6 +870,132 @@ 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 and ETH_SIGN 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