From 1015bad45149e50d0806c7f575ddec7b9599d2b2 Mon Sep 17 00:00:00 2001
From: Antoine Dujardin <antoine@harfanglab.fr>
Date: Tue, 22 Mar 2022 11:11:34 +0100
Subject: [PATCH] Don't reset cooloff time in case of login attempt during
 lockout

---
 axes/backends.py          |  3 +++
 axes/handlers/cache.py    | 14 ++++++++++++--
 axes/handlers/database.py | 11 +++++++++++
 axes/handlers/proxy.py    |  3 ++-
 tests/test_login.py       | 20 ++++++++++++++++++--
 5 files changed, 46 insertions(+), 5 deletions(-)

diff --git a/axes/backends.py b/axes/backends.py
index 5616979..926eef0 100644
--- a/axes/backends.py
+++ b/axes/backends.py
@@ -51,6 +51,9 @@ class AxesBackend(ModelBackend):
         response_context = kwargs.get("response_context", {})
         response_context["error"] = error_msg
 
+        # This flag can be used later to check if it was Axes that denied the login attempt.
+        request.axes_locked_out = True
+
         # Raise an error that stops the authentication flows at django.contrib.auth.authenticate.
         # This error stops bubbling up at the authenticate call which catches backend PermissionDenied errors.
         # After this error is caught by authenticate it emits a signal indicating user login failed,
diff --git a/axes/handlers/cache.py b/axes/handlers/cache.py
index ea0671f..38dae3c 100644
--- a/axes/handlers/cache.py
+++ b/axes/handlers/cache.py
@@ -24,7 +24,6 @@ class AxesCacheHandler(AbstractAxesHandler, AxesBaseHandler):
 
     def __init__(self):
         self.cache = get_cache()
-        self.cache_timeout = get_cache_timeout()
 
     def reset_attempts(
         self,
@@ -84,6 +83,17 @@ class AxesCacheHandler(AbstractAxesHandler, AxesBaseHandler):
             )
             return
 
+        # If axes denied access, don't record the failed attempt as that would reset the lockout time.
+        if request.axes_locked_out:
+            request.axes_credentials = credentials
+            user_locked_out.send(
+                "axes",
+                request=request,
+                username=username,
+                ip_address=request.axes_ip_address,
+            )
+            return
+
         client_str = get_client_str(
             username,
             request.axes_ip_address,
@@ -115,7 +125,7 @@ class AxesCacheHandler(AbstractAxesHandler, AxesBaseHandler):
         cache_keys = get_client_cache_key(request, credentials)
         for cache_key in cache_keys:
             failures = self.cache.get(cache_key, default=0)
-            self.cache.set(cache_key, failures + 1, self.cache_timeout)
+            self.cache.set(cache_key, failures + 1, get_cache_timeout())
 
         if (
             settings.AXES_LOCK_OUT_AT_FAILURE
diff --git a/axes/handlers/database.py b/axes/handlers/database.py
index d2fc1a7..e69e06d 100644
--- a/axes/handlers/database.py
+++ b/axes/handlers/database.py
@@ -139,6 +139,17 @@ class AxesDatabaseHandler(AbstractAxesHandler, AxesBaseHandler):
             request,
         )
 
+        # If axes denied access, don't record the failed attempt as that would reset the lockout time.
+        if request.axes_locked_out:
+            request.axes_credentials = credentials
+            user_locked_out.send(
+                "axes",
+                request=request,
+                username=username,
+                ip_address=request.axes_ip_address,
+            )
+            return
+
         # This replaces null byte chars that crash saving failures.
         get_data = get_query_str(request.GET).replace("\0", "0x00")
         post_data = get_query_str(request.POST).replace("\0", "0x00")
diff --git a/axes/handlers/proxy.py b/axes/handlers/proxy.py
index 960d820..8c0a3f3 100644
--- a/axes/handlers/proxy.py
+++ b/axes/handlers/proxy.py
@@ -78,7 +78,8 @@ class AxesProxyHandler(AbstractAxesHandler, AxesBaseHandler):
             )
             return
         if not hasattr(request, "axes_updated"):
-            request.axes_locked_out = False
+            if not hasattr(request, "axes_locked_out"):
+                request.axes_locked_out = False
             request.axes_attempt_time = now()
             request.axes_ip_address = get_client_ip_address(request)
             request.axes_user_agent = get_client_user_agent(request)
diff --git a/tests/test_login.py b/tests/test_login.py
index bb0468a..9541825 100644
--- a/tests/test_login.py
+++ b/tests/test_login.py
@@ -3,8 +3,9 @@ Integration tests for the login handling.
 
 TODO: Clean up the tests in this module.
 """
-
+from datetime import timedelta
 from importlib import import_module
+from time import sleep
 
 from django.contrib.auth import get_user_model, login, logout
 from django.http import HttpRequest
@@ -12,7 +13,7 @@ from django.test import override_settings, TestCase
 from django.urls import reverse
 
 from axes.conf import settings
-from axes.helpers import get_cache, make_cache_key_list
+from axes.helpers import get_cache, make_cache_key_list, get_cool_off
 from axes.models import AccessAttempt
 from tests.base import AxesTestCase
 
@@ -631,6 +632,21 @@ class DatabaseLoginTestCase(AxesTestCase):
         response = self.client.get(reverse("admin:login"), REMOTE_ADDR=self.IP_1)
         self.assertContains(response, self.LOGIN_FORM_KEY, status_code=200, html=True)
 
+    @override_settings(AXES_COOLOFF_TIME=timedelta(seconds=1))
+    def test_login_during_lockout_doesnt_reset_cool_off_time(self):
+        # Lockout
+        self.lockout()
+
+        # Attempt during lockout
+        sleep_time = get_cool_off().total_seconds() / 2
+        sleep(sleep_time)
+        self.login()
+        sleep(sleep_time)
+
+        # New attempt after initial lockout period: should work
+        response = self.login(is_valid_username=True, is_valid_password=True)
+        self.assertNotContains(response, self.LOCKED_MESSAGE, status_code=302)
+
 
 # Test the same logic with cache handler
 @override_settings(AXES_HANDLER="axes.handlers.cache.AxesCacheHandler")