Fix 1271 signatures for delegates (#2555)

* Fix 1271 signatures for delegates

* Fix tests

* Add test with nested safes

---------

Co-authored-by: Uxio Fuentefria <[email protected]>
Co-authored-by: Felipe Alvarado <[email protected]>

Author: Uxio0

safe_transaction_service/history/helpers.py

@@ -4,7 +4,8 @@
 
 from eth_typing import ChecksumAddress, Hash32, HexStr
 from eth_utils import keccak
-from safe_eth.eth.eip712 import eip712_encode_hash
+from safe_eth.eth.eip712 import eip712_encode, eip712_encode_hash
+from safe_eth.eth.utils import fast_keccak
 
 from safe_transaction_service.history.models import TransferDict
 from safe_transaction_service.tokens.models import Token
@@ -81,19 +82,19 @@ def calculate_hash(
 
 class DelegateSignatureHelperV2(TemporarySignatureHelper):
     @classmethod
-    def calculate_hash(
+    def calculate_hash_and_preimage(
         cls,
         delegate_address: ChecksumAddress,
         chain_id: Optional[int],
         previous_totp: bool = False,
-    ) -> Hash32:
+    ) -> tuple[Hash32, bytes]:
         """
         Builds a EIP712 object and calculates its hash
 
         :param delegate_address:
         :param chain_id:
-        :param previous_totp:
-        :return: Hash for the EIP712 generated object from the provided parameters
+        :param previous_totp: if true calculate previous totp interval
+        :return: Hash for the EIP712 generated object from the provided parameters with the preimage
         """
         totp = cls.calculate_totp(previous=previous_totp)
 
@@ -125,7 +126,8 @@ def calculate_hash(
             )
             payload["domain"]["chainId"] = chain_id
 
-        return eip712_encode_hash(payload)
+        preimage = b"".join(eip712_encode(payload))
+        return fast_keccak(preimage), preimage
 
 
 def is_valid_unique_transfer_id(unique_transfer_id: str) -> bool:

safe_transaction_service/history/serializers.py

@@ -452,14 +452,20 @@ def validate_delegator_signature(
     ) -> bool:
         ethereum_client = get_auto_ethereum_client()
         chain_id = ethereum_client.get_chain_id()
+        any_valid_signature_found = False
+
         # Accept a message with the current topt and the previous totp (to prevent replay attacks)
         for previous_totp, chain_id in list(
             itertools.product((True, False), (chain_id, None))
         ):
-            message_hash = DelegateSignatureHelperV2.calculate_hash(
-                delegate, chain_id, previous_totp=previous_totp
+            message_hash, preimage = (
+                DelegateSignatureHelperV2.calculate_hash_and_preimage(
+                    delegate, chain_id, previous_totp=previous_totp
+                )
+            )
+            safe_signatures = SafeSignature.parse_signature(
+                signature, message_hash, safe_hash_preimage=preimage
             )
-            safe_signatures = SafeSignature.parse_signature(signature, message_hash)
             if not safe_signatures:
                 raise ValidationError("Signature is not valid")
 
@@ -469,13 +475,15 @@ def validate_delegator_signature(
                 )
             safe_signature = safe_signatures[0]
             owner = safe_signature.owner
-            if not safe_signature.is_valid(ethereum_client, owner):
-                raise ValidationError(
-                    f"Signature of type={safe_signature.signature_type.name} "
-                    f"for signer={signer} is not valid"
-                )
-            if owner == signer:
-                return True
+
+            if safe_signature.is_valid(ethereum_client, owner):
+                any_valid_signature_found = True
+                if owner == signer:
+                    return True
+
+        if not any_valid_signature_found:
+            raise ValidationError(f"No valid signature found for signer={signer}")
+
         return False
 
 

safe_transaction_service/history/tests/test_helpers.py

@@ -24,7 +24,7 @@ def test_calculate_hash(self):
             "0x40ef28feb993bf127265260d48ab1a427d5b852aa8b38f511fcd368bfc9cbfdf"
         )
 
-        result_hash = DelegateSignatureHelperV2.calculate_hash(
+        result_hash, _ = DelegateSignatureHelperV2.calculate_hash_and_preimage(
             delegate_address, chain_id, False
         )
 
@@ -36,7 +36,7 @@ def test_calculate_hash(self):
             "0x37fe1c77d467e92109ef91637e30a4053bab963de2ea74f9d6c4ba1918ff32e6"
         )
 
-        result_hash = DelegateSignatureHelperV2.calculate_hash(
+        result_hash, _ = DelegateSignatureHelperV2.calculate_hash_and_preimage(
             delegate_address, chain_id, True
         )
 

safe_transaction_service/history/tests/test_views_v2.py

@@ -232,7 +232,7 @@ def test_delegates_post(self):
             # Create delegate
             self.assertEqual(SafeContractDelegate.objects.count(), 0)
             chain_id = self.ethereum_client.get_chain_id()
-            hash_to_sign = DelegateSignatureHelperV2.calculate_hash(
+            hash_to_sign, _ = DelegateSignatureHelperV2.calculate_hash_and_preimage(
                 delegate.address, chain_id, False
             )
             data["signature"] = to_0x_hex_str(
@@ -271,7 +271,7 @@ def test_delegates_post(self):
             self.assertIsNone(safe_contract_delegate.expiry_date)
 
         # Create delegate without a Safe
-        hash_to_sign = DelegateSignatureHelperV2.calculate_hash(
+        hash_to_sign, _ = DelegateSignatureHelperV2.calculate_hash_and_preimage(
             delegate.address, chain_id, False
         )
         data = {
@@ -293,7 +293,7 @@ def test_delegates_post(self):
         data["signature"] = to_0x_hex_str(signature)
         response = self.client.post(url, format="json", data=data)
         self.assertIn(
-            f"Signature of type=CONTRACT_SIGNATURE for signer={delegator.address} is not valid",
+            f"No valid signature found for signer={delegator.address}",
             response.data["non_field_errors"][0],
         )
         self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
@@ -302,7 +302,7 @@ def test_delegate_creation_without_chain_id(self):
         chain_id = None
         delegate = Account.create()
         delegator = Account.create()
-        hash_to_sign = DelegateSignatureHelperV2.calculate_hash(
+        hash_to_sign, _ = DelegateSignatureHelperV2.calculate_hash_and_preimage(
             delegate.address, chain_id, False
         )
         data = {
@@ -326,7 +326,7 @@ def test_delegate_creation_without_chain_id_with_safe(self):
         chain_id = None
         delegate = Account.create()
         delegator = Account.create()
-        hash_to_sign = DelegateSignatureHelperV2.calculate_hash(
+        hash_to_sign, _ = DelegateSignatureHelperV2.calculate_hash_and_preimage(
             delegate.address, chain_id, False
         )
         data = {
@@ -354,6 +354,79 @@ def test_delegate_creation_without_chain_id_with_safe(self):
         self.assertEqual(safe_contract_delegate.delegator, delegator.address)
         self.assertEqual(safe_contract_delegate.safe_contract_id, safe.address)
 
+    def _test_add_delegate_using_1271_signature(self, safe_deployment_fn):
+        account_a = Account.create()
+        account_b = Account.create()
+        safe_owner = safe_deployment_fn(
+            owners=[account_a.address, account_b.address], threshold=2
+        )
+        SafeContractFactory(address=safe_owner.address)
+
+        nested_safe = safe_deployment_fn(owners=[safe_owner.address])
+        SafeContractFactory(address=nested_safe.address)
+
+        chain_id = self.ethereum_client.get_chain_id()
+        delegate = Account.create()
+
+        _, preimage_to_sign = DelegateSignatureHelperV2.calculate_hash_and_preimage(
+            delegate.address, chain_id, previous_totp=False
+        )
+
+        safe_owner_message_hash, _ = safe_owner.get_message_hash_and_preimage(
+            preimage_to_sign
+        )
+
+        safe_owner_account_a_signature = account_a.unsafe_sign_hash(
+            safe_owner_message_hash
+        )["signature"]
+        safe_owner_account_b_signature = account_b.unsafe_sign_hash(
+            safe_owner_message_hash
+        )["signature"]
+
+        # Build EIP1271 signature v=0 r=safe v=dynamic_part dynamic_part=size+owner_signature
+        signatures_with_addresses = [
+            (account_a.address.lower(), safe_owner_account_a_signature),
+            (account_b.address.lower(), safe_owner_account_b_signature),
+        ]
+        signatures_with_addresses.sort(key=lambda x: x[0])
+        ordered_signatures = b"".join(sig for _, sig in signatures_with_addresses)
+
+        signature_1271 = (
+            signature_to_bytes(
+                0, int.from_bytes(HexBytes(safe_owner.address), byteorder="big"), 65
+            )
+            + eth_abi.encode(["bytes"], [ordered_signatures])[32:]
+        )
+
+        data = {
+            "label": "Nested safe delegator",
+            "safe": nested_safe.address,
+            "delegate": delegate.address,
+            "delegator": safe_owner.address,
+            "signature": to_0x_hex_str(HexBytes(signature_1271)),
+        }
+        response = self.client.post(
+            reverse("v2:history:delegates"),
+            format="json",
+            data=data,
+        )
+        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
+        self.assertEqual(SafeContractDelegate.objects.count(), 1)
+        safe_contract_delegate = SafeContractDelegate.objects.get()
+        self.assertEqual(safe_contract_delegate.delegate, delegate.address)
+        self.assertEqual(safe_contract_delegate.delegator, safe_owner.address)
+        self.assertEqual(safe_contract_delegate.safe_contract_id, nested_safe.address)
+
+    def test_add_delegate_using_1271_signature_v1_3_0(self):
+        return self._test_add_delegate_using_1271_signature(
+            self.deploy_test_safe_v1_3_0
+        )
+
+    def test_add_delegate_using_1271_signature_v1_4_1(self):
+        return self._test_add_delegate_using_1271_signature(
+            self.deploy_test_safe_v1_4_1
+        )
+
     def test_delegates_get(self):
         url = reverse("v2:history:delegates")
         response = self.client.get(url, format="json")
@@ -432,7 +505,7 @@ def test_delegate_delete(self):
         delegate = Account.create()
         delegator = Account.create()
         chain_id = self.ethereum_client.get_chain_id()
-        hash_to_sign = DelegateSignatureHelperV2.calculate_hash(
+        hash_to_sign, _ = DelegateSignatureHelperV2.calculate_hash_and_preimage(
             delegate.address, chain_id, False
         )
         # Test delete using delegate signature and then delegator signature