From 30da248ba5281686c0045c21bdcd9355d449790a Mon Sep 17 00:00:00 2001 From: Frederick Muriuki Muriithi Date: Thu, 25 Jul 2024 14:53:27 -0500 Subject: Add session module: handles session management. --- uploader/session.py | 88 +++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 88 insertions(+) create mode 100644 uploader/session.py (limited to 'uploader') diff --git a/uploader/session.py b/uploader/session.py new file mode 100644 index 0000000..4706860 --- /dev/null +++ b/uploader/session.py @@ -0,0 +1,88 @@ +"""Deal with user sessions""" +from uuid import UUID, uuid4 +from typing import Any, Optional, TypedDict + +from flask import request, session +from pymonad.either import Left, Right, Either + + +class UserDetails(TypedDict): + """Session information relating specifically to the user.""" + user_id: UUID + name: str + email: str + token: Either + logged_in: bool + + +class SessionInfo(TypedDict): + """All Session information we save.""" + session_id: UUID + user: UserDetails + anon_id: UUID + user_agent: str + ip_addr: str + masquerade: Optional[UserDetails] + auth_server_jwks: Optional[dict[str, Any]] + + +__SESSION_KEY__ = "GN::uploader::session_info" # Do not use this outside this module!! + + +def clear_session_info(): + """Clears the session.""" + session.pop(__SESSION_KEY__) + + +def save_session_info(sess_info: SessionInfo) -> SessionInfo: + """Save `session_info`.""" + # T0d0: if it is an existing session, verify that certain important security + # bits have not changed before saving. + # old_session_info = session.get(__SESSION_KEY__) + # if bool(old_session_info): + # if old_session_info["user_agent"] == request.headers.get("User-Agent"): + # session[__SESSION_KEY__] = sess_info + # return sess_info + # # request session verification + # return verify_session(sess_info) + # New session + session[__SESSION_KEY__] = sess_info + return sess_info + + +def session_info() -> SessionInfo: + """Retrieve the session information""" + anon_id = uuid4() + return save_session_info( + session.get(__SESSION_KEY__, { + "session_id": uuid4(), + "user": { + "user_id": anon_id, + "name": "Anonymous User", + "email": "anon@ymous.user", + "token": Left("INVALID-TOKEN"), + "logged_in": False + }, + "anon_id": anon_id, + "user_agent": request.headers.get("User-Agent"), + "ip_addr": request.environ.get("HTTP_X_FORWARDED_FOR", + request.remote_addr), + "masquerading": None + })) + + +def set_user_token(token: str) -> SessionInfo: + """Set the user's token.""" + info = session_info() + return save_session_info({ + **info, "user": {**info["user"], "token": Right(token)}})#type: ignore[misc] + + +def set_user_details(userdets: UserDetails) -> SessionInfo: + """Set the user details information""" + return save_session_info({**session_info(), "user": userdets})#type: ignore[misc] + + +def user_token() -> Either: + """Retrieve the user token.""" + return session_info()["user"]["token"] -- cgit v1.2.3