Centralizes Admin Authentication

Sets admin authentication on authenticator to allow the addition and
removal of admins without creating a new token.

Signed-off-by: Hassan Abouelela <[email protected]>

4 files changed, 13 insertions, 8 deletions

diff --git a/backend/authentication/backend.py b/backend/authentication/backend.py
index abe7313..bdff796 100644
--- a/backend/authentication/backend.py
+++ b/backend/authentication/backend.py
@@ -60,13 +60,8 @@ class JWTAuthenticationBackend(authentication.AuthenticationBackend):
         except Exception:
             raise authentication.AuthenticationError("Could not parse user details.")
 
-        admin = await request.state.db.admins.find_one(
-            {"_id": user_details["id"]}
-        ) is not None
-
-        if admin:
-            scopes.append("admin")
-
         user = User(token, user_details)
+        if user.fetch_admin_status(request):
+            scopes.append("admin")
 
         return authentication.AuthCredentials(scopes), user
diff --git a/backend/authentication/user.py b/backend/authentication/user.py
index a1d78e5..52baa61 100644
--- a/backend/authentication/user.py
+++ b/backend/authentication/user.py
@@ -2,6 +2,7 @@ import typing as t
 
 import jwt
 from starlette.authentication import BaseUser
+from starlette.requests import Request
 
 from backend.constants import SECRET_KEY
 from backend.discord import fetch_user_details
@@ -13,6 +14,7 @@ class User(BaseUser):
     def __init__(self, token: str, payload: dict[str, t.Any]) -> None:
         self.token = token
         self.payload = payload
+        self.admin = False
 
     @property
     def is_authenticated(self) -> bool:
@@ -32,6 +34,13 @@ class User(BaseUser):
     def decoded_token(self) -> dict[str, any]:
         return jwt.decode(self.token, SECRET_KEY, algorithms=["HS256"])
 
+    def fetch_admin_status(self, request: Request) -> bool:
+        self.admin = request.state.db.admins.find_one(
+            {"_id": self.payload["id"]}
+        ) is not None
+
+        return self.admin
+
     async def refresh_data(self) -> None:
         """Fetches user data from discord, and updates the instance."""
         self.payload = await fetch_user_details(self.decoded_token.get("token"))
diff --git a/backend/routes/forms/form.py b/backend/routes/forms/form.py
index b6b722e..e3360b1 100644
--- a/backend/routes/forms/form.py
+++ b/backend/routes/forms/form.py
@@ -26,7 +26,7 @@ class SingleForm(Route):
     @api.validate(resp=Response(HTTP_200=Form, HTTP_404=ErrorMessage), tags=["forms"])
     async def get(self, request: Request) -> JSONResponse:
         """Returns single form information by ID."""
-        admin = request.user.payload["admin"] if request.user.is_authenticated else False  # noqa
+        admin = request.user.admin if request.user.is_authenticated else False
 
         filters = {
             "_id": request.path_params["form_id"]
diff --git a/backend/routes/forms/submit.py b/backend/routes/forms/submit.py
index 55a4875..8627a29 100644
--- a/backend/routes/forms/submit.py
+++ b/backend/routes/forms/submit.py
@@ -127,6 +127,7 @@ class SubmitForm(Route):
             if constants.FormFeatures.REQUIRES_LOGIN.value in form.features:
                 if request.user.is_authenticated:
                     response["user"] = request.user.payload
+                    response["user"]["admin"] = request.user.admin
 
                     if constants.FormFeatures.COLLECT_EMAIL.value in form.features and "email" not in response["user"]:  # noqa
                         return JSONResponse({