From 7189980331dc3c11e0c4526ea5796eb713aa638c Mon Sep 17 00:00:00 2001 From: Yucheng Zhu Date: Wed, 12 Aug 2026 23:23:54 -0700 Subject: [PATCH] fix(langfuse): source the emitted metadata blob from StandardLoggingPayload Request metadata carries the whole UserAPIKeyAuth object, whose team_metadata holds the customer's own langfuse callback_vars. The only filter on the emitted blob was a four key deny list written as a circular reference crash guard, so those credentials reached the customer's own langfuse traces. The emitted blob is now the StandardLoggingPayload allowlist plus the litellm computed enrichments, and nothing is copied across from raw request metadata. That makes the credential exclusion structural rather than a filter someone has to keep correct. Steering keys keep reading raw metadata, matching literal_ai. Proxy callers are unaffected: their request metadata already rides under the allowlisted requester_metadata key, nesting intact. debug_langfuse dumped raw request metadata into the trace as a second copy of the same leak. It now emits caller scalars only. When StandardLoggingPayload is absent the trace is still emitted with the existing trace_id fallback, so failure traces survive. --- litellm/integrations/langfuse/langfuse.py | 90 +++--- .../integrations/test_langfuse.py | 282 ++++++++++++++++-- 2 files changed, 311 insertions(+), 61 deletions(-) diff --git a/litellm/integrations/langfuse/langfuse.py b/litellm/integrations/langfuse/langfuse.py index db253b1517dc..8720f561e141 100644 --- a/litellm/integrations/langfuse/langfuse.py +++ b/litellm/integrations/langfuse/langfuse.py @@ -2,8 +2,9 @@ # On success, logs events to Langfuse import os import traceback -from collections.abc import Callable, Iterable +from collections.abc import Callable, Iterable, Mapping from datetime import datetime +from types import MappingProxyType from typing import TYPE_CHECKING, Any, Final, cast from packaging.version import Version @@ -30,6 +31,7 @@ ImageResponse, ModelResponse, RerankResponse, + StandardLoggingMetadata, StandardLoggingPayload, StandardLoggingPromptManagementMetadata, TextCompletionResponse, @@ -46,6 +48,11 @@ Langfuse = Any +_DENIED_STEERING_KEYS: Final = frozenset({"headers", "endpoint", "caching_groups", "previous_models"}) +_NO_METADATA: Final[Mapping[str, Any]] = MappingProxyType({}) +_REDACTED_PROXY_HEADERS: Final[frozenset[str]] = frozenset({"authorization", "cookie", "referer"}) + + def _extract_cache_read_input_tokens(usage_obj) -> int: """ Extract cache_read_input_tokens from usage object. @@ -512,16 +519,14 @@ def _log_langfuse_v2( else [] ) - if standard_logging_object is None: - end_user_id = None - prompt_management_metadata: StandardLoggingPromptManagementMetadata | None = None - else: - end_user_id = standard_logging_object["metadata"].get("user_api_key_end_user_id", None) - - prompt_management_metadata = cast( - StandardLoggingPromptManagementMetadata | None, - standard_logging_object["metadata"].get("prompt_management_metadata", None), - ) + allowlisted_metadata: Final[StandardLoggingMetadata | dict[str, Any]] = ( + standard_logging_object["metadata"] if standard_logging_object is not None else _NO_METADATA + ) + end_user_id: Final = allowlisted_metadata.get("user_api_key_end_user_id", None) + prompt_management_metadata: Final[StandardLoggingPromptManagementMetadata | None] = cast( + StandardLoggingPromptManagementMetadata | None, + allowlisted_metadata.get("prompt_management_metadata", None), + ) # Clean Metadata before logging - never log raw metadata # the raw metadata can contain circular references which leads to infinite recursion @@ -540,12 +545,7 @@ def _log_langfuse_v2( tags.append(f"{key}:{value}") # clean litellm metadata before logging - if key in [ - "headers", - "endpoint", - "caching_groups", - "previous_models", - ]: + if key in _DENIED_STEERING_KEYS: continue else: clean_metadata[key] = value @@ -630,19 +630,18 @@ def _log_langfuse_v2( trace_params["output"] = output if not mask_output else "redacted-by-litellm" if debug is True or (isinstance(debug, str) and debug.lower() == "true"): - if "metadata" in trace_params: - # log the raw_metadata in the trace - trace_params["metadata"]["metadata_passed_to_litellm"] = metadata - else: - trace_params["metadata"] = {"metadata_passed_to_litellm": metadata} + debug_metadata: Final = { + key: value for key, value in metadata.items() if isinstance(value, (str, int, float, bool)) + } + trace_params["metadata"] = { + **(trace_params.get("metadata") or _NO_METADATA), + "metadata_passed_to_litellm": debug_metadata, + } cost: Final = kwargs.get("response_cost", None) verbose_logger.debug("trace: %s", cost) - clean_metadata["litellm_response_cost"] = cost - if standard_logging_object is not None: - hidden_params: Final = standard_logging_object.get("hidden_params", {}) - clean_metadata["hidden_params"] = filter_exceptions_from_params(hidden_params) + hidden_params: Final = standard_logging_object.get("hidden_params") if standard_logging_object else None if ( litellm.langfuse_default_tags is not None @@ -654,22 +653,24 @@ def _log_langfuse_v2( tags.append(f"proxy_base_url:{proxy_base_url}") api_base: Final = litellm_params.get("api_base", None) - if api_base: - clean_metadata["api_base"] = api_base - vertex_location: Final = kwargs.get("vertex_location", None) - if vertex_location: - clean_metadata["vertex_location"] = vertex_location - aws_region_name: Final = kwargs.get("aws_region_name", None) - if aws_region_name: - clean_metadata["aws_region_name"] = aws_region_name + + candidate_enrichments: Final = ( + ("litellm_response_cost", cost, True), + ("hidden_params", filter_exceptions_from_params(hidden_params), hidden_params is not None), + ("api_base", api_base, bool(api_base)), + ("vertex_location", vertex_location, bool(vertex_location)), + ("aws_region_name", aws_region_name, bool(aws_region_name)), + ("cache_hit", kwargs.get("cache_hit") or False, self._supports_tags() and "cache_hit" in kwargs), + ) + enrichments: Final[Mapping[str, Any]] = { + key: value for key, value, include in candidate_enrichments if include + } if self._supports_tags(): - if "cache_hit" in kwargs: - if kwargs["cache_hit"] is None: - kwargs["cache_hit"] = False - clean_metadata["cache_hit"] = kwargs["cache_hit"] + if "cache_hit" in kwargs and kwargs["cache_hit"] is None: + kwargs["cache_hit"] = False # rebind-ok: pre-existing normalization other integrations rely on if existing_trace_id is None: trace_params.update({"tags": tags}) @@ -682,13 +683,13 @@ def _log_langfuse_v2( if headers: for key, value in headers.items(): # these headers can leak our API keys and/or JWT tokens - if key.lower() not in ["authorization", "cookie", "referer"]: + if key.lower() not in _REDACTED_PROXY_HEADERS: clean_headers[key] = value trace: Final[StatefulTraceClient] = self.Langfuse.trace(**trace_params) # Log provider specific information as a span - log_provider_specific_information_as_span(trace, clean_metadata) + log_provider_specific_information_as_span(trace, enrichments) # Log guardrail information as a span self._log_guardrail_information_as_span( @@ -761,7 +762,10 @@ def _log_langfuse_v2( "output": output if not mask_output else "redacted-by-litellm", "usage": usage, "usage_details": usage_details, - "metadata": log_requester_metadata(clean_metadata), + "metadata": { + **log_requester_metadata(redact_user_api_key_info(metadata=allowlisted_metadata)), + **enrichments, + }, "level": level, "version": clean_metadata.pop("version", None), } @@ -1058,7 +1062,7 @@ def _add_prompt_to_generation_params( def log_provider_specific_information_as_span( trace, - clean_metadata, + clean_metadata: Mapping[str, Any], ): """ Logs provider-specific information as spans. @@ -1098,7 +1102,7 @@ def log_provider_specific_information_as_span( ) -def log_requester_metadata(clean_metadata: dict): +def log_requester_metadata(clean_metadata: Mapping[str, Any]): returned_metadata: Final = {} requester_metadata: Final = clean_metadata.get("requester_metadata") or {} for k, v in clean_metadata.items(): diff --git a/tests/test_litellm/integrations/test_langfuse.py b/tests/test_litellm/integrations/test_langfuse.py index c83a3fa2b73a..de04a65c3103 100644 --- a/tests/test_litellm/integrations/test_langfuse.py +++ b/tests/test_litellm/integrations/test_langfuse.py @@ -314,7 +314,7 @@ def mock_get(key, default=None): "litellm_params": {"metadata": {}}, "optional_params": {}, "litellm_call_id": "test-call-id-null-usage", - "standard_logging_object": None, + "standard_logging_object": self._build_standard_logging_payload(), "response_cost": 0.0, } @@ -382,16 +382,14 @@ def _build_standard_logging_payload(self, trace_id: Optional[str] = None): "model_id": "model-123", "model_group": "openai", "api_base": "https://api.openai.com", + # only real StandardLoggingMetadata fields: session_id, trace_name, + # headers and friends are request-metadata keys the allowlist drops, + # so a payload carrying them cannot occur in production "metadata": { "user_api_key_end_user_id": None, "prompt_management_metadata": None, - "session_id": None, - "trace_name": None, - "trace_version": None, - "headers": None, - "endpoint": None, - "caching_groups": None, - "previous_models": None, + "user_api_key_hash": "hashed-key", + "user_api_key_alias": "canary-alias", }, "hidden_params": {}, "request_tags": [], @@ -503,14 +501,251 @@ def test_log_langfuse_v2_uses_litellm_trace_id_fallback_over_call_id(self): # litellm_trace_id should be preferred over litellm_call_id assert self.last_trace_kwargs.get("id") == "trace-id-from-kwargs" - def test_log_langfuse_v2_uses_litellm_trace_id_when_standard_logging_object_none( - self, - ): + CANARY = "sk-lf-canary-SECRET-d4e5f6" + + def _canary_request_metadata(self): + """Raw request metadata shaped like the proxy builds it, credentials included.""" + from litellm.proxy._types import UserAPIKeyAuth + + team_logging = [ + { + "callback_name": "langfuse", + "callback_vars": {"langfuse_secret_key": self.CANARY}, + } + ] + return { + "user_api_key_auth": UserAPIKeyAuth( + api_key="hashed-key", + team_metadata={"logging": team_logging}, + ), + "user_api_key_team_metadata": {"logging": team_logging}, + "user_api_key_metadata": {"secret_manager_settings": {"vault_token": self.CANARY}}, + "session_id": "canary-session", + "trace_name": "canary-trace", + "first_custom": "keep-first", + "second_custom": "keep-second", + "endpoint": "/v1/chat/completions", + "headers": {"authorization": f"Bearer {self.CANARY}"}, + } + + def _emitted_payload_text(self): + """Every blob this logger handed to the langfuse SDK, as one searchable string.""" + import json + + blobs = [self.last_trace_kwargs] + if self.mock_langfuse_trace.generation.call_args is not None: + blobs.append(self.mock_langfuse_trace.generation.call_args.kwargs) + blobs.extend(call.kwargs for call in self.mock_langfuse_trace.span.call_args_list) + return json.dumps(blobs, default=repr) + + def _drive_with_canary(self, extra_metadata=None, hidden_params=None): + metadata = {**self._canary_request_metadata(), **(extra_metadata or {})} + payload = self._build_standard_logging_payload(trace_id="canary-trace-id") + if hidden_params is not None: + payload["hidden_params"] = hidden_params + kwargs = {**self._build_langfuse_kwargs(payload), "response_cost": 0.25} + self.last_trace_kwargs = {} + self.mock_langfuse_trace.generation.reset_mock() + self.mock_langfuse_trace.span.reset_mock() + + with patch( + "litellm.integrations.langfuse.langfuse._add_prompt_to_generation_params", + side_effect=lambda generation_params, **kw: generation_params, + create=True, + ): + self.logger._log_langfuse_v2( + user_id="user-1", + metadata=metadata, + litellm_params={"metadata": metadata}, + output=None, + start_time=datetime.datetime(2024, 1, 1, 12, 0, 0), + end_time=datetime.datetime(2024, 1, 1, 12, 0, 1), + kwargs=kwargs, + optional_params={}, + input=None, + response_obj=None, + level="INFO", + litellm_call_id="canary-call-id", + ) + return self.mock_langfuse_trace.generation.call_args.kwargs["metadata"] + + def test_team_callback_credentials_never_reach_langfuse(self): + """ + Regression for the credential leak: request metadata carries the whole + UserAPIKeyAuth object, whose team_metadata holds the customer's own langfuse + keys. The emitted blob is sourced from StandardLoggingPayload, so none of the + three credential carriers can ride along. + """ + generation_metadata = self._drive_with_canary() + + assert self.CANARY not in self._emitted_payload_text() + for leaked_key in ( + "user_api_key_auth", + "user_api_key_team_metadata", + "user_api_key_metadata", + ): + assert leaked_key not in generation_metadata + + def test_debug_langfuse_dump_carries_no_credentials(self): + """ + debug_langfuse dumps request metadata into the trace as a second emit site. + It must be sourced from the allowlisted payload too. + """ + self._drive_with_canary(extra_metadata={"debug_langfuse": True}) + + dumped = self.last_trace_kwargs["metadata"]["metadata_passed_to_litellm"] + assert "user_api_key_auth" not in dumped + assert self.CANARY not in self._emitted_payload_text() + + def test_raw_request_metadata_reaches_the_emitted_blob_through_no_key(self): + """ + The emitted blob is the allowlist plus litellm enrichments, nothing else. + Nothing from raw request metadata is copied across, whatever its type, which + is what makes the credential exclusion structural rather than a filter that + has to be kept correct. Proxy callers keep their own metadata under the + allowlisted requester_metadata key. + """ + generation_metadata = self._drive_with_canary() + + for caller_key in ("first_custom", "second_custom", "session_id", "trace_name"): + assert caller_key not in generation_metadata + + def test_provider_specific_span_receives_the_emitted_blob(self): + """ + The provider span reads hidden_params, which is an enrichment on the emitted + blob rather than a key of request metadata. Handing it the steering dict + instead would silently stop emitting vertex grounding spans. + """ + self._drive_with_canary(hidden_params={"vertex_ai_grounding_metadata": ["ground-a", "ground-b"]}) + + span_inputs = [call.kwargs.get("input") for call in self.mock_langfuse_trace.span.call_args_list] + assert span_inputs == ["ground-a", "ground-b"] + assert self.CANARY not in self._emitted_payload_text() + + def test_caller_cannot_spoof_an_allowlisted_identity_field(self): + """ + Request metadata never reaches the blob, so a caller naming user_api_key_alias + cannot have their value emitted in place of the proxy-resolved one. + """ + generation_metadata = self._drive_with_canary( + extra_metadata={"user_api_key_alias": "spoofed-by-caller"} + ) + + assert generation_metadata["user_api_key_alias"] == "canary-alias" + + def test_caller_nested_metadata_cannot_erase_a_litellm_enrichment(self): + """ + log_requester_metadata drops any top-level key whose name also appears inside + requester_metadata. Sourcing the blob from the allowlist populates that nested + dict for real, so a caller naming a key litellm_response_cost would otherwise + blank out the cost litellm computed. Enrichments are layered after the dedupe. + """ + payload = self._build_standard_logging_payload(trace_id="canary-trace-id") + payload["metadata"]["requester_metadata"] = {"litellm_response_cost": "caller-value", "api_base": "caller"} + kwargs = {**self._build_langfuse_kwargs(payload), "response_cost": 0.25} + metadata = self._canary_request_metadata() + self.mock_langfuse_trace.generation.reset_mock() + + with patch( + "litellm.integrations.langfuse.langfuse._add_prompt_to_generation_params", + side_effect=lambda generation_params, **kw: generation_params, + create=True, + ): + self.logger._log_langfuse_v2( + user_id="user-1", + metadata=metadata, + litellm_params={"metadata": metadata, "api_base": "https://real-api-base"}, + output=None, + start_time=datetime.datetime(2024, 1, 1, 12, 0, 0), + end_time=datetime.datetime(2024, 1, 1, 12, 0, 1), + kwargs=kwargs, + optional_params={}, + input=None, + response_obj=None, + level="INFO", + litellm_call_id="canary-call-id", + ) + + generation_metadata = self.mock_langfuse_trace.generation.call_args.kwargs["metadata"] + assert generation_metadata["litellm_response_cost"] == 0.25 + assert generation_metadata["api_base"] == "https://real-api-base" + + def test_denied_steering_keys_and_enrichments(self): + """ + endpoint is a plain string, so without the deny-list it would ride the + string re-injection straight into the emitted blob. The enrichments are + litellm-computed and must survive the move off clean_metadata. + """ + generation_metadata = self._drive_with_canary() + + assert "endpoint" not in generation_metadata + assert "headers" not in generation_metadata + assert generation_metadata["litellm_response_cost"] == 0.25 + assert "hidden_params" in generation_metadata + + def test_cache_hit_is_normalized_on_the_shared_kwargs(self): """ - When standard_logging_object is None (failure case where - get_standard_logging_object_payload threw), litellm_trace_id from kwargs - should be used as the Langfuse trace_id. This matches the DB Session ID. + kwargs here is the shared model_call_details dict. Callbacks that run after + langfuse read cache_hit off it and copy it into their own payloads, so + dropping the None to False normalization records None for datadog, logfire, + generic_api and spend tracking. """ + metadata = self._canary_request_metadata() + payload = self._build_standard_logging_payload(trace_id="canary-trace-id") + kwargs = {**self._build_langfuse_kwargs(payload), "cache_hit": None} + + with patch( + "litellm.integrations.langfuse.langfuse._add_prompt_to_generation_params", + side_effect=lambda generation_params, **kw: generation_params, + create=True, + ): + self.logger._log_langfuse_v2( + user_id="user-1", + metadata=metadata, + litellm_params={"metadata": metadata}, + output=None, + start_time=datetime.datetime(2024, 1, 1, 12, 0, 0), + end_time=datetime.datetime(2024, 1, 1, 12, 0, 1), + kwargs=kwargs, + optional_params={}, + input=None, + response_obj=None, + level="INFO", + litellm_call_id="canary-call-id", + ) + + assert kwargs["cache_hit"] is False + + def test_redact_user_api_key_info_still_strips_the_emitted_blob(self): + """ + The flag used to act on the raw-derived blob. That blob is now sourced from + StandardLoggingPayload, which is where the user_api_key_* fields live, so the + redaction has to run on the assembled payload or the flag silently stops working. + """ + with patch.object(litellm, "redact_user_api_key_info", True): + generation_metadata = self._drive_with_canary() + + assert not [key for key in generation_metadata if key.startswith("user_api_key")] + + def test_steering_keys_still_read_from_raw_metadata(self): + """ + Only the emitted payload moves to StandardLoggingPayload. The control fields + keep reading raw metadata, which is what Braintrust's migration got wrong. + """ + self._drive_with_canary() + + assert self.last_trace_kwargs.get("session_id") == "canary-session" + assert self.last_trace_kwargs.get("name") == "canary-trace" + + def test_failure_trace_survives_a_missing_standard_logging_object(self): + """ + get_standard_logging_object_payload is fail-open and returns None on any + exception, which is exactly the failed-request case Langfuse most needs to + show. The trace is still emitted with the litellm_trace_id fallback, and the + blob degrades to caller strings plus enrichments rather than falling back to + raw metadata, which would ship the UserAPIKeyAuth object. + """ + metadata = self._canary_request_metadata() kwargs = { "standard_logging_object": None, "model": "gpt-4", @@ -520,16 +755,17 @@ def test_log_langfuse_v2_uses_litellm_trace_id_when_standard_logging_object_none "litellm_trace_id": "trace-id-failure", } self.last_trace_kwargs = {} + self.mock_langfuse_trace.generation.reset_mock() with patch( "litellm.integrations.langfuse.langfuse._add_prompt_to_generation_params", side_effect=lambda generation_params, **kwargs: generation_params, create=True, ): - self.logger._log_langfuse_v2( + trace_id, _ = self.logger._log_langfuse_v2( user_id="user-1", - metadata={}, - litellm_params={"metadata": {}}, + metadata=metadata, + litellm_params={"metadata": metadata}, output=None, start_time=datetime.datetime.utcnow(), end_time=datetime.datetime.utcnow(), @@ -541,8 +777,18 @@ def test_log_langfuse_v2_uses_litellm_trace_id_when_standard_logging_object_none litellm_call_id="call-id-different", ) - # Must use litellm_trace_id, not litellm_call_id + import json + + assert trace_id == "trace-id-failure" assert self.last_trace_kwargs.get("id") == "trace-id-failure" + generation_metadata = self.mock_langfuse_trace.generation.call_args.kwargs["metadata"] + assert "user_api_key_auth" not in generation_metadata + assert self.CANARY not in self._emitted_payload_text() + assert "first_custom" not in generation_metadata + # hidden_params comes off the payload, so it is omitted rather than emitted + # as an unserializable placeholder + assert "hidden_params" not in generation_metadata + json.dumps(generation_metadata) def test_log_langfuse_v2_session_id_passed_as_trace_session_id(self): """