scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-29 06:33:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 200

fix(prometheus): count provider-side 401s in LLM failure and deployment metrics (#46668)

Browse source 
* fix(prometheus): count provider-side 401s in LLM failure and deployment metrics

The invalid virtual key skip no longer applies to an error the LLM provider
returned while the deployment was called with its own credential, so a 401
for a bad deployment credential increments litellm_llm_api_failed_requests_metric,
litellm_deployment_failure_responses and litellm_deployment_total_requests and
marks litellm_deployment_state as a partial outage again. Proxy-gate rejections
and a failing credential that belongs to the caller (a rejected per-user
credential, an api_key sent with the request, a per-user GitHub Copilot
session) keep the skip

* fix(prometheus): keep a caller's own Anthropic OAuth token failure out of the deployment metrics

A provider 401 on an Anthropic OAuth token the caller forwarded through the proxy is the caller's credential
failing, so it stays out of the deployment failure and outage series like a caller-supplied api_key. Adds
integration cells for an Anthropic deployment's own key, the forwarded caller OAuth token and a direct SDK
call that brings its own key

* test(prometheus): pin how a caller's own key or Authorization header on a deployment is counted

An SDK Router call that passes its own api_key is counted against the one-off deployment the router builds for that key, so the operator deployment keeps only its own failure. A provider 401 on an Authorization header the caller sends through the proxy counts against the deployment, as on v1.80.12

* fix(prometheus): keep caller-sent provider credentials and token-exchange 401s out of deployment metrics

A caller can send its own provider credential in the request body, such as
azure_ad_token, azure_scope, the Azure Entra app fields, watsonx token,
gigachat_access_token, OCI keys, AWS keys or Vertex credentials, at the top
level or inside extra_body. A provider 401 on that credential is the caller's
failure, so it stays out of the shared deployment's failure, request,
LLM-failed and state series, as on main. A token-exchange deployment trades the
caller's identity token on every call, so its 401s stay out as well, matching
the router cooldown.

* test(prometheus): Final locals, precise helper types and one for per comprehension in the new test helpers

* test(prometheus): pin that a provider 401 at a caller-sent api_base counts against its one-off deployment

* fix(prometheus): keep a provider 401 on caller-sent headers out of deployment metrics

A caller can override the deployment's credential with headers it sends in the
request body: headers or provider_specific_header, or an Authorization, api-key,
x-api-key, x-goog-api-key or Ocp-Apim-Subscription-Key entry in extra_headers.
A provider 401 on such a request is the caller's failure, so it stays out of the
shared deployment's failure, request, LLM-failed and state series, as on main.

* fix(prometheus): skip a provider 401 only when the caller's headers carry a credential

The proxy now records headers, extra_headers or provider_specific_header in
credential_fields only when they name a provider credential header
(Authorization, api-key, x-api-key, x-goog-api-key, Ocp-Apim-Subscription-Key),
the same way api_key is recorded. A caller that sends a plain header such as a
trace id still has a provider 401 on the deployment's own key counted against
the deployment.

* fix(prometheus): stamp credential_fields after the proxy forwards the caller's headers

A provider credential header the proxy forwards from the caller's HTTP request
(forward_llm_provider_auth_headers + forward_client_headers_to_llm_api) lands in
data["headers"] after proxy_server_request is built, so the stamp missed it and a
401 on that caller credential was counted against the deployment. Compute the
stamp after the header forwarding steps and pin both forwarded cases.

* fix(prometheus): skip a provider 401 when the caller sends any header or an OpenAI organization

The proxy now records headers, extra_headers or provider_specific_header in
credential_fields whenever the caller sends or the proxy forwards any header
there, and records organization when the caller sends one in the body or as a
forwarded OpenAI-Organization header. A forwarded header can be a gateway key
such as x-portkey-api-key or an OpenAI-Project id, and a wrong organization
gets a 401 from OpenAI, so these 401s stay out of the deployment metrics as on
main instead of marking a healthy deployment as a partial outage.

* fix(prometheus): ignore the user information the proxy adds when stamping caller headers

With add_user_information_to_llm_headers on and header forwarding enabled, the
proxy adds x-litellm-user_api_key_* headers to the forwarded headers. Those
carry the proxy's own key and user details, not the caller's, so they no longer
make a request look caller-scoped, and a provider 401 on the deployment's own
key on such a request is counted.

* fix(prometheus): skip a 401 raised before the provider is called or at the caller's own endpoint

A deployment with no key of its own fails with LiteLLM's local "Missing
Anthropic API Key" 401 before any provider call, and a caller who sends
aws_bedrock_runtime_endpoint (or another provider endpoint field) can get a
401 from their own gateway. Both were counted against the shared deployment
and marked it as a partial outage, where main skipped every 401.

Count a 401 only once the provider was called (api_call_start_time is set),
and treat caller-sent provider endpoint fields like caller credential fields.

* fix(prometheus): count the deployment's own 401 on requests that carry only anthropic-version or anthropic-beta

The Anthropic SDK sends anthropic-version on every request, and the proxy copies
anthropic-version and anthropic-beta into provider_specific_header for Anthropic,
Bedrock and Vertex. Both were treated as caller headers, so a 401 on an Anthropic
deployment's own key from any Anthropic SDK client stayed out of the deployment
metrics.

These two headers only pick the API version and beta features. Anthropic answers
a bad anthropic-version or an unknown anthropic-beta with a 400, never a 401, so
they cannot turn the deployment's own key into a 401 and no longer mark the
request as carrying a caller header.

* test(prometheus): read and write the owned proxy's own database in the expired-key and MCP cells

Since #47213 an owned proxy boots on a private database, so the expired-key update and the MCP guardrail row read now go through the gateway's database_url instead of the worker's.

* fix(prometheus): keep credential_fields as main stamps it and give Prometheus its own caller_credential_fields

credential_fields tells the Rust OCR bridge which stripped body fields the
caller sent. Repurposing it for the Prometheus caller-credential check changed
its value (organization and extra_headers added, anthropic-version-only
headers dropped). Restore main's stamp unchanged and stamp the Prometheus
classification under caller_credential_fields, which only Prometheus reads.

* fix(prometheus): skip a provider 401 on a stored credential the caller picks by litellm_credential_name

A request body litellm_credential_name swaps the deployment's stored
credential for one the caller chose, so a 401 on it is the caller's, not the
deployment's. Treat it like the other caller credential fields so it no longer
marks a healthy shared deployment as a partial outage.

---------

Co-authored-by: yucheng-berri <293926326+yucheng-berri@users.noreply.github.com>
This commit is contained in:
yucheng-berri 2026-10-26 17:57:06 -07:00 • committed by GitHub
parent b5befe2dd0
commit 715e9bb08d
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 2220 additions and 61 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

97
litellm/integrations/prometheus.py
View file

@ -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,87 @@ 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",
"litellm_credential_name",
"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, a stored credential the caller picked by name, 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 +2905,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 +2926,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",

45
litellm/proxy/litellm_pre_call_utils.py
View file

@ -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
@ -1451,6 +1456,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().removeprefix("x-pass-") in _CREDENTIAL_HEADER_NAMES
@ -2605,6 +2648,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)

162
tests/integration/_support/copilot_session_prometheus_worker.py Normal file
View file

@ -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]))

1074
tests/integration/observability/test_prometheus_provider_auth_failure.py Normal file
View file

File diff suppressed because it is too large Load diff

147
tests/integration/observability/test_prometheus_provider_auth_failure_chaos.py Normal file
View file

@ -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

99
tests/integration/sdk/test_prometheus_copilot_session_auth_failure_sdk.py Normal file
View file

@ -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,)

441
tests/unit/integrations/test_prometheus_invalid_key_filtering.py
View file

@ -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,271 @@ 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": {"body": {"litellm_credential_name": "caller-credential"}}
}
},
True,
),
(
{
"litellm_params": {
"litellm_credential_name": "deployment-credential",
"proxy_server_request": {"body": {"model": "openai-deployment"}},
}
},
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-named-credential",
"deployment-named-credential",
"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 +407,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 +430,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 +443,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 +457,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 +470,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]

216
tests/unit/proxy/test_litellm_pre_call_utils.py
View file

@ -927,6 +927,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