Hopiu/django-axes
1
0
Fork 0
mirror of https://github.com/jazzband/django-axes.git synced 2026-03-16 22:30:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

acquire a lock of AccessAttempt before updating

Browse source
This commit is contained in:
Yuta Okamoto 2021-09-08 20:02:06 +09:00 committed by Aleksi Häkli
parent b44877a085
commit 46bfc54a03
1 changed files with 45 additions and 38 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

83
axes/handlers/database.py
View file

@ -1,5 +1,6 @@
from logging import getLogger
from django.db import transaction
from django.db.models import F, Sum, Value, Q
from django.db.models.functions import Concat
from django.utils import timezone
@ -123,48 +124,54 @@ class AxesDatabaseHandler(AbstractAxesHandler, AxesBaseHandler):
"AXES: Username is None and AXES_ONLY_USER_FAILURES is enabled, new record will NOT be created."
)
else:
attempt, created = AccessAttempt.objects.get_or_create(
username=username,
ip_address=request.axes_ip_address,
user_agent=request.axes_user_agent,
defaults={
"get_data": get_data,
"post_data": post_data,
"http_accept": request.axes_http_accept,
"path_info": request.axes_path_info,
"failures_since_start": 1,
"attempt_time": request.axes_attempt_time,
},
)
# 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.
if created:
log.warning(
"AXES: New login failure by %s. Created new record in the database.",
client_str,
with transaction.atomic():
(
attempt,
created,
) = AccessAttempt.objects.select_for_update().get_or_create(
username=username,
ip_address=request.axes_ip_address,
user_agent=request.axes_user_agent,
defaults={
"get_data": get_data,
"post_data": post_data,
"http_accept": request.axes_http_accept,
"path_info": request.axes_path_info,
"failures_since_start": 1,
"attempt_time": request.axes_attempt_time,
},
)
# 3. database query if there were previous attempts in the database
# 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.
else:
separator = "\n---------\n"
# 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.
if created:
log.warning(
"AXES: New login failure by %s. Created new record in the database.",
client_str,
)
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 = F("failures_since_start") + 1
attempt.attempt_time = request.axes_attempt_time
attempt.save()
# 3. database query if there were previous attempts in the database
# 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.
else:
separator = "\n---------\n"
log.warning(
"AXES: Repeated login failure by %s. Updated existing record in the database.",
client_str,
)
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 = F("failures_since_start") + 1
attempt.attempt_time = request.axes_attempt_time
attempt.save()
log.warning(
"AXES: Repeated login failure by %s. Updated existing record in the database.",
client_str,
)
# 3. or 4. database query: Calculate the current maximum failure number from the existing attempts
failures_since_start = self.get_failures(request, credentials)