fix(security): harden Cloud Task callback dispatch

src/youtube_extension/backend/cloud_api_endpoints.py

@@ -9,9 +9,13 @@
 - Cloud Tasks for async processing
 """
 
+import asyncio
+import ipaddress
 import logging
+import socket
 from datetime import datetime, timezone
-from typing import Any, Optional
+from typing import Any, Optional, Union
+from urllib.parse import urlparse
 
 from fastapi import APIRouter, BackgroundTasks, FastAPI, Header, HTTPException, Request
 from pydantic import BaseModel, Field, ValidationError
@@ -30,6 +34,81 @@
 router = APIRouter()
 
 
+_BLOCKED_CALLBACK_HOSTS = frozenset(
+    {"localhost", "metadata", "metadata.google.internal"}
+)
+_MAX_CALLBACK_ADDRESS_ATTEMPTS = 3
+_CALLBACK_ATTEMPT_TIMEOUT = 10.0
+_CALLBACK_TOTAL_TIMEOUT = 15.0
+
+
+def _sanitize_log_value(value: Any) -> str:
+    """Strip CR/LF from untrusted values before logging."""
+    return str(value).replace("\r", "").replace("\n", "")
+
+
+def _is_blocked_ip(ip: Union[ipaddress.IPv4Address, ipaddress.IPv6Address]) -> bool:
+    """Return True unless the address is a globally routable public address."""
+    return ip.is_multicast or getattr(ip, "is_site_local", False) or not ip.is_global
+
+
+def _is_safe_callback_url(url: str, *, resolve: bool = True) -> bool:
+    """Return True only when a callback URL resolves exclusively to public IPs."""
+    return _validated_callback_addresses(url, resolve=resolve) is not None
+
+
+def _validated_callback_addresses(
+    url: str, *, resolve: bool = True
+) -> Optional[tuple[str, ...]]:
+    """Validate a callback and return the exact public addresses it resolved to."""
+    try:
+        parsed = urlparse(url)
+        port = parsed.port
+    except ValueError:
+        return None
+
+    hostname = parsed.hostname
+    if parsed.scheme not in ("http", "https") or not hostname:
+        return None
+    if hostname.rstrip(".").lower() in _BLOCKED_CALLBACK_HOSTS:
+        return None
+
+    try:
+        ip = ipaddress.ip_address(hostname)
+    except ValueError:
+        ip = None
+    if ip is not None:
+        return None if _is_blocked_ip(ip) else (str(ip),)
+    if not resolve:
+        return ()
+
+    try:
+        addrinfos = socket.getaddrinfo(
+            hostname,
+            port or (443 if parsed.scheme == "https" else 80),
+            type=socket.SOCK_STREAM,
+        )
+    except (socket.gaierror, UnicodeError, ValueError):
+        return None
+
+    addresses = []
+    for info in addrinfos:
+        if info[0] not in (socket.AF_INET, socket.AF_INET6):
+            continue
+        resolved = str(info[4][0]).split("%", 1)[0]
+        try:
+            resolved_ip = ipaddress.ip_address(resolved)
+        except ValueError:
+            return None
+        if _is_blocked_ip(resolved_ip):
+            return None
+        normalized = str(resolved_ip)
+        if normalized not in addresses:
+            addresses.append(normalized)
+
+    return tuple(addresses) if addresses else None
+
+
 
 # Pydantic models for API requests/responses
 class CloudVideoProcessingRequest(BaseModel):
@@ -91,13 +170,20 @@
     - State tracked in Firestore
     - AI reasoning via Vertex AI Agent Builder
     """
+    if request.callback_url and not _is_safe_callback_url(
+        request.callback_url, resolve=False
+    ):
+        raise HTTPException(status_code=400, detail="Invalid callback_url")
+
     try:
         processor = get_cloud_video_processor()
         video_id = processor._extract_video_id(request.video_url)
 
         logger.info(
-            f"🎬 Cloud processing request: {request.video_url} "
-            f"(async={request.async_processing}, priority={request.priority})"
+            "🎬 Cloud processing request: %s (async=%s, priority=%s)",
+            _sanitize_log_value(request.video_url),
+            request.async_processing,
+            request.priority,
         )
 
         if request.async_processing:
@@ -137,8 +223,9 @@
             )
 
     except Exception as e:
-        error_msg = f"Cloud processing failed: {str(e)}"
-        logger.error(error_msg, exc_info=True)
+        logger.error(
+            "Cloud processing failed: %s", _sanitize_log_value(e), exc_info=True
+        )
 
         # detail is a static string; error_msg (with the exception) is logged above only
         raise HTTPException(status_code=500, detail="Internal server error")
@@ -186,8 +273,9 @@
         ) from exc
 
     logger.info(
-        f"📝 Processing Cloud Task: {x_cloudtasks_taskname} "
-        f"(video_id={payload.video_id})"
+        "📝 Processing Cloud Task: %s (video_id=%s)",
+        _sanitize_log_value(x_cloudtasks_taskname),
+        _sanitize_log_value(payload.video_id),
     )
 
     try:
@@ -199,23 +287,64 @@
             force_refresh=False,
         )
 
-        # Call callback URL if provided
+        # Resolve and validate off-loop, then pin the connection to its address.
         if payload.callback_url and result.success:
-            try:
-                import httpx
-                async with httpx.AsyncClient() as client:
-                    await client.post(
-                        payload.callback_url,
-                        json={
-                            'video_id': result.video_id,
-                            'status': 'completed',
-                            'processing_time': result.processing_time,
-                        },
-                        timeout=10.0
+            callback_addresses = await asyncio.to_thread(
+                _validated_callback_addresses, payload.callback_url
+            )
+            if not callback_addresses:
+                logger.warning(
+                    "⚠️ Refusing to call unsafe callback URL: %s",
+                    _sanitize_log_value(payload.callback_url),
+                )
+            else:
+                try:
+                    import httpx
+
+                    callback_url = httpx.URL(payload.callback_url)
+                    host_header = callback_url.netloc.decode("ascii")
+                    deadline = (
+                        asyncio.get_running_loop().time() + _CALLBACK_TOTAL_TIMEOUT
                     )
-                logger.info(f"✅ Callback sent to {payload.callback_url}")
-            except Exception as e:
-                logger.warning(f"⚠️ Callback failed: {e}")
+                    sent = False
+                    last_connect_error: Optional[Exception] = None
+                    async with httpx.AsyncClient(follow_redirects=False) as client:
+                        for address in callback_addresses[
+                            :_MAX_CALLBACK_ADDRESS_ATTEMPTS
+                        ]:
+                            remaining = deadline - asyncio.get_running_loop().time()
+                            if remaining <= 0:
+                                break
+                            try:
+                                await client.post(
+                                    callback_url.copy_with(host=address),
+                                    json={
+                                        "video_id": result.video_id,
+                                        "status": "completed",
+                                        "processing_time": result.processing_time,
+                                    },
+                                    headers={"Host": host_header},
+                                    extensions={"sni_hostname": callback_url.host},
+                                    timeout=min(_CALLBACK_ATTEMPT_TIMEOUT, remaining),
+                                )
+                                sent = True
+                                break
+                            except (httpx.ConnectError, httpx.ConnectTimeout) as exc:
+                                last_connect_error = exc
+                    if sent:
+                        logger.info(
+                            "✅ Callback sent to %s",
+                            _sanitize_log_value(payload.callback_url),
+                        )
+                    elif last_connect_error is not None:
+                        raise last_connect_error
+                    else:
+                        logger.warning(
+                            "⚠️ Callback abandoned (attempt/deadline bound) for %s",
+                            _sanitize_log_value(payload.callback_url),
+                        )
+                except Exception as e:
+                    logger.warning("⚠️ Callback failed: %s", _sanitize_log_value(e))
 
         return {
             "success": result.success,
@@ -225,7 +354,9 @@
         }
 
     except Exception as e:
-        logger.error(f"Task processing failed: {e}", exc_info=True)
+        logger.error(
+            "Task processing failed: %s", _sanitize_log_value(e), exc_info=True
+        )
 
         # Update state with a static error message; raw exception is logged above only
         try:
@@ -236,7 +367,11 @@
                 error_message="Task processing failed"
             )
         except Exception as state_error:
-            logger.error(f"Failed to update error state: {state_error}")
+            logger.error(
+                "Failed to update error state: %s",
+                _sanitize_log_value(state_error),
+                exc_info=True,
+            )
 
         raise HTTPException(status_code=500, detail="Internal server error")
 

tests/unit/test_cloud_routes.py

@@ -1502,3 +1502,113 @@ def test_generate_dashboard_url_success_after_prior_error(self):
         assert ok_response.json() == {
             "embed_url": "https://looker.example.com/embed/dashboards/2?sig=def"
         }
+
+
+import youtube_extension.backend.cloud_api_endpoints as _cae
+
+
+class TestCallbackUrlSafety:
+    @pytest.mark.parametrize(
+        "value,expected",
+        [("a\r\nb", "ab"), ("line1\nline2", "line1line2"), (123, "123")],
+    )
+    def test_sanitize_log_value_strips_crlf(self, value, expected):
+        assert _cae._sanitize_log_value(value) == expected
+
+    @pytest.mark.parametrize(
+        "url",
+        [
+            "http://127.0.0.1/x",
+            "http://100.64.0.1/x",
+            "http://[fec0::1]/x",
+            "https://metadata.google.internal./x",
+            "ftp://example.com/x",
+        ],
+    )
+    def test_rejects_non_public_callback_destinations(self, url):
+        assert _cae._is_safe_callback_url(url, resolve=False) is False
+
+    def test_rejects_mixed_dns_results(self):
+        with patch.object(
+            _cae.socket,
+            "getaddrinfo",
+            return_value=[
+                (2, 1, 6, "", ("93.184.216.34", 0)),
+                (2, 1, 6, "", ("127.0.0.1", 0)),
+            ],
+        ):
+            assert _cae._is_safe_callback_url("https://mixed.example/x") is False
+
+
+class TestCallbackDispatch:
+    @staticmethod
+    def _state():
+        state = MagicMock()
+        state.video_id = "auJzb1D-fag"
+        state.processing_time = 1.0
+        state.success = True
+        return state
+
+    def test_pins_hostname_callback_to_validated_address(self):
+        mock_client = AsyncMock()
+        mock_client.__aenter__ = AsyncMock(return_value=mock_client)
+        mock_client.__aexit__ = AsyncMock(return_value=False)
+        mock_client.post = AsyncMock()
+        mock_processor = AsyncMock()
+        mock_processor.process_video_sync = AsyncMock(return_value=self._state())
+
+        with patch.object(
+            _cae.socket,
+            "getaddrinfo",
+            return_value=[
+                (2, _cae.socket.SOCK_STREAM, 6, "", ("93.184.216.34", 443))
+            ],
+        ), patch(
+            "youtube_extension.backend.cloud_api_endpoints.get_cloud_video_processor",
+            return_value=mock_processor,
+        ), patch("httpx.AsyncClient", return_value=mock_client) as client:
+            response = TestClient(_make_cloud_api_app()).post(
+                "/api/v3/process-video-task",
+                json={
+                    "video_id": "auJzb1D-fag",
+                    "video_url": "https://www.youtube.com/watch?v=auJzb1D-fag",
+                    "callback_url": "https://callbacks.example:8443/cb?job=1",
+                },
+                headers={"X-CloudTasks-TaskName": "task-1"},
+            )
+
+        assert response.status_code == 200
+        client.assert_called_once_with(follow_redirects=False)
+        url = mock_client.post.await_args.args[0]
+        kwargs = mock_client.post.await_args.kwargs
+        assert str(url) == "https://93.184.216.34:8443/cb?job=1"
+        assert kwargs["headers"]["Host"] == "callbacks.example:8443"
+        assert kwargs["extensions"]["sni_hostname"] == "callbacks.example"
+
+    def test_retries_only_untransmitted_connection_failures_up_to_three_addresses(self):
+        mock_client = AsyncMock()
+        mock_client.__aenter__ = AsyncMock(return_value=mock_client)
+        mock_client.__aexit__ = AsyncMock(return_value=False)
+        mock_client.post = AsyncMock(side_effect=_httpx_real.ConnectError("down"))
+        mock_processor = AsyncMock()
+        mock_processor.process_video_sync = AsyncMock(return_value=self._state())
+
+        with patch(
+            "youtube_extension.backend.cloud_api_endpoints._validated_callback_addresses",
+            return_value=tuple(f"8.8.8.{i}" for i in range(1, 6)),
+        ), patch(
+            "youtube_extension.backend.cloud_api_endpoints.get_cloud_video_processor",
+            return_value=mock_processor,
+        ), patch("httpx.AsyncClient", return_value=mock_client):
+            response = TestClient(_make_cloud_api_app()).post(
+                "/api/v3/process-video-task",
+                json={
+                    "video_id": "auJzb1D-fag",
+                    "video_url": "https://www.youtube.com/watch?v=auJzb1D-fag",
+                    "callback_url": "https://callbacks.example/cb",
+                },
+                headers={"X-CloudTasks-TaskName": "task-1"},
+            )
+
+        assert response.status_code == 200
+        assert mock_client.post.await_count == 3