Add test with nested safes

Author: falvaradorodriguez

safe_transaction_service/history/helpers.py

@@ -93,7 +93,7 @@ def calculate_hash_and_preimage(
 
         :param delegate_address:
         :param chain_id:
-        :param previous_totp:
+        :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)

safe_transaction_service/history/serializers.py

@@ -452,6 +452,8 @@ 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))
@@ -473,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_views_v2.py

@@ -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)
@@ -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")