feat: Add rate limiting to identity search endpoint

api/app/settings/common.py

@@ -304,6 +304,7 @@
 SIGNUP_THROTTLE_RATE = env("SIGNUP_THROTTLE_RATE", "10000/min")
 USER_THROTTLE_RATE = env("USER_THROTTLE_RATE", "500/min")
 MASTER_API_KEY_THROTTLE_RATE = env("MASTER_API_KEY_THROTTLE_RATE", USER_THROTTLE_RATE)
+IDENTITY_SEARCH_THROTTLE_RATE = env("IDENTITY_SEARCH_THROTTLE_RATE", "30/min")
 DEFAULT_THROTTLE_CLASSES = env.list("DEFAULT_THROTTLE_CLASSES", subcast=str, default=[])
 REST_FRAMEWORK = {
     "DEFAULT_PERMISSION_CLASSES": ["rest_framework.permissions.IsAuthenticated"],
@@ -324,6 +325,7 @@
         "invite": "10/min",
         "user": USER_THROTTLE_RATE,
         "influx_query": "5/min",
+        "identity_search": IDENTITY_SEARCH_THROTTLE_RATE,
     },
     "DEFAULT_FILTER_BACKENDS": ["django_filters.rest_framework.DjangoFilterBackend"],
     "DEFAULT_RENDERER_CLASSES": [

api/app/settings/test.py

@@ -13,6 +13,7 @@
     "user": "100000/day",
     "master_api_key": "100000/day",
     "influx_query": "50/min",
+    "identity_search": "100/min",
 }
 
 AWS_SSE_LOGS_BUCKET_NAME = "test_bucket"

api/environments/identities/views.py

@@ -16,6 +16,7 @@
 from rest_framework import status, viewsets
 from rest_framework.permissions import IsAuthenticated
 from rest_framework.response import Response
+from rest_framework.throttling import ScopedRateThrottle
 
 from app.pagination import CustomPagination
 from core.constants import FLAGSMITH_UPDATED_AT_HEADER, SDK_ENVIRONMENT_KEY_HEADER
@@ -41,6 +42,16 @@
 class IdentityViewSet(viewsets.ModelViewSet):  # type: ignore[type-arg]
     serializer_class = IdentitySerializer
     pagination_class = CustomPagination
+    throttle_scope = "identity_search"
+
+    def get_throttles(self):  # type: ignore[no-untyped-def]
+        """
+        Apply identity_search throttle only to list (search) requests.
+        For other actions, return the global default throttle classes.
+        """
+        if getattr(self, "action", None) == "list":
+            return [ScopedRateThrottle()]
+        return super().get_throttles()
 
     def get_queryset(self):  # type: ignore[no-untyped-def]
         if getattr(self, "swagger_fake_view", False):

api/tests/unit/environments/identities/test_unit_identities_views.py

@@ -305,6 +305,31 @@ def test_search_identities_still_allows_paging(
     assert response2.data["results"]
 
 
+def test_identity_search_is_throttled(
+    admin_client: APIClient,
+    environment: Environment,
+    reset_cache: None,
+    mocker: MockerFixture,
+) -> None:
+    # Given - mock the throttle rate to be restrictive for testing
+    mocker.patch(
+        "rest_framework.throttling.ScopedRateThrottle.get_rate", return_value="1/minute"
+    )
+    base_url = reverse(
+        "api-v1:environments:environment-identities-list",
+        args=[environment.api_key],
+    )
+    url = f"{base_url}?q=test"
+
+    # When - make 2 requests in quick succession
+    response1 = admin_client.get(url)
+    response2 = admin_client.get(url)
+
+    # Then - first should succeed, second should be throttled
+    assert response1.status_code == status.HTTP_200_OK
+    assert response2.status_code == status.HTTP_429_TOO_MANY_REQUESTS
+
+
 def test_can_delete_identity(
     environment: Environment,
     admin_client: APIClient,