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,27 @@
 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
+
+    # Atomic INCR + first-time EXPIRE for the verify-attempt counter.
+    # Using a dedicated counter key with this script makes the increment
+    # safe under concurrent wrong-code requests; a plain JSON read/modify/
+    # write would race and let parallel attackers exceed the cap.
+    _INCREMENT_VERIFY_ATTEMPTS_SCRIPT = (
+        'local count = redis.call("INCR", KEYS[1]) '
+        'if count == 1 then '
+        '    redis.call("EXPIRE", KEYS[1], tonumber(ARGV[1])) '
+        'end '
+        'return count'
+    )
+
+    @staticmethod
+    def _verify_attempts_key(token_key):
+        return f"{token_key}:verify_attempts"
+
     def __init__(self, request, key, code=None, callback=None):
         (EMAIL_HOST, ENABLE_MAGIC_LINK_LOGIN) = get_configuration_value(
             [
@@ -92,6 +113,9 @@ def initiate(self):
             expiry = 600
 
             ri.set(key, json.dumps(value), ex=expiry)
+        # Reset the verify-attempt counter so each newly issued token starts
+        # with a fresh budget of MAX_VERIFY_ATTEMPTS.
+        ri.delete(self._verify_attempts_key(key))
         return key, token
 
     def set_user_data(self):
@@ -114,12 +138,52 @@ def set_user_data(self):
                         },
                     }
                 )
-                # Delete the token from redis if the code match is successful
+                # Delete the token and its counter from redis on success.
                 ri.delete(self.key)
+                ri.delete(self._verify_attempts_key(self.key))
                 return
             else:
                 email = str(self.key).replace("magic_", "", 1)
-                if User.objects.filter(email=email).exists():
+                user_exists = User.objects.filter(email=email).exists()
+
+                # Atomically increment the verify-attempt counter in Redis.
+                # The Lua script sets the TTL only on the first increment so
+                # the lockout window matches the remaining token TTL and does
+                # not get extended by every wrong-code attempt.
+                # ri.ttl() returns -2 (missing), -1 (no expiry), 0 (sub-second
+                # remaining; Redis floors to whole seconds), or a positive int.
+                # Clamp to >=1 because EXPIRE key 0 immediately deletes the key
+                # and would let an attacker bypass the cap in the final second.
+                remaining_ttl = ri.ttl(self.key)
+                if remaining_ttl is None or remaining_ttl <= 0:
+                    remaining_ttl = 1
+                verify_attempts = int(
+                    ri.eval(
+                        self._INCREMENT_VERIFY_ATTEMPTS_SCRIPT,
+                        1,
+                        self._verify_attempts_key(self.key),
+                        remaining_ttl,
+                    )
+                )
+
+                if verify_attempts >= self.MAX_VERIFY_ATTEMPTS:
+                    # Invalidate the token (and counter) so further attempts
+                    # must regenerate; regeneration is itself attempt-counted.
+                    ri.delete(self.key)
+                    ri.delete(self._verify_attempts_key(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

@@ -2,6 +2,9 @@
 # SPDX-License-Identifier: AGPL-3.0-only
 # See the LICENSE file for details.
 
+# Python imports
+import os
+
 # Third party imports
 from rest_framework.throttling import AnonRateThrottle, UserRateThrottle
 from rest_framework import status
@@ -15,7 +18,9 @@
 
 
 class AuthenticationThrottle(AnonRateThrottle):
-    rate = "30/minute"
+    # Rate is configurable per-deployment via the AUTHENTICATION_RATE_LIMIT
+    # env var (DRF format: "<num>/<period>" where period is second/minute/hour/day).
+    rate = os.environ.get("AUTHENTICATION_RATE_LIMIT", "10/minute")
     scope = "authentication"
 
     def throttle_failure_view(self, request, *args, **kwargs):
@@ -28,6 +33,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"],