From 1dfe05d4b39ca9da61a9ec90e61fd0a79cc9b5da Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 21 Jul 2026 08:16:09 +0000 Subject: [PATCH 1/4] fix(security): SSRF-safe callback + log sanitization in cloud_api_endpoints Scoped to the still-live security gap on current main (the router refactor and exception-exposure fixes are already covered by #733/#831). Callback SSRF (user-supplied callback_url -> outbound POST): - _is_safe_callback_url() requires http(s)+hostname; rejects blocklisted internal hosts (trailing-dot / case normalized), and loopback/private/ link-local/reserved/multicast/unspecified IPs. With resolve=True it resolves via DNS and rejects if ANY resolved address is blocked, defeating obfuscated IPv4 encodings and DNS names that map to internal addresses. - Validated cheaply at request acceptance (400 on bad callback_url) and fully (DNS, off the event loop via asyncio.to_thread) immediately before dispatch; httpx client uses follow_redirects=False. - Residual DNS-rebinding TOCTOU documented (needs transport-level pinning). Log injection: - _sanitize_log_value() strips CR/LF from user-controlled values (video_url, task name, video_id, callback_url); lazy %-style logging. Tests: tests/unit/test_cloud_routes.py adds unit coverage for the guard (literal + DNS-resolution paths, obfuscated/rebinding/trailing-dot cases) and end-to-end (400 at acceptance; client never constructed for unsafe callback; safe callback POSTs with redirects disabled). 116 passed. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_018e5kEDnS1n4MKvoT6eQ6T4 --- .../backend/cloud_api_endpoints.py | 154 ++++++++++++++--- tests/unit/test_cloud_routes.py | 156 +++++++++++++++++- 2 files changed, 289 insertions(+), 21 deletions(-) diff --git a/src/youtube_extension/backend/cloud_api_endpoints.py b/src/youtube_extension/backend/cloud_api_endpoints.py index b1999ceaa..c5d97b58b 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,90 @@ 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 if the address is in a range unsafe for outbound callbacks.""" + return ( + ip.is_private + or ip.is_loopback + or ip.is_link_local + or ip.is_reserved + or ip.is_multicast + or ip.is_unspecified + ) + + +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. Because httpx + re-resolves the host at connect time, this mitigates but does not fully + eliminate DNS-rebinding TOCTOU; closing that entirely requires + transport-level pinning of the validated address. + """ + try: + parsed = urlparse(url) + except ValueError: + return False + + hostname = parsed.hostname + if parsed.scheme not in ("http", "https") or not hostname: + return False + + if hostname.rstrip(".").lower() in _BLOCKED_CALLBACK_HOSTS: + return False + + try: + ip = ipaddress.ip_address(hostname) + except ValueError: + ip = None + + if ip is not None: + return not _is_blocked_ip(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 True + + try: + addrinfos = socket.getaddrinfo(hostname, None) + except (socket.gaierror, UnicodeError, ValueError): + # Unresolvable hostname — treat as unsafe. + return False + + for info in addrinfos: + resolved = str(info[4][0]).split("%", 1)[0] # drop IPv6 scope/zone id + try: + resolved_ip = ipaddress.ip_address(resolved) + except ValueError: + return False + if _is_blocked_ip(resolved_ip): + return False + return True + # Pydantic models for API requests/responses class CloudVideoProcessingRequest(BaseModel): @@ -91,13 +179,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: @@ -164,8 +262,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 +276,38 @@ async def process_video_task_handler( force_refresh=False, ) - # Call callback URL if provided + # Call callback URL if provided. Re-validate here (with DNS resolution, + # off the event loop) immediately before dispatch — the URL was only + # cheaply validated at acceptance, and this is the actual SSRF sink. 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 + if not await asyncio.to_thread( + _is_safe_callback_url, payload.callback_url + ): + logger.warning( + "⚠️ Refusing to call unsafe callback URL: %s", + _sanitize_log_value(payload.callback_url), + ) + else: + try: + import httpx + async with httpx.AsyncClient(follow_redirects=False) as client: + await client.post( + payload.callback_url, + json={ + 'video_id': result.video_id, + 'status': 'completed', + 'processing_time': result.processing_time, + }, + timeout=10.0, + ) + logger.info( + "✅ Callback sent to %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, diff --git a/tests/unit/test_cloud_routes.py b/tests/unit/test_cloud_routes.py index 712cd678b..7157adf1a 100644 --- a/tests/unit/test_cloud_routes.py +++ b/tests/unit/test_cloud_routes.py @@ -101,7 +101,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 +1472,153 @@ 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 + "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() From 957dec20d17d609dcd4c616210dbd6a78f04a9a2 Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Tue, 21 Jul 2026 04:00:24 -0500 Subject: [PATCH 2/4] fix(security): pin callback connections after DNS validation --- .../backend/cloud_api_endpoints.py | 102 +++++++++++++----- tests/unit/test_cloud_routes.py | 89 +++++++++++++++ 2 files changed, 162 insertions(+), 29 deletions(-) diff --git a/src/youtube_extension/backend/cloud_api_endpoints.py b/src/youtube_extension/backend/cloud_api_endpoints.py index c5d97b58b..146a03748 100644 --- a/src/youtube_extension/backend/cloud_api_endpoints.py +++ b/src/youtube_extension/backend/cloud_api_endpoints.py @@ -72,22 +72,34 @@ def _is_safe_callback_url(url: str, *, resolve: bool = True) -> bool: ``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. Because httpx - re-resolves the host at connect time, this mitigates but does not fully - eliminate DNS-rebinding TOCTOU; closing that entirely requires - transport-level pinning of the validated address. + 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 False + return None hostname = parsed.hostname if parsed.scheme not in ("http", "https") or not hostname: - return False + return None if hostname.rstrip(".").lower() in _BLOCKED_CALLBACK_HOSTS: - return False + return None try: ip = ipaddress.ip_address(hostname) @@ -95,28 +107,39 @@ def _is_safe_callback_url(url: str, *, resolve: bool = True) -> bool: ip = None if ip is not None: - return not _is_blocked_ip(ip) + 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 True + return () try: - addrinfos = socket.getaddrinfo(hostname, None) + 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 False + 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 False + return None if _is_blocked_ip(resolved_ip): - return False - return True + 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 @@ -276,13 +299,15 @@ async def process_video_task_handler( force_refresh=False, ) - # Call callback URL if provided. Re-validate here (with DNS resolution, - # off the event loop) immediately before dispatch — the URL was only - # cheaply validated at acceptance, and this is the actual SSRF sink. + # 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: - if not await asyncio.to_thread( - _is_safe_callback_url, payload.callback_url - ): + 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), @@ -290,16 +315,35 @@ async def process_video_task_handler( else: try: import httpx + + callback_url = httpx.URL(payload.callback_url) async with httpx.AsyncClient(follow_redirects=False) as client: - await client.post( - payload.callback_url, - json={ - 'video_id': result.video_id, - 'status': 'completed', - 'processing_time': result.processing_time, - }, - timeout=10.0, - ) + last_connect_error = None + for address in callback_addresses: + 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": callback_url.netloc.decode("ascii") + }, + extensions={"sni_hostname": callback_url.host}, + timeout=10.0, + ) + 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 + else: + assert last_connect_error is not None + raise last_connect_error logger.info( "✅ Callback sent to %s", _sanitize_log_value(payload.callback_url), diff --git a/tests/unit/test_cloud_routes.py b/tests/unit/test_cloud_routes.py index 7157adf1a..9f2503ab7 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 @@ -1622,3 +1624,90 @@ def test_task_handler_posts_to_safe_callback(self): 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" From 92f26da2d0a46c011eb1b541d875f62eebe20b09 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 21 Jul 2026 09:12:31 +0000 Subject: [PATCH 3/4] fix(security): reject non-global callback IPs and bound callback attempts Addresses Copilot review on the current head: - _is_blocked_ip now rejects any non-global address (plus multicast and deprecated IPv6 site-local) instead of enumerating unsafe ranges. This closes shared CGNAT space (100.64.0.0/10) and fec0::/10, which Python reports as neither private nor global and which the enumerated form let through as callback targets into non-public networks. - The callback dispatch loop is now bounded: at most _MAX_CALLBACK_ADDRESS_ATTEMPTS addresses are tried under a single overall deadline (_CALLBACK_TOTAL_TIMEOUT), so a hostname resolving to many black-holing public addresses can no longer occupy a task worker for minutes instead of one timeout. Tests: CGNAT/site-local rejection cases + an attempt-cap test. 122 passed. Co-Authored-By: Claude Opus 4.8 (1M context) Claude-Session: https://claude.ai/code/session_018e5kEDnS1n4MKvoT6eQ6T4 --- .../backend/cloud_api_endpoints.py | 66 +++++++++++++------ tests/unit/test_cloud_routes.py | 33 ++++++++++ 2 files changed, 79 insertions(+), 20 deletions(-) diff --git a/src/youtube_extension/backend/cloud_api_endpoints.py b/src/youtube_extension/backend/cloud_api_endpoints.py index 146a03748..cba1ec436 100644 --- a/src/youtube_extension/backend/cloud_api_endpoints.py +++ b/src/youtube_extension/backend/cloud_api_endpoints.py @@ -46,17 +46,30 @@ def _sanitize_log_value(value: Any) -> str: def _is_blocked_ip(ip: Union[ipaddress.IPv4Address, ipaddress.IPv6Address]) -> bool: - """Return True if the address is in a range unsafe for outbound callbacks.""" + """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_private - or ip.is_loopback - or ip.is_link_local - or ip.is_reserved - or ip.is_multicast - or ip.is_unspecified + 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. @@ -317,9 +330,18 @@ async def process_video_task_handler( 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: - last_connect_error = None - for address in callback_addresses: + 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( @@ -329,25 +351,29 @@ async def process_video_task_handler( "status": "completed", "processing_time": result.processing_time, }, - headers={ - "Host": callback_url.netloc.decode("ascii") - }, + headers={"Host": host_header}, extensions={"sni_hostname": callback_url.host}, - timeout=10.0, + 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 - else: - assert last_connect_error is not None - raise last_connect_error - logger.info( - "✅ Callback sent to %s", - _sanitize_log_value(payload.callback_url), - ) + 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)) diff --git a/tests/unit/test_cloud_routes.py b/tests/unit/test_cloud_routes.py index 9f2503ab7..d8b1beffb 100644 --- a/tests/unit/test_cloud_routes.py +++ b/tests/unit/test_cloud_routes.py @@ -1510,6 +1510,8 @@ def test_allows_public_ip_literal(self): "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 @@ -1711,3 +1713,34 @@ async def observe_connect_target(url, **kwargs): 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 From 7dc9cbe46c141260c668d3871fe7681c0c2d86e5 Mon Sep 17 00:00:00 2001 From: Hayden <154503486+groupthinking@users.noreply.github.com> Date: Tue, 21 Jul 2026 04:21:41 -0500 Subject: [PATCH 4/4] fix(security): sanitize callback exception logs --- .../backend/cloud_api_endpoints.py | 17 ++++++++++++----- 1 file changed, 12 insertions(+), 5 deletions(-) diff --git a/src/youtube_extension/backend/cloud_api_endpoints.py b/src/youtube_extension/backend/cloud_api_endpoints.py index cba1ec436..7b93a8ab9 100644 --- a/src/youtube_extension/backend/cloud_api_endpoints.py +++ b/src/youtube_extension/backend/cloud_api_endpoints.py @@ -271,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") @@ -387,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: @@ -398,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")