fix(assets): validate, deduplicate, and bound job_ids query param

- Validate each token is a valid UUID (normalizes case); invalid input returns 422
- Raise on non-string list items instead of silently dropping
- Raise on unexpected input types instead of forwarding raw value
- Deduplicate tokens to avoid redundant IN clause bind params
- Cap list at max_length=100 to prevent oversized IN clauses
This commit is contained in:
Matt Miller 2026-05-11 21:18:32 -07:00
parent cfbff1bc69
commit a7078705e3

View File

@ -1,4 +1,5 @@
import json import json
import uuid
from dataclasses import dataclass from dataclasses import dataclass
from typing import Any, Literal from typing import Any, Literal
@ -52,7 +53,7 @@ class ParsedUpload:
class ListAssetsQuery(BaseModel): class ListAssetsQuery(BaseModel):
include_tags: list[str] = Field(default_factory=list) include_tags: list[str] = Field(default_factory=list)
exclude_tags: list[str] = Field(default_factory=list) exclude_tags: list[str] = Field(default_factory=list)
job_ids: list[str] = Field(default_factory=list) job_ids: list[str] = Field(default_factory=list, max_length=100)
name_contains: str | None = None name_contains: str | None = None
# Accept either a JSON string (query param) or a dict # Accept either a JSON string (query param) or a dict
@ -72,14 +73,28 @@ class ListAssetsQuery(BaseModel):
if v is None: if v is None:
return [] return []
if isinstance(v, str): if isinstance(v, str):
return [t.strip() for t in v.split(",") if t.strip()] tokens = [t.strip() for t in v.split(",") if t.strip()]
if isinstance(v, list): elif isinstance(v, list):
out: list[str] = [] tokens = []
for item in v: for item in v:
if isinstance(item, str): if not isinstance(item, str):
out.extend([t.strip() for t in item.split(",") if t.strip()]) raise ValueError(
return out f"job_ids items must be strings, got {type(item).__name__}"
return v )
tokens.extend([t.strip() for t in item.split(",") if t.strip()])
else:
raise ValueError("job_ids must be a string or list of strings")
seen: set[str] = set()
out: list[str] = []
for t in tokens:
try:
normalized = str(uuid.UUID(t))
except ValueError:
raise ValueError(f"invalid UUID in job_ids: {t!r}")
if normalized not in seen:
seen.add(normalized)
out.append(normalized)
return out
@field_validator("include_tags", "exclude_tags", mode="before") @field_validator("include_tags", "exclude_tags", mode="before")
@classmethod @classmethod