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