Add session module: handles session management.

1 files changed, 88 insertions, 0 deletions

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"]