fix(proxy): never forward the LiteLLM virtual key to Anthropic on the /anthropic passthrough

The /anthropic/{endpoint} route forwarded every incoming header upstream, so
the header carrying the caller's LiteLLM virtual key (Authorization, x-api-key,
x-litellm-api-key, or the operator-configured key header) reached Anthropic and
was rejected there as an invalid credential, with or without a proxy-side
Anthropic key layered on top.

Share the Vertex credential-less header filter: drop the proxy-only credential
headers by name, drop the value that authenticated the caller (virtual key,
master key, or JWT) from Authorization / x-api-key, keep a caller's own
Anthropic credential, layer the proxy's Anthropic credential on top, and fail
with a clean 401 when neither the proxy nor the caller supplied one.

Resolves LIT-3550

litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py

@@ -708,8 +708,7 @@ async def anthropic_proxy_route(
     endpoint_func: Final = create_pass_through_route(
         endpoint=endpoint,
         target=str(updated_url),
-        custom_headers=auth_header if auth_header is not None else {},
-        _forward_headers=True,
+        custom_headers=_upstream_headers_for_anthropic_route(request, user_api_key_dict, auth_header),
         is_streaming_request=is_streaming_request,
     )  # dynamically construct pass-through endpoint based on incoming path
     received_value: Final = await endpoint_func(
@@ -1909,6 +1908,19 @@ _HEADERS_NEVER_FORWARDED_TO_VERTEX: Final = frozenset({"content-length", "host"}
     SpecialHeaders.litellm_credential_header_names() - _VERTEX_UPSTREAM_CREDENTIAL_HEADERS
 )
 
+_CREDENTIALLESS_ANTHROPIC_MISSING_CREDENTIAL_DETAIL: Final = (
+    "No Anthropic credential is configured on this proxy and the request carried no upstream "
+    "Anthropic credential. The LiteLLM virtual key is not forwarded to Anthropic. Configure an "
+    "Anthropic credential (ANTHROPIC_API_KEY or ANTHROPIC_AUTH_TOKEN, or a model with "
+    "use_in_pass_through: true), or send your own Anthropic API key in the x-api-key header or "
+    "your own Anthropic OAuth token in the Authorization header."
+)
+
+_ANTHROPIC_UPSTREAM_CREDENTIAL_HEADERS: Final = frozenset({"authorization", "x-api-key"})
+_HEADERS_NEVER_FORWARDED_TO_ANTHROPIC: Final = frozenset({"content-length", "host", "accept-encoding"}) | (
+    SpecialHeaders.litellm_credential_header_names() - _ANTHROPIC_UPSTREAM_CREDENTIAL_HEADERS
+)
+
 
 _MAPPED_ROUTE_CALLER_KEY_HEADER: Final = "litellm_user_api_key"
 
@@ -1964,26 +1976,47 @@ def _is_authenticated_caller_secret(value: str, user_api_key_dict: UserAPIKeyAut
     return hmac.compare_digest(stored_representation.encode(), authenticated_key.encode())
 
 
+def _caller_headers_without_litellm_secrets(
+    request: Request, user_api_key_dict: UserAPIKeyAuth, never_forwarded: frozenset[str]
+) -> Mapping[str, str]:
+    """Incoming headers minus the ones only LiteLLM consumes and minus whatever value authenticated the caller."""
+    incoming: Final = _safe_get_request_headers(request)
+    dropped_by_name: Final = never_forwarded.union(
+        (_MAPPED_ROUTE_CALLER_KEY_HEADER, *_operator_configured_caller_key_header_names())
+    )
+    return MappingProxyType(
+        {
+            name: value
+            for name, value in incoming.items()
+            if name not in dropped_by_name and not _is_authenticated_caller_secret(value, user_api_key_dict)
+        }
+    )
+
+
 def _forwarded_headers_for_credentialless_vertex_passthrough(
     request: Request, user_api_key_dict: UserAPIKeyAuth
 ) -> Mapping[str, str]:
     """Caller headers to forward on the bring-your-own-credentials Vertex branch, minus LiteLLM secrets."""
-    incoming: Final = _safe_get_request_headers(request)
-    never_forwarded: Final = _HEADERS_NEVER_FORWARDED_TO_VERTEX.union(
-        (_MAPPED_ROUTE_CALLER_KEY_HEADER, *_operator_configured_caller_key_header_names())
+    forwarded: Final = _caller_headers_without_litellm_secrets(
+        request, user_api_key_dict, _HEADERS_NEVER_FORWARDED_TO_VERTEX
     )
-    forwarded: Final = MappingProxyType(
-        {
-            name: value
-            for name, value in incoming.items()
-            if name not in never_forwarded and not _is_authenticated_caller_secret(value, user_api_key_dict)
-        }
-    )
-    if "authorization" not in forwarded and "x-goog-api-key" not in forwarded:
+    if _VERTEX_UPSTREAM_CREDENTIAL_HEADERS.isdisjoint(forwarded):
         raise HTTPException(status_code=401, detail=_CREDENTIALLESS_VERTEX_MISSING_CREDENTIAL_DETAIL)
     return forwarded
 
 
+def _upstream_headers_for_anthropic_route(
+    request: Request, user_api_key_dict: UserAPIKeyAuth, proxy_auth_header: Mapping[str, str] | None
+) -> Mapping[str, str]:
+    """Caller headers minus LiteLLM secrets, with the proxy's own Anthropic credential layered on top."""
+    caller_headers: Final = _caller_headers_without_litellm_secrets(
+        request, user_api_key_dict, _HEADERS_NEVER_FORWARDED_TO_ANTHROPIC
+    )
+    if proxy_auth_header is None and _ANTHROPIC_UPSTREAM_CREDENTIAL_HEADERS.isdisjoint(caller_headers):
+        raise HTTPException(status_code=401, detail=_CREDENTIALLESS_ANTHROPIC_MISSING_CREDENTIAL_DETAIL)
+    return MappingProxyType({**caller_headers, **(proxy_auth_header or {})})
+
+
 async def _prepare_vertex_auth_headers(
     request: Request,
     vertex_credentials: VertexPassThroughCredentials | None,

tests/test_litellm/proxy/pass_through_endpoints/test_llm_pass_through_endpoints.py

@@ -28,6 +28,7 @@ from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import (
     BaseOpenAIPassThroughHandler,
     RouteChecks,
     _join_url_paths,
+    anthropic_proxy_route,
     azure_proxy_route,
     bedrock_llm_proxy_route,
     bedrock_proxy_route,
@@ -4285,6 +4286,298 @@ class TestVertexCredentiallessPassthroughVirtualKeyLeak:
         assert "sk-master-1234" not in " ".join(f"{name}:{value}" for name, value in forwarded.items())
 
 
+class TestAnthropicPassthroughVirtualKeyLeak:
+    """Regression coverage for LIT-3550.
+
+    ``/anthropic/{endpoint}`` forwarded every incoming header to Anthropic, so the
+    header that carried the caller's LiteLLM virtual key (``Authorization``,
+    ``x-api-key``, ``x-litellm-api-key``, or an operator-configured name) reached
+    Anthropic and was rejected there as an invalid credential, with or without a
+    proxy-side Anthropic key layered on top. The virtual key must never leave the
+    proxy: it is dropped by value from the headers Anthropic reads as credentials
+    (``Authorization`` / ``x-api-key``), the proxy-only credential headers are
+    dropped by name, a caller's own Anthropic credential still passes through, and
+    a request with neither a proxy credential nor a caller credential fails with a
+    clean 401 instead of reaching ``create_pass_through_route``.
+
+    The forwarded set is rebuilt the way ``pass_through_request`` builds it from
+    the captured ``create_pass_through_route`` kwargs, so a route that re-enables
+    ``_forward_headers`` fails these tests the same way the original bug did.
+    """
+
+    VKEY = "sk-litellm-victim-key"
+    PROXY_KEY = "sk-ant-api03-proxy-configured-key"
+    ENDPOINT = "v1/messages"
+
+    async def _run(
+        self,
+        monkeypatch,
+        headers: list[tuple[bytes, bytes]],
+        authenticated: UserAPIKeyAuth | None = None,
+        master_key: str | None = "sk-master-1234",
+        proxy_api_key: str | None = None,
+    ) -> tuple[HTTPException | None, dict | None]:
+        from litellm.proxy.pass_through_endpoints.pass_through_endpoints import HttpPassThroughEndpointHelpers
+        from litellm.proxy.pass_through_endpoints.passthrough_endpoint_router import (
+            PassthroughEndpointRouter,
+        )
+
+        monkeypatch.setattr("litellm.proxy.proxy_server.master_key", master_key)
+        monkeypatch.delenv("ANTHROPIC_AUTH_TOKEN", raising=False)
+        if proxy_api_key is None:
+            monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
+        else:
+            monkeypatch.setenv("ANTHROPIC_API_KEY", proxy_api_key)
+        caller: Final = authenticated if authenticated is not None else UserAPIKeyAuth(api_key=self.VKEY)
+
+        async def receive():
+            return {"type": "http.request", "body": b"{}", "more_body": False}
+
+        request = Request(
+            {
+                "type": "http",
+                "method": "POST",
+                "path": f"/anthropic/{self.ENDPOINT}",
+                "headers": headers,
+                "query_string": b"",
+            },
+            receive=receive,
+        )
+
+        captured: dict = {}
+
+        def fake_create_pass_through_route(**kwargs):
+            captured.update(kwargs)
+            return AsyncMock(return_value={"status": "success"})
+
+        module = "litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints"
+        monkeypatch.setattr(f"{module}.passthrough_endpoint_router", PassthroughEndpointRouter(lambda: None))
+        raised: HTTPException | None = None
+        with (
+            mock.patch(f"{module}.create_pass_through_route", side_effect=fake_create_pass_through_route),
+            mock.patch(f"{module}.user_api_key_auth", new=AsyncMock(return_value=caller)),
+        ):
+            try:
+                await anthropic_proxy_route(
+                    endpoint=self.ENDPOINT,
+                    request=request,
+                    fastapi_response=Response(),
+                    user_api_key_dict=caller,
+                )
+            except HTTPException as exc:
+                raised = exc
+
+        if not captured:
+            return raised, None
+        upstream: Final = HttpPassThroughEndpointHelpers.forward_headers_from_request(
+            request_headers=dict(request.headers),
+            headers=dict(captured["custom_headers"] or {}),
+            forward_headers=captured.get("_forward_headers", False),
+        )
+        return raised, upstream
+
+    @staticmethod
+    def _blob(forwarded: dict) -> str:
+        return " ".join(f"{name}:{value}" for name, value in forwarded.items())
+
+    @pytest.mark.asyncio
+    async def test_authorization_bearer_virtual_key_is_rejected_not_forwarded(self, monkeypatch):
+        raised, forwarded = await self._run(
+            monkeypatch,
+            [(b"authorization", f"Bearer {self.VKEY}".encode()), (b"content-type", b"application/json")],
+        )
+        assert forwarded is None, "credential-less request must never reach the upstream forwarder"
+        assert raised is not None and raised.status_code == 401
+        assert "ANTHROPIC_API_KEY" in str(raised.detail) and "use_in_pass_through" in str(raised.detail)
+
+    @pytest.mark.asyncio
+    async def test_x_api_key_virtual_key_is_rejected_not_forwarded(self, monkeypatch):
+        raised, forwarded = await self._run(
+            monkeypatch,
+            [(b"x-api-key", self.VKEY.encode()), (b"content-type", b"application/json")],
+        )
+        assert forwarded is None, "a virtual key that authenticated via x-api-key must be stripped, not forwarded"
+        assert raised is not None and raised.status_code == 401
+
+    @pytest.mark.asyncio
+    async def test_x_litellm_api_key_virtual_key_is_rejected_not_forwarded(self, monkeypatch):
+        raised, forwarded = await self._run(
+            monkeypatch,
+            [(b"x-litellm-api-key", self.VKEY.encode()), (b"content-type", b"application/json")],
+        )
+        assert forwarded is None, "credential-less request must never reach the upstream forwarder"
+        assert raised is not None and raised.status_code == 401
+
+    @pytest.mark.asyncio
+    async def test_master_key_in_authorization_is_rejected_not_forwarded(self, monkeypatch):
+        raised, forwarded = await self._run(
+            monkeypatch,
+            [(b"authorization", b"Bearer sk-master-1234"), (b"content-type", b"application/json")],
+            authenticated=UserAPIKeyAuth(api_key="sk-master-1234", user_role=LitellmUserRoles.PROXY_ADMIN),
+        )
+        assert forwarded is None, "the master key must never reach Anthropic"
+        assert raised is not None and raised.status_code == 401
+
+    @pytest.mark.asyncio
+    async def test_byo_anthropic_oauth_token_still_forwards_without_virtual_key(self, monkeypatch):
+        raised, forwarded = await self._run(
+            monkeypatch,
+            [
+                (b"x-litellm-api-key", self.VKEY.encode()),
+                (b"authorization", b"Bearer sk-ant-oat01-caller-oauth-token"),
+                (b"anthropic-version", b"2023-06-01"),
+                (b"content-type", b"application/json"),
+            ],
+        )
+        assert raised is None
+        assert forwarded is not None
+        assert forwarded.get("authorization") == "Bearer sk-ant-oat01-caller-oauth-token"
+        assert forwarded.get("anthropic-version") == "2023-06-01"
+        assert "x-litellm-api-key" not in forwarded
+        assert self.VKEY not in self._blob(forwarded)
+
+    @pytest.mark.asyncio
+    async def test_byo_x_api_key_still_forwards_without_virtual_key(self, monkeypatch):
+        raised, forwarded = await self._run(
+            monkeypatch,
+            [
+                (b"authorization", f"Bearer {self.VKEY}".encode()),
+                (b"x-api-key", b"sk-ant-api03-caller-own-key"),
+                (b"content-type", b"application/json"),
+            ],
+        )
+        assert raised is None
+        assert forwarded is not None
+        assert forwarded.get("x-api-key") == "sk-ant-api03-caller-own-key"
+        assert "authorization" not in forwarded
+        assert self.VKEY not in self._blob(forwarded)
+
+    @pytest.mark.asyncio
+    async def test_custom_auth_caller_keeps_own_authorization_token(self, monkeypatch):
+        raised, forwarded = await self._run(
+            monkeypatch,
+            [(b"authorization", b"Bearer sk-ant-oat01-caller-oauth-token"), (b"content-type", b"application/json")],
+            authenticated=UserAPIKeyAuth(api_key=None),
+            master_key=None,
+        )
+        assert raised is None
+        assert forwarded is not None
+        assert forwarded.get("authorization") == "Bearer sk-ant-oat01-caller-oauth-token"
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        "credential_header",
+        sorted(SpecialHeaders.litellm_credential_header_names() - {"authorization", "x-api-key", "x-litellm-api-key"}),
+    )
+    async def test_every_non_anthropic_credential_header_is_dropped_by_name(self, monkeypatch, credential_header):
+        raised, forwarded = await self._run(
+            monkeypatch,
+            [
+                (b"x-litellm-api-key", self.VKEY.encode()),
+                (b"x-api-key", b"sk-ant-api03-caller-own-key"),
+                (credential_header.encode(), b"some-distinct-caller-secret-value"),
+                (b"content-type", b"application/json"),
+            ],
+        )
+        assert raised is None
+        assert forwarded is not None
+        assert forwarded.get("x-api-key") == "sk-ant-api03-caller-own-key"
+        assert credential_header not in forwarded
+        assert "x-litellm-api-key" not in forwarded
+        assert self.VKEY not in self._blob(forwarded)
+        assert "some-distinct-caller-secret-value" not in self._blob(forwarded)
+
+    @pytest.mark.asyncio
+    async def test_virtual_key_in_operator_configured_header_is_stripped(self, monkeypatch):
+        with mock.patch.dict(  # test-quality-ok: general_settings is the real proxy config surface for litellm_key_header_name; no injection seam exists on this route
+            "litellm.proxy.proxy_server.general_settings",
+            {"litellm_key_header_name": "x-company-key"},
+        ):
+            raised, forwarded = await self._run(
+                monkeypatch,
+                [
+                    (b"x-company-key", f"Bearer {self.VKEY}".encode()),
+                    (b"x-api-key", b"sk-ant-api03-caller-own-key"),
+                    (b"content-type", b"application/json"),
+                ],
+            )
+        assert raised is None
+        assert forwarded is not None
+        assert forwarded.get("x-api-key") == "sk-ant-api03-caller-own-key"
+        assert "x-company-key" not in forwarded
+        assert self.VKEY not in self._blob(forwarded)
+
+    @pytest.mark.asyncio
+    async def test_proxy_credential_replaces_virtual_key_sent_as_bearer(self, monkeypatch):
+        raised, forwarded = await self._run(
+            monkeypatch,
+            [
+                (b"authorization", f"Bearer {self.VKEY}".encode()),
+                (b"anthropic-version", b"2023-06-01"),
+                (b"content-type", b"application/json"),
+            ],
+            proxy_api_key=self.PROXY_KEY,
+        )
+        assert raised is None
+        assert forwarded is not None
+        assert forwarded.get("x-api-key") == self.PROXY_KEY
+        assert "authorization" not in forwarded
+        assert forwarded.get("anthropic-version") == "2023-06-01"
+        assert self.VKEY not in self._blob(forwarded)
+
+    @pytest.mark.asyncio
+    async def test_proxy_credential_replaces_virtual_key_sent_as_x_api_key(self, monkeypatch):
+        raised, forwarded = await self._run(
+            monkeypatch,
+            [(b"x-api-key", self.VKEY.encode()), (b"content-type", b"application/json")],
+            proxy_api_key=self.PROXY_KEY,
+        )
+        assert raised is None
+        assert forwarded is not None
+        assert forwarded.get("x-api-key") == self.PROXY_KEY
+        assert self.VKEY not in self._blob(forwarded)
+
+    @pytest.mark.asyncio
+    async def test_proxy_credential_wins_over_callers_own_x_api_key(self, monkeypatch):
+        raised, forwarded = await self._run(
+            monkeypatch,
+            [
+                (b"x-litellm-api-key", self.VKEY.encode()),
+                (b"x-api-key", b"sk-ant-api03-caller-own-key"),
+                (b"content-type", b"application/json"),
+            ],
+            proxy_api_key=self.PROXY_KEY,
+        )
+        assert raised is None
+        assert forwarded is not None
+        assert forwarded.get("x-api-key") == self.PROXY_KEY
+        assert "sk-ant-api03-caller-own-key" not in self._blob(forwarded)
+
+    @pytest.mark.asyncio
+    async def test_x_pass_and_hop_by_hop_handling_is_unchanged(self, monkeypatch):
+        raised, forwarded = await self._run(
+            monkeypatch,
+            [
+                (b"authorization", f"Bearer {self.VKEY}".encode()),
+                (b"x-pass-anthropic-beta", b"interleaved-thinking-2025-05-14"),
+                (b"x-pass-authorization", b"Bearer smuggled"),
+                (b"content-length", b"2"),
+                (b"host", b"proxy.internal"),
+                (b"accept-encoding", b"br"),
+                (b"user-agent", b"curl/8.7.1"),
+            ],
+            proxy_api_key=self.PROXY_KEY,
+        )
+        assert raised is None
+        assert forwarded is not None
+        assert forwarded.get("anthropic-beta") == "interleaved-thinking-2025-05-14"
+        assert forwarded.get("user-agent") == "curl/8.7.1"
+        assert "authorization" not in forwarded
+        assert "content-length" not in forwarded
+        assert "host" not in forwarded
+        assert "accept-encoding" not in forwarded
+
+
 class TestVertexPassthroughDefaultLocationOnShortRoutes:
     PROJECT = "test-project"
     SHORT_ROUTE = "publishers/google/models/gemini-2.5-flash:generateContent"