Validate task status query parameter

backend/secuscan/routes.py

@@ -108,7 +108,7 @@ def build_report_filename(task: Dict[str, Any], extension: str) -> str:
     TaskCreateRequest, TaskResponse, TaskResult,
     PluginListResponse, ErrorResponse, BulkDeleteRequest,
     NotificationRuleCreate, NotificationRuleUpdate,
-    NotificationChannelType,
+    NotificationChannelType, TaskStatus,
 )
 from .config import settings
 from .database import get_db
@@ -929,6 +929,15 @@ async def list_tasks(
         where_clauses.append("plugin_id = ?")
         params.append(plugin_id)
     if status:
+        try:
+            status = TaskStatus(status).value
+        except ValueError:
+            allowed_values = ", ".join([s.value for s in TaskStatus])
+            raise HTTPException(
+                status_code=400,
+                detail=f"Invalid task status '{status}'. Allowed values: {allowed_values}"
+            )
+
         where_clauses.append("status = ?")
         params.append(status)
 

testing/backend/test_task_pagination.py

@@ -81,6 +81,23 @@ def test_invalid_pagination_is_rejected(self, test_client, qs):
         response = test_client.get(f"/api/v1/tasks?{qs}")
         assert response.status_code == 422
 
+    def test_status_filter_valid(self, test_client):
+        response = test_client.get("/api/v1/tasks?status=completed")
+        assert response.status_code == 200
+
+        data = response.json()
+        assert "tasks" in data
+        assert all(task["status"] == "completed" for task in data["tasks"])
+
+    def test_status_filter_invalid(self, test_client):
+        response = test_client.get("/api/v1/tasks?status=invalid-status")
+        assert response.status_code == 400
+
+        data = response.json()
+        assert data["detail"] == (
+            "Invalid task status 'invalid-status'. Allowed values: queued, running, completed, failed, cancelled"
+        )
+
     def test_first_page_previous_is_null(self, test_client):
         """Test that previous is None on first page"""
         response = test_client.get("/api/v1/tasks?page=1&per_page=10")
@@ -107,7 +124,7 @@ def test_next_url_preserves_filters(self, test_client):
     def test_next_url_encodes_filtered_pagination_params(self, test_client):
         """Test that filtered pagination links URL-encode query values."""
         plugin_id = "web scanner/alpha"
-        status = "queued & reviewed"
+        status = "queued"
         asyncio.run(
             _insert_task("encoded-filter-1", plugin_id, status, "2026-06-02T10:00:00")
         )
@@ -129,5 +146,5 @@ def test_next_url_encodes_filtered_pagination_params(self, test_client):
         next_url = response.json()["pagination"]["next"]
         assert next_url == (
             "/api/v1/tasks?page=2&per_page=1&"
-            "plugin_id=web+scanner%2Falpha&status=queued+%26+reviewed"
+            "plugin_id=web+scanner%2Falpha&status=queued"
         )