fix(auth): enforce model ACL on Bedrock passthrough routes

Bedrock passthrough routes (/bedrock/model/{modelId}/invoke, converse,
etc.) bypass key.models and user.models access control because
get_model_from_request() does not extract the model from the Bedrock
URL pattern. The native Bedrock request body has no top-level "model"
field, so the model lives exclusively in the URL path.

This adds:
- Regex to extract modelId from Bedrock passthrough URL patterns
- Reverse lookup from Bedrock modelId to LiteLLM model_group via
  llm_router.get_model_list() deployments
- Threading of llm_router param through all get_model_from_request
  call sites (user_api_key_auth.py, auth_checks.py)

Fixes #26399
This commit is contained in:
wangjifeng 2026-04-24 19:12:42 +08:00
parent 17fef6ee80
commit 359c191b39
4 changed files with 186 additions and 5 deletions

View file

@ -485,7 +485,7 @@ async def common_checks( # noqa: PLR0915
from litellm.proxy.proxy_server import prisma_client, user_api_key_cache
_model: Optional[Union[str, List[str]]] = get_model_from_request(
request_body, route
request_body, route, llm_router=llm_router
)
# 1. If team is blocked

View file

@ -879,8 +879,52 @@ def get_end_user_id_from_request_body(
return None
_BEDROCK_PASSTHROUGH_ROUTE_PATTERN = re.compile(
r"^/bedrock(?:/v\d+)?/model/(?P<model>.+)/"
r"(?:invoke|invoke-with-response-stream|converse|converse-stream|count_tokens|count-tokens)"
r"(?:/|$)"
)
def _resolve_bedrock_model_id_to_model_group(
bedrock_model_id: str, llm_router: Optional[Router]
) -> Optional[str]:
"""
Reverse-lookup a Bedrock modelId (as it appears in the URL path) to the
LiteLLM `model_name` (model_group) that routes to it.
The path-side id does not carry a provider prefix, whereas router
deployments are configured as `bedrock/<modelId>` or
`bedrock/converse/<modelId>`. We normalize both sides before comparing.
Returns the deployment's `model_name` if a match is found, else None.
"""
if llm_router is None:
return None
try:
deployments = llm_router.get_model_list() or []
except Exception:
return None
for deployment in deployments:
params = deployment.get("litellm_params") or {}
configured = params.get("model") or ""
if not configured:
continue
for prefix in ("bedrock/converse/", "bedrock/"):
if configured.startswith(prefix):
configured = configured[len(prefix) :]
break
if configured == bedrock_model_id:
name = deployment.get("model_name")
if name:
return name
return None
def get_model_from_request(
request_data: dict, route: str
request_data: dict,
route: str,
llm_router: Optional[Router] = None,
) -> Optional[Union[str, List[str]]]:
# First try to get model from request_data
model = request_data.get("model") or request_data.get("target_model_names")
@ -923,6 +967,19 @@ def get_model_from_request(
if vertex_match:
model = vertex_match.group(1)
# Bedrock passthrough routes carry the modelId exclusively in the URL.
# Pattern: /bedrock[/v{N}]/model/{modelId}/{invoke|converse|...}
# The modelId may contain slashes (e.g. `aws/anthropic/…`) and may be
# prefixed with `application-inference-profile/` for inference profiles.
if model is None:
bedrock_match = _BEDROCK_PASSTHROUGH_ROUTE_PATTERN.match(route)
if bedrock_match:
bedrock_model_id = bedrock_match.group("model")
resolved = _resolve_bedrock_model_id_to_model_group(
bedrock_model_id=bedrock_model_id, llm_router=llm_router
)
model = resolved or bedrock_model_id
return model

View file

@ -869,7 +869,9 @@ async def _user_api_key_auth_builder( # noqa: PLR0915
)
# Check if model has zero cost - if so, skip all budget checks
model = get_model_from_request(request_data, route)
model = get_model_from_request(
request_data, route, llm_router=llm_router
)
skip_budget_checks = False
if model is not None and llm_router is not None:
from litellm.proxy.auth.auth_checks import _is_model_cost_zero
@ -1276,7 +1278,7 @@ async def _user_api_key_auth_builder( # noqa: PLR0915
user_obj = None
# Check 2a. Check if model has zero cost - if so, skip all budget checks
model = get_model_from_request(request_data, route)
model = get_model_from_request(request_data, route, llm_router=llm_router)
skip_budget_checks = False
if model is not None and llm_router is not None:
from litellm.proxy.auth.auth_checks import _is_model_cost_zero
@ -1807,7 +1809,7 @@ async def _enforce_key_and_fallback_model_access(
):
pass
else:
model = get_model_from_request(request_data, route)
model = get_model_from_request(request_data, route, llm_router=llm_router)
fallback_models = cast(
Optional[List[ALL_FALLBACK_MODEL_VALUES]],
request_data.get("fallbacks", None),

View file

@ -255,6 +255,128 @@ def test_get_model_from_request_vertex_passthrough_still_works():
assert get_model_from_request(request_data={}, route=route) == "gemini-1.5-pro"
class TestGetModelFromRequestBedrockPassthrough:
"""
Bedrock passthrough routes put the model in the URL, not the body.
Auth ACL enforcement must extract the model from the path, otherwise
key.models / user.models restrictions are silently bypassed.
"""
BEDROCK_MODEL_ID = "global.anthropic.claude-opus-4-7"
def _router_with_bedrock_deployment(
self, model_name: str = "claude-opus-4-7", bedrock_model: Optional[str] = None
) -> MagicMock:
"""Mock an llm_router that exposes one deployment for reverse-lookup."""
router = MagicMock()
configured_model = bedrock_model or f"bedrock/{self.BEDROCK_MODEL_ID}"
router.get_model_list.return_value = [
{
"model_name": model_name,
"litellm_params": {"model": configured_model},
}
]
return router
def test_invoke_returns_bedrock_model_id_without_router(self):
route = f"/bedrock/model/{self.BEDROCK_MODEL_ID}/invoke"
assert (
get_model_from_request(request_data={}, route=route)
== self.BEDROCK_MODEL_ID
)
def test_invoke_with_response_stream_returns_bedrock_model_id(self):
route = f"/bedrock/model/{self.BEDROCK_MODEL_ID}/invoke-with-response-stream"
assert (
get_model_from_request(request_data={}, route=route)
== self.BEDROCK_MODEL_ID
)
def test_converse_returns_bedrock_model_id(self):
route = f"/bedrock/model/{self.BEDROCK_MODEL_ID}/converse"
assert (
get_model_from_request(request_data={}, route=route)
== self.BEDROCK_MODEL_ID
)
def test_converse_stream_returns_bedrock_model_id(self):
route = f"/bedrock/model/{self.BEDROCK_MODEL_ID}/converse-stream"
assert (
get_model_from_request(request_data={}, route=route)
== self.BEDROCK_MODEL_ID
)
def test_application_inference_profile_returns_full_profile_id(self):
profile_id = "arn:aws:bedrock:us-east-1:123456789012:application-inference-profile/abc123"
route = f"/bedrock/model/application-inference-profile/{profile_id}/invoke"
# The application-inference-profile path segment is part of the identifier.
result = get_model_from_request(request_data={}, route=route)
assert result == f"application-inference-profile/{profile_id}"
def test_v2_route_prefix_is_supported(self):
route = f"/bedrock/v2/model/{self.BEDROCK_MODEL_ID}/invoke"
assert (
get_model_from_request(request_data={}, route=route)
== self.BEDROCK_MODEL_ID
)
def test_model_id_with_slashes_is_preserved(self):
# e.g. aws/anthropic/bedrock-claude-3-5-sonnet-v1
slashed_model = "aws/anthropic/bedrock-claude-3-5-sonnet-v1"
route = f"/bedrock/model/{slashed_model}/converse"
assert get_model_from_request(request_data={}, route=route) == slashed_model
def test_reverse_lookup_returns_model_group_name_when_router_maps_it(self):
"""When a router deployment maps bedrock/<modelId> to a model_group,
return the model_group name so user-configured ACLs match."""
router = self._router_with_bedrock_deployment(model_name="claude-opus-4-7")
route = f"/bedrock/model/{self.BEDROCK_MODEL_ID}/invoke"
assert (
get_model_from_request(request_data={}, route=route, llm_router=router)
== "claude-opus-4-7"
)
def test_reverse_lookup_falls_back_to_raw_id_when_router_has_no_match(self):
router = MagicMock()
router.get_model_list.return_value = [
{
"model_name": "some-other-model",
"litellm_params": {"model": "bedrock/us.anthropic.claude-3-haiku"},
}
]
route = f"/bedrock/model/{self.BEDROCK_MODEL_ID}/invoke"
assert (
get_model_from_request(request_data={}, route=route, llm_router=router)
== self.BEDROCK_MODEL_ID
)
def test_reverse_lookup_handles_bedrock_converse_prefix(self):
"""litellm_params.model may be 'bedrock/converse/<modelId>'."""
router = self._router_with_bedrock_deployment(
model_name="claude-opus-4-7",
bedrock_model=f"bedrock/converse/{self.BEDROCK_MODEL_ID}",
)
route = f"/bedrock/model/{self.BEDROCK_MODEL_ID}/converse"
assert (
get_model_from_request(request_data={}, route=route, llm_router=router)
== "claude-opus-4-7"
)
def test_non_bedrock_route_ignored(self):
# sanity: make sure we don't accidentally match unrelated routes.
assert get_model_from_request(request_data={}, route="/health") is None
def test_bedrock_count_tokens_not_model_path(self):
# count_tokens / count-tokens is handled separately in the passthrough
# layer and the body usually carries `model`; the path has no modelId.
route = f"/bedrock/model/{self.BEDROCK_MODEL_ID}/count-tokens"
# We still extract the model from the path — ACL must apply.
assert (
get_model_from_request(request_data={}, route=route)
== self.BEDROCK_MODEL_ID
)
def test_get_customer_user_header_returns_none_when_no_customer_role():
from litellm.proxy.auth.auth_utils import get_customer_user_header_from_mapping