clean a dictionary of credentials of potentially sensitive info before sending to less secure functions not comprehensive intended for userloginfailed signal this value in the session is always serialized to a string so we need to convert it back to python whenever we access it if the given credentials are valid return a user object this backend doesn t accept these credentials as arguments try the next one this backend says to stop in our tracks this user should not be allowed in at all annotate the user object with the path of the backend the credentials supplied are invalid to all backends fire signal see authenticate return await synctoasyncauthenticaterequest credentials def loginrequest user backendnone sessionauthhash if user is none user request user if hasattruser getsessionauthhash sessionauthhash user getsessionauthhash if sessionkey in request session if getusersessionkeyrequest user pk or sessionauthhash and not constanttimecompare request session gethashsessionkey sessionauthhash 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 cyclekey try backend backend or user backend except attributeerror backends getbackendsreturntuplestrue if lenbackends 1 backend backends0 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 isinstancebackend str raise typeerror backend must be a dotted import path string got r backend request sessionsessionkey user meta pk valuetostringuser request sessionbackendsessionkey backend request sessionhashsessionkey sessionauthhash if hasattrrequest user request user user rotatetokenrequest userloggedin sendsenderuser class requestrequest useruser async def aloginrequest user backendnone 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 see logout return await synctoasynclogoutrequest def getusermodel try return djangoapps getmodelsettings authusermodel requirereadyfalse except valueerror raise improperlyconfigured authusermodel must be of the form applabel modelname except lookuperror raise improperlyconfigured authusermodel refers to model s that has not been installed settings authusermodel def getuserrequest from models import anonymoususer user none try userid getusersessionkeyrequest backendpath request sessionbackendsessionkey except keyerror pass else if backendpath in settings authenticationbackends backend loadbackendbackendpath user backend getuseruserid verify the session if hasattruser getsessionauthhash sessionhash request session gethashsessionkey if not sessionhash sessionhashverified false else sessionauthhash user getsessionauthhash sessionhashverified constanttimecompare sessionhash sessionauthhash if not sessionhashverified if the current secret does not verify the session try with the fallback secrets and stop when a matching one is found if sessionhash and any constanttimecomparesessionhash fallbackauthhash for fallbackauthhash in user getsessionauthfallbackhash request session cyclekey request sessionhashsessionkey sessionauthhash else request session flush user none return user or anonymoususer async def agetuserrequest return the codename of the permission for the specified action 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 see updatesessionauthhash return await synctoasyncupdatesessionauthhashrequest user clean a dictionary of credentials of potentially sensitive info before sending to less secure functions not comprehensive intended for user_login_failed signal this value in the session is always serialized to a string so we need to convert it back to python whenever we access it if the given credentials are valid return a user object this backend doesn t accept these credentials as arguments try the next one this backend says to stop in our tracks this user should not be allowed in at all annotate the user object with the path of the backend the credentials supplied are invalid to all backends fire signal see authenticate 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 to avoid reusing another user s session create a new empty session if the existing session corresponds to a different authenticated user see login 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 see logout return the user model that is active in this project return the user model instance associated with the given request session if no user is retrieved return an instance of anonymoususer verify the session if the current secret does not verify the session try with the fallback secrets and stop when a matching one is found see get_user return the codename of the permission for the specified action 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 see update_session_auth_hash
import inspect
import re

from asgiref.sync import sync_to_async

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.views.decorators.debug import sensitive_variables

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)


@sensitive_variables("credentials")
def _clean_credentials(credentials):
    
    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):
    
    
    return get_user_model()._meta.pk.to_python(request.session[SESSION_KEY])


@sensitive_variables("credentials")
def authenticate(request=None, **credentials):
    
    for backend, backend_path in _get_backends(return_tuples=True):
        backend_signature = inspect.signature(backend.authenticate)
        try:
            backend_signature.bind(request, **credentials)
        except TypeError:
            
            
            continue
        try:
            user = backend.authenticate(request, **credentials)
        except PermissionDenied:
            
            
            break
        if user is None:
            continue
        
        user.backend = backend_path
        return user

    
    user_login_failed.send(
        sender=__name__, credentials=_clean_credentials(credentials), request=request
    )


@sensitive_variables("credentials")
async def aauthenticate(request=None, **credentials):
    
    return await sync_to_async(authenticate)(request, **credentials)


def login(request, user, backend=None):
    
    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
            )
        ):
            
            
            
            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)


async def alogin(request, user, backend=None):
    
    return await sync_to_async(login)(request, user, backend)


def logout(request):
    
    
    
    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)
    request.session.flush()
    if hasattr(request, "user"):
        from django.contrib.auth.models import AnonymousUser

        request.user = AnonymousUser()


async def alogout(request):
    
    return await sync_to_async(logout)(request)


def get_user_model():
    
    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):
    
    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)
            
            if hasattr(user, "get_session_auth_hash"):
                session_hash = request.session.get(HASH_SESSION_KEY)
                if not session_hash:
                    session_hash_verified = False
                else:
                    session_auth_hash = user.get_session_auth_hash()
                    session_hash_verified = constant_time_compare(
                        session_hash, session_auth_hash
                    )
                if not session_hash_verified:
                    
                    
                    
                    if session_hash and any(
                        constant_time_compare(session_hash, fallback_auth_hash)
                        for fallback_auth_hash in user.get_session_auth_fallback_hash()
                    ):
                        request.session.cycle_key()
                        request.session[HASH_SESSION_KEY] = session_auth_hash
                    else:
                        request.session.flush()
                        user = None

    return user or AnonymousUser()


async def aget_user(request):
    
    return await sync_to_async(get_user)(request)


def get_permission_codename(action, opts):
    
    return "%s_%s" % (action, opts.model_name)


def update_session_auth_hash(request, user):
    
    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()


async def aupdate_session_auth_hash(request, user):
    
    return await sync_to_async(update_session_auth_hash)(request, user)