Merge 978fecc563 into 836f326611

litellm/proxy/_experimental/mcp_server/auth/user_api_key_auth_mcp.py

@@ -344,31 +344,31 @@ def gateway_dcr_challenge_scope(
     return " ".join(token for token in tokens if token is not None)
 
 
-def _is_gateway_dcr_challenge_scope(
-    route: str,
+def _cold_start_fallback(
+    request: Request,
+    request_route: str,
     mcp_servers: list[str] | None,
     mcp_auth_header: str | None,
     mcp_server_auth_headers: dict[str, dict[str, str]] | None,
-    exc: Exception,
-    client_ip: str | None,
-) -> bool:
-    """True when an unauthenticated MCP request should receive the RFC 9728 401
-    challenge that advertises the gateway as the authorization server.
-
-    Fires only for a genuine 401 with no client-supplied MCP auth headers (those mean
-    the caller is not a cold-start DCR client), on the scopes the gateway's keyless
-    flow serves: the aggregate ``/mcp`` endpoint, an ``x-mcp-servers``-scoped request
-    (the resource the client configured is still ``/mcp``; a single eligible header
-    target is also carried in the OAuth scope), or a per-server path whose single target
-    advertises gateway-owned sign-in. Every other named target keeps its existing
-    behavior, failing closed to the original admission error."""
-    if not _is_litellm_auth_admission_error(exc):
-        return False
+    bearer_presented: bool,
+) -> Literal["defer_to_route_challenge", "gateway_challenge"] | None:
+    """Classify the structural cold-start response without inspecting an error."""
     if _has_client_supplied_mcp_auth(mcp_auth_header, mcp_server_auth_headers):
-        return False
-    if len(MCPRequestHandler.extract_target_server_names_from_path(route)) == 0:
-        return True
-    return _gateway_dcr_challenge_target(route, mcp_servers, client_ip) is not None
+        return None
+
+    mcp_servers_from_path: Final = _parse_mcp_server_names_from_path(request_route, mcp_servers)
+    client_ip: Final = IPAddressUtils.get_mcp_client_ip(request)
+    if mcp_servers_from_path is not None and (
+        _is_mcp_passthrough_cold_start(mcp_servers_from_path, client_ip=client_ip)
+        or (not bearer_presented and _is_legacy_delegate_cold_start(mcp_servers_from_path, client_ip=client_ip))
+    ):
+        return "defer_to_route_challenge"
+    if (
+        len(MCPRequestHandler.extract_target_server_names_from_path(request_route)) == 0
+        or _gateway_dcr_challenge_target(request_route, mcp_servers, client_ip) is not None
+    ):
+        return "gateway_challenge"
+    return None
 
 
 def _gateway_dcr_challenge(
@@ -416,48 +416,45 @@ def _admission_failure_fallback(
     exc: Exception,
     bearer_presented: bool,
 ) -> UserAPIKeyAuth:
-    """Map a failed LiteLLM admission to its anonymous fallback or challenge.
+    """Apply the structural cold-start response to an admission error."""
+    if not _is_litellm_auth_admission_error(exc):
+        raise exc
 
-    Two fallbacks exist, both gated on a genuine 401 with no client-supplied
-    MCP auth headers. The pass-through cold start (RFC 9728 / MCP
-    Authorization spec discovery return) admits anonymously so the route's
-    401 emitter can produce the per-server challenge. The aggregate
-    gateway-DCR scope converts the failure into the gateway's own
-    resource_metadata challenge, with the RFC 6750 ``invalid_token`` error
-    code when the caller DID present a bearer (an expired gateway session
-    must re-authorize, not retry a dead token). Anything else re-raises the
-    original admission error unchanged."""
-    mcp_servers_from_path: Final = _parse_mcp_server_names_from_path(request_route, mcp_servers)
+    match _cold_start_fallback(
+        request,
+        request_route,
+        mcp_servers,
+        mcp_auth_header,
+        mcp_server_auth_headers,
+        bearer_presented,
+    ):
+        case "defer_to_route_challenge":
+            verbose_logger.debug("MCP pass-through cold start: deferring admission to route 401 emitter")
+            return UserAPIKeyAuth()
+        case "gateway_challenge":
+            raise _gateway_dcr_challenge(
+                request,
+                request_route,
+                mcp_servers,
+                invalid_token=bearer_presented,
+            ) from exc
+        case None:
+            raise exc
+
+
+def _has_litellm_credential_header(headers: Headers) -> bool:
     if (
-        mcp_servers_from_path is not None
-        and not _has_client_supplied_mcp_auth(mcp_auth_header, mcp_server_auth_headers)
-        and _is_litellm_auth_admission_error(exc)
-        and (
-            _is_mcp_passthrough_cold_start(
-                mcp_servers_from_path,
-                client_ip=IPAddressUtils.get_mcp_client_ip(request),
-            )
-            or (
-                not bearer_presented
-                and _is_legacy_delegate_cold_start(
-                    mcp_servers_from_path,
-                    client_ip=IPAddressUtils.get_mcp_client_ip(request),
-                )
-            )
-        )
+        MCPRequestHandler.LITELLM_API_KEY_HEADER_NAME_PRIMARY in headers
+        or MCPRequestHandler.LITELLM_API_KEY_HEADER_NAME_SECONDARY in headers
     ):
-        verbose_logger.debug("MCP pass-through cold start: deferring admission to route 401 emitter")
-        return UserAPIKeyAuth()
-    if _is_gateway_dcr_challenge_scope(
-        route=request_route,
-        mcp_servers=mcp_servers,
-        mcp_auth_header=mcp_auth_header,
-        mcp_server_auth_headers=mcp_server_auth_headers,
-        exc=exc,
-        client_ip=IPAddressUtils.get_mcp_client_ip(request),
-    ):
-        raise _gateway_dcr_challenge(request, request_route, mcp_servers, invalid_token=bearer_presented) from exc
-    raise exc
+        return True
+
+    from litellm.proxy.proxy_server import general_settings  # pyright: ignore[reportUnknownVariableType]  # legacy map
+
+    custom_header_name: Final = OPTIONAL_STRING_ADAPTER.validate_python(
+        general_settings.get("litellm_key_header_name")  # pyright: ignore[reportUnknownMemberType]  # legacy settings
+    )
+    return custom_header_name is not None and custom_header_name in headers
 
 
 @dataclass(frozen=True, slots=True)
@@ -692,6 +689,21 @@ class MCPRequestHandler:
                         bearer_presented=True,
                     )
             else:
+                from litellm.proxy.auth.auth_utils import mark_expected_credential_challenge
+
+                if not _has_litellm_credential_header(headers) and (
+                    _cold_start_fallback(
+                        request=request,
+                        request_route=request_route,
+                        mcp_servers=mcp_servers,
+                        mcp_auth_header=mcp_auth_header,
+                        mcp_server_auth_headers=mcp_server_auth_headers,
+                        bearer_presented=False,
+                    )
+                    is not None
+                ):
+                    mark_expected_credential_challenge(request)
+
                 try:
                     validated_user_api_key_auth = await user_api_key_auth(api_key=litellm_api_key, request=request)
                 except (HTTPException, ProxyException) as exc:

litellm/proxy/auth/auth_exception_handler.py

@@ -23,6 +23,7 @@ from litellm.proxy._types import (
 from litellm.proxy.auth.auth_utils import (  # noqa: F401  # legacy module exports
     _get_request_ip_address,  # pyright: ignore[reportPrivateUsage,reportUnusedImport]  # backwards-compatible package export
     get_request_ip_address,
+    is_expected_credential_challenge,
     is_invalid_virtual_key_error,
     mark_invalid_virtual_key_error,
     normalize_request_route,
@@ -203,14 +204,27 @@ class UserAPIKeyAuthExceptionHandler:
             log_extra: Final = {"requester_ip": requester_ip}
             is_invalid_virtual_key: Final = is_invalid_virtual_key_error(e)
             is_quiet_log: Final = is_invalid_virtual_key and not litellm.log_client_error_tracebacks
-            logger: Final = verbose_proxy_stdout_logger if is_quiet_log else verbose_proxy_logger
+            is_expected_challenge: Final = is_expected_credential_challenge(e, request)
+            log_level: Final = (
+                logging.DEBUG if is_expected_challenge else logging.WARNING if is_quiet_log else logging.ERROR
+            )
+            logger: Final = (
+                verbose_proxy_stdout_logger if is_quiet_log and not is_expected_challenge else verbose_proxy_logger
+            )
+            exc_info: Final = (
+                None
+                if is_expected_challenge
+                else True
+                if litellm.log_client_error_tracebacks or not is_expected_client_error(e)
+                else None
+            )
             logger.log(
-                logging.WARNING if is_quiet_log else logging.ERROR,
+                log_level,
                 "litellm.proxy.proxy_server.user_api_key_auth(): Exception occured - %s\nRequester IP Address:%s%s",
                 _auth_failure_log_message(e),
                 requester_ip,
                 _identity_log_suffix(resolved_identity),
-                exc_info=True if litellm.log_client_error_tracebacks or not is_expected_client_error(e) else None,
+                exc_info=exc_info,
                 extra=log_extra,
             )
 
@@ -260,8 +274,10 @@ class UserAPIKeyAuthExceptionHandler:
                 e = transformed_exception
 
             final_exception: Final = mark_invalid_virtual_key_error(_as_proxy_exception(e), is_invalid_virtual_key)
-            # If a quiet-logged malformed-key transform yields non-401, escalate to ERROR
-            if is_quiet_log and str(final_exception.code) != str(status.HTTP_401_UNAUTHORIZED):
+            # Escalate callback-transformed quiet logs and expected challenges with non-401 codes
+            if (is_quiet_log or is_expected_challenge) and str(final_exception.code) != str(
+                status.HTTP_401_UNAUTHORIZED
+            ):
                 verbose_proxy_logger.error(
                     "litellm.proxy.proxy_server.user_api_key_auth(): Exception occured - %s\nRequester IP Address:%s",
                     final_exception,

litellm/proxy/auth/auth_utils.py

@@ -41,6 +41,23 @@ from litellm.types.router import reject_server_owned_wif_params as _reject_serve
 from litellm.types.utils import CustomPricingLiteLLMParams, oauth_token_exchange_litellm_params
 
 
+class MissingCredentialError(Exception):
+    pass
+
+
+EXPECTED_CREDENTIAL_CHALLENGE_STATE_KEY: Final = "litellm_expected_credential_challenge"
+
+
+def mark_expected_credential_challenge(request: Request) -> None:
+    setattr(request.state, EXPECTED_CREDENTIAL_CHALLENGE_STATE_KEY, True)
+
+
+def is_expected_credential_challenge(exception: BaseException, request: Request) -> bool:
+    return isinstance(exception, MissingCredentialError) and (
+        getattr(request.state, EXPECTED_CREDENTIAL_CHALLENGE_STATE_KEY, False) is True
+    )
+
+
 def is_invalid_virtual_key_error(exception: BaseException | None) -> bool:
     """True when an authentication error rejects a malformed virtual key.
 

litellm/proxy/auth/user_api_key_auth.py

@@ -101,6 +101,7 @@ from litellm.proxy.auth.auth_object_prefetch import (
     prefetch_identity_keys,
 )
 from litellm.proxy.auth.auth_utils import (
+    MissingCredentialError,
     abbreviate_api_key,
     fallback_target_model_name,
     get_end_user_id_from_request_body,
@@ -2494,7 +2495,7 @@ async def user_api_key_auth_builder(
             raise Exception("No api key passed in.")
         elif api_key == "":
             # missing 'Bearer ' prefix
-            raise Exception("Malformed API Key passed in. Ensure Key has `Bearer ` prefix.")
+            raise MissingCredentialError("Malformed API Key passed in. Ensure Key has `Bearer ` prefix.")
 
         if route == "/user/auth":
             if general_settings.get("allow_user_auth", False) is True:

tests/integration/mcp/test_mcp_oauth_flows.py

@@ -11,7 +11,7 @@ from collections.abc import Iterator, Mapping
 from contextlib import contextmanager, nullcontext
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Final
+from typing import Final, Literal
 from urllib.parse import parse_qs, urljoin, urlsplit
 
 import httpx
@@ -39,7 +39,7 @@ from integration._support.mcp import (
 )
 from integration._support.mcp_grants import create_toolset
 from integration._support.oauth_server import AuthorizationServer, oauth_server
-from integration._support.process import owned_proxy
+from integration._support.process import owned_proxy, owned_proxy_process
 from integration._support.wire import Reply, Request, Wire, wire_server
 from pydantic import TypeAdapter
 
@@ -697,6 +697,109 @@ def test_per_user_authorization_code_with_pkce_binds_the_token_to_the_authorizin
         )
 
 
+@pytest.mark.parametrize(
+    "path_template,server_auth_type,auth_headers,expected_status,expect_www_authenticate,"
+    "expected_challenge_fragment,expected_error_lines",
+    [
+        pytest.param(
+            "/{alias}/mcp",
+            "oauth2",
+            (),
+            401,
+            True,
+            "resource_metadata=",
+            0,
+            id="per-server-oauth-no-credentials",
+        ),
+        pytest.param("/mcp", "oauth2", (), 401, True, None, 0, id="aggregate-oauth-no-credentials"),
+        pytest.param(
+            "/{alias}/mcp",
+            "oauth2",
+            (("Authorization", ""),),
+            401,
+            True,
+            None,
+            1,
+            id="oauth-empty-authorization",
+        ),
+        pytest.param(
+            "/{alias}/mcp",
+            "api_key",
+            (),
+            401,
+            False,
+            None,
+            1,
+            id="non-oauth-api-key",
+        ),
+    ],
+)
+def test_mcp_cold_start_malformed_key_log_level(
+    gateway: Gateway,
+    tmp_path: Path,
+    path_template: str,
+    server_auth_type: Literal["oauth2", "api_key"],
+    auth_headers: tuple[tuple[str, str], ...],
+    expected_status: int,
+    expect_www_authenticate: bool,
+    expected_challenge_fragment: str | None,
+    expected_error_lines: int,
+) -> None:
+    with mcp_peer() as peer, oauth_server() as auth, owned_proxy_process(gateway, tmp_path, {}) as owned:
+        with owned.gateway.scenario() as scenario:
+            alias: Final = "cold" + uuid.uuid4().hex[:8]
+            if server_auth_type == "oauth2":
+                _register_oauth(
+                    scenario,
+                    peer,
+                    auth,
+                    alias,
+                    auth_type="oauth2",
+                    oauth2_flow="authorization_code",
+                    credentials={"client_id": "cold-client", "client_secret": "cold-secret"},
+                )
+            else:
+                register_mcp(
+                    scenario,
+                    peer,
+                    alias,
+                    auth_type="api_key",
+                    credentials={"auth_value": "static-" + uuid.uuid4().hex},
+                    extra_headers=["x-api-key"],
+                )
+            baseline_log: Final = owned.log.read_text()
+            request_path: Final = path_template.format(alias=alias)
+            request_headers: Final = [*ACCEPT.items(), *auth_headers]
+            request: Final = owned.gateway.client.build_request(
+                "POST",
+                request_path,
+                headers=request_headers,
+                json=INITIALIZE,
+            )
+            expected_authorization: Final = tuple(
+                value for name, value in auth_headers if name.lower() == "authorization"
+            )
+            assert tuple(request.headers.get_list("authorization")) == expected_authorization
+            response: Final = owned.gateway.client.send(request)
+
+            assert response.status_code == expected_status, response.text
+            challenge_header: Final = response.headers.get("www-authenticate")
+            assert (challenge_header is not None) is expect_www_authenticate
+            assert expected_challenge_fragment is None or (
+                challenge_header is not None and expected_challenge_fragment in challenge_header
+            )
+            request_log: Final = eventually(
+                owned.log.read_text,
+                lambda value: f"POST {request_path}" in value[len(baseline_log) :],
+            )
+            error_lines: Final = tuple(
+                line
+                for line in request_log[len(baseline_log) :].splitlines()
+                if "Malformed API Key" in line and "ERROR" in line
+            )
+            assert len(error_lines) == expected_error_lines, error_lines
+
+
 def test_authorization_request_without_pkce_is_refused_before_reaching_the_authorization_server(
     gateway: Gateway,
 ) -> None:

tests/unit/proxy/_experimental/mcp_server/auth/test_user_api_key_auth_mcp.py

@@ -1,6 +1,7 @@
 import contextlib
 import json
 import os
+import uuid
 from collections.abc import Sequence
 from collections.abc import Set as AbstractSet
 from datetime import datetime, timedelta, timezone
@@ -12,6 +13,7 @@ import pytest
 from fastapi import HTTPException
 from fastapi.testclient import TestClient
 from starlette.datastructures import Headers
+from starlette.requests import Request
 
 from litellm.caching.caching import DualCache
 from litellm.constants import MCP_ALL_TOOLS_WILDCARD
@@ -23,6 +25,7 @@ from litellm.proxy._experimental.mcp_server.auth.user_api_key_auth_mcp import (
     is_mcp_admitted_user_subject,
 )
 from litellm.proxy._types import (
+    LiteLLM_MCPServerTable,
     LiteLLM_ObjectPermissionTable,
     LiteLLM_TeamTable,
     LiteLLM_UserTable,
@@ -31,9 +34,166 @@ from litellm.proxy._types import (
     SpecialMCPServerNames,
     UserAPIKeyAuth,
 )
+from litellm.types.mcp import MCPAuth
+from litellm.types.mcp_server.mcp_server_manager import MCPServer
 from litellm.types.agents import AgentCaller
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "path,auth_type,delegate_auth_to_upstream,pass_through,bearer_presented,mcp_auth_header,"
+    "mcp_server_auth_headers,mcp_servers,expected",
+    [
+        pytest.param(
+            "/mcp/legacy",
+            MCPAuth.oauth2,
+            True,
+            False,
+            False,
+            None,
+            None,
+            None,
+            "defer_to_route_challenge",
+            id="legacy-delegate-without-bearer",
+        ),
+        pytest.param(
+            "/mcp/legacy",
+            MCPAuth.oauth2,
+            True,
+            False,
+            True,
+            None,
+            None,
+            None,
+            None,
+            id="legacy-delegate-with-bearer",
+        ),
+        pytest.param(
+            "/mcp/passthrough",
+            None,
+            False,
+            True,
+            True,
+            None,
+            None,
+            None,
+            "defer_to_route_challenge",
+            id="pass-through-with-bearer",
+        ),
+        pytest.param(
+            "/mcp/oauth",
+            MCPAuth.oauth2,
+            False,
+            False,
+            False,
+            None,
+            None,
+            None,
+            "gateway_challenge",
+            id="gateway-dcr-per-server",
+        ),
+        pytest.param(
+            "/mcp",
+            None,
+            False,
+            False,
+            False,
+            None,
+            None,
+            None,
+            "gateway_challenge",
+            id="aggregate-mcp",
+        ),
+        pytest.param(
+            "/mcp",
+            None,
+            False,
+            False,
+            False,
+            None,
+            None,
+            ["oauth"],
+            "gateway_challenge",
+            id="aggregate-mcp-scoped-by-header",
+        ),
+        pytest.param(
+            "/mcp/oauth",
+            MCPAuth.oauth2,
+            False,
+            False,
+            False,
+            "Bearer upstream",
+            None,
+            None,
+            None,
+            id="legacy-mcp-auth-header",
+        ),
+        pytest.param(
+            "/mcp/oauth",
+            MCPAuth.oauth2,
+            False,
+            False,
+            False,
+            None,
+            {"oauth": {"Authorization": "Bearer upstream"}},
+            None,
+            None,
+            id="server-specific-mcp-auth-header",
+        ),
+    ],
+)
+async def test_cold_start_fallback_classifies_structural_challenges(
+    path: str,
+    auth_type: MCPAuth | None,
+    delegate_auth_to_upstream: bool,
+    pass_through: bool,
+    bearer_presented: bool,
+    mcp_auth_header: str | None,
+    mcp_server_auth_headers: dict[str, dict[str, str]] | None,
+    mcp_servers: list[str] | None,
+    expected: Literal["defer_to_route_challenge", "gateway_challenge"] | None,
+) -> None:
+    from litellm.proxy._experimental.mcp_server.auth.user_api_key_auth_mcp import _cold_start_fallback
+    from litellm.proxy._experimental.mcp_server.mcp_server_manager import global_mcp_server_manager
+
+    server_alias: Final = f"cold-start-{uuid.uuid4().hex}"
+    request_path: Final = f"/mcp/{server_alias}" if path != "/mcp" else path
+    request_mcp_servers: Final[list[str] | None] = [server_alias] if mcp_servers is not None else None
+    server_auth_type: Final = auth_type or (MCPAuth.oauth2 if mcp_servers is not None else MCPAuth.none)
+    server_row: Final = LiteLLM_MCPServerTable(
+        server_id=f"cold-start-{uuid.uuid4().hex}",
+        server_name=server_alias,
+        alias=server_alias,
+        url=None,
+        transport="http",
+        auth_type=server_auth_type,
+        oauth2_flow="authorization_code" if server_auth_type == MCPAuth.oauth2 else None,
+        delegate_auth_to_upstream=delegate_auth_to_upstream,
+        dcr_bridge=server_auth_type == MCPAuth.oauth2 and not delegate_auth_to_upstream,
+        oauth_passthrough=pass_through,
+        extra_headers=["Authorization"] if pass_through else [],
+        created_at=datetime(2024, 1, 1, tzinfo=timezone.utc),
+        updated_at=datetime(2024, 1, 1, tzinfo=timezone.utc),
+    )
+    request: Final = Request({"type": "http", "method": "POST", "path": request_path, "headers": []})
+    try:
+        await global_mcp_server_manager.add_server(server_row)
+        registered_server: Final[MCPServer | None] = global_mcp_server_manager.get_mcp_server_by_name(server_alias)
+        assert isinstance(registered_server, MCPServer)
+        assert registered_server.is_oauth_passthrough is pass_through
+        result: Final = _cold_start_fallback(
+            request=request,
+            request_route=request_path,
+            mcp_servers=request_mcp_servers,
+            mcp_auth_header=mcp_auth_header,
+            mcp_server_auth_headers=mcp_server_auth_headers,
+            bearer_presented=bearer_presented,
+        )
+        assert result == expected
+    finally:
+        global_mcp_server_manager.remove_server(server_row)
+
+
 @pytest.mark.parametrize(
     "headers,expected",
     [

tests/unit/proxy/auth/test_auth_exception_handler.py

@@ -1,5 +1,8 @@
 import asyncio
 import json
+import logging
+from collections.abc import Callable
+from typing import Final
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import httpx
@@ -29,6 +32,7 @@ import litellm
 from litellm._logging import verbose_proxy_logger
 from litellm.constants import INVALID_VIRTUAL_KEY_ERROR_MARKER
 from litellm.exceptions import BudgetExceededError
+from litellm.integrations.custom_logger import CustomLogger
 from litellm.proxy._types import (
     ModelAccessDeniedProxyException,
     ProxyErrorTypes,
@@ -1210,6 +1214,163 @@ async def test_handle_authentication_error_keeps_internal_message_on_model_acces
     assert [r for r in caplog.records if r.levelname == "WARNING" and "internal-models" in r.getMessage()] == []
 
 
+class _AuthenticationFailureCallback(CustomLogger):
+    def __init__(self, exception: HTTPException | None) -> None:
+        self._exception: Final[HTTPException | None] = exception
+
+    async def async_post_call_failure_hook(
+        self,
+        request_data: dict[str, object],
+        original_exception: Exception,
+        user_api_key_dict: UserAPIKeyAuth,
+        traceback_str: str | None = None,
+    ) -> HTTPException | None:
+        return self._exception
+
+
+@pytest.fixture
+def register_authentication_failure_callback(
+    monkeypatch: pytest.MonkeyPatch,
+) -> Callable[[HTTPException | None], _AuthenticationFailureCallback]:
+    def register(exception: HTTPException | None) -> _AuthenticationFailureCallback:
+        callback: Final = _AuthenticationFailureCallback(exception)
+        monkeypatch.setattr(litellm, "callbacks", [callback])
+        return callback
+
+    return register
+
+
+async def _capture_authentication_exception(
+    exception: BaseException,
+    request: Request,
+    log_client_error_tracebacks: bool,
+) -> ProxyException:
+    handler: Final = UserAPIKeyAuthExceptionHandler()
+    with (
+        patch.object(litellm, "log_client_error_tracebacks", log_client_error_tracebacks),
+        patch("litellm.proxy.proxy_server.general_settings", {"allow_requests_on_db_unavailable": False}),
+    ):
+        try:
+            await handler.handle_authentication_error(exception, request, {}, "/mcp", None, "")
+        except ProxyException as raised:
+            return raised
+    pytest.fail("expected the authentication handler to raise a ProxyException")
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("tracebacks_enabled", [True, False])
+async def test_expected_missing_credential_logs_debug_without_traceback_and_preserves_response(
+    tracebacks_enabled: bool,
+    caplog: pytest.LogCaptureFixture,
+    monkeypatch: pytest.MonkeyPatch,
+    register_authentication_failure_callback: Callable[
+        [HTTPException | None], _AuthenticationFailureCallback
+    ],
+) -> None:
+    from litellm.proxy.auth.auth_utils import MissingCredentialError, mark_expected_credential_challenge
+
+    register_authentication_failure_callback(None)
+    flagged_request: Final = Request({"type": "http", "method": "POST", "path": "/mcp", "headers": []})
+    mark_expected_credential_challenge(flagged_request)
+    unflagged_request: Final = Request({"type": "http", "method": "POST", "path": "/mcp", "headers": []})
+    message: Final = "Malformed API Key passed in. Ensure Key has `Bearer ` prefix."
+    monkeypatch.setattr(verbose_proxy_logger, "propagate", True)
+    with caplog.at_level(logging.DEBUG, logger=verbose_proxy_logger.name):
+        flagged_response: Final = await _capture_authentication_exception(
+            MissingCredentialError(message),
+            flagged_request,
+            tracebacks_enabled,
+        )
+    flagged_records: Final = tuple(
+        record for record in caplog.records if "user_api_key_auth(): Exception occured" in record.getMessage()
+    )
+    assert tuple(record.levelname for record in flagged_records) == ("DEBUG",)
+    assert flagged_records[0].exc_info is None
+
+    caplog.clear()
+    with caplog.at_level(logging.DEBUG, logger=verbose_proxy_logger.name):
+        unflagged_response: Final = await _capture_authentication_exception(
+            MissingCredentialError(message),
+            unflagged_request,
+            tracebacks_enabled,
+        )
+    unflagged_records: Final = tuple(
+        record for record in caplog.records if "user_api_key_auth(): Exception occured" in record.getMessage()
+    )
+    assert tuple(record.levelname for record in unflagged_records) == ("ERROR",)
+    assert unflagged_records[0].exc_info is not None
+    assert vars(flagged_response) == vars(unflagged_response)
+    assert flagged_response.to_dict() == unflagged_response.to_dict()
+
+
+@pytest.mark.asyncio
+async def test_expected_missing_credential_callback_403_is_escalated_and_unchanged(
+    caplog: pytest.LogCaptureFixture,
+    monkeypatch: pytest.MonkeyPatch,
+    register_authentication_failure_callback: Callable[
+        [HTTPException | None], _AuthenticationFailureCallback
+    ],
+) -> None:
+    from litellm.proxy.auth.auth_utils import MissingCredentialError, mark_expected_credential_challenge
+
+    callback_exception: Final = HTTPException(status_code=403, detail="Forbidden")
+    register_authentication_failure_callback(callback_exception)
+    flagged_request: Final = Request({"type": "http", "method": "POST", "path": "/mcp", "headers": []})
+    mark_expected_credential_challenge(flagged_request)
+    unflagged_request: Final = Request({"type": "http", "method": "POST", "path": "/mcp", "headers": []})
+    message: Final = "Malformed API Key passed in. Ensure Key has `Bearer ` prefix."
+    monkeypatch.setattr(verbose_proxy_logger, "propagate", True)
+    with caplog.at_level(logging.DEBUG, logger=verbose_proxy_logger.name):
+        flagged_response: Final = await _capture_authentication_exception(
+            MissingCredentialError(message),
+            flagged_request,
+            False,
+        )
+    flagged_records: Final = tuple(
+        record for record in caplog.records if "user_api_key_auth(): Exception occured" in record.getMessage()
+    )
+    assert tuple(record.levelname for record in flagged_records) == ("DEBUG", "ERROR")
+    assert flagged_response.code == str(status.HTTP_403_FORBIDDEN)
+
+    caplog.clear()
+    with caplog.at_level(logging.DEBUG, logger=verbose_proxy_logger.name):
+        unflagged_response: Final = await _capture_authentication_exception(
+            MissingCredentialError(message),
+            unflagged_request,
+            False,
+        )
+    assert vars(flagged_response) == vars(unflagged_response)
+    assert flagged_response.to_dict() == unflagged_response.to_dict()
+
+
+@pytest.mark.asyncio
+async def test_expected_credential_marker_does_not_quiet_policy_rejection(
+    caplog: pytest.LogCaptureFixture,
+    monkeypatch: pytest.MonkeyPatch,
+    register_authentication_failure_callback: Callable[
+        [HTTPException | None], _AuthenticationFailureCallback
+    ],
+) -> None:
+    from litellm.proxy.auth.auth_utils import mark_expected_credential_challenge
+
+    register_authentication_failure_callback(None)
+    request: Final = Request({"type": "http", "method": "POST", "path": "/mcp", "headers": []})
+    mark_expected_credential_challenge(request)
+    monkeypatch.setattr(verbose_proxy_logger, "propagate", True)
+    with caplog.at_level(logging.DEBUG, logger=verbose_proxy_logger.name):
+        response: Final = await _capture_authentication_exception(
+            HTTPException(status_code=403, detail="Policy rejected request"),
+            request,
+            False,
+        )
+
+    records: Final = tuple(
+        record for record in caplog.records if "user_api_key_auth(): Exception occured" in record.getMessage()
+    )
+    assert tuple(record.levelname for record in records) == ("ERROR",)
+    assert response.code == str(status.HTTP_403_FORBIDDEN)
+
+
 def test_as_proxy_exception_keeps_jwt_scope_denial_message_shape():
     detail = {"error": _DENIED_CLIENT_MESSAGE}
     denial = ModelAccessDeniedHTTPException(