Add Role Based Editors To Forms

Adds the ability to specify who can edit forms using discord roles.

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

4 files changed, 51 insertions, 34 deletions

diff --git a/backend/discord.py b/backend/discord.py
index 4e35216..6c8eefe 100644
--- a/backend/discord.py
+++ b/backend/discord.py
@@ -188,3 +188,8 @@ async def _verify_access_helper(
 async def verify_response_access(form_id: str, request: starlette.requests.Request) -> bool:
     """Ensure the user can access responses on the requested resource."""
     return await _verify_access_helper(form_id, request, "response_readers")
+
+
+async def verify_edit_access(form_id: str, request: starlette.requests.Request) -> bool:
+    """Ensure the user can view and modify the requested resource."""
+    return await _verify_access_helper(form_id, request, "editors")
diff --git a/backend/models/form.py b/backend/models/form.py
index 45a7e0b..4ee2804 100644
--- a/backend/models/form.py
+++ b/backend/models/form.py
@@ -44,6 +44,7 @@ class Form(BaseModel):
     webhook: _WebHook = None
     discord_role: t.Optional[str]
     response_readers: t.Optional[list[str]]
+    editors: t.Optional[list[str]]
 
     class Config:
         allow_population_by_field_name = True
@@ -68,7 +69,7 @@ class Form(BaseModel):
 
         return value
 
-    @validator("response_readers")
+    @validator("response_readers", "editors")
     def validate_role_scoping(cls, value: t.Optional[list[str]]):
         """Ensure special role based permissions aren't granted to the @everyone role."""
         if value and str(DISCORD_GUILD) in value:
diff --git a/backend/routes/forms/form.py b/backend/routes/forms/form.py
index 0f96b85..15ff9a6 100644
--- a/backend/routes/forms/form.py
+++ b/backend/routes/forms/form.py
@@ -10,13 +10,16 @@ from starlette.authentication import requires
 from starlette.requests import Request
 from starlette.responses import JSONResponse
 
-from backend import constants
+from backend import constants, discord
 from backend.models import Form
 from backend.route import Route
 from backend.routes.forms.discover import EMPTY_FORM
 from backend.routes.forms.unittesting import filter_unittests
 from backend.validation import ErrorMessage, OkayResponse, api
 
+NOT_FOUND_ERROR = JSONResponse({"error": "not_found"}, status_code=404)
+PUBLIC_FORM_FEATURES = (constants.FormFeatures.OPEN, constants.FormFeatures.DISCOVERABLE)
+
 
 class SingleForm(Route):
     """
@@ -31,9 +34,17 @@ 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.admin if request.user.is_authenticated else False
         form_id = request.path_params["form_id"].lower()
 
+        try:
+            admin = await discord.verify_edit_access(form_id, request)
+        except discord.FormNotFoundError:
+            if not constants.PRODUCTION and form_id == EMPTY_FORM.id:
+                # Empty form to help with authentication in development.
+                return JSONResponse(EMPTY_FORM.dict(admin=False))
+            else:
+                return NOT_FOUND_ERROR
+
         filters = {
             "_id": form_id
         }
@@ -41,25 +52,19 @@ class SingleForm(Route):
         if not admin:
             filters["features"] = {"$in": ["OPEN", "DISCOVERABLE"]}
 
-        if raw_form := await request.state.db.forms.find_one(filters):
-            form = Form(**raw_form)
-            if not admin:
-                form = filter_unittests(form)
-
-            return JSONResponse(form.dict(admin=admin))
-
-        elif not constants.PRODUCTION and form_id == EMPTY_FORM.id:
-            # Empty form to help with authentication in development.
-            return JSONResponse(EMPTY_FORM.dict(admin=admin))
+        form = Form(**await request.state.db.forms.find_one(filters))
+        if not admin:
+            form = filter_unittests(form)
 
-        return JSONResponse({"error": "not_found"}, status_code=404)
+        return JSONResponse(form.dict(admin=admin))
 
-    @requires(["authenticated", "admin"])
+    @requires(["authenticated"])
     @api.validate(
         resp=Response(
             HTTP_200=OkayResponse,
             HTTP_400=ErrorMessage,
-            HTTP_404=ErrorMessage
+            HTTP_401=ErrorMessage,
+            HTTP_404=ErrorMessage,
         ),
         tags=["forms"]
     )
@@ -70,10 +75,17 @@ class SingleForm(Route):
         except json.decoder.JSONDecodeError:
             return JSONResponse("Expected a JSON body.", 400)
 
-        form_id = {"_id": request.path_params["form_id"].lower()}
-        if raw_form := await request.state.db.forms.find_one(form_id):
+        form_id = request.path_params["form_id"].lower()
+
+        try:
+            if not await discord.verify_edit_access(form_id, request):
+                return JSONResponse({"error": "unauthorized"}, status_code=401)
+        except discord.FormNotFoundError:
+            return NOT_FOUND_ERROR
+
+        if raw_form := await request.state.db.forms.find_one({"id": form_id}):
             if "_id" in data or "id" in data:
-                if (data.get("id") or data.get("_id")) != form_id["_id"]:
+                if (data.get("id") or data.get("_id")) != form_id:
                     return JSONResponse({"error": "locked_field"}, status_code=400)
 
             # Build Data Merger
@@ -90,7 +102,7 @@ class SingleForm(Route):
             except ValidationError as e:
                 return JSONResponse(e.errors(), status_code=422)
 
-            await request.state.db.forms.replace_one(form_id, form.dict())
+            await request.state.db.forms.replace_one({"id": form_id}, form.dict())
 
             return JSONResponse(form.dict())
         else:
@@ -98,21 +110,20 @@ class SingleForm(Route):
 
     @requires(["authenticated", "admin"])
     @api.validate(
-        resp=Response(HTTP_200=OkayResponse, HTTP_404=ErrorMessage),
+        resp=Response(HTTP_200=OkayResponse, HTTP_401=ErrorMessage, HTTP_404=ErrorMessage),
         tags=["forms"]
     )
     async def delete(self, request: Request) -> JSONResponse:
         """Deletes form by ID."""
-        if not await request.state.db.forms.find_one(
-            {"_id": request.path_params["form_id"].lower()}
-        ):
-            return JSONResponse({"error": "not_found"}, status_code=404)
+        form_id = request.path_params["form_id"].lower()
+
+        try:
+            if not await discord.verify_edit_access(form_id, request):
+                return JSONResponse({"error": "unauthorized"}, status_code=401)
+        except discord.FormNotFoundError:
+            return NOT_FOUND_ERROR
 
-        await request.state.db.forms.delete_one(
-            {"_id": request.path_params["form_id"].lower()}
-        )
-        await request.state.db.responses.delete_many(
-            {"form_id": request.path_params["form_id"].lower()}
-        )
+        await request.state.db.forms.delete_one({"_id": form_id})
+        await request.state.db.responses.delete_many({"form_id": form_id})
 
         return JSONResponse({"status": "ok"})
diff --git a/backend/routes/forms/index.py b/backend/routes/forms/index.py
index 22171fa..38be693 100644
--- a/backend/routes/forms/index.py
+++ b/backend/routes/forms/index.py
@@ -15,13 +15,13 @@ from backend.validation import ErrorMessage, OkayResponse, api
 
 class FormsList(Route):
     """
-    List all available forms for administrator viewing.
+    List all available forms for authorized viewers.
     """
 
     name = "forms_list_create"
     path = "/"
 
-    @requires(["authenticated", "admin"])
+    @requires(["authenticated", "Admins"])
     @api.validate(resp=Response(HTTP_200=FormList), tags=["forms"])
     async def get(self, request: Request) -> JSONResponse:
         """Return a list of all forms to authenticated users."""
@@ -38,7 +38,7 @@ class FormsList(Route):
             forms
         )
 
-    @requires(["authenticated", "admin"])
+    @requires(["authenticated", "Helpers"])
     @api.validate(
         json=Form,
         resp=Response(HTTP_200=OkayResponse, HTTP_400=ErrorMessage),