utils.py 5.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167
  1. from __future__ import unicode_literals
  2. try:
  3. import win_inet_pton # pylint: disable=unused-import
  4. except ImportError:
  5. pass
  6. from inspect import getargspec
  7. from logging import getLogger
  8. from socket import error, inet_pton, AF_INET6
  9. from django.core.cache import caches
  10. from django.utils import six
  11. import ipware.ip2
  12. from axes.conf import settings
  13. from axes.models import AccessAttempt
  14. logger = getLogger(__name__)
  15. def get_axes_cache():
  16. return caches[getattr(settings, 'AXES_CACHE', 'default')]
  17. def query2str(items, max_length=1024):
  18. """Turns a dictionary into an easy-to-read list of key-value pairs.
  19. If there's a field called "password" it will be excluded from the output.
  20. The length of the output is limited to max_length to avoid a DoS attack
  21. via excessively large payloads.
  22. """
  23. return '\n'.join([
  24. '%s=%s' % (k, v) for k, v in six.iteritems(items)
  25. if k != settings.AXES_PASSWORD_FORM_FIELD
  26. ][:int(max_length / 2)])[:max_length]
  27. def get_client_str(username, ip_address, user_agent=None, path_info=None):
  28. if settings.AXES_VERBOSE:
  29. if isinstance(path_info, tuple):
  30. path_info = path_info[0]
  31. details = "{{user: '{0}', ip: '{1}', user-agent: '{2}', path: '{3}'}}"
  32. return details.format(username, ip_address, user_agent, path_info)
  33. if settings.AXES_ONLY_USER_FAILURES:
  34. client = username
  35. elif settings.AXES_LOCK_OUT_BY_COMBINATION_USER_AND_IP:
  36. client = '{0} from {1}'.format(username, ip_address)
  37. else:
  38. client = ip_address
  39. if settings.AXES_USE_USER_AGENT:
  40. client += '(user-agent={0})'.format(user_agent)
  41. return client
  42. def get_client_ip(request):
  43. client_ip_attribute = 'axes_client_ip'
  44. if not hasattr(request, client_ip_attribute):
  45. client_ip, _ = ipware.ip2.get_client_ip(
  46. request,
  47. proxy_order=settings.AXES_PROXY_ORDER,
  48. proxy_count=settings.AXES_PROXY_COUNT,
  49. proxy_trusted_ips=settings.AXES_PROXY_TRUSTED_IPS,
  50. request_header_order=settings.AXES_META_PRECEDENCE_ORDER,
  51. )
  52. setattr(request, client_ip_attribute, client_ip)
  53. return getattr(request, client_ip_attribute)
  54. def get_client_username(request, credentials=None):
  55. """Resolve client username from the given request or credentials if supplied
  56. The order of preference for fetching the username is as follows:
  57. 1. If configured, use `AXES_USERNAME_CALLABLE`, and supply either `request` or `request, credentials` as arguments
  58. depending on the function argument count (multiple signatures are supported for backwards compatibility)
  59. 2. If given, use `credentials` and fetch username from `AXES_USERNAME_FORM_FIELD` (defaults to `username`)
  60. 3. Use request.POST and fetch username from `AXES_USERNAME_FORM_FIELD` (defaults to `username`)
  61. :param request: incoming Django `HttpRequest` or similar object from authentication backend or other source
  62. :param credentials: incoming credentials `dict` or similar object from authentication backend or other source
  63. """
  64. if settings.AXES_USERNAME_CALLABLE:
  65. num_args = len(
  66. getargspec(settings.AXES_USERNAME_CALLABLE).args # pylint: disable=deprecated-method
  67. )
  68. if num_args == 2:
  69. logger.debug('Using AXES_USERNAME_CALLABLE for username with two arguments: request, credentials')
  70. return settings.AXES_USERNAME_CALLABLE(request, credentials)
  71. if num_args == 1:
  72. logger.debug('Using AXES_USERNAME_CALLABLE for username with one argument: request')
  73. return settings.AXES_USERNAME_CALLABLE(request)
  74. logger.error('Using AXES_USERNAME_CALLABLE for username failed: wrong number of arguments %s', num_args)
  75. raise TypeError('Wrong number of arguments in function call to AXES_USERNAME_CALLABLE', num_args)
  76. if credentials:
  77. logger.debug('Using `credentials` to get username with key AXES_USERNAME_FORM_FIELD')
  78. return credentials.get(settings.AXES_USERNAME_FORM_FIELD, None)
  79. logger.debug('Using `request.POST` to get username with key AXES_USERNAME_FORM_FIELD')
  80. return request.POST.get(settings.AXES_USERNAME_FORM_FIELD, None)
  81. def get_credentials(username=None, **kwargs):
  82. credentials = {settings.AXES_USERNAME_FORM_FIELD: username}
  83. credentials.update(kwargs)
  84. return credentials
  85. def is_ipv6(ip):
  86. try:
  87. inet_pton(AF_INET6, ip)
  88. except (OSError, error):
  89. return False
  90. return True
  91. def reset(ip=None, username=None):
  92. """Reset records that match ip or username, and
  93. return the count of removed attempts.
  94. """
  95. attempts = AccessAttempt.objects.all()
  96. if ip:
  97. attempts = attempts.filter(ip_address=ip)
  98. if username:
  99. attempts = attempts.filter(username=username)
  100. count, _ = attempts.delete()
  101. return count
  102. def iso8601(timestamp):
  103. """Returns datetime.timedelta translated to ISO 8601 formatted duration.
  104. """
  105. seconds = timestamp.total_seconds()
  106. minutes, seconds = divmod(seconds, 60)
  107. hours, minutes = divmod(minutes, 60)
  108. days, hours = divmod(hours, 24)
  109. date = '{:.0f}D'.format(days) if days else ''
  110. time_values = hours, minutes, seconds
  111. time_designators = 'H', 'M', 'S'
  112. time = ''.join([
  113. ('{:.0f}'.format(value) + designator)
  114. for value, designator in zip(time_values, time_designators)
  115. if value]
  116. )
  117. return 'P' + date + ('T' + time if time else '')
  118. def get_lockout_message():
  119. if settings.AXES_COOLOFF_TIME:
  120. return settings.AXES_COOLOFF_MESSAGE
  121. return settings.AXES_PERMALOCK_MESSAGE