signals.py 1.7 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061
  1. from logging import getLogger
  2. from django.contrib.auth.signals import (
  3. user_logged_in,
  4. user_logged_out,
  5. user_login_failed,
  6. )
  7. from django.core.signals import setting_changed
  8. from django.db.models.signals import post_save, post_delete
  9. from django.dispatch import Signal
  10. from django.dispatch import receiver
  11. from axes.handlers.proxy import AxesProxyHandler
  12. from axes.models import AccessAttempt
  13. log = getLogger(__name__)
  14. # This signal provides the following arguments to any listeners:
  15. # request - The current Request object.
  16. # username - The username of the User who has been locked out.
  17. # ip_address - The IP of the user who has been locked out.
  18. user_locked_out = Signal()
  19. @receiver(user_login_failed)
  20. def handle_user_login_failed(*args, **kwargs):
  21. AxesProxyHandler.user_login_failed(*args, **kwargs)
  22. @receiver(user_logged_in)
  23. def handle_user_logged_in(*args, **kwargs):
  24. AxesProxyHandler.user_logged_in(*args, **kwargs)
  25. @receiver(user_logged_out)
  26. def handle_user_logged_out(*args, **kwargs):
  27. AxesProxyHandler.user_logged_out(*args, **kwargs)
  28. @receiver(post_save, sender=AccessAttempt)
  29. def handle_post_save_access_attempt(*args, **kwargs):
  30. AxesProxyHandler.post_save_access_attempt(*args, **kwargs)
  31. @receiver(post_delete, sender=AccessAttempt)
  32. def handle_post_delete_access_attempt(*args, **kwargs):
  33. AxesProxyHandler.post_delete_access_attempt(*args, **kwargs)
  34. @receiver(setting_changed)
  35. def handle_setting_changed(
  36. sender, setting, value, enter, **kwargs
  37. ): # pylint: disable=unused-argument
  38. """
  39. Reinitialize handler implementation if a relevant setting changes
  40. in e.g. application reconfiguration or during testing.
  41. """
  42. if setting == "AXES_HANDLER":
  43. AxesProxyHandler.get_implementation(force=True)