refactor(permissions): restructured most endpoints to use Depends

Author: jbriones1

src/auth/crud.py

@@ -4,7 +4,7 @@
 import sqlalchemy
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from auth.tables import SiteUser, UserSession
+from auth.tables import SiteUserDB, UserSession
 
 _logger = logging.getLogger(__name__)
 
@@ -18,7 +18,9 @@ async def create_user_session(db_session: AsyncSession, session_id: str, computi
     existing_user_session = await db_session.scalar(
         sqlalchemy.select(UserSession).where(UserSession.computing_id == computing_id)
     )
-    existing_user = await db_session.scalar(sqlalchemy.select(SiteUser).where(SiteUser.computing_id == computing_id))
+    existing_user = await db_session.scalar(
+        sqlalchemy.select(SiteUserDB).where(SiteUserDB.computing_id == computing_id)
+    )
 
     if existing_user is None:
         if existing_user_session is not None:
@@ -27,7 +29,7 @@ async def create_user_session(db_session: AsyncSession, session_id: str, computi
 
         # add new user to User table if it's their first time logging in
         db_session.add(
-            SiteUser(computing_id=computing_id, first_logged_in=datetime.now(), last_logged_in=datetime.now())
+            SiteUserDB(computing_id=computing_id, first_logged_in=datetime.now(), last_logged_in=datetime.now())
         )
 
     if existing_user_session is not None:
@@ -68,18 +70,18 @@ async def task_clean_expired_user_sessions(db_session: AsyncSession):
 
 
 # get the site user given a session ID; returns None when session is invalid
-async def get_site_user(db_session: AsyncSession, session_id: str) -> SiteUser | None:
+async def get_site_user(db_session: AsyncSession, session_id: str) -> SiteUserDB | None:
     query = sqlalchemy.select(UserSession).where(UserSession.session_id == session_id)
     user_session = await db_session.scalar(query)
     if user_session is None:
         return None
 
-    query = sqlalchemy.select(SiteUser).where(SiteUser.computing_id == user_session.computing_id)
+    query = sqlalchemy.select(SiteUserDB).where(SiteUserDB.computing_id == user_session.computing_id)
     return await db_session.scalar(query)
 
 
 async def site_user_exists(db_session: AsyncSession, computing_id: str) -> bool:
-    user = await db_session.scalar(sqlalchemy.select(SiteUser).where(SiteUser.computing_id == computing_id))
+    user = await db_session.scalar(sqlalchemy.select(SiteUserDB).where(SiteUserDB.computing_id == computing_id))
     return user is not None
 
 
@@ -91,8 +93,8 @@ async def update_site_user(db_session: AsyncSession, session_id: str, profile_pi
         return False
 
     query = (
-        sqlalchemy.update(SiteUser)
-        .where(SiteUser.computing_id == user_session.computing_id)
+        sqlalchemy.update(SiteUserDB)
+        .where(SiteUserDB.computing_id == user_session.computing_id)
         .values(profile_picture_url=profile_picture_url)
     )
     await db_session.execute(query)

src/auth/tables.py

@@ -26,7 +26,7 @@ class UserSession(Base):
     )  # the space needed to store 256 bytes in base64
 
 
-class SiteUser(Base):
+class SiteUserDB(Base):
     # user is a reserved word in postgres
     # see: https://stackoverflow.com/questions/22256124/cannot-create-a-database-table-named-user-in-postgresql
     __tablename__ = "site_user"

src/cron/daily.py

@@ -7,7 +7,7 @@
 import google_api
 import utils
 from database import get_db_session
-from officers.crud import all_officers, get_user_by_username
+from officers.crud import get_all_officers, get_user_by_username
 
 _logger = logging.getLogger(__name__)
 
@@ -19,7 +19,7 @@ async def update_google_permissions(db_session):
 
     # TODO: for performance, only include officers with recent end-date (1 yr)
     # but measure performance first
-    for term in await all_officers(db_session):
+    for term in await get_all_officers(db_session):
         if utils.is_active(term):
             # TODO: if google drive permission is not active, update them
             pass
@@ -33,7 +33,7 @@ async def update_google_permissions(db_session):
 async def update_github_permissions(db_session):
     github_permissions, team_id_map = github.all_permissions()
 
-    for term in await all_officers(db_session):
+    for term in await get_all_officers(db_session):
         new_teams = (
             # move all active officers to their respective teams
             github.officer_teams(term.position)

src/data/semesters.py

@@ -2,6 +2,10 @@
 from enum import Enum
 from typing import assert_never
 
+JANUARY = 1
+MAY = 5
+SEPTEMBER = 9
+
 
 class Semester(Enum):
     """semester numbers are assigned by their order in the year"""
@@ -18,7 +22,7 @@ def __str__(self):
         elif self.value == 2:
             return "fall"
         else:
-            assert_never()
+            assert_never(self.value)
 
 
 def step_semesters(semester_start_date: date, num_semesters: int) -> date:
@@ -32,30 +36,32 @@ def step_semesters(semester_start_date: date, num_semesters: int) -> date:
 
 
 def current_semester_start(the_date: date) -> date:
-    if the_date.month >= 9:
-        return date(year=the_date.year, month=9, day=1)
-    elif the_date.month >= 5:
-        return date(year=the_date.year, month=5, day=1)
-    elif the_date.month >= 1:
-        return date(year=the_date.year, month=1, day=1)
+    if the_date.month >= SEPTEMBER:
+        return date(year=the_date.year, month=SEPTEMBER, day=1)
+    elif the_date.month >= MAY:
+        return date(year=the_date.year, month=MAY, day=1)
+    elif the_date.month >= JANUARY:
+        return date(year=the_date.year, month=JANUARY, day=1)
+    else:
+        raise AssertionError("unreachable")
 
 
 def current_semester(the_date: date) -> Semester:
-    if the_date.month >= 9:
+    if the_date.month >= SEPTEMBER:
         return Semester.Fall
-    elif the_date.month >= 5:
+    elif the_date.month >= MAY:
         return Semester.Summer
-    elif the_date.month >= 1:
+    elif the_date.month >= JANUARY:
         return Semester.Spring
     else:
-        assert_never()
+        raise AssertionError("unreachable")
 
 
 def get_semester_start(year: int, semester: Semester):
     match semester:
         case Semester.Fall:
-            return date(year, month=9, day=1)
+            return date(year, month=SEPTEMBER, day=1)
         case Semester.Summer:
-            return date(year, month=5, day=1)
+            return date(year, month=MAY, day=1)
         case Semester.Spring:
-            return date(year, month=1, day=1)
+            return date(year, month=JANUARY, day=1)

src/database.py

@@ -106,7 +106,7 @@ def setup_database():
     # TODO: where is sys.stdout piped to? I want all these to go to a specific logs folder
     sessionmanager = DatabaseSessionManager(
         SQLALCHEMY_TEST_DATABASE_URL if os.environ.get("LOCAL") else SQLALCHEMY_DATABASE_URL,
-        {"echo": False},
+        {"echo": True},
     )
 
 

src/dependencies.py

@@ -1,73 +1,53 @@
-from enum import Enum
 from typing import Annotated
 
-from fastapi import Depends, HTTPException, Request, status
+from fastapi import Cookie, Depends, HTTPException, status
 
 import auth
 import database
-import officers
-from officers.constants import OfficerPositionEnum
-from permission.types import WEBSITE_ADMIN_POSITIONS
+from utils.permissions import is_user_election_officer, is_user_website_admin
 
 
-# Permissions are granted if the Enum value >= the level needed
-class AdminTypeEnum(Enum):
-    Election = 1
-    Full = 2
-
-
-async def is_user_website_admin(computing_id: str, db_session: database.DBSession) -> bool:
-    for position in await officers.crud.current_officer_positions(db_session, computing_id):
-        if position in WEBSITE_ADMIN_POSITIONS:
-            return True
+async def user(db_session: database.DBSession, session_id: Annotated[str | None, Cookie()] = None) -> str | None:
+    if session_id is None:
+        return None
 
-    return False
+    session_computing_id = await auth.crud.get_computing_id(db_session, session_id)
 
+    return session_computing_id
 
-# TODO: Add an election admin version that checks the election attempting to be modified as well
-async def is_user_election_officer(computing_id: str, db_session: database.DBSession) -> bool:
-    """
-    An current election officer has access to all election, prior election officers have no access.
-    """
-    officer_terms = await officers.crud.get_current_terms_by_position(db_session, OfficerPositionEnum.ELECTIONS_OFFICER)
-    for officer in officer_terms:
-        if computing_id == officer.computing_id:
-            return True
 
-    return False
+SessionUser = Annotated[str, Depends(user)]
 
 
-async def get_user(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)
+async def logged_in_user(db_session: database.DBSession, session_id: Annotated[str | None, Cookie()] = None) -> str:
     if session_id is None:
         raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="no session id")
 
     session_computing_id = await auth.crud.get_computing_id(db_session, session_id)
     if session_computing_id is None:
         raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="no computing id")
 
-    return session_id, session_computing_id
+    return session_computing_id
+
+
+LoggedInUser = Annotated[str, Depends(logged_in_user)]
+
+
+async def perm_election(db_session: database.DBSession, computing_id: LoggedInUser) -> str:
+    if not is_user_website_admin(computing_id, db_session) or is_user_election_officer(computing_id, db_session):
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="must be an election admin")
 
+    return computing_id
 
-# Allows path functions to use this without having to add a bunch of checks
-SessionUser = Annotated[tuple[str, str], Depends(get_user)]
 
+ElectionAdmin = Annotated[str, Depends(perm_election)]
 
-async def get_admin(
-    db_session: database.DBSession, session_user: SessionUser, admin_type: AdminTypeEnum
-) -> tuple[str, str]:
-    session_id, computing_id = session_user
-    # Website admins have full permissions
-    if is_user_website_admin(computing_id, db_session):
-        return (session_id, computing_id)
 
-    # Election officers have lower permissions
-    if admin_type == AdminTypeEnum.Election and is_user_election_officer(computing_id, db_session):
-        return (session_id, computing_id)
+async def perm_admin(db_session: database.DBSession, computing_id: LoggedInUser):
+    if not is_user_website_admin(computing_id, db_session):
+        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="must be an admin")
 
-    raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="must be an admin")
+    return computing_id
 
 
-# Allows path functions to use this without having to add a bunch of checks
-SessionAdmin = Annotated[tuple[str, str], Depends(get_admin)]
+SiteAdmin = Annotated[str, Depends(perm_admin)]

src/elections/models.py

@@ -29,6 +29,7 @@ class ElectionResponse(BaseModel):
     available_positions: list[OfficerPositionEnum]
     status: ElectionStatusEnum
 
+    # Private fields
     survey_link: str | None = Field(None, description="Only available to admins")
     candidates: list[RegistrationModel] | None = Field(None, description="Only available to admins")