scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-02 07:13:02 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 150

fix(security)!: build proxy-only failure logs from an allowlist and drop caller guardrail info (#48599)

Browse source 
* fix(proxy): build proxy-only failure logs from an allowlist and ignore request-time GCS credentials

* fix(proxy): drop caller metadata from proxy-only failure logs

metadata and litellm_metadata leave the proxy-only failure-log allowlist, so a rejected request can no longer point its failure log at its own Langsmith or Langfuse host through nested metadata.

* fix(proxy): keep allowlisted metadata in proxy-only failure logs

* test(proxy): isolate proxy-only failure log metadata tests

* fix(logging): ignore non-list request tags when building the logging payload

A guardrail-rejected request with metadata tags set to a non-list value crashed
the standard logging payload, so its spend row lost model attribution.

* fix(proxy): keep router attribution and drop unauthenticated tags in proxy-only failure logs

A request rejected before routing now logs the configured deployment and model group even when its metadata or litellm_metadata carries its own values. Requests rejected at auth no longer write caller tags into failure logs and tag spend.

* fix(proxy): always strip client guardrail info and mirror failure attribution into litellm_metadata

allow_client_pricing_override no longer keeps a caller-supplied standard_logging_guardrail_information on the chat or pass-through paths, since its guardrail_cost is billed. Proxy-only failure logs now also carry key and team attribution in litellm_metadata.

* refactor(proxy): keep the proxy-only failure-log fixes inside the type gates

Types the picked helpers so basedpyright and the type-discipline gate stay at main's counts. No runtime change.

* fix(proxy): keep the WS frame guardrail-info strip and string interaction tags as sent

The guardrail-info key left the pricing strip list, so the Responses WS frame guard now drops it by name. get_request_tags no longer raises on string tags, so background interaction polling checks for them before keeping them as sent.

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
(cherry picked from commit 788868a52f)
This commit is contained in:
yucheng-berri 2026-10-27 12:40:59 -07:00 • committed by Yuneng Jiang
parent 89a2d67947
commit 375df19d63
No known key found for this signature in database
12 changed files with 469 additions and 202 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

49
litellm/integrations/gcs_bucket/gcs_bucket_base.py
View file

@ -20,7 +20,7 @@ from litellm.llms.custom_httpx.http_handler import (
httpxSpecialProvider,
)
from litellm.types.integrations.gcs_bucket import *
from litellm.types.utils import StandardCallbackDynamicParams, StandardLoggingPayload
from litellm.types.utils import StandardLoggingPayload
if TYPE_CHECKING:
from litellm.llms.vertex_ai.vertex_llm_base import VertexBase
@ -140,49 +140,16 @@ class GCSBucketBase(CustomBatchLogger):
async def get_gcs_logging_config(self, kwargs: dict[str, Any] | None = {}) -> GCSLoggingConfig:
"""
This function is used to get the GCS logging config for the GCS Bucket Logger.
It checks if the dynamic parameters are provided in the kwargs and uses them to get the GCS logging config.
If no dynamic parameters are provided, it uses the default values.
Returns the bucket and service account this logger was configured with.
"""
if kwargs is None:
kwargs = {}
standard_callback_dynamic_params: Final[StandardCallbackDynamicParams | None] = kwargs.get(
"standard_callback_dynamic_params", None
)
bucket_name: str
path_service_account: str | None
if standard_callback_dynamic_params is not None:
verbose_logger.debug("Using dynamic GCS logging")
verbose_logger.debug("standard_callback_dynamic_params: %s", standard_callback_dynamic_params)
_bucket_name: str | None = standard_callback_dynamic_params.get("gcs_bucket_name", None) or self.BUCKET_NAME
_path_service_account: Final[str | None] = (
standard_callback_dynamic_params.get("gcs_path_service_account", None) or self.path_service_account_json
if self.BUCKET_NAME is None:
raise ValueError(
"GCS_BUCKET_NAME is not set in the environment, but GCS Bucket is being used as a logging callback. Please set 'GCS_BUCKET_NAME' in the environment."
)
if _bucket_name is None:
raise ValueError(
"GCS_BUCKET_NAME is not set in the environment, but GCS Bucket is being used as a logging callback. Please set 'GCS_BUCKET_NAME' in the environment."
)
bucket_name = _bucket_name
path_service_account = _path_service_account
vertex_instance = await self.get_or_create_vertex_instance(credentials=path_service_account)
else:
# If no dynamic parameters, use the default instance
if self.BUCKET_NAME is None:
raise ValueError(
"GCS_BUCKET_NAME is not set in the environment, but GCS Bucket is being used as a logging callback. Please set 'GCS_BUCKET_NAME' in the environment."
)
bucket_name = self.BUCKET_NAME
path_service_account = self.path_service_account_json
vertex_instance = await self.get_or_create_vertex_instance(credentials=path_service_account)
return GCSLoggingConfig(
bucket_name=bucket_name,
vertex_instance=vertex_instance,
path_service_account=path_service_account,
bucket_name=self.BUCKET_NAME,
vertex_instance=await self.get_or_create_vertex_instance(credentials=self.path_service_account_json),
path_service_account=self.path_service_account_json,
)
async def get_or_create_vertex_instance(self, credentials: str | None) -> VertexBase:

20
litellm/litellm_core_utils/litellm_logging.py
View file

@ -6719,19 +6719,17 @@ class StandardLoggingPayloadSetup:
litellm_metadata: Final = cast( # cast-ok: request metadata is caller-provided
Mapping[str, object], litellm_params.get("litellm_metadata") or {}
)
if metadata.get("tags", []):
request_tags = cast(list[str], metadata.get("tags", [])).copy() # cast-ok: tags are caller-provided
elif litellm_metadata.get("tags", []):
request_tags = cast(list[str], litellm_metadata.get("tags", [])).copy() # cast-ok: tags are caller-provided
else:
request_tags = []
request_tags: Final = next(
(
cast(list[str], tags) # cast-ok: tags are caller-provided
for tags in (metadata.get("tags"), litellm_metadata.get("tags"))
if isinstance(tags, list) and tags
),
(),
)
user_agent_tags: Final = StandardLoggingPayloadSetup._get_user_agent_tags(proxy_server_request)
additional_header_tags: Final = StandardLoggingPayloadSetup._get_extra_header_tags(proxy_server_request)
if user_agent_tags is not None:
request_tags.extend(user_agent_tags)
if additional_header_tags is not None:
request_tags.extend(additional_header_tags)
return request_tags
return [*request_tags, *(user_agent_tags or ()), *(additional_header_tags or ())]
_get_request_tags = get_request_tags

19
litellm/proxy/litellm_pre_call_utils.py
View file

@ -414,10 +414,7 @@ _ALLOW_CLIENT_MESSAGE_REDACTION_OPT_OUT_METADATA_KEY: Final = "allow_client_mess
_CLIENT_PRICING_CONTROL_FIELDS: Final = frozenset(CustomPricingLiteLLMParams.model_fields.keys()) | {"base_model"}
# ``model_info`` carries the same pricing fields when read by
# ``use_custom_pricing_for_model``; strip from metadata for the same reason.
# ``standard_logging_guardrail_information`` is proxy-written telemetry summed
# into response_cost and spend; a client seeding it forges (even negative)
# guardrail cost.
_CLIENT_PRICING_METADATA_FIELDS: Final = frozenset({"model_info", "standard_logging_guardrail_information"})
_CLIENT_PRICING_METADATA_FIELDS: Final = frozenset({"model_info"})
# ``attempted_fallbacks`` and ``original_model_group`` are written by the router
# and read by spend logs as fact; a client value has no legitimate meaning and no
# key or team setting keeps it, so the strip is never gated.
@ -677,6 +674,19 @@ def _strip_client_pricing_overrides(data: dict[str, object]) -> None:
)
def strip_client_guardrail_information(data: dict[str, object]) -> None: # mutable-ok: strips the request body in place
"""Guardrail hooks write this field and its cost is billed, so no key or team setting keeps a caller's value."""
for metadata_key in ("metadata", "litellm_metadata"):
if (
isinstance(metadata := data.get(metadata_key), dict)
and "standard_logging_guardrail_information" in metadata
):
del metadata["standard_logging_guardrail_information"]
verbose_proxy_logger.debug(
"Stripped client-supplied guardrail information from request body: %s", metadata_key
)
def _strip_router_reserved_metadata(
data: dict[str, object], # mutable-ok: strips in place on the request body the pre-call pipeline threads through
) -> None:
@ -2394,6 +2404,7 @@ async def add_litellm_data_to_request(
# would silently skip the field.
if not _key_or_team_allows_client_pricing_override(user_api_key_dict):
_strip_client_pricing_overrides(data)
strip_client_guardrail_information(data)
_strip_router_reserved_metadata(data)
# Same reason as the strips above: runs after the metadata string-to-dict parse

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

@ -118,6 +118,7 @@ from litellm.proxy.litellm_pre_call_utils import ( # noqa: F401 # legacy modul
_key_or_team_allows_client_pricing_override, # pyright: ignore[reportPrivateUsage] # reuse the proxy's pricing trust policy
_strip_client_pricing_overrides, # pyright: ignore[reportPrivateUsage] # sanitize before trusted hooks add guardrail costs
get_dynamic_logging_metadata, # pyright: ignore[reportPrivateUsage] # shared proxy helper, same import style as _read_request_body above
strip_client_guardrail_information,
)
from litellm.proxy.route_llm_request import ProxyModelNotFoundError
from litellm.proxy.route_priority import (
@ -1195,10 +1196,11 @@ async def pass_through_request(
_parsed_body,
)
client_body: Final = TypeAdapter(dict[str, object]).validate_python(_parsed_body)
strip_client_guardrail_information(client_body)
if not _key_or_team_allows_client_pricing_override(user_api_key_dict):
pricing_body: Final = TypeAdapter(dict[str, object]).validate_python(_parsed_body)
_strip_client_pricing_overrides(pricing_body)
_parsed_body = pricing_body
_strip_client_pricing_overrides(client_body)
_parsed_body = client_body
if custom_llm_provider in ("laya", "bespoke"):
decision_request: Final = TypeAdapter(Mapping[str, object]).validate_python(_parsed_body)
checkpoint: Final = validate_oss_request(custom_llm_provider, decision_request)

114
litellm/proxy/utils.py
View file

@ -1270,7 +1270,8 @@ def _stamp_deployment_attribution(
callbacks key off it and must not count a proxy-side reject against the deployment. A failure
after the provider handoff keeps the metadata the router stamped; a request that never reached a
provider is flagged ``PROXY_REJECTED_BEFORE_ROUTING_KEY`` (deployment metrics key off it) whatever
its metadata says, since ``metadata.model_info`` can be caller supplied."""
its metadata says, since ``metadata.model_info`` can be caller supplied. Its ``deployment`` and
``model_group`` always come from the router config, never from the caller's metadata."""
attribution: Final = _deployment_attribution_for_model_group(model_group, team_id)
if "custom_llm_provider" in attribution:
litellm_params["custom_llm_provider"] = attribution["custom_llm_provider"]
@ -1285,9 +1286,13 @@ def _stamp_deployment_attribution(
if not isinstance(metadata, dict):
return attribution
metadata.setdefault("model_info", attribution["model_info"])
metadata.setdefault("deployment", attribution["deployment"])
if isinstance(model_group, str):
metadata.setdefault("model_group", model_group)
routing: Final = MappingProxyType(
{"deployment": attribution["deployment"], **({"model_group": model_group} if model_group else {})}
)
metadata.update(routing)
litellm_metadata: Final = litellm_params.get("litellm_metadata")
if isinstance(litellm_metadata, dict):
litellm_metadata.update(routing)
return attribution
@ -1360,6 +1365,9 @@ def _call_type_for_route(route: str | None) -> str | None:
_PROXY_ONLY_LLM_API_ERRORS: Final = (HTTPException, ProxyException, GuardrailRaisedException)
_PROXY_ONLY_ERROR_LOGGED_FIELDS: Final = frozenset(
{"model", "messages", "input", "prompt", "user", "stream", "litellm_call_id"}
)
_LOG_DB_METRICS_CALL_TYPES: Final = frozenset(("get_data", "insert_data", "update_data", "delete_data"))
@ -1394,6 +1402,57 @@ def _provider_billed_the_failure(exception: Exception) -> bool:
return _timed_out_reading_from_provider(exception) or _client_disconnected(exception)
def _as_str_mapping(value: object) -> Mapping[str, object]:
if not isinstance(value, Mapping):
return MappingProxyType({})
return cast( # cast-ok: the runtime Mapping check validates the container shape
Mapping[str, object], value
)
def _proxy_only_error_logged_metadata_keys(*, authenticated: bool, past_pre_call: bool) -> frozenset[str]:
return frozenset(
key
for key, logged in (("tags", authenticated), ("standard_logging_guardrail_information", past_pre_call))
if logged
)
def _is_logged_metadata_value(key: str, value: object) -> bool:
if key != "tags":
return True
tags: Final = cast(list[object], value) if isinstance(value, list) else None # cast-ok: isinstance checked the list
return tags is not None and all(isinstance(item, str) for item in tags)
def _proxy_only_error_log_metadata(
value: object, logged_keys: frozenset[str]
) -> dict[str, object] | None: # mutable-ok: router attribution stamps the logged metadata in place
metadata: Final = _as_str_mapping(value)
if not isinstance(value, Mapping):
return None
return {
key: value for key, value in metadata.items() if key in logged_keys and _is_logged_metadata_value(key, value)
}
def _proxy_only_error_log_fields(
request_data: Mapping[str, object], logged_metadata_keys: frozenset[str]
) -> Mapping[str, object]:
metadata_fields: Final = {
key: filtered_metadata
for key in ("metadata", "litellm_metadata")
if (filtered_metadata := _proxy_only_error_log_metadata(request_data.get(key), logged_metadata_keys))
is not None
}
return MappingProxyType(
{
**{key: value for key, value in request_data.items() if key in _PROXY_ONLY_ERROR_LOGGED_FIELDS},
**metadata_fields,
}
)
def _failure_fields_to_lift(request_data: Mapping[str, object], provider_billed: bool = False) -> Mapping[str, object]:
"""Failure-path callbacks run after ``litellm_logging_obj`` is popped from
request_data (it is not serialisable), so the caller merges these fields
@ -3650,6 +3709,7 @@ class ProxyLogging:
user_api_key_dict=user_api_key_dict,
route=route,
original_exception=original_exception,
error_type=error_type,
)
request_data.update(
@ -3742,6 +3802,7 @@ class ProxyLogging:
user_api_key_dict: UserAPIKeyAuth,
route: str | None = None,
original_exception: Exception | None = None,
error_type: ProxyErrorTypes | None = None,
):
"""
Handle logging for proxy only errors by calling `litellm_logging_obj.async_failure_handler`
@ -3750,19 +3811,25 @@ class ProxyLogging:
"""
litellm_logging_obj: Logging | None = request_data.get("litellm_logging_obj", None)
set_up_by_request: Final = litellm_logging_obj is not None
request_fields: Final = cast(Mapping[str, object], request_data) # cast-ok: request bodies are str-keyed JSON
logged_metadata_keys: Final = _proxy_only_error_logged_metadata_keys(
authenticated=error_type != ProxyErrorTypes.auth_error, past_pre_call=set_up_by_request
)
user_api_key_logged_metadata: Final = LiteLLMProxyRequestSetup.get_sanitized_user_information_from_key(
user_api_key_dict=user_api_key_dict
)
if litellm_logging_obj is None:
from litellm._uuid import uuid
request_data.setdefault("litellm_call_id", str(uuid.uuid4()))
user_api_key_logged_metadata: Final = LiteLLMProxyRequestSetup.get_sanitized_user_information_from_key(
user_api_key_dict=user_api_key_dict
)
litellm_logging_obj, data = litellm.utils.function_setup(
original_function=route or "IGNORE_THIS",
rules_obj=litellm.utils.Rules(),
start_time=datetime.now(),
**request_data,
**_proxy_only_error_log_fields( # pyright: ignore[reportArgumentType] # the allowlist never carries is_async_call
request_fields, logged_metadata_keys
),
)
request_data["litellm_logging_obj"] = litellm_logging_obj # rebind-ok: lifted then popped by the caller
if "metadata" not in request_data:
@ -3771,18 +3838,39 @@ class ProxyLogging:
if litellm_logging_obj is not None:
## UPDATE LOGGING INPUT
_optional_params: Final = {}
_litellm_params: Final = {}
_optional_params: Final[dict[str, object]] = {} # mutable-ok: filled below, then handed to the logger
_litellm_params: Final[dict[str, object]] = {} # mutable-ok: filled below, then handed to the logger
litellm_param_keys: Final = LoggedLiteLLMParams.__annotations__.keys()
for k, v in request_data.items():
for k, v in _proxy_only_error_log_fields(request_fields, logged_metadata_keys).items():
if k in litellm_param_keys:
_litellm_params[k] = v
elif k not in ("model", "user", "litellm_logging_obj"):
_optional_params[k] = v
existing: Final = _as_str_mapping(
litellm_logging_obj.litellm_params # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # Logging.litellm_params is untyped
)
litellm_params: Final = {
**_litellm_params,
"metadata": {
**_as_str_mapping(existing.get("metadata")),
**_as_str_mapping(_litellm_params.get("metadata")),
**user_api_key_logged_metadata,
},
**{
key: {
**_as_str_mapping(existing.get(key)),
**_as_str_mapping(value),
**user_api_key_logged_metadata,
}
for key, value in _litellm_params.items()
if key == "litellm_metadata"
},
}
attribution: Final = _stamp_deployment_attribution(
_litellm_params,
litellm_params,
request_data.get("model"),
user_api_key_dict.team_id,
dispatched=_reached_deployment(litellm_logging_obj),
@ -3792,7 +3880,7 @@ class ProxyLogging:
model=request_data.get("model", ""),
user=request_data.get("user", ""),
optional_params=_optional_params,
litellm_params=_litellm_params,
litellm_params=litellm_params,
**(
{"custom_llm_provider": attribution["custom_llm_provider"]}
if "custom_llm_provider" in attribution

18
tests/unit/integrations/gcs_bucket/test_gcs_bucket_base.py
View file

@ -120,6 +120,24 @@ class TestGCSBucketBase:
custom_llm_provider="vertex_ai",
)
@pytest.mark.asyncio
async def test_gcs_logging_config_ignores_request_time_bucket_and_credentials(self, monkeypatch):
monkeypatch.setenv("GCS_PATH_SERVICE_ACCOUNT", "/admin/sa.json")
handler = GCSBucketBase(bucket_name="admin-bucket")
with patch.object(handler, "get_or_create_vertex_instance", new_callable=AsyncMock) as mock_vertex:
config = await handler.get_gcs_logging_config(
{
"standard_callback_dynamic_params": {
"gcs_bucket_name": "caller-bucket",
"gcs_path_service_account": "/caller/sa.json",
}
}
)
assert (config["bucket_name"], config["path_service_account"]) == ("admin-bucket", "/admin/sa.json")
mock_vertex.assert_awaited_once_with(credentials="/admin/sa.json")
@pytest.mark.asyncio
async def test_log_json_data_on_gcs_url_encodes_object_name(self):
handler = GCSBucketBase(bucket_name="test-bucket")

15
tests/unit/litellm_core_utils/test_litellm_logging.py
View file

@ -3193,6 +3193,21 @@ def test_get_request_tags_from_metadata_and_litellm_metadata():
assert "User-Agent: litellm/1.0.0" in tags
@pytest.mark.parametrize("unusable_metadata_tags", [123, "tag", {"tag": True}, []])
def test_get_request_tags_falls_back_to_litellm_metadata_when_metadata_tags_are_unusable(unusable_metadata_tags):
from litellm.litellm_core_utils.litellm_logging import StandardLoggingPayloadSetup
tags = StandardLoggingPayloadSetup._get_request_tags(
litellm_params={
"metadata": {"tags": unusable_metadata_tags},
"litellm_metadata": {"tags": ["litellm-metadata-tag"]},
},
proxy_server_request={},
)
assert tags == ["litellm-metadata-tag"]
def test_get_request_tags_does_not_mutate_original_tags():
"""
Test that _get_request_tags does not mutate the original tags list in metadata.

84
tests/unit/llms/vertex_ai/files/test_vertex_ai_files_handler.py
View file

@ -260,87 +260,3 @@ class TestVertexAIFilesHandler:
)
assert service_account == '{"type": "service_account", "project_id": "p"}'
@pytest.mark.asyncio
async def test_afile_content_honors_per_model_bucket_over_env(self, monkeypatch):
"""
Regression for #32640: a batch output written to a per-model gcs_bucket_name must be
readable even when the global GCS_BUCKET_NAME points at a different bucket. Before the
fix the read path resolved the bucket from env only and raised
"file_id bucket does not match the configured storage bucket".
"""
monkeypatch.setenv("GCS_BUCKET_NAME", "env-default-bucket")
monkeypatch.delenv("GCS_PATH_SERVICE_ACCOUNT", raising=False)
file_id = "gs%3A%2F%2Fmy-model-bucket%2Flitellm-vertex-files%2Fuploads%2Fabc-batch-output.jsonl"
file_content_request = FileContentRequest(file_id=file_id, extra_headers=None, extra_body=None)
with (
patch.object(self.handler, "download_gcs_object", new_callable=AsyncMock) as mock_download,
patch.object(
self.handler,
"get_or_create_vertex_instance",
new_callable=AsyncMock,
return_value=object(),
),
):
mock_download.return_value = b"batch output"
result = await self.handler.afile_content(
file_content_request=file_content_request,
vertex_credentials="/model/sa.json",
vertex_project="test-project",
vertex_location="us-central1",
timeout=60.0,
max_retries=0,
litellm_params={
"gcs_bucket_name": "my-model-bucket",
"vertex_credentials": "/model/sa.json",
},
)
assert isinstance(result, HttpxBinaryResponseContent)
assert result.response.content == b"batch output"
dynamic_params = mock_download.call_args.kwargs["standard_callback_dynamic_params"]
assert dynamic_params["gcs_bucket_name"] == "my-model-bucket"
assert dynamic_params["gcs_path_service_account"] == "/model/sa.json"
assert mock_download.call_args.kwargs["object_name"] == "litellm-vertex-files/uploads/abc-batch-output.jsonl"
@pytest.mark.asyncio
async def test_afile_content_reads_without_global_env_bucket(self, monkeypatch):
"""
Regression for #32640: with no global GCS_BUCKET_NAME set, a model-group-level
deployment (per-model gcs_bucket_name) must still be readable. Before the fix the read
path raised "GCS_BUCKET_NAME is not set in the environment".
"""
monkeypatch.delenv("GCS_BUCKET_NAME", raising=False)
monkeypatch.delenv("GCS_PATH_SERVICE_ACCOUNT", raising=False)
file_id = "gs%3A%2F%2Fmy-model-bucket%2Flitellm-vertex-files%2Fuploads%2Fabc-batch-output.jsonl"
file_content_request = FileContentRequest(file_id=file_id, extra_headers=None, extra_body=None)
with (
patch.object(self.handler, "download_gcs_object", new_callable=AsyncMock) as mock_download,
patch.object(
self.handler,
"get_or_create_vertex_instance",
new_callable=AsyncMock,
return_value=object(),
),
):
mock_download.return_value = b"batch output"
result = await self.handler.afile_content(
file_content_request=file_content_request,
vertex_credentials="/model/sa.json",
vertex_project="test-project",
vertex_location="us-central1",
timeout=60.0,
max_retries=0,
litellm_params={"gcs_bucket_name": "my-model-bucket"},
)
assert isinstance(result, HttpxBinaryResponseContent)
dynamic_params = mock_download.call_args.kwargs["standard_callback_dynamic_params"]
assert dynamic_params["gcs_bucket_name"] == "my-model-bucket"

34
tests/unit/proxy/pass_through_endpoints/test_pass_through_endpoints.py
View file

@ -1922,6 +1922,40 @@ async def test_pass_through_request_uses_resolved_timeout():
assert mock_get_client.call_args[1]["params"]["timeout"] == 1500
@pytest.mark.asyncio
@pytest.mark.parametrize("allow_client_pricing_override", [True, False])
async def test_pass_through_request_strips_client_guardrail_information(allow_client_pricing_override: bool):
mock_request = MagicMock(spec=Request)
mock_request.method = "POST"
mock_request.headers = Headers({})
mock_request.query_params = QueryParams({})
with patch("litellm.proxy.proxy_server.proxy_logging_obj") as mock_proxy_logging:
mock_proxy_logging.pre_call_hook = AsyncMock(side_effect=RuntimeError("stop before upstream"))
mock_proxy_logging.post_call_failure_hook = AsyncMock()
with pytest.raises(ProxyException):
await pass_through_request(
request=mock_request,
target="http://test.com",
custom_headers={},
user_api_key_dict=UserAPIKeyAuth(
metadata={"allow_client_pricing_override": allow_client_pricing_override}
),
custom_body={
"model": "gpt-4o",
"input_cost_per_token": 0.5,
"metadata": {
"trace_id": "keep-me",
"standard_logging_guardrail_information": [{"guardrail_name": "forged", "guardrail_cost": 5.0}],
},
},
)
hooked_body: Final = mock_proxy_logging.pre_call_hook.call_args.kwargs["data"]
assert hooked_body["metadata"] == {"trace_id": "keep-me"}
assert ("input_cost_per_token" in hooked_body) is allow_client_pricing_override
@pytest.mark.asyncio
async def test_create_pass_through_route_forwards_timeout():
unique_path = "/test/path/unique/timeout"

31
tests/unit/proxy/test_pricing_field_strip.py
View file

@ -22,6 +22,7 @@ from litellm.proxy.litellm_pre_call_utils import (
_CLIENT_PRICING_METADATA_FIELDS,
_strip_client_pricing_overrides,
add_litellm_data_to_request,
strip_client_guardrail_information,
)
from litellm.types.utils import CustomPricingLiteLLMParams
@ -133,7 +134,7 @@ class TestStripClientPricingOverrides:
"standard_logging_guardrail_information": [{"guardrail_cost": 5.0}],
},
}
_strip_client_pricing_overrides(data)
strip_client_guardrail_information(data)
assert data["metadata"] == {"user_session": "keep-me"}
assert data["litellm_metadata"] == {}
@ -164,7 +165,6 @@ class TestStripClientPricingOverrides:
def test_metadata_field_set_contains_model_info(self):
assert "model_info" in _CLIENT_PRICING_METADATA_FIELDS
assert "standard_logging_guardrail_information" in _CLIENT_PRICING_METADATA_FIELDS
def test_strip_emits_debug_log_listing_dropped_fields(self, caplog):
# Operators need a paper trail so they can diagnose why a previously
@ -315,6 +315,33 @@ async def test_add_litellm_data_to_request_skips_strip_with_key_opt_in():
assert updated["base_model"] == "gpt-4o-mini"
@pytest.mark.asyncio
@pytest.mark.parametrize("opt_in_field", ["metadata", "team_metadata"])
async def test_add_litellm_data_to_request_strips_guardrail_information_despite_pricing_opt_in(opt_in_field):
data = {
"model": "gpt-4",
"messages": [{"role": "user", "content": "hi"}],
"input_cost_per_token": 0.0001,
"metadata": {
"user_session": "keep-me",
"standard_logging_guardrail_information": [{"guardrail_name": "forged", "guardrail_cost": 5.0}],
},
}
updated = await add_litellm_data_to_request(
data=data,
request=_make_request_mock(),
user_api_key_dict=_user_api_key_auth(**{opt_in_field: {"allow_client_pricing_override": True}}),
proxy_config=MagicMock(),
general_settings={},
version="test-version",
)
assert updated["input_cost_per_token"] == 0.0001
assert updated["metadata"]["user_session"] == "keep-me"
assert "standard_logging_guardrail_information" not in updated["metadata"]
@pytest.mark.asyncio
async def test_add_litellm_data_to_request_strips_json_string_litellm_metadata():
"""``litellm_metadata`` may arrive as a JSON-encoded string (multipart/

157
tests/unit/proxy/test_proxy_utils_model_creation_and_error_logging.py
View file

@ -116,51 +116,132 @@ async def test_proxy_only_error_log_marks_no_upstream_llm_call():
@pytest.mark.asyncio
async def test_proxy_only_error_log_keeps_litellm_metadata_in_litellm_params():
"""Responses API requests carry guardrail info under ``litellm_metadata``
(not ``metadata``). It must land in litellm_params so
``merge_litellm_metadata`` can surface ``guardrail_information`` in the
spend-log failure row, matching the chat completions path."""
from litellm.proxy._types import UserAPIKeyAuth
@pytest.mark.parametrize("metadata_field", ["metadata", "litellm_metadata"])
async def test_proxy_only_error_log_drops_caller_guardrail_information_without_logging_object(
metadata_field: str,
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(litellm, "callbacks", [])
monkeypatch.setattr(litellm, "input_callback", [])
monkeypatch.setattr(litellm, "failure_callback", [])
monkeypatch.setattr(litellm, "_async_failure_callback", [])
guardrail_info: Final = [
{
"guardrail_name": "caller-guardrail",
"guardrail_status": "guardrail_intervened",
"guardrail_cost": 123.45,
}
]
request_data: Final[dict[str, object]] = {
"model": "gpt-4o",
"messages": [{"role": "user", "content": "hi"}],
metadata_field: {"tags": ["t"], "standard_logging_guardrail_information": guardrail_info},
}
proxy_logging_obj = ProxyLogging(user_api_key_cache=DualCache())
captured = {}
guardrail_info = [{"guardrail_name": "test-guard", "guardrail_status": "blocked"}]
await ProxyLogging(user_api_key_cache=DualCache())._handle_logging_proxy_only_error(
request_data=request_data,
user_api_key_dict=UserAPIKeyAuth(api_key="sk-rejected", team_id="team-a", request_route="/v1/chat/completions"),
route="/v1/chat/completions",
original_exception=HTTPException(status_code=400, detail="blocked"),
)
def fake_update_environment_variables(self, *args, **kwargs):
captured["litellm_params"] = kwargs.get("litellm_params")
captured["optional_params"] = kwargs.get("optional_params")
details: Final = request_data["litellm_logging_obj"].model_call_details
assert details["standard_logging_object"] is not None
assert details["litellm_params"][metadata_field]["tags"] == ["t"]
assert "standard_logging_guardrail_information" not in details["litellm_params"][metadata_field]
assert "guardrail_cost" not in str(details["standard_logging_object"])
assert "litellm_metadata" not in details["optional_params"]
from litellm.litellm_core_utils.litellm_logging import Logging
orig_update_env = Logging.update_environment_variables
orig_pre_call = Logging.pre_call
orig_async_failure = Logging.async_failure_handler
@pytest.mark.asyncio
@pytest.mark.parametrize("metadata_field", ["metadata", "litellm_metadata"])
async def test_proxy_only_error_log_keeps_tags_and_drops_caller_logging_fields(
metadata_field: str,
) -> None:
request_data: Final[dict[str, object]] = {
"model": "gpt-4o",
"messages": [{"role": "user", "content": "hi"}],
metadata_field: {
"langsmith_base_url": "https://caller.invalid",
"langsmith_api_key": "caller-key",
"tags": ["t"],
"user_api_key_team_id": "caller-team",
"user_api_key_hash": "caller-hash",
"user_api_key_auth_metadata": {"caller": "metadata"},
},
"langsmith_base_url": "https://caller.invalid",
"standard_callback_dynamic_params": {"gcs_path_service_account": "/caller/sa.json"},
}
async def _noop_async_failure(self, *args, **kwargs):
return None
await ProxyLogging(user_api_key_cache=DualCache())._handle_logging_proxy_only_error(
request_data=request_data,
user_api_key_dict=UserAPIKeyAuth(api_key="sk-rejected", team_id="team-a", request_route="/v1/chat/completions"),
route="/v1/chat/completions",
original_exception=Exception("bad key"),
)
Logging.update_environment_variables = fake_update_environment_variables
Logging.pre_call = lambda self, *args, **kwargs: None
Logging.async_failure_handler = _noop_async_failure
try:
await proxy_logging_obj._handle_logging_proxy_only_error(
request_data={
"model": "gpt-4o",
"input": "blocked prompt",
"litellm_metadata": {"standard_logging_guardrail_information": guardrail_info},
},
user_api_key_dict=UserAPIKeyAuth(api_key="sk-9876", request_route="/v1/responses"),
route="/v1/responses",
original_exception=HTTPException(status_code=400, detail="blocked"),
)
finally:
Logging.update_environment_variables = orig_update_env
Logging.pre_call = orig_pre_call
Logging.async_failure_handler = orig_async_failure
details: Final = request_data["litellm_logging_obj"].model_call_details
assert "caller" not in str(details["standard_callback_dynamic_params"])
assert "langsmith_base_url" not in details
assert "caller" not in str(details["litellm_params"])
assert details["litellm_params"][metadata_field]["tags"] == ["t"]
assert details["litellm_params"]["metadata"]["user_api_key_team_id"] == "team-a"
assert details["litellm_params"][metadata_field]["user_api_key_team_id"] == "team-a"
assert captured["litellm_params"]["litellm_metadata"]["standard_logging_guardrail_information"] == guardrail_info
assert "litellm_metadata" not in captured["optional_params"]
@pytest.mark.asyncio
@pytest.mark.parametrize("metadata_field", ["metadata", "litellm_metadata"])
async def test_proxy_only_error_log_drops_malformed_metadata_tags(metadata_field: str) -> None:
guardrail_info: Final = [{"guardrail_name": "g", "guardrail_status": "guardrail_intervened"}]
request_data: Final[dict[str, object]] = {
"model": "gpt-4o",
"messages": [{"role": "user", "content": "hi"}],
metadata_field: {"tags": 123, "standard_logging_guardrail_information": guardrail_info},
}
await ProxyLogging(user_api_key_cache=DualCache())._handle_logging_proxy_only_error(
request_data=request_data,
user_api_key_dict=UserAPIKeyAuth(api_key="sk-rejected", team_id="team-a", request_route="/v1/chat/completions"),
route="/v1/chat/completions",
original_exception=Exception("bad key"),
)
details: Final = request_data["litellm_logging_obj"].model_call_details
assert details["standard_logging_object"] is not None
assert "tags" not in details["litellm_params"][metadata_field]
assert "standard_logging_guardrail_information" not in details["litellm_params"][metadata_field]
@pytest.mark.asyncio
@pytest.mark.parametrize("metadata_field", ["metadata", "litellm_metadata"])
async def test_proxy_only_error_log_builds_payload_when_logging_object_has_malformed_tags(
metadata_field: str,
) -> None:
request_data: Final[dict[str, object]] = {
"model": "gpt-4o",
"messages": [{"role": "user", "content": "hi"}],
"litellm_call_id": "malformed-tags-call",
metadata_field: {"tags": 123},
}
logging_obj, _ = litellm.utils.function_setup(
original_function="acompletion",
rules_obj=litellm.utils.Rules(),
start_time=real_datetime.datetime.now(),
**request_data,
)
await ProxyLogging(user_api_key_cache=DualCache())._handle_logging_proxy_only_error(
request_data={**request_data, "litellm_logging_obj": logging_obj},
user_api_key_dict=UserAPIKeyAuth(api_key="sk-blocked", team_id="team-a", request_route="/v1/chat/completions"),
route="/v1/chat/completions",
original_exception=HTTPException(status_code=400, detail="Content blocked"),
)
standard_logging_object: Final = logging_obj.model_call_details["standard_logging_object"]
assert standard_logging_object is not None
assert standard_logging_object["model"] == request_data["model"]
assert standard_logging_object["metadata"]["user_api_key_team_id"] == "team-a"
assert 123 not in standard_logging_object["request_tags"]
@pytest.mark.asyncio

122
tests/unit/proxy/utils/proxy_logging/test_post_call_failure_hook.py
View file

@ -400,6 +400,103 @@ async def test_post_call_failure_hook_flags_pre_routing_reject_despite_caller_mo
assert kwargs["litellm_params"]["custom_llm_provider"] == "openai"
@pytest.mark.asyncio
@pytest.mark.parametrize("metadata_field", ["metadata", "litellm_metadata"])
async def test_pre_routing_reject_logs_the_configured_deployment_over_caller_metadata(
proxy_logging, make_user_api_key_auth, monkeypatch, metadata_field: str
):
from litellm.proxy import proxy_server
from litellm.proxy.spend_tracking.spend_tracking_utils import get_logging_payload
recorded: list[dict] = []
class _RecordingLogger(CustomLogger):
async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
recorded.append(kwargs)
monkeypatch.setattr(
proxy_server,
"llm_router",
litellm.Router(
model_list=[
{
"model_name": "internal-model",
"litellm_params": {"model": "openai/gpt-4.1", "api_key": "sk-test"},
}
]
),
)
monkeypatch.setattr(litellm, "callbacks", [_RecordingLogger()])
proxy_logging.alert_types = []
request_data: Final[dict[str, object]] = {
"model": "internal-model",
"messages": [{"role": "user", "content": "hi"}],
"litellm_call_id": "caller-routing-metadata",
metadata_field: {"deployment": "openai/caller-chosen", "model_group": "caller-group"},
}
logging_obj, _ = litellm.utils.function_setup(
original_function="acompletion", rules_obj=litellm.utils.Rules(), start_time=datetime.now(), **request_data
)
await proxy_logging.post_call_failure_hook(
request_data={**request_data, "litellm_logging_obj": logging_obj},
original_exception=HTTPException(status_code=400, detail="Content blocked"),
user_api_key_dict=make_user_api_key_auth(request_route="/chat/completions"),
route="/chat/completions",
)
assert len(recorded) == 1
now: Final = datetime.now()
payload: Final = get_logging_payload(
kwargs={**recorded[0], "completion_start_time": now}, response_obj=None, start_time=now, end_time=now
)
standard_logging_object: Final = recorded[0]["standard_logging_object"]
assert {
"logged_model": standard_logging_object["model"],
"logged_model_group": standard_logging_object["model_group"],
"spend_model": payload["model"],
"spend_model_group": payload["model_group"],
} == {
"logged_model": "openai/gpt-4.1",
"logged_model_group": "internal-model",
"spend_model": "openai/gpt-4.1",
"spend_model_group": "internal-model",
}
@pytest.mark.asyncio
@pytest.mark.parametrize(("error_type", "logged_tags"), [(ProxyErrorTypes.auth_error, []), (None, ["checkout"])])
async def test_proxy_only_failure_log_keeps_caller_tags_only_after_auth(
proxy_logging, monkeypatch, error_type: ProxyErrorTypes | None, logged_tags: list[str]
):
from litellm.proxy import proxy_server
recorded: list[dict] = []
class _RecordingLogger(CustomLogger):
async def async_log_failure_event(self, kwargs, response_obj, start_time, end_time):
recorded.append(kwargs)
monkeypatch.setattr(proxy_server, "llm_router", None)
monkeypatch.setattr(litellm, "callbacks", [_RecordingLogger()])
proxy_logging.alert_types = []
await proxy_logging.post_call_failure_hook(
request_data={
"model": "internal-model",
"messages": [{"role": "user", "content": "hi"}],
"metadata": {"tags": ["checkout"]},
},
original_exception=HTTPException(status_code=401, detail="Invalid proxy server token passed"),
user_api_key_dict=UserAPIKeyAuth(request_route="/chat/completions"),
error_type=error_type,
route="/chat/completions",
)
assert len(recorded) == 1
assert recorded[0]["standard_logging_object"]["request_tags"] == logged_tags
@pytest.mark.asyncio
async def test_post_call_failure_hook_attribution_does_not_count_against_the_deployment(
proxy_logging, make_user_api_key_auth, monkeypatch
@ -572,9 +669,8 @@ async def test_post_call_failure_hook_omits_provider_when_a_deployment_does_not_
async def test_handle_logging_proxy_only_path_attributes_with_read_only_metadata(
proxy_logging, make_user_api_key_auth, monkeypatch
):
"""With a logging object already on the request, its metadata is taken as given;
a read-only mapping there must not crash the stamp, and the failure handler
still receives the provider attribution."""
"""With a logging object already on the request, allowlisted metadata is kept,
a read-only mapping must not crash the stamp, and the failure handler receives provider attribution."""
from litellm.proxy import proxy_server
monkeypatch.setattr(
@ -593,16 +689,26 @@ async def test_handle_logging_proxy_only_path_attributes_with_read_only_metadata
logging_obj = MagicMock()
logging_obj.call_type = "acompletion"
logging_obj.model_call_details = {}
logging_obj.litellm_params = {"metadata": {"preexisting": "logged"}}
logging_obj.async_failure_handler = AsyncMock()
guardrail_info: Final = [
{
"guardrail_name": "local-block",
"guardrail_status": "guardrail_intervened",
"guardrail_cost": 0.25,
}
]
await proxy_logging._handle_logging_proxy_only_error(
request_data={
"litellm_logging_obj": logging_obj,
"model": "internal-model",
"messages": [{"role": "user", "content": "hi"}],
"metadata": MappingProxyType({"user_api_key_alias": "frozen"}),
"metadata": MappingProxyType(
{"tags": ["checkout"], "standard_logging_guardrail_information": guardrail_info}
),
},
user_api_key_dict=make_user_api_key_auth(request_route="/chat/completions"),
user_api_key_dict=make_user_api_key_auth(request_route="/chat/completions", key_alias="frozen"),
route="/chat/completions",
original_exception=HTTPException(status_code=403, detail="blocked"),
)
@ -611,7 +717,11 @@ async def test_handle_logging_proxy_only_path_attributes_with_read_only_metadata
update_kwargs = logging_obj.update_environment_variables.call_args.kwargs
assert update_kwargs["custom_llm_provider"] == "openai"
assert update_kwargs["litellm_params"]["custom_llm_provider"] == "openai"
assert update_kwargs["litellm_params"]["metadata"] == {"user_api_key_alias": "frozen"}
assert {
key: update_kwargs["litellm_params"]["metadata"][key] for key in ("preexisting", "user_api_key_alias")
} == {"preexisting": "logged", "user_api_key_alias": "frozen"}
assert update_kwargs["litellm_params"]["metadata"]["tags"] == ["checkout"]
assert update_kwargs["litellm_params"]["metadata"]["standard_logging_guardrail_information"] == guardrail_info
@pytest.mark.asyncio