From 715e9bb08d7567d58beebff640a7b32c4370342d Mon Sep 17 00:00:00 2001 From: yucheng-berri Date: Mon, 26 Oct 2026 17:57:06 -0700 Subject: [PATCH] fix(prometheus): count provider-side 401s in LLM failure and deployment metrics (#46668) * 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> --- litellm/integrations/prometheus.py | 97 +- litellm/proxy/litellm_pre_call_utils.py | 45 + .../copilot_session_prometheus_worker.py | 162 +++ .../test_prometheus_provider_auth_failure.py | 1074 +++++++++++++++++ ..._prometheus_provider_auth_failure_chaos.py | 147 +++ ...etheus_copilot_session_auth_failure_sdk.py | 99 ++ .../test_prometheus_invalid_key_filtering.py | 441 ++++++- .../unit/proxy/test_litellm_pre_call_utils.py | 216 ++++ 8 files changed, 2220 insertions(+), 61 deletions(-) create mode 100644 tests/integration/_support/copilot_session_prometheus_worker.py create mode 100644 tests/integration/observability/test_prometheus_provider_auth_failure.py create mode 100644 tests/integration/observability/test_prometheus_provider_auth_failure_chaos.py create mode 100644 tests/integration/sdk/test_prometheus_copilot_session_auth_failure_sdk.py diff --git a/litellm/integrations/prometheus.py b/litellm/integrations/prometheus.py index 19d161b8aa4..bc8344fa686 100644 --- a/litellm/integrations/prometheus.py +++ b/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,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", diff --git a/litellm/proxy/litellm_pre_call_utils.py b/litellm/proxy/litellm_pre_call_utils.py index b51eea7a876..5f1948d1c1c 100644 --- a/litellm/proxy/litellm_pre_call_utils.py +++ b/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--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) diff --git a/tests/integration/_support/copilot_session_prometheus_worker.py b/tests/integration/_support/copilot_session_prometheus_worker.py new file mode 100644 index 00000000000..a6b2c5d4692 --- /dev/null +++ b/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])) diff --git a/tests/integration/observability/test_prometheus_provider_auth_failure.py b/tests/integration/observability/test_prometheus_provider_auth_failure.py new file mode 100644 index 00000000000..13d264ca6b0 --- /dev/null +++ b/tests/integration/observability/test_prometheus_provider_auth_failure.py @@ -0,0 +1,1074 @@ +from __future__ import annotations + +import asyncio +import json +import threading +import uuid +from collections import Counter +from collections.abc import Iterator, Sequence +from dataclasses import dataclass +from hashlib import sha256 +from pathlib import Path +from typing import Final + +import httpx +import openai +import pytest +import yaml +from integration._support.client import Gateway, Scenario, eventually, gateway_from_environment +from integration._support.database import write_rows +from integration._support.mcp import ENTRY_POINTS, McpCaller, McpPeer, echo_tool, register_mcp, scripted_peer +from integration._support.process import owned_proxy +from integration._support.prometheus_series import Sample, scrape +from integration._support.wire import Reply, Request, Wire, wire_server +from pydantic import JsonValue + +RUN: Final = uuid.uuid4().hex[:8] +REJECTS: Final = "provider-rejects-credential" +HELD_REJECTS: Final = "provider-rejects-credential-when-released" +KEY_NOT_VALID: Final = "provider-key-not-valid" +RATE_LIMITS: Final = "provider-rate-limits" +RECOVERS: Final = "provider-rejects-once" +HEALTHY: Final = "provider-healthy" +AGENT_365_GUARDRAIL: Final = f"agent-365-{RUN}" +CALLER_OAUTH_TOKEN: Final = f"sk-ant-oat01-caller-{RUN}" +REVOKED_GATEWAY_KEY: Final = f"revoked-gateway-key-{RUN}" +UNKNOWN_ORGANIZATION: Final = f"org-unknown-{RUN}" +REVOKED_STORED_KEY: Final = f"sk-revoked-stored-credential-{RUN}" +AZURE_API_VERSION: Final = "2024-02-01" +CALLER_AZURE_CELLS: Final = ( + ("chat", False), + ("embeddings", False), + ("responses", False), + ("messages", False), + ("responses", True), +) +CALLER_CREDENTIAL_CELLS: Final = ( + ("caller-watsonx-token", "token"), + ("caller-gigachat-access-token", "gigachat_access_token"), +) +TOKEN_EXCHANGE_CELLS: Final = ("token-exchange", "token-exchange-credential") +KEYLESS_CELLS: Final = ("keyless", "messages-keyless") +TOKEN_EXCHANGE_SETTINGS: Final = { + "token_exchange_endpoint": "https://login.microsoftonline.com/00000000-0000-4000-8000-0000000c0b01/oauth2/v2.0/token", + "client_id": "synthetic-obo-client-id", + "client_secret": "synthetic-obo-client-secret", +} + +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" +PROXY_FAILED: Final = "litellm_proxy_failed_requests_metric_total" +SUCCESSES: Final = "litellm_deployment_success_responses_total" +DEPLOYMENT_SERIES: Final = (FAILURES, TOTALS, LLM_FAILED, STATE) + +ENDPOINT_CELLS: Final = ( + "chat", + "chat-stream", + "messages", + "messages-stream", + "responses", + "responses-stream", + "embeddings", +) +REJECTED_CELLS: Final = ( + *ENDPOINT_CELLS, + "sdk-sync", + "sdk-async", + "repeated", + "fallback", + "burst", + "caller-api-key", + "caller-extra-headers", + "caller-headers", + "caller-provider-specific-header", + "plain-extra-headers", + "plain-headers", + "plain-provider-specific-header", + "forwarded-provider-header", + "forwarded-plain-header", + "forwarded-user-information", +) +GATED_CELLS: Final = ("invalid-key", "missing-key", "expired-key", "blocked-key", "no-model-access") +CALLER_SCOPE_CELLS: Final = ("forwarded-gateway-key", "caller-organization", "forwarded-organization") +HEALTHY_CELLS: Final = (*GATED_CELLS, *CALLER_SCOPE_CELLS, "healthy", "key-rpm-limit") +GUARDRAIL_ROWS: Final = ( + "SELECT COALESCE(jsonb_path_query_first(metadata, " + "'$.guardrail_information[*] ? (@.guardrail_name == $name).guardrail_status', " + "jsonb_build_object('name', %s::text)) #>> '{}', 'none') AS status " + 'FROM "LiteLLM_SpendLogs" WHERE api_key = %s AND call_type = %s' +) + + +def name(cell: str) -> str: + return f"prometheus-{cell}-{RUN}" + + +def identity(cell: str) -> str: + return f"{name(cell)}-deployment" + + +def _error(status: int, message: str, kind: str, code: str) -> Reply: + return Reply(status=status, body=json.dumps({"error": {"message": message, "type": kind, "code": code}}).encode()) + + +def _completion() -> Reply: + return Reply( + body=json.dumps( + { + "id": f"chatcmpl-{uuid.uuid4().hex}", + "object": "chat.completion", + "created": 1, + "model": "gpt-4o-mini", + "choices": [{"index": 0, "message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"}], + "usage": {"prompt_tokens": 3, "completion_tokens": 1, "total_tokens": 4}, + } + ).encode() + ) + + +class Provider: + """Answers by upstream model name. `HELD_REJECTS` waits for `release` before rejecting the credential.""" + + def __init__(self, release: threading.Event) -> None: + self._lock: Final = threading.Lock() + self._seen: Final[Counter[str]] = Counter() + self._release: Final = release + + def __call__(self, request: Request) -> Reply: + sent: Final = json.loads(request.body) + model: Final = str(sent.get("model") or sent.get("model_id") or request.target.split("/")[2]) + if request.headers.get("x-portkey-api-key") == REVOKED_GATEWAY_KEY: + return _error(401, "Invalid gateway API key", "invalid_request_error", "invalid_api_key") + if request.headers.get("authorization") == f"Bearer {REVOKED_STORED_KEY}": + return _error(401, "Incorrect API key provided: sk-revok*****", "invalid_request_error", "invalid_api_key") + if request.headers.get("openai-organization") == UNKNOWN_ORGANIZATION: + return _error( + 401, + "OpenAI-Organization header should match organization for API key", + "invalid_request_error", + "mismatched_organization", + ) + with self._lock: + self._seen[model] += 1 + attempt: Final = self._seen[model] + if model == HELD_REJECTS: + assert self._release.wait(timeout=30), "held provider reply was never released" + if model in (REJECTS, HELD_REJECTS) or (model == RECOVERS and attempt == 1): + return _error(401, "Incorrect API key provided: sk-rotat*****", "invalid_request_error", "invalid_api_key") + if model == KEY_NOT_VALID: + return _error(400, "API key not valid. Please pass a valid API key.", "invalid_request_error", "400") + if model == RATE_LIMITS: + return _error(429, "Rate limit reached for requests", "rate_limit_error", "rate_limit_exceeded") + return _completion() + + +def deployment(cell: str, upstream_model: str, api_base: str) -> dict[str, JsonValue]: + return { + "model_name": name(cell), + "litellm_params": { + "model": f"openai/{upstream_model}", + "api_key": f"sk-deployment-credential-{cell}", + "api_base": api_base, + }, + "model_info": {"id": identity(cell)}, + } + + +def anthropic_deployment(cell: str, api_base: str) -> dict[str, JsonValue]: + return { + "model_name": name(cell), + "litellm_params": { + "model": f"anthropic/{REJECTS}", + "api_key": f"sk-ant-api03-deployment-credential-{cell}", + "api_base": api_base, + }, + "model_info": {"id": identity(cell)}, + } + + +def bedrock_deployment(cell: str, opt_in: dict[str, JsonValue]) -> dict[str, JsonValue]: + return { + "model_name": name(cell), + "litellm_params": { + "model": f"bedrock/converse/{REJECTS}", + "aws_access_key_id": "AKIADEPLOYMENTCREDENTIAL", + "aws_secret_access_key": f"deployment-credential-{cell}", + "aws_region_name": "us-east-1", + **opt_in, + }, + "model_info": {"id": identity(cell)}, + } + + +def keyless_anthropic_deployment(cell: str, api_base: str) -> dict[str, JsonValue]: + return { + "model_name": name(cell), + "litellm_params": {"model": f"anthropic/{REJECTS}", "api_base": api_base}, + "model_info": {"id": identity(cell)}, + } + + +def azure_deployment(cell: str, api_base: str, credential: dict[str, JsonValue]) -> dict[str, JsonValue]: + return { + "model_name": name(cell), + "litellm_params": { + "model": f"azure/{REJECTS}", + "api_base": api_base, + "api_version": AZURE_API_VERSION, + **credential, + }, + "model_info": {"id": identity(cell)}, + } + + +def caller_credential_deployment(cell: str, field: str, api_base: str) -> dict[str, JsonValue]: + provider: Final = "watsonx" if field == "token" else "gigachat" + project: Final[dict[str, JsonValue]] = {"project_id": f"project-{RUN}"} if provider == "watsonx" else {} + return { + "model_name": name(cell), + "litellm_params": { + "model": f"{provider}/{REJECTS}", + "api_base": api_base, + field: f"deployment-credential-{cell}", + **project, + }, + "model_info": {"id": identity(cell)}, + } + + +def token_exchange_deployment(cell: str) -> dict[str, JsonValue]: + settings: Final[dict[str, JsonValue]] = ( + {"litellm_credential_name": f"m365-obo-{RUN}"} if cell.endswith("-credential") else {**TOKEN_EXCHANGE_SETTINGS} + ) + return { + "model_name": name(cell), + "litellm_params": {"model": "microsoft_365_copilot/chat", **settings}, + "model_info": {"id": identity(cell)}, + } + + +def caller_azure_cell(opt_in: str, route: str, inside_extra_body: bool) -> str: + return f"{opt_in}-caller-azure-token-{route}{'-extra-body' if inside_extra_body else ''}" + + +def config(directory: Path, upstream: Wire) -> Path: + api_base: Final = f"{upstream.url}/v1" + configuration: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()) + configuration["model_list"] = [ + *configuration["model_list"], + *(deployment(cell, REJECTS, api_base) for cell in REJECTED_CELLS), + *(deployment(cell, HEALTHY, api_base) for cell in HEALTHY_CELLS), + deployment("held", HELD_REJECTS, api_base), + deployment("key-not-valid", KEY_NOT_VALID, api_base), + deployment("rate-limited", RATE_LIMITS, api_base), + deployment("recovers", RECOVERS, api_base), + anthropic_deployment("anthropic-own-key", upstream.url), + anthropic_deployment("anthropic-sdk-own-key", upstream.url), + anthropic_deployment("caller-oauth", upstream.url), + *(keyless_anthropic_deployment(cell, upstream.url) for cell in KEYLESS_CELLS), + bedrock_deployment( + "deployment-caller-bedrock-endpoint", + {"configurable_clientside_auth_params": ["aws_bedrock_runtime_endpoint"]}, + ), + azure_deployment("azure-own-token", upstream.url, {"azure_ad_token": "deployment-entra-token-azure-own-token"}), + *( + azure_deployment( + caller_azure_cell("deployment", route, inside_extra_body), + upstream.url, + {"configurable_clientside_auth_params": ["azure_ad_token"]}, + ) + for route, inside_extra_body in CALLER_AZURE_CELLS + ), + *(caller_credential_deployment(cell, field, upstream.url) for cell, field in CALLER_CREDENTIAL_CELLS), + *(token_exchange_deployment(cell) for cell in TOKEN_EXCHANGE_CELLS), + { + "model_name": name("caller-api-base"), + "litellm_params": { + "model": f"openai/{REJECTS}", + "api_key": "sk-deployment-credential-caller-api-base", + "api_base": api_base, + "configurable_clientside_auth_params": ["api_base"], + }, + "model_info": {"id": identity("caller-api-base")}, + }, + { + "model_name": name("caller-credential"), + "litellm_params": { + "model": "github_copilot/gpt-4o", + "litellm_credential_name": f"copilot-per-user-{RUN}", + "github_copilot_auth_type": "per_user_oauth", + }, + "model_info": {"id": identity("caller-credential")}, + }, + { + "model_name": name("caller-credential-name"), + "litellm_params": { + "model": f"openai/{HEALTHY}", + "api_base": api_base, + "litellm_credential_name": f"deployment-key-{RUN}", + }, + "model_info": {"id": identity("caller-credential-name")}, + }, + ] + configuration["credential_list"] = [ + *configuration.get("credential_list", []), + {"credential_name": f"m365-obo-{RUN}", "credential_values": {**TOKEN_EXCHANGE_SETTINGS}, "credential_info": {}}, + { + "credential_name": f"deployment-key-{RUN}", + "credential_values": {"api_key": "sk-deployment-credential-caller-credential-name"}, + "credential_info": {}, + }, + { + "credential_name": f"revoked-key-{RUN}", + "credential_values": {"api_key": REVOKED_STORED_KEY}, + "credential_info": {}, + }, + ] + configuration["guardrails"] = [ + { + "guardrail_name": AGENT_365_GUARDRAIL, + "litellm_params": { + "guardrail": "agent_365", + "mode": "pre_mcp_call", + "default_on": True, + "tenant_id": "00000000-0000-4000-8000-0000000a3650", + "client_id": "synthetic-client-id", + "client_secret": "synthetic-client-secret", + }, + } + ] + configuration["litellm_settings"]["callbacks"] = ["prometheus"] + configuration["litellm_settings"]["model_group_settings"] = { + "forward_client_headers_to_llm_api": [ + name("forwarded-provider-header"), + name("forwarded-plain-header"), + name("forwarded-gateway-key"), + name("forwarded-user-information"), + ] + } + configuration["litellm_settings"]["add_user_information_to_llm_headers"] = True + configuration["general_settings"]["forward_llm_provider_auth_headers"] = True + configuration["general_settings"]["forward_openai_org_id"] = True + configuration.setdefault("router_settings", {}).update({"num_retries": 0, "disable_cooldowns": True}) + path: Final = directory / "prometheus-provider-auth-failure.yaml" + path.write_text(yaml.safe_dump(configuration)) + return path + + +@dataclass(frozen=True, slots=True) +class Rig: + gateway: Gateway + upstream: Wire + + @property + def api_base(self) -> str: + return f"{self.upstream.url}/v1" + + +@pytest.fixture(scope="module") +def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Rig]: + directory: Final = tmp_path_factory.mktemp("prometheus-provider-auth-failure") + released: Final = threading.Event() + released.set() + with ( + wire_server(Provider(released)) as upstream, + gateway_from_environment() as shared, + owned_proxy( + shared, + directory, + {"PROXY_CONFIG_RELOAD_INTERVAL_SECONDS": "2"}, + config=config(directory, upstream), + remove_environment=("ANTHROPIC_API_KEY", "ANTHROPIC_AUTH_TOKEN"), + ) as owned, + ): + yield Rig(owned, upstream) + + +def opted_in_config(directory: Path, upstream: Wire) -> Path: + configuration: Final = yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()) + configuration["model_list"] = [ + *configuration["model_list"], + *( + azure_deployment(caller_azure_cell("proxy-wide", route, inside_extra_body), upstream.url, {}) + for route, inside_extra_body in CALLER_AZURE_CELLS + ), + bedrock_deployment("proxy-wide-caller-bedrock-endpoint", {}), + ] + configuration["general_settings"]["allow_client_side_credentials"] = True + configuration["litellm_settings"]["callbacks"] = ["prometheus"] + configuration.setdefault("router_settings", {}).update({"num_retries": 0, "disable_cooldowns": True}) + path: Final = directory / "prometheus-caller-credentials-opted-in.yaml" + path.write_text(yaml.safe_dump(configuration)) + return path + + +@pytest.fixture(scope="module") +def opted_in(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Rig]: + directory: Final = tmp_path_factory.mktemp("prometheus-caller-credentials-opted-in") + released: Final = threading.Event() + released.set() + with ( + wire_server(Provider(released)) as upstream, + gateway_from_environment() as shared, + owned_proxy(shared, directory, {}, config=opted_in_config(directory, upstream)) as owned, + ): + yield Rig(owned, upstream) + + +def series(samples: Sequence[Sample], metric: str, **labels: str) -> tuple[Sample, ...]: + return tuple( + sample + for sample in samples + if sample.name == metric and all(sample.labels.get(label) == value for label, value in labels.items()) + ) + + +def values(samples: Sequence[Sample], metric: str, **labels: str) -> list[float]: + return [sample.value for sample in series(samples, metric, **labels)] + + +def until(gateway: Gateway, metric: str, **labels: str) -> tuple[Sample, ...]: + return eventually(lambda: scrape(gateway), lambda samples: bool(series(samples, metric, **labels)), 30) + + +def messages(cell: str) -> list[JsonValue]: + return [{"role": "user", "content": f"{cell} {uuid.uuid4().hex}"}] + + +def send(gateway: Gateway, cell: str, model: str | None = None, key: str | None = None) -> httpx.Response: + asked: Final = model or name(cell) + stream: Final = cell.endswith("-stream") + if cell.startswith("messages"): + anthropic: Final = {"model": asked, "max_tokens": 16, "messages": messages(cell), "stream": stream} + return gateway.request("POST", "/v1/messages", anthropic, key=key) + if cell.startswith("responses"): + return gateway.request("POST", "/v1/responses", {"model": asked, "input": cell, "stream": stream}, key=key) + if cell == "embeddings": + return gateway.request("POST", "/v1/embeddings", {"model": asked, "input": cell}, key=key) + return gateway.request( + "POST", "/v1/chat/completions", {"model": asked, "messages": messages(cell), "stream": stream}, key=key + ) + + +def assert_reached_with_the_deployment_credential(rig: Rig, cell: str, attempts: int = 1) -> None: + reached: Final = rig.upstream.drain() + assert [request.headers["authorization"] for request in reached] == [ + f"Bearer sk-deployment-credential-{cell}" + ] * attempts, [request.target for request in reached] + + +def recorded(samples: Sequence[Sample], **labels: str) -> list[tuple[str, dict[str, str], float]]: + return [(sample.name, dict(sample.labels), sample.value) for sample in series_of(samples, **labels)] + + +def series_of(samples: Sequence[Sample], **labels: str) -> tuple[Sample, ...]: + return tuple( + sample + for sample in samples + if sample.name.startswith("litellm_") + and all(sample.labels.get(label) == value for label, value in labels.items()) + ) + + +def assert_kept_out_of_the_deployment_metrics(rig: Rig, cell: str) -> None: + samples: Final = until(rig.gateway, PROXY_FAILED, requested_model=name(cell)) + assert {metric: values(samples, metric, model_id=identity(cell)) for metric in DEPLOYMENT_SERIES} == { + metric: [] for metric in DEPLOYMENT_SERIES + }, recorded(samples, model_id=identity(cell)) + + +def assert_recorded_as_a_deployment_failure(rig: Rig, cell: str, status: str, count: float = 1.0) -> None: + samples: Final = until(rig.gateway, FAILURES, model_id=identity(cell)) + observed: Final = recorded(samples, model_id=identity(cell)) + assert values(samples, FAILURES, model_id=identity(cell), exception_status=status) == [count], observed + assert values(samples, TOTALS, model_id=identity(cell)) == [count], observed + assert values(samples, LLM_FAILED, model_id=identity(cell)) == [count], observed + assert values(samples, STATE, model_id=identity(cell)) == [1.0], observed + + +@pytest.mark.parametrize("cell", ENDPOINT_CELLS) +def test_a_provider_401_on_every_endpoint_counts_as_a_deployment_failure_and_a_partial_outage( + rig: Rig, cell: str +) -> None: + rig.upstream.drain() + response: Final = send(rig.gateway, cell) + assert response.status_code == 401, response.text + assert_reached_with_the_deployment_credential(rig, cell) + assert_recorded_as_a_deployment_failure(rig, cell, "401") + + +def test_a_provider_401_seen_by_the_sync_openai_client_is_recorded(rig: Rig) -> None: + rig.upstream.drain() + with ( + openai.OpenAI( + base_url=f"{rig.gateway.client.base_url}/v1", + api_key=rig.gateway.key, + max_retries=0, + http_client=httpx.Client(trust_env=False), + ) as client, + pytest.raises(openai.AuthenticationError, match="Incorrect API key provided"), + ): + client.chat.completions.create(model=name("sdk-sync"), messages=[{"role": "user", "content": "sync"}]) + assert_reached_with_the_deployment_credential(rig, "sdk-sync") + assert_recorded_as_a_deployment_failure(rig, "sdk-sync", "401") + + +def test_a_provider_401_seen_by_the_async_openai_client_is_recorded(rig: Rig) -> None: + async def ask() -> None: + async with openai.AsyncOpenAI( + base_url=f"{rig.gateway.client.base_url}/v1", + api_key=rig.gateway.key, + max_retries=0, + http_client=httpx.AsyncClient(trust_env=False), + ) as client: + await client.chat.completions.create(model=name("sdk-async"), messages=[{"role": "user", "content": "a"}]) + + rig.upstream.drain() + with pytest.raises(openai.AuthenticationError, match="Incorrect API key provided"): + asyncio.run(ask()) + assert_reached_with_the_deployment_credential(rig, "sdk-async") + assert_recorded_as_a_deployment_failure(rig, "sdk-async", "401") + + +def test_each_repeated_provider_401_is_counted_once(rig: Rig) -> None: + rig.upstream.drain() + assert [send(rig.gateway, "repeated").status_code for _ in range(3)] == [401, 401, 401] + assert_reached_with_the_deployment_credential(rig, "repeated", attempts=3) + assert_recorded_as_a_deployment_failure(rig, "repeated", "401", count=3.0) + + +def test_a_provider_401_rescued_by_a_fallback_still_counts_against_the_rejected_deployment(rig: Rig) -> None: + rig.upstream.drain() + response: Final = rig.gateway.request( + "POST", + "/v1/chat/completions", + {"model": name("fallback"), "messages": messages("fallback"), "fallbacks": [name("healthy")]}, + ) + assert response.status_code == 200, response.text + assert [request.headers["authorization"] for request in rig.upstream.drain()] == [ + "Bearer sk-deployment-credential-fallback", + "Bearer sk-deployment-credential-healthy", + ] + assert_recorded_as_a_deployment_failure(rig, "fallback", "401") + rescued: Final = eventually( + lambda: scrape(rig.gateway), lambda samples: values(samples, SUCCESSES, model_id=identity("healthy")) == [1.0] + ) + assert values(rescued, FAILURES, model_id=identity("healthy")) == [] + + +def test_a_provider_auth_message_on_a_non_401_status_is_recorded(rig: Rig) -> None: + rig.upstream.drain() + response: Final = send(rig.gateway, "key-not-valid") + assert response.status_code == 400, response.text + assert_reached_with_the_deployment_credential(rig, "key-not-valid") + assert_recorded_as_a_deployment_failure(rig, "key-not-valid", "400") + + +def test_a_provider_429_is_recorded_as_before(rig: Rig) -> None: + rig.upstream.drain() + response: Final = send(rig.gateway, "rate-limited") + assert response.status_code == 429, response.text + assert_reached_with_the_deployment_credential(rig, "rate-limited") + assert_recorded_as_a_deployment_failure(rig, "rate-limited", "429") + + +def test_a_deployment_marked_down_by_a_provider_401_is_healthy_again_after_a_success(rig: Rig) -> None: + rig.upstream.drain() + assert send(rig.gateway, "recovers").status_code == 401 + assert_recorded_as_a_deployment_failure(rig, "recovers", "401") + recovered: Final = send(rig.gateway, "recovers") + assert recovered.status_code == 200, recovered.text + assert len(rig.upstream.drain()) == 2 + eventually( + lambda: scrape(rig.gateway), + lambda samples: values(samples, STATE, model_id=identity("recovers"), api_base=rig.api_base) == [0.0], + ) + + +def test_a_per_caller_credential_rejection_stays_out_of_the_deployment_metrics(rig: Rig) -> None: + rig.upstream.drain() + response: Final = send(rig.gateway, "caller-credential") + assert response.status_code == 401, response.text + assert "Connect GitHub Copilot" in response.text, response.text + assert rig.upstream.drain() == () + samples: Final = until(rig.gateway, PROXY_FAILED, requested_model=name("caller-credential")) + assert values(samples, FAILURES, model_id=identity("caller-credential")) == [] + assert values(samples, TOTALS, model_id=identity("caller-credential")) == [] + assert values(samples, LLM_FAILED, model_id=identity("caller-credential")) == [] + assert values(samples, STATE, model_id=identity("caller-credential")) == [] + + +def deployment_series(samples: Sequence[Sample]) -> frozenset[tuple[str, tuple[tuple[str, str], ...]]]: + return frozenset((sample.name, sample.identity()) for sample in samples if sample.name in DEPLOYMENT_SERIES) + + +def test_a_provider_401_on_an_api_key_the_caller_sent_stays_out_of_the_deployment_metrics(rig: Rig) -> None: + caller_key: Final = f"sk-caller-supplied-{uuid.uuid4().hex}" + before: Final = deployment_series(scrape(rig.gateway)) + rig.upstream.drain() + response: Final = rig.gateway.request( + "POST", + "/v1/chat/completions", + {"model": name("caller-api-key"), "messages": messages("caller-api-key"), "api_key": caller_key}, + ) + assert response.status_code == 401, response.text + assert [request.headers["authorization"] for request in rig.upstream.drain()] == [f"Bearer {caller_key}"] + samples: Final = until(rig.gateway, PROXY_FAILED, requested_model=name("caller-api-key")) + assert deployment_series(samples) - before == frozenset(), recorded(samples, requested_model=name("caller-api-key")) + + +def test_a_provider_401_on_a_stored_credential_the_caller_names_leaves_a_healthy_deployment_healthy(rig: Rig) -> None: + cell: Final = "caller-credential-name" + rig.upstream.drain() + assert send(rig.gateway, cell).status_code == 200 + assert_reached_with_the_deployment_credential(rig, cell) + response: Final = rig.gateway.request( + "POST", + "/v1/chat/completions", + {"model": name(cell), "messages": messages(cell), "litellm_credential_name": f"revoked-key-{RUN}"}, + ) + assert response.status_code == 401, response.text + assert [request.headers["authorization"] for request in rig.upstream.drain()] == [f"Bearer {REVOKED_STORED_KEY}"] + samples: Final = until(rig.gateway, PROXY_FAILED, requested_model=name(cell)) + observed: Final = recorded(samples, model_id=identity(cell)) + assert values(samples, FAILURES, model_id=identity(cell)) == [], observed + assert values(samples, LLM_FAILED, model_id=identity(cell)) == [], observed + assert values(samples, TOTALS, model_id=identity(cell)) == [1.0], observed + assert values(samples, STATE, model_id=identity(cell)) == [0.0], observed + + +def anthropic_messages(gateway: Gateway, cell: str, key: str, headers: dict[str, str]) -> httpx.Response: + body: Final[dict[str, JsonValue]] = {"model": name(cell), "max_tokens": 16, "messages": messages(cell)} + return gateway.request("POST", "/v1/messages", body, key=key, headers=headers) + + +def caller_header(field: str, header: str, value: str) -> JsonValue: + if field == "provider_specific_header": + return {"custom_llm_provider": "openai", "extra_headers": {header: value}} + return {header: value} + + +@pytest.mark.parametrize( + "cell,field", + [ + ("caller-extra-headers", "extra_headers"), + ("caller-headers", "headers"), + ("caller-provider-specific-header", "provider_specific_header"), + ], +) +def test_a_provider_401_on_an_authorization_header_the_caller_sends_stays_out_of_the_deployment_metrics( + rig: Rig, cell: str, field: str +) -> None: + caller_value: Final = f"Bearer sk-{cell}-{uuid.uuid4().hex}" + rig.upstream.drain() + response: Final = rig.gateway.request( + "POST", + "/v1/chat/completions", + {"model": name(cell), "messages": messages(cell), field: caller_header(field, "Authorization", caller_value)}, + ) + assert response.status_code == 401, response.text + assert [request.headers["authorization"] for request in rig.upstream.drain()] == [caller_value] + samples: Final = until(rig.gateway, PROXY_FAILED, requested_model=name(cell)) + observed: Final = recorded(samples, model_id=identity(cell)) + assert values(samples, FAILURES, model_id=identity(cell)) == [], observed + assert values(samples, TOTALS, model_id=identity(cell)) == [], observed + assert values(samples, LLM_FAILED, model_id=identity(cell)) == [], observed + assert values(samples, STATE, model_id=identity(cell)) == [], observed + + +@pytest.mark.parametrize( + "cell,field", + [ + ("plain-extra-headers", "extra_headers"), + ("plain-headers", "headers"), + ("plain-provider-specific-header", "provider_specific_header"), + ], +) +def test_a_provider_401_on_the_deployments_own_key_stays_out_when_the_caller_sends_a_plain_header( + rig: Rig, cell: str, field: str +) -> None: + trace: Final = f"{cell}-{uuid.uuid4().hex}" + rig.upstream.drain() + response: Final = rig.gateway.request( + "POST", + "/v1/chat/completions", + {"model": name(cell), "messages": messages(cell), field: caller_header(field, "X-Caller-Trace", trace)}, + ) + assert response.status_code == 401, response.text + reached: Final = rig.upstream.drain() + assert [(request.headers["authorization"], request.headers.get("x-caller-trace")) for request in reached] == [ + (f"Bearer sk-deployment-credential-{cell}", trace) + ] + assert_kept_out_of_the_deployment_metrics(rig, cell) + + +def test_a_provider_401_on_a_provider_credential_header_the_proxy_forwards_stays_out_of_the_deployment_metrics( + rig: Rig, +) -> None: + caller_key: Final = f"AIza-forwarded-{uuid.uuid4().hex}" + rig.upstream.drain() + response: Final = rig.gateway.request( + "POST", + "/v1/chat/completions", + {"model": name("forwarded-provider-header"), "messages": messages("forwarded-provider-header")}, + headers={"x-goog-api-key": caller_key}, + ) + assert response.status_code == 401, response.text + assert [request.headers.get("x-goog-api-key") for request in rig.upstream.drain()] == [caller_key] + samples: Final = until(rig.gateway, PROXY_FAILED, requested_model=name("forwarded-provider-header")) + observed: Final = recorded(samples, model_id=identity("forwarded-provider-header")) + assert values(samples, FAILURES, model_id=identity("forwarded-provider-header")) == [], observed + assert values(samples, TOTALS, model_id=identity("forwarded-provider-header")) == [], observed + assert values(samples, LLM_FAILED, model_id=identity("forwarded-provider-header")) == [], observed + assert values(samples, STATE, model_id=identity("forwarded-provider-header")) == [], observed + + +def test_a_provider_401_on_the_deployments_own_key_stays_out_when_the_proxy_forwards_a_plain_header(rig: Rig) -> None: + trace: Final = f"forwarded-plain-header-{uuid.uuid4().hex}" + rig.upstream.drain() + response: Final = rig.gateway.request( + "POST", + "/v1/chat/completions", + {"model": name("forwarded-plain-header"), "messages": messages("forwarded-plain-header")}, + headers={"x-caller-trace": trace}, + ) + assert response.status_code == 401, response.text + assert [ + (request.headers["authorization"], request.headers.get("x-caller-trace")) for request in rig.upstream.drain() + ] == [("Bearer sk-deployment-credential-forwarded-plain-header", trace)] + assert_kept_out_of_the_deployment_metrics(rig, "forwarded-plain-header") + + +def test_a_provider_401_on_the_deployments_own_key_counts_when_the_proxy_forwards_only_its_own_user_information( + rig: Rig, +) -> None: + rig.upstream.drain() + response: Final = send(rig.gateway, "forwarded-user-information") + assert response.status_code == 401, response.text + reached: Final = rig.upstream.drain() + assert [ + (request.headers["authorization"], "x-litellm-user_api_key_hash" in request.headers) for request in reached + ] == [("Bearer sk-deployment-credential-forwarded-user-information", True)] + assert_recorded_as_a_deployment_failure(rig, "forwarded-user-information", "401") + + +def test_a_provider_401_on_a_gateway_key_the_proxy_forwards_stays_out_of_a_healthy_deployments_metrics( + rig: Rig, +) -> None: + rig.upstream.drain() + response: Final = rig.gateway.request( + "POST", + "/v1/chat/completions", + {"model": name("forwarded-gateway-key"), "messages": messages("forwarded-gateway-key")}, + headers={"x-portkey-api-key": REVOKED_GATEWAY_KEY}, + ) + assert response.status_code == 401, response.text + assert [ + (request.headers["authorization"], request.headers.get("x-portkey-api-key")) for request in rig.upstream.drain() + ] == [("Bearer sk-deployment-credential-forwarded-gateway-key", REVOKED_GATEWAY_KEY)] + assert_kept_out_of_the_deployment_metrics(rig, "forwarded-gateway-key") + + +@pytest.mark.parametrize( + "cell,body,headers", + [ + ("caller-organization", {"organization": UNKNOWN_ORGANIZATION}, {}), + ("forwarded-organization", {}, {"OpenAI-Organization": UNKNOWN_ORGANIZATION}), + ], +) +def test_a_provider_401_on_an_organization_the_caller_sends_stays_out_of_a_healthy_deployments_metrics( + rig: Rig, cell: str, body: dict[str, JsonValue], headers: dict[str, str] +) -> None: + rig.upstream.drain() + response: Final = rig.gateway.request( + "POST", "/v1/chat/completions", {"model": name(cell), "messages": messages(cell), **body}, headers=headers + ) + assert response.status_code == 401, response.text + assert [ + (request.headers["authorization"], request.headers.get("openai-organization")) + for request in rig.upstream.drain() + ] == [(f"Bearer sk-deployment-credential-{cell}", UNKNOWN_ORGANIZATION)] + assert_kept_out_of_the_deployment_metrics(rig, cell) + + +@pytest.mark.parametrize( + ("cell", "headers"), + [ + ("anthropic-own-key", {}), + ("anthropic-sdk-own-key", {"anthropic-version": "2023-06-01", "anthropic-beta": "tools-2024-04-04"}), + ], + ids=["no-client-headers", "anthropic-sdk-headers"], +) +def test_a_provider_401_on_an_anthropic_deployments_own_key_is_recorded( + rig: Rig, cell: str, headers: dict[str, str] +) -> None: + with rig.gateway.scenario() as scenario: + key: Final = scenario.key(models=[name(cell)]) + rig.upstream.drain() + response: Final = anthropic_messages(rig.gateway, cell, key, headers) + assert response.status_code == 401, response.text + assert [request.headers.get("x-api-key") for request in rig.upstream.drain()] == [ + f"sk-ant-api03-deployment-credential-{cell}" + ] + assert_recorded_as_a_deployment_failure(rig, cell, "401") + + +def test_a_provider_401_on_an_anthropic_oauth_token_the_caller_sent_stays_out_of_the_deployment_metrics( + rig: Rig, +) -> None: + with rig.gateway.scenario() as scenario: + key: Final = scenario.key(models=[name("caller-oauth")]) + rig.upstream.drain() + response: Final = anthropic_messages( + rig.gateway, + "caller-oauth", + CALLER_OAUTH_TOKEN, + {"x-litellm-api-key": key}, + ) + assert response.status_code == 401, response.text + assert [request.headers.get("authorization") for request in rig.upstream.drain()] == [ + f"Bearer {CALLER_OAUTH_TOKEN}" + ] + samples: Final = until(rig.gateway, PROXY_FAILED, requested_model=name("caller-oauth")) + assert {metric: values(samples, metric, model_id=identity("caller-oauth")) for metric in DEPLOYMENT_SERIES} == { + metric: [] for metric in DEPLOYMENT_SERIES + }, recorded(samples, requested_model=name("caller-oauth")) + + +def test_a_provider_401_on_an_azure_deployments_own_ad_token_is_recorded(rig: Rig) -> None: + rig.upstream.drain() + response: Final = send(rig.gateway, "azure-own-token") + assert response.status_code == 401, response.text + assert [request.headers.get("authorization") for request in rig.upstream.drain()] == [ + "Bearer deployment-entra-token-azure-own-token" + ] + assert_recorded_as_a_deployment_failure(rig, "azure-own-token", "401") + + +def caller_azure_request(cell: str, route: str, credential: dict[str, JsonValue]) -> tuple[str, dict[str, JsonValue]]: + if route == "embeddings": + return "/v1/embeddings", {"model": name(cell), "input": cell, **credential} + if route == "responses": + return "/v1/responses", {"model": name(cell), "input": cell, **credential} + if route == "messages": + return "/v1/messages", {"model": name(cell), "max_tokens": 16, "messages": messages(cell), **credential} + return "/v1/chat/completions", {"model": name(cell), "messages": messages(cell), **credential} + + +@pytest.mark.parametrize( + ("route", "inside_extra_body"), + CALLER_AZURE_CELLS, + ids=[f"{route}{'-extra-body' if inside_extra_body else ''}" for route, inside_extra_body in CALLER_AZURE_CELLS], +) +@pytest.mark.parametrize("opt_in", ("deployment", "proxy-wide")) +def test_a_provider_401_on_an_azure_ad_token_the_caller_sent_stays_out_of_the_deployment_metrics( + request: pytest.FixtureRequest, opt_in: str, route: str, inside_extra_body: bool +) -> None: + opted: Final[Rig] = request.getfixturevalue("rig" if opt_in == "deployment" else "opted_in") + cell: Final = caller_azure_cell(opt_in, route, inside_extra_body) + token: Final = f"caller-entra-token-{uuid.uuid4().hex}" + credential: Final[dict[str, JsonValue]] = ( + {"extra_body": {"azure_ad_token": token}} if inside_extra_body else {"azure_ad_token": token} + ) + opted.upstream.drain() + response: Final = opted.gateway.request("POST", *caller_azure_request(cell, route, credential)) + assert response.status_code == 401, response.text + assert [reached.headers.get("authorization") for reached in opted.upstream.drain()] == [ + None if inside_extra_body else f"Bearer {token}" + ] + samples: Final = until(opted.gateway, PROXY_FAILED, requested_model=name(cell)) + assert {metric: values(samples, metric, model_id=identity(cell)) for metric in DEPLOYMENT_SERIES} == { + metric: [] for metric in DEPLOYMENT_SERIES + }, recorded(samples, requested_model=name(cell)) + + +@pytest.mark.parametrize(("cell", "field"), CALLER_CREDENTIAL_CELLS, ids=[cell for cell, _ in CALLER_CREDENTIAL_CELLS]) +def test_a_provider_401_on_a_provider_credential_the_caller_sent_stays_out_of_the_deployment_metrics( + rig: Rig, cell: str, field: str +) -> None: + caller_credential: Final = f"caller-credential-{uuid.uuid4().hex}" + rig.upstream.drain() + response: Final = rig.gateway.request( + "POST", "/v1/chat/completions", {"model": name(cell), "messages": messages(cell), field: caller_credential} + ) + assert response.status_code == 401, response.text + assert [reached.headers.get("authorization") for reached in rig.upstream.drain()] == [f"Bearer {caller_credential}"] + samples: Final = until(rig.gateway, PROXY_FAILED, requested_model=name(cell)) + assert {metric: values(samples, metric, model_id=identity(cell)) for metric in DEPLOYMENT_SERIES} == { + metric: [] for metric in DEPLOYMENT_SERIES + }, recorded(samples, requested_model=name(cell)) + + +@pytest.mark.parametrize("cell", TOKEN_EXCHANGE_CELLS) +def test_a_token_exchange_deployment_rejecting_a_caller_without_an_identity_token_stays_out_of_the_deployment_metrics( + rig: Rig, cell: str +) -> None: + rig.upstream.drain() + response: Final = send(rig.gateway, cell) + assert response.status_code == 401, response.text + assert "IdP-issued access token" in response.text, response.text + assert rig.upstream.drain() == () + samples: Final = until(rig.gateway, PROXY_FAILED, requested_model=name(cell)) + assert {metric: values(samples, metric, model_id=identity(cell)) for metric in DEPLOYMENT_SERIES} == { + metric: [] for metric in DEPLOYMENT_SERIES + }, recorded(samples, requested_model=name(cell)) + + +@pytest.mark.parametrize("cell", KEYLESS_CELLS) +def test_a_missing_deployment_key_reported_before_the_provider_is_called_stays_out_of_the_deployment_metrics( + rig: Rig, cell: str +) -> None: + rig.upstream.drain() + response: Final = send(rig.gateway, cell) + assert response.status_code == 401, response.text + assert "Missing Anthropic API Key" in response.text, response.text + assert rig.upstream.drain() == () + assert_kept_out_of_the_deployment_metrics(rig, cell) + + +@pytest.mark.parametrize("opt_in", ("deployment", "proxy-wide")) +def test_a_provider_401_at_a_bedrock_endpoint_the_caller_sends_stays_out_of_the_deployment_metrics( + request: pytest.FixtureRequest, opt_in: str +) -> None: + opted: Final[Rig] = request.getfixturevalue("rig" if opt_in == "deployment" else "opted_in") + cell: Final = f"{opt_in}-caller-bedrock-endpoint" + opted.upstream.drain() + response: Final = opted.gateway.request( + "POST", + "/v1/chat/completions", + {"model": name(cell), "messages": messages(cell), "aws_bedrock_runtime_endpoint": opted.upstream.url}, + ) + assert response.status_code == 401, response.text + assert [reached.target for reached in opted.upstream.drain()] == [f"/model/{REJECTS}/converse"] + assert_kept_out_of_the_deployment_metrics(opted, cell) + + +def test_a_provider_401_at_an_api_base_the_caller_sends_counts_against_a_one_off_deployment(rig: Rig) -> None: + caller_api_base: Final = f"{rig.upstream.url}/caller-endpoint-{RUN}/v1" + rig.upstream.drain() + response: Final = rig.gateway.request( + "POST", + "/v1/chat/completions", + {"model": name("caller-api-base"), "messages": messages("caller-api-base"), "api_base": caller_api_base}, + ) + assert response.status_code == 401, response.text + assert [(reached.target, reached.headers["authorization"]) for reached in rig.upstream.drain()] == [ + (f"/caller-endpoint-{RUN}/v1/chat/completions", "Bearer sk-deployment-credential-caller-api-base") + ] + samples: Final = until(rig.gateway, FAILURES, api_base=caller_api_base) + one_off: Final = {sample.labels["model_id"] for sample in series(samples, FAILURES, api_base=caller_api_base)} + assert len(one_off) == 1 and identity("caller-api-base") not in one_off, recorded(samples, api_base=caller_api_base) + assert values(samples, FAILURES, api_base=caller_api_base, exception_status="401") == [1.0] + assert values(samples, TOTALS, api_base=caller_api_base) == [1.0] + assert values(samples, STATE, api_base=caller_api_base) == [1.0] + assert values(samples, LLM_FAILED, model_id=next(iter(one_off))) == [1.0] + assert {metric: values(samples, metric, model_id=identity("caller-api-base")) for metric in DEPLOYMENT_SERIES} == { + metric: [] for metric in DEPLOYMENT_SERIES + } + + +def _gated_key(scenario: Scenario, cell: str) -> str | None: + if cell == "invalid-key": + return f"sk-not-a-virtual-key-{uuid.uuid4().hex}" + if cell == "missing-key": + return None + if cell == "no-model-access": + return scenario.key(models=[name("healthy")]) + key: Final = scenario.key(models=[name(cell)]) + if cell == "expired-key": + write_rows( + "UPDATE \"LiteLLM_VerificationToken\" SET expires = now() - interval '1 hour' WHERE token = %s", + (sha256(key.encode()).hexdigest(),), + database_url=scenario.gateway.database_url, + ) + return key + scenario.gateway.post("/key/block", {"key": key}) + return key + + +def _gated(gateway: Gateway, cell: str, key: str | None) -> httpx.Response: + if key is None: + return gateway.client.post("/v1/chat/completions", json={"model": name(cell), "messages": messages(cell)}) + return send(gateway, cell, key=key) + + +def recorded_before(cell: str) -> list[tuple[str, str, str, float]]: + if cell in ("invalid-key", "expired-key"): + return [] + return sorted([(FAILURES, "", "", 1.0), (TOTALS, "", "", 1.0), (LLM_FAILED, identity(cell), "", 1.0)]) + + +def deployment_rows(samples: Sequence[Sample], cell: str) -> list[tuple[str, str, str, float]]: + return sorted( + (sample.name, sample.labels.get("model_id", ""), sample.labels.get("api_base", ""), sample.value) + for sample in samples + if sample.name in DEPLOYMENT_SERIES + and name(cell) in (sample.labels.get("requested_model"), sample.labels.get("model")) + ) + + +@pytest.mark.parametrize("cell", GATED_CELLS) +def test_a_request_the_proxy_rejects_itself_never_marks_the_deployment_down_and_is_recorded_as_before( + rig: Rig, cell: str +) -> None: + with rig.gateway.scenario() as scenario: + key: Final = _gated_key(scenario, cell) + rig.upstream.drain() + response: Final = _gated(rig.gateway, cell, key) + assert response.status_code == (403 if cell == "no-model-access" else 401), response.text + assert rig.upstream.drain() == () + samples: Final = until(rig.gateway, PROXY_FAILED, requested_model=name(cell)) + assert deployment_rows(samples, cell) == recorded_before(cell), recorded(samples, requested_model=name(cell)) + + +def test_a_key_rate_limit_rejected_at_the_proxy_is_still_counted_as_before(rig: Rig) -> None: + with rig.gateway.scenario() as scenario: + key: Final = scenario.key(models=[name("key-rpm-limit")], rpm_limit=1) + rig.upstream.drain() + admitted: Final = send(rig.gateway, "key-rpm-limit", key=key) + assert admitted.status_code == 200, admitted.text + limited: Final = send(rig.gateway, "key-rpm-limit", key=key) + assert limited.status_code == 429, limited.text + assert len(rig.upstream.drain()) == 1 + samples: Final = until(rig.gateway, FAILURES, requested_model=name("key-rpm-limit"), exception_status="429") + assert values(samples, FAILURES, requested_model=name("key-rpm-limit"), exception_status="429") == [1.0] + eventually( + lambda: scrape(rig.gateway), + lambda observed: ( + values(observed, STATE, model_id=identity("key-rpm-limit"), api_base=rig.api_base) == [0.0] + ), + ) + + +@pytest.fixture(scope="module") +def mcp_peer() -> Iterator[McpPeer]: + with scripted_peer(echo_tool("add")) as peer: + yield peer + + +def test_an_mcp_tool_call_the_proxy_rejects_for_the_caller_stays_out_of_the_llm_failure_metrics( + rig: Rig, mcp_peer: McpPeer +) -> None: + alias: Final = "a365" + RUN + with rig.gateway.scenario() as scenario: + server: Final = register_mcp(scenario, mcp_peer, alias) + key: Final = scenario.key(object_permission={"mcp_servers": [server]}) + hashed: Final = sha256(key.encode()).hexdigest() + outcomes: Final = { + entry: McpCaller(rig.gateway, key, entry, alias).call(f"{alias}-add", {"entry": entry}, server) + for entry in ENTRY_POINTS + } + assert all( + outcome.error is not None and "Agent 365 guardrail rejected" in outcome.raw for outcome in outcomes.values() + ), {entry: outcome.raw for entry, outcome in outcomes.items()} + eventually( + lambda: rig.gateway.rows(GUARDRAIL_ROWS, (AGENT_365_GUARDRAIL, hashed, "call_mcp_tool")), + lambda rows: [row["status"] for row in rows] == ["guardrail_intervened"] * len(ENTRY_POINTS), + seconds=70, + ) + samples: Final = scrape(rig.gateway) + assert values(samples, LLM_FAILED, hashed_api_key=hashed) == [] + assert values(samples, FAILURES, hashed_api_key=hashed) == [] + assert values(samples, TOTALS, hashed_api_key=hashed) == [] diff --git a/tests/integration/observability/test_prometheus_provider_auth_failure_chaos.py b/tests/integration/observability/test_prometheus_provider_auth_failure_chaos.py new file mode 100644 index 00000000000..a1fe1cf4f33 --- /dev/null +++ b/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 diff --git a/tests/integration/sdk/test_prometheus_copilot_session_auth_failure_sdk.py b/tests/integration/sdk/test_prometheus_copilot_session_auth_failure_sdk.py new file mode 100644 index 00000000000..1b5e990fe41 --- /dev/null +++ b/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,) diff --git a/tests/unit/integrations/test_prometheus_invalid_key_filtering.py b/tests/unit/integrations/test_prometheus_invalid_key_filtering.py index 9e8348e860a..453fd28636d 100644 --- a/tests/unit/integrations/test_prometheus_invalid_key_filtering.py +++ b/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,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] diff --git a/tests/unit/proxy/test_litellm_pre_call_utils.py b/tests/unit/proxy/test_litellm_pre_call_utils.py index ed7817bd660..e03fd9850c1 100644 --- a/tests/unit/proxy/test_litellm_pre_call_utils.py +++ b/tests/unit/proxy/test_litellm_pre_call_utils.py @@ -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