attempts.py 3.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899
  1. from logging import getLogger
  2. from typing import List
  3. from django.db.models import QuerySet
  4. from django.utils.timezone import datetime, now
  5. from axes.conf import settings
  6. from axes.helpers import get_client_username, get_client_parameters, get_cool_off
  7. from axes.models import AccessAttempt
  8. log = getLogger(__name__)
  9. def get_cool_off_threshold(attempt_time: datetime = None) -> datetime:
  10. """
  11. Get threshold for fetching access attempts from the database.
  12. """
  13. cool_off = get_cool_off()
  14. if cool_off is None:
  15. raise TypeError(
  16. "Cool off threshold can not be calculated with settings.AXES_COOLOFF_TIME set to None"
  17. )
  18. if attempt_time is None:
  19. return now() - cool_off
  20. return attempt_time - cool_off
  21. def filter_user_attempts(request, credentials: dict = None) -> List[QuerySet]:
  22. """
  23. Return a list querysets of AccessAttempts that match the given request and credentials.
  24. """
  25. username = get_client_username(request, credentials)
  26. filter_kwargs_list = get_client_parameters(
  27. username, request.axes_ip_address, request.axes_user_agent
  28. )
  29. attempts_list = [
  30. AccessAttempt.objects.filter(**filter_kwargs)
  31. for filter_kwargs in filter_kwargs_list
  32. ]
  33. return attempts_list
  34. def get_user_attempts(request, credentials: dict = None) -> List[QuerySet]:
  35. """
  36. Get list of querysets with valid user attempts that match the given request and credentials.
  37. """
  38. attempts_list = filter_user_attempts(request, credentials)
  39. if settings.AXES_COOLOFF_TIME is None:
  40. log.debug(
  41. "AXES: Getting all access attempts from database because no AXES_COOLOFF_TIME is configured"
  42. )
  43. return attempts_list
  44. threshold = get_cool_off_threshold(request.axes_attempt_time)
  45. log.debug("AXES: Getting access attempts that are newer than %s", threshold)
  46. return [attempts.filter(attempt_time__gte=threshold) for attempts in attempts_list]
  47. def clean_expired_user_attempts(attempt_time: datetime = None) -> int:
  48. """
  49. Clean expired user attempts from the database.
  50. """
  51. if settings.AXES_COOLOFF_TIME is None:
  52. log.debug(
  53. "AXES: Skipping clean for expired access attempts because no AXES_COOLOFF_TIME is configured"
  54. )
  55. return 0
  56. threshold = get_cool_off_threshold(attempt_time)
  57. count, _ = AccessAttempt.objects.filter(attempt_time__lt=threshold).delete()
  58. log.info(
  59. "AXES: Cleaned up %s expired access attempts from database that were older than %s",
  60. count,
  61. threshold,
  62. )
  63. return count
  64. def reset_user_attempts(request, credentials: dict = None) -> int:
  65. """
  66. Reset all user attempts that match the given request and credentials.
  67. """
  68. attempts_list = filter_user_attempts(request, credentials)
  69. count = 0
  70. for attempts in attempts_list:
  71. _count, _ = attempts.delete()
  72. count += _count
  73. log.info("AXES: Reset %s access attempts from database.", count)
  74. return count