226 lines
7.8 KiB
Python
226 lines
7.8 KiB
Python
import inspect
|
|
import re
|
|
|
|
from django.apps import apps as django_apps
|
|
from django.conf import settings
|
|
from django.core.exceptions import ImproperlyConfigured, PermissionDenied
|
|
from django.middleware.csrf import rotate_token
|
|
from django.utils.crypto import constant_time_compare
|
|
from django.utils.module_loading import import_string
|
|
from django.utils.translation import LANGUAGE_SESSION_KEY
|
|
|
|
from .signals import user_logged_in, user_logged_out, user_login_failed
|
|
|
|
SESSION_KEY = '_auth_user_id'
|
|
BACKEND_SESSION_KEY = '_auth_user_backend'
|
|
HASH_SESSION_KEY = '_auth_user_hash'
|
|
REDIRECT_FIELD_NAME = 'next'
|
|
|
|
|
|
def load_backend(path):
|
|
return import_string(path)()
|
|
|
|
|
|
def _get_backends(return_tuples=False):
|
|
backends = []
|
|
for backend_path in settings.AUTHENTICATION_BACKENDS:
|
|
backend = load_backend(backend_path)
|
|
backends.append((backend, backend_path) if return_tuples else backend)
|
|
if not backends:
|
|
raise ImproperlyConfigured(
|
|
'No authentication backends have been defined. Does '
|
|
'AUTHENTICATION_BACKENDS contain anything?'
|
|
)
|
|
return backends
|
|
|
|
|
|
def get_backends():
|
|
return _get_backends(return_tuples=False)
|
|
|
|
|
|
def _clean_credentials(credentials):
|
|
"""
|
|
Clean a dictionary of credentials of potentially sensitive info before
|
|
sending to less secure functions.
|
|
|
|
Not comprehensive - intended for user_login_failed signal
|
|
"""
|
|
SENSITIVE_CREDENTIALS = re.compile('api|token|key|secret|password|signature', re.I)
|
|
CLEANSED_SUBSTITUTE = '********************'
|
|
for key in credentials:
|
|
if SENSITIVE_CREDENTIALS.search(key):
|
|
credentials[key] = CLEANSED_SUBSTITUTE
|
|
return credentials
|
|
|
|
|
|
def _get_user_session_key(request):
|
|
# This value in the session is always serialized to a string, so we need
|
|
# to convert it back to Python whenever we access it.
|
|
return get_user_model()._meta.pk.to_python(request.session[SESSION_KEY])
|
|
|
|
|
|
def authenticate(request=None, **credentials):
|
|
"""
|
|
If the given credentials are valid, return a User object.
|
|
"""
|
|
for backend, backend_path in _get_backends(return_tuples=True):
|
|
try:
|
|
inspect.getcallargs(backend.authenticate, request, **credentials)
|
|
except TypeError:
|
|
# This backend doesn't accept these credentials as arguments. Try the next one.
|
|
continue
|
|
try:
|
|
user = backend.authenticate(request, **credentials)
|
|
except PermissionDenied:
|
|
# This backend says to stop in our tracks - this user should not be allowed in at all.
|
|
break
|
|
if user is None:
|
|
continue
|
|
# Annotate the user object with the path of the backend.
|
|
user.backend = backend_path
|
|
return user
|
|
|
|
# The credentials supplied are invalid to all backends, fire signal
|
|
user_login_failed.send(sender=__name__, credentials=_clean_credentials(credentials), request=request)
|
|
|
|
|
|
def login(request, user, backend=None):
|
|
"""
|
|
Persist a user id and a backend in the request. This way a user doesn't
|
|
have to reauthenticate on every request. Note that data set during
|
|
the anonymous session is retained when the user logs in.
|
|
"""
|
|
session_auth_hash = ''
|
|
if user is None:
|
|
user = request.user
|
|
if hasattr(user, 'get_session_auth_hash'):
|
|
session_auth_hash = user.get_session_auth_hash()
|
|
|
|
if SESSION_KEY in request.session:
|
|
if _get_user_session_key(request) != user.pk or (
|
|
session_auth_hash and
|
|
not constant_time_compare(request.session.get(HASH_SESSION_KEY, ''), session_auth_hash)):
|
|
# To avoid reusing another user's session, create a new, empty
|
|
# session if the existing session corresponds to a different
|
|
# authenticated user.
|
|
request.session.flush()
|
|
else:
|
|
request.session.cycle_key()
|
|
|
|
try:
|
|
backend = backend or user.backend
|
|
except AttributeError:
|
|
backends = _get_backends(return_tuples=True)
|
|
if len(backends) == 1:
|
|
_, backend = backends[0]
|
|
else:
|
|
raise ValueError(
|
|
'You have multiple authentication backends configured and '
|
|
'therefore must provide the `backend` argument or set the '
|
|
'`backend` attribute on the user.'
|
|
)
|
|
else:
|
|
if not isinstance(backend, str):
|
|
raise TypeError('backend must be a dotted import path string (got %r).' % backend)
|
|
|
|
request.session[SESSION_KEY] = user._meta.pk.value_to_string(user)
|
|
request.session[BACKEND_SESSION_KEY] = backend
|
|
request.session[HASH_SESSION_KEY] = session_auth_hash
|
|
if hasattr(request, 'user'):
|
|
request.user = user
|
|
rotate_token(request)
|
|
user_logged_in.send(sender=user.__class__, request=request, user=user)
|
|
|
|
|
|
def logout(request):
|
|
"""
|
|
Remove the authenticated user's ID from the request and flush their session
|
|
data.
|
|
"""
|
|
# Dispatch the signal before the user is logged out so the receivers have a
|
|
# chance to find out *who* logged out.
|
|
user = getattr(request, 'user', None)
|
|
if not getattr(user, 'is_authenticated', True):
|
|
user = None
|
|
user_logged_out.send(sender=user.__class__, request=request, user=user)
|
|
|
|
# remember language choice saved to session
|
|
language = request.session.get(LANGUAGE_SESSION_KEY)
|
|
|
|
request.session.flush()
|
|
|
|
if language is not None:
|
|
request.session[LANGUAGE_SESSION_KEY] = language
|
|
|
|
if hasattr(request, 'user'):
|
|
from django.contrib.auth.models import AnonymousUser
|
|
request.user = AnonymousUser()
|
|
|
|
|
|
def get_user_model():
|
|
"""
|
|
Return the User model that is active in this project.
|
|
"""
|
|
try:
|
|
return django_apps.get_model(settings.AUTH_USER_MODEL, require_ready=False)
|
|
except ValueError:
|
|
raise ImproperlyConfigured("AUTH_USER_MODEL must be of the form 'app_label.model_name'")
|
|
except LookupError:
|
|
raise ImproperlyConfigured(
|
|
"AUTH_USER_MODEL refers to model '%s' that has not been installed" % settings.AUTH_USER_MODEL
|
|
)
|
|
|
|
|
|
def get_user(request):
|
|
"""
|
|
Return the user model instance associated with the given request session.
|
|
If no user is retrieved, return an instance of `AnonymousUser`.
|
|
"""
|
|
from .models import AnonymousUser
|
|
user = None
|
|
try:
|
|
user_id = _get_user_session_key(request)
|
|
backend_path = request.session[BACKEND_SESSION_KEY]
|
|
except KeyError:
|
|
pass
|
|
else:
|
|
if backend_path in settings.AUTHENTICATION_BACKENDS:
|
|
backend = load_backend(backend_path)
|
|
user = backend.get_user(user_id)
|
|
# Verify the session
|
|
if hasattr(user, 'get_session_auth_hash'):
|
|
session_hash = request.session.get(HASH_SESSION_KEY)
|
|
session_hash_verified = session_hash and constant_time_compare(
|
|
session_hash,
|
|
user.get_session_auth_hash()
|
|
)
|
|
if not session_hash_verified:
|
|
request.session.flush()
|
|
user = None
|
|
|
|
return user or AnonymousUser()
|
|
|
|
|
|
def get_permission_codename(action, opts):
|
|
"""
|
|
Return the codename of the permission for the specified action.
|
|
"""
|
|
return '%s_%s' % (action, opts.model_name)
|
|
|
|
|
|
def update_session_auth_hash(request, user):
|
|
"""
|
|
Updating a user's password logs out all sessions for the user.
|
|
|
|
Take the current request and the updated user object from which the new
|
|
session hash will be derived and update the session hash appropriately to
|
|
prevent a password change from logging out the session from which the
|
|
password was changed.
|
|
"""
|
|
request.session.cycle_key()
|
|
if hasattr(user, 'get_session_auth_hash') and request.user == user:
|
|
request.session[HASH_SESSION_KEY] = user.get_session_auth_hash()
|
|
|
|
|
|
default_app_config = 'django.contrib.auth.apps.AuthConfig'
|