scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-08 22:21:35 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 78

fix(proxy): wire team-level logging callbacks into passthrough endpoints (#38979)

Browse source 
* 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
This commit is contained in:
yucheng-berri 2026-08-31 19:38:39 -07:00 • committed by GitHub
parent 3fadcd7155
commit ccd76dac50
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 251 additions and 5 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

95
litellm/proxy/pass_through_endpoints/pass_through_endpoints.py
View file

@ -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

161
tests/test_litellm/proxy/pass_through_endpoints/test_pass_through_endpoints.py
View file

@ -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