diff options
| -rw-r--r-- | backend/discord.py | 25 | ||||
| -rw-r--r-- | backend/routes/forms/form.py | 24 | ||||
| -rw-r--r-- | backend/routes/forms/response.py | 9 | ||||
| -rw-r--r-- | backend/routes/forms/responses.py | 8 |
4 files changed, 26 insertions, 40 deletions
diff --git a/backend/discord.py b/backend/discord.py index f972f5f..856e878 100644 --- a/backend/discord.py +++ b/backend/discord.py @@ -7,6 +7,7 @@ import typing import httpx import starlette.requests from pymongo.database import Database +from starlette import exceptions from backend import constants, models @@ -151,22 +152,26 @@ async def get_member( return member -class FormNotFoundError(Exception): +class FormNotFoundError(exceptions.HTTPException): """The requested form was not found.""" +class UnauthorizedError(exceptions.HTTPException): + """You are not authorized to use this resource.""" + + async def _verify_access_helper( form_id: str, request: starlette.requests.Request, attribute: str -) -> bool: +) -> None: """A low level helper to validate access to a form resource based on the user's scopes.""" # Short circuit all resources for admins if "admin" in request.auth.scopes: - return True + return form = await request.state.db.forms.find_one({"id": form_id}) if not form: - raise FormNotFoundError() + raise FormNotFoundError(status_code=404) form = models.Form(**form) @@ -178,16 +183,16 @@ async def _verify_access_helper( role = models.DiscordRole(**json.loads(role["data"])) if role.name in request.auth.scopes: - return True + return - return False + raise UnauthorizedError(status_code=401) -async def verify_response_access(form_id: str, request: starlette.requests.Request) -> bool: +async def verify_response_access(form_id: str, request: starlette.requests.Request) -> None: """Ensure the user can access responses on the requested resource.""" - return await _verify_access_helper(form_id, request, "response_readers") + await _verify_access_helper(form_id, request, "response_readers") -async def verify_edit_access(form_id: str, request: starlette.requests.Request) -> bool: +async def verify_edit_access(form_id: str, request: starlette.requests.Request) -> None: """Ensure the user can view and modify the requested resource.""" - return await _verify_access_helper(form_id, request, "editors") + await _verify_access_helper(form_id, request, "editors") diff --git a/backend/routes/forms/form.py b/backend/routes/forms/form.py index 15ff9a6..b6c6f1d 100644 --- a/backend/routes/forms/form.py +++ b/backend/routes/forms/form.py @@ -17,7 +17,6 @@ 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) @@ -37,13 +36,15 @@ class SingleForm(Route): form_id = request.path_params["form_id"].lower() try: - admin = await discord.verify_edit_access(form_id, request) + await discord.verify_edit_access(form_id, request) + admin = True 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 + raise + except discord.UnauthorizedError: + admin = False filters = { "_id": form_id @@ -63,7 +64,6 @@ class SingleForm(Route): resp=Response( HTTP_200=OkayResponse, HTTP_400=ErrorMessage, - HTTP_401=ErrorMessage, HTTP_404=ErrorMessage, ), tags=["forms"] @@ -76,12 +76,7 @@ class SingleForm(Route): return JSONResponse("Expected a JSON body.", 400) 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 discord.verify_edit_access(form_id, request) if raw_form := await request.state.db.forms.find_one({"id": form_id}): if "_id" in data or "id" in data: @@ -116,12 +111,7 @@ class SingleForm(Route): async def delete(self, request: Request) -> JSONResponse: """Deletes form by 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 + await discord.verify_edit_access(form_id, request) await request.state.db.forms.delete_one({"_id": form_id}) await request.state.db.responses.delete_many({"form_id": form_id}) diff --git a/backend/routes/forms/response.py b/backend/routes/forms/response.py index fbf8e99..565701f 100644 --- a/backend/routes/forms/response.py +++ b/backend/routes/forms/response.py @@ -21,18 +21,13 @@ class Response(Route): @requires(["authenticated"]) @api.validate( - resp=RouteResponse(HTTP_200=FormResponse, HTTP_401=ErrorMessage, HTTP_404=ErrorMessage), + resp=RouteResponse(HTTP_200=FormResponse, HTTP_404=ErrorMessage), tags=["forms", "responses"] ) async def get(self, request: Request) -> JSONResponse: """Return a single form response by ID.""" form_id = request.path_params["form_id"] - - try: - if not await discord.verify_response_access(form_id, request): - return JSONResponse({"error": "unauthorized"}, status_code=401) - except discord.FormNotFoundError: - return JSONResponse({"error": "form_not_found"}, status_code=404) + await discord.verify_response_access(form_id, request) if raw_response := await request.state.db.responses.find_one( { diff --git a/backend/routes/forms/responses.py b/backend/routes/forms/responses.py index 1c8ebe3..818ebce 100644 --- a/backend/routes/forms/responses.py +++ b/backend/routes/forms/responses.py @@ -27,17 +27,13 @@ class Responses(Route): @requires(["authenticated"]) @api.validate( - resp=Response(HTTP_200=ResponseList, HTTP_401=ErrorMessage, HTTP_404=ErrorMessage), + resp=Response(HTTP_200=ResponseList), tags=["forms", "responses"] ) async def get(self, request: Request) -> JSONResponse: """Returns all form responses by form ID.""" form_id = request.path_params["form_id"] - try: - if not await discord.verify_response_access(form_id, request): - return JSONResponse({"error": "unauthorized"}, 401) - except discord.FormNotFoundError: - return JSONResponse({"error": "not_found"}, 404) + await discord.verify_response_access(form_id, request) cursor = request.state.db.responses.find( {"form_id": form_id} |