## 28) Login failure delay can enable app-layer DoS

backend/app/utils/constants.py

@@ -45,6 +45,19 @@ SESSION_TOKEN_SIGNATURE_SEPARATOR: Final[str] = "."
 SESSION_COOKIE_NAME: Final[str] = "bangui_session"
 """Name of the session cookie used by the browser SPA."""
 
+# ---------------------------------------------------------------------------
+# Authentication penalty (brute-force resistance)
+# ---------------------------------------------------------------------------
+
+LOGIN_PENALTY_BASE_SECONDS: Final[float] = 1.0
+"""Base penalty (seconds) for a failed login attempt."""
+
+LOGIN_PENALTY_MAX_SECONDS: Final[float] = 10.0
+"""Maximum penalty (seconds) for failed login attempts."""
+
+LOGIN_PENALTY_MULTIPLIER: Final[float] = 2.0
+"""Exponential multiplier applied per failed attempt."""
+
 # ---------------------------------------------------------------------------
 # Time-range presets (used by dashboard and history endpoints)
 # ---------------------------------------------------------------------------

backend/app/utils/rate_limiter.py

@@ -7,6 +7,10 @@ Old entries are cleaned up by a background task to prevent unbounded growth.
 Process-local implementation — in multi-worker setups, each worker has
 independent counters. This constraint limits the blast radius of brute-force
 attacks to a single worker.
+
+The penalty strategy for failed login attempts is also managed here:
+record_failure() records a failure timestamp and returns the penalty delay
+to apply, enabling progressive back-off without exhausting request capacity.
 """
 
 from __future__ import annotations
@@ -17,6 +21,12 @@ from typing import TYPE_CHECKING
 
 import structlog
 
+from app.utils.constants import (
+    LOGIN_PENALTY_BASE_SECONDS,
+    LOGIN_PENALTY_MAX_SECONDS,
+    LOGIN_PENALTY_MULTIPLIER,
+)
+
 if TYPE_CHECKING:
     from collections.abc import Mapping
 
@@ -48,6 +58,8 @@ class RateLimiter:
         self.max_attempts: int = max_attempts
         self.window_seconds: int = window_seconds
         self._attempts: dict[str, deque[float]] = {}
+        self._failures: dict[str, deque[float]] = {}
+        self._lock_counts: dict[str, int] = {}
 
     def is_allowed(self, ip_address: str) -> bool:
         """Check if a request from *ip_address* is allowed.
@@ -126,3 +138,84 @@ class RateLimiter:
     def reset(self) -> None:
         """Clear all tracked attempts (for testing)."""
         self._attempts.clear()
+        self._failures.clear()
+        self._lock_counts.clear()
+
+    # ---------------------------------------------------------------------------
+    # Penalty strategy for failed login attempts
+    # ---------------------------------------------------------------------------
+
+    def record_failure(self, ip_address: str) -> float:
+        """Record a failed login attempt and return the penalty delay in seconds.
+
+        Tracks consecutive failures per IP. Penalty grows exponentially with
+        each failure, bounded by :data:`~app.utils.constants.LOGIN_PENALTY_MAX_SECONDS`,
+        then resets the failure counter. This provides brute-force resistance
+        without exhausting request capacity.
+
+        A concurrency guard (``_lock_counts``) prevents a single IP from
+        accumulating many concurrent penalty tasks.
+
+        Args:
+            ip_address: The client IP address whose login attempt failed.
+
+        Returns:
+            The penalty delay in seconds to apply.
+        """
+        now = time()
+
+        if ip_address not in self._failures:
+            self._failures[ip_address] = deque()
+        if ip_address not in self._lock_counts:
+            self._lock_counts[ip_address] = 0
+
+        failures = self._failures[ip_address]
+        lock_count = self._lock_counts[ip_address]
+
+        # Reset if last failure is outside the window
+        cutoff = now - self.window_seconds
+        while failures and failures[0] < cutoff:
+            failures.popleft()
+
+        consecutive = len(failures)
+        penalty = min(
+            LOGIN_PENALTY_BASE_SECONDS * (LOGIN_PENALTY_MULTIPLIER ** consecutive),
+            LOGIN_PENALTY_MAX_SECONDS,
+        )
+
+        failures.append(now)
+
+        # Concurrency protection: if too many concurrent sleeps are already
+        # running for this IP, cap the penalty to avoid thread exhaustion.
+        if lock_count >= 3:
+            penalty = min(penalty, LOGIN_PENALTY_BASE_SECONDS)
+
+        return penalty
+
+    def acquire(self, ip_address: str) -> bool:
+        """Acquire a concurrency slot for a penalty task.
+
+        Args:
+            ip_address: The client IP address.
+
+        Returns:
+            ``True`` if the slot was acquired, ``False`` if the IP already has
+            the maximum number of concurrent penalty tasks running.
+        """
+        if ip_address not in self._lock_counts:
+            self._lock_counts[ip_address] = 0
+
+        if self._lock_counts[ip_address] >= 3:
+            return False
+
+        self._lock_counts[ip_address] += 1
+        return True
+
+    def release(self, ip_address: str) -> None:
+        """Release a concurrency slot when a penalty task completes.
+
+        Args:
+            ip_address: The client IP address.
+        """
+        if ip_address in self._lock_counts and self._lock_counts[ip_address] > 0:
+            self._lock_counts[ip_address] -= 1