➕ Add admin dependency for admin invite routes

Co-authored-by: Andreas Müller <[email protected]>

backend/src/appointment/dependencies/auth.py

@@ -10,7 +10,7 @@
 from ..database import repo, schemas
 from ..dependencies.database import get_db
 from ..exceptions import validation
-from ..exceptions.validation import InvalidTokenException
+from ..exceptions.validation import InvalidTokenException, InvalidPermissionLevelException
 
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token", auto_error=False)
 
@@ -55,6 +55,27 @@ def get_subscriber(
     return user
 
 
+def get_admin_subscriber(
+    token: Annotated[str, Depends(oauth2_scheme)],
+    db: Session = Depends(get_db),
+):
+    if token is None:
+        raise InvalidTokenException()
+
+    """Automatically retrieve and return the subscriber"""
+    user = get_user_from_token(db, token)
+
+    if user is None:
+        raise InvalidTokenException()
+
+    # check admin allow list
+    ALLOW_LIST = os.getenv("APP_ADMIN_ALLOW_LIST", '').split(',')
+    if not user.email in ALLOW_LIST:
+        raise InvalidPermissionLevelException()
+
+    return user
+
+
 def get_subscriber_from_signed_url(
     url: str = Body(..., embed=True),
     db: Session = Depends(get_db),

backend/src/appointment/exceptions/validation.py

@@ -20,6 +20,15 @@ def get_msg(self):
         return l10n('unknown-error')
 
 
+class InvalidPermissionLevelException(APIException):
+    """Raise when the subscribers permission level is too low for the action"""
+    id_code = 'INVALID_PERMISSION_LEVEL'
+    status_code = 401
+
+    def get_msg(self):
+        return l10n('protected-route-fail')
+
+
 class InvalidTokenException(APIException):
     """Raise when the subscriber could not be parsed from the auth token"""
     id_code = 'INVALID_TOKEN'

backend/src/appointment/routes/invite.py

@@ -4,7 +4,8 @@
 from sqlalchemy.orm import Session
 
 from ..database import repo, schemas, models
-from ..dependencies.auth import get_subscriber, get_subscriber_from_signed_url
+from ..database.models import Subscriber
+from ..dependencies.auth import get_admin_subscriber
 from ..dependencies.database import get_db
 
 from ..exceptions import validation
@@ -13,13 +14,14 @@
 
 
 @router.get('/', response_model=list[schemas.Invite])
-def get_all_invites(db: Session = Depends(get_db)):
+def get_all_invites(db: Session = Depends(get_db), admin: Subscriber = Depends(get_admin_subscriber)):
+    """List all existing invites, needs admin permissions"""
     return db.query(models.Invite).all()
 
 
 @router.post("/generate/{n}", response_model=list[schemas.Invite])
-def generate_invite_codes(n: int, db: Session = Depends(get_db)):
-    """endpoint to generate n invite codes"""
+def generate_invite_codes(n: int, db: Session = Depends(get_db), admin: Subscriber = Depends(get_admin_subscriber)):
+    """endpoint to generate n invite codes, needs admin permissions"""
     return repo.invite.generate_codes(db, n)
 
 
@@ -39,8 +41,8 @@ def use_invite_code(code: str, db: Session = Depends(get_db)):
 
 
 @router.put("/revoke/{code}")
-def revoke_invite_code(code: str, db: Session = Depends(get_db)):
-    """endpoint to revoke a given invite code and mark in unavailable"""
+def revoke_invite_code(code: str, db: Session = Depends(get_db), admin: Subscriber = Depends(get_admin_subscriber)):
+    """endpoint to revoke a given invite code and mark in unavailable, needs admin permissions"""
     if not repo.invite.code_exists(db, code):
         raise validation.InviteCodeNotFoundException()
     if not repo.invite.code_is_available(db, code):