fix(cloud): authenticate task requests before payload validation (#1134)

src/youtube_extension/backend/cloud_api_endpoints.py

@@ -18,7 +18,7 @@
 
 from fastapi import APIRouter, FastAPI, HTTPException, BackgroundTasks, Request, Header
 from fastapi.responses import JSONResponse
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, ValidationError
 
 # Import cloud services
 from ..services.cloud import (
@@ -148,9 +148,24 @@ async def process_video_cloud(
         # detail is a static string; error_msg (with the exception) is logged above only
         raise HTTPException(status_code=500, detail="Internal server error")
 
-@router.post("/api/v3/process-video-task")
+@router.post(
+    "/api/v3/process-video-task",
+    # The body is parsed manually after the Cloud Tasks header check, so the
+    # schema is declared here to keep it documented in OpenAPI.
+    openapi_extra={
+        "requestBody": {
+            "required": True,
+            "content": {
+                "application/json": {
+                    "schema": CloudTaskPayload.model_json_schema(
+                        ref_template="#/components/schemas/{model}"
+                    )
+                }
+            },
+        }
+    },
+)
 async def process_video_task_handler(
-    payload: CloudTaskPayload,
     request: Request,
     x_cloudtasks_taskname: Optional[str] = Header(None),
 ):
@@ -168,6 +183,13 @@ async def process_video_task_handler(
             detail="Only Cloud Tasks can call this endpoint"
         )
 
+    try:
+        payload = CloudTaskPayload.model_validate_json(await request.body())
+    except ValidationError as exc:
+        raise HTTPException(
+            status_code=422, detail=exc.errors(include_input=False)
+        ) from exc
+
     logger.info(
         f"📝 Processing Cloud Task: {x_cloudtasks_taskname} "
         f"(video_id={payload.video_id})"

tests/unit/test_cloud_routes.py

@@ -598,6 +598,13 @@ def test_process_video_task_no_header_returns_403(self):
         })
         assert response.status_code == 403
 
+    def test_process_video_task_no_header_malformed_payload_still_403(self):
+        client = self._build_app()
+        response = client.post("/api/v3/process-video-task", json={
+            "video_id": "auJzb1D-fag",
+        })
+        assert response.status_code == 403
+
     def test_process_video_task_with_header_success(self):
         state = self._make_state()
 
@@ -620,6 +627,33 @@ def test_process_video_task_with_header_success(self):
         assert data["success"] is True
         assert data["video_id"] == "auJzb1D-fag"
 
+    def test_process_video_task_with_header_malformed_payload_returns_422(self):
+        client = self._build_app()
+        response = client.post(
+            "/api/v3/process-video-task",
+            json={"video_id": "auJzb1D-fag"},
+            headers={"X-CloudTasks-TaskName": "task-abc-123"},
+        )
+        assert response.status_code == 422
+
+    def test_process_video_task_with_header_invalid_utf8_returns_422(self):
+        client = self._build_app()
+        response = client.post(
+            "/api/v3/process-video-task",
+            content=b"\xff",
+            headers={"X-CloudTasks-TaskName": "task-abc-123"},
+        )
+        assert response.status_code == 422
+        assert response.json()["detail"][0]["type"] == "json_invalid"
+
+    def test_process_video_task_documents_request_body_schema(self):
+        client = self._build_app()
+        spec = client.app.openapi()
+        operation = spec["paths"]["/api/v3/process-video-task"]["post"]
+        schema = operation["requestBody"]["content"]["application/json"]["schema"]
+        assert operation["requestBody"]["required"] is True
+        assert set(schema["required"]) == {"video_id", "video_url"}
+
     def test_process_video_task_exception(self):
         mock_processor = AsyncMock()
         mock_processor.process_video_sync = AsyncMock(side_effect=Exception("processing failed"))