Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 9 additions & 20 deletions app/query_validation.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,16 +12,17 @@
CANONICAL_BOOLEAN_QUERY_VALUES = {"true", "false"}


def _raise_bad_query_param(detail: str) -> None:
raise HTTPException(status_code=400, detail=detail)


def _query_param_values(request: Request, name: str) -> list[str]:
return request.query_params.getlist(name)


def _reject_control_char_value(value: str, name: str) -> None:
if contains_control_character(value):
raise HTTPException(
status_code=400,
detail=f"{name} must not contain control characters",
)
_raise_bad_query_param(f"{name} must not contain control characters")


def reject_control_char_query_param(request: Request, name: str) -> None:
Expand All @@ -35,30 +36,21 @@ def reject_noncanonical_int_query_param(request: Request, name: str) -> None:
for value in _query_param_values(request, name):
_reject_control_char_value(value, name)
if not CANONICAL_INTEGER_QUERY_RE.fullmatch(value):
raise HTTPException(
status_code=400,
detail=f"{name} must be a canonical positive integer",
)
_raise_bad_query_param(f"{name} must be a canonical positive integer")


def reject_noncanonical_bool_query_param(request: Request, name: str) -> None:
"""Reject boolean query aliases before application code trusts coerced values."""
for value in _query_param_values(request, name):
_reject_control_char_value(value, name)
if value not in CANONICAL_BOOLEAN_QUERY_VALUES:
raise HTTPException(
status_code=400,
detail=f"{name} must be true or false",
)
_raise_bad_query_param(f"{name} must be true or false")


def reject_repeated_query_param(request: Request, name: str) -> None:
"""Reject ambiguous scalar query parameters before FastAPI chooses one value."""
if len(_query_param_values(request, name)) > 1:
raise HTTPException(
status_code=400,
detail=f"{name} must be provided at most once",
)
_raise_bad_query_param(f"{name} must be provided at most once")


def reject_unsupported_query_params(
Expand All @@ -67,7 +59,4 @@ def reject_unsupported_query_params(
"""Reject query parameters that are valid elsewhere but unsupported here."""
for name in names:
if _query_param_values(request, name):
raise HTTPException(
status_code=400,
detail=f"{name} is not supported on {context}",
)
_raise_bad_query_param(f"{name} is not supported on {context}")
Loading