From 8a879703fd8d73f567385d3122e7020e0c7b4f4b Mon Sep 17 00:00:00 2001
From: Hayden Garvey <154503486+groupthinking@users.noreply.github.com>
Date: Tue, 4 Aug 2026 06:06:41 +0000
Subject: [PATCH] fix(cloud): authenticate task requests before payload
 validation

Port the canonical two-file artifact from PR #1132 (exact head
42939e0012ed2d3bab18de445655c98da0ad6980) onto current main:
/api/v3/process-video-task now checks the X-CloudTasks-TaskName gate
before parsing the body, so unauthorized malformed calls return 403
while authorized malformed calls keep strict 422 validation (including
invalid UTF-8 bodies via errors(include_input=False)). The request-body
schema stays documented in OpenAPI via openapi_extra.

Generated with [Linear](https://linear.app/myxstack/issue/GRV-198/execution-enforce-cloud-tasks-authentication-before-payload-validation#agent-session-95a9c398)

Co-authored-by: linear-code[bot] <222613912+linear-code[bot]@users.noreply.github.com>
---
 .../backend/cloud_api_endpoints.py            | 28 +++++++++++++--
 tests/unit/test_cloud_routes.py               | 34 +++++++++++++++++++
 2 files changed, 59 insertions(+), 3 deletions(-)

diff --git a/src/youtube_extension/backend/cloud_api_endpoints.py b/src/youtube_extension/backend/cloud_api_endpoints.py
index b1999ceaa..fede7a724 100644
--- a/src/youtube_extension/backend/cloud_api_endpoints.py
+++ b/src/youtube_extension/backend/cloud_api_endpoints.py
@@ -14,7 +14,7 @@
 from typing import Any, Optional
 
 from fastapi import APIRouter, BackgroundTasks, FastAPI, Header, HTTPException, Request
-from pydantic import BaseModel, Field
+from pydantic import BaseModel, Field, ValidationError
 
 # Import cloud services
 from ..services.cloud import (
@@ -143,9 +143,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),
 ):
@@ -163,6 +178,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})"
diff --git a/tests/unit/test_cloud_routes.py b/tests/unit/test_cloud_routes.py
index 712cd678b..4231a5f2c 100644
--- a/tests/unit/test_cloud_routes.py
+++ b/tests/unit/test_cloud_routes.py
@@ -604,6 +604,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()
 
@@ -626,6 +633,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"))