feat(mcp): enforce per-user MCP tool-call entitlements in the auth module

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

@@ -1,6 +1,6 @@
 import re
 from datetime import datetime, timezone
-from typing import Dict, List, Optional, Set, Tuple, cast
+from typing import TYPE_CHECKING, Dict, List, Optional, Sequence, Set, Tuple, cast
 
 from fastapi import HTTPException
 from starlette.datastructures import Headers
@@ -30,6 +30,7 @@
 )
 from litellm.proxy._types import (
     UI_TEAM_ID,
+    LiteLLM_ObjectPermissionTable,
     LiteLLM_TeamTable,
     ProxyException,
     SpecialHeaders,
@@ -43,13 +44,27 @@
     user_api_key_auth,
 )
 from litellm.proxy.common_utils.http_parsing_utils import _read_request_body
-from litellm.proxy.common_utils.user_api_key_cache import get_management_object_ttl
+from litellm.proxy.common_utils.user_api_key_cache import (
+    USER_NO_MCP_PERMISSION_SENTINEL,
+    get_management_object_ttl,
+    user_object_permission_id_cache_key,
+)
 from litellm.repositories.table_repositories import (
     AgentsRepository,
     MCPServerRepository,
 )
+from litellm.repositories.user_repository import UserRepository
 from litellm.types.mcp_server.mcp_server_manager import MCPServer
 
+if TYPE_CHECKING:
+    from litellm.proxy.utils import PrismaClient
+
+
+def _as_list(values: Sequence[str] | None) -> list[str] | None:  # mutable-ok: resolver returns a list
+    """Widen a read-only allowlist back to the mutable list the resolver's own contract returns,
+    preserving the ``None`` that means "no restriction"."""
+    return None if values is None else list(values)
+
 
 def _parse_mcp_server_names_from_path(path: str, mcp_servers_header: Optional[List[str]] = None) -> Optional[List[str]]:
     """Resolve the single MCP server name a cold-start passthrough bypass may
@@ -1408,6 +1423,15 @@ async def get_allowed_mcp_servers(
                         f"Applied agent intersection filter. Final allowed servers: {allowed_mcp_servers}"
                     )
 
+            #########################################################
+            # Apply the internal user's own ceiling (the entitlement attached to the human)
+            #########################################################
+            capped, user_restricts = await MCPRequestHandler._apply_user_server_ceiling(
+                allowed_mcp_servers, user_api_key_auth, keyless_source=keyless_source
+            )
+            allowed_mcp_servers = list(capped)
+            has_lower_level_mcp_restrictions = has_lower_level_mcp_restrictions or user_restricts
+
             #########################################################
             # Apply org-level ceiling if org_id is set
             #########################################################
@@ -1831,6 +1855,12 @@ async def get_allowed_tools_for_server(
                 # No team restrictions → use key restrictions
                 allowed_tools = cast(List[str], key_tools)
 
+            allowed_tools = _as_list(
+                await MCPRequestHandler._apply_user_tool_ceiling(
+                    allowed_tools, server_id, user_api_key_auth, keyless_source=keyless_source
+                )
+            )
+
             return await MCPRequestHandler._apply_agent_and_org_tool_ceilings(
                 allowed_tools, server_id, user_api_key_auth, keyless_source=keyless_source
             )
@@ -2376,6 +2406,203 @@ async def _get_allowed_mcp_servers_for_end_user(
             verbose_logger.warning(f"Failed to get allowed MCP servers for end_user: {str(e)}")
             return []
 
+    @staticmethod
+    async def _get_user_object_permission(
+        user_api_key_auth: UserAPIKeyAuth | None = None,
+    ) -> LiteLLM_ObjectPermissionTable | None:
+        """The internal user's OWN object_permission: the entitlement attached to the HUMAN rather
+        than to the credential they authenticated with.
+
+        A key's object_permission is the credential's scope and a team's is the group's; this one
+        answers "which MCP servers and tools is this person entitled to", independent of how many keys
+        they hold. Caches the ``user_id -> object_permission_id`` mapping (with a sentinel for "no
+        entitlement") exactly as the agent path does, then reuses the shared ``object_permission_id``
+        cache, so a warm request reads no rows.
+
+        ``None`` means the human places NO ceiling: no user row, or a row naming no permission. The
+        two fault classes are deliberately NOT collapsed into that: a user row we cannot read leaves
+        us unable to say whether they are entitled at all, which is exactly the state before this
+        level existed, so it places no ceiling; a row that NAMES a permission we cannot read is a
+        KNOWN entitlement with unknown contents, so it raises and the caller denies.
+        """
+        from litellm.proxy.auth.auth_checks import get_object_permission
+        from litellm.proxy.proxy_server import (
+            prisma_client,
+            proxy_logging_obj,
+            user_api_key_cache,
+        )
+
+        if not user_api_key_auth or not user_api_key_auth.user_id:
+            return None
+
+        if prisma_client is None:
+            verbose_logger.debug("prisma_client is None")
+            return None
+
+        user_id = user_api_key_auth.user_id
+        object_permission_id = await MCPRequestHandler._user_object_permission_id(user_id, prisma_client)
+        if object_permission_id is None:
+            return None
+
+        object_permission = await get_object_permission(
+            object_permission_id=object_permission_id,
+            prisma_client=prisma_client,
+            user_api_key_cache=user_api_key_cache,
+            parent_otel_span=user_api_key_auth.parent_otel_span,
+            proxy_logging_obj=proxy_logging_obj,
+        )
+        if object_permission is None:
+            raise ValueError(
+                f"user {user_id!r} names object_permission_id {object_permission_id!r} which could not be loaded"
+            )
+        return object_permission
+
+    @staticmethod
+    async def _user_object_permission_id(user_id: str, prisma_client: "PrismaClient") -> str | None:
+        """The permission row this human's user row links to, or None when they link none.
+
+        Caches the link (with a sentinel for "links none") so a human without an entitlement costs no
+        DB read per MCP request. Anything other than an id string is treated as a cache MISS rather
+        than carried into the permission lookup, and a read that fails answers None: not knowing
+        whether someone is entitled is the state that existed before this level, so it places no
+        ceiling. Only a link we DID resolve can make the caller deny.
+        """
+        from litellm.proxy.proxy_server import user_api_key_cache
+
+        cache_key = user_object_permission_id_cache_key(user_id)
+        try:
+            cached: object = await user_api_key_cache.async_get_cache(key=cache_key)
+            if cached == USER_NO_MCP_PERMISSION_SENTINEL:
+                return None
+            if isinstance(cached, str) and cached:
+                return cached
+            user_row = await UserRepository(prisma_client).table.find_unique(where={"user_id": user_id})
+            linked: object = getattr(user_row, "object_permission_id", None) if user_row is not None else None
+            object_permission_id = linked if isinstance(linked, str) and linked else None
+            await user_api_key_cache.async_set_cache(
+                key=cache_key,
+                value=object_permission_id or USER_NO_MCP_PERMISSION_SENTINEL,
+                ttl=get_management_object_ttl(user_api_key_cache),
+            )
+            return object_permission_id
+        except Exception as e:  # noqa: BLE001  # unknown whether entitled at all: no ceiling, as before
+            verbose_logger.warning(f"MCP user entitlement: link for {user_id!r} unresolved, no ceiling: {str(e)}")
+            return None
+
+    @staticmethod
+    async def _get_allowed_mcp_servers_for_user(
+        user_api_key_auth: UserAPIKeyAuth | None = None,
+    ) -> Sequence[str] | None:
+        """The MCP servers the internal user is entitled to, as server ids.
+
+        ``[]`` means this human places no restriction (allow-all from this level); ``None`` means the
+        ceiling is UNRESOLVED, which the caller denies on. Servers named only under
+        ``mcp_tool_permissions`` count as entitled, exactly as they do for a key or a team, so
+        granting one tool never requires naming its server twice.
+        """
+        from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+            global_mcp_server_manager,
+        )
+
+        try:
+            object_permissions = await MCPRequestHandler._get_user_object_permission(user_api_key_auth)
+            if object_permissions is None:
+                return []
+
+            direct_mcp_servers = global_mcp_server_manager.expand_permission_list(object_permissions.mcp_servers or [])
+            access_group_servers = await MCPRequestHandler._get_mcp_servers_from_access_groups(
+                object_permissions.mcp_access_groups or []
+            )
+            tool_perm_servers = list(
+                global_mcp_server_manager.expand_tool_permissions(object_permissions.mcp_tool_permissions).keys()
+            )
+            return list(set(direct_mcp_servers + access_group_servers + tool_perm_servers))
+        except Exception as e:  # noqa: BLE001  # any resolution fault is an unresolved ceiling, never "no ceiling"
+            verbose_logger.warning(f"Failed to get allowed MCP servers for user: {str(e)}")
+            return None
+
+    @staticmethod
+    async def _apply_user_server_ceiling(
+        allowed_mcp_servers: Sequence[str],
+        user_api_key_auth: UserAPIKeyAuth | None = None,
+        *,
+        keyless_source: bool = False,
+    ) -> tuple[tuple[str, ...], bool]:
+        """Narrow a resolved server list by the internal user's own entitlement.
+
+        Returns the capped list and whether this human restricted it at all; the caller needs the
+        second value because an org list may only CAP a lower-level restriction, never replace one, so
+        a user ceiling has to be visible to the org step.
+
+        RAISES when the entitlement is known but unreadable, which the resolver's own handler turns
+        into deny-all. That is the point of the level: dropping a ceiling we know exists is exactly the
+        silent widening it is there to prevent.
+        """
+        if keyless_source:
+            return tuple(allowed_mcp_servers), False
+        entitled = await MCPRequestHandler._get_allowed_mcp_servers_for_user(user_api_key_auth)
+        if entitled is None:
+            raise ValueError(
+                f"MCP user ceiling unresolvable for user_id="
+                f"{user_api_key_auth.user_id if user_api_key_auth else None!r}"
+            )
+        if not entitled:
+            return tuple(allowed_mcp_servers), False
+        capped = tuple(server for server in allowed_mcp_servers if server in set(entitled))
+        verbose_logger.debug(f"Applied user ceiling filter. Final allowed servers: {capped}")
+        return capped, True
+
+    @staticmethod
+    async def _user_places_mcp_ceiling(user_api_key_auth: UserAPIKeyAuth | None = None) -> bool:
+        """Whether this human's own entitlement bounds their MCP access at all.
+
+        True when they are entitled to a specific set of servers, and also when that entitlement is
+        UNRESOLVED — a caller uses this to decide whether it may skip the resolver, and skipping it on
+        a transient fault would widen access.
+        """
+        entitled_servers = await MCPRequestHandler._get_allowed_mcp_servers_for_user(user_api_key_auth)
+        return entitled_servers is None or len(entitled_servers) > 0
+
+    @staticmethod
+    async def _apply_user_tool_ceiling(
+        allowed_tools: Sequence[str] | None,
+        server_id: str,
+        user_api_key_auth: UserAPIKeyAuth | None = None,
+        *,
+        keyless_source: bool = False,
+    ) -> Sequence[str] | None:
+        """Narrow a key/team tool allowlist by the internal user's own tool entitlement.
+
+        The human's entitlement can only ever narrow: a user naming tools on ``server_id`` intersects
+        (and becomes the allowlist when no lower level restricts), while a user naming none places no
+        restriction. Returns ``[]`` (deny every tool on this server) when the entitlement cannot be
+        resolved, because the caller's own except-handler treats a raise as allow-all for key auth.
+        """
+        from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
+            global_mcp_server_manager,
+        )
+
+        if keyless_source:
+            return allowed_tools
+
+        try:
+            object_permissions = await MCPRequestHandler._get_user_object_permission(user_api_key_auth)
+        except Exception as e:  # noqa: BLE001  # an unresolved human entitlement must deny, not widen
+            verbose_logger.warning(f"MCP user tool ceiling unresolvable, denying tools on {server_id!r}: {str(e)}")
+            return []
+
+        if object_permissions is None or not object_permissions.mcp_tool_permissions:
+            return allowed_tools
+
+        user_tools = global_mcp_server_manager.expand_tool_permissions(object_permissions.mcp_tool_permissions).get(
+            server_id
+        )
+        if user_tools is None:
+            return allowed_tools
+        if allowed_tools is None:
+            return list(user_tools)
+        return list(set(allowed_tools) & set(user_tools))
+
     # Sentinel stored in cache when an agent has no object_permission, so we
     # don't re-query the DB on every MCP request for that agent.
     _AGENT_NO_PERMISSION_SENTINEL = "__agent_no_mcp_permission__"

litellm/proxy/_experimental/mcp_server/mcp_server_manager.py

@@ -2418,6 +2418,11 @@ async def get_allowed_mcp_servers(self, user_api_key_auth: Optional[UserAPIKeyAu
                 and not is_admitted_subject
                 and _user_has_admin_view(user_api_key_auth)
                 and not has_explicit_object_permission
+                # An entitlement attached to the HUMAN binds them whatever their role: it is the
+                # person's scope, not the credential's, so an admin role is not a waiver of it. An
+                # UNRESOLVED entitlement also skips the shortcut, so the resolver denies rather than
+                # handing over the whole registry on a transient fault.
+                and not await MCPRequestHandler._user_places_mcp_ceiling(user_api_key_auth)
             ):
                 verbose_logger.debug("Admin user without explicit object_permission - returning all servers")
                 return list(self.get_registry().keys())

litellm/proxy/_types.py

@@ -2782,6 +2782,7 @@ class UserInfoV2Response(LiteLLMPydanticObjectBase):
     updated_at: Optional[datetime] = None
     sso_user_id: Optional[str] = None
     teams: List[str] = []  # Just team IDs, not full team objects
+    object_permission: LiteLLM_ObjectPermissionTable | None = None
 
 
 from litellm.models.config import LiteLLM_Config as LiteLLM_Config  # noqa: E402

litellm/proxy/auth/auth_checks.py

@@ -74,6 +74,7 @@
 from litellm.proxy.common_utils.user_api_key_cache import (
     UserApiKeyCache,
     get_management_object_ttl,
+    object_permission_cache_key,
 )
 from litellm.proxy.db.exception_handler import PrismaDBExceptionHandler
 from litellm.proxy.guardrails.tool_name_extraction import (
@@ -2609,7 +2610,7 @@ async def get_object_permission(
         raise Exception("No DB Connected. See - https://docs.litellm.ai/docs/proxy/virtual_keys")
 
     # check if in cache
-    key = "object_permission_id:{}".format(object_permission_id)
+    key = object_permission_cache_key(object_permission_id)
     deserialized_perm = await user_api_key_cache.async_get_cache(
         key=key,
         model_type=LiteLLM_ObjectPermissionTable,

litellm/proxy/common_utils/user_api_key_cache.py

@@ -150,6 +150,28 @@ async def async_set_cache_pipeline(  # type: ignore[override]
         return await super().async_set_cache_pipeline(cache_list=normalized, local_only=local_only, **kwargs)
 
 
+#: Value cached under ``user_object_permission_id_cache_key`` when the user links no permission row,
+#: so a human without an entitlement costs no DB read per request. Lives beside the key builder
+#: because it is part of the same cache protocol: a reader that knows the key must know this value.
+USER_NO_MCP_PERMISSION_SENTINEL = "__user_no_mcp_permission__"
+
+
+def user_object_permission_id_cache_key(user_id: str) -> str:
+    """Cache key for the ``user_id -> object_permission_id`` link.
+
+    Lives here rather than next to either user because two modules own the two halves: the MCP auth
+    resolver writes it on read, and ``/user/update`` deletes it after changing the link. A key format
+    duplicated across those two drifts silently, and the failure is an entitlement change that never
+    takes effect.
+    """
+    return f"user_object_permission_id:{user_id}"
+
+
+def object_permission_cache_key(object_permission_id: str) -> str:
+    """Cache key ``get_object_permission`` stores a permission row under."""
+    return f"object_permission_id:{object_permission_id}"
+
+
 def get_management_object_ttl(cache: DualCache) -> float:
     """
     In-memory TTL for management-object cache writes (keys, teams, users, budgets, ...).