plain.auth

  1from plain.csrf.middleware import rotate_token
  2from plain.exceptions import ImproperlyConfigured
  3from plain.packages import packages as plain_packages
  4from plain.runtime import settings
  5from plain.utils.crypto import constant_time_compare, salted_hmac
  6
  7from .signals import user_logged_in, user_logged_out
  8
  9USER_ID_SESSION_KEY = "_auth_user_id"
 10USER_HASH_SESSION_KEY = "_auth_user_hash"
 11
 12
 13def _get_user_id_from_session(request):
 14    # This value in the session is always serialized to a string, so we need
 15    # to convert it back to Python whenever we access it.
 16    return get_user_model()._meta.pk.to_python(request.session[USER_ID_SESSION_KEY])
 17
 18
 19def get_session_auth_hash(user):
 20    """
 21    Return an HMAC of the password field.
 22    """
 23    return _get_session_auth_hash(user)
 24
 25
 26def get_session_auth_fallback_hash(user):
 27    for fallback_secret in settings.SECRET_KEY_FALLBACKS:
 28        yield _get_session_auth_hash(user, secret=fallback_secret)
 29
 30
 31def _get_session_auth_hash(user, secret=None):
 32    key_salt = "plain.auth.get_session_auth_hash"
 33    return salted_hmac(
 34        key_salt,
 35        getattr(user, settings.AUTH_USER_SESSION_HASH_FIELD),
 36        secret=secret,
 37        algorithm="sha256",
 38    ).hexdigest()
 39
 40
 41def login(request, user):
 42    """
 43    Persist a user id and a backend in the request. This way a user doesn't
 44    have to reauthenticate on every request. Note that data set during
 45    the anonymous session is retained when the user logs in.
 46    """
 47    if settings.AUTH_USER_SESSION_HASH_FIELD:
 48        session_auth_hash = get_session_auth_hash(user)
 49    else:
 50        session_auth_hash = ""
 51
 52    if USER_ID_SESSION_KEY in request.session:
 53        if _get_user_id_from_session(request) != user.pk:
 54            # To avoid reusing another user's session, create a new, empty
 55            # session if the existing session corresponds to a different
 56            # authenticated user.
 57            request.session.flush()
 58        elif session_auth_hash and not constant_time_compare(
 59            request.session.get(USER_HASH_SESSION_KEY, ""), session_auth_hash
 60        ):
 61            # If the session hash does not match the current hash, reset the
 62            # session. Most likely this means the password was changed.
 63            request.session.flush()
 64    else:
 65        request.session.cycle_key()
 66
 67    request.session[USER_ID_SESSION_KEY] = user._meta.pk.value_to_string(user)
 68    request.session[USER_HASH_SESSION_KEY] = session_auth_hash
 69    if hasattr(request, "user"):
 70        request.user = user
 71    rotate_token(request)
 72    user_logged_in.send(sender=user.__class__, request=request, user=user)
 73
 74
 75def logout(request):
 76    """
 77    Remove the authenticated user's ID from the request and flush their session
 78    data.
 79    """
 80    # Dispatch the signal before the user is logged out so the receivers have a
 81    # chance to find out *who* logged out.
 82    user = getattr(request, "user", None)
 83    user_logged_out.send(sender=user.__class__, request=request, user=user)
 84    request.session.flush()
 85    if hasattr(request, "user"):
 86        request.user = None
 87
 88
 89def get_user_model():
 90    """
 91    Return the User model that is active in this project.
 92    """
 93    try:
 94        return plain_packages.get_model(settings.AUTH_USER_MODEL, require_ready=False)
 95    except ValueError:
 96        raise ImproperlyConfigured(
 97            "AUTH_USER_MODEL must be of the form 'package_label.model_name'"
 98        )
 99    except LookupError:
100        raise ImproperlyConfigured(
101            "AUTH_USER_MODEL refers to model '%s' that has not been installed"
102            % settings.AUTH_USER_MODEL
103        )
104
105
106def get_user(request):
107    """
108    Return the user model instance associated with the given request session.
109    If no user is retrieved, return None.
110    """
111    if USER_ID_SESSION_KEY not in request.session:
112        return None
113
114    user_id = _get_user_id_from_session(request)
115
116    UserModel = get_user_model()
117    try:
118        user = UserModel._default_manager.get(pk=user_id)
119    except UserModel.DoesNotExist:
120        return None
121
122    # If the user models defines a specific field to also hash and compare
123    # (like password), then we verify that the hash of that field is still
124    # the same as when the session was created.
125    #
126    # If it has changed (i.e. password changed), then the session
127    # is no longer valid and cleared out.
128    if settings.AUTH_USER_SESSION_HASH_FIELD:
129        session_hash = request.session.get(USER_HASH_SESSION_KEY)
130        if not session_hash:
131            session_hash_verified = False
132        else:
133            session_auth_hash = get_session_auth_hash(user)
134            session_hash_verified = constant_time_compare(
135                session_hash, session_auth_hash
136            )
137        if not session_hash_verified:
138            # If the current secret does not verify the session, try
139            # with the fallback secrets and stop when a matching one is
140            # found.
141            if session_hash and any(
142                constant_time_compare(session_hash, fallback_auth_hash)
143                for fallback_auth_hash in get_session_auth_fallback_hash(user)
144            ):
145                request.session.cycle_key()
146                request.session[USER_HASH_SESSION_KEY] = session_auth_hash
147            else:
148                request.session.flush()
149                user = None
150
151    return user