feat: introduce progressive lockout tiers for dynamic cool-off periods based on failure count.

Author: rodrigo.nogueira

axes/checks.py

@@ -6,7 +6,7 @@
 from django.utils.module_loading import import_string
 
 from axes.backends import AxesStandaloneBackend
-from axes.conf import settings
+from axes.conf import LockoutTier, settings
 
 
 class Messages:
@@ -26,6 +26,14 @@ class Messages:
         "AXES_LOCKOUT_PARAMETERS does not contain 'ip_address'."
         " This configuration allows attackers to bypass rate limits by rotating User-Agents or Cookies."
     )
+    LOCKOUT_TIERS_CONFLICT = (
+        "AXES_LOCKOUT_TIERS is set alongside AXES_COOLOFF_TIME."
+        " When tiers are active, AXES_COOLOFF_TIME is ignored."
+        " Remove AXES_COOLOFF_TIME to silence this warning."
+    )
+    LOCKOUT_TIERS_INVALID = (
+        "AXES_LOCKOUT_TIERS must be a list of LockoutTier instances."
+    )
 
 
 class Hints:
@@ -35,6 +43,10 @@ class Hints:
     SETTING_DEPRECATED = None
     CALLABLE_INVALID = None
     LOCKOUT_PARAMETERS_INVALID = "Add 'ip_address' to AXES_LOCKOUT_PARAMETERS."
+    LOCKOUT_TIERS_CONFLICT = "Remove AXES_COOLOFF_TIME when using AXES_LOCKOUT_TIERS."
+    LOCKOUT_TIERS_INVALID = (
+        "Use: AXES_LOCKOUT_TIERS = [LockoutTier(failures=3, cooloff=timedelta(minutes=15)), ...]"
+    )
 
 
 class Codes:
@@ -44,6 +56,8 @@ class Codes:
     SETTING_DEPRECATED = "axes.W004"
     CALLABLE_INVALID = "axes.W005"
     LOCKOUT_PARAMETERS_INVALID = "axes.W006"
+    LOCKOUT_TIERS_CONFLICT = "axes.W007"
+    LOCKOUT_TIERS_INVALID = "axes.W008"
 
 
 @register(Tags.security, Tags.caches, Tags.compatibility)
@@ -192,6 +206,41 @@ def axes_lockout_params_check(app_configs, **kwargs):  # pylint: disable=unused-
     return warnings
 
 
+@register(Tags.security)
+def axes_lockout_tiers_check(app_configs, **kwargs):  # pylint: disable=unused-argument
+    warnings = []
+    tiers = getattr(settings, "AXES_LOCKOUT_TIERS", None)
+    if tiers is None:
+        return warnings
+
+    if not _is_valid_tiers_list(tiers):
+        warnings.append(
+            Warning(
+                msg=Messages.LOCKOUT_TIERS_INVALID,
+                hint=Hints.LOCKOUT_TIERS_INVALID,
+                id=Codes.LOCKOUT_TIERS_INVALID,
+            )
+        )
+        return warnings
+
+    if getattr(settings, "AXES_COOLOFF_TIME", None) is not None:
+        warnings.append(
+            Warning(
+                msg=Messages.LOCKOUT_TIERS_CONFLICT,
+                hint=Hints.LOCKOUT_TIERS_CONFLICT,
+                id=Codes.LOCKOUT_TIERS_CONFLICT,
+            )
+        )
+
+    return warnings
+
+
+def _is_valid_tiers_list(tiers) -> bool:
+    if not isinstance(tiers, (list, tuple)):
+        return False
+    return all(isinstance(t, LockoutTier) for t in tiers)
+
+
 @register
 def axes_conf_check(app_configs, **kwargs):  # pylint: disable=unused-argument
     warnings = []

axes/conf.py

@@ -1,7 +1,16 @@
+from dataclasses import dataclass
+from datetime import timedelta
+
 from django.conf import settings
 from django.contrib.auth import get_user_model
 from django.utils.translation import gettext_lazy as _
 
+
+@dataclass(frozen=True, order=True)
+class LockoutTier:
+    failures: int
+    cooloff: timedelta
+
 # disable plugin when set to False
 settings.AXES_ENABLED = getattr(settings, "AXES_ENABLED", True)
 
@@ -87,6 +96,10 @@
 
 settings.AXES_COOLOFF_TIME = getattr(settings, "AXES_COOLOFF_TIME", None)
 
+# Progressive lockout tiers: list of LockoutTier(failures, cooloff) instances.
+# When set, overrides AXES_FAILURE_LIMIT and AXES_COOLOFF_TIME.
+settings.AXES_LOCKOUT_TIERS = getattr(settings, "AXES_LOCKOUT_TIERS", None)
+
 settings.AXES_USE_ATTEMPT_EXPIRATION = getattr(settings, "AXES_USE_ATTEMPT_EXPIRATION", False)
 
 settings.AXES_VERBOSE = getattr(settings, "AXES_VERBOSE", settings.AXES_ENABLED)

axes/helpers.py

@@ -11,7 +11,7 @@
 from django.utils.encoding import force_bytes
 from django.utils.module_loading import import_string
 
-from axes.conf import settings
+from axes.conf import LockoutTier, settings
 from axes.models import AccessBase
 
 log = getLogger(__name__)
@@ -60,9 +60,16 @@ def get_cool_off(request: Optional[HttpRequest] = None) -> Optional[timedelta]:
     offers a unified _timedelta or None_ representation of that configuration for use with the
     Axes internal implementations.
 
+    When ``AXES_LOCKOUT_TIERS`` is configured, the cooloff is resolved from the
+    matching tier based on the failure count attached to the request.
+
     :exception TypeError: if settings.AXES_COOLOFF_TIME is of wrong type.
     """
 
+    tier = _resolve_tier_from_request(request)
+    if tier is not None:
+        return tier.cooloff
+
     cool_off = settings.AXES_COOLOFF_TIME
 
     if isinstance(cool_off, int):
@@ -100,6 +107,31 @@ def get_cool_off_iso8601(delta: timedelta) -> str:
         return f"P{days_str}T{time_str}"
     return f"P{days_str}"
 
+
+def get_lockout_tier(failures: int) -> Optional[LockoutTier]:
+    """Return the matching ``LockoutTier`` for *failures*, or ``None``."""
+    tiers = settings.AXES_LOCKOUT_TIERS
+    if not tiers:
+        return None
+    sorted_tiers = sorted(tiers, key=lambda t: t.failures)
+    matched = None
+    for tier in sorted_tiers:
+        if failures >= tier.failures:
+            matched = tier
+    return matched
+
+
+def _resolve_tier_from_request(
+    request: Optional[HttpRequest],
+) -> Optional[LockoutTier]:
+    """Extract failure count from *request* and resolve the tier."""
+    if not settings.AXES_LOCKOUT_TIERS or request is None:
+        return None
+    failures = getattr(request, "axes_failures_since_start", None)
+    if failures is None:
+        return None
+    return get_lockout_tier(failures)
+
 def get_attempt_expiration(request: Optional[HttpRequest] = None) -> datetime:
     """
     Get threshold for fetching access attempts from the database.
@@ -442,6 +474,11 @@ def get_query_str(query: Type[QueryDict], max_length: int = 1024) -> str:
 
 
 def get_failure_limit(request: HttpRequest, credentials) -> int:
+    tiers = settings.AXES_LOCKOUT_TIERS
+    if tiers:
+        sorted_tiers = sorted(tiers, key=lambda t: t.failures)
+        return sorted_tiers[0].failures
+
     if callable(settings.AXES_FAILURE_LIMIT):
         return settings.AXES_FAILURE_LIMIT(  # pylint: disable=not-callable
             request, credentials
@@ -454,7 +491,7 @@ def get_failure_limit(request: HttpRequest, credentials) -> int:
 
 
 def get_lockout_message() -> str:
-    if settings.AXES_COOLOFF_TIME:
+    if settings.AXES_COOLOFF_TIME or settings.AXES_LOCKOUT_TIERS:
         return settings.AXES_COOLOFF_MESSAGE
     return settings.AXES_PERMALOCK_MESSAGE
 
@@ -486,8 +523,10 @@ def get_lockout_response(
         )
 
     status = settings.AXES_HTTP_RESPONSE_CODE
+    failures = getattr(request, "axes_failures_since_start", None) or 0
     context = {
         "failure_limit": get_failure_limit(request, credentials),
+        "failure_count": failures,
         "username": get_client_username(request, credentials) or "",
     }
 

docs/4_configuration.rst

@@ -25,6 +25,8 @@ The following ``settings.py`` options are available for customizing Axes behavio
 +------------------------------------------------------+----------------------------------------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
 | AXES_COOLOFF_TIME                                    | None                                         | If set, defines a period of inactivity after which  old failed login attempts will be cleared.  Can be set to a Python timedelta object, an integer, a float, a callable, or a string path to a callable which takes the request as argument.  If an integer or float, will be interpreted as a number of hours:  ``AXES_COOLOFF_TIME = 2`` 2 hours,   ``AXES_COOLOFF_TIME = 2.0`` 2 hours, 120 minutes,  ``AXES_COOLOFF_TIME = 1.7`` 1.7 hours, 102 minutes, 6120 seconds                                                                                                                                                                                                                                                                |
 +------------------------------------------------------+----------------------------------------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
+| AXES_LOCKOUT_TIERS                                   | None                                         | A list of ``LockoutTier(failures, cooloff)`` instances that define progressive lockout durations.  When set, overrides ``AXES_FAILURE_LIMIT`` and ``AXES_COOLOFF_TIME``.  The lowest tier threshold becomes the effective failure limit, and each subsequent tier applies a longer cool-off.  Example: ``from datetime import timedelta``, ``from axes.conf import LockoutTier``, ``AXES_LOCKOUT_TIERS = [LockoutTier(failures=3, cooloff=timedelta(minutes=15)), LockoutTier(failures=6, cooloff=timedelta(hours=2)), LockoutTier(failures=10, cooloff=timedelta(days=1))]``.  With this configuration: 3 failures → 15 min lockout, 6 failures → 2 h, 10+ failures → 24 h.                                                               |
++------------------------------------------------------+----------------------------------------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
 | AXES_ONLY_ADMIN_SITE                                 | False                                        | If ``True``, lock is only enabled for admin site. Admin site is determined by checking request path against the path of ``"admin:index"`` view. If admin urls are not registered in current urlconf, all requests will not be locked.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                     |
 +------------------------------------------------------+----------------------------------------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
 | AXES_ONLY_USER_FAILURES                              | False                                        | DEPRECATED: USE ``AXES_LOCKOUT_PARAMETERS`` INSTEAD. If ``True``, only lock based on username, and never lock based on IP if attempts exceed the limit. Otherwise utilize the existing IP and user locking logic.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                         |

tests/test_checks.py

@@ -1,8 +1,11 @@
+from datetime import timedelta
+
 from django.core.checks import run_checks, Warning  # pylint: disable=redefined-builtin
 from django.test import override_settings, modify_settings
 
 from axes.backends import AxesStandaloneBackend
 from axes.checks import Messages, Hints, Codes
+from axes.conf import LockoutTier
 from tests.base import AxesTestCase
 
 
@@ -129,3 +132,46 @@ def test_invalid_import_path(self):
     def test_valid_callable(self):
         warnings = run_checks()
         self.assertEqual(warnings, [])
+
+
+class LockoutTiersCheckTestCase(AxesTestCase):
+    SAMPLE_TIERS = [
+        LockoutTier(failures=3, cooloff=timedelta(minutes=15)),
+        LockoutTier(failures=6, cooloff=timedelta(hours=2)),
+    ]
+
+    @override_settings(AXES_LOCKOUT_TIERS=SAMPLE_TIERS, AXES_COOLOFF_TIME=None)
+    def test_tiers_alone_no_warning(self):
+        warnings = run_checks()
+        self.assertEqual(warnings, [])
+
+    @override_settings(AXES_LOCKOUT_TIERS=SAMPLE_TIERS, AXES_COOLOFF_TIME=1)
+    def test_tiers_with_cooloff_time_warns(self):
+        warnings = run_checks()
+        warning = Warning(
+            msg=Messages.LOCKOUT_TIERS_CONFLICT,
+            hint=Hints.LOCKOUT_TIERS_CONFLICT,
+            id=Codes.LOCKOUT_TIERS_CONFLICT,
+        )
+        self.assertIn(warning, warnings)
+
+    @override_settings(AXES_LOCKOUT_TIERS="not a list")
+    def test_tiers_invalid_format_warns(self):
+        warnings = run_checks()
+        warning = Warning(
+            msg=Messages.LOCKOUT_TIERS_INVALID,
+            hint=Hints.LOCKOUT_TIERS_INVALID,
+            id=Codes.LOCKOUT_TIERS_INVALID,
+        )
+        self.assertIn(warning, warnings)
+
+    @override_settings(AXES_LOCKOUT_TIERS=[(3, timedelta(minutes=15))])
+    def test_tiers_plain_tuples_warns(self):
+        warnings = run_checks()
+        warning = Warning(
+            msg=Messages.LOCKOUT_TIERS_INVALID,
+            hint=Hints.LOCKOUT_TIERS_INVALID,
+            id=Codes.LOCKOUT_TIERS_INVALID,
+        )
+        self.assertIn(warning, warnings)
+