fix(api): rate-limit magic-code verify, bound per-token attempts (GHSA-9pvm-fcf6-9234)

apps/api/plane/authentication/provider/credentials/magic_code.py

@@ -22,6 +22,11 @@
 class MagicCodeProvider(CredentialAdapter):
     provider = "magic-code"
 
+    # Max wrong-code verification attempts per issued token before the token
+    # is invalidated. Prevents brute-forcing the 6-digit code space within
+    # the token TTL window.
+    MAX_VERIFY_ATTEMPTS = 5
+
     def __init__(self, request, key, code=None, callback=None):
         (EMAIL_HOST, ENABLE_MAGIC_LINK_LOGIN) = get_configuration_value(
             [
@@ -119,7 +124,34 @@ def set_user_data(self):
                 return
             else:
                 email = str(self.key).replace("magic_", "", 1)
-                if User.objects.filter(email=email).exists():
+                user_exists = User.objects.filter(email=email).exists()
+
+                # Increment the verify-attempt counter and persist with the
+                # remaining TTL so the lockout window matches the token window.
+                verify_attempts = data.get("verify_attempts", 0) + 1
+                data["verify_attempts"] = verify_attempts
+                remaining_ttl = ri.ttl(self.key)
+                if remaining_ttl is None or remaining_ttl < 0:
+                    remaining_ttl = 1
+                ri.set(self.key, json.dumps(data), ex=remaining_ttl)
+
+                if verify_attempts >= self.MAX_VERIFY_ATTEMPTS:
+                    # Invalidate the token so further attempts must regenerate
+                    # (which is itself attempt-counted on the generation path).
+                    ri.delete(self.key)
+                    if user_exists:
+                        raise AuthenticationException(
+                            error_code=AUTHENTICATION_ERROR_CODES["EMAIL_CODE_ATTEMPT_EXHAUSTED_SIGN_IN"],
+                            error_message="EMAIL_CODE_ATTEMPT_EXHAUSTED_SIGN_IN",
+                            payload={"email": str(email)},
+                        )
+                    raise AuthenticationException(
+                        error_code=AUTHENTICATION_ERROR_CODES["EMAIL_CODE_ATTEMPT_EXHAUSTED_SIGN_UP"],
+                        error_message="EMAIL_CODE_ATTEMPT_EXHAUSTED_SIGN_UP",
+                        payload={"email": str(email)},
+                    )
+
+                if user_exists:
                     raise AuthenticationException(
                         error_code=AUTHENTICATION_ERROR_CODES["INVALID_MAGIC_CODE_SIGN_IN"],
                         error_message="INVALID_MAGIC_CODE_SIGN_IN",

apps/api/plane/authentication/rate_limit.py

@@ -28,6 +28,22 @@ def throttle_failure_view(self, request, *args, **kwargs):
             return Response(e.get_error_dict(), status=status.HTTP_429_TOO_MANY_REQUESTS)
 
 
+def authentication_throttle_allows(request):
+    """
+    Apply AuthenticationThrottle to a plain django.views.View request.
+
+    DRF's throttle_classes only run inside APIView.initial(); the magic
+    sign-in / sign-up endpoints extend django.views.View to return
+    HttpResponseRedirect from a form POST flow, so they need a manual
+    throttle check. Returns True if the request is allowed through,
+    False if it should be rejected with a RATE_LIMIT_EXCEEDED error.
+    """
+    throttle = AuthenticationThrottle()
+    # SimpleRateThrottle.allow_request only reads request.META and
+    # request.user, both available on a plain Django HttpRequest.
+    return throttle.allow_request(request, None)
+
+
 class EmailVerificationThrottle(UserRateThrottle):
     """
     Throttle for email verification code generation.

apps/api/plane/authentication/views/app/magic.py

@@ -26,7 +26,10 @@
     AuthenticationException,
     AUTHENTICATION_ERROR_CODES,
 )
-from plane.authentication.rate_limit import AuthenticationThrottle
+from plane.authentication.rate_limit import (
+    AuthenticationThrottle,
+    authentication_throttle_allows,
+)
 from plane.utils.path_validator import get_safe_redirect_url
 
 
@@ -65,6 +68,18 @@ def post(self, request):
         email = request.POST.get("email", "").strip().lower()
         next_path = request.POST.get("next_path")
 
+        if not authentication_throttle_allows(request):
+            exc = AuthenticationException(
+                error_code=AUTHENTICATION_ERROR_CODES["RATE_LIMIT_EXCEEDED"],
+                error_message="RATE_LIMIT_EXCEEDED",
+            )
+            url = get_safe_redirect_url(
+                base_url=base_host(request=request, is_app=True),
+                next_path=next_path,
+                params=exc.get_error_dict(),
+            )
+            return HttpResponseRedirect(url)
+
         if code == "" or email == "":
             exc = AuthenticationException(
                 error_code=AUTHENTICATION_ERROR_CODES["MAGIC_SIGN_IN_EMAIL_CODE_REQUIRED"],
@@ -136,6 +151,18 @@ def post(self, request):
         email = request.POST.get("email", "").strip().lower()
         next_path = request.POST.get("next_path")
 
+        if not authentication_throttle_allows(request):
+            exc = AuthenticationException(
+                error_code=AUTHENTICATION_ERROR_CODES["RATE_LIMIT_EXCEEDED"],
+                error_message="RATE_LIMIT_EXCEEDED",
+            )
+            url = get_safe_redirect_url(
+                base_url=base_host(request=request, is_app=True),
+                next_path=next_path,
+                params=exc.get_error_dict(),
+            )
+            return HttpResponseRedirect(url)
+
         if code == "" or email == "":
             exc = AuthenticationException(
                 error_code=AUTHENTICATION_ERROR_CODES["MAGIC_SIGN_UP_EMAIL_CODE_REQUIRED"],

apps/api/plane/authentication/views/space/magic.py

@@ -25,12 +25,18 @@
     AuthenticationException,
     AUTHENTICATION_ERROR_CODES,
 )
+from plane.authentication.rate_limit import (
+    AuthenticationThrottle,
+    authentication_throttle_allows,
+)
 from plane.utils.path_validator import get_safe_redirect_url, validate_next_path, get_allowed_hosts
 
 
 class MagicGenerateSpaceEndpoint(APIView):
     permission_classes = [AllowAny]
 
+    throttle_classes = [AuthenticationThrottle]
+
     def post(self, request):
         # Check if instance is configured
         instance = Instance.objects.first()
@@ -60,6 +66,18 @@ def post(self, request):
         email = request.POST.get("email", "").strip().lower()
         next_path = request.POST.get("next_path")
 
+        if not authentication_throttle_allows(request):
+            exc = AuthenticationException(
+                error_code=AUTHENTICATION_ERROR_CODES["RATE_LIMIT_EXCEEDED"],
+                error_message="RATE_LIMIT_EXCEEDED",
+            )
+            url = get_safe_redirect_url(
+                base_url=base_host(request=request, is_space=True),
+                next_path=next_path,
+                params=exc.get_error_dict(),
+            )
+            return HttpResponseRedirect(url)
+
         if code == "" or email == "":
             exc = AuthenticationException(
                 error_code=AUTHENTICATION_ERROR_CODES["MAGIC_SIGN_IN_EMAIL_CODE_REQUIRED"],
@@ -119,6 +137,18 @@ def post(self, request):
         email = request.POST.get("email", "").strip().lower()
         next_path = request.POST.get("next_path")
 
+        if not authentication_throttle_allows(request):
+            exc = AuthenticationException(
+                error_code=AUTHENTICATION_ERROR_CODES["RATE_LIMIT_EXCEEDED"],
+                error_message="RATE_LIMIT_EXCEEDED",
+            )
+            url = get_safe_redirect_url(
+                base_url=base_host(request=request, is_space=True),
+                next_path=next_path,
+                params=exc.get_error_dict(),
+            )
+            return HttpResponseRedirect(url)
+
         if code == "" or email == "":
             exc = AuthenticationException(
                 error_code=AUTHENTICATION_ERROR_CODES["MAGIC_SIGN_UP_EMAIL_CODE_REQUIRED"],