152 lines
5.1 KiB
Python
152 lines
5.1 KiB
Python
import inspect
|
|
import re
|
|
|
|
from django.conf import settings
|
|
from django.core.exceptions import ImproperlyConfigured, PermissionDenied
|
|
from django.utils.module_loading import import_by_path
|
|
from django.middleware.csrf import rotate_token
|
|
|
|
from .signals import user_logged_in, user_logged_out, user_login_failed
|
|
|
|
SESSION_KEY = '_auth_user_id'
|
|
BACKEND_SESSION_KEY = '_auth_user_backend'
|
|
REDIRECT_FIELD_NAME = 'next'
|
|
|
|
|
|
def load_backend(path):
|
|
return import_by_path(path)()
|
|
|
|
|
|
def get_backends():
|
|
backends = []
|
|
for backend_path in settings.AUTHENTICATION_BACKENDS:
|
|
backends.append(load_backend(backend_path))
|
|
if not backends:
|
|
raise ImproperlyConfigured('No authentication backends have been defined. Does AUTHENTICATION_BACKENDS contain anything?')
|
|
return backends
|
|
|
|
|
|
def _clean_credentials(credentials):
|
|
"""
|
|
Cleans 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 authenticate(**credentials):
|
|
"""
|
|
If the given credentials are valid, return a User object.
|
|
"""
|
|
for backend in get_backends():
|
|
try:
|
|
inspect.getcallargs(backend.authenticate, **credentials)
|
|
except TypeError:
|
|
# This backend doesn't accept these credentials as arguments. Try the next one.
|
|
continue
|
|
|
|
try:
|
|
user = backend.authenticate(**credentials)
|
|
except PermissionDenied:
|
|
# This backend says to stop in our tracks - this user should not be allowed in at all.
|
|
return None
|
|
if user is None:
|
|
continue
|
|
# Annotate the user object with the path of the backend.
|
|
user.backend = "%s.%s" % (backend.__module__, backend.__class__.__name__)
|
|
return user
|
|
|
|
# The credentials supplied are invalid to all backends, fire signal
|
|
user_login_failed.send(sender=__name__,
|
|
credentials=_clean_credentials(credentials))
|
|
|
|
|
|
def login(request, user):
|
|
"""
|
|
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.
|
|
"""
|
|
if user is None:
|
|
user = request.user
|
|
# TODO: It would be nice to support different login methods, like signed cookies.
|
|
if SESSION_KEY in request.session:
|
|
if request.session[SESSION_KEY] != user.pk:
|
|
# 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()
|
|
request.session[SESSION_KEY] = user.pk
|
|
request.session[BACKEND_SESSION_KEY] = user.backend
|
|
if hasattr(request, 'user'):
|
|
request.user = user
|
|
rotate_token(request)
|
|
user_logged_in.send(sender=user.__class__, request=request, user=user)
|
|
|
|
|
|
def logout(request):
|
|
"""
|
|
Removes the authenticated user's ID from the request and flushes 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 hasattr(user, 'is_authenticated') and not user.is_authenticated():
|
|
user = None
|
|
user_logged_out.send(sender=user.__class__, request=request, user=user)
|
|
|
|
request.session.flush()
|
|
if hasattr(request, 'user'):
|
|
from django.contrib.auth.models import AnonymousUser
|
|
request.user = AnonymousUser()
|
|
|
|
|
|
def get_user_model():
|
|
"""
|
|
Returns the User model that is active in this project.
|
|
"""
|
|
from django.db.models import get_model
|
|
|
|
try:
|
|
app_label, model_name = settings.AUTH_USER_MODEL.split('.')
|
|
except ValueError:
|
|
raise ImproperlyConfigured("AUTH_USER_MODEL must be of the form 'app_label.model_name'")
|
|
user_model = get_model(app_label, model_name)
|
|
if user_model is None:
|
|
raise ImproperlyConfigured("AUTH_USER_MODEL refers to model '%s' that has not been installed" % settings.AUTH_USER_MODEL)
|
|
return user_model
|
|
|
|
|
|
def get_user(request):
|
|
"""
|
|
Returns the user model instance associated with the given request session.
|
|
If no user is retrieved an instance of `AnonymousUser` is returned.
|
|
"""
|
|
from .models import AnonymousUser
|
|
try:
|
|
user_id = request.session[SESSION_KEY]
|
|
backend_path = request.session[BACKEND_SESSION_KEY]
|
|
assert backend_path in settings.AUTHENTICATION_BACKENDS
|
|
backend = load_backend(backend_path)
|
|
user = backend.get_user(user_id) or AnonymousUser()
|
|
except (KeyError, AssertionError):
|
|
user = AnonymousUser()
|
|
return user
|
|
|
|
|
|
def get_permission_codename(action, opts):
|
|
"""
|
|
Returns the codename of the permission for the specified action.
|
|
"""
|
|
return '%s_%s' % (action, opts.model_name)
|