add session management (initially for /survey)

compendium_v2/auth/session_management.py

-from flask_login import LoginManager, UserMixin  # type: ignore
+from flask import jsonify
+from datetime import datetime
+from compendium_v2.db import session_scope
+from compendium_v2.db.auth_model import User
+from flask_login import LoginManager  # type: ignore
 
 
-# TODO: implement user model as SQLAlchemy model
-class User(UserMixin):
-    pass
+def create_user(email: str, fullname: str, oidc_sub: str):
+    """
+    Function used to create a new user in the database.
+
+    :param email: The email of the user
+    :param fullname: The full name of the user
+    :param oidc_sub: The OIDC subject identifier (ID) of the user
+    :return: The user object
+    """
+
+    with session_scope() as session:
+        user = User(email=email, fullname=fullname, oidc_sub=oidc_sub)
+        session.add(user)
+        return user
 
 
-def fetch_user(email: str):
+def fetch_user(profile: dict):
+    """
+    Function used to resolve an OIDC profile to a user in the database.
+
+    :param profile: OIDC profile information
+    :return: User object if the user exists, None otherwise.
+    """
+
+    with session_scope() as session:
+        sub_id = profile['sub']
+        user = session.query(User).filter(User.oidc_sub == sub_id).first()
+        if user is None:
+            return None
+        user.last_login = datetime.utcnow()
+        return user
+
+
+def user_loader(user_id: str):
     """
     Function used to retrieve the internal user model for the user attempting login.
 
-    :param profile: The email of the user attempting login.
+    :param user_id: The ID of the user attempting login.
     :return: User object if the user exists, None otherwise.
     """
 
-    # TODO: fetch user from database instead of just creating a user object
-    user = User()
-    user.id = email
-    return user
+    with session_scope() as session:
+        user = session.query(User).filter(User.id == user_id).first()
+        if user is None:
+            return None
+        user.last_login = datetime.utcnow()
+        return user
+
+
+def unauth_handler():
+    return jsonify(success=False,
+                   data={'login_required': True},
+                   message='Authorize to access this page.'), 401
 
 
 def setup_login_manager(login_manager: LoginManager):
 
-    login_manager.user_loader(fetch_user)
+    login_manager.user_loader(user_loader)
+    login_manager.unauthorized_handler(unauth_handler)

compendium_v2/routes/authentication.py

@@ -2,7 +2,7 @@
 from flask_login import login_required, login_user, logout_user  # type: ignore
 from flask import Blueprint, url_for, redirect
 from compendium_v2.auth import get_client
-from compendium_v2.auth.session_management import fetch_user
+from compendium_v2.auth.session_management import fetch_user, create_user
 
 routes = Blueprint('authentication', __name__)
 
@@ -26,7 +26,10 @@ def authorize():
     if 'email' not in profile or not profile['email']:
         return '<h3>Authentication failed: Invalid user response from provider</h3>', 400
 
-    user = fetch_user(profile['email'])
+    user = fetch_user(profile)
+    if user is None:
+        # create a new user
+        user = create_user(profile['email'], profile['name'], profile['sub'])
     login_user(user)
 
     # redirect to /survey since we only require login for this part of the app

compendium_v2/routes/default.py

 import pkg_resources
-from flask import Blueprint, jsonify, render_template, Response
-from flask_login import login_required  # type: ignore
+from flask import Blueprint, jsonify, render_template, Response, redirect, url_for
+from flask_login import current_user  # type: ignore
 from compendium_v2.routes import common
 
 routes = Blueprint('compendium-v2-default', __name__)
@@ -47,10 +47,13 @@ def index(path):
 @routes.route('/survey', defaults={'path': ''}, methods=['GET'])
 @routes.route('/survey/', defaults={'path': ''}, methods=['GET'])
 @routes.route('/survey/<path:path>', methods=['GET'])
-@login_required
 def survey_index(path):
     is_api = path.startswith('api')
 
+    # implement login_required with a redirect to login instead of a 401
+    if not current_user.is_authenticated:
+        return redirect(url_for('authentication.login'))
+
     if is_api:
         # return 404 for API requests that don't match a route
         return Response(status=404, mimetype='application/json', response='{"error": "Not Found"}')