b1c78fe4c8
* fix(api): rate-limit magic-code verification and bound per-token attempts
The magic-link sign-in / sign-up endpoints accept a 6-digit numeric code
(900k-value space, 600s TTL) but never increment a failure counter on a
wrong-code verify and extend django.views.View rather than DRF APIView,
so DRF's AuthenticationThrottle never runs against them. The space-side
generate endpoint also lacked throttle_classes. Combined, this allowed
an unauthenticated attacker who knew a victim's email to brute-force
the code within the TTL window and log in as the victim.
- Add MAX_VERIFY_ATTEMPTS=5 in MagicCodeProvider.set_user_data: failed
comparisons now persist verify_attempts in Redis under the remaining
TTL and, on hitting the limit, delete the key and raise
EMAIL_CODE_ATTEMPT_EXHAUSTED. This is the load-bearing fix - it caps
total attempts per issued token regardless of request rate.
- Add authentication_throttle_allows() so plain Django Views can apply
AuthenticationThrottle without converting to APIView (would change
CSRF + request-parsing semantics for the redirect-flow endpoints).
- Apply the throttle to MagicSignIn/UpEndpoint and the space variants;
add throttle_classes to MagicGenerateSpaceEndpoint to match its app
sibling.
Refs GHSA-9pvm-fcf6-9234.
* fix(api): make verify-attempt increment atomic, expose throttle rate via env
Address PR review feedback:
- Replace the JSON read-modify-write of verify_attempts with a Lua
EVAL script that INCRs a dedicated counter key and EXPIREs it only
on the first increment. The previous round-trip was racy: parallel
wrong-code requests could read the same value and both write the
same incremented count, letting an attacker exceed MAX_VERIFY_ATTEMPTS
under concurrency. Counter is now reset on each new token issuance
and cleared on successful verify / exhaustion.
- Make AuthenticationThrottle.rate configurable via the
AUTHENTICATION_RATE_LIMIT env var (default 10/minute, down from 30
to tighten the budget on unauth auth-adjacent endpoints). Document
it in deployments/aio and deployments/cli variables.env.
* test(api): cover magic-code attempt cap, counter reset, and auth throttle
Add the contract tests called out in the PR test plan:
- TestMagicSignInVerifyAttempts:
- test_exhausted_after_max_wrong_attempts: after MAX_VERIFY_ATTEMPTS
wrong codes the next verify redirects with EMAIL_CODE_ATTEMPT_
EXHAUSTED_SIGN_IN and both Redis keys are deleted; a follow-up
verify reports EXPIRED.
- test_counter_increments_on_each_wrong_attempt: the dedicated
verify_attempts counter advances by exactly one per wrong POST,
matching the atomic Lua INCR.
- test_counter_resets_on_token_regeneration: regenerating the
magic-link clears the counter so the user isn't pre-locked-out by
a prior session's wrong attempts.
- TestMagicSignUpVerifyAttempts.test_signup_exhausted_after_max_wrong_attempts:
the sign-up endpoint returns EMAIL_CODE_ATTEMPT_EXHAUSTED_SIGN_UP on
the exhausting attempt.
- TestAuthenticationThrottle: exercises authentication_throttle_allows
on the plain-View redirect-flow endpoints by patching the rate down
and asserting RATE_LIMIT_EXCEEDED is appended to the redirect URL
once the per-IP budget is exceeded, for both magic-sign-in and
magic-sign-up.
Each new class clears Django cache (DRF throttle storage) and the
per-email Redis keys around every test so runs are independent.
* fix(api): clamp remaining_ttl to >=1 for verify-attempt counter EXPIRE
ri.ttl() returns 0 when the token has less than one second remaining
(Redis floors to whole seconds). The previous clamp only caught
None and < 0, so a sub-second TTL would pass through and the Lua
script's EXPIRE counter 0 would immediately delete the key — letting
an attacker bypass MAX_VERIFY_ATTEMPTS during the final second of the
token's life. Switch the comparison to <= 0.
Narrow real-world impact (sub-second window, throttle still bounds
the rate) but the cap should hold regardless of timing.
200 lines
8 KiB
Python
200 lines
8 KiB
Python
# Copyright (c) 2023-present Plane Software, Inc. and contributors
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
# See the LICENSE file for details.
|
|
|
|
# Django imports
|
|
from django.core.validators import validate_email
|
|
from django.http import HttpResponseRedirect
|
|
from django.views import View
|
|
from django.utils.http import url_has_allowed_host_and_scheme
|
|
|
|
# Third party imports
|
|
from rest_framework import status
|
|
from rest_framework.permissions import AllowAny
|
|
from rest_framework.response import Response
|
|
from rest_framework.views import APIView
|
|
|
|
# Module imports
|
|
from plane.authentication.provider.credentials.magic_code import MagicCodeProvider
|
|
from plane.authentication.utils.login import user_login
|
|
from plane.bgtasks.magic_link_code_task import magic_link
|
|
from plane.license.models import Instance
|
|
from plane.authentication.utils.host import base_host
|
|
from plane.db.models import User
|
|
from plane.authentication.adapter.error import (
|
|
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()
|
|
if instance is None or not instance.is_setup_done:
|
|
exc = AuthenticationException(
|
|
error_code=AUTHENTICATION_ERROR_CODES["INSTANCE_NOT_CONFIGURED"],
|
|
error_message="INSTANCE_NOT_CONFIGURED",
|
|
)
|
|
return Response(exc.get_error_dict(), status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
email = request.data.get("email", "").strip().lower()
|
|
try:
|
|
validate_email(email)
|
|
adapter = MagicCodeProvider(request=request, key=email)
|
|
key, token = adapter.initiate()
|
|
# If the smtp is configured send through here
|
|
magic_link.delay(email, key, token)
|
|
return Response({"key": str(key)}, status=status.HTTP_200_OK)
|
|
except AuthenticationException as e:
|
|
return Response(e.get_error_dict(), status=status.HTTP_400_BAD_REQUEST)
|
|
|
|
|
|
class MagicSignInSpaceEndpoint(View):
|
|
def post(self, request):
|
|
# set the referer as session to redirect after login
|
|
code = request.POST.get("code", "").strip()
|
|
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"],
|
|
error_message="MAGIC_SIGN_IN_EMAIL_CODE_REQUIRED",
|
|
)
|
|
params = exc.get_error_dict()
|
|
url = get_safe_redirect_url(
|
|
base_url=base_host(request=request, is_space=True),
|
|
next_path=next_path,
|
|
params=params,
|
|
)
|
|
return HttpResponseRedirect(url)
|
|
|
|
existing_user = User.objects.filter(email=email).first()
|
|
|
|
if not existing_user:
|
|
exc = AuthenticationException(
|
|
error_code=AUTHENTICATION_ERROR_CODES["USER_DOES_NOT_EXIST"],
|
|
error_message="USER_DOES_NOT_EXIST",
|
|
)
|
|
params = exc.get_error_dict()
|
|
url = get_safe_redirect_url(
|
|
base_url=base_host(request=request, is_space=True),
|
|
next_path=next_path,
|
|
params=params,
|
|
)
|
|
return HttpResponseRedirect(url)
|
|
|
|
# Active User
|
|
try:
|
|
provider = MagicCodeProvider(request=request, key=f"magic_{email}", code=code)
|
|
user = provider.authenticate()
|
|
# Login the user and record his device info
|
|
user_login(request=request, user=user, is_space=True)
|
|
# redirect to referer path
|
|
next_path = validate_next_path(next_path=next_path)
|
|
url = f"{base_host(request=request, is_space=True).rstrip('/')}{next_path}"
|
|
if url_has_allowed_host_and_scheme(url, allowed_hosts=get_allowed_hosts()):
|
|
return HttpResponseRedirect(url)
|
|
else:
|
|
return HttpResponseRedirect(base_host(request=request, is_space=True))
|
|
|
|
except AuthenticationException as e:
|
|
params = e.get_error_dict()
|
|
url = get_safe_redirect_url(
|
|
base_url=base_host(request=request, is_space=True),
|
|
next_path=next_path,
|
|
params=params,
|
|
)
|
|
return HttpResponseRedirect(url)
|
|
|
|
|
|
class MagicSignUpSpaceEndpoint(View):
|
|
def post(self, request):
|
|
# set the referer as session to redirect after login
|
|
code = request.POST.get("code", "").strip()
|
|
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"],
|
|
error_message="MAGIC_SIGN_UP_EMAIL_CODE_REQUIRED",
|
|
)
|
|
params = exc.get_error_dict()
|
|
url = get_safe_redirect_url(
|
|
base_url=base_host(request=request, is_space=True),
|
|
next_path=next_path,
|
|
params=params,
|
|
)
|
|
return HttpResponseRedirect(url)
|
|
# Existing User
|
|
existing_user = User.objects.filter(email=email).first()
|
|
# Already existing
|
|
if existing_user:
|
|
exc = AuthenticationException(
|
|
error_code=AUTHENTICATION_ERROR_CODES["USER_ALREADY_EXIST"],
|
|
error_message="USER_ALREADY_EXIST",
|
|
)
|
|
params = exc.get_error_dict()
|
|
url = get_safe_redirect_url(
|
|
base_url=base_host(request=request, is_space=True),
|
|
next_path=next_path,
|
|
params=params,
|
|
)
|
|
return HttpResponseRedirect(url)
|
|
|
|
try:
|
|
provider = MagicCodeProvider(request=request, key=f"magic_{email}", code=code)
|
|
user = provider.authenticate()
|
|
# Login the user and record his device info
|
|
user_login(request=request, user=user, is_space=True)
|
|
# redirect to referer path
|
|
next_path = validate_next_path(next_path=next_path)
|
|
url = f"{base_host(request=request, is_space=True).rstrip('/')}{next_path}"
|
|
if url_has_allowed_host_and_scheme(url, allowed_hosts=get_allowed_hosts()):
|
|
return HttpResponseRedirect(url)
|
|
else:
|
|
return HttpResponseRedirect(base_host(request=request, is_space=True))
|
|
|
|
except AuthenticationException as e:
|
|
params = e.get_error_dict()
|
|
url = get_safe_redirect_url(
|
|
base_url=base_host(request=request, is_space=True),
|
|
next_path=next_path,
|
|
params=params,
|
|
)
|
|
return HttpResponseRedirect(url)
|