helpers.py 16 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478
  1. from datetime import timedelta
  2. from hashlib import md5
  3. from logging import getLogger
  4. from string import Template
  5. from typing import Callable, Optional, Type, Union
  6. from urllib.parse import urlencode
  7. import ipware.ip
  8. from django.core.cache import caches, BaseCache
  9. from django.http import HttpRequest, HttpResponse, JsonResponse, QueryDict
  10. from django.shortcuts import render, redirect
  11. from django.utils.module_loading import import_string
  12. from axes.conf import settings
  13. from axes.models import AccessBase
  14. log = getLogger(__name__)
  15. def get_cache() -> BaseCache:
  16. """
  17. Get the cache instance Axes is configured to use with ``settings.AXES_CACHE`` and use ``'default'`` if not set.
  18. """
  19. return caches[getattr(settings, "AXES_CACHE", "default")]
  20. def get_cache_timeout() -> Optional[int]:
  21. """
  22. Return the cache timeout interpreted from settings.AXES_COOLOFF_TIME.
  23. The cache timeout can be either None if not configured or integer of seconds if configured.
  24. Notice that the settings.AXES_COOLOFF_TIME can be None, timedelta, integer, callable, or str path,
  25. and this function offers a unified _integer or None_ representation of that configuration
  26. for use with the Django cache backends.
  27. """
  28. cool_off = get_cool_off()
  29. if cool_off is None:
  30. return None
  31. return int(cool_off.total_seconds())
  32. def get_cool_off() -> Optional[timedelta]:
  33. """
  34. Return the login cool off time interpreted from settings.AXES_COOLOFF_TIME.
  35. The return value is either None or timedelta.
  36. Notice that the settings.AXES_COOLOFF_TIME is either None, timedelta, or integer of hours,
  37. and this function offers a unified _timedelta or None_ representation of that configuration
  38. for use with the Axes internal implementations.
  39. :exception TypeError: if settings.AXES_COOLOFF_TIME is of wrong type.
  40. """
  41. cool_off = settings.AXES_COOLOFF_TIME
  42. if isinstance(cool_off, int):
  43. return timedelta(hours=cool_off)
  44. if isinstance(cool_off, str):
  45. return import_string(cool_off)()
  46. if callable(cool_off):
  47. return cool_off()
  48. return cool_off
  49. def get_cool_off_iso8601(delta: timedelta) -> str:
  50. """
  51. Return datetime.timedelta translated to ISO 8601 formatted duration for use in e.g. cool offs.
  52. """
  53. seconds = delta.total_seconds()
  54. minutes, seconds = divmod(seconds, 60)
  55. hours, minutes = divmod(minutes, 60)
  56. days, hours = divmod(hours, 24)
  57. days_str = f"{days:.0f}D" if days else ""
  58. time_str = "".join(
  59. f"{value:.0f}{designator}"
  60. for value, designator in [[hours, "H"], [minutes, "M"], [seconds, "S"]]
  61. if value
  62. )
  63. if time_str:
  64. return f"P{days_str}T{time_str}"
  65. return f"P{days_str}"
  66. def get_credentials(username: str = None, **kwargs) -> dict:
  67. """
  68. Calculate credentials for Axes to use internally from given username and kwargs.
  69. Axes will set the username value into the key defined with ``settings.AXES_USERNAME_FORM_FIELD``
  70. and update the credentials dictionary with the kwargs given on top of that.
  71. """
  72. credentials = {settings.AXES_USERNAME_FORM_FIELD: username}
  73. credentials.update(kwargs)
  74. return credentials
  75. def get_client_username(request, credentials: dict = None) -> str:
  76. """
  77. Resolve client username from the given request or credentials if supplied.
  78. The order of preference for fetching the username is as follows:
  79. 1. If configured, use ``AXES_USERNAME_CALLABLE``, and supply ``request, credentials`` as arguments
  80. 2. If given, use ``credentials`` and fetch username from ``AXES_USERNAME_FORM_FIELD`` (defaults to ``username``)
  81. 3. Use request.POST and fetch username from ``AXES_USERNAME_FORM_FIELD`` (defaults to ``username``)
  82. :param request: incoming Django ``HttpRequest`` or similar object from authentication backend or other source
  83. :param credentials: incoming credentials ``dict`` or similar object from authentication backend or other source
  84. """
  85. if settings.AXES_USERNAME_CALLABLE:
  86. log.debug("Using settings.AXES_USERNAME_CALLABLE to get username")
  87. if callable(settings.AXES_USERNAME_CALLABLE):
  88. return settings.AXES_USERNAME_CALLABLE(request, credentials)
  89. if isinstance(settings.AXES_USERNAME_CALLABLE, str):
  90. return import_string(settings.AXES_USERNAME_CALLABLE)(request, credentials)
  91. raise TypeError(
  92. "settings.AXES_USERNAME_CALLABLE needs to be a string, callable, or None."
  93. )
  94. if credentials:
  95. log.debug(
  96. "Using parameter credentials to get username with key settings.AXES_USERNAME_FORM_FIELD"
  97. )
  98. return credentials.get(settings.AXES_USERNAME_FORM_FIELD, None)
  99. log.debug(
  100. "Using parameter request.POST to get username with key settings.AXES_USERNAME_FORM_FIELD"
  101. )
  102. request_data = getattr(request, "data", request.POST)
  103. return request_data.get(settings.AXES_USERNAME_FORM_FIELD, None)
  104. def get_client_ip_address(request) -> str:
  105. """
  106. Get client IP address as configured by the user.
  107. The django-ipware package is used for address resolution
  108. and parameters can be configured in the Axes package.
  109. """
  110. client_ip_address, _ = ipware.ip.get_client_ip(
  111. request,
  112. proxy_order=settings.AXES_PROXY_ORDER,
  113. proxy_count=settings.AXES_PROXY_COUNT,
  114. proxy_trusted_ips=settings.AXES_PROXY_TRUSTED_IPS,
  115. request_header_order=settings.AXES_META_PRECEDENCE_ORDER,
  116. )
  117. return client_ip_address
  118. def get_client_user_agent(request) -> str:
  119. return request.META.get("HTTP_USER_AGENT", "<unknown>")[:255]
  120. def get_client_path_info(request) -> str:
  121. return request.META.get("PATH_INFO", "<unknown>")[:255]
  122. def get_client_http_accept(request) -> str:
  123. return request.META.get("HTTP_ACCEPT", "<unknown>")[:1025]
  124. def get_client_parameters(username: str, ip_address: str, user_agent: str) -> list:
  125. """
  126. Get query parameters for filtering AccessAttempt queryset.
  127. This method returns a dict that guarantees iteration order for keys and values,
  128. and can so be used in e.g. the generation of hash keys or other deterministic functions.
  129. Returns list of dict, every item of list are separate parameters
  130. """
  131. if settings.AXES_ONLY_USER_FAILURES:
  132. # 1. Only individual usernames can be tracked with parametrization
  133. filter_query = [{"username": username}]
  134. else:
  135. if settings.AXES_LOCK_OUT_BY_USER_OR_IP:
  136. # One of `username` or `IP address` is used
  137. filter_query = [{"username": username}, {"ip_address": ip_address}]
  138. elif settings.AXES_LOCK_OUT_BY_COMBINATION_USER_AND_IP:
  139. # 2. A combination of username and IP address can be used as well
  140. filter_query = [{"username": username, "ip_address": ip_address}]
  141. else:
  142. # 3. Default case is to track the IP address only, which is the most secure option
  143. filter_query = [{"ip_address": ip_address}]
  144. if settings.AXES_USE_USER_AGENT:
  145. # 4. The HTTP User-Agent can be used to track e.g. one browser
  146. filter_query.append({"user_agent": user_agent})
  147. return filter_query
  148. def make_cache_key_list(filter_kwargs_list):
  149. cache_keys = []
  150. for filter_kwargs in filter_kwargs_list:
  151. cache_key_components = "".join(
  152. value for value in filter_kwargs.values() if value
  153. )
  154. cache_key_digest = md5(cache_key_components.encode()).hexdigest()
  155. cache_keys.append(f"axes-{cache_key_digest}")
  156. return cache_keys
  157. def get_client_cache_key(
  158. request_or_attempt: Union[HttpRequest, AccessBase], credentials: dict = None
  159. ) -> str:
  160. """
  161. Build cache key name from request or AccessAttempt object.
  162. :param request_or_attempt: HttpRequest or AccessAttempt object
  163. :param credentials: credentials containing user information
  164. :return cache_key: Hash key that is usable for Django cache backends
  165. """
  166. if isinstance(request_or_attempt, AccessBase):
  167. username = request_or_attempt.username
  168. ip_address = request_or_attempt.ip_address
  169. user_agent = request_or_attempt.user_agent
  170. else:
  171. username = get_client_username(request_or_attempt, credentials)
  172. ip_address = get_client_ip_address(request_or_attempt)
  173. user_agent = get_client_user_agent(request_or_attempt)
  174. filter_kwargs_list = get_client_parameters(username, ip_address, user_agent)
  175. return make_cache_key_list(filter_kwargs_list)
  176. def get_client_str(
  177. username: str, ip_address: str, user_agent: str, path_info: str
  178. ) -> str:
  179. """
  180. Get a readable string that can be used in e.g. logging to distinguish client requests.
  181. Example log format would be
  182. ``{username: "example", ip_address: "127.0.0.1", path_info: "/example/"}``
  183. """
  184. client_dict = dict()
  185. if settings.AXES_VERBOSE:
  186. # Verbose mode logs every attribute that is available
  187. client_dict["username"] = username
  188. client_dict["ip_address"] = ip_address
  189. client_dict["user_agent"] = user_agent
  190. else:
  191. # Other modes initialize the attributes that are used for the actual lockouts
  192. client_list = get_client_parameters(username, ip_address, user_agent)
  193. client_dict = {}
  194. for client in client_list:
  195. client_dict.update(client)
  196. # Path info is always included as last component in the client string for traceability purposes
  197. if path_info and isinstance(path_info, (tuple, list)):
  198. path_info = path_info[0]
  199. client_dict["path_info"] = path_info
  200. # Template the internal dictionary representation into a readable and concatenated {key: "value"} format
  201. template = Template('$key: "$value"')
  202. items = [{"key": k, "value": v} for k, v in client_dict.items()]
  203. client_str = ", ".join(template.substitute(item) for item in items)
  204. client_str = "{" + client_str + "}"
  205. return client_str
  206. def get_query_str(query: Type[QueryDict], max_length: int = 1024) -> str:
  207. """
  208. Turns a query dictionary into an easy-to-read list of key-value pairs.
  209. If a field is called either ``'password'`` or ``settings.AXES_PASSWORD_FORM_FIELD`` it will be excluded.
  210. The length of the output is limited to max_length to avoid a DoS attack via excessively large payloads.
  211. """
  212. query_dict = query.copy()
  213. query_dict.pop("password", None)
  214. query_dict.pop(settings.AXES_PASSWORD_FORM_FIELD, None)
  215. template = Template("$key=$value")
  216. items = [{"key": k, "value": v} for k, v in query_dict.items()]
  217. query_str = "\n".join(template.substitute(item) for item in items)
  218. return query_str[:max_length]
  219. def get_failure_limit(request, credentials) -> int:
  220. if callable(settings.AXES_FAILURE_LIMIT):
  221. return settings.AXES_FAILURE_LIMIT(request, credentials)
  222. if isinstance(settings.AXES_FAILURE_LIMIT, str):
  223. return import_string(settings.AXES_FAILURE_LIMIT)(request, credentials)
  224. if isinstance(settings.AXES_FAILURE_LIMIT, int):
  225. return settings.AXES_FAILURE_LIMIT
  226. raise TypeError("settings.AXES_FAILURE_LIMIT needs to be a callable or an integer")
  227. def get_lockout_message() -> str:
  228. if settings.AXES_COOLOFF_TIME:
  229. return settings.AXES_COOLOFF_MESSAGE
  230. return settings.AXES_PERMALOCK_MESSAGE
  231. def get_lockout_response(request, credentials: dict = None) -> HttpResponse:
  232. if settings.AXES_LOCKOUT_CALLABLE:
  233. if callable(settings.AXES_LOCKOUT_CALLABLE):
  234. return settings.AXES_LOCKOUT_CALLABLE(request, credentials)
  235. if isinstance(settings.AXES_LOCKOUT_CALLABLE, str):
  236. return import_string(settings.AXES_LOCKOUT_CALLABLE)(request, credentials)
  237. raise TypeError(
  238. "settings.AXES_LOCKOUT_CALLABLE needs to be a string, callable, or None."
  239. )
  240. status = 403
  241. context = {
  242. "failure_limit": get_failure_limit(request, credentials),
  243. "username": get_client_username(request, credentials) or "",
  244. }
  245. cool_off = get_cool_off()
  246. if cool_off:
  247. context.update(
  248. {
  249. "cooloff_time": get_cool_off_iso8601(
  250. cool_off
  251. ), # differing old name is kept for backwards compatibility
  252. "cooloff_timedelta": cool_off,
  253. }
  254. )
  255. if request.META.get("HTTP_X_REQUESTED_WITH") == "XMLHttpRequest":
  256. json_response = JsonResponse(context, status=status)
  257. json_response[
  258. "Access-Control-Allow-Origin"
  259. ] = settings.AXES_ALLOWED_CORS_ORIGINS
  260. json_response["Access-Control-Allow-Methods"] = "POST, OPTIONS"
  261. json_response[
  262. "Access-Control-Allow-Headers"
  263. ] = "Origin, Content-Type, Accept, Authorization, x-requested-with"
  264. return json_response
  265. if settings.AXES_LOCKOUT_TEMPLATE:
  266. return render(request, settings.AXES_LOCKOUT_TEMPLATE, context, status=status)
  267. if settings.AXES_LOCKOUT_URL:
  268. lockout_url = settings.AXES_LOCKOUT_URL
  269. query_string = urlencode({"username": context["username"]})
  270. url = "{}?{}".format(lockout_url, query_string)
  271. return redirect(url)
  272. return HttpResponse(get_lockout_message(), status=status)
  273. def is_ip_address_in_whitelist(ip_address: str) -> bool:
  274. if not settings.AXES_IP_WHITELIST:
  275. return False
  276. return ip_address in settings.AXES_IP_WHITELIST
  277. def is_ip_address_in_blacklist(ip_address: str) -> bool:
  278. if not settings.AXES_IP_BLACKLIST:
  279. return False
  280. return ip_address in settings.AXES_IP_BLACKLIST
  281. def is_client_ip_address_whitelisted(request):
  282. """
  283. Check if the given request refers to a whitelisted IP.
  284. """
  285. if settings.AXES_NEVER_LOCKOUT_WHITELIST and is_ip_address_in_whitelist(
  286. request.axes_ip_address
  287. ):
  288. return True
  289. if settings.AXES_ONLY_WHITELIST and is_ip_address_in_whitelist(
  290. request.axes_ip_address
  291. ):
  292. return True
  293. return False
  294. def is_client_ip_address_blacklisted(request) -> bool:
  295. """
  296. Check if the given request refers to a blacklisted IP.
  297. """
  298. if is_ip_address_in_blacklist(request.axes_ip_address):
  299. return True
  300. if settings.AXES_ONLY_WHITELIST and not is_ip_address_in_whitelist(
  301. request.axes_ip_address
  302. ):
  303. return True
  304. return False
  305. def is_client_method_whitelisted(request) -> bool:
  306. """
  307. Check if the given request uses a whitelisted method.
  308. """
  309. if settings.AXES_NEVER_LOCKOUT_GET and request.method == "GET":
  310. return True
  311. return False
  312. def is_user_attempt_whitelisted(request, credentials: dict = None) -> bool:
  313. """
  314. Check if the given request or credentials refer to a whitelisted username.
  315. This method invokes the ``settings.AXES_WHITELIST`` callable
  316. with ``request`` and ``credentials`` arguments.
  317. This function could use the following implementation for checking
  318. the lockout flags from a specific property in the user object:
  319. .. code-block: python
  320. username_value = get_client_username(request, credentials)
  321. username_field = getattr(
  322. get_user_model(),
  323. "USERNAME_FIELD",
  324. "username"
  325. )
  326. kwargs = {username_field: username_value}
  327. user_model = get_user_model()
  328. user = user_model.objects.get(**kwargs)
  329. return user.nolockout
  330. """
  331. whitelist_callable = settings.AXES_WHITELIST_CALLABLE
  332. if whitelist_callable is None:
  333. return False
  334. if callable(whitelist_callable):
  335. return whitelist_callable(request, credentials)
  336. if isinstance(whitelist_callable, str):
  337. return import_string(whitelist_callable)(request, credentials)
  338. raise TypeError(
  339. "settings.AXES_WHITELIST_CALLABLE needs to be a string, callable, or None."
  340. )
  341. def toggleable(func) -> Callable:
  342. """
  343. Decorator that toggles function execution based on settings.
  344. If the ``settings.AXES_ENABLED`` flag is set to ``False``
  345. the decorated function never runs and a None is returned.
  346. This decorator is only suitable for functions that do not
  347. require return values to be passed back to callers.
  348. """
  349. def inner(*args, **kwargs): # pylint: disable=inconsistent-return-statements
  350. if settings.AXES_ENABLED:
  351. return func(*args, **kwargs)
  352. return inner