Repository: jazzband/django-axes
Pull request #1222 (2024-08-14 08:59:23)

feat: pass request to `AXES_COOLOFF_TIME` callback
# What does this PR do?

<!--
Congratulations! You've made it this far! You're not quite done yet though.

Please replace this with a description of the change and which issue is fixed (if applicable). Please also include relevant motivation and context. List any dependencies (if any) that are required for this change.

Once you're done, someone will review your PR shortly. They may suggest changes to make the code even better.
-->

<!-- Remove if not applicable -->

If the `AXES_COOLOFF_TIME` is a callable or path to a callable taking an argument, pass the username to it.

This should enable users to customize the cool off to be user dependant, and possibly implement a growing cool-off time:

- First lockout cools off after 5 mins
- Second one after 10 mins
- etc...

Past references of this:
- https://github.com/jazzband/django-axes/issues/667#issuecomment-737751694
- #218

This PR should be backwards compatible with previous behaviour: if the `AXES_COOLOFF_TIME` is a callable that takes no argument, it will be called without parameters.

## Before submitting
- [ ] This PR fixes a typo or improves the docs (you can dismiss the other checks if that's the case).
- [x] Did you make sure to update the documentation with your changes?
- [x] Did you write any new necessary tests?


browniebroke: Looks like the test failures are the same as another PR (https://github.com/jazzband/django-axes/pull/1221) and they don't look related to my change so I'm going to ignore them for this. Any idea how to fix them? I'll try to take a look if I find time
aleksihakli: > Looks like the test failures are the same as another PR (#1221) and they don't look related to my change so I'm going to ignore them for this. Any idea how to fix them? I'll try to take a look if I find time

There was a failing test that was produced by a change in the Django upstream implementation that limits header sizes. I updated the test, if you rebase this PR changeset on top of jazzband/django-axes repository master branch the test should be fixed in this branch as well.

Patch:
diff --git a/axes/attempts.py b/axes/attempts.py
index 80055c1..d518b53 100644
--- a/axes/attempts.py
+++ b/axes/attempts.py
@@ -12,17 +12,18 @@ from axes.models import AccessAttempt
 log = getLogger(__name__)
 
 
-def get_cool_off_threshold(attempt_time: Optional[datetime] = None) -> datetime:
+def get_cool_off_threshold(request: Optional[HttpRequest] = None) -> datetime:
     """
     Get threshold for fetching access attempts from the database.
     """
 
-    cool_off = get_cool_off()
+    cool_off = get_cool_off(request)
     if cool_off is None:
         raise TypeError(
             "Cool off threshold can not be calculated with settings.AXES_COOLOFF_TIME set to None"
         )
 
+    attempt_time = request.axes_attempt_time
     if attempt_time is None:
         return now() - cool_off
     return attempt_time - cool_off
@@ -62,12 +63,12 @@ def get_user_attempts(
         )
         return attempts_list
 
-    threshold = get_cool_off_threshold(request.axes_attempt_time)
+    threshold = get_cool_off_threshold(request)
     log.debug("AXES: Getting access attempts that are newer than %s", threshold)
     return [attempts.filter(attempt_time__gte=threshold) for attempts in attempts_list]
 
 
-def clean_expired_user_attempts(attempt_time: Optional[datetime] = None) -> int:
+def clean_expired_user_attempts(request: Optional[HttpRequest] = None) -> int:
     """
     Clean expired user attempts from the database.
     """
@@ -78,7 +79,7 @@ def clean_expired_user_attempts(attempt_time: Optional[datetime] = None) -> int:
         )
         return 0
 
-    threshold = get_cool_off_threshold(attempt_time)
+    threshold = get_cool_off_threshold(request)
     count, _ = AccessAttempt.objects.filter(attempt_time__lt=threshold).delete()
     log.info(
         "AXES: Cleaned up %s expired access attempts from database that were older than %s",
diff --git a/axes/handlers/cache.py b/axes/handlers/cache.py
index a797f39..7646c71 100644
--- a/axes/handlers/cache.py
+++ b/axes/handlers/cache.py
@@ -113,7 +113,7 @@ class AxesCacheHandler(AbstractAxesHandler, AxesBaseHandler):
             return
 
         cache_keys = get_client_cache_keys(request, credentials)
-        cache_timeout = get_cache_timeout()
+        cache_timeout = get_cache_timeout(request)
         failures = []
         for cache_key in cache_keys:
             added = self.cache.add(key=cache_key, value=1, timeout=cache_timeout)
diff --git a/axes/handlers/database.py b/axes/handlers/database.py
index be2ded4..1b5f1d0 100644
--- a/axes/handlers/database.py
+++ b/axes/handlers/database.py
@@ -132,7 +132,7 @@ class AxesDatabaseHandler(AbstractAxesHandler, AxesBaseHandler):
             return
 
         # 1. database query: Clean up expired user attempts from the database before logging new attempts
-        clean_expired_user_attempts(request.axes_attempt_time)
+        clean_expired_user_attempts(request)
 
         username = get_client_username(request, credentials)
         client_str = get_client_str(
@@ -262,7 +262,7 @@ class AxesDatabaseHandler(AbstractAxesHandler, AxesBaseHandler):
         """
 
         # 1. database query: Clean up expired user attempts from the database
-        clean_expired_user_attempts(request.axes_attempt_time)
+        clean_expired_user_attempts(request)
 
         username = user.get_username()
         credentials = get_credentials(username)
@@ -305,7 +305,7 @@ class AxesDatabaseHandler(AbstractAxesHandler, AxesBaseHandler):
         """
 
         # 1. database query: Clean up expired user attempts from the database
-        clean_expired_user_attempts(request.axes_attempt_time)
+        clean_expired_user_attempts(request)
 
         username = user.get_username() if user else None
         client_str = get_client_str(
diff --git a/axes/helpers.py b/axes/helpers.py
index 78cc668..198460f 100644
--- a/axes/helpers.py
+++ b/axes/helpers.py
@@ -32,7 +32,7 @@ def get_cache() -> BaseCache:
     return caches[getattr(settings, "AXES_CACHE", "default")]
 
 
-def get_cache_timeout() -> Optional[int]:
+def get_cache_timeout(request: Optional[HttpRequest] = None) -> Optional[int]:
     """
     Return the cache timeout interpreted from settings.AXES_COOLOFF_TIME.
 
@@ -43,21 +43,22 @@ def get_cache_timeout() -> Optional[int]:
     for use with the Django cache backends.
     """
 
-    cool_off = get_cool_off()
+    cool_off = get_cool_off(request)
     if cool_off is None:
         return None
     return int(cool_off.total_seconds())
 
 
-def get_cool_off() -> Optional[timedelta]:
+def get_cool_off(request: Optional[HttpRequest] = None) -> Optional[timedelta]:
     """
     Return the login cool off time interpreted from settings.AXES_COOLOFF_TIME.
 
     The return value is either None or timedelta.
 
-    Notice that the settings.AXES_COOLOFF_TIME is either None, timedelta, or integer/float of hours,
-    and this function offers a unified _timedelta or None_ representation of that configuration
-    for use with the Axes internal implementations.
+    Notice that the settings.AXES_COOLOFF_TIME is either None, timedelta, integer/float of hours,
+    a path to a callable or a callable taking 1 argument (the request). This function
+    offers a unified _timedelta or None_ representation of that configuration for use with the
+    Axes internal implementations.
 
     :exception TypeError: if settings.AXES_COOLOFF_TIME is of wrong type.
     """
@@ -69,9 +70,10 @@ def get_cool_off() -> Optional[timedelta]:
     if isinstance(cool_off, float):
         return timedelta(minutes=cool_off * 60)
     if isinstance(cool_off, str):
-        return import_string(cool_off)()
+        cool_off_func = import_string(cool_off)
+        return cool_off_func(request)
     if callable(cool_off):
-        return cool_off()  # pylint: disable=not-callable
+        return cool_off(request)  # pylint: disable=not-callable
 
     return cool_off
 
@@ -462,7 +464,7 @@ def get_lockout_response(
         "username": get_client_username(request, credentials) or "",
     }
 
-    cool_off = get_cool_off()
+    cool_off = get_cool_off(request)
     if cool_off:
         context.update(
             {
diff --git a/docs/4_configuration.rst b/docs/4_configuration.rst
index 1a60594..9572244 100644
--- a/docs/4_configuration.rst
+++ b/docs/4_configuration.rst
@@ -23,7 +23,7 @@ The following ``settings.py`` options are available for customizing Axes behavio
 +------------------------------------------------------+----------------------------------------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
 | AXES_LOCK_OUT_AT_FAILURE                             | True                                         | After the number of allowed login attempts are exceeded, should we lock out this IP (and optional user agent)?                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                            |
 +------------------------------------------------------+----------------------------------------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
-| 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 no arguments.  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_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_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.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                     |
 +------------------------------------------------------+----------------------------------------------+-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
diff --git a/requirements.txt b/requirements.txt
index 8fbb0bd..2a2ac47 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -2,4 +2,4 @@
 -r requirements-qa.txt
 -r requirements-test.txt
 sphinx_rtd_theme==2.0.0
-tox==4.20.0
+tox==4.21.0


Test patch:
diff --git a/tests/test_attempts.py b/tests/test_attempts.py
index 04af617..188071b 100644
--- a/tests/test_attempts.py
+++ b/tests/test_attempts.py
@@ -1,7 +1,7 @@
 from unittest.mock import patch
 
 from django.http import HttpRequest
-from django.test import override_settings
+from django.test import override_settings, RequestFactory
 from django.utils.timezone import now
 
 from axes.attempts import get_cool_off_threshold
@@ -15,12 +15,13 @@ class GetCoolOffThresholdTestCase(AxesTestCase):
     def test_get_cool_off_threshold(self):
         timestamp = now()
 
+        request = RequestFactory().post("/")
         with patch("axes.attempts.now", return_value=timestamp):
-            attempt_time = timestamp
-            threshold_now = get_cool_off_threshold(attempt_time)
+            request.axes_attempt_time = timestamp
+            threshold_now = get_cool_off_threshold(request)
 
-            attempt_time = None
-            threshold_none = get_cool_off_threshold(attempt_time)
+            request.axes_attempt_time = None
+            threshold_none = get_cool_off_threshold(request)
 
         self.assertEqual(threshold_now, threshold_none)
 
diff --git a/tests/test_helpers.py b/tests/test_helpers.py
index 0201fbb..1d5c8aa 100644
--- a/tests/test_helpers.py
+++ b/tests/test_helpers.py
@@ -59,6 +59,38 @@ class CacheTestCase(AxesTestCase):
     def test_get_cache_timeout_none(self):
         self.assertEqual(get_cache_timeout(), None)
 
+    def test_get_increasing_cache_timeout_by_username(self):
+        user_durations = {
+            "ben": timedelta(minutes=5),
+            "jen": timedelta(minutes=10),
+        }
+
+        def _callback(request):
+            username = request.POST["username"] if request else object()
+            previous_duration = user_durations.get(username, timedelta())
+            user_durations[username] = previous_duration + timedelta(minutes=5)
+            return user_durations[username]
+
+        rf = RequestFactory()
+        ben_req = rf.post("/", data={"username": "ben"})
+        jen_req = rf.post("/", data={"username": "jen"})
+        james_req = rf.post("/", data={"username": "james"})
+
+        with override_settings(AXES_COOLOFF_TIME=_callback):
+            with self.subTest("no username"):
+                self.assertEqual(get_cache_timeout(), 300)
+
+            with self.subTest("ben"):
+                self.assertEqual(get_cache_timeout(ben_req), 600)
+                self.assertEqual(get_cache_timeout(ben_req), 900)
+                self.assertEqual(get_cache_timeout(ben_req), 1200)
+
+            with self.subTest("jen"):
+                self.assertEqual(get_cache_timeout(jen_req), 900)
+
+            with self.subTest("james"):
+                self.assertEqual(get_cache_timeout(james_req), 300)
+
 
 class TimestampTestCase(AxesTestCase):
     def test_iso8601(self):
@@ -915,7 +947,7 @@ class LockoutResponseTestCase(AxesTestCase):
         self.assertEqual(type(response), HttpResponse)
 
 
-def mock_get_cool_off_str():
+def mock_get_cool_off_str(req):
     return timedelta(seconds=30)
 
 
@@ -940,7 +972,7 @@ class AxesCoolOffTestCase(AxesTestCase):
     def test_get_cool_off_float_gt_0(self):
         self.assertEqual(get_cool_off(), timedelta(seconds=6120))
 
-    @override_settings(AXES_COOLOFF_TIME=lambda: timedelta(seconds=30))
+    @override_settings(AXES_COOLOFF_TIME=lambda r: timedelta(seconds=30))
     def test_get_cool_off_callable(self):
         self.assertEqual(get_cool_off(), timedelta(seconds=30))