fix(proxy): re-assert the authenticated identity on passthrough requests (#36121)

* fix(proxy): re-assert the authenticated identity on passthrough requests

The passthrough merges the client's litellm_metadata into the request metadata
and then re-asserts only user_api_key and the parent span. Every other identity
field the spend and budget pipeline reads stays whatever the request body set,
so a body carrying user_api_key_user_id, user_api_key_team_id,
user_api_key_org_id or user_api_key_end_user_id charges that user, team, org or
end user instead of the caller.

Re-assert the whole sanitized identity after the merge, so the client's copy of
any of those fields is overwritten by the authenticated key's own values.

* test(passthrough): assert no authenticated identity field is client settable

The existing regression names seven fields; the re-assertion covers every field
get_sanitized_user_information_from_key returns, which is twenty today. Derive
the set from the helper so a field added to StandardLoggingUserAPIKeyMetadata is
covered without touching the test.

Two of the twenty were not covered before, including user_api_key_hash, which is
distinct from user_api_key and was client settable.

litellm/proxy/pass_through_endpoints/pass_through_endpoints.py

@@ -565,6 +565,9 @@ class HttpPassThroughEndpointHelpers(BasePassthroughUtils):
         # real parent span.
         _metadata["user_api_key"] = user_api_key_dict.api_key
         _metadata["litellm_parent_otel_span"] = user_api_key_dict.parent_otel_span
+        _metadata.update(
+            LiteLLMProxyRequestSetup.get_sanitized_user_information_from_key(user_api_key_dict=user_api_key_dict)
+        )
 
         kwargs: Final = {
             "litellm_params": {

tests/pass_through_unit_tests/test_pass_through_unit_tests.py

@@ -30,6 +30,7 @@ from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (
 )
 from fastapi import Request
 from litellm.proxy._types import UserAPIKeyAuth
+from litellm.proxy.litellm_pre_call_utils import LiteLLMProxyRequestSetup
 from litellm.proxy.pass_through_endpoints.pass_through_endpoints import (
     _update_metadata_with_tags_in_header,
     HttpPassThroughEndpointHelpers,
@@ -652,3 +653,119 @@ def test_custom_pricing_used_in_cost_calculation():
 
     print(f"Cache-aware cost: {cache_cost}")
     print("✅ Custom pricing parameters are correctly used in cost calculation")
+
+
+def test_init_kwargs_client_metadata_cannot_spoof_authenticated_identity(
+    mock_request, mock_user_api_key_dict
+):
+    request = mock_request()
+    passthrough_payload = PassthroughStandardLoggingPayload(
+        url="https://test.com",
+        request_body={},
+    )
+    authenticated_key = UserAPIKeyAuth(
+        api_key="test-key",
+        user_id="test-user",
+        team_id="test-team",
+        end_user_id="test-user",
+        key_alias="real-key",
+        team_alias="Real Team",
+        user_email="real@example.com",
+        org_id="real-org",
+    )
+
+    result = HttpPassThroughEndpointHelpers._init_kwargs_for_pass_through_endpoint(
+        request=request,
+        user_api_key_dict=authenticated_key,
+        passthrough_logging_payload=passthrough_payload,
+        litellm_call_id="test-call-id",
+        logging_obj=LiteLLMLoggingObj(
+            model="test-model",
+            messages=[],
+            stream=False,
+            call_type="test-call-type",
+            start_time=datetime.now(),
+            litellm_call_id="test-call-id",
+            function_id="test-function-id",
+        ),
+        _parsed_body={
+            "litellm_metadata": {
+                "user_api_key_org_id": "victim-org",
+                "user_api_key_end_user_id": "victim-end-user",
+                "user_api_key_user_id": "victim-user",
+                "user_api_key_team_id": "victim-team",
+                "user_api_key_team_alias": "Victim Team",
+                "user_api_key_alias": "victim-key",
+                "user_api_key_user_email": "victim@example.com",
+            }
+        },
+    )
+
+    metadata = result["litellm_params"]["metadata"]
+    assert metadata["user_api_key_user_id"] == "test-user"
+    assert metadata["user_api_key_team_id"] == "test-team"
+    assert metadata["user_api_key_team_alias"] == "Real Team"
+    assert metadata["user_api_key_alias"] == "real-key"
+    assert metadata["user_api_key_user_email"] == "real@example.com"
+    assert metadata["user_api_key_org_id"] == "real-org"
+    assert metadata["user_api_key_end_user_id"] == "test-user"
+
+
+def test_init_kwargs_no_authenticated_identity_field_is_client_settable(
+    mock_request, mock_user_api_key_dict
+):
+    authenticated_key = UserAPIKeyAuth(
+        api_key="test-key",
+        user_id="test-user",
+        team_id="test-team",
+        end_user_id="test-end-user",
+        key_alias="real-key",
+        team_alias="Real Team",
+        user_email="real@example.com",
+        org_id="real-org",
+        organization_alias="Real Org",
+        project_id="real-project",
+        project_alias="Real Project",
+        spend=1.5,
+        max_budget=10.0,
+        user_spend=2.5,
+        user_max_budget=20.0,
+        team_spend=3.5,
+        team_max_budget=30.0,
+        metadata={"real": "auth-metadata"},
+    )
+    expected = dict(
+        LiteLLMProxyRequestSetup.get_sanitized_user_information_from_key(
+            user_api_key_dict=authenticated_key
+        )
+    )
+    assert len(expected) >= 20
+
+    spoofed = {key: f"SPOOFED-{key}" for key in expected}
+
+    result = HttpPassThroughEndpointHelpers._init_kwargs_for_pass_through_endpoint(
+        request=mock_request(),
+        user_api_key_dict=authenticated_key,
+        passthrough_logging_payload=PassthroughStandardLoggingPayload(
+            url="https://test.com", request_body={}
+        ),
+        litellm_call_id="test-call-id",
+        logging_obj=LiteLLMLoggingObj(
+            model="test-model",
+            messages=[],
+            stream=False,
+            call_type="test-call-type",
+            start_time=datetime.now(),
+            litellm_call_id="test-call-id",
+            function_id="test-function-id",
+        ),
+        _parsed_body={"litellm_metadata": dict(spoofed), "metadata": dict(spoofed)},
+    )
+
+    metadata = result["litellm_params"]["metadata"]
+    survived = {
+        key: metadata.get(key)
+        for key in expected
+        if metadata.get(key) != expected[key]
+    }
+    assert survived == {}, f"client-supplied values survived for: {sorted(survived)}"