1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
|
import typing as t
import httpx
from pydantic import BaseModel, Field, constr, root_validator, validator
from pydantic.error_wrappers import ErrorWrapper, ValidationError
from backend.constants import DISCORD_GUILD, FormFeatures, WebHook
from .question import Question
PUBLIC_FIELDS = [
"id",
"features",
"questions",
"name",
"description",
"submitted_text",
"discord_role"
]
class _WebHook(BaseModel):
"""Schema model of discord webhooks."""
url: str
message: t.Optional[str]
@validator("url")
def validate_url(cls, url: str) -> str:
"""Validates URL parameter."""
if "discord.com/api/webhooks/" not in url:
raise ValueError("URL must be a discord webhook.")
return url
class Form(BaseModel):
"""Schema model for form."""
id: constr(to_lower=True) = Field(alias="_id")
features: list[str]
questions: list[Question]
name: str
description: str
submitted_text: t.Optional[str] = None
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
@validator("features")
def validate_features(cls, value: list[str]) -> t.Optional[list[str]]:
"""Validates is all features in allowed list."""
# Uppercase everything to avoid mixed case in DB
value = [v.upper() for v in value]
allowed_values = [v.value for v in FormFeatures.__members__.values()]
if any(v not in allowed_values for v in value):
raise ValueError("Form features list contains one or more invalid values.")
if FormFeatures.REQUIRES_LOGIN.value not in value:
if FormFeatures.COLLECT_EMAIL.value in value:
raise ValueError(
"COLLECT_EMAIL feature require REQUIRES_LOGIN feature."
)
if FormFeatures.ASSIGN_ROLE.value in value:
raise ValueError("ASSIGN_ROLE feature require REQUIRES_LOGIN feature.")
return value
@validator("response_readers", "editors")
def validate_role_scoping(cls, value: t.Optional[list[str]]) -> t.Optional[list[str]]:
"""Ensure special role based permissions aren't granted to the @everyone role."""
if value and str(DISCORD_GUILD) in value:
raise ValueError("You can not add the everyone role as an access scope.")
return value
@root_validator
def validate_role(cls, values: dict[str, t.Any]) -> t.Optional[dict[str, t.Any]]:
"""Validates does Discord role provided when flag provided."""
if (
FormFeatures.ASSIGN_ROLE.value in values.get("features", [])
and not values.get("discord_role")
):
raise ValueError(
"discord_role field is required when ASSIGN_ROLE flag is provided."
)
return values
def dict(self, admin: bool = True, **kwargs) -> dict[str, t.Any]:
"""Wrapper for original function to exclude private data for public access."""
data = super().dict(**kwargs)
returned_data = {}
if not admin:
for field in PUBLIC_FIELDS:
if field == "id" and kwargs.get("by_alias"):
fetch_field = "_id"
else:
fetch_field = field
returned_data[field] = data[fetch_field]
else:
returned_data = data
return returned_data
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() -> t.Optional[str]:
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 = (
WebHook.__name__.lower(),
WebHook.URL.value
)
return ValidationError([ErrorWrapper(e, loc=loc)], _WebHook)
|
