e30605d419
* fix: add rate limiting to email/password sign-in and sign-up endpoints All four password authentication views (app sign-in, app sign-up, space sign-in, space sign-up) extended django.views.View, so DRF's global AnonRateThrottle never ran and the endpoints accepted unlimited credential guesses with no friction (brute-force / credential stuffing, GHSA-349j). Add authentication_throttle_allows(request) at the top of each post() method — before any DB access — using the same AuthenticationThrottle already guarding the magic-code views. On rejection the view redirects with RATE_LIMIT_EXCEEDED, consistent with all other throttled auth endpoints. Default limit remains 10/minute, overridable via AUTHENTICATION_RATE_LIMIT. Co-authored-by: Plane AI <noreply@plane.so> * refactor: consolidate auth throttle into a decorator + add tests - Extract the repeated throttle-and-redirect block from the six redirect-flow auth views (email + magic, app + space) into a single throttle_auth_redirect decorator in rate_limit.py. Behaviour is unchanged: the throttle still runs before any DB access; brute-force traffic is rejected without a DB hit. - Add regression tests for the password sign-in/sign-up throttle on both app and space endpoints, mirroring the existing magic-code throttle tests. - Reset the shared AuthenticationThrottle bucket before every test in test_authentication.py. All auth endpoints share one per-IP throttle scope, so the newly-throttled password requests exhausted the budget mid-file and caused unrelated tests to trip RATE_LIMIT_EXCEEDED. --------- Co-authored-by: Plane AI <noreply@plane.so> Co-authored-by: sriramveeraghanta <veeraghanta.sriram@gmail.com>
244 lines
9.2 KiB
Python
244 lines
9.2 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.exceptions import ValidationError
|
|
from django.core.validators import validate_email
|
|
from django.http import HttpResponseRedirect
|
|
from django.views import View
|
|
|
|
# Module imports
|
|
from plane.authentication.provider.credentials.email import EmailProvider
|
|
from plane.authentication.utils.login import user_login
|
|
from plane.authentication.rate_limit import throttle_auth_redirect
|
|
from plane.license.models import Instance
|
|
from plane.authentication.utils.host import base_host
|
|
from plane.authentication.utils.redirection_path import get_redirection_path
|
|
from plane.authentication.utils.user_auth_workflow import post_user_auth_workflow
|
|
from plane.db.models import User
|
|
from plane.authentication.adapter.error import (
|
|
AuthenticationException,
|
|
AUTHENTICATION_ERROR_CODES,
|
|
)
|
|
from plane.utils.path_validator import get_safe_redirect_url
|
|
|
|
|
|
class SignInAuthEndpoint(View):
|
|
# Rate-limit password auth attempts before any DB access, using the same
|
|
# AuthenticationThrottle the magic-code views use (default 10/minute,
|
|
# configurable via the AUTHENTICATION_RATE_LIMIT env var).
|
|
@throttle_auth_redirect(is_app=True)
|
|
def post(self, request):
|
|
next_path = request.POST.get("next_path")
|
|
# Check instance configuration
|
|
instance = Instance.objects.first()
|
|
if instance is None or not instance.is_setup_done:
|
|
# Redirection params
|
|
exc = AuthenticationException(
|
|
error_code=AUTHENTICATION_ERROR_CODES["INSTANCE_NOT_CONFIGURED"],
|
|
error_message="INSTANCE_NOT_CONFIGURED",
|
|
)
|
|
params = exc.get_error_dict()
|
|
# Base URL join
|
|
url = get_safe_redirect_url(
|
|
base_url=base_host(request=request, is_app=True),
|
|
next_path=next_path,
|
|
params=params,
|
|
)
|
|
return HttpResponseRedirect(url)
|
|
|
|
# set the referer as session to redirect after login
|
|
email = request.POST.get("email", False)
|
|
password = request.POST.get("password", False)
|
|
|
|
## Raise exception if any of the above are missing
|
|
if not email or not password:
|
|
# Redirection params
|
|
exc = AuthenticationException(
|
|
error_code=AUTHENTICATION_ERROR_CODES["REQUIRED_EMAIL_PASSWORD_SIGN_IN"],
|
|
error_message="REQUIRED_EMAIL_PASSWORD_SIGN_IN",
|
|
payload={"email": str(email)},
|
|
)
|
|
params = exc.get_error_dict()
|
|
# Next path
|
|
url = get_safe_redirect_url(
|
|
base_url=base_host(request=request, is_app=True),
|
|
next_path=next_path,
|
|
params=params,
|
|
)
|
|
return HttpResponseRedirect(url)
|
|
|
|
# Validate email
|
|
email = email.strip().lower()
|
|
try:
|
|
validate_email(email)
|
|
except ValidationError:
|
|
exc = AuthenticationException(
|
|
error_code=AUTHENTICATION_ERROR_CODES["INVALID_EMAIL_SIGN_IN"],
|
|
error_message="INVALID_EMAIL_SIGN_IN",
|
|
payload={"email": str(email)},
|
|
)
|
|
params = exc.get_error_dict()
|
|
url = get_safe_redirect_url(
|
|
base_url=base_host(request=request, is_app=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",
|
|
payload={"email": str(email)},
|
|
)
|
|
params = exc.get_error_dict()
|
|
url = get_safe_redirect_url(
|
|
base_url=base_host(request=request, is_app=True),
|
|
next_path=next_path,
|
|
params=params,
|
|
)
|
|
return HttpResponseRedirect(url)
|
|
|
|
try:
|
|
provider = EmailProvider(
|
|
request=request,
|
|
key=email,
|
|
code=password,
|
|
is_signup=False,
|
|
callback=post_user_auth_workflow,
|
|
)
|
|
user = provider.authenticate()
|
|
# Login the user and record his device info
|
|
user_login(request=request, user=user, is_app=True)
|
|
# Get the redirection path
|
|
if next_path:
|
|
path = next_path
|
|
else:
|
|
path = get_redirection_path(user=user)
|
|
|
|
# Get the safe redirect URL
|
|
url = get_safe_redirect_url(
|
|
base_url=base_host(request=request, is_app=True),
|
|
next_path=path,
|
|
params={},
|
|
)
|
|
return HttpResponseRedirect(url)
|
|
except AuthenticationException as e:
|
|
params = e.get_error_dict()
|
|
url = get_safe_redirect_url(
|
|
base_url=base_host(request=request, is_app=True),
|
|
next_path=next_path,
|
|
params=params,
|
|
)
|
|
return HttpResponseRedirect(url)
|
|
|
|
|
|
class SignUpAuthEndpoint(View):
|
|
@throttle_auth_redirect(is_app=True)
|
|
def post(self, request):
|
|
next_path = request.POST.get("next_path")
|
|
# Check instance configuration
|
|
instance = Instance.objects.first()
|
|
if instance is None or not instance.is_setup_done:
|
|
# Redirection params
|
|
exc = AuthenticationException(
|
|
error_code=AUTHENTICATION_ERROR_CODES["INSTANCE_NOT_CONFIGURED"],
|
|
error_message="INSTANCE_NOT_CONFIGURED",
|
|
)
|
|
params = exc.get_error_dict()
|
|
url = get_safe_redirect_url(
|
|
base_url=base_host(request=request, is_app=True),
|
|
next_path=next_path,
|
|
params=params,
|
|
)
|
|
return HttpResponseRedirect(url)
|
|
|
|
email = request.POST.get("email", False)
|
|
password = request.POST.get("password", False)
|
|
## Raise exception if any of the above are missing
|
|
if not email or not password:
|
|
# Redirection params
|
|
exc = AuthenticationException(
|
|
error_code=AUTHENTICATION_ERROR_CODES["REQUIRED_EMAIL_PASSWORD_SIGN_UP"],
|
|
error_message="REQUIRED_EMAIL_PASSWORD_SIGN_UP",
|
|
payload={"email": str(email)},
|
|
)
|
|
params = exc.get_error_dict()
|
|
url = get_safe_redirect_url(
|
|
base_url=base_host(request=request, is_app=True),
|
|
next_path=next_path,
|
|
params=params,
|
|
)
|
|
return HttpResponseRedirect(url)
|
|
# Validate the email
|
|
email = email.strip().lower()
|
|
try:
|
|
validate_email(email)
|
|
except ValidationError:
|
|
# Redirection params
|
|
exc = AuthenticationException(
|
|
error_code=AUTHENTICATION_ERROR_CODES["INVALID_EMAIL_SIGN_UP"],
|
|
error_message="INVALID_EMAIL_SIGN_UP",
|
|
payload={"email": str(email)},
|
|
)
|
|
params = exc.get_error_dict()
|
|
url = get_safe_redirect_url(
|
|
base_url=base_host(request=request, is_app=True),
|
|
next_path=next_path,
|
|
params=params,
|
|
)
|
|
return HttpResponseRedirect(url)
|
|
|
|
# Existing user
|
|
existing_user = User.objects.filter(email=email).first()
|
|
|
|
if existing_user:
|
|
# Existing User
|
|
exc = AuthenticationException(
|
|
error_code=AUTHENTICATION_ERROR_CODES["USER_ALREADY_EXIST"],
|
|
error_message="USER_ALREADY_EXIST",
|
|
payload={"email": str(email)},
|
|
)
|
|
params = exc.get_error_dict()
|
|
url = get_safe_redirect_url(
|
|
base_url=base_host(request=request, is_app=True),
|
|
next_path=next_path,
|
|
params=params,
|
|
)
|
|
return HttpResponseRedirect(url)
|
|
|
|
try:
|
|
provider = EmailProvider(
|
|
request=request,
|
|
key=email,
|
|
code=password,
|
|
is_signup=True,
|
|
callback=post_user_auth_workflow,
|
|
)
|
|
user = provider.authenticate()
|
|
# Login the user and record his device info
|
|
user_login(request=request, user=user, is_app=True)
|
|
# Get the redirection path
|
|
if next_path:
|
|
path = next_path
|
|
else:
|
|
path = get_redirection_path(user=user)
|
|
|
|
url = get_safe_redirect_url(
|
|
base_url=base_host(request=request, is_app=True),
|
|
next_path=path,
|
|
params={},
|
|
)
|
|
return HttpResponseRedirect(url)
|
|
except AuthenticationException as e:
|
|
params = e.get_error_dict()
|
|
url = get_safe_redirect_url(
|
|
base_url=base_host(request=request, is_app=True),
|
|
next_path=next_path,
|
|
params=params,
|
|
)
|
|
return HttpResponseRedirect(url)
|