aboutsummaryrefslogtreecommitdiffstats
path: root/backend
diff options
context:
space:
mode:
Diffstat (limited to 'backend')
-rw-r--r--backend/discord.py5
-rw-r--r--backend/models/form.py3
-rw-r--r--backend/routes/forms/form.py71
-rw-r--r--backend/routes/forms/index.py6
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),
generated by cgit v1.2.3 (git 2.39.1) at 2025-10-31 12:32:16 +0000