add LOCK_OUT_BY_USER_OR_IP option

store all AccessAttempt records

axes/apps.py

@@ -41,6 +41,8 @@ def initialize(cls):
             log.info("AXES: blocking by username only.")
         elif settings.AXES_LOCK_OUT_BY_COMBINATION_USER_AND_IP:
             log.info("AXES: blocking by combination of username and IP.")
+        elif settings.AXES_LOCK_OUT_BY_USER_OR_IP:
+            log.info("AXES: blocking by username or IP.")
         else:
             log.info("AXES: blocking by IP only.")
 

axes/attempts.py

@@ -33,29 +33,32 @@ def filter_user_attempts(request, credentials: dict = None) -> QuerySet:
 
     username = get_client_username(request, credentials)
 
-    filter_kwargs = get_client_parameters(
+    filter_kwargs_list = get_client_parameters(
         username, request.axes_ip_address, request.axes_user_agent
     )
-
-    return AccessAttempt.objects.filter(**filter_kwargs)
+    attempts_list = [
+        AccessAttempt.objects.filter(**filter_kwargs)
+        for filter_kwargs in filter_kwargs_list
+    ]
+    return attempts_list
 
 
 def get_user_attempts(request, credentials: dict = None) -> QuerySet:
     """
     Get valid user attempts that match the given request and credentials.
     """
 
-    attempts = filter_user_attempts(request, credentials)
+    attempts_list = filter_user_attempts(request, credentials)
 
     if settings.AXES_COOLOFF_TIME is None:
         log.debug(
             "AXES: Getting all access attempts from database because no AXES_COOLOFF_TIME is configured"
         )
-        return attempts
+        return attempts_list
 
     threshold = get_cool_off_threshold(request.axes_attempt_time)
     log.debug("AXES: Getting access attempts that are newer than %s", threshold)
-    return attempts.filter(attempt_time__gte=threshold)
+    return [attempts.filter(attempt_time__gte=threshold) for attempts in attempts_list]
 
 
 def clean_expired_user_attempts(attempt_time: datetime = None) -> int:
@@ -84,9 +87,12 @@ def reset_user_attempts(request, credentials: dict = None) -> int:
     Reset all user attempts that match the given request and credentials.
     """
 
-    attempts = filter_user_attempts(request, credentials)
+    attempts_list = filter_user_attempts(request, credentials)
 
-    count, _ = attempts.delete()
+    count = 0
+    for attempts in attempts_list:
+        _count, _ = attempts.delete()
+        count += _count
     log.info("AXES: Reset %s access attempts from database.", count)
 
     return count

axes/conf.py

@@ -20,6 +20,9 @@ class Meta:
     # lock out with the combination of username and IP address
     LOCK_OUT_BY_COMBINATION_USER_AND_IP = False
 
+    # lock out with the username or IP address
+    LOCK_OUT_BY_USER_OR_IP = False
+
     # lock out with username and never the IP or user agent
     ONLY_USER_FAILURES = False
 

axes/handlers/base.py

@@ -146,7 +146,13 @@ def is_admin_site(self, request) -> bool:
 
         return False
 
-    def reset_attempts(self, *, ip_address: str = None, username: str = None) -> int:
+    def reset_attempts(
+        self,
+        *,
+        ip_address: str = None,
+        username: str = None,
+        ip_or_username: bool = False,
+    ) -> int:
         """
         Resets access attempts that match the given IP address or username.
 

axes/handlers/cache.py

@@ -26,8 +26,11 @@ def __init__(self):
         self.cache_timeout = get_cache_timeout()
 
     def get_failures(self, request, credentials: dict = None) -> int:
-        cache_key = get_client_cache_key(request, credentials)
-        return self.cache.get(cache_key, default=0)
+        cache_keys = get_client_cache_key(request, credentials)
+        failure_count = max(
+            self.cache.get(cache_key, default=0) for cache_key in cache_keys
+        )
+        return failure_count
 
     def user_login_failed(
         self, sender, credentials: dict, request=None, **kwargs
@@ -71,8 +74,10 @@ def user_login_failed(
                 client_str,
             )
 
-        cache_key = get_client_cache_key(request, credentials)
-        self.cache.set(cache_key, failures_since_start, self.cache_timeout)
+        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)
 
         if (
             settings.AXES_LOCK_OUT_AT_FAILURE
@@ -109,14 +114,15 @@ def user_logged_in(
         log.info("AXES: Successful login by %s.", client_str)
 
         if settings.AXES_RESET_ON_SUCCESS:
-            cache_key = get_client_cache_key(request, credentials)
-            failures_since_start = self.cache.get(cache_key, default=0)
-            self.cache.delete(cache_key)
-            log.info(
-                "AXES: Deleted %d failed login attempts by %s from cache.",
-                failures_since_start,
-                client_str,
-            )
+            cache_keys = get_client_cache_key(request, credentials)
+            for cache_key in cache_keys:
+                failures_since_start = self.cache.get(cache_key, default=0)
+                self.cache.delete(cache_key)
+                log.info(
+                    "AXES: Deleted %d failed login attempts by %s from cache.",
+                    failures_since_start,
+                    client_str,
+                )
 
     def user_logged_out(self, sender, request, user, **kwargs):
         username = user.get_username() if user else None

axes/handlers/database.py

@@ -1,6 +1,6 @@
 from logging import getLogger
 
-from django.db.models import Max, Value
+from django.db.models import Sum, Value, Q
 from django.db.models.functions import Concat
 from django.utils import timezone
 
@@ -33,13 +33,22 @@ class AxesDatabaseHandler(AbstractAxesHandler, AxesBaseHandler):
               process, caching its output can be dangerous.
     """
 
-    def reset_attempts(self, *, ip_address: str = None, username: str = None) -> int:
+    def reset_attempts(
+        self,
+        *,
+        ip_address: str = None,
+        username: str = None,
+        ip_or_username: bool = False,
+    ) -> int:
         attempts = AccessAttempt.objects.all()
 
-        if ip_address:
-            attempts = attempts.filter(ip_address=ip_address)
-        if username:
-            attempts = attempts.filter(username=username)
+        if ip_or_username:
+            attempts = attempts.filter(Q(ip_address=ip_address) | Q(username=username))
+        else:
+            if ip_address:
+                attempts = attempts.filter(ip_address=ip_address)
+            if username:
+                attempts = attempts.filter(username=username)
 
         count, _ = attempts.delete()
         log.info("AXES: Reset %d access attempts from database.", count)
@@ -62,11 +71,17 @@ def reset_logs(self, *, age_days: int = None) -> int:
         return count
 
     def get_failures(self, request, credentials: dict = None) -> int:
-        attempts = get_user_attempts(request, credentials)
-        return (
-            attempts.aggregate(Max("failures_since_start"))["failures_since_start__max"]
-            or 0
+        attempts_list = get_user_attempts(request, credentials)
+        attempt_count = max(
+            (
+                attempts.aggregate(Sum("failures_since_start"))[
+                    "failures_since_start__sum"
+                ]
+                or 0
+            )
+            for attempts in attempts_list
         )
+        return attempt_count
 
     def user_login_failed(
         self, sender, credentials: dict, request=None, **kwargs
@@ -106,31 +121,33 @@ def user_login_failed(
         failures_since_start = 1 + self.get_failures(request, credentials)
 
         # 3. database query: Insert or update access records with the new failure data
-        if failures_since_start > 1:
+        try:
+            attempt = AccessAttempt.objects.get(
+                username=username,
+                ip_address=request.axes_ip_address,
+                user_agent=request.axes_user_agent,
+            )
             # Update failed attempt information but do not touch the username, IP address, or user agent fields,
             # because attackers can request the site with multiple different configurations
             # in order to bypass the defense mechanisms that are used by the site.
 
             log.warning(
                 "AXES: Repeated login failure by %s. Count = %d of %d. Updating existing record in the database.",
                 client_str,
-                failures_since_start,
+                attempt.failures_since_start,
                 get_failure_limit(request, credentials),
             )
 
             separator = "\n---------\n"
 
-            attempts = get_user_attempts(request, credentials)
-            attempts.update(
-                get_data=Concat("get_data", Value(separator + get_data)),
-                post_data=Concat("post_data", Value(separator + post_data)),
-                http_accept=request.axes_http_accept,
-                path_info=request.axes_path_info,
-                failures_since_start=failures_since_start,
-                attempt_time=request.axes_attempt_time,
-                username=username,
-            )
-        else:
+            attempt.get_data = Concat("get_data", Value(separator + get_data))
+            attempt.post_data = Concat("post_data", Value(separator + post_data))
+            attempt.http_accept = request.axes_http_accept
+            attempt.path_info = request.axes_path_info
+            attempt.failures_since_start += 1
+            attempt.attempt_time = request.axes_attempt_time
+            attempt.save()
+        except AccessAttempt.DoesNotExist:
             # Record failed attempt with all the relevant information.
             # Filtering based on username, IP address and user agent handled elsewhere,
             # and this handler just records the available information for further use.
@@ -148,7 +165,7 @@ def user_login_failed(
                 post_data=post_data,
                 http_accept=request.axes_http_accept,
                 path_info=request.axes_path_info,
-                failures_since_start=failures_since_start,
+                failures_since_start=1,
                 attempt_time=request.axes_attempt_time,
             )
 

axes/handlers/proxy.py

@@ -43,9 +43,15 @@ def get_implementation(cls, force: bool = False) -> AxesHandler:
         return cls.implementation
 
     @classmethod
-    def reset_attempts(cls, *, ip_address: str = None, username: str = None) -> int:
+    def reset_attempts(
+        cls,
+        *,
+        ip_address: str = None,
+        username: str = None,
+        ip_or_username: bool = False,
+    ) -> int:
         return cls.get_implementation().reset_attempts(
-            ip_address=ip_address, username=username
+            ip_address=ip_address, username=username, ip_or_username=ip_or_username
         )
 
     @classmethod

axes/handlers/test.py

@@ -6,7 +6,13 @@ class AxesTestHandler(AxesHandler):  # pylint: disable=unused-argument
     Signal handler implementation that does nothing, ideal for a test suite.
     """
 
-    def reset_attempts(self, *, ip_address: str = None, username: str = None) -> int:
+    def reset_attempts(
+        self,
+        *,
+        ip_address: str = None,
+        username: str = None,
+        ip_or_username: bool = False,
+    ) -> int:
         return 0
 
     def reset_logs(self, *, age_days: int = None) -> int:

axes/helpers.py

@@ -174,40 +174,46 @@ def get_client_http_accept(request) -> str:
     return request.META.get("HTTP_ACCEPT", "<unknown>")[:1025]
 
 
-def get_client_parameters(username: str, ip_address: str, user_agent: str) -> dict:
+def get_client_parameters(username: str, ip_address: str, user_agent: str) -> list:
     """
     Get query parameters for filtering AccessAttempt queryset.
 
     This method returns a dict that guarantees iteration order for keys and values,
     and can so be used in e.g. the generation of hash keys or other deterministic functions.
-    """
 
-    filter_kwargs = dict()
+    Returns list of dict, every item of list are separate parameters
+    """
 
     if settings.AXES_ONLY_USER_FAILURES:
         # 1. Only individual usernames can be tracked with parametrization
-        filter_kwargs["username"] = username
+        filter_query = [{"username": username}]
     else:
-        if settings.AXES_LOCK_OUT_BY_COMBINATION_USER_AND_IP:
+        if settings.AXES_LOCK_OUT_BY_USER_OR_IP:
+            # One of `username` or `IP address` is used
+            filter_query = [{"username": username}, {"ip_address": ip_address}]
+        elif settings.AXES_LOCK_OUT_BY_COMBINATION_USER_AND_IP:
             # 2. A combination of username and IP address can be used as well
-            filter_kwargs["username"] = username
-            filter_kwargs["ip_address"] = ip_address
+            filter_query = [{"username": username, "ip_address": ip_address}]
         else:
             # 3. Default case is to track the IP address only, which is the most secure option
-            filter_kwargs["ip_address"] = ip_address
+            filter_query = [{"ip_address": ip_address}]
 
         if settings.AXES_USE_USER_AGENT:
             # 4. The HTTP User-Agent can be used to track e.g. one browser
-            filter_kwargs["user_agent"] = user_agent
+            filter_query.append({"user_agent": user_agent})
 
-    return filter_kwargs
+    return filter_query
 
 
-def make_cache_key(filter_kwargs):
-    cache_key_components = "".join(value for value in filter_kwargs.values() if value)
-    cache_key_digest = md5(cache_key_components.encode()).hexdigest()
-    cache_key = f"axes-{cache_key_digest}"
-    return cache_key
+def make_cache_key_list(filter_kwargs_list):
+    cache_keys = []
+    for filter_kwargs in filter_kwargs_list:
+        cache_key_components = "".join(
+            value for value in filter_kwargs.values() if value
+        )
+        cache_key_digest = md5(cache_key_components.encode()).hexdigest()
+        cache_keys.append(f"axes-{cache_key_digest}")
+    return cache_keys
 
 
 def get_client_cache_key(
@@ -230,9 +236,9 @@ def get_client_cache_key(
         ip_address = get_client_ip_address(request_or_attempt)
         user_agent = get_client_user_agent(request_or_attempt)
 
-    filter_kwargs = get_client_parameters(username, ip_address, user_agent)
+    filter_kwargs_list = get_client_parameters(username, ip_address, user_agent)
 
-    return make_cache_key(filter_kwargs)
+    return make_cache_key_list(filter_kwargs_list)
 
 
 def get_client_str(
@@ -254,7 +260,10 @@ def get_client_str(
         client_dict["user_agent"] = user_agent
     else:
         # Other modes initialize the attributes that are used for the actual lockouts
-        client_dict = get_client_parameters(username, ip_address, user_agent)
+        client_list = get_client_parameters(username, ip_address, user_agent)
+        client_dict = {}
+        for d in client_list:
+            client_dict.update(d)
 
     # Path info is always included as last component in the client string for traceability purposes
     if path_info and isinstance(path_info, (tuple, list)):

axes/tests/test_handlers.py

@@ -191,8 +191,8 @@ def test_whitelist(self, log):
 
     def test_user_login_failed_multiple_username(self):
         configurations = (
-            (1, 2, {}, ["admin", "admin1"]),
-            (1, 2, {"AXES_USE_USER_AGENT": True}, ["admin", "admin1"]),
+            (2, 1, {}, ["admin", "admin1"]),
+            (2, 1, {"AXES_USE_USER_AGENT": True}, ["admin", "admin1"]),
             (2, 1, {"AXES_ONLY_USER_FAILURES": True}, ["admin", "admin1"]),
             (
                 2,
@@ -206,6 +206,8 @@ def test_user_login_failed_multiple_username(self):
                 {"AXES_LOCK_OUT_BY_COMBINATION_USER_AND_IP": True},
                 ["admin", "admin"],
             ),
+            (1, 2, {"AXES_LOCK_OUT_BY_USER_OR_IP": True}, ["admin", "admin"]),
+            (2, 1, {"AXES_LOCK_OUT_BY_USER_OR_IP": True}, ["admin", "admin1"]),
         )
 
         for (