From 5c23d6e237c4cb560f3f3d94c88e7cc3dca36932 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 8 Sep 2026 22:48:39 +0000 Subject: [PATCH 1/2] Initial plan From 21cfb34332b22787e65f2a78b282538095f875af Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 8 Sep 2026 22:54:55 +0000 Subject: [PATCH 2/2] fix(security): harden cloud callback dispatch Co-authored-by: groupthinking <154503486+groupthinking@users.noreply.github.com> --- .../backend/cloud_api_endpoints.py | 183 +++++++++++++++--- tests/unit/test_cloud_routes.py | 110 +++++++++++ 2 files changed, 269 insertions(+), 24 deletions(-) diff --git a/src/youtube_extension/backend/cloud_api_endpoints.py b/src/youtube_extension/backend/cloud_api_endpoints.py index fede7a724..580c0480e 100644 --- a/src/youtube_extension/backend/cloud_api_endpoints.py +++ b/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 @@ async def process_video_cloud( - 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 @@ async def process_video_cloud( ) 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 @@ async def process_video_task_handler( ) 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 @@ async def process_video_task_handler( 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 @@ async def process_video_task_handler( } 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 @@ async def process_video_task_handler( 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") diff --git a/tests/unit/test_cloud_routes.py b/tests/unit/test_cloud_routes.py index 4231a5f2c..14b6d5c81 100644 --- a/tests/unit/test_cloud_routes.py +++ b/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