Add scoped throttling for public auth endpoints

.env.example

@@ -8,6 +8,12 @@ EMAIL_PASSWORD=
 EMAIL_HOST=
 EMAIL_PORT=
 
+DRF_THROTTLE_AUTH_REGISTER=5/min
+DRF_THROTTLE_AUTH_RESEND_EMAIL=3/min
+DRF_THROTTLE_AUTH_RESET_PASSWORD=3/min
+DRF_THROTTLE_TOKEN_OBTAIN=10/min
+DRF_THROTTLE_PROGRAM_REGISTER_NEW=10/min
+
 SENTRY_DSN=
 
 DATABASE_NAME=

core/throttling.py

@@ -0,0 +1,14 @@
+from rest_framework.throttling import ScopedRateThrottle
+from rest_framework.settings import api_settings
+
+
+class PostOnlyScopedRateThrottle(ScopedRateThrottle):
+    def get_rate(self):
+        if not getattr(self, "scope", None):
+            return None
+        return api_settings.DEFAULT_THROTTLE_RATES.get(self.scope)
+
+    def allow_request(self, request, view):
+        if request.method != "POST":
+            return True
+        return super().allow_request(request, view)

docs/api-safety-fix-plan.md

@@ -0,0 +1,27 @@
+# API Safety Fix Plan: Scoped Throttling
+
+Дата: 2026-07-19.
+
+Implemented in this PR:
+
+- Added `PostOnlyScopedRateThrottle`, a small DRF throttle helper that applies scoped throttling only to `POST` requests.
+- Added scoped throttling to public/user-facing high-risk endpoints:
+  - `POST /auth/users/`
+  - `POST /auth/resend_email/`
+  - `POST /auth/reset_password/`
+  - `POST /api/token/`
+  - `POST /programs/<id>/register_new/`
+- Added env-configurable rates in `REST_FRAMEWORK["DEFAULT_THROTTLE_RATES"]` without enabling global `DEFAULT_THROTTLE_CLASSES`.
+- Scoped rate keys are `auth_register`, `auth_resend_email`, `auth_reset_password`, `token_obtain`, and `program_register_new`.
+- Added `.env.example` entries for the new rates.
+- Added targeted tests that override each selected scope to `1/min` and assert the second request is throttled.
+
+Not changed in this PR:
+
+- No serializers, payload schemas, success response bodies, or existing business validation branches were changed.
+- No production/dev GitHub Actions, deploy files, release process, nginx, or proxy configuration were changed.
+- No global throttling was enabled for the rest of the API.
+
+Proxy/IP note:
+
+DRF throttles anonymous requests by client ident. The current backend has `SECURE_PROXY_SSL_HEADER`, but broader trusted proxy/IP handling must be coordinated with infrastructure before relying on IP throttles as an abuse boundary. Reverse proxy config should ensure client IP headers are overwritten by trusted infrastructure, not accepted from arbitrary clients.

partner_programs/tests/test_program_throttling.py

@@ -0,0 +1,61 @@
+from unittest.mock import patch
+
+from django.conf import settings
+from django.core.cache import cache
+from django.test import TestCase, override_settings
+from rest_framework.test import APIClient
+
+from partner_programs.tests.helpers import create_partner_program
+
+
+def throttle_settings(**rates):
+    rest_framework = dict(settings.REST_FRAMEWORK)
+    rest_framework["DEFAULT_THROTTLE_RATES"] = {
+        **rest_framework.get("DEFAULT_THROTTLE_RATES", {}),
+        **rates,
+    }
+    return rest_framework
+
+
+class PartnerProgramThrottleTests(TestCase):
+    def setUp(self):
+        cache.clear()
+        self.client = APIClient()
+        self.program = create_partner_program()
+
+    @override_settings(
+        REST_FRAMEWORK=throttle_settings(program_register_new="1/min")
+    )
+    @patch("partner_programs.services.registration.send_email.delay")
+    def test_external_program_registration_post_is_scoped_throttled(
+        self,
+        send_email_delay,
+    ):
+        first_response = self.client.post(
+            f"/programs/{self.program.id}/register_new/",
+            {
+                "email": "external-throttle-one@example.com",
+                "password": "pass",
+                "first_name": "External",
+                "last_name": "User",
+                "birthday": "01-01-1990",
+            },
+            format="json",
+            REMOTE_ADDR="203.0.113.20",
+        )
+        second_response = self.client.post(
+            f"/programs/{self.program.id}/register_new/",
+            {
+                "email": "external-throttle-two@example.com",
+                "password": "pass",
+                "first_name": "External",
+                "last_name": "User",
+                "birthday": "01-01-1990",
+            },
+            format="json",
+            REMOTE_ADDR="203.0.113.20",
+        )
+
+        self.assertEqual(first_response.status_code, 201)
+        self.assertEqual(second_response.status_code, 429)
+        send_email_delay.assert_called_once()

partner_programs/views.py

@@ -15,6 +15,7 @@
 
 from core.serializers import EmptySerializer, SetLikedSerializer, SetViewedSerializer
 from core.services import add_view, set_like
+from core.throttling import PostOnlyScopedRateThrottle
 from core.utils import build_xlsx_download_response
 from partner_programs.models import (
     PartnerProgram,
@@ -182,6 +183,8 @@ class PartnerProgramCreateUserAndRegister(generics.GenericAPIView):
     permission_classes = [AllowAny]
     serializer_class = PartnerProgramNewUserSerializer
     queryset = PartnerProgram.objects.all()
+    throttle_classes = [PostOnlyScopedRateThrottle]
+    throttle_scope = "program_register_new"
 
     def post(self, request, *args, **kwargs):
         data = request.data

procollab/settings.py

@@ -163,6 +163,23 @@
         "rest_framework.renderers.BrowsableAPIRenderer",
         "rest_framework.renderers.AdminRenderer",
     ],
+    "DEFAULT_THROTTLE_RATES": {
+        "auth_register": config(
+            "DRF_THROTTLE_AUTH_REGISTER", default="5/min", cast=str
+        ),
+        "auth_resend_email": config(
+            "DRF_THROTTLE_AUTH_RESEND_EMAIL", default="3/min", cast=str
+        ),
+        "auth_reset_password": config(
+            "DRF_THROTTLE_AUTH_RESET_PASSWORD", default="3/min", cast=str
+        ),
+        "token_obtain": config(
+            "DRF_THROTTLE_TOKEN_OBTAIN", default="10/min", cast=str
+        ),
+        "program_register_new": config(
+            "DRF_THROTTLE_PROGRAM_REGISTER_NEW", default="10/min", cast=str
+        ),
+    },
 }
 
 ASGI_APPLICATION = "procollab.asgi.application"

procollab/urls.py

@@ -6,11 +6,11 @@
 from drf_yasg.views import get_schema_view
 from rest_framework import authentication, permissions
 from rest_framework_simplejwt.views import (
-    TokenObtainPairView,
     TokenRefreshView,
     TokenVerifyView,
 )
 from users.authentication import ActivityTrackingJWTAuthentication
+from users.token_views import ThrottledTokenObtainPairView
 
 schema_view = get_schema_view(
     openapi.Info(
@@ -58,7 +58,11 @@
     path("courses/", include("courses.urls", namespace="courses")),
     path("rate-project/", include(("project_rates.urls", "rate_projects"))),
     path("feed/", include("feed.urls", namespace="feed")),
-    path("api/token/", TokenObtainPairView.as_view(), name="token_obtain_pair"),
+    path(
+        "api/token/",
+        ThrottledTokenObtainPairView.as_view(),
+        name="token_obtain_pair",
+    ),
     path("api/token/refresh/", TokenRefreshView.as_view(), name="token_refresh"),
     path("api/token/verify/", TokenVerifyView.as_view(), name="token_verify"),
     path("", include("metrics.urls", namespace="metrics")),

users/password_reset_urls.py

@@ -0,0 +1,26 @@
+from django.urls import path
+from django_rest_passwordreset.views import (
+    ResetPasswordConfirm,
+    ResetPasswordRequestToken,
+    ResetPasswordValidateToken,
+)
+
+from core.throttling import PostOnlyScopedRateThrottle
+
+app_name = "password_reset"
+
+
+class ThrottledResetPasswordRequestToken(ResetPasswordRequestToken):
+    throttle_classes = [PostOnlyScopedRateThrottle]
+    throttle_scope = "auth_reset_password"
+
+
+urlpatterns = [
+    path("", ThrottledResetPasswordRequestToken.as_view(), name="reset-password-request"),
+    path("confirm/", ResetPasswordConfirm.as_view(), name="reset-password-confirm"),
+    path(
+        "validate_token/",
+        ResetPasswordValidateToken.as_view(),
+        name="reset-password-validate",
+    ),
+]

users/tests/test_auth_throttling.py

@@ -0,0 +1,173 @@
+from unittest.mock import patch
+
+from django.conf import settings
+from django.core.cache import cache
+from django.test import TestCase, override_settings
+from rest_framework.permissions import AllowAny
+from rest_framework.response import Response
+from rest_framework.test import APIClient, APIRequestFactory
+from rest_framework.views import APIView
+
+from core.throttling import PostOnlyScopedRateThrottle
+from tests.constants import USER_CREATE_DATA
+from users.tests.helpers import build_user
+
+
+def throttle_settings(**rates):
+    rest_framework = dict(settings.REST_FRAMEWORK)
+    rest_framework["DEFAULT_THROTTLE_RATES"] = {
+        **rest_framework.get("DEFAULT_THROTTLE_RATES", {}),
+        **rates,
+    }
+    return rest_framework
+
+
+class DummyPostOnlyThrottleView(APIView):
+    permission_classes = [AllowAny]
+    throttle_classes = [PostOnlyScopedRateThrottle]
+    throttle_scope = "auth_register"
+
+    def get(self, _request):
+        return Response({"ok": True})
+
+    def post(self, _request):
+        return Response({"ok": True})
+
+
+class AuthThrottleTests(TestCase):
+    def setUp(self):
+        cache.clear()
+        self.client = APIClient()
+        self.factory = APIRequestFactory()
+
+    @override_settings(
+        REST_FRAMEWORK=throttle_settings(auth_register="1/min")
+    )
+    @patch("users.views.verify_email")
+    def test_user_registration_post_is_scoped_throttled(self, verify_email_mock):
+        first_response = self.client.post(
+            "/auth/users/",
+            USER_CREATE_DATA,
+            format="json",
+            REMOTE_ADDR="203.0.113.10",
+        )
+        second_payload = {
+            **USER_CREATE_DATA,
+            "email": "second-user@example.com",
+        }
+        second_response = self.client.post(
+            "/auth/users/",
+            second_payload,
+            format="json",
+            REMOTE_ADDR="203.0.113.10",
+        )
+
+        self.assertEqual(first_response.status_code, 201)
+        self.assertEqual(second_response.status_code, 429)
+        verify_email_mock.assert_called_once()
+
+    @override_settings(
+        REST_FRAMEWORK=throttle_settings(auth_resend_email="1/min")
+    )
+    @patch("users.views.verify_email")
+    def test_resend_verify_email_post_is_scoped_throttled(self, verify_email_mock):
+        user = build_user(email="inactive-throttle@example.com", is_active=False)
+
+        first_response = self.client.post(
+            "/auth/resend_email/",
+            {"email": user.email},
+            format="json",
+            REMOTE_ADDR="203.0.113.11",
+        )
+        second_response = self.client.post(
+            "/auth/resend_email/",
+            {"email": user.email},
+            format="json",
+            REMOTE_ADDR="203.0.113.11",
+        )
+
+        self.assertEqual(first_response.status_code, 200)
+        self.assertEqual(second_response.status_code, 429)
+        verify_email_mock.assert_called_once()
+
+    @override_settings(
+        REST_FRAMEWORK=throttle_settings(token_obtain="1/min")
+    )
+    def test_token_obtain_post_is_scoped_throttled(self):
+        user = build_user(email="token-throttle@example.com")
+        payload = {"email": user.email, "password": "very_strong_password"}
+
+        first_response = self.client.post(
+            "/api/token/",
+            payload,
+            format="json",
+            REMOTE_ADDR="203.0.113.12",
+        )
+        second_response = self.client.post(
+            "/api/token/",
+            payload,
+            format="json",
+            REMOTE_ADDR="203.0.113.12",
+        )
+
+        self.assertEqual(first_response.status_code, 200)
+        self.assertEqual(second_response.status_code, 429)
+
+    @override_settings(
+        REST_FRAMEWORK=throttle_settings(auth_reset_password="1/min")
+    )
+    @patch("users.signals.EmailMultiAlternatives")
+    def test_password_reset_request_post_is_scoped_throttled(self, _email_message):
+        user = build_user(email="reset-throttle@example.com")
+
+        first_response = self.client.post(
+            "/auth/reset_password/",
+            {"email": user.email},
+            format="json",
+            REMOTE_ADDR="203.0.113.13",
+        )
+        second_response = self.client.post(
+            "/auth/reset_password/",
+            {"email": user.email},
+            format="json",
+            REMOTE_ADDR="203.0.113.13",
+        )
+
+        self.assertNotEqual(first_response.status_code, 429)
+        self.assertEqual(second_response.status_code, 429)
+
+    @override_settings(
+        REST_FRAMEWORK=throttle_settings(auth_register="1/min")
+    )
+    def test_post_only_throttle_allows_get_and_options(self):
+        view = DummyPostOnlyThrottleView.as_view()
+
+        for method in ("get", "options"):
+            for _ in range(2):
+                response = view(
+                    getattr(self.factory, method)(
+                        "/unused/",
+                        REMOTE_ADDR="203.0.113.14",
+                    )
+                )
+                self.assertNotEqual(response.status_code, 429)
+
+        first_response = view(
+            self.factory.post(
+                "/unused/",
+                {},
+                format="json",
+                REMOTE_ADDR="203.0.113.14",
+            )
+        )
+        second_response = view(
+            self.factory.post(
+                "/unused/",
+                {},
+                format="json",
+                REMOTE_ADDR="203.0.113.14",
+            )
+        )
+
+        self.assertEqual(first_response.status_code, 200)
+        self.assertEqual(second_response.status_code, 429)

users/token_views.py

@@ -0,0 +1,8 @@
+from rest_framework_simplejwt.views import TokenObtainPairView
+
+from core.throttling import PostOnlyScopedRateThrottle
+
+
+class ThrottledTokenObtainPairView(TokenObtainPairView):
+    throttle_classes = [PostOnlyScopedRateThrottle]
+    throttle_scope = "token_obtain"