mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-05 02:41:56 +00:00
feat(mcp): user->agent delegation core for on-behalf-of credential resolution
Feature-flagged (mcp_user_delegation_enabled, default off) delegation/consent layer for MCP: an agent-bound key whose agent has mcp_can_delegate and an active consent record from the asserted user resolves upstream per-user credentials as that user. Header, capability, and consent are validated once at admission for every transport; only the credential-resolution subject swaps, permissions and attribution stay on the agent key. Consent and email lookups are cached with grant/revoke invalidation. Part of LIT-4448.
This commit is contained in:
Tin Chi Lo
parent
e5a9f3f5d7
commit
490028b0ef
22 changed files with 1194 additions and 7 deletions
|
|
@ -0,0 +1,19 @@
|
|||
-- AlterTable
|
||||
ALTER TABLE "LiteLLM_ObjectPermissionTable" ADD COLUMN "mcp_can_delegate" BOOLEAN;
|
||||
|
||||
-- CreateTable
|
||||
CREATE TABLE IF NOT EXISTS "LiteLLM_UserAgentDelegationTable" (
|
||||
"delegation_id" TEXT NOT NULL,
|
||||
"user_id" TEXT NOT NULL,
|
||||
"agent_id" TEXT NOT NULL,
|
||||
"granted_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
"granted_by" TEXT NOT NULL,
|
||||
"revoked_at" TIMESTAMP(3),
|
||||
"revoked_by" TEXT,
|
||||
"updated_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
|
||||
|
||||
CONSTRAINT "LiteLLM_UserAgentDelegationTable_pkey" PRIMARY KEY ("delegation_id")
|
||||
);
|
||||
|
||||
-- CreateIndex
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS "LiteLLM_UserAgentDelegationTable_user_id_agent_id_key" ON "LiteLLM_UserAgentDelegationTable"("user_id", "agent_id");
|
||||
|
|
@ -280,6 +280,7 @@ model LiteLLM_ObjectPermissionTable {
|
|||
mcp_toolsets String[] @default([]) // Toolset IDs granted to this key/team/user
|
||||
search_tools String[] @default([]) // search_tool_name values this key/team/user may call
|
||||
mcp_tool_search_enabled Boolean?
|
||||
mcp_can_delegate Boolean? // Agents with this permission may act on behalf of consenting users on MCP routes
|
||||
teams LiteLLM_TeamTable[]
|
||||
projects LiteLLM_ProjectTable[]
|
||||
verification_tokens LiteLLM_VerificationToken[]
|
||||
|
|
@ -369,6 +370,20 @@ model LiteLLM_MCPToolsetTable {
|
|||
updated_by String?
|
||||
}
|
||||
|
||||
// User consent records authorizing an agent to act on the user's behalf on MCP routes
|
||||
model LiteLLM_UserAgentDelegationTable {
|
||||
delegation_id String @id @default(uuid())
|
||||
user_id String
|
||||
agent_id String
|
||||
granted_at DateTime @default(now())
|
||||
granted_by String
|
||||
revoked_at DateTime?
|
||||
revoked_by String?
|
||||
updated_at DateTime @default(now()) @updatedAt
|
||||
|
||||
@@unique([user_id, agent_id])
|
||||
}
|
||||
|
||||
// Per-user BYOK credentials for MCP servers
|
||||
model LiteLLM_MCPUserCredentials {
|
||||
id String @id @default(uuid())
|
||||
|
|
|
|||
|
|
@ -25,3 +25,4 @@ class LiteLLM_ObjectPermissionTable(LiteLLMPydanticObjectBase):
|
|||
blocked_tools: Optional[List[str]] = []
|
||||
search_tools: Optional[List[str]] = []
|
||||
mcp_tool_search_enabled: Optional[bool] = None
|
||||
mcp_can_delegate: bool | None = None
|
||||
|
|
|
|||
|
|
@ -322,6 +322,10 @@ class MCPRequestHandler:
|
|||
else:
|
||||
raise
|
||||
|
||||
validated_user_api_key_auth = await MCPRequestHandler.resolve_delegated_user_auth(
|
||||
validated_user_api_key_auth, headers
|
||||
)
|
||||
|
||||
return (
|
||||
validated_user_api_key_auth,
|
||||
mcp_auth_header,
|
||||
|
|
@ -331,6 +335,115 @@ class MCPRequestHandler:
|
|||
dict(headers),
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
async def resolve_delegated_user_auth(
|
||||
user_api_key_auth: UserAPIKeyAuth | None,
|
||||
headers: Headers,
|
||||
) -> UserAPIKeyAuth | None:
|
||||
"""Validate an x-litellm-delegated-user assertion and stamp the delegated subject.
|
||||
|
||||
The asserted email is untrusted input: it grants nothing unless the gateway
|
||||
enables delegation, the calling key is bound to an agent whose object
|
||||
permission carries mcp_can_delegate, AND the asserted user holds an active
|
||||
consent record for that agent. Every failure is a 403 (never a silent
|
||||
ignore, which would let an agent believe it acted as the user while acting
|
||||
as itself). Requests without the header are returned unchanged.
|
||||
|
||||
Scope of the swap: the stamped delegated_user_id changes only the
|
||||
credential-resolution subject for the per-user OAuth (authorization_code)
|
||||
arm, where the delegated user's own stored token is injected upstream (or
|
||||
the request fails closed if they have none). For modes that do not resolve
|
||||
a per-user stored token (token_exchange/OBO, passthrough, client
|
||||
credentials, BYOK), the agent's own credentials are used unchanged; this
|
||||
is not an escalation (the agent never gains the user's identity upstream),
|
||||
and per-user token exchange for OBO servers is a follow-up (the mint arm).
|
||||
"""
|
||||
asserted_email = (headers.get(SpecialHeaders.mcp_delegated_user.value) or "").strip()
|
||||
if not asserted_email:
|
||||
return user_api_key_auth
|
||||
|
||||
from litellm.proxy._experimental.mcp_server.delegation_db import (
|
||||
get_active_user_agent_delegation,
|
||||
)
|
||||
from litellm.proxy.proxy_server import general_settings, prisma_client
|
||||
|
||||
if not general_settings.get("mcp_user_delegation_enabled", False):
|
||||
raise HTTPException(
|
||||
status_code=403,
|
||||
detail={"error": "User delegation is not enabled on this gateway (mcp_user_delegation_enabled)."},
|
||||
)
|
||||
|
||||
if user_api_key_auth is None or not user_api_key_auth.agent_id:
|
||||
raise HTTPException(
|
||||
status_code=403,
|
||||
detail={"error": "Delegation requires a key bound to an agent (key has no agent_id)."},
|
||||
)
|
||||
|
||||
agent_object_permission = await MCPRequestHandler._get_agent_object_permission(user_api_key_auth)
|
||||
if agent_object_permission is None or not getattr(agent_object_permission, "mcp_can_delegate", False):
|
||||
raise HTTPException(
|
||||
status_code=403,
|
||||
detail={
|
||||
"error": f"Agent {user_api_key_auth.agent_id} is not permitted to act on behalf of users (mcp_can_delegate)."
|
||||
},
|
||||
)
|
||||
|
||||
if prisma_client is None:
|
||||
raise HTTPException(
|
||||
status_code=403,
|
||||
detail={"error": "Delegation requires a connected database."},
|
||||
)
|
||||
|
||||
delegated_user_id = await MCPRequestHandler._resolve_single_user_id_by_email(asserted_email)
|
||||
active_delegation = (
|
||||
await get_active_user_agent_delegation(
|
||||
prisma_client, user_id=delegated_user_id, agent_id=user_api_key_auth.agent_id
|
||||
)
|
||||
if delegated_user_id is not None
|
||||
else None
|
||||
)
|
||||
if active_delegation is None:
|
||||
verbose_logger.warning(
|
||||
"MCP delegation rejected: no active consent for asserted user (agent_id=%s, user_resolved=%s)",
|
||||
user_api_key_auth.agent_id,
|
||||
delegated_user_id is not None,
|
||||
)
|
||||
raise HTTPException(
|
||||
status_code=403,
|
||||
detail={"error": "Delegation not authorized for the asserted user."},
|
||||
)
|
||||
|
||||
return user_api_key_auth.model_copy(update={"delegated_user_id": delegated_user_id})
|
||||
|
||||
@staticmethod
|
||||
async def _resolve_single_user_id_by_email(email: str) -> str | None:
|
||||
"""The user_id for an email, or None when unknown OR ambiguous.
|
||||
|
||||
user_email has no unique constraint, so a multi-match must fail closed
|
||||
rather than picking an arbitrary row. This is intentionally NOT cached:
|
||||
the ambiguity guard is a security control, and a cached mapping would go
|
||||
stale on user creation/deletion/email-reassignment (there is no
|
||||
user-mutation invalidation hook), which could resolve the assertion to
|
||||
the wrong or a deleted user. The hot path stays light because the
|
||||
downstream consent lookup is cached; this resolves the identity live.
|
||||
"""
|
||||
from litellm.proxy.proxy_server import prisma_client
|
||||
from litellm.repositories.user_repository import UserRepository
|
||||
|
||||
if prisma_client is None:
|
||||
return None
|
||||
|
||||
users = await UserRepository(prisma_client).table.find_many(
|
||||
where={"user_email": {"equals": email, "mode": "insensitive"}}, take=2
|
||||
)
|
||||
if len(users) != 1:
|
||||
verbose_logger.warning(
|
||||
"MCP delegation user lookup returned %d matches for the asserted email; failing closed",
|
||||
len(users),
|
||||
)
|
||||
return None
|
||||
return users[0].user_id
|
||||
|
||||
@staticmethod
|
||||
def _extract_target_server_names_from_path(path: str) -> List[str]:
|
||||
"""
|
||||
|
|
|
|||
118
litellm/proxy/_experimental/mcp_server/delegation_db.py
Normal file
118
litellm/proxy/_experimental/mcp_server/delegation_db.py
Normal file
|
|
@ -0,0 +1,118 @@
|
|||
from datetime import datetime, timezone
|
||||
|
||||
from litellm.proxy.common_utils.user_api_key_cache import get_management_object_ttl
|
||||
from litellm.proxy.utils import PrismaClient
|
||||
from litellm.repositories.table_repositories import UserAgentDelegationRepository
|
||||
from litellm.types.mcp_server.user_agent_delegation import UserAgentDelegation
|
||||
|
||||
# Sentinel cached for a (user, agent) pair with no active consent, so a repeated
|
||||
# unauthorized assertion is answered from cache instead of re-querying the DB.
|
||||
_NO_ACTIVE_DELEGATION = "__no_active_delegation__"
|
||||
|
||||
|
||||
def _delegation_cache_key(user_id: str, agent_id: str) -> str:
|
||||
return f"mcp_user_delegation:{user_id}:{agent_id}"
|
||||
|
||||
|
||||
async def grant_user_agent_delegation(
|
||||
prisma_client: PrismaClient,
|
||||
user_id: str,
|
||||
agent_id: str,
|
||||
granted_by: str,
|
||||
) -> UserAgentDelegation:
|
||||
"""Grant consent for `agent_id` to act on behalf of `user_id`.
|
||||
|
||||
Upserts so re-granting after a revocation reactivates the same row
|
||||
(revoked_at cleared) instead of violating the (user_id, agent_id) unique.
|
||||
"""
|
||||
row = await UserAgentDelegationRepository(prisma_client).table.upsert(
|
||||
where={"user_id_agent_id": {"user_id": user_id, "agent_id": agent_id}},
|
||||
data={
|
||||
"create": {"user_id": user_id, "agent_id": agent_id, "granted_by": granted_by},
|
||||
"update": {
|
||||
"granted_at": datetime.now(timezone.utc),
|
||||
"granted_by": granted_by,
|
||||
"revoked_at": None,
|
||||
"revoked_by": None,
|
||||
},
|
||||
},
|
||||
)
|
||||
await _invalidate_delegation_cache(user_id, agent_id)
|
||||
return UserAgentDelegation(**row.model_dump())
|
||||
|
||||
|
||||
async def revoke_user_agent_delegation(
|
||||
prisma_client: PrismaClient,
|
||||
user_id: str,
|
||||
agent_id: str,
|
||||
revoked_by: str,
|
||||
) -> UserAgentDelegation | None:
|
||||
"""Revoke an ACTIVE consent. Returns None when there is nothing active to
|
||||
revoke (no record, or already revoked), so the caller surfaces a 404 and a
|
||||
repeat revoke never overwrites the original revoked_at/revoked_by audit."""
|
||||
existing = await UserAgentDelegationRepository(prisma_client).table.find_unique(
|
||||
where={"user_id_agent_id": {"user_id": user_id, "agent_id": agent_id}}
|
||||
)
|
||||
if existing is None or existing.revoked_at is not None:
|
||||
return None
|
||||
row = await UserAgentDelegationRepository(prisma_client).table.update(
|
||||
where={"user_id_agent_id": {"user_id": user_id, "agent_id": agent_id}},
|
||||
data={"revoked_at": datetime.now(timezone.utc), "revoked_by": revoked_by},
|
||||
)
|
||||
await _invalidate_delegation_cache(user_id, agent_id)
|
||||
return UserAgentDelegation(**row.model_dump()) if row else None
|
||||
|
||||
|
||||
async def get_active_user_agent_delegation(
|
||||
prisma_client: PrismaClient,
|
||||
user_id: str,
|
||||
agent_id: str,
|
||||
) -> UserAgentDelegation | None:
|
||||
"""The active (never-revoked) consent row for this pair, if any.
|
||||
|
||||
Cached in ``user_api_key_cache`` (positive and negative) so a delegated
|
||||
request does not hit the DB on the hot path; grant and revoke both bust the
|
||||
entry, so a consent change still takes effect immediately across workers
|
||||
rather than only at TTL.
|
||||
"""
|
||||
from litellm.proxy.proxy_server import user_api_key_cache
|
||||
|
||||
cache_key = _delegation_cache_key(user_id, agent_id)
|
||||
cached = await user_api_key_cache.async_get_cache(key=cache_key)
|
||||
if cached == _NO_ACTIVE_DELEGATION:
|
||||
return None
|
||||
if isinstance(cached, dict):
|
||||
return UserAgentDelegation(**cached)
|
||||
|
||||
row = await UserAgentDelegationRepository(prisma_client).table.find_first(
|
||||
where={"user_id": user_id, "agent_id": agent_id, "revoked_at": None}
|
||||
)
|
||||
ttl = get_management_object_ttl(user_api_key_cache)
|
||||
if row is None:
|
||||
await user_api_key_cache.async_set_cache(key=cache_key, value=_NO_ACTIVE_DELEGATION, ttl=ttl)
|
||||
return None
|
||||
delegation = UserAgentDelegation(**row.model_dump())
|
||||
await user_api_key_cache.async_set_cache(key=cache_key, value=delegation.model_dump(mode="json"), ttl=ttl)
|
||||
return delegation
|
||||
|
||||
|
||||
async def list_user_agent_delegations(
|
||||
prisma_client: PrismaClient,
|
||||
user_id: str | None = None,
|
||||
) -> list[UserAgentDelegation]:
|
||||
"""List delegations, scoped to `user_id` when given.
|
||||
|
||||
`user_id=None` means "all rows across all users" and is an ADMIN-ONLY view;
|
||||
callers gating on a non-admin identity must pass that identity's user_id (and
|
||||
handle a None identity themselves) rather than falling through to this
|
||||
unscoped listing.
|
||||
"""
|
||||
where = {"user_id": user_id} if user_id is not None else {}
|
||||
rows = await UserAgentDelegationRepository(prisma_client).table.find_many(where=where, order={"granted_at": "desc"})
|
||||
return [UserAgentDelegation(**r.model_dump()) for r in rows]
|
||||
|
||||
|
||||
async def _invalidate_delegation_cache(user_id: str, agent_id: str) -> None:
|
||||
from litellm.proxy.proxy_server import user_api_key_cache
|
||||
|
||||
await user_api_key_cache.async_delete_cache(key=_delegation_cache_key(user_id, agent_id))
|
||||
|
|
@ -3,7 +3,7 @@
|
|||
These edge functions translate v1's request objects into the resolver's typed inputs and map
|
||||
its typed errors onto the proxy's public exception contract. They import v1 and live outside the
|
||||
package's public surface so the resolver core (``resolver.py`` / ``types.py``) stays v1-free.
|
||||
Nothing wires them into ``_create_mcp_client`` yet.
|
||||
``_create_mcp_client`` wires them in via ``_resolve_v2_auth`` for every migrated mode.
|
||||
|
||||
``to_server_spec`` maps only the modes the resolver has gone live for, returning ``None`` for
|
||||
every other mode so the caller defers to v1 (parity-safe); it grows one branch per migrated mode.
|
||||
|
|
@ -40,14 +40,17 @@ def to_subject(user_api_key_auth: Optional[UserAPIKeyAuth], subject_token: Optio
|
|||
"""Map v1's authenticated principal onto the resolver's Subject.
|
||||
|
||||
tenant_id / subject_id are empty for an unauthenticated caller; the per-user arms must reject
|
||||
an empty subject rather than share one credential slot across callers.
|
||||
an empty subject rather than share one credential slot across callers. A validated delegation
|
||||
assertion (UserAPIKeyAuth.delegated_user_id, stamped at MCP admission after the consent check)
|
||||
replaces the credential subject so per-user upstream credentials resolve as the delegated
|
||||
user; admission, permissions, and attribution stay on the calling key.
|
||||
"""
|
||||
inbound = SecretStr(subject_token) if subject_token else None
|
||||
if user_api_key_auth is None:
|
||||
return Subject(tenant_id="", subject_id="", inbound_token=inbound)
|
||||
return Subject(
|
||||
tenant_id=user_api_key_auth.org_id or user_api_key_auth.team_id or "",
|
||||
subject_id=user_api_key_auth.user_id or "",
|
||||
subject_id=user_api_key_auth.delegated_user_id or user_api_key_auth.user_id or "",
|
||||
inbound_token=inbound,
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -706,6 +706,12 @@ if MCP_AVAILABLE:
|
|||
MCPRequestHandler,
|
||||
)
|
||||
|
||||
# Validate any delegation assertion up front, outside the try below, so a
|
||||
# delegation 403 propagates instead of being swallowed into a 200 empty-tools
|
||||
# response; also runs before the virtual-tools early return so that path
|
||||
# cannot bypass the assertion check.
|
||||
user_api_key_dict = await MCPRequestHandler.resolve_delegated_user_auth(user_api_key_dict, request.headers)
|
||||
|
||||
try:
|
||||
mcp_server_name = _as_query_str(mcp_server_name)
|
||||
toolset_name = _as_query_str(toolset_name)
|
||||
|
|
@ -899,6 +905,8 @@ if MCP_AVAILABLE:
|
|||
)
|
||||
|
||||
try:
|
||||
user_api_key_dict = await MCPRequestHandler.resolve_delegated_user_auth(user_api_key_dict, request.headers)
|
||||
|
||||
data = await request.json()
|
||||
|
||||
tool_name = data.get("name")
|
||||
|
|
|
|||
|
|
@ -1009,6 +1009,7 @@ class LiteLLM_ObjectPermissionBase(LiteLLMPydanticObjectBase):
|
|||
models: Optional[List[str]] = None
|
||||
search_tools: Optional[List[str]] = None
|
||||
mcp_tool_search_enabled: Optional[bool] = None
|
||||
mcp_can_delegate: bool | None = None
|
||||
|
||||
|
||||
from litellm.types.object_permission import ( # noqa: E402
|
||||
|
|
@ -2365,6 +2366,10 @@ class ConfigGeneralSettings(LiteLLMPydanticObjectBase):
|
|||
None,
|
||||
description="Controls how non-admin users interact with MCP servers in the dashboard. 'restricted' shows only accessible servers, 'view_all' lists every server in read-only mode.",
|
||||
)
|
||||
mcp_user_delegation_enabled: bool | None = Field(
|
||||
None,
|
||||
description="Enables the x-litellm-delegated-user assertion on MCP routes: an agent-bound key whose agent has the mcp_can_delegate permission and an active consent record from the asserted user resolves upstream per-user credentials as that user. Default False; when disabled, requests carrying the header are rejected.",
|
||||
)
|
||||
store_prompts_in_spend_logs: Optional[bool] = Field(
|
||||
None,
|
||||
description="If True, stores request messages and responses in spend logs. Default is False.",
|
||||
|
|
@ -2574,6 +2579,10 @@ class UserAPIKeyAuth(LiteLLM_VerificationTokenView): # the expected response ob
|
|||
api_key: Optional[str] = None
|
||||
user_role: Optional[LitellmUserRoles] = None
|
||||
allowed_model_region: Optional[AllowedModelRegion] = None
|
||||
# Set only after a validated MCP delegation assertion: upstream per-user
|
||||
# credential resolution runs as this user while admission, permissions,
|
||||
# and attribution stay on the calling agent key
|
||||
delegated_user_id: str | None = None
|
||||
parent_otel_span: Optional[Span] = None
|
||||
rpm_limit_per_model: Optional[Dict[str, int]] = None
|
||||
tpm_limit_per_model: Optional[Dict[str, int]] = None
|
||||
|
|
@ -3857,6 +3866,7 @@ class SpecialHeaders(enum.Enum):
|
|||
mcp_auth = "x-mcp-auth"
|
||||
mcp_servers = "x-mcp-servers"
|
||||
mcp_access_groups = "x-mcp-access-groups"
|
||||
mcp_delegated_user = "x-litellm-delegated-user"
|
||||
|
||||
@classmethod
|
||||
def litellm_credential_header_names(cls) -> "frozenset[str]":
|
||||
|
|
|
|||
|
|
@ -2774,3 +2774,173 @@ if MCP_AVAILABLE:
|
|||
|
||||
global_mcp_server_manager.invalidate_toolset_cache(toolset_id)
|
||||
return Response(status_code=status.HTTP_202_ACCEPTED)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# User -> agent delegation (consent records for MCP on-behalf-of resolution)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
from litellm.proxy._experimental.mcp_server.delegation_db import (
|
||||
grant_user_agent_delegation,
|
||||
list_user_agent_delegations,
|
||||
revoke_user_agent_delegation,
|
||||
)
|
||||
from litellm.types.mcp_server.user_agent_delegation import (
|
||||
NewUserAgentDelegationRequest,
|
||||
RevokeUserAgentDelegationRequest,
|
||||
)
|
||||
|
||||
# Module-scope dependency/header singletons so route defaults reference a name
|
||||
# rather than calling Depends()/Header() inline (B008); FastAPI resolves them
|
||||
# the same way. The changed-by header keeps audit attribution consistent with
|
||||
# the other management endpoints.
|
||||
_delegation_user_auth = Depends(user_api_key_auth)
|
||||
_delegation_changed_by = Header(None)
|
||||
|
||||
async def _authorized_delegation_target_user_id(
|
||||
prisma_client: Any,
|
||||
user_api_key_dict: UserAPIKeyAuth,
|
||||
payload_user_id: str | None,
|
||||
payload_user_email: str | None,
|
||||
action: str,
|
||||
) -> str:
|
||||
"""Single authorization + resolution gate shared by grant/revoke.
|
||||
|
||||
A non-admin may only manage its own delegations, and is resolved to its
|
||||
own user_id WITHOUT ever running an email lookup, so this path cannot be
|
||||
used to probe which emails exist. An admin may target any user by id or
|
||||
email; a user_id target is checked for existence so a typo cannot create
|
||||
a phantom, never-usable consent record.
|
||||
"""
|
||||
forbidden = HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail={"error": f"Only proxy admins or the user themselves can {action} a delegation."},
|
||||
)
|
||||
if LitellmUserRoles.PROXY_ADMIN != user_api_key_dict.user_role:
|
||||
caller_user_id = user_api_key_dict.user_id
|
||||
if not caller_user_id:
|
||||
raise forbidden
|
||||
if payload_user_email is not None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail={"error": "Only proxy admins may target a user by email; use your own user_id."},
|
||||
)
|
||||
if payload_user_id is not None and payload_user_id != caller_user_id:
|
||||
raise forbidden
|
||||
return caller_user_id
|
||||
|
||||
if payload_user_id:
|
||||
existing = await prisma_client.db.litellm_usertable.find_unique(where={"user_id": payload_user_id})
|
||||
if existing is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail={"error": f"User '{payload_user_id}' not found."},
|
||||
)
|
||||
return payload_user_id
|
||||
from litellm.proxy._experimental.mcp_server.auth.user_api_key_auth_mcp import (
|
||||
MCPRequestHandler,
|
||||
)
|
||||
|
||||
resolved = await MCPRequestHandler._resolve_single_user_id_by_email(payload_user_email or "")
|
||||
if resolved is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail={"error": "user_email did not resolve to exactly one user."},
|
||||
)
|
||||
return resolved
|
||||
|
||||
@router.post(
|
||||
"/delegation",
|
||||
description="Grant consent for an agent to act on behalf of a user on MCP routes",
|
||||
status_code=status.HTTP_201_CREATED,
|
||||
)
|
||||
@management_endpoint_wrapper
|
||||
async def grant_delegation(
|
||||
payload: NewUserAgentDelegationRequest,
|
||||
user_api_key_dict: UserAPIKeyAuth = _delegation_user_auth,
|
||||
litellm_changed_by: str | None = _delegation_changed_by,
|
||||
):
|
||||
prisma_client = get_prisma_client_or_throw("Database not connected. Connect a database to your proxy")
|
||||
target_user_id = await _authorized_delegation_target_user_id(
|
||||
prisma_client, user_api_key_dict, payload.user_id, payload.user_email, "grant"
|
||||
)
|
||||
|
||||
agent_row = await prisma_client.db.litellm_agentstable.find_unique(
|
||||
where={"agent_id": payload.agent_id}, include={"object_permission": True}
|
||||
)
|
||||
if agent_row is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail={"error": f"Agent '{payload.agent_id}' not found."},
|
||||
)
|
||||
if not getattr(getattr(agent_row, "object_permission", None), "mcp_can_delegate", False):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail={
|
||||
"error": f"Agent '{payload.agent_id}' does not have mcp_can_delegate; grant the capability before consenting."
|
||||
},
|
||||
)
|
||||
|
||||
granted_by = (
|
||||
get_audit_log_changed_by(
|
||||
litellm_changed_by=litellm_changed_by,
|
||||
user_api_key_dict=user_api_key_dict,
|
||||
litellm_proxy_admin_name=LITELLM_PROXY_ADMIN_NAME,
|
||||
)
|
||||
or LITELLM_PROXY_ADMIN_NAME
|
||||
)
|
||||
return await grant_user_agent_delegation(
|
||||
prisma_client, user_id=target_user_id, agent_id=payload.agent_id, granted_by=granted_by
|
||||
)
|
||||
|
||||
@router.post(
|
||||
"/delegation/revoke",
|
||||
description="Revoke an agent's consent to act on behalf of a user",
|
||||
)
|
||||
@management_endpoint_wrapper
|
||||
async def revoke_delegation(
|
||||
payload: RevokeUserAgentDelegationRequest,
|
||||
user_api_key_dict: UserAPIKeyAuth = _delegation_user_auth,
|
||||
litellm_changed_by: str | None = _delegation_changed_by,
|
||||
):
|
||||
prisma_client = get_prisma_client_or_throw("Database not connected. Connect a database to your proxy")
|
||||
target_user_id = await _authorized_delegation_target_user_id(
|
||||
prisma_client, user_api_key_dict, payload.user_id, None, "revoke"
|
||||
)
|
||||
|
||||
revoked_by = (
|
||||
get_audit_log_changed_by(
|
||||
litellm_changed_by=litellm_changed_by,
|
||||
user_api_key_dict=user_api_key_dict,
|
||||
litellm_proxy_admin_name=LITELLM_PROXY_ADMIN_NAME,
|
||||
)
|
||||
or LITELLM_PROXY_ADMIN_NAME
|
||||
)
|
||||
revoked = await revoke_user_agent_delegation(
|
||||
prisma_client, user_id=target_user_id, agent_id=payload.agent_id, revoked_by=revoked_by
|
||||
)
|
||||
if revoked is None:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail={"error": "No delegation found for that user and agent."},
|
||||
)
|
||||
return revoked
|
||||
|
||||
@router.get(
|
||||
"/delegation",
|
||||
description="List delegation consent records (admins see all; users see their own)",
|
||||
)
|
||||
@management_endpoint_wrapper
|
||||
async def list_delegations(
|
||||
user_id: str | None = None,
|
||||
user_api_key_dict: UserAPIKeyAuth = _delegation_user_auth,
|
||||
):
|
||||
prisma_client = get_prisma_client_or_throw("Database not connected. Connect a database to your proxy")
|
||||
if LitellmUserRoles.PROXY_ADMIN == user_api_key_dict.user_role:
|
||||
return await list_user_agent_delegations(prisma_client, user_id=user_id)
|
||||
# A non-admin caller may only see its own consents. A key with no user
|
||||
# association (e.g. a team key, user_id=None) owns none; return empty
|
||||
# rather than passing None through, which the store would read as an
|
||||
# unscoped "list all" and leak every user's records.
|
||||
if not user_api_key_dict.user_id:
|
||||
return []
|
||||
return await list_user_agent_delegations(prisma_client, user_id=user_api_key_dict.user_id)
|
||||
|
|
|
|||
|
|
@ -280,6 +280,7 @@ model LiteLLM_ObjectPermissionTable {
|
|||
mcp_toolsets String[] @default([]) // Toolset IDs granted to this key/team/user
|
||||
search_tools String[] @default([]) // search_tool_name values this key/team/user may call
|
||||
mcp_tool_search_enabled Boolean?
|
||||
mcp_can_delegate Boolean? // Agents with this permission may act on behalf of consenting users on MCP routes
|
||||
teams LiteLLM_TeamTable[]
|
||||
projects LiteLLM_ProjectTable[]
|
||||
verification_tokens LiteLLM_VerificationToken[]
|
||||
|
|
@ -369,6 +370,20 @@ model LiteLLM_MCPToolsetTable {
|
|||
updated_by String?
|
||||
}
|
||||
|
||||
// User consent records authorizing an agent to act on the user's behalf on MCP routes
|
||||
model LiteLLM_UserAgentDelegationTable {
|
||||
delegation_id String @id @default(uuid())
|
||||
user_id String
|
||||
agent_id String
|
||||
granted_at DateTime @default(now())
|
||||
granted_by String
|
||||
revoked_at DateTime?
|
||||
revoked_by String?
|
||||
updated_at DateTime @default(now()) @updatedAt
|
||||
|
||||
@@unique([user_id, agent_id])
|
||||
}
|
||||
|
||||
// Per-user BYOK credentials for MCP servers
|
||||
model LiteLLM_MCPUserCredentials {
|
||||
id String @id @default(uuid())
|
||||
|
|
|
|||
|
|
@ -113,6 +113,10 @@ class MCPToolsetRepository(PrismaTableRepository):
|
|||
table_name = "litellm_mcptoolsettable"
|
||||
|
||||
|
||||
class UserAgentDelegationRepository(PrismaTableRepository):
|
||||
table_name = "litellm_useragentdelegationtable"
|
||||
|
||||
|
||||
class ToolRepository(PrismaTableRepository):
|
||||
table_name = "litellm_tooltable"
|
||||
|
||||
|
|
|
|||
|
|
@ -174,6 +174,7 @@ class AgentObjectPermission(TypedDict, total=False):
|
|||
mcp_tool_permissions: Optional[Dict[str, List[str]]]
|
||||
models: Optional[List[str]]
|
||||
agents: Optional[List[str]]
|
||||
mcp_can_delegate: Optional[bool]
|
||||
|
||||
|
||||
class AgentConfig(TypedDict, total=False):
|
||||
|
|
|
|||
40
litellm/types/mcp_server/user_agent_delegation.py
Normal file
40
litellm/types/mcp_server/user_agent_delegation.py
Normal file
|
|
@ -0,0 +1,40 @@
|
|||
from datetime import datetime
|
||||
|
||||
from pydantic import BaseModel, field_validator, model_validator
|
||||
|
||||
|
||||
class UserAgentDelegation(BaseModel):
|
||||
delegation_id: str
|
||||
user_id: str
|
||||
agent_id: str
|
||||
granted_at: datetime
|
||||
granted_by: str
|
||||
revoked_at: datetime | None = None
|
||||
revoked_by: str | None = None
|
||||
|
||||
|
||||
def _reject_blank(value: str | None) -> str | None:
|
||||
if value is not None and not value.strip():
|
||||
raise ValueError("must not be blank")
|
||||
return value
|
||||
|
||||
|
||||
class NewUserAgentDelegationRequest(BaseModel):
|
||||
user_id: str | None = None
|
||||
user_email: str | None = None
|
||||
agent_id: str
|
||||
|
||||
_no_blanks = field_validator("user_id", "user_email", "agent_id")(_reject_blank)
|
||||
|
||||
@model_validator(mode="after")
|
||||
def exactly_one_target(self) -> "NewUserAgentDelegationRequest":
|
||||
if bool(self.user_id) == bool(self.user_email):
|
||||
raise ValueError("provide exactly one of user_id or user_email")
|
||||
return self
|
||||
|
||||
|
||||
class RevokeUserAgentDelegationRequest(BaseModel):
|
||||
user_id: str
|
||||
agent_id: str
|
||||
|
||||
_no_blanks = field_validator("user_id", "agent_id")(_reject_blank)
|
||||
|
|
@ -25,3 +25,4 @@ class ObjectPermissionDict(TypedDict, total=False):
|
|||
models: Optional[list[str]]
|
||||
search_tools: Optional[list[str]]
|
||||
mcp_tool_search_enabled: Optional[bool]
|
||||
mcp_can_delegate: Optional[bool]
|
||||
|
|
|
|||
|
|
@ -280,6 +280,7 @@ model LiteLLM_ObjectPermissionTable {
|
|||
mcp_toolsets String[] @default([]) // Toolset IDs granted to this key/team/user
|
||||
search_tools String[] @default([]) // search_tool_name values this key/team/user may call
|
||||
mcp_tool_search_enabled Boolean?
|
||||
mcp_can_delegate Boolean? // Agents with this permission may act on behalf of consenting users on MCP routes
|
||||
teams LiteLLM_TeamTable[]
|
||||
projects LiteLLM_ProjectTable[]
|
||||
verification_tokens LiteLLM_VerificationToken[]
|
||||
|
|
@ -369,6 +370,20 @@ model LiteLLM_MCPToolsetTable {
|
|||
updated_by String?
|
||||
}
|
||||
|
||||
// User consent records authorizing an agent to act on the user's behalf on MCP routes
|
||||
model LiteLLM_UserAgentDelegationTable {
|
||||
delegation_id String @id @default(uuid())
|
||||
user_id String
|
||||
agent_id String
|
||||
granted_at DateTime @default(now())
|
||||
granted_by String
|
||||
revoked_at DateTime?
|
||||
revoked_by String?
|
||||
updated_at DateTime @default(now()) @updatedAt
|
||||
|
||||
@@unique([user_id, agent_id])
|
||||
}
|
||||
|
||||
// Per-user BYOK credentials for MCP servers
|
||||
model LiteLLM_MCPUserCredentials {
|
||||
id String @id @default(uuid())
|
||||
|
|
|
|||
|
|
@ -6131,3 +6131,214 @@ class TestMCPDcrBridgeDelegateAdmission:
|
|||
route="/mcp/bridge_delegate_server",
|
||||
)
|
||||
assert exc_info.value.status_code == 500
|
||||
|
||||
|
||||
class TestMCPUserDelegation:
|
||||
"""The x-litellm-delegated-user assertion ladder: untrusted input that grants
|
||||
nothing without the flag, an agent binding, the agent capability, and an
|
||||
active consent record; absent header = byte-identical behavior."""
|
||||
|
||||
def _headers(self, email=None):
|
||||
return Headers({"x-litellm-delegated-user": email} if email else {})
|
||||
|
||||
def _agent_auth(self, agent_id="agent-1"):
|
||||
return UserAPIKeyAuth(api_key="test-key", user_id="agent-svc-user", agent_id=agent_id)
|
||||
|
||||
def _general_settings(self, enabled):
|
||||
return {"mcp_user_delegation_enabled": enabled}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_no_header_returns_auth_unchanged(self):
|
||||
auth = self._agent_auth()
|
||||
result = await MCPRequestHandler.resolve_delegated_user_auth(auth, self._headers())
|
||||
assert result is auth
|
||||
assert result.delegated_user_id is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_header_with_flag_disabled_rejected(self):
|
||||
with patch("litellm.proxy.proxy_server.general_settings", self._general_settings(False)):
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
await MCPRequestHandler.resolve_delegated_user_auth(
|
||||
self._agent_auth(), self._headers("alice@example.com")
|
||||
)
|
||||
assert exc_info.value.status_code == 403
|
||||
assert "mcp_user_delegation_enabled" in str(exc_info.value.detail)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_header_on_key_without_agent_binding_rejected(self):
|
||||
auth = UserAPIKeyAuth(api_key="test-key", user_id="someone")
|
||||
with patch("litellm.proxy.proxy_server.general_settings", self._general_settings(True)):
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
await MCPRequestHandler.resolve_delegated_user_auth(auth, self._headers("alice@example.com"))
|
||||
assert exc_info.value.status_code == 403
|
||||
assert "agent" in str(exc_info.value.detail).lower()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_header_on_anonymous_auth_rejected(self):
|
||||
with patch("litellm.proxy.proxy_server.general_settings", self._general_settings(True)):
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
await MCPRequestHandler.resolve_delegated_user_auth(None, self._headers("alice@example.com"))
|
||||
assert exc_info.value.status_code == 403
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_agent_without_can_delegate_rejected(self):
|
||||
agent_op = MagicMock()
|
||||
agent_op.mcp_can_delegate = None
|
||||
with (
|
||||
patch("litellm.proxy.proxy_server.general_settings", self._general_settings(True)),
|
||||
patch.object(MCPRequestHandler, "_get_agent_object_permission", AsyncMock(return_value=agent_op)),
|
||||
):
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
await MCPRequestHandler.resolve_delegated_user_auth(
|
||||
self._agent_auth(), self._headers("alice@example.com")
|
||||
)
|
||||
assert exc_info.value.status_code == 403
|
||||
assert "mcp_can_delegate" in str(exc_info.value.detail)
|
||||
|
||||
def _capable_agent_op(self):
|
||||
agent_op = MagicMock()
|
||||
agent_op.mcp_can_delegate = True
|
||||
return agent_op
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_unknown_asserted_user_rejected_uniformly(self):
|
||||
with (
|
||||
patch("litellm.proxy.proxy_server.general_settings", self._general_settings(True)),
|
||||
patch("litellm.proxy.proxy_server.prisma_client", MagicMock()),
|
||||
patch.object(
|
||||
MCPRequestHandler, "_get_agent_object_permission", AsyncMock(return_value=self._capable_agent_op())
|
||||
),
|
||||
patch.object(MCPRequestHandler, "_resolve_single_user_id_by_email", AsyncMock(return_value=None)),
|
||||
):
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
await MCPRequestHandler.resolve_delegated_user_auth(
|
||||
self._agent_auth(), self._headers("nobody@example.com")
|
||||
)
|
||||
assert exc_info.value.status_code == 403
|
||||
assert "not authorized" in str(exc_info.value.detail)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_missing_consent_rejected_with_same_error_as_unknown_user(self):
|
||||
with (
|
||||
patch("litellm.proxy.proxy_server.general_settings", self._general_settings(True)),
|
||||
patch("litellm.proxy.proxy_server.prisma_client", MagicMock()),
|
||||
patch.object(
|
||||
MCPRequestHandler, "_get_agent_object_permission", AsyncMock(return_value=self._capable_agent_op())
|
||||
),
|
||||
patch.object(MCPRequestHandler, "_resolve_single_user_id_by_email", AsyncMock(return_value="alice-id")),
|
||||
patch(
|
||||
"litellm.proxy._experimental.mcp_server.delegation_db.get_active_user_agent_delegation",
|
||||
AsyncMock(return_value=None),
|
||||
),
|
||||
):
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
await MCPRequestHandler.resolve_delegated_user_auth(
|
||||
self._agent_auth(), self._headers("alice@example.com")
|
||||
)
|
||||
assert exc_info.value.status_code == 403
|
||||
assert "not authorized" in str(exc_info.value.detail)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_active_consent_stamps_delegated_user_id(self):
|
||||
delegation = MagicMock()
|
||||
auth = self._agent_auth()
|
||||
with (
|
||||
patch("litellm.proxy.proxy_server.general_settings", self._general_settings(True)),
|
||||
patch("litellm.proxy.proxy_server.prisma_client", MagicMock()),
|
||||
patch.object(
|
||||
MCPRequestHandler, "_get_agent_object_permission", AsyncMock(return_value=self._capable_agent_op())
|
||||
),
|
||||
patch.object(MCPRequestHandler, "_resolve_single_user_id_by_email", AsyncMock(return_value="alice-id")),
|
||||
patch(
|
||||
"litellm.proxy._experimental.mcp_server.delegation_db.get_active_user_agent_delegation",
|
||||
AsyncMock(return_value=delegation),
|
||||
),
|
||||
):
|
||||
result = await MCPRequestHandler.resolve_delegated_user_auth(auth, self._headers("alice@example.com"))
|
||||
|
||||
assert result is not None
|
||||
assert result.delegated_user_id == "alice-id"
|
||||
assert result.user_id == "agent-svc-user"
|
||||
assert auth.delegated_user_id is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_email_lookup_fails_closed_on_ambiguity(self):
|
||||
two_rows = [MagicMock(user_id="u1"), MagicMock(user_id="u2")]
|
||||
mock_prisma = MagicMock()
|
||||
mock_table = MagicMock()
|
||||
mock_table.find_many = AsyncMock(return_value=two_rows)
|
||||
with (
|
||||
patch("litellm.proxy.proxy_server.prisma_client", mock_prisma),
|
||||
patch(
|
||||
"litellm.repositories.user_repository.UserRepository",
|
||||
MagicMock(return_value=MagicMock(table=mock_table)),
|
||||
),
|
||||
):
|
||||
resolved = await MCPRequestHandler._resolve_single_user_id_by_email("dupe@example.com")
|
||||
assert resolved is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_email_lookup_single_match_case_insensitive(self):
|
||||
one_row = [MagicMock(user_id="alice-id")]
|
||||
mock_table = MagicMock()
|
||||
mock_table.find_many = AsyncMock(return_value=one_row)
|
||||
with (
|
||||
patch("litellm.proxy.proxy_server.prisma_client", MagicMock()),
|
||||
patch(
|
||||
"litellm.repositories.user_repository.UserRepository",
|
||||
MagicMock(return_value=MagicMock(table=mock_table)),
|
||||
),
|
||||
):
|
||||
resolved = await MCPRequestHandler._resolve_single_user_id_by_email("Alice@Example.com")
|
||||
assert resolved == "alice-id"
|
||||
where = mock_table.find_many.await_args.kwargs["where"]
|
||||
assert where["user_email"]["mode"] == "insensitive"
|
||||
|
||||
|
||||
class TestMCPUserDelegationWiring:
|
||||
"""Pins the SEAMS: admission calls the delegation resolver and uses its
|
||||
result. The resolver's own ladder is covered above; these fail if the call
|
||||
site is deleted."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_process_mcp_request_routes_auth_through_delegation_resolver(self):
|
||||
scope = {
|
||||
"type": "http",
|
||||
"method": "POST",
|
||||
"path": "/mcp",
|
||||
"headers": [
|
||||
(b"x-litellm-api-key", b"Bearer sk-agent"),
|
||||
(b"x-litellm-delegated-user", b"alice@example.com"),
|
||||
],
|
||||
}
|
||||
admitted = UserAPIKeyAuth(api_key="hashed", user_id="agent-svc-user", agent_id="agent-1")
|
||||
delegated = admitted.model_copy(update={"delegated_user_id": "alice-id"})
|
||||
|
||||
async def mock_user_api_key_auth(api_key, request):
|
||||
return admitted
|
||||
|
||||
resolve_mock = AsyncMock(return_value=delegated)
|
||||
with (
|
||||
patch(
|
||||
"litellm.proxy._experimental.mcp_server.auth.user_api_key_auth_mcp.user_api_key_auth",
|
||||
side_effect=mock_user_api_key_auth,
|
||||
),
|
||||
patch.object(MCPRequestHandler, "resolve_delegated_user_auth", resolve_mock),
|
||||
):
|
||||
auth_result, *_ = await MCPRequestHandler.process_mcp_request(scope)
|
||||
|
||||
assert auth_result is delegated
|
||||
assert resolve_mock.await_args.args[0] is admitted
|
||||
assert resolve_mock.await_args.args[1].get("x-litellm-delegated-user") == "alice@example.com"
|
||||
|
||||
|
||||
def test_object_permission_table_model_surfaces_mcp_can_delegate():
|
||||
"""The read model returned by get_object_permission must carry the field, or
|
||||
the capability check silently reads a missing attribute as False even when
|
||||
the DB column is true. Both the base and the table model must declare it."""
|
||||
from litellm.models.object_permission import LiteLLM_ObjectPermissionTable
|
||||
from litellm.proxy._types import LiteLLM_ObjectPermissionBase
|
||||
|
||||
assert "mcp_can_delegate" in LiteLLM_ObjectPermissionBase.model_fields
|
||||
op = LiteLLM_ObjectPermissionTable(object_permission_id="x", mcp_can_delegate=True)
|
||||
assert op.mcp_can_delegate is True
|
||||
|
|
|
|||
|
|
@ -267,13 +267,22 @@ def test_to_subject_unauthenticated_is_empty_with_inbound_token():
|
|||
|
||||
|
||||
def test_to_subject_maps_principal_fields():
|
||||
principal = SimpleNamespace(org_id="org1", team_id="team1", user_id="user1")
|
||||
principal = SimpleNamespace(org_id="org1", team_id="team1", user_id="user1", delegated_user_id=None)
|
||||
subject = to_subject(principal, None)
|
||||
assert subject.tenant_id == "org1"
|
||||
assert subject.subject_id == "user1"
|
||||
assert subject.inbound_token is None
|
||||
|
||||
|
||||
def test_to_subject_delegated_user_replaces_credential_subject():
|
||||
principal = SimpleNamespace(
|
||||
org_id=None, team_id="team1", user_id="agent-svc-user", delegated_user_id="delegated-user"
|
||||
)
|
||||
subject = to_subject(principal, None)
|
||||
assert subject.subject_id == "delegated-user"
|
||||
assert subject.tenant_id == "team1"
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"error, status",
|
||||
[
|
||||
|
|
|
|||
|
|
@ -0,0 +1,184 @@
|
|||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from litellm.caching.caching import DualCache
|
||||
from litellm.proxy._experimental.mcp_server import delegation_db
|
||||
|
||||
|
||||
def _mock_prisma_returning(rows):
|
||||
"""A prisma client whose delegation table.find_first pops from `rows`."""
|
||||
table = MagicMock()
|
||||
table.find_first = AsyncMock(side_effect=list(rows))
|
||||
client = MagicMock()
|
||||
client.db.litellm_useragentdelegationtable = table
|
||||
return client, table
|
||||
|
||||
|
||||
def _row(user_id="alice", agent_id="agent-1", revoked_at=None):
|
||||
# revoked_at is a real attribute (not an auto-Mock) because production reads
|
||||
# row.revoked_at to decide active vs already-revoked.
|
||||
return MagicMock(
|
||||
revoked_at=revoked_at,
|
||||
model_dump=lambda: {
|
||||
"delegation_id": "d1",
|
||||
"user_id": user_id,
|
||||
"agent_id": agent_id,
|
||||
"granted_at": "2026-07-17T00:00:00Z",
|
||||
"granted_by": "admin",
|
||||
"revoked_at": revoked_at,
|
||||
"revoked_by": None,
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
class TestActiveDelegationCache:
|
||||
@pytest.mark.asyncio
|
||||
async def test_positive_result_served_from_cache_second_call(self):
|
||||
client, table = _mock_prisma_returning([_row()])
|
||||
with patch("litellm.proxy.proxy_server.user_api_key_cache", DualCache()):
|
||||
first = await delegation_db.get_active_user_agent_delegation(client, "alice", "agent-1")
|
||||
second = await delegation_db.get_active_user_agent_delegation(client, "alice", "agent-1")
|
||||
assert first is not None and second is not None
|
||||
assert second.user_id == "alice"
|
||||
table.find_first.assert_awaited_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_negative_result_cached_so_repeated_denial_skips_db(self):
|
||||
client, table = _mock_prisma_returning([None])
|
||||
with patch("litellm.proxy.proxy_server.user_api_key_cache", DualCache()):
|
||||
first = await delegation_db.get_active_user_agent_delegation(client, "bob", "agent-1")
|
||||
second = await delegation_db.get_active_user_agent_delegation(client, "bob", "agent-1")
|
||||
assert first is None and second is None
|
||||
table.find_first.assert_awaited_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_revoke_busts_cache_so_next_read_hits_db_and_denies(self):
|
||||
"""Immediate revocation: after a cached positive, revoke must invalidate
|
||||
the entry so the next resolve fails closed rather than serving stale
|
||||
consent until TTL."""
|
||||
cache = DualCache()
|
||||
active_table = MagicMock()
|
||||
active_table.find_first = AsyncMock(side_effect=[_row(), None])
|
||||
active_table.find_unique = AsyncMock(return_value=_row())
|
||||
active_table.update = AsyncMock(
|
||||
return_value=MagicMock(
|
||||
model_dump=lambda: {
|
||||
"delegation_id": "d1",
|
||||
"user_id": "alice",
|
||||
"agent_id": "agent-1",
|
||||
"granted_at": "2026-07-17T00:00:00Z",
|
||||
"granted_by": "admin",
|
||||
"revoked_at": "2026-07-17T01:00:00Z",
|
||||
"revoked_by": "admin",
|
||||
}
|
||||
)
|
||||
)
|
||||
client = MagicMock()
|
||||
client.db.litellm_useragentdelegationtable = active_table
|
||||
with patch("litellm.proxy.proxy_server.user_api_key_cache", cache):
|
||||
assert await delegation_db.get_active_user_agent_delegation(client, "alice", "agent-1") is not None
|
||||
await delegation_db.revoke_user_agent_delegation(client, "alice", "agent-1", revoked_by="admin")
|
||||
after = await delegation_db.get_active_user_agent_delegation(client, "alice", "agent-1")
|
||||
assert after is None
|
||||
assert active_table.find_first.await_count == 2
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_grant_busts_negative_cache_so_new_consent_is_immediate(self):
|
||||
cache = DualCache()
|
||||
table = MagicMock()
|
||||
table.find_first = AsyncMock(side_effect=[None, _row()])
|
||||
table.upsert = AsyncMock(return_value=_row())
|
||||
client = MagicMock()
|
||||
client.db.litellm_useragentdelegationtable = table
|
||||
with patch("litellm.proxy.proxy_server.user_api_key_cache", cache):
|
||||
assert await delegation_db.get_active_user_agent_delegation(client, "alice", "agent-1") is None
|
||||
await delegation_db.grant_user_agent_delegation(client, "alice", "agent-1", granted_by="admin")
|
||||
after = await delegation_db.get_active_user_agent_delegation(client, "alice", "agent-1")
|
||||
assert after is not None
|
||||
|
||||
|
||||
class TestListDelegationScoping:
|
||||
"""A non-admin key with no user association must not fall through to the
|
||||
unscoped 'list all' view (which would leak every user's consent records)."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_all_with_none_user_id_returns_all_rows(self):
|
||||
"""Pins the footgun the endpoint guard defends against: the store treats
|
||||
user_id=None as no filter (admin-only view)."""
|
||||
table = MagicMock()
|
||||
table.find_many = AsyncMock(return_value=[_row("alice"), _row("bob")])
|
||||
client = MagicMock()
|
||||
client.db.litellm_useragentdelegationtable = table
|
||||
result = await delegation_db.list_user_agent_delegations(client, user_id=None)
|
||||
assert len(result) == 2
|
||||
assert table.find_many.await_args.kwargs["where"] == {}
|
||||
|
||||
|
||||
class TestDelegationRequestValidation:
|
||||
def test_grant_request_rejects_both_targets(self):
|
||||
import pytest
|
||||
from pydantic import ValidationError
|
||||
|
||||
from litellm.types.mcp_server.user_agent_delegation import NewUserAgentDelegationRequest
|
||||
|
||||
with pytest.raises(ValidationError):
|
||||
NewUserAgentDelegationRequest(user_id="u1", user_email="u@x.com", agent_id="a1")
|
||||
|
||||
def test_grant_request_rejects_neither_target(self):
|
||||
import pytest
|
||||
from pydantic import ValidationError
|
||||
|
||||
from litellm.types.mcp_server.user_agent_delegation import NewUserAgentDelegationRequest
|
||||
|
||||
with pytest.raises(ValidationError):
|
||||
NewUserAgentDelegationRequest(agent_id="a1")
|
||||
|
||||
def test_grant_request_rejects_blank_agent(self):
|
||||
import pytest
|
||||
from pydantic import ValidationError
|
||||
|
||||
from litellm.types.mcp_server.user_agent_delegation import NewUserAgentDelegationRequest
|
||||
|
||||
with pytest.raises(ValidationError):
|
||||
NewUserAgentDelegationRequest(user_id="u1", agent_id=" ")
|
||||
|
||||
def test_grant_request_accepts_single_target(self):
|
||||
from litellm.types.mcp_server.user_agent_delegation import NewUserAgentDelegationRequest
|
||||
|
||||
assert NewUserAgentDelegationRequest(user_email="u@x.com", agent_id="a1").agent_id == "a1"
|
||||
|
||||
|
||||
class TestRevokeAuditIntegrity:
|
||||
@pytest.mark.asyncio
|
||||
async def test_revoke_already_revoked_returns_none_and_does_not_restamp(self):
|
||||
"""Re-revoking an already-revoked consent must be a no-op: return None
|
||||
(so the endpoint 404s) and never overwrite the original revoked_at/by."""
|
||||
already_revoked = MagicMock(revoked_at="2026-07-17T00:00:00Z")
|
||||
table = MagicMock()
|
||||
table.find_unique = AsyncMock(return_value=already_revoked)
|
||||
table.update = AsyncMock()
|
||||
client = MagicMock()
|
||||
client.db.litellm_useragentdelegationtable = table
|
||||
|
||||
with patch("litellm.proxy.proxy_server.user_api_key_cache", DualCache()):
|
||||
result = await delegation_db.revoke_user_agent_delegation(
|
||||
client, "alice", "agent-1", revoked_by="admin2"
|
||||
)
|
||||
|
||||
assert result is None
|
||||
table.update.assert_not_awaited()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_revoke_nonexistent_returns_none(self):
|
||||
table = MagicMock()
|
||||
table.find_unique = AsyncMock(return_value=None)
|
||||
table.update = AsyncMock()
|
||||
client = MagicMock()
|
||||
client.db.litellm_useragentdelegationtable = table
|
||||
|
||||
with patch("litellm.proxy.proxy_server.user_api_key_cache", DualCache()):
|
||||
result = await delegation_db.revoke_user_agent_delegation(client, "ghost", "agent-1", revoked_by="a")
|
||||
|
||||
assert result is None
|
||||
table.update.assert_not_awaited()
|
||||
|
|
@ -2699,9 +2699,7 @@ class TestRestListToolsetFiltering:
|
|||
|
||||
mock_manager = MagicMock()
|
||||
mock_manager.expand_tool_permissions = MagicMock(side_effect=lambda perms: perms or {})
|
||||
mock_manager.resolve_toolset_tool_permissions = AsyncMock(
|
||||
return_value={"server-a": ["lookup_status"]}
|
||||
)
|
||||
mock_manager.resolve_toolset_tool_permissions = AsyncMock(return_value={"server-a": ["lookup_status"]})
|
||||
|
||||
monkeypatch.setattr(
|
||||
rest_endpoints.global_mcp_server_manager,
|
||||
|
|
@ -2725,3 +2723,91 @@ class TestRestListToolsetFiltering:
|
|||
)
|
||||
|
||||
assert [tool.name for tool in result] == ["lookup_status"]
|
||||
|
||||
|
||||
class TestRestDelegationWiring:
|
||||
"""Pins that both REST runtime endpoints route auth through the delegation
|
||||
resolver before any permission or credential resolution."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_tools_uses_delegation_resolved_auth(self, monkeypatch):
|
||||
from unittest.mock import patch
|
||||
|
||||
original = UserAPIKeyAuth(api_key="hashed", user_id="agent-svc-user", agent_id="agent-1")
|
||||
delegated = original.model_copy(update={"delegated_user_id": "alice-id"})
|
||||
seen_auth = {}
|
||||
|
||||
async def fake_contexts(auth):
|
||||
seen_auth["auth"] = auth
|
||||
return [auth]
|
||||
|
||||
async def fake_get_allowed(user_api_key_auth=None):
|
||||
return []
|
||||
|
||||
monkeypatch.setattr(rest_endpoints, "build_effective_auth_contexts", fake_contexts, raising=False)
|
||||
monkeypatch.setattr(
|
||||
rest_endpoints.global_mcp_server_manager,
|
||||
"get_allowed_mcp_servers",
|
||||
fake_get_allowed,
|
||||
raising=False,
|
||||
)
|
||||
|
||||
request = _build_request(
|
||||
path="/mcp-rest/tools/list",
|
||||
method="GET",
|
||||
headers={"x-litellm-delegated-user": "alice@example.com"},
|
||||
)
|
||||
with patch.object(auth_mcp.MCPRequestHandler, "resolve_delegated_user_auth", AsyncMock(return_value=delegated)):
|
||||
await rest_endpoints.list_tool_rest_api(request, user_api_key_dict=original)
|
||||
|
||||
assert seen_auth["auth"] is delegated
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_call_tool_rejection_propagates_before_any_resolution(self):
|
||||
from unittest.mock import patch
|
||||
|
||||
from fastapi import HTTPException
|
||||
|
||||
original = UserAPIKeyAuth(api_key="hashed", user_id="agent-svc-user", agent_id="agent-1")
|
||||
sentinel = HTTPException(status_code=403, detail={"error": "delegation-sentinel"})
|
||||
|
||||
request = _build_request(
|
||||
path="/mcp-rest/tools/call",
|
||||
method="POST",
|
||||
json_body={"server_id": "server-1", "name": "lookup_status", "arguments": {}},
|
||||
headers={"x-litellm-delegated-user": "alice@example.com"},
|
||||
)
|
||||
with patch.object(auth_mcp.MCPRequestHandler, "resolve_delegated_user_auth", AsyncMock(side_effect=sentinel)):
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
await rest_endpoints.call_tool_rest_api(request, user_api_key_dict=original)
|
||||
|
||||
assert exc_info.value.status_code == 403
|
||||
assert "delegation-sentinel" in str(exc_info.value.detail)
|
||||
|
||||
|
||||
class TestRestListDelegationPropagates:
|
||||
@pytest.mark.asyncio
|
||||
async def test_list_tools_delegation_403_is_not_swallowed_into_200(self):
|
||||
"""A delegation rejection on tools/list must surface as a 403, not be
|
||||
caught by the endpoint's broad except-HTTPException handler and returned
|
||||
as a 200 empty-tools response."""
|
||||
from unittest.mock import patch
|
||||
|
||||
from fastapi import HTTPException
|
||||
|
||||
original = UserAPIKeyAuth(api_key="hashed", user_id="agent-svc-user", agent_id="agent-1")
|
||||
sentinel = HTTPException(status_code=403, detail={"error": "delegation-denied"})
|
||||
|
||||
request = _build_request(
|
||||
path="/mcp-rest/tools/list",
|
||||
method="GET",
|
||||
headers={"x-litellm-delegated-user": "alice@example.com"},
|
||||
)
|
||||
with patch.object(
|
||||
auth_mcp.MCPRequestHandler, "resolve_delegated_user_auth", AsyncMock(side_effect=sentinel)
|
||||
):
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
await rest_endpoints.list_tool_rest_api(request, user_api_key_dict=original)
|
||||
|
||||
assert exc_info.value.status_code == 403
|
||||
assert "delegation-denied" in str(exc_info.value.detail)
|
||||
|
|
|
|||
|
|
@ -719,6 +719,7 @@ _EXPECTED_CUSTOMER = {
|
|||
"blocked_tools": [],
|
||||
"search_tools": [],
|
||||
"mcp_tool_search_enabled": None,
|
||||
"mcp_can_delegate": None,
|
||||
},
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -5376,3 +5376,155 @@ async def test_edit_mcp_server_snapshot_failure_skips_purge_but_edit_succeeds():
|
|||
|
||||
assert result.server_id == server_id
|
||||
mock_purge.assert_not_awaited()
|
||||
|
||||
|
||||
class TestDelegationListScoping:
|
||||
"""GET /v1/mcp/delegation must not leak other users' consent records to a
|
||||
non-admin key with no user association (user_id=None)."""
|
||||
|
||||
def _client_and_table(self, auth):
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
|
||||
|
||||
table = MagicMock()
|
||||
table.find_many = AsyncMock(return_value=[])
|
||||
prisma = MagicMock()
|
||||
prisma.db.litellm_useragentdelegationtable = table
|
||||
|
||||
client = create_mcp_router_test_client()
|
||||
client.app.dependency_overrides[user_api_key_auth] = lambda: auth
|
||||
return client, table, prisma
|
||||
|
||||
def test_non_admin_key_without_user_id_never_queries_and_returns_empty(self):
|
||||
from unittest.mock import patch
|
||||
|
||||
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
|
||||
|
||||
auth = UserAPIKeyAuth(api_key="team-key", user_role=LitellmUserRoles.INTERNAL_USER, user_id=None)
|
||||
client, table, prisma = self._client_and_table(auth)
|
||||
|
||||
with patch(
|
||||
"litellm.proxy.management_endpoints.mcp_management_endpoints.get_prisma_client_or_throw",
|
||||
return_value=prisma,
|
||||
):
|
||||
resp = client.get("/v1/mcp/delegation")
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert resp.json() == []
|
||||
table.find_many.assert_not_awaited()
|
||||
|
||||
def test_non_admin_key_with_user_id_is_scoped_to_that_user(self):
|
||||
from unittest.mock import patch
|
||||
|
||||
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
|
||||
|
||||
auth = UserAPIKeyAuth(api_key="user-key", user_role=LitellmUserRoles.INTERNAL_USER, user_id="alice")
|
||||
client, table, prisma = self._client_and_table(auth)
|
||||
|
||||
with patch(
|
||||
"litellm.proxy.management_endpoints.mcp_management_endpoints.get_prisma_client_or_throw",
|
||||
return_value=prisma,
|
||||
):
|
||||
resp = client.get("/v1/mcp/delegation")
|
||||
|
||||
assert resp.status_code == 200
|
||||
table.find_many.assert_awaited_once()
|
||||
assert table.find_many.await_args.kwargs["where"] == {"user_id": "alice"}
|
||||
|
||||
def test_admin_key_lists_all_unscoped(self):
|
||||
from unittest.mock import patch
|
||||
|
||||
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
|
||||
|
||||
auth = UserAPIKeyAuth(api_key="admin-key", user_role=LitellmUserRoles.PROXY_ADMIN, user_id=None)
|
||||
client, table, prisma = self._client_and_table(auth)
|
||||
|
||||
with patch(
|
||||
"litellm.proxy.management_endpoints.mcp_management_endpoints.get_prisma_client_or_throw",
|
||||
return_value=prisma,
|
||||
):
|
||||
resp = client.get("/v1/mcp/delegation")
|
||||
|
||||
assert resp.status_code == 200
|
||||
table.find_many.assert_awaited_once()
|
||||
assert table.find_many.await_args.kwargs["where"] == {}
|
||||
|
||||
|
||||
class TestDelegationGrantAuthz:
|
||||
"""The consolidated authz gate: non-admins act only on themselves and never
|
||||
trigger an email lookup (no enumeration oracle); admin user_id targets are
|
||||
checked for existence so a typo cannot create a phantom consent record."""
|
||||
|
||||
def _client(self, auth):
|
||||
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
|
||||
|
||||
client = create_mcp_router_test_client()
|
||||
client.app.dependency_overrides[user_api_key_auth] = lambda: auth
|
||||
return client
|
||||
|
||||
def test_non_admin_targeting_by_email_is_forbidden_without_lookup(self):
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
|
||||
|
||||
auth = UserAPIKeyAuth(api_key="k", user_role=LitellmUserRoles.INTERNAL_USER, user_id="alice")
|
||||
client = self._client(auth)
|
||||
email_lookup = AsyncMock(return_value="someone")
|
||||
|
||||
with (
|
||||
patch(
|
||||
"litellm.proxy.management_endpoints.mcp_management_endpoints.get_prisma_client_or_throw",
|
||||
return_value=MagicMock(),
|
||||
),
|
||||
patch(
|
||||
"litellm.proxy._experimental.mcp_server.auth.user_api_key_auth_mcp.MCPRequestHandler._resolve_single_user_id_by_email",
|
||||
email_lookup,
|
||||
),
|
||||
):
|
||||
resp = client.post("/v1/mcp/delegation", json={"user_email": "victim@x.com", "agent_id": "a1"})
|
||||
|
||||
assert resp.status_code == 403
|
||||
email_lookup.assert_not_awaited()
|
||||
|
||||
def test_non_admin_targeting_foreign_user_id_is_forbidden(self):
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
|
||||
|
||||
auth = UserAPIKeyAuth(api_key="k", user_role=LitellmUserRoles.INTERNAL_USER, user_id="alice")
|
||||
client = self._client(auth)
|
||||
|
||||
with patch(
|
||||
"litellm.proxy.management_endpoints.mcp_management_endpoints.get_prisma_client_or_throw",
|
||||
return_value=MagicMock(),
|
||||
):
|
||||
resp = client.post("/v1/mcp/delegation", json={"user_id": "bob", "agent_id": "a1"})
|
||||
|
||||
assert resp.status_code == 403
|
||||
|
||||
def test_admin_grant_with_nonexistent_user_id_is_404_not_phantom(self):
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
|
||||
|
||||
auth = UserAPIKeyAuth(api_key="admin", user_role=LitellmUserRoles.PROXY_ADMIN, user_id="admin")
|
||||
client = self._client(auth)
|
||||
prisma = MagicMock()
|
||||
prisma.db.litellm_usertable.find_unique = AsyncMock(return_value=None)
|
||||
grant_spy = AsyncMock()
|
||||
|
||||
with (
|
||||
patch(
|
||||
"litellm.proxy.management_endpoints.mcp_management_endpoints.get_prisma_client_or_throw",
|
||||
return_value=prisma,
|
||||
),
|
||||
patch(
|
||||
"litellm.proxy._experimental.mcp_server.delegation_db.grant_user_agent_delegation",
|
||||
grant_spy,
|
||||
),
|
||||
):
|
||||
resp = client.post("/v1/mcp/delegation", json={"user_id": "ghost", "agent_id": "a1"})
|
||||
|
||||
assert resp.status_code == 404
|
||||
grant_spy.assert_not_awaited()
|
||||
|
|
|
|||
11
ui/litellm-dashboard/src/lib/http/schema.d.ts
generated
vendored
11
ui/litellm-dashboard/src/lib/http/schema.d.ts
generated
vendored
|
|
@ -22599,6 +22599,11 @@ export interface components {
|
|||
* @description CIDR ranges of trusted reverse proxies. When set, X-Forwarded-For and X-Forwarded-* origin headers are only trusted from these IPs.
|
||||
*/
|
||||
mcp_trusted_proxy_ranges?: string[] | null;
|
||||
/**
|
||||
* Mcp User Delegation Enabled
|
||||
* @description Enables the x-litellm-delegated-user assertion on MCP routes: an agent-bound key whose agent has the mcp_can_delegate permission and an active consent record from the asserted user resolves upstream per-user credentials as that user. Default False; when disabled, requests carrying the header are rejected.
|
||||
*/
|
||||
mcp_user_delegation_enabled?: boolean | null;
|
||||
/**
|
||||
* Mcp Xff Num Trusted Hops
|
||||
* @description Number of trusted reverse proxies/load balancers in front of the gateway that append to X-Forwarded-For. When set (and mcp_trusted_proxy_ranges validates the direct peer), the client IP for MCP access control is read this many entries from the right of the chain instead of the spoofable leftmost value, defeating append-style X-Forwarded-For forgery.
|
||||
|
|
@ -25414,6 +25419,8 @@ export interface components {
|
|||
blocked_tools?: string[] | null;
|
||||
/** Mcp Access Groups */
|
||||
mcp_access_groups?: string[] | null;
|
||||
/** Mcp Can Delegate */
|
||||
mcp_can_delegate?: boolean | null;
|
||||
/** Mcp Servers */
|
||||
mcp_servers?: string[] | null;
|
||||
/** Mcp Tool Permissions */
|
||||
|
|
@ -25456,6 +25463,8 @@ export interface components {
|
|||
* @default []
|
||||
*/
|
||||
mcp_access_groups: string[] | null;
|
||||
/** Mcp Can Delegate */
|
||||
mcp_can_delegate?: boolean | null;
|
||||
/**
|
||||
* Mcp Servers
|
||||
* @default []
|
||||
|
|
@ -32814,6 +32823,8 @@ export interface components {
|
|||
created_by?: string | null;
|
||||
/** Created By User */
|
||||
created_by_user?: unknown | null;
|
||||
/** Delegated User Id */
|
||||
delegated_user_id?: string | null;
|
||||
/** End User Id */
|
||||
end_user_id?: string | null;
|
||||
/** End User Max Budget */
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue