fix: RBAC bypass vulnerabilities in model access (#4270)

Closes security gaps where RBAC checks could be bypassed:

o Inference router: Added RBAC enforcement in the fallback
  path to ensure access control is applied consistently.

o Model listing: Dynamic models fetched via provider_data were returned
  without RBAC checks. Added filtering to ensure users only see models
  they have permission to access.

Both fixes create temporary ModelWithOwner objects for RBAC validation,
maintaining security through consistent access control enforcement.

Closes: #4269

Signed-off-by: Derek Higgins <derekh@redhat.com>
This commit is contained in:
Derek Higgins 2025-12-03 13:42:22 +00:00 committed by GitHub
parent 7f43051a63
commit 8940be23c4
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 229 additions and 5 deletions

View file

@ -14,6 +14,9 @@ from openai.types.chat import ChatCompletionToolChoiceOptionParam as OpenAIChatC
from openai.types.chat import ChatCompletionToolParam as OpenAIChatCompletionToolParam
from pydantic import TypeAdapter
from llama_stack.core.access_control.access_control import is_action_allowed
from llama_stack.core.datatypes import ModelWithOwner
from llama_stack.core.request_headers import get_authenticated_user
from llama_stack.log import get_logger
from llama_stack.providers.utils.inference.inference_store import InferenceStore
from llama_stack_api import (
@ -93,15 +96,41 @@ class InferenceRouter(Inference):
provider = await self.routing_table.get_provider_impl(model.identifier)
return provider, model.provider_resource_id
# Handles cases where clients use the provider format directly
return await self._get_provider_by_fallback(model_id, expected_model_type)
async def _get_provider_by_fallback(self, model_id: str, expected_model_type: str) -> tuple[Inference, str]:
"""
Handle fallback case where model_id is in provider_id/provider_resource_id format.
"""
splits = model_id.split("/", maxsplit=1)
if len(splits) != 2:
raise ModelNotFoundError(model_id)
provider_id, provider_resource_id = splits
# Check if provider exists
if provider_id not in self.routing_table.impls_by_provider_id:
logger.warning(f"Provider {provider_id} not found for model {model_id}")
raise ModelNotFoundError(model_id)
# Create a temporary model object for RBAC check
temp_model = ModelWithOwner(
identifier=model_id,
provider_id=provider_id,
provider_resource_id=provider_resource_id,
model_type=expected_model_type,
metadata={}, # Empty metadata for temporary object
)
# Perform RBAC check
user = get_authenticated_user()
if not is_action_allowed(self.routing_table.policy, "read", temp_model, user):
logger.debug(
f"Access denied to model '{model_id}' via fallback path for user {user.principal if user else 'anonymous'}"
)
raise ModelNotFoundError(model_id)
return self.routing_table.impls_by_provider_id[provider_id], provider_resource_id
async def rerank(