mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-06 02:48:13 +00:00
0c5c96dbf6
/v1/batches now rejects a missing input_file_id with a 400 through raise_if_required_body_param_missing, so the contract negative that was skipped for "500s instead of 400" passes as written. Verified against a live proxy. The three Datadog MCP tests were skipped because each one sent a `telemetry` argument that search_datadog_logs rejects with "unexpected additional properties". That argument was never a documented Datadog parameter and no assertion reads it, so it is dropped and the tests run again unchanged otherwise.
82 lines
3.3 KiB
Python
82 lines
3.3 KiB
Python
"""Live e2e: a virtual key without MCP access is denied a real MCP server's tools.
|
|
|
|
An admin registers the Datadog remote MCP server through the management API
|
|
(persisted in the DB, picked up without a restart) and queues its deletion. Two
|
|
keys are created against that one server: one granted access through
|
|
`object_permission.mcp_servers` and one with no MCP grant at all. The permitted
|
|
key is the control that proves the upstream is alive and the tool is callable,
|
|
so a failure on the denied key is an authorization denial rather than a dead
|
|
server. The denied key must then see none of the server's tools on `tools/list`
|
|
and must be refused with a 403 on `tools/call`.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from datadog_mcp import SEARCH_LOGS_TOOL, register_datadog_mcp
|
|
from e2e_config import DD_SEARCH_FROM, unique_marker
|
|
from e2e_http import unwrap
|
|
from lifecycle import ResourceManager
|
|
from mcp_client import McpClient
|
|
|
|
pytestmark = pytest.mark.e2e
|
|
|
|
|
|
def _key(client: McpClient, resources: ResourceManager, *, mcp_servers: list[str] | None) -> str:
|
|
label = "allowed" if mcp_servers else "denied"
|
|
key = client.generate_key(user_id=f"e2e-mcp-{label}-{unique_marker()}", mcp_servers=mcp_servers)
|
|
resources.defer(lambda: client.proxy.delete_key(key))
|
|
return key
|
|
|
|
|
|
class TestMcpKeyWithoutAccessIsDenied:
|
|
@pytest.mark.covers("mcp.list_tools.api_key.denied_without_permission")
|
|
def test_list_tools_denied_without_permission(
|
|
self,
|
|
client: McpClient,
|
|
resources: ResourceManager,
|
|
) -> None:
|
|
server_id = register_datadog_mcp(client, resources)
|
|
client.await_registered(server_id)
|
|
|
|
permitted_key = _key(client, resources, mcp_servers=[server_id])
|
|
denied_key = _key(client, resources, mcp_servers=None)
|
|
|
|
_ = client.await_tool(permitted_key, server_id, SEARCH_LOGS_TOOL)
|
|
|
|
denied_tools = unwrap(client.list_tools(denied_key)).tool_names_for_server(server_id)
|
|
assert denied_tools == frozenset(), (
|
|
f"ungranted key saw the server's tools; tools/list leaked across the permission "
|
|
f"boundary: {denied_tools}"
|
|
)
|
|
|
|
@pytest.mark.covers("mcp.call_tool.api_key.denied_without_permission")
|
|
def test_call_tool_denied_without_permission(
|
|
self,
|
|
client: McpClient,
|
|
resources: ResourceManager,
|
|
) -> None:
|
|
server_id = register_datadog_mcp(client, resources)
|
|
client.await_registered(server_id)
|
|
|
|
permitted_key = _key(client, resources, mcp_servers=[server_id])
|
|
denied_key = _key(client, resources, mcp_servers=None)
|
|
|
|
tool_name = client.await_tool(permitted_key, server_id, SEARCH_LOGS_TOOL)
|
|
|
|
search_args = {
|
|
"query": "service:litellm",
|
|
"from": DD_SEARCH_FROM,
|
|
"to": "now",
|
|
"max_tokens": 1000,
|
|
}
|
|
permitted_call = client.await_call_tool(
|
|
permitted_key, server_id=server_id, name=tool_name, arguments=search_args
|
|
)
|
|
assert permitted_call.is_error is not True, f"granted key's tool call errored: {permitted_call}"
|
|
|
|
denied = client.await_call_tool_denied(
|
|
denied_key, server_id=server_id, name=tool_name, arguments=search_args
|
|
)
|
|
assert "access_denied" in denied.body, f"403 was not an MCP access denial: {denied.body}"
|