1from __future__ import annotations
  2
  3import hmac
  4from collections.abc import Generator
  5from typing import TYPE_CHECKING
  6
  7from plain.runtime import settings
  8from plain.sessions import get_request_session
  9from plain.utils.crypto import salted_hmac
 10from plain.utils.encoding import force_bytes
 11
 12from .requests import get_request_user, set_request_user
 13
 14if TYPE_CHECKING:
 15    from app.users.models import User
 16    from plain.http import Request
 17
 18_USER_ID_SESSION_KEY = "_auth_user_id"
 19_USER_HASH_SESSION_KEY = "_auth_user_hash"
 20
 21
 22def get_session_auth_hash(user: User) -> str:
 23    """
 24    Return an HMAC of the password field.
 25    """
 26    return _get_session_auth_hash(user)
 27
 28
 29def update_session_auth_hash(request: Request, user: User) -> None:
 30    """
 31    Updating a user's password (for example) logs out all sessions for the user.
 32
 33    Take the current request and the updated user object from which the new
 34    session hash will be derived and update the session hash appropriately to
 35    prevent a password change from logging out the session from which the
 36    password was changed.
 37    """
 38
 39    session = get_request_session(request)
 40    session.cycle_key()
 41    if get_request_user(request) == user:
 42        session[_USER_HASH_SESSION_KEY] = get_session_auth_hash(user)
 43
 44
 45def _get_session_auth_fallback_hash(user: User) -> Generator[str]:
 46    for fallback_secret in settings.SECRET_KEY_FALLBACKS:
 47        yield _get_session_auth_hash(user, secret=fallback_secret)
 48
 49
 50def _get_session_auth_hash(user: User, secret: str | None = None) -> str:
 51    key_salt = "plain.auth.get_session_auth_hash"
 52    return salted_hmac(
 53        key_salt,
 54        getattr(user, settings.AUTH_USER_SESSION_HASH_FIELD),
 55        secret=secret,
 56        algorithm="sha256",
 57    ).hexdigest()
 58
 59
 60def login(request: Request, user: User) -> None:
 61    """
 62    Persist a user id and a backend in the request. This way a user doesn't
 63    have to reauthenticate on every request. Note that data set during
 64    the anonymous session is retained when the user logs in.
 65    """
 66    session = get_request_session(request)
 67
 68    if settings.AUTH_USER_SESSION_HASH_FIELD:
 69        session_auth_hash = get_session_auth_hash(user)
 70    else:
 71        session_auth_hash = ""
 72
 73    if _USER_ID_SESSION_KEY in session:
 74        if int(session[_USER_ID_SESSION_KEY]) != user.id:
 75            # To avoid reusing another user's session, create a new, empty
 76            # session if the existing session corresponds to a different
 77            # authenticated user.
 78            session.flush()
 79        elif session_auth_hash and not hmac.compare_digest(
 80            force_bytes(session.get(_USER_HASH_SESSION_KEY, "")),
 81            force_bytes(session_auth_hash),
 82        ):
 83            # If the session hash does not match the current hash, reset the
 84            # session. Most likely this means the password was changed.
 85            session.flush()
 86    else:
 87        # Invalidate the current session key and generate a new one to enhance security,
 88        # typically done after user login to prevent session fixation attacks.
 89        session.cycle_key()
 90
 91    session[_USER_ID_SESSION_KEY] = user.id
 92    session[_USER_HASH_SESSION_KEY] = session_auth_hash
 93    set_request_user(request, user)
 94
 95
 96def logout(request: Request) -> None:
 97    """
 98    Remove the authenticated user's ID from the request and flush their session
 99    data.
100    """
101    # Dispatch the signal before the user is logged out so the receivers have a
102    # chance to find out *who* logged out.
103    session = get_request_session(request)
104    session.flush()
105    set_request_user(request, None)
106
107
108def get_user(request: Request) -> User | None:
109    """
110    Return the user model instance associated with the given request session.
111    If no user is retrieved, return None.
112    """
113    from app.users.models import User
114
115    session = get_request_session(request)
116
117    if _USER_ID_SESSION_KEY not in session:
118        return None
119
120    try:
121        user = User.query.get(id=session[_USER_ID_SESSION_KEY])
122    except User.DoesNotExist:
123        return None
124
125    # If the user models defines a specific field to also hash and compare
126    # (like password), then we verify that the hash of that field is still
127    # the same as when the session was created.
128    #
129    # If it has changed (i.e. password changed), then the session
130    # is no longer valid and cleared out.
131    if settings.AUTH_USER_SESSION_HASH_FIELD:
132        session_hash = session.get(_USER_HASH_SESSION_KEY)
133        if not session_hash:
134            session_hash_verified = False
135        else:
136            session_auth_hash = get_session_auth_hash(user)
137            session_hash_verified = hmac.compare_digest(
138                force_bytes(session_hash), force_bytes(session_auth_hash)
139            )
140        if not session_hash_verified:
141            # If the current secret does not verify the session, try
142            # with the fallback secrets and stop when a matching one is
143            # found.
144            if session_hash and any(
145                hmac.compare_digest(
146                    force_bytes(session_hash), force_bytes(fallback_auth_hash)
147                )
148                for fallback_auth_hash in _get_session_auth_fallback_hash(user)
149            ):
150                session.cycle_key()
151                session[_USER_HASH_SESSION_KEY] = session_auth_hash
152            else:
153                session.flush()
154                user = None
155
156    return user