[SECUR-236] fix: harden pagination bounds and auth brute-force rate limiting

AppScan DAST remediation, ported from the plane-ee fix (#8591) and
re-verified against a live plane-ce instance.

- paginator: reject non-positive per_page (per_page=0 -> ZeroDivisionError ->
  HTTP 500) and bound the client-supplied cursor value/offset. The grouped
  paginators use cursor.value as the per-group page size: a negative value
  slices the queryset with a negative stop (ValueError -> HTTP 500) and a huge
  value fetches far more than max_per_page rows per group (cap bypass / DoS).
  One central guard in BasePaginator.paginate(); regression tests added.
- auth: sign-in/sign-up (app + space) were plain Views with no rate limiting.
  Add the IP-based AuthenticationThrottle check plus a per-account throttle
  keyed on the normalized email (AuthenticationAccountThrottle). The IP key is
  bypassable by spoofing X-Forwarded-For (NUM_PROXIES unset); the per-account
  limiter caps credential guessing against a single account regardless of IP.
- project serializer: mark created_by/updated_by read-only (fields="__all__"
  left them client-writable, allowing project ownership/attribution forgery).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Manish Gupta
2026-07-28 14:49:12 +05:30
parent a8e53b6ac7
commit 47ab37cade
6 changed files with 206 additions and 2 deletions

View File

@@ -34,7 +34,12 @@ class ProjectSerializer(BaseSerializer):
class Meta:
model = Project
fields = "__all__"
read_only_fields = ["workspace", "deleted_at"]
# created_by/updated_by are audit fields set server-side by BaseModel.save()
# from the request user; with fields="__all__" they are otherwise client-writable,
# letting a caller forge project ownership/attribution. save() only backfills
# created_by when it is None, so a supplied value would survive — mark them
# read-only so the client value is ignored.
read_only_fields = ["workspace", "deleted_at", "created_by", "updated_by"]
def validate_name(self, name):
project_id = self.instance.id if self.instance else None

View File

@@ -6,7 +6,7 @@
import os
# Third party imports
from rest_framework.throttling import AnonRateThrottle, UserRateThrottle
from rest_framework.throttling import AnonRateThrottle, SimpleRateThrottle, UserRateThrottle
from rest_framework import status
from rest_framework.response import Response
@@ -49,6 +49,31 @@ def authentication_throttle_allows(request):
return throttle.allow_request(request, None)
class AuthenticationAccountThrottle(SimpleRateThrottle):
"""Per-account (submitted email) authentication throttle.
AuthenticationThrottle keys on DRF get_ident, which honors X-Forwarded-For when
NUM_PROXIES is unset — an attacker can rotate that header to get a fresh bucket per
request and brute-force credentials unthrottled. Bucketing a second limiter by the
normalized email caps guesses against any single account regardless of source IP.
Email is normalized (strip + lower) to match the login lookup so casing tricks cannot
multiply the allowance; requests without an email fall back to the client identity.
"""
scope = "authentication_account"
rate = os.environ.get("AUTHENTICATION_ACCOUNT_RATE_LIMIT", "5/minute")
def get_cache_key(self, request, view=None):
email = (request.POST.get("email") or "").strip().lower()
ident = f"email:{email}" if email else f"ip:{self.get_ident(request)}"
return self.cache_format % {"scope": self.scope, "ident": ident}
def authentication_account_throttle_allows(request):
"""Per-account counterpart to authentication_throttle_allows (see above)."""
return AuthenticationAccountThrottle().allow_request(request, None)
class EmailVerificationThrottle(UserRateThrottle):
"""
Throttle for email verification code generation.

View File

@@ -10,6 +10,10 @@ from django.views import View
# Module imports
from plane.authentication.provider.credentials.email import EmailProvider
from plane.authentication.rate_limit import (
authentication_throttle_allows,
authentication_account_throttle_allows,
)
from plane.authentication.utils.login import user_login
from plane.license.models import Instance
from plane.authentication.utils.host import base_host
@@ -26,6 +30,22 @@ from plane.utils.path_validator import get_safe_redirect_url
class SignInAuthEndpoint(View):
def post(self, request):
next_path = request.POST.get("next_path")
# Rate-limit password sign-in per IP to prevent credential brute-force.
# This is a plain django View, so DRF throttle_classes do not apply and
# the throttle must be invoked manually (as in the magic-link endpoints).
if not authentication_throttle_allows(request) or not authentication_account_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)
# Check instance configuration
instance = Instance.objects.first()
if instance is None or not instance.is_setup_done:
@@ -135,6 +155,21 @@ class SignInAuthEndpoint(View):
class SignUpAuthEndpoint(View):
def post(self, request):
next_path = request.POST.get("next_path")
# Rate-limit password sign-up per IP to prevent automated abuse,
# mirroring the sign-in path and the magic-link endpoints.
if not authentication_throttle_allows(request) or not authentication_account_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)
# Check instance configuration
instance = Instance.objects.first()
if instance is None or not instance.is_setup_done:

View File

@@ -11,6 +11,10 @@ from django.utils.http import url_has_allowed_host_and_scheme
# Module imports
from plane.authentication.provider.credentials.email import EmailProvider
from plane.authentication.rate_limit import (
authentication_throttle_allows,
authentication_account_throttle_allows,
)
from plane.authentication.utils.login import user_login
from plane.license.models import Instance
from plane.authentication.utils.host import base_host
@@ -25,6 +29,22 @@ from plane.utils.path_validator import get_safe_redirect_url, validate_next_path
class SignInAuthSpaceEndpoint(View):
def post(self, request):
next_path = request.POST.get("next_path")
# Rate-limit password sign-in per IP to prevent credential brute-force.
# Plain django View → DRF throttle_classes do not apply, so invoke the
# throttle manually (as in the magic-link endpoints).
if not authentication_throttle_allows(request) or not authentication_account_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)
# Check instance configuration
instance = Instance.objects.first()
if instance is None or not instance.is_setup_done:
@@ -110,6 +130,21 @@ class SignInAuthSpaceEndpoint(View):
class SignUpAuthSpaceEndpoint(View):
def post(self, request):
next_path = request.POST.get("next_path")
# Rate-limit password sign-up per IP to prevent automated abuse,
# mirroring the sign-in path and the magic-link endpoints.
if not authentication_throttle_allows(request) or not authentication_account_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)
# Check instance configuration
instance = Instance.objects.first()
if instance is None or not instance.is_setup_done:

View File

@@ -121,3 +121,89 @@ class TestPaginateGroupByValidation:
paginator_cls=_StubGroupedPaginator,
)
assert response.data["grouped_by"] is None
@pytest.mark.unit
class TestGetPerPageBounds:
"""get_per_page() must reject non-positive per_page before it reaches the
paginator. A per_page of 0 divides by zero in math.ceil(count / limit) and
a negative per_page slices the queryset with garbage bounds — both would
otherwise surface as an unhandled HTTP 500 (flagged by AppScan as
"Integer Overflow" on the stickies per_page parameter)."""
@pytest.mark.parametrize("per_page", ["0", "-1", "-1000"])
def test_non_positive_per_page_raises_parse_error(self, per_page):
request = _make_request(per_page=per_page)
with pytest.raises(ParseError):
BasePaginator().get_per_page(request)
def test_over_max_per_page_still_rejected(self):
request = _make_request(per_page="5000")
with pytest.raises(ParseError):
BasePaginator().get_per_page(request, default_per_page=20, max_per_page=1000)
def test_non_integer_per_page_raises_parse_error(self):
request = _make_request(per_page="abc")
with pytest.raises(ParseError):
BasePaginator().get_per_page(request)
def test_valid_per_page_passes_through(self):
request = _make_request(per_page="30")
assert BasePaginator().get_per_page(request, default_per_page=20, max_per_page=1000) == 30
def test_per_page_of_one_is_allowed(self):
# The exact lower boundary must be accepted.
request = _make_request(per_page="1")
assert BasePaginator().get_per_page(request) == 1
class _ExplodingPaginator:
"""Fails if constructed — proves the cursor guard rejects BEFORE any paginator runs."""
def __init__(self, **kwargs):
raise AssertionError("paginator_cls must not be constructed for an invalid cursor")
@pytest.mark.unit
class TestCursorBounds:
"""paginate() must reject an out-of-bounds client cursor before it drives slicing.
The grouped paginators use cursor.value as the per-group page size
(stop = offset + (cursor.value or limit) + 1). A negative value slices the queryset
with a negative stop -> ValueError('Negative indexing is not supported') -> HTTP 500;
a huge value fetches far more than max_per_page rows per group (cap bypass / DoS).
cursor.offset must be non-negative."""
@pytest.mark.parametrize("cursor", ["-1:0:0", "1000000:0:0", "20:-1:0"])
def test_out_of_bounds_cursor_rejected_before_paginator(self, cursor):
request = _make_request(cursor=cursor)
with pytest.raises(ParseError):
BasePaginator().paginate(
request=request,
queryset=None,
paginator_cls=_ExplodingPaginator,
default_per_page=20,
max_per_page=1000,
)
def test_valid_cursor_passes_the_guard(self):
request = _make_request(cursor="20:0:0")
response = BasePaginator().paginate(
request=request,
queryset=None,
paginator_cls=_StubGroupedPaginator,
default_per_page=20,
max_per_page=1000,
)
assert response.data["results"] == []
def test_cursor_value_at_max_is_allowed(self):
request = _make_request(cursor="1000:0:0")
response = BasePaginator().paginate(
request=request,
queryset=None,
paginator_cls=_StubGroupedPaginator,
default_per_page=20,
max_per_page=1000,
)
assert response.data["results"] == []

View File

@@ -646,6 +646,13 @@ class BasePaginator:
except ValueError:
raise ParseError(detail="Invalid per_page parameter.")
# Reject non-positive values before they reach the paginator, where a
# zero limit divides by zero in math.ceil(count / limit) and a negative
# limit slices the queryset with garbage bounds — both surface as an
# unhandled HTTP 500 instead of a clean client error.
if per_page < 1:
raise ParseError(detail="Invalid per_page value. Must be at least 1.")
max_per_page = max(max_per_page, default_per_page)
if per_page > max_per_page:
raise ParseError(detail=f"Invalid per_page value. Cannot exceed {max_per_page}.")
@@ -680,6 +687,17 @@ class BasePaginator:
except ValueError:
raise ParseError(detail="Invalid cursor parameter.")
# Bound the client-supplied cursor before it drives any slicing. The grouped
# paginators use cursor.value as the per-group page size
# (stop = offset + (cursor.value or limit) + 1). Left unbounded, a negative value
# slices the queryset with a negative stop -> "Negative indexing is not supported"
# (HTTP 500), and a huge value fetches far more than max_per_page rows per group
# (max_per_page cap bypass / resource-exhaustion DoS). cursor.offset is the page
# index and must be non-negative.
effective_max_per_page = max(max_per_page, default_per_page)
if not (0 <= input_cursor.value <= effective_max_per_page) or input_cursor.offset < 0:
raise ParseError(detail="Invalid cursor parameter.")
if not paginator:
if group_by_field_name:
# Validate against the allowlist before the field name reaches