attempts.py 6.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218
  1. from __future__ import unicode_literals
  2. from datetime import timedelta
  3. from hashlib import md5
  4. from django.contrib.auth import get_user_model
  5. from django.utils import timezone
  6. from axes.conf import settings
  7. from axes.models import AccessAttempt
  8. from axes.utils import get_axes_cache, get_client_ip, get_client_username
  9. def _query_user_attempts(request, credentials=None):
  10. """Returns access attempt record if it exists.
  11. Otherwise return None.
  12. """
  13. ip = get_client_ip(request)
  14. username = get_client_username(request, credentials)
  15. if settings.AXES_ONLY_USER_FAILURES:
  16. attempts = AccessAttempt.objects.filter(username=username)
  17. elif settings.AXES_USE_USER_AGENT:
  18. ua = request.META.get('HTTP_USER_AGENT', '<unknown>')[:255]
  19. attempts = AccessAttempt.objects.filter(
  20. user_agent=ua, ip_address=ip, username=username, trusted=True
  21. )
  22. else:
  23. attempts = AccessAttempt.objects.filter(
  24. ip_address=ip, username=username, trusted=True
  25. )
  26. if not attempts:
  27. params = {'trusted': False}
  28. if settings.AXES_ONLY_USER_FAILURES:
  29. params['username'] = username
  30. elif settings.AXES_LOCK_OUT_BY_COMBINATION_USER_AND_IP:
  31. params['username'] = username
  32. params['ip_address'] = ip
  33. else:
  34. params['ip_address'] = ip
  35. if settings.AXES_USE_USER_AGENT and not settings.AXES_ONLY_USER_FAILURES:
  36. params['user_agent'] = ua
  37. attempts = AccessAttempt.objects.filter(**params)
  38. return attempts
  39. def get_cache_key(request_or_obj, credentials=None):
  40. """
  41. Build cache key name from request or AccessAttempt object.
  42. :param request_or_obj: Request or AccessAttempt object
  43. :return cache-key: String, key to be used in cache system
  44. """
  45. if isinstance(request_or_obj, AccessAttempt):
  46. ip = request_or_obj.ip_address
  47. un = request_or_obj.username
  48. ua = request_or_obj.user_agent
  49. else:
  50. ip = get_client_ip(request_or_obj)
  51. un = get_client_username(request_or_obj, credentials)
  52. ua = request_or_obj.META.get('HTTP_USER_AGENT', '<unknown>')[:255]
  53. ip = ip.encode('utf-8') if ip else ''.encode('utf-8')
  54. un = un.encode('utf-8') if un else ''.encode('utf-8')
  55. ua = ua.encode('utf-8') if ua else ''.encode('utf-8')
  56. if settings.AXES_ONLY_USER_FAILURES:
  57. attributes = un
  58. elif settings.AXES_LOCK_OUT_BY_COMBINATION_USER_AND_IP:
  59. attributes = ip + un
  60. else:
  61. attributes = ip
  62. if settings.AXES_USE_USER_AGENT and not settings.AXES_ONLY_USER_FAILURES:
  63. attributes += ua
  64. cache_hash_key = 'axes-{}'.format(md5(attributes).hexdigest())
  65. return cache_hash_key
  66. def get_cache_timeout():
  67. """Returns timeout according to COOLOFF_TIME."""
  68. cache_timeout = None
  69. cool_off = settings.AXES_COOLOFF_TIME
  70. if cool_off:
  71. if isinstance(cool_off, (int, float)):
  72. cache_timeout = timedelta(hours=cool_off).total_seconds()
  73. else:
  74. cache_timeout = cool_off.total_seconds()
  75. return cache_timeout
  76. def get_user_attempts(request, credentials=None):
  77. force_reload = False
  78. attempts = _query_user_attempts(request, credentials)
  79. cache_hash_key = get_cache_key(request, credentials)
  80. cache_timeout = get_cache_timeout()
  81. cool_off = settings.AXES_COOLOFF_TIME
  82. if cool_off:
  83. if isinstance(cool_off, (int, float)):
  84. cool_off = timedelta(hours=cool_off)
  85. for attempt in attempts:
  86. if attempt.attempt_time + cool_off < timezone.now():
  87. if attempt.trusted:
  88. attempt.failures_since_start = 0
  89. attempt.save()
  90. get_axes_cache().set(cache_hash_key, 0, cache_timeout)
  91. else:
  92. attempt.delete()
  93. force_reload = True
  94. failures_cached = get_axes_cache().get(cache_hash_key)
  95. if failures_cached is not None:
  96. get_axes_cache().set(
  97. cache_hash_key, failures_cached - 1, cache_timeout
  98. )
  99. # If objects were deleted, we need to update the queryset to reflect this,
  100. # so force a reload.
  101. if force_reload:
  102. attempts = _query_user_attempts(request, credentials)
  103. return attempts
  104. def reset_user_attempts(request, credentials=None):
  105. attempts = _query_user_attempts(request, credentials)
  106. count, _ = attempts.delete()
  107. return count
  108. def ip_in_whitelist(ip):
  109. if not settings.AXES_IP_WHITELIST:
  110. return False
  111. return ip in settings.AXES_IP_WHITELIST
  112. def ip_in_blacklist(ip):
  113. if not settings.AXES_IP_BLACKLIST:
  114. return False
  115. return ip in settings.AXES_IP_BLACKLIST
  116. def is_user_lockable(request, credentials=None):
  117. """Check if the user has a profile with nolockout
  118. If so, then return the value to see if this user is special
  119. and doesn't get their account locked out
  120. """
  121. if request.method != 'POST':
  122. return True
  123. try:
  124. field = getattr(get_user_model(), 'USERNAME_FIELD', 'username')
  125. kwargs = {
  126. field: get_client_username(request, credentials)
  127. }
  128. user = get_user_model().objects.get(**kwargs)
  129. if hasattr(user, 'nolockout'):
  130. # need to invert since we need to return
  131. # false for users that can't be blocked
  132. return not user.nolockout
  133. except get_user_model().DoesNotExist:
  134. # not a valid user
  135. return True
  136. # Default behavior for a user to be lockable
  137. return True
  138. def is_already_locked(request, credentials=None):
  139. ip = get_client_ip(request)
  140. if (
  141. settings.AXES_ONLY_USER_FAILURES or
  142. settings.AXES_LOCK_OUT_BY_COMBINATION_USER_AND_IP
  143. ) and request.method == 'GET':
  144. return False
  145. if settings.AXES_NEVER_LOCKOUT_WHITELIST and ip_in_whitelist(ip):
  146. return False
  147. if settings.AXES_ONLY_WHITELIST and not ip_in_whitelist(ip):
  148. return True
  149. if ip_in_blacklist(ip):
  150. return True
  151. if not is_user_lockable(request, credentials):
  152. return False
  153. cache_hash_key = get_cache_key(request, credentials)
  154. failures_cached = get_axes_cache().get(cache_hash_key)
  155. if failures_cached is not None:
  156. return (
  157. failures_cached >= settings.AXES_FAILURE_LIMIT and
  158. settings.AXES_LOCK_OUT_AT_FAILURE
  159. )
  160. for attempt in get_user_attempts(request, credentials):
  161. if (
  162. attempt.failures_since_start >= settings.AXES_FAILURE_LIMIT and
  163. settings.AXES_LOCK_OUT_AT_FAILURE
  164. ):
  165. return True
  166. return False