refactor(permissions): rewrote permissions to integrate admin types

Author: jbriones1

src/__init__.py

@@ -0,0 +1 @@
+from . import dependencies

src/auth/__init__.py

@@ -0,0 +1 @@
+from auth import crud

src/dependencies.py

@@ -0,0 +1,73 @@
+from enum import Enum
+from typing import Annotated
+
+from fastapi import Depends, HTTPException, Request, status
+
+import auth
+import database
+import officers
+from officers.constants import OfficerPositionEnum
+from permission.types import WEBSITE_ADMIN_POSITIONS
+
+
+# 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
+
+    return False
+
+
+# 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
+
+
+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)
+    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
+
+
+# Allows path functions to use this without having to add a bunch of checks
+SessionUser = Annotated[tuple[str, str], Depends(get_user)]
+
+
+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)
+
+    raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="must be an admin")
+
+
+# Allows path functions to use this without having to add a bunch of checks
+SessionAdmin = Annotated[tuple[str, str], Depends(get_admin)]

src/elections/urls.py

@@ -8,6 +8,7 @@
 import elections.tables
 import nominees.crud
 import registrations.crud
+from dependencies import SessionUser
 from elections.models import (
     ElectionParams,
     ElectionResponse,
@@ -18,7 +19,7 @@
 from officers.constants import COUNCIL_REP_ELECTION_POSITIONS, GENERAL_ELECTION_POSITIONS, OfficerPositionEnum
 from permission.types import ElectionOfficer, WebsiteAdmin
 from utils.shared_models import DetailModel, SuccessResponse
-from utils.urls import get_current_user, slugify
+from utils.urls import slugify
 
 router = APIRouter(
     prefix="/election",
@@ -27,10 +28,10 @@
 
 
 async def get_election_permissions(
-    request: Request,
+    session_user: SessionUser,
     db_session: database.DBSession,
 ) -> tuple[bool, str | None, str | None]:
-    session_id, computing_id = await get_current_user(request, db_session)
+    session_id, computing_id = session_user
     if not session_id or not computing_id:
         return False, None, None
 
@@ -90,14 +91,14 @@ def _raise_if_bad_election_data(
     "",
     description="Returns a list of all election & their status",
     response_model=list[ElectionResponse],
-    responses={404: {"description": "No election found", "model": DetailModel}},
+    responses={status.HTTP_404_NOT_FOUND: {"description": "No election found", "model": DetailModel}},
     operation_id="get_all_elections",
 )
 async def list_elections(
-    request: Request,
+    session_user: SessionUser,
     db_session: database.DBSession,
 ):
-    is_admin, _, _ = await get_election_permissions(request, db_session)
+    is_admin, _, _ = await get_election_permissions(session_user, db_session)
     election_list = await elections.crud.get_all_elections(db_session)
     if election_list is None or len(election_list) == 0:
         raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="no election found")

src/officers/__init__.py

@@ -0,0 +1 @@
+import crud

src/officers/crud.py

@@ -1,5 +1,5 @@
 from collections.abc import Sequence
-from datetime import date, datetime
+from datetime import date
 
 import sqlalchemy
 from fastapi import HTTPException
@@ -11,7 +11,7 @@
 import utils
 from data import semesters
 from officers.constants import OfficerPosition
-from officers.models import OfficerInfoResponse, OfficerTermCreate
+from officers.models import OfficerInfoResponse
 from officers.tables import OfficerInfo, OfficerTerm
 
 # NOTE: this module should not do any data validation; that should be done in the urls.py or higher layer
@@ -59,6 +59,29 @@ async def current_officers(
     return officer_list
 
 
+async def get_current_terms_by_position(db_session: database.DBSession, position: str) -> list[OfficerInfoResponse]:
+    """
+    Get current officer that holds a position
+    """
+    curr_time = date.today()
+    query = (
+        sqlalchemy.select(OfficerTerm)
+        .join(OfficerInfo, OfficerTerm.computing_id)
+        .where(
+            (OfficerTerm.start_date <= curr_time) & (OfficerTerm.end_date >= curr_time) & OfficerTerm.position
+            == position
+        )
+        .order_by(OfficerTerm.start_date.desc())
+    )
+
+    result = (await db_session.execute(query)).all()
+    officer_list = []
+    for term in result:
+        officer_list.append(OfficerTerm(*term))
+
+    return officer_list
+
+
 async def all_officers(db_session: AsyncSession, include_future_terms: bool) -> list[OfficerInfoResponse]:
     """
     This could be a lot of data, so be careful

src/officers/models.py

@@ -1,7 +1,8 @@
 from datetime import date
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, Field
 
+from constants import COMPUTING_ID_LEN
 from officers.constants import OFFICER_LEGAL_NAME_MAX, OfficerPositionEnum
 
 OFFICER_PRIVATE_INFO = {

src/permission/types.py

@@ -10,6 +10,14 @@
 from data.semesters import step_semesters
 from officers.constants import OfficerPositionEnum
 
+WEBSITE_ADMIN_POSITIONS: list[OfficerPositionEnum] = [
+    OfficerPositionEnum.PRESIDENT,
+    OfficerPositionEnum.VICE_PRESIDENT,
+    OfficerPositionEnum.DIRECTOR_OF_ARCHIVES,
+    OfficerPositionEnum.SYSTEM_ADMINISTRATOR,
+    OfficerPositionEnum.WEBMASTER,
+]
+
 
 class OfficerPrivateInfo:
     @staticmethod
@@ -29,47 +37,3 @@ async def has_permission(db_session: database.DBSession, computing_id: str) -> b
                 return True
 
         return False
-
-
-class ElectionOfficer:
-    @staticmethod
-    async def has_permission(db_session: database.DBSession, computing_id: str) -> bool:
-        """
-        An current election officer has access to all election, prior election officers have no access.
-        """
-        officer_terms = await officers.crud.current_officers(db_session, True)
-        current_election_officer = officer_terms.get(officers.constants.OfficerPositionEnum.ELECTIONS_OFFICER)
-        if current_election_officer is not None:
-            for election_officer in current_election_officer[1]:
-                if election_officer.private_data.computing_id == computing_id and election_officer.is_current_officer:
-                    return True
-
-        return False
-
-
-class WebsiteAdmin:
-    WEBSITE_ADMIN_POSITIONS: ClassVar[list[OfficerPositionEnum]] = [
-        OfficerPositionEnum.PRESIDENT,
-        OfficerPositionEnum.VICE_PRESIDENT,
-        OfficerPositionEnum.DIRECTOR_OF_ARCHIVES,
-        OfficerPositionEnum.SYSTEM_ADMINISTRATOR,
-        OfficerPositionEnum.WEBMASTER,
-    ]
-
-    @staticmethod
-    async def has_permission(db_session: database.DBSession, computing_id: str) -> bool:
-        """
-        A website admin has to be an active officer who has one of the above positions
-        """
-        for position in await officers.crud.current_officer_positions(db_session, computing_id):
-            if position in WebsiteAdmin.WEBSITE_ADMIN_POSITIONS:
-                return True
-        return False
-
-    @staticmethod
-    async def has_permission_or_raise(
-        db_session: database.DBSession, computing_id: str, errmsg: str = "must have website admin permissions"
-    ) -> bool:
-        if not await WebsiteAdmin.has_permission(db_session, computing_id):
-            raise HTTPException(status_code=403, detail=errmsg)
-        return True

src/utils/urls.py

@@ -1,83 +1,7 @@
 import re
-from enum import Enum
-
-from fastapi import HTTPException, Request, status
-
-import auth
-import auth.crud
-import database
-from permission.types import ElectionOfficer, WebsiteAdmin
-
-
-class AdminTypeEnum(Enum):
-    Full = 1
-    Election = 2
 
 
 # TODO: move other utils into this module
 def slugify(text: str) -> str:
     """Creates a unique slug based on text passed in. Assumes non-unicode text."""
     return re.sub(r"[\W_]+", "-", text.strip().replace("/", "").replace("&", ""))
-
-
-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, 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=401, detail="no computing id")
-
-    return session_id, session_computing_id
-
-
-async def get_current_user(request: Request, db_session: database.DBSession) -> tuple[str, str] | tuple[None, None]:
-    """
-    Gets information about the currently logged in user.
-
-    Args:
-        request: The request being checked
-        db_session: The current database session
-
-    Returns:
-        A tuple of either (None, None) if there is no logged in user or a tuple (session ID, computing ID)
-    """
-    session_id = request.cookies.get("session_id", None)
-    if session_id is None:
-        return None, None
-
-    session_computing_id = await auth.crud.get_computing_id(db_session, session_id)
-    if session_computing_id is None:
-        return None, None
-
-    return session_id, session_computing_id
-
-
-# TODO: Add an election admin version that checks the election attempting to be modified as well
-async def admin_or_raise(
-    request: Request, db_session: database.DBSession, admintype: AdminTypeEnum = AdminTypeEnum.Full
-) -> tuple[str, str]:
-    session_id, computing_id = await get_current_user(request, db_session)
-    if not session_id or not computing_id:
-        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="must be logged in")
-
-    # where valid means election officer or website admin
-    if (await WebsiteAdmin.has_permission(db_session, computing_id)) or (
-        admintype is AdminTypeEnum.Election and await ElectionOfficer.has_permission(db_session, computing_id)
-    ):
-        return session_id, computing_id
-
-    raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="must be an admin")
-
-
-async def is_website_admin(request: Request, db_session: database.DBSession) -> tuple[bool, str | None, str | None]:
-    session_id, computing_id = await get_current_user(request, db_session)
-    if session_id is None or computing_id is None:
-        return False, session_id, computing_id
-
-    if await WebsiteAdmin.has_permission(db_session, computing_id):
-        return True, session_id, computing_id
-
-    return False, session_id, computing_id