From ccd76dac505035885b34e54d325ea9bfe9a6718d Mon Sep 17 00:00:00 2001 From: yucheng-berri Date: Mon, 31 Aug 2026 19:38:39 -0700 Subject: [PATCH] fix(proxy): wire team-level logging callbacks into passthrough endpoints (#38979) * fix(proxy): wire team-level logging callbacks into passthrough endpoints LIT-5152: passthrough routes now wire dynamic team-level callbacks (success_callback, failure_callback, callback_vars) into Logging constructor, mirroring the add_litellm_data_to_request behavior. Three hardening fixes: 1. Catch TypeError/AttributeError in _get_validated_callback_metadata when team logging metadata has wrong shape (e.g., logging list instead of dict), preventing HTTP 500 on passthrough routes with malformed config. 2. Wrap websocket passthrough logging initialization in try/except, since the socket is already accepted at that point; errors after accept() yield abrupt close (1006/1011) rather than clean HTTP error response. 3. Handle malformed deprecated callback_settings gracefully with try/except. 4. Wrap HTTP passthrough callback resolution in try/except to prevent 500 on malformed team metadata (backward-compatibility fix). Changes: - pass_through_endpoints.py: wire dynamic callbacks in HTTP+WS paths, handle malformed metadata gracefully with try/except fallbacks - litellm_pre_call_utils.py: expand exception handling in validators - test file: regression test for happy-path team callback wiring * refactor(proxy): share passthrough team-callback resolution and cover its fail-open path Collapse the duplicated callback wiring on the HTTP and websocket passthrough paths into one helper that returns a frozen wiring value, log resolution failures at error level so a broken logging config stays visible, and add regression tests for malformed team metadata and an operational lookup failure. Reverts the _get_validated_callback_metadata except widening: it changed behavior for normal LLM routes, which is outside this ticket's scope. * fix(proxy): keep passthrough alive when team callback vars hold env references The deprecated team_metadata.callback_settings branch builds TeamCallbackMetadata directly, skipping the AddTeamCallback validation that strips os.environ/ references from the newer logging list. Stamping those vars onto the Logging object made its constructor raise, so a team on the legacy shape got HTTP 500 on every passthrough call. Validate the resolved vars inside the fail-open boundary instead, so the request goes through with dynamic callbacks skipped and the reason logged. * fix(proxy): lint violations in team callback wiring helper * style: format lint --- .../pass_through_endpoints.py | 95 ++++++++++- .../test_pass_through_endpoints.py | 161 ++++++++++++++++++ 2 files changed, 251 insertions(+), 5 deletions(-) diff --git a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py index 09d3dedaafa..ddd27f0fc2a 100644 --- a/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py +++ b/litellm/proxy/pass_through_endpoints/pass_through_endpoints.py @@ -6,9 +6,10 @@ import posixpath import traceback from base64 import b64encode from collections.abc import AsyncGenerator, Callable, Iterable, Mapping, Sequence +from dataclasses import dataclass from datetime import datetime from itertools import groupby -from typing import Any, Final, TypedDict, cast +from typing import TYPE_CHECKING, Any, Final, TypedDict, cast from urllib.parse import urlencode, urlparse import httpx @@ -47,6 +48,7 @@ from litellm.litellm_core_utils.core_helpers import ( get_metadata_variable_name_from_kwargs, get_or_create_metadata_bucket, ) +from litellm.litellm_core_utils.initialize_dynamic_callback_params import validate_no_callback_env_reference from litellm.litellm_core_utils.internal_call_metadata import MODEL_ACCESS_GROUP_METADATA_KEY from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER @@ -78,7 +80,10 @@ from litellm.proxy.common_utils.http_parsing_utils import ( from litellm.proxy.common_utils.sse_keepalive import ( wrap_passthrough_sse_bytes_with_keepalive_pings, ) -from litellm.proxy.litellm_pre_call_utils import LiteLLMProxyRequestSetup +from litellm.proxy.litellm_pre_call_utils import ( + LiteLLMProxyRequestSetup, + _get_dynamic_logging_metadata, # pyright: ignore[reportPrivateUsage] # shared proxy helper, same import style as _read_request_body above +) from litellm.proxy.utils import normalize_route_for_root_path from litellm.repositories.team_repository import TeamRepository from litellm.secret_managers.main import get_secret_str @@ -90,7 +95,7 @@ from litellm.types.passthrough_endpoints.pass_through_endpoints import ( EndpointType, PassthroughStandardLoggingPayload, ) -from litellm.types.utils import Usage +from litellm.types.utils import TRUSTED_CALLBACK_VARS_FIELD, Usage from .streaming_handler import PassThroughStreamingHandler from .success_handler import PassThroughEndpointLogging @@ -99,6 +104,9 @@ from .upstream_usage_headers import ( apply_upstream_reported_usage, ) +if TYPE_CHECKING: + from litellm.proxy.proxy_server import ProxyConfig + router: Final = APIRouter() pass_through_endpoint_logging: Final = PassThroughEndpointLogging() @@ -752,6 +760,67 @@ def _build_passthrough_failure_request_payload( return request_payload +@dataclass(frozen=True, slots=True) +class _TeamCallbackWiring: + success_callbacks: "list[str | Callable | CustomLogger] | None" = None # mutable-ok: Logging.__init__ arg + failure_callbacks: "list[str | Callable | CustomLogger] | None" = None # mutable-ok: Logging.__init__ arg + logging_kwargs: dict[str, str | dict[str, str]] | None = None # mutable-ok: Logging.__init__ arg + + +def _resolve_team_callback_wiring( + user_api_key_dict: UserAPIKeyAuth, + proxy_config: "ProxyConfig", + route_description: str, +) -> _TeamCallbackWiring: + """Resolve key/team dynamic logging callbacks for a passthrough request. + + Mirrors add_litellm_data_to_request: callback_vars are unpacked top-level + (read by initialize_standard_callback_dynamic_params) and also stamped on + the proxy-owned trusted-vars field (read by get_trusted_callback_params). + + Fails open: a callback resolution or validation error is logged at error + level and the request proceeds without dynamic callbacks, since a broken + logging config must not fail the customer's upstream call (and the + websocket is already accepted by the time this runs on that path). The + env-reference check runs here because the deprecated callback_settings + branch skips AddTeamCallback validation, and Logging.__init__ would + otherwise reject the vars mid-request. + """ + try: + callback_settings_obj: Final = _get_dynamic_logging_metadata( + user_api_key_dict=user_api_key_dict, proxy_config=proxy_config + ) + if callback_settings_obj and callback_settings_obj.callback_vars: + for ( + item + ) in callback_settings_obj.callback_vars.items(): # rebind-ok: dict.items iteration for env-ref validation + validate_no_callback_env_reference(item[0], item[1], source="key/team callback metadata") + except Exception: # noqa: BLE001 - a broken logging config must never fail the passthrough request + verbose_proxy_logger.exception( + "%s: failed to resolve team logging callbacks, continuing without them", + route_description, + ) + return _TeamCallbackWiring() + if callback_settings_obj is None: + return _TeamCallbackWiring() + callback_vars: Final = callback_settings_obj.callback_vars + success_callbacks: Final = callback_settings_obj.success_callback + failure_callbacks: Final = callback_settings_obj.failure_callback + logging_kwargs: Final = ( + None + if not callback_vars + else { # mutable-ok: Logging arg + **callback_vars, + TRUSTED_CALLBACK_VARS_FIELD: callback_vars, + } + ) + return _TeamCallbackWiring( + success_callbacks=None if success_callbacks is None else [*success_callbacks], # mutable-ok: Logging arg + failure_callbacks=None if failure_callbacks is None else [*failure_callbacks], # mutable-ok: Logging arg + logging_kwargs=logging_kwargs, + ) + + async def _log_passthrough_upstream_failure( response: httpx.Response, user_api_key_dict: UserAPIKeyAuth, @@ -845,7 +914,7 @@ async def pass_through_request( from litellm.proxy.pass_through_endpoints.passthrough_guardrails import ( PassthroughGuardrailHandler, ) - from litellm.proxy.proxy_server import proxy_logging_obj + from litellm.proxy.proxy_server import proxy_config, proxy_logging_obj ######################################################### # Initialize variables @@ -930,6 +999,11 @@ async def pass_through_request( # read e.g. ``chat gpt-4o`` instead of ``chat unknown``. passthrough_model: Final = (_parsed_body.get("model") if isinstance(_parsed_body, dict) else None) or "unknown" start_time: Final = datetime.now() + team_callbacks: Final = _resolve_team_callback_wiring( + user_api_key_dict=user_api_key_dict, + proxy_config=proxy_config, + route_description="pass_through_endpoint", + ) logging_obj = Logging( model=passthrough_model, messages=[{"role": "user", "content": safe_dumps(_parsed_body)}], @@ -938,6 +1012,9 @@ async def pass_through_request( start_time=start_time, litellm_call_id=litellm_call_id, function_id="1245", + dynamic_success_callbacks=team_callbacks.success_callbacks, + dynamic_failure_callbacks=team_callbacks.failure_callbacks, + kwargs=team_callbacks.logging_kwargs, ) # Store passthrough guardrails config on logging_obj for field targeting @@ -2022,7 +2099,7 @@ async def websocket_passthrough_request( setup_model_rewriter: Optional rewrite of the setup frame's model before it reaches the upstream """ from litellm.litellm_core_utils.litellm_logging import Logging - from litellm.proxy.proxy_server import proxy_logging_obj + from litellm.proxy.proxy_server import proxy_config, proxy_logging_obj from litellm.types.passthrough_endpoints.pass_through_endpoints import ( PassthroughStandardLoggingPayload, ) @@ -2055,6 +2132,11 @@ async def websocket_passthrough_request( upstream_headers[header_name] = header_value # Initialize logging object similar to HTTP passthrough + team_callbacks: Final = _resolve_team_callback_wiring( + user_api_key_dict=user_api_key_dict, + proxy_config=proxy_config, + route_description="websocket_passthrough", + ) logging_obj: Final = Logging( model="unknown", messages=[{"role": "user", "content": "WebSocket connection"}], @@ -2063,6 +2145,9 @@ async def websocket_passthrough_request( start_time=start_time, litellm_call_id=litellm_call_id, function_id="websocket_passthrough", + dynamic_success_callbacks=team_callbacks.success_callbacks, + dynamic_failure_callbacks=team_callbacks.failure_callbacks, + kwargs=team_callbacks.logging_kwargs, ) # Create passthrough logging payload diff --git a/tests/test_litellm/proxy/pass_through_endpoints/test_pass_through_endpoints.py b/tests/test_litellm/proxy/pass_through_endpoints/test_pass_through_endpoints.py index a3f56adb86f..f5ae0fe5977 100644 --- a/tests/test_litellm/proxy/pass_through_endpoints/test_pass_through_endpoints.py +++ b/tests/test_litellm/proxy/pass_through_endpoints/test_pass_through_endpoints.py @@ -5462,3 +5462,164 @@ def test_the_marker_check_distinguishes_the_two_route_kinds(): builtin = MagicMock(spec=Request) builtin.scope = {"endpoint": llm_passthrough_endpoints.anthropic_proxy_route} assert request_dispatched_to_pass_through_endpoint(builtin) is False + + +async def _drive_passthrough_request_and_capture_logging(user_api_key_dict: UserAPIKeyAuth) -> tuple[int, object]: + import litellm + from litellm.llms.custom_httpx.http_handler import get_async_httpx_client + from litellm.types.llms.custom_http import httpxSpecialProvider + + def transport_handler(upstream_request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json={"ok": True}) + + real_handler = get_async_httpx_client( + llm_provider=httpxSpecialProvider.PassThroughEndpoint, + params={"timeout": resolve_pass_through_request_timeout(None)}, + ) + cache_dict = litellm.in_memory_llm_clients_cache.cache_dict + cache_key = next((key for key, cached in cache_dict.items() if cached is real_handler), None) + assert cache_key is not None + cache_dict[cache_key] = SimpleNamespace(client=httpx.AsyncClient(transport=httpx.MockTransport(transport_handler))) + + mock_request = MagicMock(spec=Request) + mock_request.method = "POST" + mock_request.headers = Headers({}) + mock_request.query_params = QueryParams({}) + mock_request.body = AsyncMock(return_value=b'{"model": "gemini-2.0-flash"}') + + captured_data: dict = {} + + async def capture_pre_call_hook(user_api_key_dict, data, call_type): + captured_data.update(data) + return data + + mock_proxy_logging = MagicMock() + mock_proxy_logging.pre_call_hook = AsyncMock(side_effect=capture_pre_call_hook) + mock_proxy_logging.post_call_failure_hook = AsyncMock() + mock_proxy_logging.post_call_response_headers_hook = AsyncMock(return_value={}) + mock_proxy_logging.get_proxy_hook = MagicMock(return_value=None) + + try: + with patch( # test-quality-ok: proxy_logging_obj is a proxy_server module global read inside pass_through_request; there is no injection seam + "litellm.proxy.proxy_server.proxy_logging_obj", mock_proxy_logging + ): + response = await pass_through_request( + request=mock_request, + target="https://upstream.example.test/v1/generate", + custom_headers={}, + user_api_key_dict=user_api_key_dict, + ) + finally: + cache_dict[cache_key] = real_handler + + return response.status_code, captured_data.get("litellm_logging_obj") + + +@pytest.mark.asyncio +async def test_pass_through_request_wires_team_callbacks(): + """LIT-5152 regression: pass_through_request must resolve team-level logging + callbacks from key/team metadata and wire them into the Logging object, the + same way add_litellm_data_to_request does for normal LLM routes.""" + user_api_key_dict = UserAPIKeyAuth( + api_key="test-key", + team_id="test-team", + team_metadata={ + "logging": [ + { + "callback_name": "langfuse", + "callback_type": "success_and_failure", + "callback_vars": { + "langfuse_public_key": "pk_test", + "langfuse_secret_key": "sk_test", + "langfuse_host": "https://langfuse.example.test", + }, + } + ] + }, + ) + + status_code, logging_obj = await _drive_passthrough_request_and_capture_logging(user_api_key_dict) + + assert status_code == 200 + assert logging_obj is not None + assert logging_obj.dynamic_success_callbacks, "team success callbacks not wired into Logging" + assert logging_obj.dynamic_failure_callbacks, "team failure callbacks not wired into Logging" + assert logging_obj.standard_callback_dynamic_params.get("langfuse_public_key") == "pk_test" + assert logging_obj.standard_callback_dynamic_params.get("langfuse_secret_key") == "sk_test" + assert logging_obj.standard_callback_dynamic_params.get("langfuse_host") == "https://langfuse.example.test" + assert ("langfuse_public_key", "pk_test") in logging_obj._trusted_callback_vars + + +@pytest.mark.asyncio +async def test_pass_through_request_survives_malformed_team_logging_metadata(): + """LIT-5152 fail-open: a malformed team ``logging`` value (here a non-iterable) + raises inside callback resolution; the passthrough request must still succeed, + just without dynamic callbacks.""" + user_api_key_dict = UserAPIKeyAuth( + api_key="test-key", + team_id="test-team", + team_metadata={"logging": 5}, + ) + + status_code, logging_obj = await _drive_passthrough_request_and_capture_logging(user_api_key_dict) + + assert status_code == 200 + assert logging_obj is not None + assert not logging_obj.dynamic_success_callbacks + assert not logging_obj.dynamic_failure_callbacks + + +@pytest.mark.asyncio +async def test_pass_through_request_survives_env_reference_in_deprecated_callback_settings(): + """LIT-5152 fail-open: the deprecated ``callback_settings`` team metadata skips + AddTeamCallback validation, so an ``os.environ/`` callback var would otherwise + blow up inside ``Logging.__init__`` and fail the request; the passthrough must + instead succeed without dynamic callbacks.""" + user_api_key_dict = UserAPIKeyAuth( + api_key="test-key", + team_id="test-team", + team_metadata={ + "callback_settings": { + "success_callback": ["langfuse"], + "failure_callback": ["langfuse"], + "callback_vars": { + "langfuse_public_key": "os.environ/LANGFUSE_PUBLIC_KEY", + "langfuse_secret_key": "os.environ/LANGFUSE_SECRET_KEY", + "langfuse_host": "https://langfuse.example.test", + }, + } + }, + ) + + status_code, logging_obj = await _drive_passthrough_request_and_capture_logging(user_api_key_dict) + + assert status_code == 200 + assert logging_obj is not None + assert not logging_obj.dynamic_success_callbacks + assert not logging_obj.dynamic_failure_callbacks + assert not logging_obj.standard_callback_dynamic_params.get("langfuse_public_key") + + +@pytest.mark.asyncio +async def test_resolve_team_callback_wiring_fails_open_on_operational_error(): + """LIT-5152 fail-open: an operational error while resolving callback metadata + (e.g. team config lookup hitting a dead secret manager) must not raise; the + request proceeds without dynamic callbacks and the error is logged.""" + from litellm.proxy.pass_through_endpoints.pass_through_endpoints import ( + _resolve_team_callback_wiring, + ) + from litellm.proxy.proxy_server import ProxyConfig + + class RaisingTeamConfig(ProxyConfig): + def load_team_config(self, team_id: str) -> dict: + raise RuntimeError("secret manager unavailable") + + wiring = _resolve_team_callback_wiring( + user_api_key_dict=UserAPIKeyAuth(api_key="test-key", team_id="test-team"), + proxy_config=RaisingTeamConfig(), + route_description="pass_through_endpoint", + ) + + assert wiring.success_callbacks is None + assert wiring.failure_callbacks is None + assert wiring.logging_kwargs is None