test: pin credential route gating, resolver-failure resilience, and OtelDestination immutability

Adds regression coverage the gauntlet had only proven ad-hoc:
- self_managed_routes contains exactly the /credentials list + single-name routes
  (no wildcard), and the ^/credentials/[^/]+$ pattern matches a single-segment
  name but NOT /credentials/by_name/... or /credentials/by_model/...
- a plain internal user (the role a team-admin/org-admin key carries) is denied
  by_name/by_model, while PROXY_ADMIN_VIEW_ONLY reaches them via the read-parity
  safe-GET default-allow (the documented masking inconsistency, not a leak)
- _hoist_request_destinations swallows a resolver failure and leaves the
  destinations ContextVar empty, so telemetry setup can never break auth
- OtelDestination is frozen (a request cannot rewrite where its traces go) and
  header_string renders the k=v form
This commit is contained in:
Yucheng Zhu 2026-07-27 13:19:23 -07:00
parent e05a7a3655
commit 5d907fc29d
3 changed files with 135 additions and 1 deletions

View file

@ -530,3 +530,18 @@ def test_generic_backend_resolves_generic_destination():
got = OpenTelemetryV2._destinations_for_backend(_Shim(), event)
assert [d.endpoint for d in got] == ["http://collector:4318"]
def test_otel_destination_is_frozen_and_renders_header_string():
"""OtelDestination is the immutable value the resolver hands to the runtime;
mutating one after resolution must fail (a request can never rewrite where its
traces go), and header_string renders the k=v,k2=v2 form the exporter expects."""
import pytest
from pydantic import ValidationError
dest = OtelDestination(
endpoint="https://c/v1", headers={"Authorization": "Bearer x", "x-k": "y"}
)
with pytest.raises(ValidationError):
dest.endpoint = "https://evil/v1"
assert dest.header_string() == "Authorization=Bearer x,x-k=y"

View file

@ -9,7 +9,12 @@ sys.path.insert(
import pytest
from fastapi import HTTPException, Request
from litellm.proxy._types import LiteLLM_UserTable, LitellmUserRoles, UserAPIKeyAuth
from litellm.proxy._types import (
LiteLLM_UserTable,
LiteLLMRoutes,
LitellmUserRoles,
UserAPIKeyAuth,
)
from litellm.proxy.auth.route_checks import RouteChecks
@ -3084,3 +3089,91 @@ def test_internal_user_blocked_from_search_tool_writes(route):
assert "Only proxy admin" in str(exc_info.value)
assert f"Route={route}" in str(exc_info.value)
assert "Your role=internal_user" in str(exc_info.value)
# --- Credential route gating (PR #30873: /credentials opened to self-managed) --- #
def test_self_managed_routes_includes_credentials_entries():
"""The PR adds exactly the credential list + single-name routes to the
self-managed set (handlers do their own authz). No wildcard is added, so the
by_name/by_model subpaths are NOT covered here."""
routes = LiteLLMRoutes.self_managed_routes.value
assert "/credentials" in routes
assert "/credentials/{credential_name}" in routes
assert "/credentials/by_name/{credential_name}" not in routes
assert "/credentials/*" not in routes
@pytest.mark.parametrize(
"route,expected",
[
("/credentials", True),
("/credentials/my-dest", True), # single segment -> {credential_name}
("/credentials/by_name/my-dest", False), # extra segment -> no match
("/credentials/by_model/model-123", False), # extra segment -> no match
],
)
def test_credentials_self_managed_pattern_matches_single_segment_only(route, expected):
"""`/credentials/{credential_name}` compiles to ^/credentials/[^/]+$, so a caller
who only reaches self-managed routes (team-admin, org-admin, plain internal user)
can hit the list/single-name routes but NOT the two-segment by_name/by_model
endpoints."""
assert (
RouteChecks.check_route_access(
route=route, allowed_routes=LiteLLMRoutes.self_managed_routes.value
)
is expected
)
@pytest.mark.parametrize(
"route",
["/credentials/by_name/my-dest", "/credentials/by_model/model-123"],
)
def test_by_name_by_model_forbidden_for_internal_user(route):
"""A plain internal user (also the role a team-admin/org-admin key carries) is
denied by_name/by_model: the route matches no allow-list, so the gate raises."""
user_obj = LiteLLM_UserTable(
user_id="u1", user_role=LitellmUserRoles.INTERNAL_USER.value
)
valid_token = UserAPIKeyAuth(
user_id="u1", user_role=LitellmUserRoles.INTERNAL_USER.value
)
request = MagicMock(spec=Request)
request.method = "GET"
request.query_params = {}
with pytest.raises(Exception) as exc:
RouteChecks.non_proxy_admin_allowed_routes_check(
user_obj=user_obj,
_user_role=LitellmUserRoles.INTERNAL_USER.value,
route=route,
request=request,
valid_token=valid_token,
request_data={},
)
assert "Only proxy admin" in str(exc.value)
assert f"Route={route}" in str(exc.value)
@pytest.mark.parametrize(
"route",
["/credentials/by_name/my-dest", "/credentials/by_model/model-123"],
)
def test_by_name_by_model_reachable_by_admin_viewer(route):
"""PROXY_ADMIN_VIEW_ONLY reaches by_name/by_model via the read-parity safe-GET
default-allow (documented in the PR body as a masking inconsistency, not a
cross-tenant leak). Pins that the viewer is NOT blocked at the route gate."""
request = MagicMock(spec=Request)
request.method = "GET"
request.query_params = {}
# returns None (allow) rather than raising
assert (
RouteChecks._check_proxy_admin_viewer_access(
route=route,
_user_role=LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY.value,
request_data={},
request=request,
)
is None
)

View file

@ -4271,6 +4271,32 @@ async def test_builder_hoists_destinations_before_post_lookup_auth_checks():
mock_enforce.assert_awaited_once()
@pytest.mark.asyncio
async def test_hoist_destinations_resolver_failure_never_breaks_auth():
"""Destination resolution is best-effort telemetry setup: if the resolver raises,
_hoist_request_destinations must swallow it and leave the ContextVar at its empty
default so auth proceeds and the fan-out processor no-ops. A raise here would take
down every request."""
from unittest.mock import AsyncMock, MagicMock, patch
from litellm.integrations.otel.plumbing.context import request_destinations
from litellm.proxy.auth.user_api_key_auth import _hoist_request_destinations
request = MagicMock()
request.state = MagicMock()
valid_token = UserAPIKeyAuth(api_key="sk-x", token="hashed")
with patch(
"litellm.proxy.litellm_pre_call_utils._resolve_logging_exporters",
new_callable=AsyncMock,
side_effect=RuntimeError("resolver blew up"),
):
# must not raise
await _hoist_request_destinations(request, valid_token)
assert request_destinations() == ()
def _mint_cli_session_token(monkeypatch, *, user_id="cli-admin"):
"""Mint a CLI session token for a PROXY_ADMIN user so auth resolves on the
admin early-return path (no prisma/common_checks needed)."""