diff --git a/src/youtube_extension/backend/cloud_api_endpoints.py b/src/youtube_extension/backend/cloud_api_endpoints.py index b1999ceaa..7b93a8ab9 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 @@ -30,6 +34,126 @@ router = APIRouter() +# Well-known internal hostnames that must never receive an outbound callback. +_BLOCKED_CALLBACK_HOSTS = frozenset( + {"localhost", "metadata", "metadata.google.internal"} +) + + +def _sanitize_log_value(value: Any) -> str: + """Strip CR/LF from untrusted values before logging to prevent log injection.""" + 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. + + Rejecting every non-global destination (rather than enumerating unsafe + ranges) also blocks addresses that Python reports as neither private nor + global — e.g. shared CGNAT space (``100.64.0.0/10``) and deprecated IPv6 + site-local (``fec0::/10``) — which the enumerated form let through. + Multicast and deprecated IPv6 site-local (``fec0::/10``, which some Python + versions still report as global) are rejected explicitly. + """ + return ( + ip.is_multicast + or getattr(ip, "is_site_local", False) + or not ip.is_global + ) + + +# Bounds for outbound callback dispatch: a hostname can resolve to many public +# addresses, so cap how many are attempted and the total wall-clock spent so a +# black-holing DNS answer cannot tie up a task worker far beyond one timeout. +_MAX_CALLBACK_ADDRESS_ATTEMPTS = 3 +_CALLBACK_ATTEMPT_TIMEOUT = 10.0 +_CALLBACK_TOTAL_TIMEOUT = 15.0 + + +def _is_safe_callback_url(url: str, *, resolve: bool = True) -> bool: + """Return True only for callback URLs safe for the server to POST to. + + Mitigates SSRF against the user-supplied Cloud Task callback: + * requires an http(s) scheme with a hostname; + * rejects well-known internal hostnames (trailing-dot / case normalized); + * rejects loopback / private / link-local / reserved / multicast / + unspecified IP literals; + * when ``resolve`` is True, resolves the hostname via DNS and rejects if + ANY resolved address is blocked — this defeats obfuscated IPv4 + encodings (decimal/hex/octal) and DNS names that map to internal + addresses. + + ``resolve=False`` runs only the cheap, network-free checks; it is used for + early request-time validation, while the full resolving check is run off + the event loop immediately before the outbound request. + """ + + 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. + + A non-``None`` empty tuple means the URL passed the network-free validation. + A resolving validation returns at least one numeric address; callers must use + one of those addresses as the connection target instead of resolving the + attacker-controlled hostname again. + """ + 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: + # Non-literal host clears the cheap gate; it is fully resolved and + # re-validated before any outbound request is actually made. + return () + + try: + addrinfos = socket.getaddrinfo( + hostname, + port or (443 if parsed.scheme == "https" else 80), + type=socket.SOCK_STREAM, + ) + except (socket.gaierror, UnicodeError, ValueError): + # Unresolvable hostname — treat as unsafe. + 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] # drop IPv6 scope/zone id + 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 +215,23 @@ async def process_video_cloud( - State tracked in Firestore - AI reasoning via Vertex AI Agent Builder """ + # Reject an unsafe callback URL up front (cheap, no DNS) so the caller gets + # immediate feedback instead of a silently-dropped callback later. Raised + # before the try/except below so it surfaces as 400, not 500. + 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,10 +271,11 @@ 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 + # The exception is sanitized for logs and never returned to the client. raise HTTPException(status_code=500, detail="Internal server error") @router.post("/api/v3/process-video-task") @@ -164,8 +299,9 @@ async def process_video_task_handler( ) 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: @@ -177,23 +313,72 @@ async def process_video_task_handler( force_refresh=False, ) - # Call callback URL if provided + # Call callback URL if provided. Resolve and validate off the event loop, + # then connect to that exact numeric address. The logical URL remains in + # Host and TLS SNI so routing and certificate verification still target + # the callback hostname without allowing connect-time DNS rebinding. 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") + loop = asyncio.get_running_loop() + deadline = loop.time() + _CALLBACK_TOTAL_TIMEOUT + 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 - loop.time() + if remaining <= 0: + break + pinned_url = callback_url.copy_with(host=address) + try: + await client.post( + pinned_url, + 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: + # httpx has not sent the request when connection + # establishment fails, so another already-validated + # address is safe to try without duplicating a POST. + 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(str(e)) ) - logger.info(f"✅ Callback sent to {payload.callback_url}") - except Exception as e: - logger.warning(f"⚠️ Callback failed: {e}") return { "success": result.success, @@ -203,7 +388,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: @@ -214,7 +401,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 712cd678b..d8b1beffb 100644 --- a/tests/unit/test_cloud_routes.py +++ b/tests/unit/test_cloud_routes.py @@ -11,10 +11,12 @@ from __future__ import annotations +import ipaddress import sys import types as _types from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch +from urllib.parse import urlparse import pytest @@ -101,7 +103,11 @@ def _make_pkg(name: str) -> _types.ModuleType: from youtube_extension.backend.cloud_ai_routes import ( router as cloud_ai_router, ) - from youtube_extension.backend.cloud_api_endpoints import setup_cloud_api_endpoints + from youtube_extension.backend.cloud_api_endpoints import ( + setup_cloud_api_endpoints, + _is_safe_callback_url, + _sanitize_log_value, + ) # The modules under test are now imported and hold their own references to the # leaf stubs above. Remove those import-time stubs from sys.modules so they do @@ -1468,3 +1474,273 @@ 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" } + + +# =========================================================================== +# Regression tests for the callback SSRF guard + log sanitizer +# =========================================================================== + +import youtube_extension.backend.cloud_api_endpoints as _cae + + +class TestCallbackUrlSafety: + """Direct unit tests for _is_safe_callback_url / _sanitize_log_value.""" + + @pytest.mark.parametrize( + "value,expected", + [ + ("plain", "plain"), + ("a\r\nb", "ab"), + ("line1\nline2", "line1line2"), + ("carriage\rreturn", "carriagereturn"), + ("forged\n2026 [ERROR] injected", "forged2026 [ERROR] injected"), + (123, "123"), + ], + ) + def test_sanitize_log_value_strips_crlf(self, value, expected): + assert _sanitize_log_value(value) == expected + + def test_allows_public_ip_literal(self): + assert _is_safe_callback_url("https://93.184.216.34/callback") is True + + @pytest.mark.parametrize( + "url", + [ + "http://127.0.0.1/x", # loopback literal + "http://10.0.0.5/x", # private + "http://192.168.1.1/x", # private + "http://169.254.169.254/meta", # link-local / GCP metadata + "http://100.64.0.1/x", # CGNAT: neither private nor global + "http://[fec0::1]/x", # deprecated IPv6 site-local + "https://metadata.google.internal/x", # blocklisted hostname + "https://metadata.google.internal./x", # trailing-dot bypass attempt + "http://LOCALHOST/x", # case-insensitive blocklist + "ftp://example.com/x", # non-http scheme + "file:///etc/passwd", # non-http scheme + "not a url", # malformed / no host + "", # empty + ], + ) + def test_rejects_unsafe_without_dns(self, url): + # resolve=False path must reject all of these without any DNS lookup. + assert _is_safe_callback_url(url, resolve=False) is False + + def test_resolve_false_allows_nonliteral_host_without_dns(self): + # Cheap gate lets a plain DNS name through; full resolution happens + # later at dispatch. Guard against accidental DNS on the acceptance path. + with patch.object(_cae.socket, "getaddrinfo", + side_effect=AssertionError("no DNS on cheap path")): + assert _is_safe_callback_url("https://example.com/x", resolve=False) is True + + def test_rejects_obfuscated_ip_resolving_to_loopback(self): + with patch.object(_cae.socket, "getaddrinfo", + return_value=[(2, 1, 6, "", ("127.0.0.1", 0))]): + assert _is_safe_callback_url("http://2130706433/x") is False + + def test_rejects_dns_alias_to_private(self): + with patch.object(_cae.socket, "getaddrinfo", + return_value=[(2, 1, 6, "", ("10.0.0.5", 0))]): + assert _is_safe_callback_url("http://127.0.0.1.nip.io/x") is False + + def test_rejects_when_any_resolved_address_blocked(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 _is_safe_callback_url("http://mixed.example/x") is False + + def test_allows_hostname_resolving_public(self): + with patch.object(_cae.socket, "getaddrinfo", + return_value=[(2, 1, 6, "", ("93.184.216.34", 0))]): + assert _is_safe_callback_url("https://example.com/x") is True + + def test_rejects_unresolvable_hostname(self): + with patch.object(_cae.socket, "getaddrinfo", + side_effect=_cae.socket.gaierror("nope")): + assert _is_safe_callback_url("https://no-such-host.invalid/x") is False + + +class TestCallbackSsrfEndToEnd: + """End-to-end: unsafe callbacks are rejected at acceptance and never POSTed.""" + + @staticmethod + def _state(): + state = MagicMock() + state.video_id = "auJzb1D-fag" + state.video_url = "https://yt.com/watch?v=auJzb1D-fag" + state.processing_time = 1.0 + state.success = True + return state + + def _post_task(self, callback_url, mock_client_cls): + mock_processor = AsyncMock() + mock_processor.process_video_sync = AsyncMock(return_value=self._state()) + with patch( + "youtube_extension.backend.cloud_api_endpoints.get_cloud_video_processor", + return_value=mock_processor, + ), patch("httpx.AsyncClient", mock_client_cls): + return TestClient(_make_cloud_api_app()).post( + "/api/v3/process-video-task", + json={ + "video_id": "auJzb1D-fag", + "video_url": "https://yt.com/watch?v=auJzb1D-fag", + "callback_url": callback_url, + }, + headers={"X-CloudTasks-TaskName": "task-1"}, + ) + + def test_acceptance_rejects_unsafe_callback_with_400(self): + # process-video validates the callback up front (no DNS) and 400s. + mock_processor = AsyncMock() + mock_processor._extract_video_id = MagicMock(return_value="auJzb1D-fag") + with patch( + "youtube_extension.backend.cloud_api_endpoints.get_cloud_video_processor", + return_value=mock_processor, + ): + response = TestClient(_make_cloud_api_app()).post( + "/api/v3/process-video", + json={ + "video_url": "https://www.youtube.com/watch?v=auJzb1D-fag", + "callback_url": "http://127.0.0.1/steal", + }, + ) + assert response.status_code == 400 + + def test_task_handler_never_posts_to_unsafe_callback(self): + mock_client_cls = MagicMock() # httpx.AsyncClient must not be constructed + response = self._post_task("http://169.254.169.254/steal", mock_client_cls) + assert response.status_code == 200 + mock_client_cls.assert_not_called() + + def test_task_handler_posts_to_safe_callback(self): + mock_client = AsyncMock() + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + mock_client.post = AsyncMock() + mock_client_cls = MagicMock(return_value=mock_client) + + # Public IP literal -> guard allows without DNS; redirects disabled. + response = self._post_task("https://93.184.216.34/cb", mock_client_cls) + assert response.status_code == 200 + mock_client_cls.assert_called_once() + assert mock_client_cls.call_args.kwargs.get("follow_redirects") is False + mock_client.post.assert_awaited_once() + + def test_task_handler_falls_back_only_to_prevalidated_public_address(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("first address down"), None] + ) + mock_client_cls = MagicMock(return_value=mock_client) + + with patch.object( + _cae.socket, + "getaddrinfo", + return_value=[ + (2, _cae.socket.SOCK_STREAM, 6, "", ("93.184.216.34", 443)), + (2, _cae.socket.SOCK_STREAM, 6, "", ("8.8.8.8", 443)), + ], + ): + response = self._post_task("https://callbacks.example/cb", mock_client_cls) + + assert response.status_code == 200 + assert [ + str(awaited.args[0]) for awaited in mock_client.post.await_args_list + ] == ["https://93.184.216.34/cb", "https://8.8.8.8/cb"] + assert all( + awaited.kwargs["headers"]["Host"] == "callbacks.example" + and awaited.kwargs["extensions"]["sni_hostname"] == "callbacks.example" + for awaited in mock_client.post.await_args_list + ) + + @pytest.mark.parametrize( + "callback_url,pinned_url,host_header", + [ + ( + "http://callbacks.example:8080/cb?job=1", + "http://93.184.216.34:8080/cb?job=1", + "callbacks.example:8080", + ), + ( + "https://callbacks.example:8443/cb?job=1", + "https://93.184.216.34:8443/cb?job=1", + "callbacks.example:8443", + ), + ], + ) + def test_task_handler_pins_validated_address_against_dns_rebind( + self, callback_url, pinned_url, host_header + ): + mock_client = AsyncMock() + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + connected_addresses = [] + + async def observe_connect_target(url, **kwargs): + parsed = urlparse(str(url)) + host = parsed.hostname + try: + connected_addresses.append(str(ipaddress.ip_address(host))) + except ValueError: + # Model httpx's independent connect-time resolution. A vulnerable + # implementation passes the attacker-controlled name here and sees + # the rebound private address from the second DNS response. + connected_addresses.append( + _cae.socket.getaddrinfo( + host, parsed.port, type=_cae.socket.SOCK_STREAM + )[0][4][0] + ) + + mock_client.post = AsyncMock(side_effect=observe_connect_target) + mock_client_cls = MagicMock(return_value=mock_client) + public_answer = [(2, _cae.socket.SOCK_STREAM, 6, "", ("93.184.216.34", 443))] + rebound_answer = [(2, _cae.socket.SOCK_STREAM, 6, "", ("127.0.0.1", 443))] + + with patch.object( + _cae.socket, + "getaddrinfo", + side_effect=[public_answer, rebound_answer], + ): + response = self._post_task(callback_url, mock_client_cls) + + assert response.status_code == 200 + assert connected_addresses == ["93.184.216.34"] + call_args, call_kwargs = mock_client.post.await_args + assert str(call_args[0]) == pinned_url + assert call_kwargs["headers"]["Host"] == host_header + assert call_kwargs["extensions"]["sni_hostname"] == "callbacks.example" + assert call_kwargs["json"]["status"] == "completed" + + def test_task_handler_caps_black_holing_address_attempts(self): + # A hostname resolving to many black-holing public addresses must not + # exceed the attempt cap, bounding worst-case task-worker time. + addresses = tuple(f"203.0.113.{i}" for i in range(1, 6)) # 5 public IPs + 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_client_cls = MagicMock(return_value=mock_client) + + mock_processor = AsyncMock() + mock_processor.process_video_sync = AsyncMock(return_value=self._state()) + with patch( + "youtube_extension.backend.cloud_api_endpoints.get_cloud_video_processor", + return_value=mock_processor, + ), patch( + "youtube_extension.backend.cloud_api_endpoints._validated_callback_addresses", + return_value=addresses, + ), patch("httpx.AsyncClient", mock_client_cls): + response = TestClient(_make_cloud_api_app()).post( + "/api/v3/process-video-task", + json={ + "video_id": "auJzb1D-fag", + "video_url": "https://yt.com/watch?v=auJzb1D-fag", + "callback_url": "https://many.example/cb", + }, + headers={"X-CloudTasks-TaskName": "task-cap"}, + ) + assert response.status_code == 200 + assert mock_client.post.await_count == _cae._MAX_CALLBACK_ADDRESS_ATTEMPTS