Allow banning EOAs

- Don't allow certain EOAs to interact with the service

config/settings/base.py

@@ -6,6 +6,7 @@
 
 import environ
 from corsheaders.defaults import default_headers as default_cors_headers
+from eth_typing import ChecksumAddress, HexAddress, HexStr
 
 from safe_transaction_service import __version__
 
@@ -717,3 +718,62 @@
     ],
     "SORT_OPERATION_PARAMETERS": False,
 }
+
+# Addresses not allowed to interact with the service
+# List taken from https://www.ic3.gov/PSA/2025/PSA250226
+BANNED_EOAS: set[ChecksumAddress] = {
+    ChecksumAddress(HexAddress(HexStr(address)))
+    for address in (
+        "0x51E9d833Ecae4E8D9D8Be17300AEE6D3398C135D",
+        "0x96244D83DC15d36847C35209bBDc5bdDE9bEc3D8",
+        "0x83c7678492D623fb98834F0fbcb2E7b7f5Af8950",
+        "0x83Ef5E80faD88288F770152875Ab0bb16641a09E",
+        "0xAF620E6d32B1c67f3396EF5d2F7d7642Dc2e6CE9",
+        "0x3A21F4E6Bbe527D347ca7c157F4233c935779847",
+        "0xfa3FcCCB897079fD83bfBA690E7D47Eb402d6c49",
+        "0xFc926659Dd8808f6e3e0a8d61B20B871F3Fa6465",
+        "0xb172F7e99452446f18FF49A71bfEeCf0873003b4",
+        "0x6d46bd3AfF100f23C194e5312f93507978a6DC91",
+        "0xf0a16603289eAF35F64077Ba3681af41194a1c09",
+        "0x23Db729908137cb60852f2936D2b5c6De0e1c887",
+        "0x40e98FeEEbaD7Ddb0F0534Ccaa617427eA10187e",
+        "0x140c9Ab92347734641b1A7c124ffDeE58c20C3E3",
+        "0x684d4b58Dc32af786BF6D572A792fF7A883428B9",
+        "0xBC3e5e8C10897a81b63933348f53f2e052F89a7E",
+        "0x5Af75eAB6BEC227657fA3E749a8BFd55f02e4b1D",
+        "0xBCA02B395747D62626a65016F2e64A20bd254A39",
+        "0x4C198B3B5F3a4b1Aa706daC73D826c2B795ccd67",
+        "0xCd7eC020121Ead6f99855cbB972dF502dB5bC63a",
+        "0xbdE2Cc5375fa9E0383309A2cA31213f2D6cabcbd",
+        "0xD3C611AeD139107DEC2294032da3913BC26507fb",
+        "0xB72334cB9D0b614D30C4c60e2bd12fF5Ed03c305",
+        "0x8c7235e1A6EeF91b980D0FcA083347FBb7EE1806",
+        "0x1bb0970508316DC735329752a4581E0a4bAbc6B4",
+        "0x1eB27f136BFe7947f80d6ceE3Cf0bfDf92b45e57",
+        "0xCd1a4A457cA8b0931c3BF81Df3CFa227ADBdb6E9",
+        "0x09278b36863bE4cCd3d0c22d643E8062D7a11377",
+        "0x660BfcEa3A5FAF823e8f8bF57dd558db034dea1d",
+        "0xE9bc552fdFa54b30296d95F147e3e0280FF7f7e6",
+        "0x30a822CDD2782D2B2A12a08526452e885978FA1D",
+        "0xB4a862A81aBB2f952FcA4C6f5510962e18c7f1A2",
+        "0x0e8C1E2881F35Ef20343264862A242FB749d6b35",
+        "0x9271EDdda0F0f2bB7b1A0c712bdF8dbD0A38d1Ab",
+        "0xe69753Ddfbedbd249E703EB374452E78dae1ae49",
+        "0x2290937A4498C96eFfb87b8371a33D108F8D433f",
+        "0x959c4CA19c4532C97A657D82d97acCBAb70e6fb4",
+        "0x52207Ec7B1b43AA5DB116931a904371ae2C1619e",
+        "0x9eF42873Ae015AA3da0c4354AeF94a18D2B3407b",
+        "0x1542368a03ad1f03d96D51B414f4738961Cf4443",
+        "0x21032176B43d9f7E9410fB37290a78f4fEd6044C",
+        "0xA4B2Fd68593B6F34E51cB9eDB66E71c1B4Ab449e",
+        "0x55CCa2f5eB07907696afe4b9Db5102bcE5feB734",
+        "0xA5A023E052243b7cce34Cbd4ba20180e8Dea6Ad6",
+        "0xdD90071D52F20e85c89802e5Dc1eC0A7B6475f92",
+        "0x1512fcb09463A61862B73ec09B9b354aF1790268",
+        "0xF302572594a68aA8F951faE64ED3aE7DA41c72Be",
+        "0x723a7084028421994d4a7829108D63aB44658315",
+        "0xf03AfB1c6A11A7E370920ad42e6eE735dBedF0b1",
+        "0xEB0bAA3A556586192590CAD296b1e48dF62a8549",
+        "0xD5b58Cf7813c1eDC412367b97876bD400ea5c489",
+    )
+}

safe_transaction_service/history/serializers.py

@@ -5,6 +5,7 @@
 from enum import Enum
 from typing import Any, Dict, List, Optional
 
+from django.conf import settings
 from django.http import Http404
 from django.utils import timezone
 
@@ -125,6 +126,10 @@ def validate_signature(self, signature: bytes):
         ethereum_client = get_auto_ethereum_client()
         for safe_signature in parsed_signatures:
             owner = safe_signature.owner
+            if owner in settings.BANNED_EOAS:
+                raise ValidationError(
+                    f"Signer={owner} is not authorized to interact with the service"
+                )
             if owner not in safe_owners:
                 raise ValidationError(
                     f"Signer={owner} is not an owner. Current owners={safe_owners}"
@@ -264,6 +269,11 @@ def validate(self, attrs):
                     f"Signature={to_0x_hex_str(safe_signature.signature)} for owner={owner} is not valid"
                 )
 
+            if owner in settings.BANNED_EOAS:
+                raise ValidationError(
+                    f"Signer={owner} is not authorized to interact with the service"
+                )
+
             if owner in delegates and len(parsed_signatures) > 1:
                 raise ValidationError(
                     "Just one signature is expected if using delegates"
@@ -713,9 +723,11 @@ def get_block_number(self, obj: MultisigTransaction) -> Optional[int]:
 
     def get_confirmations(self, obj: MultisigTransaction) -> Dict[str, Any]:
         """
-        Filters confirmations queryset
+        Validate and check integrity of confirmations queryset
+
         :param obj: MultisigConfirmation instance
         :return: Serialized queryset
+        :raises InternalValidationError: If any inconsistency is detected
         """
         if obj.ethereum_tx_id:
             return SafeMultisigConfirmationResponseSerializer(

safe_transaction_service/history/tests/test_views.py

@@ -858,6 +858,43 @@ def test_post_multisig_confirmation(self):
         self.assertEqual(response.status_code, status.HTTP_201_CREATED)
         self.assertEqual(MultisigConfirmation.objects.count(), 2)
 
+    def test_post_multisig_confirmation_banned(self):
+        owner_account_1 = Account.create()
+        owner_account_2 = Account.create()
+        safe = self.deploy_test_safe(
+            owners=[owner_account_1.address, owner_account_2.address]
+        )
+        safe_address = safe.address
+        multisig_transaction = MultisigTransactionFactory(
+            safe=safe_address, trusted=True, ethereum_tx=None
+        )
+        safe_tx_hash = multisig_transaction.safe_tx_hash
+        data = {
+            "signature": to_0x_hex_str(
+                owner_account_1.unsafe_sign_hash(safe_tx_hash)["signature"]
+            )
+        }
+        self.assertEqual(MultisigConfirmation.objects.count(), 0)
+        with self.settings(BANNED_EOAS={owner_account_1.address}):
+            response = self.client.post(
+                reverse(
+                    "v1:history:multisig-transaction-confirmations",
+                    args=(safe_tx_hash,),
+                ),
+                format="json",
+                data=data,
+            )
+            self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+            self.assertEqual(
+                response.json(),
+                {
+                    "signature": [
+                        f"Signer={owner_account_1.address} is not authorized to interact with the service"
+                    ]
+                },
+            )
+            self.assertEqual(MultisigConfirmation.objects.count(), 0)
+
     def test_get_multisig_transaction(self):
         safe_tx_hash = to_0x_hex_str(fast_keccak_text("gnosis"))
         response = self.client.get(
@@ -2221,6 +2258,66 @@ def test_post_multisig_transactions_with_delegate(self):
             response.data["non_field_errors"][0],
         )
 
+    def test_post_multisig_transactions_with_banned_signatures(self):
+        safe_owners = [Account.create() for _ in range(4)]
+        safe_owner_addresses = [s.address for s in safe_owners]
+        safe = self.deploy_test_safe(owners=safe_owner_addresses, threshold=3)
+        safe_address = safe.address
+
+        data = {
+            "to": Account.create().address,
+            "value": 100000000000000000,
+            "data": None,
+            "operation": 0,
+            "nonce": 0,
+            "safeTxGas": 0,
+            "baseGas": 0,
+            "gasPrice": 0,
+            "gasToken": "0x0000000000000000000000000000000000000000",
+            "refundReceiver": "0x0000000000000000000000000000000000000000",
+            # "contractTransactionHash": "0x1c2c77b29086701ccdda7836c399112a9b715c6a153f6c8f75c84da4297f60d3",
+            "sender": safe_owners[0].address,
+            "origin": "Testing origin field",
+        }
+
+        safe_tx = safe.build_multisig_tx(
+            data["to"],
+            data["value"],
+            data["data"],
+            data["operation"],
+            data["safeTxGas"],
+            data["baseGas"],
+            data["gasPrice"],
+            data["gasToken"],
+            data["refundReceiver"],
+            safe_nonce=data["nonce"],
+        )
+        safe_tx_hash = safe_tx.safe_tx_hash
+        data["contractTransactionHash"] = to_0x_hex_str(safe_tx_hash)
+        data["signature"] = to_0x_hex_str(
+            b"".join(
+                [
+                    safe_owner.unsafe_sign_hash(safe_tx_hash)["signature"]
+                    for safe_owner in safe_owners
+                ]
+            )
+        )
+        with self.settings(BANNED_EOAS={safe_owners[0].address}):
+            response = self.client.post(
+                reverse("v1:history:multisig-transactions", args=(safe_address,)),
+                format="json",
+                data=data,
+            )
+            self.assertEqual(response.status_code, status.HTTP_422_UNPROCESSABLE_ENTITY)
+            self.assertEqual(
+                response.json(),
+                {
+                    "nonFieldErrors": [
+                        f"Signer={safe_owners[0].address} is not authorized to interact with the service"
+                    ]
+                },
+            )
+
     def test_safe_balances_view(self):
         safe_address = Account.create().address
         response = self.client.get(

safe_transaction_service/safe_messages/serializers.py

@@ -1,6 +1,8 @@
 import json
 from typing import Any, Dict, Optional, Sequence, Tuple, Union
 
+from django.conf import settings
+
 import safe_eth.eth.django.serializers as eth_serializers
 from eth_typing import ChecksumAddress, HexStr
 from hexbytes import HexBytes
@@ -54,6 +56,11 @@ def get_valid_owner_from_signatures(
                 raise ValidationError(f"Signature for owner {owner} already exists")
 
         owners = get_safe_owners(safe_address)
+        if owner in settings.BANNED_EOAS:
+            raise ValidationError(
+                f"Signer={owner} is not authorized to interact with the service"
+            )
+
         if owner not in owners:
             raise ValidationError(f"{owner} is not an owner of the Safe")
 

safe_transaction_service/safe_messages/tests/test_views.py

@@ -236,6 +236,23 @@ def test_safe_messages_create_view(self, get_owners_mock: MagicMock):
                     )
 
                 get_owners_mock.return_value = [account.address]
+
+                with self.settings(BANNED_EOAS={account.address}):
+                    response = self.client.post(
+                        reverse("v1:safe_messages:safe-messages", args=(safe_address,)),
+                        format="json",
+                        data=data,
+                    )
+                    self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
+                    self.assertEqual(
+                        response.json(),
+                        {
+                            "nonFieldErrors": [
+                                f"Signer={account.address} is not authorized to interact with the service"
+                            ]
+                        },
+                    )
+
                 response = self.client.post(
                     reverse("v1:safe_messages:safe-messages", args=(safe_address,)),
                     format="json",
@@ -256,7 +273,8 @@ def test_safe_messages_create_view(self, get_owners_mock: MagicMock):
                     {
                         "non_field_errors": [
                             ErrorDetail(
-                                string=f"Message with hash {to_0x_hex_str(safe_message_hash)} for safe {safe_address} already exists in DB",
+                                string=f"Message with hash {to_0x_hex_str(safe_message_hash)} for safe {safe_address} "
+                                f"already exists in DB",
                                 code="invalid",
                             )
                         ]