adjust CAS api usage & do some slight refactoring

Author: EarthenSky

src/auth/types.py

@@ -11,6 +11,16 @@ class SessionType:
     ALUMNI = "alumni"
     SFU = "sfu"
 
+    @staticmethod
+    def valid_session_type_list():
+        # values taken from https://www.sfu.ca/information-systems/services/cas/cas-for-web-applications.html
+        return [
+            "faculty",
+            "student",
+            "alumni",
+            "sfu"
+        ]
+
 @dataclass
 class SiteUserData:
     computing_id: str

src/auth/urls.py

@@ -47,7 +47,11 @@ async def login_user(
     # verify the ticket is valid
     service = urllib.parse.quote(f"{FRONTEND_ROOT_URL}/api/auth/login?redirect_path={redirect_path}&redirect_fragment={redirect_fragment}")
     service_validate_url = f"https://cas.sfu.ca/cas/serviceValidate?service={service}&ticket={ticket}"
-    cas_response = xmltodict.parse(requests.get(service_validate_url).text)
+    cas_response_text = requests.get(service_validate_url).text
+    cas_response = xmltodict.parse(cas_response_text)
+
+    print("CAS RESPONSE ::")
+    print(cas_response_text)
 
     if "cas:authenticationFailure" in cas_response["cas:serviceResponse"]:
         _logger.info(f"User failed to login, with response {cas_response}")
@@ -65,14 +69,19 @@ async def login_user(
             if cas_response["cas:serviceResponse"]["cas:authenticationSuccess"]["cas:maillist"] == "cmpt-students":
                 session_type = SessionType.CSSS_MEMBER
             else:
-                raise HTTPException(status_code=500, details="malformed authentication response; this is an SFU CAS error")
+                raise HTTPException(status_code=500, details="malformed cas:maillist authentication response; this is an SFU CAS error")
         elif "cas:authtype" in cas_response["cas:serviceResponse"]["cas:authenticationSuccess"]:
             # sfu, alumni, faculty, student
             session_type = cas_response["cas:serviceResponse"]["cas:authenticationSuccess"]["cas:authtype"]
-            if session_type not in SessionType.value_list():
+            if session_type not in SessionType.valid_session_type_list():
                 raise HTTPException(status_code=500, detail=f"unexpected session type from SFU CAS of {session_type}")
+
+            if session_type == "alumni":
+                if "@" not in computing_id:
+                    raise HTTPException(status_code=500, detail=f"invalid alumni computing_id response from CAS AUTH with value {session_type}")
+                computing_id = computing_id.split("@")[0]
         else:
-            raise HTTPException(status_code=500, detail="malformed authentication response; this is an SFU CAS error")
+            raise HTTPException(status_code=500, detail="malformed unknown authentication response; this is an SFU CAS error")
 
         await crud.create_user_session(db_session, session_id, computing_id, session_type)
         await db_session.commit()

src/auth/utils.py

@@ -0,0 +1,20 @@
+from fastapi import HTTPException, Request
+
+import auth.crud
+import database
+
+
+async def logged_in_or_raise(
+    request: Request,
+    db_session: database.DBSession
+) -> tuple[str, str]:
+    """gets the user's computing_id, or raises an exception if the current request is not logged in"""
+    session_id = request.cookies.get("session_id", None)
+    if session_id is None:
+        raise HTTPException(status_code=401)
+
+    session_computing_id = await auth.crud.get_computing_id(db_session, session_id)
+    if session_computing_id is None:
+        raise HTTPException(status_code=401)
+
+    return session_id, session_computing_id

src/exambank/crud.py

@@ -10,11 +10,13 @@ class ExamMetadata:
     course_id: str
 
 async def create_exam():
-    # TODO: these are for admins to run manually
+    # for admins to run manually
+    # TODO: implement this later; for now just upload data manually
     pass
 
 async def update_exam():
-    # TODO: these are for admins to run manually
+    # for admins to run manually
+    # TODO: implement this later; for now just upload data manually
     pass
 
 async def all_exams(

src/exambank/tables.py

@@ -1,4 +1,5 @@
 from datetime import datetime
+from types import ExamKind
 
 from sqlalchemy import Column, DateTime, ForeignKey, Integer, String, Text
 from sqlalchemy.orm import relationship
@@ -8,15 +9,6 @@
 
 # TODO: determine what info will need to be in the spreadsheet, then moved here
 
-# TODO: move this to types.py
-class ExamKind:
-    FINAL = "final"
-    MIDTERM = "midterm"
-    QUIZ = "quiz"
-    ASSIGNMENT = "assignment"
-    NOTES = "notes"
-    MISC = "misc"
-
 class ExamMetadata(Base):
     __tablename__ = "exam_metadata"
 
@@ -41,16 +33,16 @@ class ExamMetadata(Base):
 class Professor(Base):
     __tablename__ = "professor"
 
-    professor_id = Column(Integer, primary_key=True, autoincrement=True)
-    name = Column(String(64), nullable=False)
-    info_url = Column(String(128), nullable=False)
-
     computing_id = Column(
         String(COMPUTING_ID_LEN),
+        ForeignKey("user_session.computing_id"),
+        primary_key=True,
         # Foreign key constriant w/ users table
-        #ForeignKey("user_session.computing_id"),
         nullable=True,
     )
 
+    name = Column(String(64), nullable=False)
+    info_url = Column(String(128), nullable=False)
+
 # TODO: eventually implement a table for courses & course info; hook it in with the rest of the site & coursys api
 

src/exambank/types.py

@@ -0,0 +1,7 @@
+class ExamKind:
+    FINAL = "final"
+    MIDTERM = "midterm"
+    QUIZ = "quiz"
+    ASSIGNMENT = "assignment"
+    NOTES = "notes"
+    MISC = "misc"

src/exambank/urls.py

@@ -1,11 +1,10 @@
 import os
-from typing import Optional
 
 from fastapi import APIRouter, HTTPException, JSONResponse, Request, Response
 
-import auth.crud
 import database
 import exambank.crud
+from auth.utils import logged_in_or_raise
 from exambank.watermark import apply_watermark, create_watermark, raster_pdf
 from permission.types import ExamBankAccess
 from utils import path_in_dir
@@ -17,7 +16,7 @@
     tags=["exam-bank"],
 )
 
-# TODO: update endpoints to use crud functions
+# TODO: update endpoints to use crud functions -> don't use crud actually; refactor to do that later
 
 @router.get(
     "/list/exams"
@@ -34,17 +33,6 @@ async def all_exams(
     exam_list = exambank.crud.all_exams(db_session, course_id_starts_with)
     return JSONResponse([exam.serializable_dict() for exam in exam_list])
 
-@router.get(
-    "/list/courses"
-)
-async def all_courses(
-    _request: Request,
-    _db_session: database.DBSession,
-):
-    # TODO: replace this with a table eventually
-    courses = [f.name for f in os.scandir(f"{EXAM_BANK_DIR}") if f.is_dir()]
-    return JSONResponse(courses)
-
 @router.get(
     "/get/{exam_id}"
 )
@@ -53,17 +41,8 @@ async def get_exam(
     db_session: database.DBSession,
     exam_id: int,
 ):
-    session_id = request.cookies.get("session_id", None)
-    if session_id is None:
-        raise HTTPException(status_code=401)
-
-    computing_id = await auth.crud.get_computing_id(db_session, session_id)
-    if computing_id is None:
-        raise HTTPException(status_code=401)
-
-    # TODO: clean this checking into one function & one computing_id check
-    if not await ExamBankAccess.has_permission(request):
-        raise HTTPException(status_code=401, detail="user must have exam bank access permission")
+    _, session_computing_id = await logged_in_or_raise(request, db_session)
+    await ExamBankAccess.has_permission_or_raise(request, errmsg="user must have exam bank access permission")
 
     # number exams with an exam_id pkey
     # TODO: store resource locations in a db table & simply look them up
@@ -77,10 +56,9 @@ async def get_exam(
         raise HTTPException(status_code=500, detail="Found dangerous pdf path, exiting")
 
     # TODO: test this works nicely
-    watermark = create_watermark(computing_id, 20)
+    watermark = create_watermark(session_computing_id, 20)
     watermarked_pdf = apply_watermark(exam_path, watermark)
     image_bytes = raster_pdf(watermarked_pdf)
 
-    headers = { "Content-Disposition": f'inline; filename="{meta.course_id}_{exam_id}_{computing_id}.pdf"' }
+    headers = { "Content-Disposition": f'inline; filename="{meta.course_id}_{exam_id}_{session_computing_id}.pdf"' }
     return Response(content=image_bytes, headers=headers, media_type="application/pdf")
-

src/officers/urls.py

@@ -7,6 +7,7 @@
 import database
 import officers.crud
 import utils
+from auth.utils import logged_in_or_raise
 from officers.tables import OfficerInfo, OfficerTerm
 from officers.types import InitialOfficerInfo, OfficerInfoUpload, OfficerTermUpload
 from permission.types import OfficerPrivateInfo, WebsiteAdmin
@@ -37,21 +38,6 @@ async def has_officer_private_info_access(
     has_private_access = await OfficerPrivateInfo.has_permission(db_session, computing_id)
     return session_id, computing_id, has_private_access
 
-async def logged_in_or_raise(
-    request: Request,
-    db_session: database.DBSession
-) -> tuple[str, str]:
-    """gets the user's computing_id, or raises an exception if the current request is not logged in"""
-    session_id = request.cookies.get("session_id", None)
-    if session_id is None:
-        raise HTTPException(status_code=401)
-
-    session_computing_id = await auth.crud.get_computing_id(db_session, session_id)
-    if session_computing_id is None:
-        raise HTTPException(status_code=401)
-
-    return session_id, session_computing_id
-
 # ---------------------------------------- #
 # endpoints
 

src/permission/types.py

@@ -51,19 +51,18 @@ async def has_permission(db_session: database.DBSession, computing_id: str) -> b
         return False
 
     @staticmethod
-    async def validate_request(db_session: database.DBSession, request: Request) -> bool:
+    async def validate_request(db_session: database.DBSession, request: Request):
         """
         Checks if the provided request satisfies these permissions, and raises the neccessary
         exceptions if not
         """
-        # TODO: does this function return bool???
         session_id = request.cookies.get("session_id", None)
         if session_id is None:
             raise HTTPException(status_code=401, detail="must be logged in")
-        else:
-            computing_id = await auth.crud.get_computing_id(db_session, session_id)
-            if not await WebsiteAdmin.has_permission(db_session, computing_id):
-                raise HTTPException(status_code=401, detail="must have website admin permissions")
+
+        computing_id = await auth.crud.get_computing_id(db_session, session_id)
+        if not await WebsiteAdmin.has_permission(db_session, computing_id):
+            raise HTTPException(status_code=401, detail="must have website admin permissions")
 
     @staticmethod
     async def has_permission_or_raise(
@@ -84,11 +83,18 @@ async def has_permission(
         if session_id is None:
             return False
 
-        # TODO: allow CSSS officers to access the exam bank, in addition to faculty
-
         if await auth.crud.get_session_type(db_session, session_id) == SessionType.FACULTY:
            return True
 
         # the only non-faculty who can view exams are website admins
         computing_id = await auth.crud.get_computing_id(db_session, session_id)
         return await WebsiteAdmin.has_permission(db_session, computing_id)
+
+    @staticmethod
+    async def has_permission_or_raise(
+        db_session: database.DBSession,
+        request: Request,
+        errmsg: str = "must have exam bank access permissions"
+    ):
+        if not await ExamBankAccess.has_permission(db_session, request):
+            raise HTTPException(status_code=401, detail=errmsg)