signals.py 6.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184
  1. from __future__ import unicode_literals
  2. import logging
  3. from django.contrib.auth.signals import user_logged_in
  4. from django.contrib.auth.signals import user_logged_out
  5. from django.contrib.auth.signals import user_login_failed
  6. from django.db.models.signals import post_save, post_delete
  7. from django.dispatch import receiver
  8. from django.dispatch import Signal
  9. from django.utils import timezone
  10. from axes.conf import settings
  11. from axes.attempts import get_cache_key
  12. from axes.attempts import get_cache_timeout
  13. from axes.attempts import get_user_attempts
  14. from axes.attempts import is_user_lockable
  15. from axes.attempts import ip_in_whitelist
  16. from axes.attempts import reset_user_attempts
  17. from axes.models import AccessLog, AccessAttempt
  18. from axes.utils import get_client_str
  19. from axes.utils import query2str
  20. from axes.utils import get_axes_cache, get_client_ip, get_client_username, get_credentials
  21. log = logging.getLogger(settings.AXES_LOGGER)
  22. user_locked_out = Signal(providing_args=['request', 'username', 'ip_address'])
  23. @receiver(user_login_failed)
  24. def log_user_login_failed(sender, credentials, request, **kwargs): # pylint: disable=unused-argument
  25. """ Create an AccessAttempt record if the login wasn't successful
  26. """
  27. if request is None:
  28. log.warning('Attempt to authenticate with a custom backend failed.')
  29. return
  30. ip_address = get_client_ip(request)
  31. username = get_client_username(request, credentials)
  32. user_agent = request.META.get('HTTP_USER_AGENT', '<unknown>')[:255]
  33. path_info = request.META.get('PATH_INFO', '<unknown>')[:255]
  34. http_accept = request.META.get('HTTP_ACCEPT', '<unknown>')[:1025]
  35. if settings.AXES_NEVER_LOCKOUT_WHITELIST and ip_in_whitelist(ip_address):
  36. return
  37. failures = 0
  38. attempts = get_user_attempts(request, credentials)
  39. cache_hash_key = get_cache_key(request, credentials)
  40. cache_timeout = get_cache_timeout()
  41. failures_cached = get_axes_cache().get(cache_hash_key)
  42. if failures_cached is not None:
  43. failures = failures_cached
  44. else:
  45. for attempt in attempts:
  46. failures = max(failures, attempt.failures_since_start)
  47. # add a failed attempt for this user
  48. failures += 1
  49. get_axes_cache().set(cache_hash_key, failures, cache_timeout)
  50. # has already attempted, update the info
  51. if attempts:
  52. for attempt in attempts:
  53. attempt.get_data = '%s\n---------\n%s' % (
  54. attempt.get_data,
  55. query2str(request.GET),
  56. )
  57. attempt.post_data = '%s\n---------\n%s' % (
  58. attempt.post_data,
  59. query2str(request.POST)
  60. )
  61. attempt.http_accept = http_accept
  62. attempt.path_info = path_info
  63. attempt.failures_since_start = failures
  64. attempt.attempt_time = timezone.now()
  65. attempt.save()
  66. log.info(
  67. 'AXES: Repeated login failure by %s. Count = %d of %d',
  68. get_client_str(username, ip_address, user_agent, path_info),
  69. failures,
  70. settings.AXES_FAILURE_LIMIT
  71. )
  72. else:
  73. # Record failed attempt. Whether or not the IP address or user agent is
  74. # used in counting failures is handled elsewhere, so we just record
  75. # everything here.
  76. AccessAttempt.objects.create(
  77. user_agent=user_agent,
  78. ip_address=ip_address,
  79. username=username,
  80. get_data=query2str(request.GET),
  81. post_data=query2str(request.POST),
  82. http_accept=http_accept,
  83. path_info=path_info,
  84. failures_since_start=failures,
  85. )
  86. log.info(
  87. 'AXES: New login failure by %s. Creating access record.',
  88. get_client_str(username, ip_address, user_agent, path_info)
  89. )
  90. # no matter what, we want to lock them out if they're past the number of
  91. # attempts allowed, unless the user is set to notlockable
  92. if (
  93. failures >= settings.AXES_FAILURE_LIMIT and
  94. settings.AXES_LOCK_OUT_AT_FAILURE and
  95. is_user_lockable(request, credentials)
  96. ):
  97. log.warning(
  98. 'AXES: locked out %s after repeated login attempts.',
  99. get_client_str(username, ip_address, user_agent, path_info)
  100. )
  101. # send signal when someone is locked out.
  102. user_locked_out.send(
  103. 'axes', request=request, username=username, ip_address=ip_address
  104. )
  105. @receiver(user_logged_in)
  106. def log_user_logged_in(sender, request, user, **kwargs): # pylint: disable=unused-argument
  107. """ When a user logs in, update the access log
  108. """
  109. username = user.get_username()
  110. credentials = get_credentials(username)
  111. ip_address = get_client_ip(request)
  112. user_agent = request.META.get('HTTP_USER_AGENT', '<unknown>')[:255]
  113. path_info = request.META.get('PATH_INFO', '<unknown>')[:255]
  114. http_accept = request.META.get('HTTP_ACCEPT', '<unknown>')[:1025]
  115. log.info(
  116. 'AXES: Successful login by %s.',
  117. get_client_str(username, ip_address, user_agent, path_info)
  118. )
  119. if not settings.AXES_DISABLE_SUCCESS_ACCESS_LOG:
  120. AccessLog.objects.create(
  121. user_agent=user_agent,
  122. ip_address=ip_address,
  123. username=username,
  124. http_accept=http_accept,
  125. path_info=path_info,
  126. trusted=True,
  127. )
  128. if settings.AXES_RESET_ON_SUCCESS:
  129. count = reset_user_attempts(request, credentials)
  130. log.info(
  131. 'AXES: Deleted %d failed login attempts by %s.',
  132. count,
  133. get_client_str(username, ip_address, user_agent, path_info)
  134. )
  135. @receiver(user_logged_out)
  136. def log_user_logged_out(sender, request, user, **kwargs): # pylint: disable=unused-argument
  137. """ When a user logs out, update the access log
  138. """
  139. log.info('AXES: Successful logout by %s.', user)
  140. if user and not settings.AXES_DISABLE_ACCESS_LOG:
  141. AccessLog.objects.filter(
  142. username=user.get_username(),
  143. logout_time__isnull=True,
  144. ).update(logout_time=timezone.now())
  145. @receiver(post_save, sender=AccessAttempt)
  146. def update_cache_after_save(instance, **kwargs): # pylint: disable=unused-argument
  147. cache_hash_key = get_cache_key(instance)
  148. if not get_axes_cache().get(cache_hash_key):
  149. cache_timeout = get_cache_timeout()
  150. get_axes_cache().set(cache_hash_key, instance.failures_since_start, cache_timeout)
  151. @receiver(post_delete, sender=AccessAttempt)
  152. def delete_cache_after_delete(instance, **kwargs): # pylint: disable=unused-argument
  153. cache_hash_key = get_cache_key(instance)
  154. get_axes_cache().delete(cache_hash_key)