diff options
| -rw-r--r-- | backend/constants.py | 9 | ||||
| -rw-r--r-- | backend/models/form.py | 87 | ||||
| -rw-r--r-- | backend/routes/forms/index.py | 18 |
3 files changed, 81 insertions, 33 deletions
diff --git a/backend/constants.py b/backend/constants.py index f5e4304..bfcf261 100644 --- a/backend/constants.py +++ b/backend/constants.py @@ -62,3 +62,12 @@ class FormFeatures(Enum): COLLECT_EMAIL = "COLLECT_EMAIL" DISABLE_ANTISPAM = "DISABLE_ANTISPAM" WEBHOOK_ENABLED = "WEBHOOK_ENABLED" + + +class WebHook(Enum): + URL = "url" + MESSAGE = "message" + + +class Meta(Enum): + WEB_HOOK = WebHook diff --git a/backend/models/form.py b/backend/models/form.py index 9a3a8a2..96362d4 100644 --- a/backend/models/form.py +++ b/backend/models/form.py @@ -2,8 +2,9 @@ import typing as t import httpx from pydantic import BaseModel, Field, validator +from pydantic.error_wrappers import ErrorWrapper, ValidationError -from backend.constants import FormFeatures +from backend.constants import FormFeatures, Meta, WebHook from .question import Question PUBLIC_FIELDS = ["id", "features", "questions", "name", "description"] @@ -16,40 +17,10 @@ class _WebHook(BaseModel): @validator("url") def validate_url(cls, url: str) -> str: - """Checks if discord webhook urls are valid.""" - if not isinstance(url, str): - raise ValueError("Webhook URL must be a string.") - + """Validates URL parameter.""" if "discord.com/api/webhooks/" not in url: raise ValueError("URL must be a discord webhook.") - # Attempt to connect to URL - try: - httpx.get(url).raise_for_status() - - except httpx.RequestError as e: - # Catch exceptions in request format - raise ValueError( - f"Encountered error while trying to connect to url: `{e}`" - ) - - except httpx.HTTPStatusError as e: - # Catch exceptions in response - status = e.response.status_code - - if status == 401: - raise ValueError( - "Could not authenticate with target. Please check the webhook url." - ) - elif status == 404: - raise ValueError( - "Target could not find webhook url. Please check the webhook url." - ) - else: - raise ValueError( - f"Unknown error ({status}) while connecting to target: {e}" - ) - return url @@ -107,3 +78,55 @@ class Form(BaseModel): class FormList(BaseModel): __root__: t.List[Form] + + +async def validate_hook_url(url: str) -> t.Optional[ValidationError]: + """Validator for discord webhook urls.""" + async def validate(): + if not isinstance(url, str): + raise ValueError("Webhook URL must be a string.") + + if "discord.com/api/webhooks/" not in url: + raise ValueError("URL must be a discord webhook.") + + try: + async with httpx.AsyncClient() as client: + response = await client.get(url) + response.raise_for_status() + + except httpx.RequestError as error: + # Catch exceptions in request format + raise ValueError( + f"Encountered error while trying to connect to url: `{error}`" + ) + + except httpx.HTTPStatusError as error: + # Catch exceptions in response + status = error.response.status_code + + if status == 401: + raise ValueError( + "Could not authenticate with target. Please check the webhook url." + ) + elif status == 404: + raise ValueError( + "Target could not find webhook url. Please check the webhook url." + ) + else: + raise ValueError( + f"Unknown error ({status}) while connecting to target: {error}" + ) + + return url + + # Validate, and return errors, if any + try: + await validate() + except Exception as e: + loc = ( + Meta.__name__.lower(), + WebHook.__name__.lower(), + WebHook.URL.value + ) + + return ValidationError([ErrorWrapper(e, loc=loc)], _WebHook) diff --git a/backend/routes/forms/index.py b/backend/routes/forms/index.py index d1373e4..0e1dee8 100644 --- a/backend/routes/forms/index.py +++ b/backend/routes/forms/index.py @@ -6,8 +6,10 @@ from starlette.authentication import requires from starlette.requests import Request from starlette.responses import JSONResponse -from backend.route import Route +from backend.constants import Meta, WebHook from backend.models import Form, FormList +from backend.models.form import validate_hook_url +from backend.route import Route from backend.validation import ErrorMessage, OkayResponse, api @@ -46,6 +48,20 @@ class FormsList(Route): """Create a new form.""" form_data = await request.json() + # Verify Webhook + try: + # Get url from request + path = (Meta.__name__.lower(), WebHook.__name__.lower(), WebHook.URL.value) + url = form_data[path[0]][path[1]][path[2]] + + # Validate URL + validation = await validate_hook_url(url) + if validation: + return JSONResponse(validation.errors(), status_code=422) + + except KeyError: + pass + form = Form(**form_data) if await request.state.db.forms.find_one({"_id": form.id}): |