fix(06): CR-07 validate event_type against allowlist before LIKE filter

Add _VALID_EVENT_PREFIXES frozenset and validate event_type at all three
filter sites in audit.py (_build_filtered_query, _build_filtered_query
_with_handles, and the inline count query in list_audit_log). Invalid
values return HTTP 422. Also change like() pattern from prefix% to
prefix.% to enforce true prefix-match semantics.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
curo1305
2026-06-04 23:15:48 +02:00
co-authored by Claude Sonnet 4.6
parent a37a91071c
commit 10970d9557
+11 -3
View File
@@ -43,6 +43,8 @@ from storage.minio_backend import MinIOBackend
router = APIRouter(prefix="/api/admin", tags=["audit"])
_VALID_EVENT_PREFIXES = frozenset({"auth", "document", "folder", "share", "admin", "cloud"})
# ── Safe response helpers ─────────────────────────────────────────────────────
@@ -121,7 +123,9 @@ def _build_filtered_query(
if user_id is not None:
q = q.where(AuditLog.user_id == user_id)
if event_type is not None:
q = q.where(AuditLog.event_type.like(f"{event_type}%"))
if event_type not in _VALID_EVENT_PREFIXES:
raise HTTPException(status_code=422, detail="Invalid event_type prefix")
q = q.where(AuditLog.event_type.like(f"{event_type}.%"))
return q
@@ -161,7 +165,9 @@ def _build_filtered_query_with_handles(
if user_uuid is not None:
q = q.where(AuditLog.user_id == user_uuid)
if event_type is not None:
q = q.where(AuditLog.event_type.like(f"{event_type}%"))
if event_type not in _VALID_EVENT_PREFIXES:
raise HTTPException(status_code=422, detail="Invalid event_type prefix")
q = q.where(AuditLog.event_type.like(f"{event_type}.%"))
return q
@@ -288,7 +294,9 @@ async def list_audit_log(
if user_uuid is not None:
count_q = count_q.where(AuditLog.user_id == user_uuid)
if event_type is not None:
count_q = count_q.where(AuditLog.event_type.like(f"{event_type}%"))
if event_type not in _VALID_EVENT_PREFIXES:
raise HTTPException(status_code=422, detail="Invalid event_type prefix")
count_q = count_q.where(AuditLog.event_type.like(f"{event_type}.%"))
count_result = await session.execute(count_q)
total = count_result.scalar_one()