Merge 2c5b009c2f into adf3373381

litellm/integrations/prometheus.py

@@ -14,6 +14,7 @@ from functools import partial
 from types import MappingProxyType
 from typing import TYPE_CHECKING, Annotated, Any, Final, Literal, Protocol, TypeAlias, TypeVar, cast
 
+import openai
 from pydantic import BaseModel, Field, TypeAdapter, ValidationError
 from typing_extensions import ReadOnly, TypedDict, assert_never
 
@@ -22,11 +23,13 @@ from litellm._internal_context import with_service_target
 from litellm._logging import print_verbose, verbose_logger
 from litellm.constants import (
     AUTH_OBJECTS_TARGET,
+    LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL,
     PROMETHEUS_OVERFLOW_SERIES_LABEL_VALUE,
     PROXY_LLM_PROVIDER_FALLBACK,
     PROXY_REJECTED_BEFORE_ROUTING_KEY,
 )
 from litellm.exceptions import (
+    CallerCredentialAuthenticationError,
     MalformedRequestFieldError,
     validate_rate_limit_category,
     validate_rate_limit_type,
@@ -51,6 +54,7 @@ from litellm.litellm_core_utils.get_llm_provider_logic import is_registered_cust
 from litellm.litellm_core_utils.service_tier_utils import (
     get_service_tier_from_standard_logging_payload,
 )
+from litellm.llms.github_copilot.per_user_auth import github_copilot_user_session_from
 from litellm.llms.openai_like.json_loader import JSONProviderRegistry
 from litellm.models.end_user import LiteLLM_EndUserTable
 from litellm.models.team import LiteLLM_TeamTable
@@ -269,7 +273,10 @@ _NO_LOGGED_FIELDS: Final[Mapping[str, object]] = MappingProxyType({})
 
 
 def _logged_fields(value: object) -> Mapping[str, object]:
-    return _LOGGED_FIELDS.validate_python(value) if isinstance(value, Mapping) else _NO_LOGGED_FIELDS
+    try:
+        return _LOGGED_FIELDS.validate_python(value)
+    except ValidationError:
+        return _NO_LOGGED_FIELDS
 
 
 def _is_llm_provider(name: str) -> bool:
@@ -359,6 +366,86 @@ def _get_proxy_llm_router() -> Router | None:
     return llm_router
 
 
+_CALLER_CREDENTIAL_FIELDS: Final = frozenset(
+    {
+        "aws_access_key_id",
+        "aws_secret_access_key",
+        "aws_session_token",
+        "aws_web_identity_token",
+        "azure_ad_token",
+        "azure_password",
+        "azure_scope",
+        "azure_username",
+        "client_id",
+        "client_secret",
+        "gigachat_access_token",
+        "oci_fingerprint",
+        "oci_key",
+        "oci_key_file",
+        "oci_tenancy",
+        "oci_user",
+        "tenant_id",
+        "token",
+        "user_config",
+        "vertex_ai_credentials",
+        "vertex_credentials",
+        "watsonx_credentials",
+        "wx_credentials",
+        "zen_api_key",
+    }
+)
+
+
+_CALLER_ENDPOINT_FIELDS: Final = frozenset(
+    {"aws_bedrock_runtime_endpoint", "aws_sts_endpoint", "deployment_url", "s3_endpoint_url", "sagemaker_base_url"}
+)
+
+
+_CALLER_TRANSPORT_FIELDS: Final = frozenset(
+    {"api_key", "extra_headers", "headers", "organization", "provider_specific_header"}
+)
+
+
+def _sent_a_provider_credential_or_endpoint(caller_body: Mapping[str, object]) -> bool:
+    """True when the caller's request body, or its ``extra_body``, carries a provider credential field or a
+    provider endpoint field."""
+    sent: Final = {*caller_body, *_logged_fields(caller_body.get("extra_body"))}
+    return not (_CALLER_CREDENTIAL_FIELDS.isdisjoint(sent) and _CALLER_ENDPOINT_FIELDS.isdisjoint(sent))
+
+
+def _used_a_caller_credential(logged_call: Mapping[str, object]) -> bool:
+    """True when the provider was called with a credential or at an endpoint that belongs to the caller rather than
+    the deployment: an api_key, header, OpenAI organization or other provider credential or endpoint field the caller
+    sent with the request, the caller's own Anthropic OAuth token, the caller's own GitHub Copilot connection, or the
+    caller's identity token that a token-exchange deployment trades in on every call."""
+    litellm_params: Final = _logged_fields(logged_call.get("litellm_params"))
+    proxy_server_request: Final = _logged_fields(litellm_params.get("proxy_server_request"))
+    caller_credential_fields: Final = proxy_server_request.get("caller_credential_fields")
+    logged_metadata: Final = _logged_fields(_logged_fields(logged_call.get("standard_logging_object")).get("metadata"))
+    return (
+        (
+            isinstance(caller_credential_fields, tuple)
+            and any(field in caller_credential_fields for field in _CALLER_TRANSPORT_FIELDS)
+        )
+        or _sent_a_provider_credential_or_endpoint(_logged_fields(proxy_server_request.get("body")))
+        or bool(litellm_params.get("token_exchange_endpoint"))
+        or logged_metadata.get("used_client_oauth_token") is True
+        or github_copilot_user_session_from(litellm_params) is not None
+    )
+
+
+def _is_deployment_failure(exception: openai.APIError, logged_call: Mapping[str, object]) -> bool:
+    """True for an error raised while calling the deployment with its own credential, on a request the proxy
+    did not reject itself. An error LiteLLM raises before it calls the provider, such as a missing deployment key,
+    is not one."""
+    return not (
+        isinstance(exception, CallerCredentialAuthenticationError)
+        or bool(logged_call.get(LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL))
+        or logged_call.get("api_call_start_time") is None
+        or _used_a_caller_credential(logged_call)
+    )
+
+
 def _bounded_requested_model_label(requested_model: object, router_originated: bool = False) -> str | None:
     """
     Bound ``requested_model`` label cardinality: names the router recognizes
@@ -2817,6 +2904,8 @@ class PrometheusLogger(CustomLogger):
         information from various callback function signatures and determines if the request
         represents an invalid API key attempt that should be filtered from metrics.
 
+        An error raised while calling the deployment with its own credential is never skipped.
+
         Args:
             kwargs: Dictionary potentially containing exception and other data
             user_api_key_dict: User API key authentication object (currently unused)
@@ -2836,6 +2925,11 @@ class PrometheusLogger(CustomLogger):
         if exception is None and kwargs:
             exception = kwargs.get("exception")
 
+        if isinstance(exception, openai.APIError) and _is_deployment_failure(
+            exception, _LOGGED_FIELDS.validate_python(kwargs or {})
+        ):
+            return False
+
         if self._is_invalid_api_key_request(status_code, exception=exception):
             verbose_logger.debug(
                 "Skipping Prometheus metrics for invalid API key request: status_code=%s, exception=%s",

litellm/proxy/litellm_pre_call_utils.py

@@ -99,6 +99,8 @@ _CREDENTIAL_HEADER_NAMES: Final = SpecialHeaders.litellm_credential_header_names
     {"cookie", "proxy-authorization"}
 )
 _TRANSPORT_ONLY_CREDENTIAL_KEYS: Final = frozenset({"provider_specific_header", "headers", "api_key"})
+_REQUEST_FIELDS: Final = TypeAdapter(Mapping[str, object])
+_REQUEST_ENTRIES: Final = TypeAdapter(tuple[object, ...])
 
 # Matches any header of the form x-<something>-session-id (case-insensitive).
 # Excludes the two explicit litellm headers which are handled with higher priority.
@@ -289,6 +291,9 @@ _DESTINATION_ONLY_CALLBACK_VARS: Final[frozenset[str]] = frozenset({CAPTURE_MESS
 _CALLBACK_CREDENTIAL_KEYS: Final = frozenset(StandardCallbackDynamicParams.__annotations__) | frozenset(
     {TRUSTED_CALLBACK_VARS_FIELD}
 )
+_HEADERS_THAT_CANNOT_CHANGE_PROVIDER_AUTH: Final = frozenset(
+    {*(f"x-litellm-{key}" for key in StandardLoggingUserAPIKeyMetadata.__annotations__), *ANTHROPIC_API_HEADERS}
+)
 service_logger_obj: Final = ServiceLogging()  # used for tracking latency on OTEL
 # Bounded dedup for stale-alias warnings (FIFO eviction when over cap).
 _MAX_STALE_ALIAS_WARNING_KEYS: Final = 10_000
@@ -1448,6 +1453,44 @@ def clean_headers(
     return clean_headers
 
 
+def _request_fields(value: object) -> Mapping[str, object]:
+    try:
+        return _REQUEST_FIELDS.validate_python(value)
+    except PydanticValidationError:
+        return MappingProxyType({})
+
+
+def _request_entries(value: object) -> tuple[object, ...]:
+    if value is None:
+        return ()
+    try:
+        return _REQUEST_ENTRIES.validate_python(value)
+    except PydanticValidationError:
+        return (value,)
+
+
+def _carries_a_caller_header(headers: object) -> bool:
+    return any(name.lower() not in _HEADERS_THAT_CANNOT_CHANGE_PROVIDER_AUTH for name in _request_fields(headers))
+
+
+def _caller_credential_fields(data: Mapping[str, object]) -> tuple[str, ...]:
+    """Names of the request fields through which the caller can change how the provider authenticates the call:
+    ``api_key``, ``organization``, and ``headers``, ``extra_headers`` or a ``provider_specific_header`` entry's
+    ``extra_headers`` that carry any header besides the key and user details the proxy adds under
+    ``add_user_information_to_llm_headers`` and the ``anthropic-version`` and ``anthropic-beta`` headers."""
+    carried: Final = {
+        "api_key": "api_key" in data,
+        "extra_headers": _carries_a_caller_header(data.get("extra_headers")),
+        "headers": _carries_a_caller_header(data.get("headers")),
+        "organization": "organization" in data,
+        "provider_specific_header": any(
+            _carries_a_caller_header(_request_fields(entry).get("extra_headers"))
+            for entry in _request_entries(data.get("provider_specific_header"))
+        ),
+    }
+    return tuple(name for name, sent in sorted(carried.items()) if sent)
+
+
 def _is_credential_header(header: str) -> bool:
     """Whether `header` carries a caller credential rather than request context."""
     return header.lower() in _CREDENTIAL_HEADER_NAMES
@@ -2594,6 +2637,8 @@ async def add_litellm_data_to_request(
     data[_metadata_variable_name]["used_client_oauth_token"] = add_provider_specific_headers_to_request(
         data=data, headers=_headers
     )
+    caller_fields: Final = _caller_credential_fields(_REQUEST_FIELDS.validate_python(data))
+    data["proxy_server_request"]["caller_credential_fields"] = caller_fields  # rebind-ok: data is an out-param
 
     ## Cache Controls
     cache_control_header: Final = _headers.get("Cache-Control", None)

tests/integration/_support/copilot_session_prometheus_worker.py

@@ -0,0 +1,162 @@
+"""One SDK process for the SDK provider-401 metric cells: a provider 401 on a caller's own Copilot session, one on
+the router deployment's own key, one on a key the router caller passes and one on a direct call's own key, then a
+report of the deployment series on that process's default registry. Run as a script by ``run_copilot_session_worker`` so each process registers the
+Prometheus metrics exactly once."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import subprocess
+import sys
+from itertools import chain
+from dataclasses import dataclass
+from typing import Final
+
+import litellm
+import openai
+from litellm.integrations.prometheus import PrometheusLogger
+from litellm.llms.github_copilot.per_user_auth import GithubCopilotUserSession
+from prometheus_client import REGISTRY
+from pydantic import TypeAdapter
+
+COPILOT_DEPLOYMENT: Final = "copilot-per-user-deployment"
+OPERATOR_DEPLOYMENT: Final = "operator-key-deployment"
+CALLER_SESSION_TOKEN: Final = "caller-copilot-session-token"
+OPERATOR_KEY: Final = "rotated-operator-key"
+ROUTER_CALLER_KEY: Final = "rotated-router-caller-key"
+DIRECT_CALL: Final = "direct-sdk-call"
+DIRECT_KEY: Final = "rotated-direct-call-key"
+FAILURES: Final = "litellm_deployment_failure_responses_total"
+TOTALS: Final = "litellm_deployment_total_requests_total"
+LLM_FAILED: Final = "litellm_llm_api_failed_requests_metric_total"
+STATE: Final = "litellm_deployment_state"
+DEPLOYMENT_SERIES: Final = frozenset({FAILURES, TOTALS, LLM_FAILED, STATE})
+WORKER_TIMEOUT_SECONDS: Final = 120
+
+
+@dataclass(frozen=True, slots=True)
+class Outcome:
+    deployment: str
+    status: int | None
+
+
+@dataclass(frozen=True, slots=True)
+class Series:
+    metric: str
+    model_id: str
+    api_base: str
+    value: float
+
+
+@dataclass(frozen=True, slots=True)
+class WorkerReport:
+    api_base: str
+    outcomes: tuple[Outcome, ...]
+    series: tuple[Series, ...]
+
+    def values(self, metric: str, model_id: str) -> tuple[float, ...]:
+        return tuple(s.value for s in self.series if s.metric == metric and s.model_id == model_id)
+
+    def on_the_endpoint(self, metric: str, model_id: str) -> tuple[float, ...]:
+        return tuple(
+            s.value for s in self.series if (s.metric, s.model_id, s.api_base) == (metric, model_id, self.api_base)
+        )
+
+    def outside_the_router(self, metric: str) -> tuple[float, ...]:
+        return tuple(s.value for s in self.series if s.metric == metric and not s.model_id)
+
+    def on_a_one_off_deployment(self, metric: str) -> tuple[float, ...]:
+        named_ids: Final = ("", COPILOT_DEPLOYMENT, OPERATOR_DEPLOYMENT)
+        return tuple(s.value for s in self.series if s.metric == metric and s.model_id not in named_ids)
+
+
+_REPORT: Final = TypeAdapter(WorkerReport)
+
+
+def run_copilot_session_worker(api_base: str) -> WorkerReport:
+    completed: Final = subprocess.run(
+        (sys.executable, "-P", __file__, api_base),
+        capture_output=True,
+        text=True,
+        timeout=WORKER_TIMEOUT_SECONDS,
+        check=False,
+    )
+    assert completed.returncode == 0, completed.stderr[-3000:]
+    return _REPORT.validate_json(completed.stdout.splitlines()[-1])
+
+
+def _deployment_series() -> tuple[Series, ...]:
+    samples: Final = tuple(chain.from_iterable(family.samples for family in REGISTRY.collect()))
+    return tuple(
+        Series(
+            metric=sample.name,
+            model_id=str(sample.labels.get("model_id", "")),
+            api_base=str(sample.labels.get("api_base", "")),
+            value=float(sample.value),
+        )
+        for sample in samples
+        if sample.name in DEPLOYMENT_SERIES
+    )
+
+
+def _router(api_base: str) -> litellm.Router:
+    return litellm.Router(
+        model_list=[
+            {
+                "model_name": COPILOT_DEPLOYMENT,
+                "litellm_params": {
+                    "model": "github_copilot/gpt-4o",
+                    "litellm_credential_name": "copilot-per-user",
+                    "github_copilot_auth_type": "per_user_oauth",
+                },
+                "model_info": {"id": COPILOT_DEPLOYMENT},
+            },
+            {
+                "model_name": OPERATOR_DEPLOYMENT,
+                "litellm_params": {"model": "openai/gpt-5.6", "api_base": api_base, "api_key": OPERATOR_KEY},
+                "model_info": {"id": OPERATOR_DEPLOYMENT},
+            },
+        ],
+        num_retries=0,
+        disable_cooldowns=True,
+    )
+
+
+async def _attempt(router: litellm.Router, model: str, **extra: object) -> Outcome:
+    try:
+        await router.acompletion(model=model, messages=[{"role": "user", "content": f"{model} worker"}], **extra)
+    except openai.APIStatusError as error:
+        return Outcome(deployment=model, status=error.status_code)
+    return Outcome(deployment=model, status=None)
+
+
+async def _direct_attempt(api_base: str) -> Outcome:
+    try:
+        await litellm.acompletion(
+            model="openai/gpt-5.6",
+            api_base=api_base,
+            api_key=DIRECT_KEY,
+            messages=[{"role": "user", "content": f"{DIRECT_CALL} worker"}],
+        )
+    except openai.APIStatusError as error:
+        return Outcome(deployment=DIRECT_CALL, status=error.status_code)
+    return Outcome(deployment=DIRECT_CALL, status=None)
+
+
+async def main(api_base: str) -> None:
+    litellm.callbacks = [PrometheusLogger()]  # test-quality-ok: this worker process exits right after the report
+    session: Final = GithubCopilotUserSession(token=CALLER_SESSION_TOKEN, api_base=api_base)
+    router: Final = _router(api_base)
+    outcomes: Final = (
+        await _attempt(router, COPILOT_DEPLOYMENT, github_copilot_user_session=session),
+        await _attempt(router, OPERATOR_DEPLOYMENT),
+        await _attempt(router, OPERATOR_DEPLOYMENT, api_key=ROUTER_CALLER_KEY),
+        await _direct_attempt(api_base),
+    )
+    report: Final = WorkerReport(api_base=api_base, outcomes=outcomes, series=_deployment_series())
+    sys.stdout.write(json.dumps(_REPORT.dump_python(report)) + "\n")
+
+
+if __name__ == "__main__":
+    asyncio.run(main(sys.argv[1]))

tests/integration/observability/test_prometheus_provider_auth_failure_chaos.py

@@ -0,0 +1,147 @@
+import threading
+import uuid
+from collections.abc import Iterator, Sequence
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import contextmanager
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Final
+
+import httpx
+import psutil
+from integration._support.client import Gateway, eventually
+from integration._support.process import OwnedProxy, owned_proxy_process
+from integration._support.prometheus_series import Sample, scrape
+from integration._support.wire import Wire, wire_server
+from test_prometheus_provider_auth_failure import (
+    FAILURES,
+    LLM_FAILED,
+    STATE,
+    TOTALS,
+    Provider,
+    config,
+    identity,
+    name,
+    send,
+    values,
+)
+
+BURST: Final = 12
+ENDPOINTS: Final = ("chat", "chat-stream", "messages", "responses")
+
+
+@dataclass(frozen=True, slots=True)
+class ChaosRig:
+    owned: OwnedProxy
+    upstream: Wire
+    release: threading.Event
+
+
+@contextmanager
+def _rig(gateway: Gateway, tmp_path: Path) -> Iterator[ChaosRig]:
+    release: Final = threading.Event()
+    prom_dir: Final = tmp_path / "prom"
+    prom_dir.mkdir()
+    with (
+        wire_server(Provider(release)) as upstream,
+        owned_proxy_process(
+            gateway,
+            tmp_path,
+            {"PROMETHEUS_MULTIPROC_DIR": str(prom_dir)},
+            config=config(tmp_path, upstream),
+            workers=2,
+        ) as owned,
+    ):
+        try:
+            yield ChaosRig(owned, upstream, release)
+        finally:
+            release.set()
+
+
+def _status(gateway: Gateway, cell: str, model: str, key: str | None = None) -> int:
+    try:
+        return send(gateway, cell, model=model, key=key).status_code
+    except httpx.HTTPError:
+        return -1
+
+
+def _holds_a_connection_to(pid: int, port: int) -> bool:
+    return any(connection.raddr[1:] == (port,) for connection in psutil.Process(pid).net_connections(kind="tcp"))
+
+
+def _workers_calling_upstream(rig: ChaosRig) -> frozenset[int]:
+    port: Final = int(rig.upstream.url.rsplit(":", 1)[1])
+    return frozenset(pid for pid in rig.owned.worker_pids() if _holds_a_connection_to(pid, port))
+
+
+def _counted(samples: Sequence[Sample], cell: str) -> tuple[float, float, float]:
+    return (
+        sum(values(samples, FAILURES, model_id=identity(cell))),
+        sum(values(samples, TOTALS, model_id=identity(cell))),
+        sum(values(samples, LLM_FAILED, model_id=identity(cell))),
+    )
+
+
+def test_a_burst_of_provider_401s_and_proxy_rejects_across_two_workers_counts_only_the_provider_401s(
+    gateway: Gateway, tmp_path: Path
+) -> None:
+    with _rig(gateway, tmp_path) as rig:
+        rig.release.set()
+        workers: Final = eventually(rig.owned.worker_pids, lambda pids: len(pids) == 2)
+        invalid: Final = f"sk-not-a-virtual-key-{uuid.uuid4().hex}"
+        calls: Final[tuple[tuple[str, str, str | None], ...]] = tuple(
+            (ENDPOINTS[index % len(ENDPOINTS)], name("burst"), None) for index in range(BURST)
+        ) + tuple(("chat", name("invalid-key"), invalid) for _ in range(BURST))
+
+        def call(cell: str, model: str, key: str | None) -> int:
+            return _status(rig.owned.gateway, cell, model, key)
+
+        with ThreadPoolExecutor(max_workers=8) as pool:
+            statuses: Final = tuple(pool.map(call, *zip(*calls, strict=True)))
+        assert statuses == (401,) * (2 * BURST), statuses
+        assert len(rig.upstream.drain()) == BURST
+        samples: Final = eventually(
+            lambda: scrape(rig.owned.gateway),
+            lambda observed: _counted(observed, "burst") == (BURST, BURST, BURST),
+            seconds=70,
+            return_last_on_timeout=True,
+        )
+        assert _counted(samples, "burst") == (BURST, BURST, BURST), (workers, _counted(samples, "burst"))
+        assert set(values(samples, STATE, model_id=identity("burst"))) == {1.0}
+        assert values(samples, FAILURES, requested_model=name("invalid-key")) == []
+        assert values(samples, LLM_FAILED, model=name("invalid-key")) == []
+
+
+def test_a_worker_killed_while_provider_401s_are_in_flight_loses_no_answered_failure(
+    gateway: Gateway, tmp_path: Path
+) -> None:
+    with _rig(gateway, tmp_path) as rig:
+        eventually(rig.owned.worker_pids, lambda pids: len(pids) == 2)
+        with ThreadPoolExecutor(max_workers=8) as pool:
+            futures: Final = tuple(pool.submit(_status, rig.owned.gateway, "chat", name("held")) for _ in range(BURST))
+            eventually(lambda: rig.upstream.received.qsize() >= 8, bool, seconds=30)
+            victim: Final = min(eventually(lambda: _workers_calling_upstream(rig), bool))
+            with rig.owned.replacing_worker(victim) as kill:
+                kill()
+                rig.release.set()
+            statuses: Final = tuple(future.result() for future in futures)
+        reached: Final = len(rig.upstream.drain())
+        answered: Final = statuses.count(401)
+        lost: Final = statuses.count(-1)
+        assert answered + lost == BURST and lost > 0, statuses
+        samples: Final = eventually(
+            lambda: scrape(rig.owned.gateway),
+            lambda observed: answered <= _counted(observed, "held")[0] <= reached,
+            seconds=70,
+            return_last_on_timeout=True,
+        )
+        counted: Final = _counted(samples, "held")
+        assert answered <= counted[0] <= reached and counted[0] == counted[1] == counted[2], (statuses, counted)
+        follow_up: Final = _status(rig.owned.gateway, "chat", name("held"))
+        assert follow_up == 401
+        recounted: Final = eventually(
+            lambda: _counted(scrape(rig.owned.gateway), "held"),
+            lambda observed: observed[0] == counted[0] + 1,
+            seconds=70,
+        )
+        assert recounted == (counted[0] + 1,) * 3

tests/integration/sdk/test_prometheus_copilot_session_auth_failure_sdk.py

@@ -0,0 +1,99 @@
+"""A provider 401 on a caller's own per-user GitHub Copilot session is the caller's credential failing, so the
+shared Copilot deployment keeps no failure, request or outage series for it, while a provider 401 on the router
+deployment's own key, on a key a router caller passes or on the key a direct SDK call brings is counted as a failed
+request, the router caller's key against its own one-off deployment."""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Iterator
+from typing import Final
+
+import pytest
+from integration._support.copilot_session_prometheus_worker import (
+    CALLER_SESSION_TOKEN,
+    COPILOT_DEPLOYMENT,
+    DEPLOYMENT_SERIES,
+    DIRECT_CALL,
+    DIRECT_KEY,
+    FAILURES,
+    LLM_FAILED,
+    OPERATOR_DEPLOYMENT,
+    OPERATOR_KEY,
+    ROUTER_CALLER_KEY,
+    STATE,
+    TOTALS,
+    Outcome,
+    WorkerReport,
+    run_copilot_session_worker,
+)
+from integration._support.wire import Reply, Request, wire_server
+
+
+@pytest.fixture(scope="module")
+def ran() -> Iterator[tuple[WorkerReport, tuple[Request, ...]]]:
+    def reject(request: Request) -> Reply:
+        assert request.method == "POST", request.method
+        return Reply(
+            status=401,
+            body=json.dumps({"error": {"message": "bad credentials", "type": "authentication_error"}}).encode(),
+        )
+
+    with wire_server(reject) as upstream:
+        report: Final = run_copilot_session_worker(upstream.url)
+        yield report, upstream.drain()
+
+
+def test_every_call_reaches_the_provider_with_its_own_credential_and_fails_with_its_401(
+    ran: tuple[WorkerReport, tuple[Request, ...]],
+) -> None:
+    report, received = ran
+    assert report.outcomes == (
+        Outcome(deployment=COPILOT_DEPLOYMENT, status=401),
+        Outcome(deployment=OPERATOR_DEPLOYMENT, status=401),
+        Outcome(deployment=OPERATOR_DEPLOYMENT, status=401),
+        Outcome(deployment=DIRECT_CALL, status=401),
+    )
+    assert [r.headers.get("authorization") for r in received] == [
+        f"Bearer {CALLER_SESSION_TOKEN}",
+        f"Bearer {OPERATOR_KEY}",
+        f"Bearer {ROUTER_CALLER_KEY}",
+        f"Bearer {DIRECT_KEY}",
+    ]
+
+
+def test_a_provider_401_on_a_callers_copilot_session_stays_out_of_the_shared_deployment_metrics(
+    ran: tuple[WorkerReport, tuple[Request, ...]],
+) -> None:
+    report, _ = ran
+    assert {s.metric for s in report.series if s.model_id == COPILOT_DEPLOYMENT} & DEPLOYMENT_SERIES == set()
+
+
+def test_a_provider_401_on_the_deployments_own_key_counts_as_a_deployment_failure(
+    ran: tuple[WorkerReport, tuple[Request, ...]],
+) -> None:
+    report, _ = ran
+    assert report.values(FAILURES, OPERATOR_DEPLOYMENT) == (1.0,)
+    assert report.values(TOTALS, OPERATOR_DEPLOYMENT) == (1.0,)
+    assert report.values(LLM_FAILED, OPERATOR_DEPLOYMENT) == (1.0,)
+    assert report.on_the_endpoint(STATE, OPERATOR_DEPLOYMENT) == (1.0,)
+
+
+def test_a_provider_401_on_the_key_a_direct_sdk_call_brings_is_counted_as_a_failed_request(
+    ran: tuple[WorkerReport, tuple[Request, ...]],
+) -> None:
+    report, _ = ran
+    assert report.outside_the_router(FAILURES) == (1.0,)
+    assert report.outside_the_router(TOTALS) == (1.0,)
+    assert report.outside_the_router(LLM_FAILED) == (1.0,)
+
+
+def test_a_provider_401_on_a_key_a_router_caller_passes_counts_against_its_own_one_off_deployment(
+    ran: tuple[WorkerReport, tuple[Request, ...]],
+) -> None:
+    report, _ = ran
+    assert report.on_a_one_off_deployment(FAILURES) == (1.0,)
+    assert report.on_a_one_off_deployment(TOTALS) == (1.0,)
+    assert report.on_a_one_off_deployment(LLM_FAILED) == (1.0,)
+    assert report.values(FAILURES, OPERATOR_DEPLOYMENT) == (1.0,)
+    assert report.values(TOTALS, OPERATOR_DEPLOYMENT) == (1.0,)

tests/unit/integrations/test_prometheus_invalid_key_filtering.py

@@ -2,19 +2,63 @@
 Unit tests for Prometheus invalid API key request filtering.
 
 Tests the 401 detection helpers, that LLM-level metrics skip invalid API key
-requests, and that the proxy-level failed request counter still records them.
+requests rejected at the proxy gate while recording a provider's 401, and that
+the proxy-level failed request counter still records them.
 """
 
+from collections.abc import Iterator
+from datetime import datetime
+from typing import Final
 from unittest.mock import Mock, patch
 
 import pytest
 from fastapi import HTTPException
 from prometheus_client import REGISTRY
+from prometheus_client.samples import Sample
 
+import litellm
+from litellm.constants import LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL
 from litellm.integrations.prometheus import PrometheusLogger
+from litellm.llms.github_copilot.per_user_auth import GithubCopilotUserSession
 from litellm.proxy._types import ProxyErrorTypes, ProxyException, UserAPIKeyAuth
 
 
+PROVIDER_CREDENTIAL_FIELDS = [
+    "aws_access_key_id",
+    "aws_secret_access_key",
+    "aws_session_token",
+    "aws_web_identity_token",
+    "azure_ad_token",
+    "azure_password",
+    "azure_scope",
+    "azure_username",
+    "client_id",
+    "client_secret",
+    "gigachat_access_token",
+    "oci_fingerprint",
+    "oci_key",
+    "oci_key_file",
+    "oci_tenancy",
+    "oci_user",
+    "tenant_id",
+    "token",
+    "user_config",
+    "vertex_ai_credentials",
+    "vertex_credentials",
+    "watsonx_credentials",
+    "wx_credentials",
+    "zen_api_key",
+]
+PROVIDER_ENDPOINT_FIELDS = [
+    "aws_bedrock_runtime_endpoint",
+    "aws_sts_endpoint",
+    "deployment_url",
+    "s3_endpoint_url",
+    "sagemaker_base_url",
+]
+CALLED_THE_PROVIDER: Final = {"api_call_start_time": datetime(2026, 10, 19, 12, 0, 0)}
+
+
 @pytest.fixture(scope="function")
 def prometheus_logger():
     """Create a PrometheusLogger instance for testing."""
@@ -48,18 +92,13 @@ class TestExtractStatusCode:
             (ExceptionWithStatusCode, 401, 401),
         ],
     )
-    def test_extract_from_exception(
-        self, prometheus_logger, exception_class, code_value, expected
-    ):
+    def test_extract_from_exception(self, prometheus_logger, exception_class, code_value, expected):
         exception = exception_class(code_value)
         assert prometheus_logger._extract_status_code(exception=exception) == expected
 
     def test_extract_from_kwargs(self, prometheus_logger):
         exception = ExceptionWithCode("401")
-        assert (
-            prometheus_logger._extract_status_code(kwargs={"exception": exception})
-            == 401
-        )
+        assert prometheus_logger._extract_status_code(kwargs={"exception": exception}) == 401
 
     def test_extract_from_enum_values(self, prometheus_logger):
         enum_values = Mock(status_code="401")
@@ -79,50 +118,252 @@ class TestInvalidAPIKeyDetection:
         ],
     )
     def test_status_code_detection(self, prometheus_logger, status_code, expected):
-        assert (
-            prometheus_logger._is_invalid_api_key_request(status_code=status_code)
-            == expected
-        )
+        assert prometheus_logger._is_invalid_api_key_request(status_code=status_code) == expected
 
     def test_auth_error_message_detection(self, prometheus_logger):
         exception = AssertionError(
             "LiteLLM Virtual Key expected. Received=invalid-key-12345, expected to start with 'sk-'."
         )
-        assert (
-            prometheus_logger._is_invalid_api_key_request(
-                status_code=None, exception=exception
-            )
-            is True
-        )
+        assert prometheus_logger._is_invalid_api_key_request(status_code=None, exception=exception) is True
 
     def test_non_auth_exception_not_detected(self, prometheus_logger):
         exception = ValueError("Some other error")
-        assert (
-            prometheus_logger._is_invalid_api_key_request(
-                status_code=None, exception=exception
-            )
-            is False
-        )
+        assert prometheus_logger._is_invalid_api_key_request(status_code=None, exception=exception) is False
 
 
 class TestSkipMetricsValidation:
     """Test high-level validation method that orchestrates detection and extraction."""
 
-    def test_skip_for_401_exception(self, prometheus_logger):
-        """Test full flow: extraction -> detection -> skip decision."""
-        exception = ExceptionWithCode("401")
-        assert (
-            prometheus_logger._should_skip_metrics_for_invalid_key(exception=exception)
-            is True
-        )
+    @pytest.mark.parametrize(
+        "exception,proxy_gate,expected",
+        [
+            (ExceptionWithCode("401"), True, True),
+            (AssertionError("expected to start with 'sk-'"), True, True),
+            (ExceptionWithStatusCode(429), True, False),
+            (ExceptionWithCode("401"), False, True),
+            (AssertionError("expected to start with 'sk-'"), False, True),
+            (
+                litellm.AuthenticationError(
+                    message="Incorrect API key provided: sk-rotat*****",
+                    llm_provider="openai",
+                    model="gpt-4o-mini",
+                ),
+                False,
+                False,
+            ),
+            (
+                litellm.AuthenticationError(
+                    message="API key not valid. Please pass a valid API key.",
+                    llm_provider="gemini",
+                    model="gemini-2.5-flash",
+                ),
+                False,
+                False,
+            ),
+            (
+                litellm.AuthenticationError(
+                    message="Authentication Error, Invalid proxy server token passed.",
+                    llm_provider="openai",
+                    model="gpt-4o-mini",
+                ),
+                True,
+                True,
+            ),
+            (
+                litellm.CallerCredentialAuthenticationError(
+                    message="GitHub Copilot connection is missing for this user",
+                    llm_provider="github_copilot",
+                    model="gpt-4o",
+                ),
+                False,
+                True,
+            ),
+            (
+                HTTPException(status_code=401, detail="caller did not present a token"),
+                False,
+                True,
+            ),
+            (
+                litellm.BadRequestError(
+                    message="API key not valid. Please pass a valid API key.",
+                    llm_provider="gemini",
+                    model="gemini-2.5-flash",
+                ),
+                False,
+                False,
+            ),
+        ],
+        ids=[
+            "proxy-gate-401",
+            "proxy-gate-auth-message",
+            "proxy-gate-429",
+            "unmarked-proxy-401",
+            "unmarked-auth-message",
+            "provider-401",
+            "provider-auth-message",
+            "proxy-gate-litellm-auth-error",
+            "caller-credential-401",
+            "unmarked-http-exception-401",
+            "provider-400-auth-message",
+        ],
+    )
+    def test_skip_except_for_provider_errors_the_proxy_did_not_raise(
+        self, prometheus_logger, exception, proxy_gate, expected
+    ):
+        kwargs: Final[dict[str, object]] = {"exception": exception, **CALLED_THE_PROVIDER}
+        if proxy_gate:
+            kwargs[LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL] = True
+        assert prometheus_logger._should_skip_metrics_for_invalid_key(kwargs=kwargs) is expected
 
-    def test_skip_for_auth_error_message(self, prometheus_logger):
-        """Test full flow: exception message -> detection -> skip decision."""
-        exception = AssertionError("expected to start with 'sk-'")
-        assert (
-            prometheus_logger._should_skip_metrics_for_invalid_key(exception=exception)
-            is True
+    @pytest.mark.parametrize(
+        "logged_call,expected",
+        [
+            ({"litellm_params": {"proxy_server_request": {"caller_credential_fields": ("api_key",)}}}, True),
+            (
+                {
+                    "litellm_params": {
+                        "github_copilot_user_session": GithubCopilotUserSession(
+                            token="caller-session-token", api_base="https://api.githubcopilot.com"
+                        )
+                    }
+                },
+                True,
+            ),
+            ({"standard_logging_object": {"metadata": {"used_client_oauth_token": True}}}, True),
+            ({"litellm_params": {"proxy_server_request": {"caller_credential_fields": ("headers",)}}}, True),
+            (
+                {
+                    "litellm_params": {
+                        "proxy_server_request": {"caller_credential_fields": ("provider_specific_header",)}
+                    }
+                },
+                True,
+            ),
+            ({"litellm_params": {"proxy_server_request": {"caller_credential_fields": ()}}}, False),
+            ({"litellm_params": {"proxy_server_request": {"caller_credential_fields": "api_key"}}}, False),
+            ({"standard_logging_object": {"metadata": {"used_client_oauth_token": False}}}, False),
+            ({"litellm_params": {"proxy_server_request": {"body": {"azure_ad_token": "caller-token"}}}}, True),
+            (
+                {
+                    "litellm_params": {
+                        "proxy_server_request": {"body": {"extra_body": {"azure_ad_token": "caller-token"}}}
+                    }
+                },
+                True,
+            ),
+            (
+                {
+                    "litellm_params": {
+                        "azure_ad_token": "deployment-token",
+                        "proxy_server_request": {"body": {"model": "azure-deployment"}},
+                    }
+                },
+                False,
+            ),
+            ({"litellm_params": {"proxy_server_request": {"body": {"extra_body": "azure_ad_token"}}}}, False),
+            ({"litellm_params": {"proxy_server_request": {"caller_credential_fields": ("extra_headers",)}}}, True),
+            ({"litellm_params": {"proxy_server_request": {"caller_credential_fields": ("organization",)}}}, True),
+            (
+                {
+                    "litellm_params": {
+                        "proxy_server_request": {"credential_fields": ("headers",), "caller_credential_fields": ()}
+                    }
+                },
+                False,
+            ),
+            (
+                {
+                    "litellm_params": {
+                        "proxy_server_request": {"body": {"extra_headers": {"Authorization": "Bearer sk-caller"}}}
+                    }
+                },
+                False,
+            ),
+        ],
+        ids=[
+            "caller-supplied-api-key",
+            "caller-copilot-session",
+            "caller-anthropic-oauth-token",
+            "caller-supplied-headers",
+            "caller-supplied-provider-specific-header",
+            "no-caller-credential-fields",
+            "malformed-credential-fields",
+            "deployment-anthropic-key",
+            "caller-supplied-azure-ad-token",
+            "caller-supplied-azure-ad-token-in-extra-body",
+            "deployment-azure-ad-token",
+            "malformed-extra-body",
+            "caller-supplied-extra-headers",
+            "caller-supplied-organization",
+            "headers-that-cannot-change-provider-auth",
+            "extra-headers-not-stamped-by-the-proxy",
+        ],
+    )
+    def test_skip_a_provider_401_on_a_credential_the_caller_owns(self, prometheus_logger, logged_call, expected):
+        exception: Final = litellm.AuthenticationError(
+            message="Incorrect API key provided: sk-calle*****",
+            llm_provider="openai",
+            model="gpt-4o-mini",
         )
+        kwargs: Final = {"exception": exception, **CALLED_THE_PROVIDER, **logged_call}
+        assert prometheus_logger._should_skip_metrics_for_invalid_key(kwargs=kwargs) is expected
+
+    @pytest.mark.parametrize(
+        "logged_call,expected",
+        [(CALLED_THE_PROVIDER, False), ({}, True), ({"api_call_start_time": None}, True)],
+        ids=["provider-401", "401-before-the-provider-call", "blank-provider-call-start"],
+    )
+    def test_skip_a_401_raised_before_the_provider_was_called(self, prometheus_logger, logged_call, expected):
+        exception: Final = litellm.AuthenticationError(
+            message="Missing Anthropic API Key - A call is being made to anthropic but no key is set",
+            llm_provider="anthropic",
+            model="claude-haiku-4-5",
+        )
+        kwargs: Final = {"exception": exception, **logged_call}
+        assert prometheus_logger._should_skip_metrics_for_invalid_key(kwargs=kwargs) is expected
+
+    @pytest.mark.parametrize("field", PROVIDER_CREDENTIAL_FIELDS + PROVIDER_ENDPOINT_FIELDS)
+    @pytest.mark.parametrize("inside_extra_body", [False, True], ids=["body", "extra-body"])
+    def test_skip_a_provider_401_on_a_provider_credential_field_the_caller_sent(
+        self, prometheus_logger, field, inside_extra_body
+    ):
+        sent: Final = {field: "caller-credential"}
+        body: Final = {"model": "shared-deployment", **({"extra_body": sent} if inside_extra_body else sent)}
+        kwargs: Final = {
+            "exception": litellm.AuthenticationError(message="Unauthorized", llm_provider="azure", model="gpt-4o"),
+            "litellm_params": {"proxy_server_request": {"body": body}},
+            **CALLED_THE_PROVIDER,
+        }
+        assert prometheus_logger._should_skip_metrics_for_invalid_key(kwargs=kwargs) is True
+
+    @pytest.mark.parametrize("field", PROVIDER_CREDENTIAL_FIELDS + PROVIDER_ENDPOINT_FIELDS)
+    def test_count_a_provider_401_on_a_provider_credential_field_the_deployment_owns(self, prometheus_logger, field):
+        kwargs: Final = {
+            "exception": litellm.AuthenticationError(message="Unauthorized", llm_provider="azure", model="gpt-4o"),
+            **CALLED_THE_PROVIDER,
+            "litellm_params": {
+                field: "deployment-credential",
+                "proxy_server_request": {"body": {"model": "shared-deployment"}},
+            },
+        }
+        assert prometheus_logger._should_skip_metrics_for_invalid_key(kwargs=kwargs) is False
+
+    @pytest.mark.parametrize(
+        "token_exchange_endpoint,expected",
+        [("https://login.example.com/oauth2/v2.0/token", True), ("", False), (None, False)],
+        ids=["token-exchange-deployment", "blank-endpoint", "no-endpoint"],
+    )
+    def test_skip_a_provider_401_on_a_token_exchange_deployment(
+        self, prometheus_logger, token_exchange_endpoint, expected
+    ):
+        kwargs: Final = {
+            "exception": litellm.AuthenticationError(
+                message="Unauthorized", llm_provider="microsoft_365_copilot", model="copilot"
+            ),
+            "litellm_params": {"token_exchange_endpoint": token_exchange_endpoint, "client_secret": "deployment"},
+            **CALLED_THE_PROVIDER,
+        }
+        assert prometheus_logger._should_skip_metrics_for_invalid_key(kwargs=kwargs) is expected
 
     def test_no_skip_for_valid_request(self, prometheus_logger):
         assert prometheus_logger._should_skip_metrics_for_invalid_key() is False
@@ -147,19 +388,13 @@ class TestAsyncHooks:
             ),
         ],
     )
-    async def test_post_call_failure_hook_counts_401_without_key_hash(
-        self, prometheus_logger, exception
-    ):
+    async def test_post_call_failure_hook_counts_401_without_key_hash(self, prometheus_logger, exception):
         unauthenticated = UserAPIKeyAuth(request_route="/v1/chat/completions")
         unauthenticated.api_key = "notakeyatall"
 
         with (
-            patch.object(
-                prometheus_logger, "litellm_proxy_failed_requests_metric"
-            ) as mock_failed,
-            patch.object(
-                prometheus_logger, "litellm_proxy_total_requests_metric"
-            ) as mock_total,
+            patch.object(prometheus_logger, "litellm_proxy_failed_requests_metric") as mock_failed,
+            patch.object(prometheus_logger, "litellm_proxy_total_requests_metric") as mock_total,
         ):
             await prometheus_logger.async_post_call_failure_hook(
                 request_data={"model": "test-model"},
@@ -176,9 +411,7 @@ class TestAsyncHooks:
         mock_total.labels.return_value.inc.assert_called_once()
 
     @pytest.mark.asyncio
-    async def test_post_call_failure_hook_keeps_resolved_identity_labels_for_401(
-        self, prometheus_logger
-    ):
+    async def test_post_call_failure_hook_keeps_resolved_identity_labels_for_401(self, prometheus_logger):
         expired_key = UserAPIKeyAuth(
             api_key="sk-expired",
             key_alias="expired-alias",
@@ -191,9 +424,7 @@ class TestAsyncHooks:
             code=401,
         )
 
-        with patch.object(
-            prometheus_logger, "litellm_proxy_failed_requests_metric"
-        ) as mock_failed:
+        with patch.object(prometheus_logger, "litellm_proxy_failed_requests_metric") as mock_failed:
             await prometheus_logger.async_post_call_failure_hook(
                 request_data={"model": "test-model"},
                 original_exception=exception,
@@ -207,7 +438,7 @@ class TestAsyncHooks:
         assert failed_labels["team"] == "team-1"
 
     @pytest.mark.asyncio
-    async def test_log_failure_event_skips_401(self, prometheus_logger):
+    async def test_log_failure_event_skips_401_rejected_at_the_proxy_gate(self, prometheus_logger):
         exception = ExceptionWithCode("401")
         kwargs = {
             "model": "test-model",
@@ -220,20 +451,91 @@ class TestAsyncHooks:
             },
             "exception": exception,
             "litellm_params": {},
+            LITELLM_LOGGING_NO_UPSTREAM_LLM_CALL: True,
         }
 
         with (
-            patch.object(
-                prometheus_logger, "litellm_llm_api_failed_requests_metric"
-            ) as mock_failed,
-            patch.object(
-                prometheus_logger, "set_llm_deployment_failure_metrics"
-            ) as mock_deployment,
+            patch.object(prometheus_logger, "litellm_llm_api_failed_requests_metric") as mock_failed,
+            patch.object(prometheus_logger, "set_llm_deployment_failure_metrics") as mock_deployment,
         ):
-
             await prometheus_logger.async_log_failure_event(
                 kwargs=kwargs, response_obj=None, start_time=None, end_time=None
             )
 
             mock_failed.labels.assert_not_called()
             mock_deployment.assert_not_called()
+
+
+def _provider_auth_failure_kwargs() -> dict[str, object]:
+    exception: Final = litellm.AuthenticationError(
+        message="Incorrect API key provided: sk-rotat*****",
+        llm_provider="openai",
+        model="gpt-4o-mini",
+    )
+    return {
+        "model": "gpt-4o-mini",
+        "standard_logging_object": {
+            "metadata": {
+                "user_api_key_hash": "test-key",
+                "user_api_key_alias": "test-alias",
+                "user_api_key_user_id": "test-user",
+                "user_api_key_team_id": "test-team",
+                "user_api_key_team_alias": "test-team-alias",
+            },
+            "model_group": "rotated-credential",
+            "model_id": "deployment-1",
+            "api_base": "https://api.openai.com/v1",
+            "custom_llm_provider": "openai",
+        },
+        "exception": exception,
+        "litellm_params": {"custom_llm_provider": "openai", "metadata": {}},
+        **CALLED_THE_PROVIDER,
+    }
+
+
+def _metric_values(name: str, model_id: str) -> list[float]:
+    return [
+        sample.value
+        for sample in _registry_samples()
+        if sample.name == name and sample.labels.get("model_id") == model_id
+    ]
+
+
+def _registry_samples() -> Iterator[Sample]:
+    for metric in REGISTRY.collect():
+        yield from metric.samples
+
+
+class TestProviderAuthFailures:
+    """A 401 raised by the provider is an upstream failure and is recorded."""
+
+    @pytest.mark.asyncio
+    async def test_log_failure_event_records_a_provider_401(self, prometheus_logger):
+        await prometheus_logger.async_log_failure_event(
+            kwargs=_provider_auth_failure_kwargs(),
+            response_obj=None,
+            start_time=None,
+            end_time=None,
+        )
+
+        assert _metric_values("litellm_llm_api_failed_requests_metric_total", "deployment-1") == [1.0]
+        assert _metric_values("litellm_deployment_failure_responses_total", "deployment-1") == [1.0]
+        assert _metric_values("litellm_deployment_state", "deployment-1") == [1.0]
+
+    def test_deployment_failure_metrics_record_a_provider_401(self, prometheus_logger):
+        prometheus_logger.set_llm_deployment_failure_metrics(_provider_auth_failure_kwargs())
+
+        failures: Final = [
+            sample for sample in _registry_samples() if sample.name == "litellm_deployment_failure_responses_total"
+        ]
+        assert [
+            (
+                sample.labels["model_id"],
+                sample.labels["exception_status"],
+                sample.labels["api_provider"],
+                sample.value,
+            )
+            for sample in failures
+        ] == [("deployment-1", "401", "openai", 1.0)]
+        assert _metric_values("litellm_deployment_total_requests_total", "deployment-1") == [1.0]
+        assert _metric_values("litellm_deployment_state", "deployment-1") == [1.0]

tests/unit/proxy/test_litellm_pre_call_utils.py

@@ -879,6 +879,222 @@ async def test_add_litellm_data_to_request_body_snapshot_excludes_proxy_server_r
     assert snapshot_body["messages"] == [{"role": "user", "content": "hello"}]
 
 
+FORWARD_PROVIDER_AUTH_HEADERS: Final = {
+    "forward_llm_provider_auth_headers": True,
+    "forward_client_headers_to_llm_api": True,
+}
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "sent,request_headers,general_settings,expected",
+    [
+        ({"headers": {"Authorization": "Bearer sk-caller"}}, {}, {}, ("headers",)),
+        ({"headers": {"X-Trace-Id": "trace-1"}}, {}, {}, ("headers",)),
+        ({"headers": {}}, {}, {}, ()),
+        ({"extra_headers": {"API-Key": "caller-azure-key"}}, {}, {}, ("extra_headers",)),
+        ({"extra_headers": {"OpenAI-Project": "proj_caller"}}, {}, {}, ("extra_headers",)),
+        (
+            {"provider_specific_header": {"custom_llm_provider": "openai", "extra_headers": {"x-api-key": "caller"}}},
+            {},
+            {},
+            ("provider_specific_header",),
+        ),
+        (
+            {
+                "provider_specific_header": [
+                    {"custom_llm_provider": "anthropic", "extra_headers": {"anthropic-beta": "tools"}},
+                    {"custom_llm_provider": "gemini", "extra_headers": {"X-Goog-Api-Key": "caller"}},
+                ]
+            },
+            {},
+            {},
+            ("provider_specific_header",),
+        ),
+        (
+            {
+                "provider_specific_header": {
+                    "custom_llm_provider": "anthropic",
+                    "extra_headers": {"x-caller-trace": "x"},
+                }
+            },
+            {},
+            {},
+            ("provider_specific_header",),
+        ),
+        (
+            {
+                "provider_specific_header": {
+                    "custom_llm_provider": "anthropic",
+                    "extra_headers": {"anthropic-version": "2023-06-01", "anthropic-beta": "tools-2024-04-04"},
+                }
+            },
+            {},
+            {},
+            (),
+        ),
+        ({"headers": {"Anthropic-Version": "2023-06-01"}}, {}, {}, ()),
+        ({"organization": "org-caller"}, {}, {}, ("organization",)),
+        ({"headers": "authorization", "provider_specific_header": "x-api-key"}, {}, {}, ()),
+        (
+            {"api_key": "sk-caller", "headers": {"Ocp-Apim-Subscription-Key": "caller"}},
+            {},
+            {},
+            ("api_key", "headers"),
+        ),
+        ({}, {"anthropic-beta": "tools-2024-04-04"}, {}, ()),
+        ({}, {"anthropic-version": "2023-06-01", "anthropic-beta": "tools-2024-04-04"}, {}, ()),
+        ({}, {"openai-organization": "org-caller"}, {"forward_openai_org_id": True}, ("organization",)),
+        (
+            {},
+            {"x-litellm-api-key": "sk-virtual", "authorization": "Bearer sk-ant-oat01-caller"},
+            {},
+            ("provider_specific_header",),
+        ),
+        (
+            {},
+            {"authorization": "Bearer sk-virtual", "x-goog-api-key": "AIza-caller"},
+            FORWARD_PROVIDER_AUTH_HEADERS,
+            ("headers",),
+        ),
+        (
+            {},
+            {"authorization": "Bearer sk-virtual", "x-portkey-api-key": "caller-gateway-key"},
+            FORWARD_PROVIDER_AUTH_HEADERS,
+            ("headers",),
+        ),
+        ({}, {"authorization": "Bearer sk-virtual"}, FORWARD_PROVIDER_AUTH_HEADERS, ()),
+    ],
+    ids=[
+        "authorization-in-headers",
+        "plain-headers",
+        "empty-headers",
+        "api-key-in-extra-headers",
+        "openai-project-in-extra-headers",
+        "credential-in-provider-specific-header",
+        "credential-in-one-provider-specific-header-entry",
+        "plain-provider-specific-header",
+        "anthropic-version-and-beta-in-provider-specific-header",
+        "anthropic-version-in-headers",
+        "organization",
+        "malformed-header-fields",
+        "api-key-and-credential-headers",
+        "proxy-forwarded-anthropic-beta",
+        "proxy-forwarded-anthropic-sdk-headers",
+        "proxy-forwarded-openai-organization",
+        "proxy-forwarded-anthropic-oauth-token",
+        "proxy-forwarded-provider-auth-header",
+        "proxy-forwarded-gateway-key",
+        "proxy-forwarding-nothing",
+    ],
+)
+async def test_caller_credential_fields_name_every_field_that_carries_a_caller_header_or_credential(
+    sent: dict[str, object],
+    request_headers: dict[str, str],
+    general_settings: dict[str, object],
+    expected: tuple[str, ...],
+) -> None:
+    updated: Final = await _request_data_added_by_the_proxy(sent, request_headers, general_settings)
+
+    assert updated["proxy_server_request"]["caller_credential_fields"] == expected
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "request_headers,expected",
+    [({}, ()), ({"x-caller-trace": "trace-1"}, ("headers",))],
+    ids=["user-information-only", "user-information-and-caller-header"],
+)
+async def test_caller_credential_fields_ignore_the_user_information_the_proxy_adds_to_forwarded_headers(
+    monkeypatch: pytest.MonkeyPatch, request_headers: dict[str, str], expected: tuple[str, ...]
+) -> None:
+    monkeypatch.setattr(litellm, "add_user_information_to_llm_headers", True)
+
+    updated: Final = await _request_data_added_by_the_proxy({}, request_headers, FORWARD_PROVIDER_AUTH_HEADERS)
+
+    assert updated["headers"]["x-litellm-user_api_key_user_id"] == "test-user"
+    assert updated["proxy_server_request"]["caller_credential_fields"] == expected
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "sent,request_headers,general_settings,expected",
+    [
+        ({"headers": {"Anthropic-Version": "2023-06-01"}}, {}, {}, ("headers",)),
+        (
+            {
+                "provider_specific_header": {
+                    "custom_llm_provider": "anthropic",
+                    "extra_headers": {"anthropic-version": "2023-06-01"},
+                }
+            },
+            {},
+            {},
+            ("provider_specific_header",),
+        ),
+        ({"organization": "org-caller"}, {}, {}, ()),
+        ({"extra_headers": {"API-Key": "caller-azure-key"}}, {}, {}, ()),
+        (
+            {"api_key": "sk-caller", "headers": {"Ocp-Apim-Subscription-Key": "caller"}},
+            {},
+            {},
+            ("api_key", "headers"),
+        ),
+        ({}, {"openai-organization": "org-caller"}, {"forward_openai_org_id": True}, ()),
+        (
+            {},
+            {"authorization": "Bearer sk-virtual", "x-goog-api-key": "AIza-caller"},
+            FORWARD_PROVIDER_AUTH_HEADERS,
+            (),
+        ),
+    ],
+    ids=[
+        "anthropic-version-in-headers",
+        "anthropic-version-in-provider-specific-header",
+        "organization",
+        "api-key-in-extra-headers",
+        "api-key-and-credential-headers",
+        "proxy-forwarded-openai-organization",
+        "proxy-forwarded-provider-auth-header",
+    ],
+)
+async def test_credential_fields_name_the_stripped_fields_the_caller_sent(
+    sent: dict[str, object],
+    request_headers: dict[str, str],
+    general_settings: dict[str, object],
+    expected: tuple[str, ...],
+) -> None:
+    updated: Final = await _request_data_added_by_the_proxy(sent, request_headers, general_settings)
+
+    assert updated["proxy_server_request"]["credential_fields"] == expected
+
+
+async def _request_data_added_by_the_proxy(
+    sent: dict[str, object], request_headers: dict[str, str], general_settings: dict[str, object]
+) -> dict:
+    from litellm.proxy.litellm_pre_call_utils import add_litellm_data_to_request
+
+    request_mock: Final = MagicMock(spec=Request)
+    request_mock.url = MagicMock()
+    request_mock.url.__str__.return_value = "http://localhost/v1/chat/completions"
+    request_mock.url.path = "/v1/chat/completions"
+    request_mock.method = "POST"
+    request_mock.query_params = {}
+    request_mock.headers = {"content-type": "application/json", **request_headers}
+    request_mock.client = MagicMock()
+    request_mock.client.host = "127.0.0.1"
+    data: Final = {"model": "gpt-4o-mini", "messages": [{"role": "user", "content": "hello"}], **sent}
+
+    return await add_litellm_data_to_request(
+        data=data,
+        request=request_mock,
+        user_api_key_dict=UserAPIKeyAuth(api_key="hashed-key", user_id="test-user"),
+        proxy_config=MagicMock(),
+        general_settings=dict(general_settings),
+        version="test-version",
+    )
+
+
 def test_initial_snapshot_refresh_clears_a_previous_guardrail_checkpoint() -> None:
     from litellm.integrations.shadow_eval_logger import GuardrailRequestSnapshot
     from litellm.litellm_core_utils.litellm_logging import Logging