diff --git a/litellm/integrations/otel/plumbing/context.py b/litellm/integrations/otel/plumbing/context.py index 5290d87ae9d..b7a68f77373 100644 --- a/litellm/integrations/otel/plumbing/context.py +++ b/litellm/integrations/otel/plumbing/context.py @@ -40,7 +40,7 @@ _PROPAGATOR: Final = TraceContextTextMapPropagator() # request task, so there is nothing to leak. _request_root_span: Final["ContextVar[Span | None]"] = ContextVar("litellm_otel_request_root_span", default=None) -_request_destinations: 'ContextVar[tuple["OtelDestination", ...]]' = ContextVar( +_request_destinations: Final['ContextVar[tuple["OtelDestination", ...]]'] = ContextVar( "litellm_otel_request_destinations", default=() ) diff --git a/litellm/proxy/auth/user_api_key_auth.py b/litellm/proxy/auth/user_api_key_auth.py index 03eee8b4c78..9c35902689b 100644 --- a/litellm/proxy/auth/user_api_key_auth.py +++ b/litellm/proxy/auth/user_api_key_auth.py @@ -11,7 +11,9 @@ import asyncio import fnmatch import re import secrets +from collections.abc import Mapping from datetime import datetime, timezone +from types import MappingProxyType from typing import Any, Final, NamedTuple, Protocol, Union, cast import fastapi @@ -108,6 +110,9 @@ except ImportError as e: user_api_key_service_logger_obj: Final = ServiceLogging() # used for tracking latency on OTEL +_EMPTY_MAPPING: Final[Mapping[str, str]] = MappingProxyType({}) + + def _normalize_public_auth_route(route: str) -> str: if route != "/" and route.endswith("/"): return route.rstrip("/") @@ -1034,22 +1039,22 @@ async def _hoist_request_destinations(request: Request, user_api_key_dict: UserA ) destinations_raw, _backends = await _resolve_logging_exporters(user_api_key_dict) - destinations = tuple( + destinations: Final = tuple( OtelDestination( callback_name=item.get("callback_name"), endpoint=item.get("endpoint", ""), - headers=item.get("headers") or {}, - resource_attributes=item.get("resource_attributes") or {}, + headers=item.get("headers") or _EMPTY_MAPPING, + resource_attributes=item.get("resource_attributes") or _EMPTY_MAPPING, protocol=item.get("protocol"), ) for item in destinations_raw - if isinstance(item, dict) and item.get("endpoint") + if item.get("endpoint") ) set_request_destinations(destinations) try: request.state.otel_destinations = destinations_raw - except Exception: # noqa: BLE001 # request.state mirror is best-effort; the ContextVar is the source of truth - pass + except Exception as mirror_exc: # noqa: BLE001 # the ContextVar is the source of truth + verbose_proxy_logger.debug("OTel V2: request.state mirror failed: %s", mirror_exc) except Exception as exc: # noqa: BLE001 # destination hoist is best-effort telemetry setup; it must never fail auth verbose_proxy_logger.debug("OTel V2: hoist destination resolution failed: %s", exc) diff --git a/litellm/proxy/litellm_pre_call_utils.py b/litellm/proxy/litellm_pre_call_utils.py index 97bc7a7fded..93a59390fce 100644 --- a/litellm/proxy/litellm_pre_call_utils.py +++ b/litellm/proxy/litellm_pre_call_utils.py @@ -719,7 +719,7 @@ async def _effective_org_id(user_api_key_dict: UserAPIKeyAuth) -> str | None: """ if user_api_key_dict.org_id is not None: return user_api_key_dict.org_id - team_id = user_api_key_dict.team_id + team_id: Final = user_api_key_dict.team_id if team_id is None: return None from litellm.proxy import proxy_server @@ -728,7 +728,7 @@ async def _effective_org_id(user_api_key_dict: UserAPIKeyAuth) -> str | None: if proxy_server.prisma_client is None: return None try: - team_obj = await get_team_object( + team_obj: Final = await get_team_object( team_id=team_id, prisma_client=proxy_server.prisma_client, user_api_key_cache=proxy_server.user_api_key_cache, @@ -769,7 +769,7 @@ async def _resolve_logging_exporters( if not is_otel_v2_enabled(): return (), () - logging_credentials = tuple( + logging_credentials: Final = tuple( (credential, info) for credential in litellm.credential_list if (info := parse_credential_info(credential.credential_info)) is not None and info.credential_type == "logging" @@ -777,21 +777,21 @@ async def _resolve_logging_exporters( if not logging_credentials: return (), () - team_id = user_api_key_dict.team_id - org_id = ( + team_id: Final = user_api_key_dict.team_id + org_id: Final = ( await _effective_org_id(user_api_key_dict) if any(info.access is not None and info.access.orgs for _, info in logging_credentials) else None ) team_ids, org_ids = identity_scope(team_id, org_id) - built = tuple( + built: Final = tuple( result for credential, info in logging_credentials if access_grants(info.access, team_ids, org_ids) if (result := destination_for_credential(credential)) is not None ) - deduped = { + deduped: Final = { # mutable-ok: dedup accumulator, consumed immediately below ( destination.endpoint, tuple(sorted(destination.headers.items())), @@ -802,8 +802,8 @@ async def _resolve_logging_exporters( ) for backend, destination in built } - destinations: tuple[OtelDestinationParams, ...] = tuple( - { + destinations: Final[tuple[OtelDestinationParams, ...]] = tuple( + { # mutable-ok: OtelDestinationParams is a TypedDict, so each entry is a real dict "callback_name": backend, "endpoint": destination.endpoint, "headers": destination.headers, @@ -812,7 +812,7 @@ async def _resolve_logging_exporters( } for backend, destination in deduped.values() ) - backends = tuple(dict.fromkeys(backend for backend, _ in deduped.values())) + backends: Final = tuple(dict.fromkeys(backend for backend, _ in deduped.values())) return destinations, backends @@ -883,12 +883,13 @@ async def _apply_admin_logging_exporters( if not is_otel_v2_enabled(): return if cached_destinations is not None: - destinations = tuple(cached_destinations) + destinations = tuple(cached_destinations) # rebind-ok: the else branch resolves it instead else: try: - destinations, _backends = await _resolve_logging_exporters(user_api_key_dict) + resolved, _backends = await _resolve_logging_exporters(user_api_key_dict) except Exception: # noqa: BLE001 # best-effort telemetry setup must never break the request return + destinations = resolved # rebind-ok: the cached branch assigns it instead _set_request_otel_destinations(destinations) @@ -2206,7 +2207,7 @@ async def add_litellm_data_to_request( # proxy-owned field instead of the raw request kwargs. data[TRUSTED_CALLBACK_VARS_FIELD] = callback_settings_obj.callback_vars - cached = getattr(getattr(request, "state", None), "otel_destinations", None) + cached: Final = getattr(getattr(request, "state", None), "otel_destinations", None) await _apply_admin_logging_exporters(user_api_key_dict, cached_destinations=cached) # Add disabled callbacks from key metadata diff --git a/litellm/types/utils.py b/litellm/types/utils.py index 05311ce940e..9239b98e3e9 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -39,7 +39,7 @@ from pydantic import ( field_serializer, field_validator, ) -from typing_extensions import Required, TypedDict +from typing_extensions import ReadOnly, Required, TypedDict from litellm._logging import verbose_logger from litellm._uuid import uuid @@ -3208,11 +3208,11 @@ class OtelDestinationParams(TypedDict, total=False): belongs to, so fan-out routes each destination to the right backend's logger. """ - callback_name: str - endpoint: str - headers: Mapping[str, str] - resource_attributes: Mapping[str, str] - protocol: str | None + callback_name: ReadOnly[str] + endpoint: ReadOnly[str] + headers: ReadOnly[Mapping[str, str]] + resource_attributes: ReadOnly[Mapping[str, str]] + protocol: ReadOnly[str | None] class StandardCallbackDynamicParams(TypedDict, total=False): diff --git a/tests/test_litellm/proxy/management_endpoints/test_logging_exporter_access.py b/tests/test_litellm/proxy/management_endpoints/test_logging_exporter_access.py index 603a15f72a0..cc66bc57c6b 100644 --- a/tests/test_litellm/proxy/management_endpoints/test_logging_exporter_access.py +++ b/tests/test_litellm/proxy/management_endpoints/test_logging_exporter_access.py @@ -299,4 +299,33 @@ def test_resolved_names_keep_every_grant_sharing_one_target(monkeypatch): _cred("distinct", access={"teams": ["t1"]}), ], ) - assert resolved_logging_exporter_names("t1", None) == ("dup-one", "dup-two", "distinct") \ No newline at end of file + assert resolved_logging_exporter_names("t1", None) == ("dup-one", "dup-two", "distinct") + + + +@pytest.mark.asyncio +async def test_disclosure_agrees_with_the_resolver_on_a_shared_target(monkeypatch): + """Regression: the disclosed names and the resolver's selection are derived from the + same grants, so no credential is disclosed that the resolver dropped entirely. + + Pins the two sides together: the resolver collapses the duplicate target to a single + export, and every name it kept a destination for is disclosed.""" + from litellm.proxy._types import UserAPIKeyAuth + from litellm.proxy.litellm_pre_call_utils import _resolve_logging_exporters + + monkeypatch.setenv("LITELLM_OTEL_V2", "true") + is_otel_v2_enabled.cache_clear() + same = "http://collector.example/shared/v1/traces" + monkeypatch.setattr( + litellm, + "credential_list", + [ + _cred("dup-one", access={"global": True}, endpoint=same), + _cred("dup-two", access={"global": True}, endpoint=same), + ], + ) + + destinations, _backends = await _resolve_logging_exporters(UserAPIKeyAuth(api_key="k")) + + assert len(destinations) == 1 + assert resolved_logging_exporter_names(None, None) == ("dup-one", "dup-two") diff --git a/tests/test_litellm/proxy/test_litellm_pre_call_utils.py b/tests/test_litellm/proxy/test_litellm_pre_call_utils.py index 84b8f25bf10..e0994f88d65 100644 --- a/tests/test_litellm/proxy/test_litellm_pre_call_utils.py +++ b/tests/test_litellm/proxy/test_litellm_pre_call_utils.py @@ -5536,27 +5536,6 @@ async def test_client_cannot_control_otel_destinations(_seeded_logging_credentia assert not dynamic_params.get("otel_destinations") -@pytest.mark.asyncio -async def test_apply_admin_logging_exporters_registers_on_failure(_seeded_logging_credentials, monkeypatch): - """An admin-owned destination must capture a FAILED upstream call, not only a - successful one. - - The destination sink is one process-wide logger, so it has to sit on the failure - list as well as the success list; registering it on success alone means a - 401/timeout never reaches the destination and the trace lands with no error - gen-AI span. Registration is idempotent. - """ - import litellm - from litellm.integrations.otel.destination_logger import admin_destination_logger - from litellm.integrations.otel.logger import publish_global_otel_v2_provider - - sink = admin_destination_logger() - publish_global_otel_v2_provider([], lambda provider: None) - publish_global_otel_v2_provider([], lambda provider: None) - for bucket in (litellm._async_success_callback, litellm._async_failure_callback): - assert sum(1 for callback in bucket if callback is sink) == 1 - - @pytest.mark.asyncio async def test_add_litellm_data_to_request_merges_metadata_tags_on_responses_route(): """Regression for #31584: user-supplied metadata.tags must be merged into