langchain-ai/deepagents
1
0
Fork 0
mirror of https://github.com/langchain-ai/deepagents.git synced 2026-07-22 09:45:24 -04:00
Code Issues 169 Packages Projects Releases 16 Wiki Activity

feat(code): selective per-server project MCP trust (#4507)

Browse Source 
Add `[mcp].enabled_project_servers` / `disabled_project_servers` (and
env equivalents) to selectively pre-approve or reject individual project
MCP servers by name from user-level config.

---

Mirrors Claude Code's `enabledMcpjsonServers` /
`disabledMcpjsonServers`: a user can pre-approve or reject specific
project MCP servers (from a repo's `.mcp.json`) by name via a new
`[mcp]` `config.toml` table (`enabled_project_servers` /
`disabled_project_servers`) or the
`DEEPAGENTS_CODE_ENABLED_PROJECT_MCP_SERVERS` / `..._DISABLED_...` env
vars — without the interactive fingerprint prompt and without trusting
the whole config. Reject wins over approval and over full trust.

The allow/deny lists are read **only** from the user-level
`~/.deepagents/config.toml` and process env (via the new
`model_config.load_mcp_server_trust_lists`), never from `.mcp.json` or
any repo-committed file, so a committed config cannot self-approve its
own servers. The untrusted path changes from "drop all" to "drop all
except allowlisted", and only allowlisted (or fully trusted) names
survive into the merged config — so the remote-SSRF /
header-exfiltration gate stays intact.

Made by [Open
SWE](https://openswe.vercel.app/agents/0b9a59e3-2c90-47ce-65f9-efac2532321a)

## References
- Plan:
https://openswe.vercel.app/agents/0b9a59e3-2c90-47ce-65f9-efac2532321a/plan

---------

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
This commit is contained in:
Mason Daugherty
2026-07-06 10:40:20 -04:00
committed by GitHub
parent 5d1c3928f7
commit aaa22a9340
13 changed files with 1982 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/code/THREAT_MODEL.md
+1 -1
View File
@@ -232,7 +232,7 @@
#### TB4: MCP Config → Process / Network
- **Inside**: `mcp_tools._validate_server_config` validates JSON structure and field types. `mcp_trust.compute_config_fingerprint` computes SHA-256 over config file contents. Project-level stdio servers require trust approval (fingerprint prompt) unless `--trust-project-mcp` is set.
- **Inside**: `mcp_tools._validate_server_config` validates JSON structure and field types. `mcp_trust.compute_config_fingerprint` computes SHA-256 over config file contents. Project-level stdio servers require trust approval (fingerprint prompt) unless `--trust-project-mcp` is set. A user can selectively pre-approve or reject individual project servers by name via `[mcp].enabled_project_servers` / `disabled_project_servers` (or the `DEEPAGENTS_CODE_ENABLED_PROJECT_MCP_SERVERS` / `..._DISABLED_...` env vars); reject wins over approval and over full trust. These lists are read by `model_config.load_mcp_server_trust_lists` only from user-controlled sources — the user-level `~/.deepagents/config.toml`, the global `~/.deepagents/.env`, and shell-exported env — never from `.mcp.json` or any repo-committed file, so a committed config cannot self-approve its own servers. In particular, the env forms (`DEEPAGENTS_CODE_ENABLED_PROJECT_MCP_SERVERS` / `..._DISABLED_...`) are added to `config._PROJECT_DOTENV_DENIED_ENV_KEYS`, so a committed *project* `.env` cannot inject them into `os.environ` even though it is loaded at bootstrap. Enabled resolves env-over-TOML, but disabled *unions* TOML and env so a deny can never be silently emptied by the other source. If the user's `config.toml` exists but cannot be read or parsed, the loader fails closed (project configs are treated as untrusted and the error is surfaced) rather than proceeding with an empty deny list. Only allowlisted (or fully trusted) names survive into the merged config, so a non-allowlisted remote entry is never preflighted and its interpolated headers are never resolved. Note that the allow list binds by server *name*, not by fingerprint: an allowlisted name is loaded whatever definition the repo currently gives it, bypassing the fingerprint change-detection that re-prompts on full-config trust — so users should only allowlist names in repos they would otherwise trust.
- **Outside**: Stdio server `command`, `args`, and `env` fields are user-controlled strings passed directly to `StdioConnection`. The `env` dict from MCP config is forwarded without filtering — users can set arbitrary environment variables (including `PATH`, `LD_PRELOAD`, `PYTHONPATH`) for the MCP subprocess.
- **Crossing mechanism**: `subprocess.Popen` (via `langchain_mcp_adapters`) for stdio; HTTP/SSE for remote servers.
libs/code/deepagents_code/_env_vars.py
+36
View File
@@ -93,6 +93,42 @@ real PyPI release.
Any non-empty value enables the flag (including `"0"` or `"false"`).
"""
DISABLED_PROJECT_MCP_SERVERS = "DEEPAGENTS_CODE_DISABLED_PROJECT_MCP_SERVERS"
"""Comma-separated project MCP server names to always reject by name.
A user-level equivalent of `[mcp].disabled_project_servers`.
Rejection wins over approval: a name listed here is dropped even when it also
appears in `ENABLED_PROJECT_MCP_SERVERS` (or `[mcp].enabled_project_servers`)
and even when the project config is otherwise trusted. Unlike the enabled list,
this env var *unions* with (rather than replaces)
`[mcp].disabled_project_servers` — denies accumulate across sources, so neither
can silently empty a deny set in the other. This is process env the user
controls, not a repo file, so it does not weaken the user-level-only
trust boundary: a committed *project* `.env` is blocked from setting it
(see `config._PROJECT_DOTENV_DENIED_ENV_KEYS`); only the user's shell,
launch env, or global `~/.deepagents/.env` can.
"""
ENABLED_PROJECT_MCP_SERVERS = "DEEPAGENTS_CODE_ENABLED_PROJECT_MCP_SERVERS"
"""Comma-separated project MCP server names to pre-approve by name.
A user-level equivalent of `[mcp].enabled_project_servers`.
Servers named here load from an otherwise-untrusted project `.mcp.json` without
prompting (they are omitted from the interactive approval prompt), while
non-listed servers stay dropped. Like `DISABLED_PROJECT_MCP_SERVERS`, this is
user-controlled process env, not a repo file, so it does not weaken
the user-level-only trust boundary (a committed *project* `.env` cannot set it;
see `config._PROJECT_DOTENV_DENIED_ENV_KEYS`). This contract is name-based:
a project command or URL change under the same server name still matches.
When set, this replaces (takes precedence over) the
`[mcp].enabled_project_servers` TOML list.
(`DISABLED_PROJECT_MCP_SERVERS` instead *unions* with its TOML list, so a deny
is never silently emptied.)
"""
EXTERNAL_EVENT_SOCKET = "DEEPAGENTS_CODE_EXTERNAL_EVENT_SOCKET"
"""Enable the local Unix-socket external event listener.
libs/code/deepagents_code/config.py
+55 -8
View File
@@ -19,7 +19,12 @@ from pathlib import Path
from typing import TYPE_CHECKING, Any, Protocol
from urllib.parse import unquote, urlparse
from deepagents_code._env_vars import HIDE_SPLASH_VERSION, is_env_truthy
from deepagents_code._env_vars import (
DISABLED_PROJECT_MCP_SERVERS,
ENABLED_PROJECT_MCP_SERVERS,
HIDE_SPLASH_VERSION,
is_env_truthy,
)
from deepagents_code._git import resolve_git_branch
from deepagents_code._version import __version__
from deepagents_code.config_manifest import (
@@ -155,6 +160,32 @@ lowercase `bash_env` injected into the environment is inert. Any future entry
that some consumer reads case-insensitively would need a different check.
"""
_PROJECT_DOTENV_DENIED_ENV_KEYS = frozenset(
{
ENABLED_PROJECT_MCP_SERVERS,
DISABLED_PROJECT_MCP_SERVERS,
}
)
"""Env keys a *project* `.env` must not inject, even though they are otherwise
safe process-env inputs.
These two vars are the env form of the user-level project-MCP allow/deny lists
(`model_config.load_mcp_server_trust_lists`). Their whole purpose is to be a
*user-level* decision: naming a project MCP server here pre-approves it from an
untrusted `.mcp.json` (stdio → local command execution; remote → SSRF and
`${VAR}` header exfiltration during the discovery preflight). A project `.env`
travels with a cloned repo, so honoring it would let an attacker commit
`.mcp.json` + `.env` and self-approve their own servers — exactly the trust
boundary the feature exists to hold.
Unlike `_DOTENV_DENIED_ENV_KEYS` (denied from *any* `.env` because they turn
`.env` loading into code execution), these are denied only from the *project*
`.env`: the user's own global `~/.deepagents/.env` and their shell exports are
legitimate, trusted sources and continue to set them. The loader reads plain
`os.environ`, so blocking injection here — before the value ever reaches
`os.environ` — is what keeps that read trustworthy.
"""
def _find_dotenv_from_start_path(start_path: Path) -> Path | None:
"""Find the nearest `.env` file from an explicit start path upward.
@@ -201,7 +232,7 @@ def _preview_dotenv_environ(*, start_path: Path | None = None) -> dict[str, str]
if env.get(key) == value:
env.pop(key)
def apply_dotenv(dotenv_path: Path | None) -> None:
def apply_dotenv(dotenv_path: Path | None, *, is_project: bool) -> None:
if dotenv_path is None:
return
try:
@@ -221,6 +252,13 @@ def _preview_dotenv_environ(*, start_path: Path | None = None) -> dict[str, str]
# Log the key only — the value is attacker-controlled.
logger.debug("Ignoring denied env key %r from %s", key, dotenv_path)
continue
if is_project and key in _PROJECT_DOTENV_DENIED_ENV_KEYS:
# Mirror `_load_dotenv`: a project `.env` cannot preview-set a
# user-level MCP trust decision (the global `.env`/shell can).
logger.debug(
"Ignoring project-denied env key %r from %s", key, dotenv_path
)
continue
env[key] = value
project_dotenv: Path | None = None
@@ -237,7 +275,7 @@ def _preview_dotenv_environ(*, start_path: Path | None = None) -> dict[str, str]
start_path or "cwd",
exc_info=True,
)
apply_dotenv(project_dotenv)
apply_dotenv(project_dotenv, is_project=True)
try:
global_dotenv = _GLOBAL_DOTENV_PATH if _GLOBAL_DOTENV_PATH.is_file() else None
@@ -249,7 +287,7 @@ def _preview_dotenv_environ(*, start_path: Path | None = None) -> dict[str, str]
exc_info=True,
)
global_dotenv = None
apply_dotenv(global_dotenv)
apply_dotenv(global_dotenv, is_project=False)
return env
@@ -312,7 +350,7 @@ def _load_dotenv(
os.environ.pop(key)
_dotenv_loaded_values.clear()
def apply_dotenv(dotenv_path: Path) -> bool:
def apply_dotenv(dotenv_path: Path, *, is_project: bool) -> bool:
values = dotenv.dotenv_values(dotenv_path=dotenv_path)
applied = False
for key, value in values.items():
@@ -322,6 +360,13 @@ def _load_dotenv(
# Log the key only — the value is attacker-controlled.
logger.debug("Ignoring denied env key %r from %s", key, dotenv_path)
continue
if is_project and key in _PROJECT_DOTENV_DENIED_ENV_KEYS:
# A committed project `.env` must not set a user-level MCP trust
# decision; the global `.env` and shell may (is_project=False).
logger.debug(
"Ignoring project-denied env key %r from %s", key, dotenv_path
)
continue
os.environ[key] = value
_dotenv_loaded_values[key] = value
applied = True
@@ -335,11 +380,11 @@ def _load_dotenv(
found = dotenv.find_dotenv(usecwd=True)
if found:
dotenv_path = found
loaded = apply_dotenv(Path(found)) or loaded
loaded = apply_dotenv(Path(found), is_project=True) or loaded
else:
dotenv_path = _find_dotenv_from_start_path(start_path)
if dotenv_path is not None:
loaded = apply_dotenv(dotenv_path) or loaded
loaded = apply_dotenv(dotenv_path, is_project=True) or loaded
except (OSError, ValueError):
logger.warning(
"Could not read project dotenv at %s; project env vars will not be loaded",
@@ -352,7 +397,9 @@ def _load_dotenv(
# try/except wraps both is_file() and load_dotenv() to cover the TOCTOU
# window where the file can vanish between stat and open.
try:
if _GLOBAL_DOTENV_PATH.is_file() and apply_dotenv(_GLOBAL_DOTENV_PATH):
if _GLOBAL_DOTENV_PATH.is_file() and apply_dotenv(
_GLOBAL_DOTENV_PATH, is_project=False
):
loaded = True
logger.debug("Loaded global dotenv: %s", _GLOBAL_DOTENV_PATH)
except (OSError, ValueError):
libs/code/deepagents_code/config_manifest.py
+44
View File
@@ -1104,6 +1104,44 @@ _STATIC_OPTIONS: tuple[ConfigOption, ...] = (
default=False,
env_var=_env_vars.SUPPRESS_ENV_OVERRIDE_WARNING,
),
# --- MCP ------------------------------------------------------------
# Project trust lists are parsed by `model_config.load_mcp_server_trust_lists`,
# which reads them only from the user-level config.toml (never a project file),
# so they are STRUCTURED-for-discovery here rather than env-backed scalars. The
# env overrides are named in the summaries instead of `env_var` because the
# scalar resolver rejects env-backed STRUCTURED options by design.
ConfigOption(
key="mcp.enabled_project_servers",
group="MCP",
summary=(
"Project MCP server names to pre-approve by name from an untrusted "
".mcp.json; command/URL changes under the same name still match "
"(env: DEEPAGENTS_CODE_ENABLED_PROJECT_MCP_SERVERS)."
),
kind=OptionKind.STRUCTURED,
toml_keys=("mcp", "enabled_project_servers"),
),
ConfigOption(
key="mcp.disabled_project_servers",
group="MCP",
summary=(
"Project MCP server names to always reject; reject wins over approval "
"and trust (env: DEEPAGENTS_CODE_DISABLED_PROJECT_MCP_SERVERS)."
),
kind=OptionKind.STRUCTURED,
toml_keys=("mcp", "disabled_project_servers"),
),
# Unlike the two trust lists above, this one is managed by `mcp_disabled`
# (the server viewer's disable toggle), not `load_mcp_server_trust_lists`;
# it plays no part in the project-trust security boundary. It is STRUCTURED
# here purely for discovery.
ConfigOption(
key="mcp.disabled_servers",
group="MCP",
summary="MCP server names disabled by the user from the server viewer.",
kind=OptionKind.STRUCTURED,
toml_keys=("mcp", "disabled_servers"),
),
# --- Updates --------------------------------------------------------
ConfigOption(
key="update.auto_update",
@@ -1203,6 +1241,12 @@ NON_OPTION_ENV_VARS: frozenset[str] = frozenset(
# Set then popped during the self-update restart handshake (main.py);
# never user-configured.
_env_vars.RESTARTED_AFTER_UPDATE,
# Env equivalents of the STRUCTURED `[mcp]` lists. They are read by the
# dedicated `model_config.load_mcp_server_trust_lists` loader (which the
# `mcp.*` STRUCTURED options describe for discovery), not by the scalar
# resolver, so they intentionally have no scalar `env_var` ConfigOption.
_env_vars.ENABLED_PROJECT_MCP_SERVERS,
_env_vars.DISABLED_PROJECT_MCP_SERVERS,
}
)
"""`_env_vars` constants intentionally excluded from the option catalog."""
libs/code/deepagents_code/main.py
+83 -8
View File
@@ -1897,7 +1897,8 @@ async def run_textual_cli_async(
Merged on top of auto-discovered configs (highest precedence).
no_mcp: Disable all MCP tool loading.
trust_project_mcp: Controls project-level stdio server trust.
trust_project_mcp: Controls project-level server trust (stdio and
remote alike).
`True` to allow, `False` to deny, `None` to check trust store.
enable_interpreter: Enable `CodeInterpreterMiddleware` (`js_eval`) on
@@ -2056,7 +2057,8 @@ async def _run_acp_cli_async(
profile_override: Extra profile fields from `--profile-override`.
mcp_config_path: Optional path to MCP servers JSON configuration file.
no_mcp: Disable all MCP tool loading.
trust_project_mcp: Controls project-level stdio server trust.
trust_project_mcp: Controls project-level server trust (stdio and
remote alike).
Returns:
Exit code for ACP mode.
@@ -2351,12 +2353,20 @@ def _check_mcp_project_trust(*, trust_flag: bool = False) -> bool | None:
returns `True`. Otherwise checks the persistent trust store; if
untrusted, shows an interactive approval prompt.
Servers already resolved by the user's `enabled_project_servers` /
`disabled_project_servers` lists are shown for transparency but not
prompted for (enabled ones load regardless; disabled ones never load).
`None` is returned when that leaves nothing to decide. If the user's own
allow/deny policy cannot be read, returns `False` (fail closed) rather than
prompting under an unknown deny list.
Args:
trust_flag: Whether `--trust-project-mcp` was passed.
Returns:
`True` to allow project servers, `False` to deny, or `None`
when no project servers exist.
`True` to allow project servers, `False` to deny (including when the
user's trust policy could not be read), or `None` when there are no
project servers whose fate this prompt decides.
"""
from deepagents_code.mcp_tools import (
classify_discovered_configs,
@@ -2421,20 +2431,85 @@ def _check_mcp_project_trust(*, trust_flag: bool = False) -> bool | None:
if not debug_prompt and is_project_mcp_trusted(project_root, fingerprint):
return True
# Interactive prompt
# Partition by the user's own allow/deny lists (read only from home config,
# never the repo — the same boundary the loader enforces). Enabled servers
# load regardless of the answer here and disabled ones never load, so the
# prompt must not *ask* about them. It still *shows* them, though, so the
# user sees what their config decided — notably a repo redefining an
# allowlisted name, which binds by name rather than by fingerprint.
from deepagents_code.model_config import load_mcp_server_trust_lists
trust_lists = load_mcp_server_trust_lists()
from rich.console import Console as _Console
from rich.markup import escape
prompt_console = _Console(stderr=True)
prompt_servers: list[tuple[str, str, str]] = []
preapproved: list[tuple[str, str, str]] = []
blocked: list[tuple[str, str, str]] = []
for name, kind, summary in all_servers:
# Disabled first: reject precedence (a name in both lists is disabled).
if name in trust_lists.disabled:
blocked.append((name, kind, summary))
elif name in trust_lists.enabled:
preapproved.append((name, kind, summary))
else:
prompt_servers.append((name, kind, summary))
def _print_auto_resolved() -> None:
"""List servers the config already decided, without asking about them."""
if not preapproved and not blocked:
return
prompt_console.print()
prompt_console.print(
"[dim]Resolved by your config (not prompted):[/dim]", highlight=False
)
for name, kind, summary in preapproved:
prompt_console.print(
f' [green]"{escape(name)}"[/green] ({escape(kind)}): pre-approved '
f"(enabled_project_servers): {escape(summary)}",
highlight=False,
)
for name, kind, summary in blocked:
prompt_console.print(
f' [red]"{escape(name)}"[/red] ({escape(kind)}): blocked '
f"(disabled_project_servers): {escape(summary)}",
highlight=False,
)
if trust_lists.read_error is not None:
# The user's allow/deny policy could not be read. Fail closed here too
# (matching the loader, which forces the config untrusted) instead of
# prompting and possibly persisting fingerprint trust under an unknown
# deny list. Any env-enabled names still load — the loader re-applies the
# lists downstream — but nothing is approved via this prompt.
prompt_console.print(
f"[yellow]Warning: {escape(trust_lists.read_error)}; treating "
"project MCP servers as untrusted.[/yellow]",
highlight=False,
)
_print_auto_resolved()
return False
if not prompt_servers:
# Nothing left to decide interactively, but surface what the lists
# resolved so a load driven purely by config is never fully silent.
_print_auto_resolved()
return None
docs_url = (
"https://docs.langchain.com/oss/python/deepagents/code/"
"mcp-tools#project-level-trust"
)
prompt_console = _Console(stderr=True)
prompt_console.print()
prompt_console.print(
"[bold yellow]Project MCP servers require approval:[/bold yellow]"
)
for name, kind, summary in all_servers:
prompt_console.print(f' [bold]"{name}"[/bold] ({kind}): {summary}')
for name, kind, summary in prompt_servers:
prompt_console.print(
f' [bold]"{escape(name)}"[/bold] ({escape(kind)}): {escape(summary)}'
)
_print_auto_resolved()
prompt_console.print()
prompt_console.print(
f"[dim]Learn more: [link={docs_url}]{docs_url}[/link][/dim]",
libs/code/deepagents_code/mcp_disabled.py
+52 -22
View File
@@ -2,8 +2,8 @@
Disabled servers are skipped at config merge time so their tools never
reach the agent and no connection is attempted. State lives under
`[mcp_disabled]` in `~/.deepagents/config.toml`, alongside the user's
other configuration sections.
`[mcp].disabled_servers` in `~/.deepagents/config.toml`, alongside the
user's other MCP configuration.
The store keys on server *name* alone. Two configs that both declare a
`github` server will both be disabled by a single entry — intentional,
@@ -24,8 +24,10 @@ from deepagents_code.model_config import DEFAULT_CONFIG_PATH as _DEFAULT_CONFIG_
logger = logging.getLogger(__name__)
_SECTION = "mcp_disabled"
_KEY = "servers"
_SECTION = "mcp"
_KEY = "disabled_servers"
_LEGACY_SECTION = "mcp_disabled"
_LEGACY_KEY = "servers"
class _ConfigLoadError(Exception):
@@ -92,6 +94,43 @@ def _save_config(data: dict[str, Any], config_path: Path) -> bool:
return True
def _coerce_entries(entries: object) -> set[str] | None:
"""Return valid server names from a TOML value, or `None` when unset."""
if not isinstance(entries, list):
return None
return {name for name in entries if isinstance(name, str) and name}
def _disabled_entries(data: dict[str, Any]) -> set[str]:
"""Return disabled names from the current config shape with legacy fallback."""
section = data.get(_SECTION)
if isinstance(section, dict):
entries = _coerce_entries(section.get(_KEY))
if entries is not None:
return entries
legacy_section = data.get(_LEGACY_SECTION)
if isinstance(legacy_section, dict):
entries = _coerce_entries(legacy_section.get(_LEGACY_KEY))
if entries is not None:
return entries
return set()
def _remove_legacy_disabled_section(data: dict[str, Any]) -> None:
"""Drop the old top-level section after writing the folded config shape."""
legacy_section = data.get(_LEGACY_SECTION)
if not isinstance(legacy_section, dict):
data.pop(_LEGACY_SECTION, None)
return
legacy_section.pop(_LEGACY_KEY, None)
if legacy_section:
data[_LEGACY_SECTION] = legacy_section
else:
data.pop(_LEGACY_SECTION, None)
def get_disabled_servers(*, config_path: Path | None = None) -> set[str]:
"""Return the set of server names the user has disabled.
@@ -108,13 +147,7 @@ def get_disabled_servers(*, config_path: Path | None = None) -> set[str]:
data = _load_config(config_path)
except _ConfigLoadError:
return set()
section = data.get(_SECTION)
if not isinstance(section, dict):
return set()
entries = section.get(_KEY)
if not isinstance(entries, list):
return set()
return {name for name in entries if isinstance(name, str) and name}
return _disabled_entries(data)
def is_server_disabled(server_name: str, *, config_path: Path | None = None) -> bool:
@@ -159,24 +192,21 @@ def set_server_disabled(
data = _load_config(config_path)
except _ConfigLoadError as exc:
return False, str(exc)
section = data.get(_SECTION)
if not isinstance(section, dict):
section = {}
entries_raw = section.get(_KEY)
entries: list[str] = (
[name for name in entries_raw if isinstance(name, str) and name]
if isinstance(entries_raw, list)
else []
)
current = set(entries)
current = _disabled_entries(data)
previous = set(current)
if disabled:
current.add(server_name)
else:
current.discard(server_name)
if current == set(entries):
if current == previous and _LEGACY_SECTION not in data:
return True, None
section = data.get(_SECTION)
if not isinstance(section, dict):
section = {}
section[_KEY] = sorted(current)
data[_SECTION] = section
_remove_legacy_disabled_section(data)
if _save_config(data, config_path):
return True, None
return False, f"could not write {config_path}"
libs/code/deepagents_code/mcp_tools.py
+283 -86
View File
@@ -27,6 +27,7 @@ if TYPE_CHECKING:
from langchain_mcp_adapters.client import Connection
from mcp import ClientSession
from deepagents_code.model_config import McpServerTrustLists
from deepagents_code.project_utils import ProjectContext
logger = logging.getLogger(__name__)
@@ -710,6 +711,93 @@ def _json_error_snippet(
return f" {source}\n {' ' * caret_col}^"
def _load_mcp_config_json(config_path: str) -> dict[str, Any]:
"""Load MCP configuration JSON with parser diagnostics.
Args:
config_path: Path to the MCP JSON configuration file.
Returns:
Parsed configuration dictionary.
Raises:
FileNotFoundError: If config file doesn't exist.
json.JSONDecodeError: If config file contains invalid JSON.
"""
path = Path(config_path)
if not path.exists():
error_msg = f"MCP config file not found: {config_path}"
raise FileNotFoundError(error_msg)
try:
with path.open(encoding="utf-8") as file_obj:
return json.load(file_obj)
except json.JSONDecodeError as exc:
# Build a layered message: core reason, an actionable hint for common
# mistakes, then a caret snippet last so the auto-appended
# "line X column Y" suffix reads as the location of the caret.
parts = [f"Invalid JSON in MCP config file: {exc.msg}"]
hint = _json_error_hint(exc)
if hint is not None:
parts.append(hint)
snippet = _json_error_snippet(exc.doc, exc.lineno, exc.colno, pos=exc.pos)
if snippet is not None:
parts.append(snippet)
error_msg = "\n".join(parts)
raise json.JSONDecodeError(error_msg, exc.doc, exc.pos) from exc
def _validate_mcp_config_top_level(config: dict[str, Any]) -> None:
"""Validate top-level MCP configuration fields.
Args:
config: Parsed MCP config dictionary.
Raises:
TypeError: If top-level fields have wrong types.
ValueError: If required top-level fields are missing.
"""
if "mcpServers" not in config:
error_msg = (
"MCP config must contain 'mcpServers' field. "
'Expected format: {"mcpServers": {"server-name": {...}}}'
)
raise ValueError(error_msg)
if not isinstance(config["mcpServers"], dict):
error_msg = "'mcpServers' field must be a dictionary"
raise TypeError(error_msg)
if not config["mcpServers"]:
error_msg = "'mcpServers' field is empty - no servers configured"
raise ValueError(error_msg)
def _validate_mcp_config_servers(config: dict[str, Any]) -> None:
"""Validate every server in an MCP configuration.
Args:
config: Parsed MCP config dictionary.
"""
for server_name, server_config in config["mcpServers"].items():
_validate_server_config(server_name, server_config)
def _load_mcp_config_top_level(config_path: Path) -> dict[str, Any]:
"""Load an MCP config file and validate only its top-level shape.
Args:
config_path: Config path to load.
Returns:
Parsed configuration dictionary with a valid `mcpServers` mapping.
"""
config = _load_mcp_config_json(str(config_path))
_validate_mcp_config_top_level(config)
return config
def load_mcp_config(config_path: str) -> dict[str, Any]:
"""Load and validate MCP configuration from a JSON file.
@@ -741,48 +829,9 @@ def load_mcp_config(config_path: str) -> dict[str, Any]:
json.JSONDecodeError: If config file contains invalid JSON.
TypeError: If config fields have wrong types.
ValueError: If config is missing required fields.
RuntimeError: If header env-var interpolation references an unset var.
""" # noqa: DOC502 - `_validate_server_config()` raises `RuntimeError` indirectly
path = Path(config_path)
if not path.exists():
error_msg = f"MCP config file not found: {config_path}"
raise FileNotFoundError(error_msg)
try:
with path.open(encoding="utf-8") as file_obj:
config = json.load(file_obj)
except json.JSONDecodeError as exc:
# Build a layered message: core reason, an actionable hint for common
# mistakes, then a caret snippet last so the auto-appended
# "line X column Y" suffix reads as the location of the caret.
parts = [f"Invalid JSON in MCP config file: {exc.msg}"]
hint = _json_error_hint(exc)
if hint is not None:
parts.append(hint)
snippet = _json_error_snippet(exc.doc, exc.lineno, exc.colno, pos=exc.pos)
if snippet is not None:
parts.append(snippet)
error_msg = "\n".join(parts)
raise json.JSONDecodeError(error_msg, exc.doc, exc.pos) from exc
if "mcpServers" not in config:
error_msg = (
"MCP config must contain 'mcpServers' field. "
'Expected format: {"mcpServers": {"server-name": {...}}}'
)
raise ValueError(error_msg)
if not isinstance(config["mcpServers"], dict):
error_msg = "'mcpServers' field must be a dictionary"
raise TypeError(error_msg)
if not config["mcpServers"]:
error_msg = "'mcpServers' field is empty - no servers configured"
raise ValueError(error_msg)
for server_name, server_config in config["mcpServers"].items():
_validate_server_config(server_name, server_config)
""" # noqa: DOC502 - raised indirectly by `_load_mcp_config_json` / `_validate_server_config` (which does shape-only checks; `${VAR}` header interpolation is deferred to activation time, so no RuntimeError here)
config = _load_mcp_config_top_level(Path(config_path))
_validate_mcp_config_servers(config)
return config
@@ -924,6 +973,11 @@ def extract_project_server_summaries(
return results
for name, server in servers.items():
if not isinstance(server, dict):
logger.debug(
"Skipping malformed MCP server entry %r: expected a table, got %s",
name,
type(server).__name__,
)
continue
kind = _resolve_server_type(server)
if kind == "stdio":
@@ -994,6 +1048,31 @@ def load_mcp_config_with_error(
return None, str(exc)
def _load_mcp_config_top_level_with_error(
config_path: Path,
) -> tuple[dict[str, Any] | None, str | None]:
"""Load an MCP config file, validating only its top-level structure.
Args:
config_path: Config path to load.
Returns:
`(parsed_config, None)` on success, `(None, None)` when the file
doesn't exist, or `(None, error_message)` on load/top-level validate
failure.
"""
try:
return _load_mcp_config_top_level(config_path), None
except FileNotFoundError:
return None, None
except OSError as exc:
logger.warning("Skipping unreadable MCP config %s: %s", config_path, exc)
return None, f"Unreadable: {exc}"
except (json.JSONDecodeError, ValueError, TypeError) as exc:
logger.warning("Skipping invalid MCP config %s: %s", config_path, exc)
return None, str(exc)
def _check_stdio_server(server_name: str, server_config: dict[str, Any]) -> None:
"""Verify that a stdio server's command exists on PATH.
@@ -1753,6 +1832,47 @@ async def get_mcp_tools(
return await _load_tools_from_config(config)
def _log_skipped_project_servers(
dropped: list[tuple[str, str, str]],
*,
trust_project_mcp: bool | None,
config_trusted: bool,
) -> None:
"""Log project MCP servers that were dropped, explaining why.
Split out so the trust/drop loop stays readable. The message distinguishes an
explicit reject on an otherwise-trusted config from the untrusted-drop cases,
which themselves differ by whether trust was declined outright
(`--trust-project-mcp` off) or merely not yet granted.
Args:
dropped: `(name, kind, summary)` tuples for each skipped server.
trust_project_mcp: The caller's tri-state trust flag.
config_trusted: Whether the project config was otherwise trusted (so the
only reason to drop is an explicit user-level deny entry).
"""
skipped_list = "\n".join(
f"- {name} [{kind}]: {summary}" for name, kind, summary in dropped
)
if config_trusted:
logger.warning(
"Skipped project MCP servers rejected by user config "
"(disabled_project_servers):\n%s",
skipped_list,
)
elif trust_project_mcp is False:
logger.warning(
"Skipped untrusted project MCP servers:\n%s",
skipped_list,
)
else:
logger.warning(
"Skipped untrusted project MCP servers "
"(config changed or not yet approved):\n%s",
skipped_list,
)
async def resolve_and_load_mcp_tools(
*,
explicit_config_path: str | None = None,
@@ -1772,12 +1892,23 @@ async def resolve_and_load_mcp_tools(
explicit_config_path: Extra config file to layer on top of
auto-discovered configs.
no_mcp: If `True`, disable all MCP loading.
trust_project_mcp: Controls project-level stdio server trust.
trust_project_mcp: Controls project-level server trust.
- `True`: always trust project configs, including stdio servers.
- `False`: drop stdio entries from project configs.
Applies to stdio and remote (http/sse) servers alike — remote entries
are gated too because an attacker-controlled `.mcp.json` can SSRF or
exfiltrate `${VAR}` headers during the discovery preflight.
- `True`: always trust project configs (all servers).
- `False`: drop all project servers (stdio and remote).
- `None`: consult the persistent trust store — trusted configs
load fully, untrusted project stdio servers are dropped.
load fully; all servers from untrusted project configs are
dropped.
Regardless of this flag, the user-level allow/deny lists
(`[mcp].enabled_project_servers` /`disabled_project_servers` and
their env equivalents, via `load_mcp_server_trust_lists`) are
applied: named servers load from an otherwise-untrusted config,
and explicitly denied servers are dropped even from a trusted one.
project_context: Explicit project path context for config discovery
and trust resolution.
stateless: When `True`, do not return an owned runtime session manager.
@@ -1795,9 +1926,10 @@ async def resolve_and_load_mcp_tools(
types.
ValueError: If `explicit_config_path` is missing required fields
or declares an unsupported transport.
RuntimeError: If the merged MCP config is malformed, or header
env-var interpolation in `explicit_config_path` references an
unset variable.
RuntimeError: If the merged MCP config is malformed. (Header `${VAR}`
interpolation is deferred to activation inside
`_load_tools_from_config`, which captures such failures into the
returned `server_infos` rather than raising here.)
""" # noqa: DOC502 - FileNotFoundError / JSONDecodeError / TypeError / ValueError surface via `load_mcp_config`
if no_mcp:
return [], None, []
@@ -1822,8 +1954,9 @@ async def resolve_and_load_mcp_tools(
configs.append(config)
project_trusted: bool | None = None
trust_lists: McpServerTrustLists | None = None
for path in project_configs:
config, error = load_mcp_config_with_error(path)
config, error = _load_mcp_config_top_level_with_error(path)
if error is not None:
config_load_errors.append((path, error))
if config is None:
@@ -1831,46 +1964,110 @@ async def resolve_and_load_mcp_tools(
project_servers = extract_project_server_summaries(config)
if not project_servers:
configs.append(config)
# No dict servers yielded a summary, so every entry is malformed.
# Re-validate the already-loaded config (no second file read) to
# surface a precise per-server error; this always fails today, but
# the append path is kept so a future validator that accepts such a
# shape would still load it.
try:
_validate_mcp_config_servers(config)
except (ValueError, TypeError, RuntimeError) as exc:
config_load_errors.append((path, str(exc)))
else:
configs.append(config)
continue
# Whether the config as a whole is trusted (flag/env/fingerprint). This
# governs the default for un-listed servers; the user-level allow/deny
# lists below refine it per server.
if trust_project_mcp is True:
configs.append(config)
continue
if trust_project_mcp is None and project_trusted is None:
from deepagents_code.mcp_trust import (
compute_config_fingerprint,
is_project_mcp_trusted,
)
project_root = str(_resolve_project_config_base(project_context).resolve())
fingerprint = compute_config_fingerprint(project_configs)
project_trusted = is_project_mcp_trusted(project_root, fingerprint)
if project_trusted is True:
configs.append(config)
continue
# Untrusted project config: drop ALL servers (stdio + remote). Remote
# entries from an attacker-controlled .mcp.json can SSRF localhost or
# cloud metadata endpoints during the preflight HEAD probe, and any
# `${VAR}` references in their `headers` would exfiltrate the value
# to the attacker URL during the discovery handshake.
skipped = [
f"- {name} [{kind}]: {summary}" for name, kind, summary in project_servers
]
skipped_list = "\n".join(skipped)
if trust_project_mcp is False:
logger.warning(
"Skipped untrusted project MCP servers:\n%s",
skipped_list,
)
config_trusted = True
elif trust_project_mcp is False:
config_trusted = False
else:
logger.warning(
"Skipped untrusted project MCP servers "
"(config changed or not yet approved):\n%s",
skipped_list,
if project_trusted is None:
from deepagents_code.mcp_trust import (
compute_config_fingerprint,
is_project_mcp_trusted,
)
project_root = str(
_resolve_project_config_base(project_context).resolve()
)
fingerprint = compute_config_fingerprint(project_configs)
project_trusted = is_project_mcp_trusted(project_root, fingerprint)
config_trusted = project_trusted
# The allow/deny lists are sourced only from the user's own config (home
# config.toml + env) — never from the repo — so a committed .mcp.json
# cannot self-approve. Loaded lazily and reused across project configs.
if trust_lists is None:
from deepagents_code.model_config import (
DEFAULT_CONFIG_PATH,
load_mcp_server_trust_lists,
)
trust_lists = load_mcp_server_trust_lists()
if trust_lists.read_error is not None:
# Surface the read failure as a visible config error (a bare
# logger.warning has no handler outside debug mode).
config_load_errors.append((DEFAULT_CONFIG_PATH, trust_lists.read_error))
if trust_lists.load_failed:
# Fail closed: the user's allow/deny policy could not be read, so do
# not honor whole-config trust — otherwise a server the user meant to
# deny would load. Names explicitly enabled via a readable source
# (shell env) still survive the filter below.
config_trusted = False
# Keep only servers that survive the trust decision. Dropping the rest
# here (rather than loading all or none) preserves the SSRF/header-
# exfiltration gate: a non-allowlisted remote entry from an attacker-
# controlled .mcp.json never reaches the preflight HEAD probe or the
# `${VAR}` header interpolation during the discovery handshake.
servers = config["mcpServers"]
kept: dict[str, Any] = {}
for name, server in servers.items():
if name in trust_lists.disabled:
# Explicit reject always wins, even for a trusted config.
continue
if config_trusted or name in trust_lists.enabled:
kept[name] = server
if kept:
filtered = {**config, "mcpServers": kept}
try:
_validate_mcp_config_servers(filtered)
except (ValueError, TypeError, RuntimeError) as exc:
# The whole filtered config is dropped, so name the kept
# (trusted/allowlisted) servers that will NOT load — otherwise
# they vanish silently (the skip-log below only covers servers
# dropped by the trust decision, not by this validation failure).
logger.warning(
"Skipping invalid MCP config %s after project trust "
"filtering; these trusted/allowlisted servers will not "
"load: %s (%s)",
path,
", ".join(kept),
exc,
)
config_load_errors.append((path, str(exc)))
else:
configs.append(filtered)
# Servers dropped by the trust *decision* (disabled, or not allowlisted
# in an untrusted config). A validation failure above is reported
# separately, so those kept-but-unloaded names are intentionally not
# re-listed here with a trust-based reason.
dropped = [
(name, kind, summary)
for name, kind, summary in project_servers
if name not in kept
]
if dropped:
_log_skipped_project_servers(
dropped,
trust_project_mcp=trust_project_mcp,
config_trusted=config_trusted,
)
if explicit_config_path:
libs/code/deepagents_code/model_config.py
+243
View File
@@ -3046,6 +3046,249 @@ def unsuppress_warning(key: str, config_path: Path | None = None) -> bool:
return True
@dataclass(frozen=True)
class McpServerTrustLists:
"""User-level allow/deny lists for project MCP servers, by server name.
Sourced only from the user's own configuration — the home `config.toml`, the
global `~/.deepagents/.env`, and shell-exported env — never from a repo, so a
committed `.mcp.json` cannot self-approve. A committed *project* `.env` is
specifically prevented from setting the env forms of these lists (see
`config._PROJECT_DOTENV_DENIED_ENV_KEYS`). See `load_mcp_server_trust_lists`.
The "reject wins" invariant — a name in both lists is only rejected — is
enforced in `__post_init__`, so every instance is disjoint no matter how it
was constructed; callers need not pre-subtract.
"""
enabled: frozenset[str]
"""Server names pre-approved to load from an untrusted project config."""
disabled: frozenset[str]
"""Server names always rejected; reject wins over `enabled` and over trust."""
read_error: str | None = field(default=None, compare=False)
"""Non-`None` when the user's `config.toml` existed but its trust policy
could not be fully read: the file was unreadable/unparseable, its `[mcp]`
value was not a table, or its `disabled_project_servers` was a wrong type
that could not be interpreted as a deny list. Callers must treat this as
fail-closed (do not grant whole-config project trust) and surface it, rather
than proceeding with a deny list that may not have loaded — use `load_failed`
for that check. Note the resolved `enabled`/`disabled` sets are not
necessarily empty here: names from a still-readable source (the env vars)
continue to apply. Excluded from equality so a failed load still compares
equal to empty lists for tests that only care about the resolved names."""
def __post_init__(self) -> None:
"""Enforce reject precedence by removing disabled names from enabled.
A rejected name must never survive in `enabled`, whatever the caller
passed, so a future allow-first consumer can't be tricked into loading
a denied server. Frozen dataclass, so assign via `object.__setattr__`.
"""
if self.enabled & self.disabled:
object.__setattr__(self, "enabled", self.enabled - self.disabled)
@property
def load_failed(self) -> bool:
"""Whether the user's trust policy failed to load (see `read_error`).
Callers gating on trust MUST check this and fail closed: a failed load
means a configured deny may be missing, so whole-config project trust
must not be honored. Named so the fail-closed contract is discoverable
rather than resting on every caller remembering the `read_error`
sentinel.
"""
return self.read_error is not None
def _parse_csv_env(name: str) -> list[str] | None:
"""Parse a comma-separated env var into a list of trimmed, non-empty names.
Returns:
The parsed list when the variable is set (possibly empty after
trimming), or `None` when the variable is unset so callers can
distinguish "unset, fall back to TOML" from "set but empty".
"""
raw = os.environ.get(name)
if raw is None:
return None
return [item.strip() for item in raw.split(",") if item.strip()]
def _toml_str_list(
value: object, *, key: str, config_path: Path
) -> tuple[list[str], bool]:
"""Coerce a raw TOML value into a list of trimmed, non-empty server names.
A bare string is *split on commas* (e.g. `disabled_project_servers = "a, b"`
yields `["a", "b"]`), so a scalar written in the TOML parses identically to
the comma-separated env form in `_parse_csv_env` — the two forms can never
silently diverge into one bogus `"a, b"` token that matches no server. Non-
string list elements are dropped (with a log) while the surrounding valid
names survive. A genuinely wrong type (number, table, bool) cannot be
interpreted as names at all: it yields an empty list *and* flags `malformed`,
so a caller enforcing a deny list can fail closed rather than silently drop
the rejection.
Args:
value: The raw value read from the `[mcp]` table (or `None` when the
key is absent).
key: The TOML key name, used only for log context.
config_path: The config file the value came from, for log context.
Returns:
`(names, malformed)`. `names` are the trimmed, non-empty server names.
`malformed` is `True` only when `value` is present but neither a
string nor a list (so it could not be read as names); it is `False`
for an absent value, a string, or any list — even one whose non-
string elements were dropped.
"""
if value is None:
return [], False
if isinstance(value, str):
# Split on commas so a bare string parses exactly like the env form; a
# single name with no comma still yields a one-element list.
return [item.strip() for item in value.split(",") if item.strip()], False
if not isinstance(value, list):
logger.warning(
"[mcp].%s in %s should be a list of strings, got %s; ignoring it",
key,
config_path,
type(value).__name__,
)
return [], True
result: list[str] = []
discarded = 0
for item in value:
if isinstance(item, str) and item.strip():
result.append(item.strip())
else:
discarded += 1
if discarded:
logger.warning(
"[mcp].%s in %s: ignored %d non-string or empty entr%s",
key,
config_path,
discarded,
"y" if discarded == 1 else "ies",
)
return result, False
def load_mcp_server_trust_lists(
config_path: Path | None = None,
) -> McpServerTrustLists:
"""Load per-server project MCP allow/deny lists from user-level config.
Security boundary: this reads the `[mcp]` table only from the user-level
`config.toml` (`DEFAULT_CONFIG_PATH`, i.e. `~/.deepagents/config.toml`) and
the `DEEPAGENTS_CODE_ENABLED_PROJECT_MCP_SERVERS` /
`DEEPAGENTS_CODE_DISABLED_PROJECT_MCP_SERVERS` process env vars — never from
a project's `.mcp.json` or any repo-committed file. There is no
project-level `config.toml` discovery, so an attacker who commits a
malicious `.mcp.json` plus an in-repo config cannot pre-approve their own
servers; the approval must live in the user's home config. This mirrors
Claude Code's "untrusted folder → only non-checked-in settings" rule.
Source resolution differs by list, matching each one's security direction:
- `enabled` (permissive): the env var, when set, *replaces* the TOML list
(env-beats-config, as elsewhere). Clearing it via an empty env value is
fail-closed — it only ever pre-approves fewer servers.
- `disabled` (restrictive): the env var *unions* with the TOML list — denies
accumulate and a lower-effort source can never silently empty a deny
entry set in the other, which would be a fail-open. There is
deliberately no way to *remove* a configured deny via env.
Rejection wins: a name appearing in both the enabled and disabled result is
reported only in `disabled`.
Args:
config_path: Config file to read. Defaults to `DEFAULT_CONFIG_PATH`;
callers should not point this at a project path — doing so would
defeat the boundary above.
Returns:
The resolved `McpServerTrustLists`. A missing file yields empty lists
(the normal "unset" case). `read_error` is set (so callers can fail
closed instead of treating a broken config as "nothing denied") when
the file exists but cannot be read/parsed, when `[mcp]` is not a
table, or when `disabled_project_servers` is a wrong type that cannot
be read as a deny list; env-sourced names still apply in that case.
"""
if config_path is None:
config_path = DEFAULT_CONFIG_PATH
toml_enabled: list[str] = []
toml_disabled: list[str] = []
read_error: str | None = None
try:
if config_path.exists():
with config_path.open("rb") as f:
data = tomllib.load(f)
mcp_section = data.get("mcp", {})
if isinstance(mcp_section, dict):
# A wrong-typed `enabled` value degrades to an empty allowlist:
# approving nothing extra is already fail-closed, so the
# `malformed` flag is intentionally ignored here.
toml_enabled, _ = _toml_str_list(
mcp_section.get("enabled_project_servers"),
key="enabled_project_servers",
config_path=config_path,
)
toml_disabled, disabled_malformed = _toml_str_list(
mcp_section.get("disabled_project_servers"),
key="disabled_project_servers",
config_path=config_path,
)
if disabled_malformed:
# A wrong-typed deny list cannot be read, so proceeding as
# if nothing were denied would be a fail-open. Surface it and
# fail closed, mirroring the unreadable-file path below.
read_error = (
f"[mcp].disabled_project_servers in {config_path} must be "
"a list of strings; refusing to proceed with an "
"unenforced deny list"
)
else:
# An `[mcp]` value that is not a table means the deny list is
# unreadable too; fail closed rather than leave it unenforced.
read_error = (
f"[mcp] in {config_path} must be a table, got "
f"{type(mcp_section).__name__}"
)
logger.warning(
"[mcp] in %s should be a table, got %s; treating project "
"configs as untrusted",
config_path,
type(mcp_section).__name__,
)
except (OSError, tomllib.TOMLDecodeError) as exc:
# The file exists but is unreadable/unparseable. Record it so callers
# fail closed rather than silently proceeding with an empty deny list.
read_error = f"Could not read MCP trust lists from {config_path}: {exc}"
logger.warning(
"Could not read %s for MCP server trust lists; treating project "
"configs as untrusted",
config_path,
exc_info=True,
)
env_enabled = _parse_csv_env(_env_vars.ENABLED_PROJECT_MCP_SERVERS)
env_disabled = _parse_csv_env(_env_vars.DISABLED_PROJECT_MCP_SERVERS)
# Enabled: env replaces TOML. Disabled: env unions with TOML (denies
# accumulate; env can add a deny but never clear a configured one).
enabled = frozenset(env_enabled if env_enabled is not None else toml_enabled)
disabled = frozenset(toml_disabled) | frozenset(env_disabled or ())
# Reject precedence (a name in both lists ends up only in `disabled`) is
# enforced by `McpServerTrustLists.__post_init__`, so no subtraction here.
return McpServerTrustLists(
enabled=enabled, disabled=disabled, read_error=read_error
)
THREAD_COLUMN_DEFAULTS: dict[str, bool] = {
"thread_id": False,
"messages": True,
libs/code/tests/unit_tests/test_main.py
+221
View File
@@ -2379,6 +2379,227 @@ class TestCheckMcpProjectTrustPrompt:
assert decision is True
assert "could not be saved" in capsys.readouterr().err
def test_all_servers_list_resolved_shows_context_without_prompt(
self,
capsys: pytest.CaptureFixture[str],
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""List-resolved server rows escape project-controlled Rich markup."""
from deepagents_code import model_config
from deepagents_code.main import _check_mcp_project_trust
project_root = tmp_path / "proj"
project_root.mkdir()
project_cfg = project_root / ".mcp.json"
project_cfg.write_text("{}")
user_config = tmp_path / "config.toml"
user_config.write_text(
"[mcp]\n"
'enabled_project_servers = ["docs[/green]"]\n'
'disabled_project_servers = ["blocked[/red]"]\n'
)
monkeypatch.setattr(model_config, "DEFAULT_CONFIG_PATH", user_config)
project_context = SimpleNamespace(
project_root=project_root, user_cwd=project_root
)
def _no_input(_prompt: str = "") -> str:
msg = "prompt must be skipped when all servers are list-resolved"
raise AssertionError(msg)
with (
patch(
"deepagents_code.project_utils.ProjectContext.from_user_cwd",
return_value=project_context,
),
patch(
"deepagents_code.mcp_tools.discover_mcp_configs",
return_value=[project_cfg],
),
patch(
"deepagents_code.mcp_tools.classify_discovered_configs",
return_value=([], [project_cfg]),
),
patch(
"deepagents_code.mcp_tools.load_mcp_config_lenient",
return_value={
"mcpServers": {
"docs[/green]": {"command": "echo"},
"blocked[/red]": {"command": "echo"},
}
},
),
patch(
"deepagents_code.mcp_tools.extract_project_server_summaries",
return_value=[
("docs[/green]", "stdio[/green]", "echo [/green]"),
("blocked[/red]", "http[/red]", "https://x.test/[/red]"),
],
),
patch(
"deepagents_code.mcp_trust.is_project_mcp_trusted",
return_value=False,
),
patch("builtins.input", _no_input),
):
decision = _check_mcp_project_trust(trust_flag=False)
assert decision is None
err = capsys.readouterr().err
flattened = err.replace("\n", "")
# No approval question, but the config's decisions are surfaced.
assert "require approval" not in err
assert "Resolved by your config" in err
assert (
'"docs[/green]" (stdio[/green]): pre-approved '
"(enabled_project_servers): echo [/green]" in flattened
)
assert (
'"blocked[/red]" (http[/red]): blocked '
"(disabled_project_servers): https://x.test/[/red]" in flattened
)
def test_prompt_asks_only_about_unlisted_but_shows_preapproved(
self,
capsys: pytest.CaptureFixture[str],
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""The prompt asks only about unlisted servers; pre-approved ones show."""
from deepagents_code import model_config
from deepagents_code.main import _check_mcp_project_trust
project_root = tmp_path / "proj"
project_root.mkdir()
project_cfg = project_root / ".mcp.json"
project_cfg.write_text("{}")
user_config = tmp_path / "config.toml"
user_config.write_text('[mcp]\nenabled_project_servers = ["docs"]\n')
monkeypatch.setattr(model_config, "DEFAULT_CONFIG_PATH", user_config)
project_context = SimpleNamespace(
project_root=project_root, user_cwd=project_root
)
with (
patch(
"deepagents_code.project_utils.ProjectContext.from_user_cwd",
return_value=project_context,
),
patch(
"deepagents_code.mcp_tools.discover_mcp_configs",
return_value=[project_cfg],
),
patch(
"deepagents_code.mcp_tools.classify_discovered_configs",
return_value=([], [project_cfg]),
),
patch(
"deepagents_code.mcp_tools.load_mcp_config_lenient",
return_value={
"mcpServers": {
"docs": {"command": "echo"},
"other": {"command": "echo"},
}
},
),
patch(
"deepagents_code.mcp_tools.extract_project_server_summaries",
return_value=[
("docs", "stdio", "echo docs"),
("other", "stdio", "echo other"),
],
),
patch(
"deepagents_code.mcp_trust.is_project_mcp_trusted",
return_value=False,
),
patch("builtins.input", return_value="n"),
):
decision = _check_mcp_project_trust(trust_flag=False)
assert decision is False
err = capsys.readouterr().err
# The unlisted server is the one actually being asked about.
assert ' "other" (stdio): echo other' in err
# The pre-approved server is shown as resolved, not asked about.
assert (
' "docs" (stdio): pre-approved (enabled_project_servers): echo docs'
in err
)
def test_unreadable_policy_fails_closed_without_prompting(
self,
capsys: pytest.CaptureFixture[str],
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""A corrupt user config.toml makes the prompt fail closed (return False).
The allow/deny policy could not be read, so the prompt must not ask (and
possibly persist trust) under an unknown deny list; it warns and denies,
matching the loader's fail-closed behavior.
"""
from deepagents_code import model_config
from deepagents_code.main import _check_mcp_project_trust
project_root = tmp_path / "proj"
project_root.mkdir()
project_cfg = project_root / ".mcp.json"
project_cfg.write_text("{}")
user_config = tmp_path / "config.toml"
user_config.write_text("[[not valid toml")
monkeypatch.setattr(model_config, "DEFAULT_CONFIG_PATH", user_config)
project_context = SimpleNamespace(
project_root=project_root, user_cwd=project_root
)
def _no_input(_prompt: str = "") -> str:
msg = "prompt must be skipped when the trust policy is unreadable"
raise AssertionError(msg)
with (
patch(
"deepagents_code.project_utils.ProjectContext.from_user_cwd",
return_value=project_context,
),
patch(
"deepagents_code.mcp_tools.discover_mcp_configs",
return_value=[project_cfg],
),
patch(
"deepagents_code.mcp_tools.classify_discovered_configs",
return_value=([], [project_cfg]),
),
patch(
"deepagents_code.mcp_tools.load_mcp_config_lenient",
return_value={"mcpServers": {"docs": {"command": "echo"}}},
),
patch(
"deepagents_code.mcp_tools.extract_project_server_summaries",
return_value=[("docs", "stdio", "echo docs")],
),
patch(
"deepagents_code.mcp_trust.is_project_mcp_trusted",
return_value=False,
),
patch("builtins.input", _no_input),
):
decision = _check_mcp_project_trust(trust_flag=False)
assert decision is False
err = capsys.readouterr().err
# Rich may wrap the warning across lines; flatten before matching.
flattened = err.replace("\n", "")
assert "treating project MCP servers as untrusted" in flattened
assert "require approval" not in err
class TestCheckMcpProjectTrustDedupe:
"""Regression tests for the project MCP approval prompt deduplication.
libs/code/tests/unit_tests/test_mcp_disabled.py
+59 -2
View File
@@ -22,13 +22,45 @@ class TestGetDisabledServers:
assert get_disabled_servers(config_path=cfg) == set()
def test_reads_existing_entries(self, tmp_path: Path) -> None:
cfg = tmp_path / "config.toml"
cfg.write_text('[mcp]\ndisabled_servers = ["github", "slack"]\n')
assert get_disabled_servers(config_path=cfg) == {"github", "slack"}
def test_reads_legacy_entries_when_folded_key_missing(self, tmp_path: Path) -> None:
cfg = tmp_path / "config.toml"
cfg.write_text('[mcp_disabled]\nservers = ["github", "slack"]\n')
assert get_disabled_servers(config_path=cfg) == {"github", "slack"}
def test_folded_key_wins_over_legacy_entries(self, tmp_path: Path) -> None:
cfg = tmp_path / "config.toml"
cfg.write_text(
'[mcp]\ndisabled_servers = ["github"]\n'
'[mcp_disabled]\nservers = ["slack"]\n'
)
assert get_disabled_servers(config_path=cfg) == {"github"}
def test_empty_folded_key_shadows_legacy(self, tmp_path: Path) -> None:
# An empty (but present) folded list is authoritative: once the new
# shape exists it is the source of truth, so legacy is not consulted.
cfg = tmp_path / "config.toml"
cfg.write_text(
'[mcp]\ndisabled_servers = []\n[mcp_disabled]\nservers = ["slack"]\n'
)
assert get_disabled_servers(config_path=cfg) == set()
def test_malformed_folded_key_falls_back_to_legacy(self, tmp_path: Path) -> None:
# A wrong-typed folded value is treated as "unset" (not "empty"), so the
# legacy list still applies. This is a best-effort convenience list, not
# a security deny list, so falling back rather than failing closed is fine.
cfg = tmp_path / "config.toml"
cfg.write_text(
'[mcp]\ndisabled_servers = "github"\n[mcp_disabled]\nservers = ["slack"]\n'
)
assert get_disabled_servers(config_path=cfg) == {"slack"}
def test_filters_non_string_entries(self, tmp_path: Path) -> None:
cfg = tmp_path / "config.toml"
cfg.write_text('[mcp_disabled]\nservers = ["ok", ""]\n')
cfg.write_text('[mcp]\ndisabled_servers = ["ok", ""]\n')
assert get_disabled_servers(config_path=cfg) == {"ok"}
def test_returns_empty_on_corrupt_toml(self, tmp_path: Path) -> None:
@@ -75,7 +107,32 @@ class TestSetServerDisabled:
contents = cfg.read_text()
assert "[other]" in contents
assert 'key = "value"' in contents
assert "[mcp_disabled]" in contents
assert "[mcp]" in contents
assert "disabled_servers" in contents
assert "github" in contents
def test_preserves_existing_mcp_keys(self, tmp_path: Path) -> None:
cfg = tmp_path / "config.toml"
cfg.write_text('[mcp]\nenabled_project_servers = ["docs"]\n')
ok, detail = set_server_disabled("github", True, config_path=cfg)
assert ok
assert detail is None
contents = cfg.read_text()
assert "enabled_project_servers" in contents
assert "docs" in contents
assert "disabled_servers" in contents
assert "github" in contents
def test_migrates_legacy_section_on_write(self, tmp_path: Path) -> None:
cfg = tmp_path / "config.toml"
cfg.write_text('[mcp_disabled]\nservers = ["github"]\n')
ok, detail = set_server_disabled("slack", True, config_path=cfg)
assert ok
assert detail is None
contents = cfg.read_text()
assert "[mcp_disabled]" not in contents
assert "[mcp]" in contents
assert get_disabled_servers(config_path=cfg) == {"github", "slack"}
def test_entries_sorted(self, tmp_path: Path) -> None:
cfg = tmp_path / "config.toml"
libs/code/tests/unit_tests/test_mcp_tools.py
+462
View File
@@ -1697,6 +1697,37 @@ class TestResolveAndLoadMcpTools:
with pytest.raises(json.JSONDecodeError):
await resolve_and_load_mcp_tools(explicit_config_path=str(bad))
@patch("deepagents_code.mcp_tools._load_tools_from_config")
@patch("deepagents_code.mcp_tools.classify_discovered_configs")
@patch("deepagents_code.mcp_tools.discover_mcp_configs")
async def test_malformed_project_config_without_summaries_is_nonfatal(
self,
mock_discover: MagicMock,
mock_classify: MagicMock,
mock_load: AsyncMock,
tmp_path: Path,
) -> None:
"""Malformed-only project configs are reported instead of crashing."""
project_cfg = tmp_path / ".mcp.json"
project_cfg.write_text(
json.dumps({"mcpServers": {"bad": ["not", "a", "dict"]}})
)
mock_discover.return_value = [project_cfg]
mock_classify.return_value = ([], [project_cfg])
mock_load.return_value = ([], None, [])
tools, manager, infos = await resolve_and_load_mcp_tools(
trust_project_mcp=True,
)
assert tools == []
assert manager is None
assert mock_load.call_count == 0
assert len(infos) == 1
assert infos[0].name == "<config:.mcp.json>"
assert infos[0].status == "error"
assert "must be a dictionary" in (infos[0].error or "")
@patch("deepagents_code.mcp_tools._load_tools_from_config")
@patch("deepagents_code.mcp_tools.classify_discovered_configs")
@patch("deepagents_code.mcp_tools.discover_mcp_configs")
@@ -1706,6 +1737,7 @@ class TestResolveAndLoadMcpTools:
mock_classify: MagicMock,
mock_load: AsyncMock,
tmp_path: Path,
monkeypatch: pytest.MonkeyPatch,
caplog: pytest.LogCaptureFixture,
) -> None:
"""Project remote MCP entries do not reach the loader without trust.
@@ -1734,6 +1766,14 @@ class TestResolveAndLoadMcpTools:
mock_discover.return_value = [project_cfg]
mock_classify.return_value = ([], [project_cfg])
mock_load.return_value = ([], None, [])
monkeypatch.setattr(
"deepagents_code.model_config.DEFAULT_CONFIG_PATH",
tmp_path / "config.toml",
)
monkeypatch.delenv("DEEPAGENTS_CODE_ENABLED_PROJECT_MCP_SERVERS", raising=False)
monkeypatch.delenv(
"DEEPAGENTS_CODE_DISABLED_PROJECT_MCP_SERVERS", raising=False
)
caplog.set_level(logging.WARNING, logger="deepagents_code.mcp_tools")
tools, _manager, _infos = await resolve_and_load_mcp_tools(
@@ -3136,3 +3176,425 @@ class TestNormalizeMCPArguments:
"dropped empty-string keys" in r.message and "ctx" in r.message
for r in caplog.records
)
class TestSelectiveProjectMcpTrust:
"""Per-server allow/deny filtering of project MCP servers.
The user-level allow/deny lists are honored only from the user's own
`config.toml` (via `DEFAULT_CONFIG_PATH`), never from a repo-committed
file, and only allowlisted (or fully trusted) names reach the loader — so
the SSRF/exfiltration gate on untrusted remote entries stays intact.
"""
@staticmethod
def _write_project_config(project_root: Path, servers: dict[str, Any]) -> None:
(project_root / ".mcp.json").write_text(json.dumps({"mcpServers": servers}))
@staticmethod
def _stdio(command: str = "echo") -> dict[str, Any]:
return {"command": command, "args": []}
@staticmethod
def _remote(url: str = "https://example.test/mcp") -> dict[str, Any]:
return {"type": "sse", "url": url}
async def _resolve_merged(
self,
project_root: Path,
monkeypatch: pytest.MonkeyPatch,
*,
user_config: Path,
trust_project_mcp: bool | None,
) -> dict[str, Any] | None:
"""Run resolution and return the merged config passed to the loader.
Returns `None` when the loader is never reached (i.e. every project
server was dropped and no other config remained).
"""
# Isolate discovery and the trust store from the developer's real home.
home = project_root.parent / "home"
(home / ".deepagents").mkdir(parents=True, exist_ok=True)
monkeypatch.setenv("HOME", str(home))
monkeypatch.setattr(
"deepagents_code.model_config.DEFAULT_CONFIG_PATH", user_config
)
loader = AsyncMock(return_value=([], None, []))
monkeypatch.setattr("deepagents_code.mcp_tools._load_tools_from_config", loader)
ctx = ProjectContext(user_cwd=project_root, project_root=project_root)
await resolve_and_load_mcp_tools(
project_context=ctx,
trust_project_mcp=trust_project_mcp,
)
if not loader.called:
return None
return loader.call_args.args[0]
async def test_allowlisted_loads_and_sibling_dropped(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""An allowlisted server loads while a non-listed sibling is dropped."""
project = tmp_path / "project"
project.mkdir()
self._write_project_config(
project, {"docs": self._stdio(), "other": self._stdio()}
)
user_config = tmp_path / "config.toml"
user_config.write_text('[mcp]\nenabled_project_servers = ["docs"]\n')
merged = await self._resolve_merged(
project, monkeypatch, user_config=user_config, trust_project_mcp=False
)
assert merged is not None
assert set(merged["mcpServers"]) == {"docs"}
async def test_allowlisted_loads_with_invalid_unlisted_sibling(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""An invalid unlisted server cannot block an allowlisted sibling."""
project = tmp_path / "project"
project.mkdir()
self._write_project_config(
project,
{
"docs": self._stdio(),
"broken": ["not", "a", "server"],
},
)
user_config = tmp_path / "config.toml"
user_config.write_text('[mcp]\nenabled_project_servers = ["docs"]\n')
merged = await self._resolve_merged(
project, monkeypatch, user_config=user_config, trust_project_mcp=False
)
assert merged is not None
assert set(merged["mcpServers"]) == {"docs"}
async def test_disabled_dropped_even_when_trusted(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""An explicitly disabled server is dropped even from a trusted config."""
project = tmp_path / "project"
project.mkdir()
self._write_project_config(
project, {"docs": self._stdio(), "blocked": self._stdio()}
)
user_config = tmp_path / "config.toml"
user_config.write_text('[mcp]\ndisabled_project_servers = ["blocked"]\n')
merged = await self._resolve_merged(
project, monkeypatch, user_config=user_config, trust_project_mcp=True
)
assert merged is not None
assert set(merged["mcpServers"]) == {"docs"}
async def test_disabled_invalid_server_dropped_before_validation(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""An invalid disabled server cannot block a trusted sibling."""
project = tmp_path / "project"
project.mkdir()
self._write_project_config(
project,
{
"docs": self._stdio(),
"blocked": ["not", "a", "server"],
},
)
user_config = tmp_path / "config.toml"
user_config.write_text('[mcp]\ndisabled_project_servers = ["blocked"]\n')
merged = await self._resolve_merged(
project, monkeypatch, user_config=user_config, trust_project_mcp=True
)
assert merged is not None
assert set(merged["mcpServers"]) == {"docs"}
async def test_repo_committed_allowlist_is_ignored(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""An allowlist committed in the repo does not self-approve servers."""
project = tmp_path / "project"
project.mkdir()
self._write_project_config(project, {"evil": self._stdio()})
# Attacker-committed project config that tries to approve its own server.
(project / "config.toml").write_text(
'[mcp]\nenabled_project_servers = ["evil"]\n'
)
(project / ".deepagents").mkdir()
(project / ".deepagents" / "config.toml").write_text(
'[mcp]\nenabled_project_servers = ["evil"]\n'
)
# User has no allowlist of their own.
user_config = tmp_path / "config.toml"
merged = await self._resolve_merged(
project, monkeypatch, user_config=user_config, trust_project_mcp=False
)
# The repo allowlist is never read, so the server stays dropped.
assert merged is None
async def test_name_in_both_lists_is_disabled(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A server named in both lists is disabled (reject precedence)."""
project = tmp_path / "project"
project.mkdir()
self._write_project_config(project, {"both": self._stdio()})
user_config = tmp_path / "config.toml"
user_config.write_text(
"[mcp]\n"
'enabled_project_servers = ["both"]\n'
'disabled_project_servers = ["both"]\n'
)
merged = await self._resolve_merged(
project, monkeypatch, user_config=user_config, trust_project_mcp=False
)
assert merged is None
async def test_malformed_table_falls_back_to_full_drop(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A wrong-typed enabled value drops all untrusted servers, no crash.
(A bare string is coerced to a single name, so use a genuinely wrong
type — an integer — which degrades to an empty allowlist and full drop.)
"""
project = tmp_path / "project"
project.mkdir()
self._write_project_config(project, {"docs": self._stdio()})
user_config = tmp_path / "config.toml"
user_config.write_text(
"[mcp]\nenabled_project_servers = 123\n"
) # not a list or string
merged = await self._resolve_merged(
project, monkeypatch, user_config=user_config, trust_project_mcp=False
)
assert merged is None
async def test_env_allowlist_honored(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""The env allowlist approves a server with no TOML allowlist set."""
project = tmp_path / "project"
project.mkdir()
self._write_project_config(
project, {"docs": self._stdio(), "other": self._stdio()}
)
user_config = tmp_path / "config.toml" # no [mcp] table
monkeypatch.setenv("DEEPAGENTS_CODE_ENABLED_PROJECT_MCP_SERVERS", "docs")
merged = await self._resolve_merged(
project, monkeypatch, user_config=user_config, trust_project_mcp=False
)
assert merged is not None
assert set(merged["mcpServers"]) == {"docs"}
async def test_allowlisted_remote_kept_and_sibling_dropped(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Remote entries are gated by name too: only the allowlisted one loads.
This is the SSRF/exfiltration case the design exists for — a
non-allowlisted remote entry must never survive into the merged config
(and so never reach the preflight probe or `${VAR}` header resolution).
"""
project = tmp_path / "project"
project.mkdir()
self._write_project_config(
project, {"docs": self._remote(), "evil": self._remote()}
)
user_config = tmp_path / "config.toml"
user_config.write_text('[mcp]\nenabled_project_servers = ["docs"]\n')
merged = await self._resolve_merged(
project, monkeypatch, user_config=user_config, trust_project_mcp=False
)
assert merged is not None
assert set(merged["mcpServers"]) == {"docs"}
assert merged["mcpServers"]["docs"]["type"] == "sse"
async def test_allow_and_deny_combined_in_untrusted_config(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""One untrusted config: allowed kept, denied and unlisted both dropped."""
project = tmp_path / "project"
project.mkdir()
self._write_project_config(
project,
{
"keep": self._stdio(),
"block": self._stdio(),
"other": self._stdio(),
},
)
user_config = tmp_path / "config.toml"
user_config.write_text(
"[mcp]\n"
'enabled_project_servers = ["keep"]\n'
'disabled_project_servers = ["block"]\n'
)
merged = await self._resolve_merged(
project, monkeypatch, user_config=user_config, trust_project_mcp=False
)
assert merged is not None
assert set(merged["mcpServers"]) == {"keep"}
async def test_disabled_dropped_when_fingerprint_trusted(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Deny wins over fingerprint trust, not just the explicit flag."""
project = tmp_path / "project"
project.mkdir()
self._write_project_config(
project, {"docs": self._stdio(), "blocked": self._stdio()}
)
user_config = tmp_path / "config.toml"
user_config.write_text('[mcp]\ndisabled_project_servers = ["blocked"]\n')
# Fingerprint store reports the whole config as trusted.
monkeypatch.setattr(
"deepagents_code.mcp_trust.is_project_mcp_trusted", lambda *_a, **_k: True
)
merged = await self._resolve_merged(
project, monkeypatch, user_config=user_config, trust_project_mcp=None
)
assert merged is not None
assert set(merged["mcpServers"]) == {"docs"}
async def test_allowlisted_loads_when_fingerprint_untrusted(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""On the `None` path with untrusted fingerprint, allowlist still loads."""
project = tmp_path / "project"
project.mkdir()
self._write_project_config(
project, {"docs": self._stdio(), "other": self._stdio()}
)
user_config = tmp_path / "config.toml"
user_config.write_text('[mcp]\nenabled_project_servers = ["docs"]\n')
monkeypatch.setattr(
"deepagents_code.mcp_trust.is_project_mcp_trusted", lambda *_a, **_k: False
)
merged = await self._resolve_merged(
project, monkeypatch, user_config=user_config, trust_project_mcp=None
)
assert merged is not None
assert set(merged["mcpServers"]) == {"docs"}
async def test_allowlisted_but_invalid_server_is_nonfatal(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""An allowlisted server that is itself invalid is dropped, not fatal.
Exercises the deferred per-server validation branch: the kept subset is
validated after trust filtering, and a validation failure drops the
config rather than crashing resolution.
"""
project = tmp_path / "project"
project.mkdir()
# A dict (so it yields a summary and skips the empty-summaries fast path),
# but setting both tool filters at once — a documented per-server error.
self._write_project_config(
project,
{
"docs": {
"command": "echo",
"args": [],
"allowedTools": ["a"],
"disabledTools": ["b"],
}
},
)
user_config = tmp_path / "config.toml"
user_config.write_text('[mcp]\nenabled_project_servers = ["docs"]\n')
merged = await self._resolve_merged(
project, monkeypatch, user_config=user_config, trust_project_mcp=False
)
# Invalid kept server -> whole filtered config dropped -> loader unreached.
assert merged is None
async def test_unreadable_user_config_fails_closed_and_surfaces_error(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A corrupt user config.toml drops project servers even under --trust.
The allow/deny policy could not be read, so the loader records a
`read_error`; resolution then treats the project config as untrusted
(fail closed) and surfaces the error as an MCP config error rather than
loading a server the user might have meant to deny.
"""
project = tmp_path / "project"
project.mkdir()
self._write_project_config(project, {"docs": self._stdio()})
home = tmp_path / "home"
(home / ".deepagents").mkdir(parents=True, exist_ok=True)
monkeypatch.setenv("HOME", str(home))
user_config = tmp_path / "config.toml"
user_config.write_text("[[not valid toml")
monkeypatch.setattr(
"deepagents_code.model_config.DEFAULT_CONFIG_PATH", user_config
)
loader = AsyncMock(return_value=([], None, []))
monkeypatch.setattr("deepagents_code.mcp_tools._load_tools_from_config", loader)
ctx = ProjectContext(user_cwd=project, project_root=project)
_tools, _manager, infos = await resolve_and_load_mcp_tools(
project_context=ctx, trust_project_mcp=True
)
# Fail closed: even with trust_project_mcp=True, nothing loads.
assert loader.call_count == 0
# The read failure is surfaced (not just a debug-only warning).
assert any(
info.status == "error" and "config.toml" in (info.error or "")
for info in infos
)
async def test_env_enabled_survives_unreadable_user_config(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A read_error fails closed, but an env-enabled name still loads.
On a corrupt config.toml the loader forces the config untrusted, yet the
allowlist read from a still-readable source (the env var) applies: the
env-enabled server loads while a non-listed sibling is dropped. Pins the
"readable source (shell env) still survives" branch so a future hardening
that also empties `enabled` on read_error doesn't silently drop a server
the user explicitly allowlisted.
"""
project = tmp_path / "project"
project.mkdir()
self._write_project_config(
project, {"docs": self._stdio(), "other": self._stdio()}
)
user_config = tmp_path / "config.toml"
user_config.write_text("[[not valid toml")
monkeypatch.setenv("DEEPAGENTS_CODE_ENABLED_PROJECT_MCP_SERVERS", "docs")
merged = await self._resolve_merged(
project, monkeypatch, user_config=user_config, trust_project_mcp=True
)
assert merged is not None
assert set(merged["mcpServers"]) == {"docs"}
libs/code/tests/unit_tests/test_model_config.py
+323
View File
@@ -18,6 +18,7 @@ from deepagents_code.model_config import (
PROVIDER_BASE_URL_ENV,
RETRY_PARAM_BY_PROVIDER,
THREAD_COLUMN_DEFAULTS,
McpServerTrustLists,
ModelConfig,
ModelConfigError,
ModelProfileEntry,
@@ -39,6 +40,7 @@ from deepagents_code.model_config import (
has_provider_credentials,
is_warning_suppressed,
load_default_agent,
load_mcp_server_trust_lists,
load_recent_agent,
load_recent_models,
load_thread_columns,
@@ -5214,6 +5216,327 @@ class TestUnsuppressWarning:
assert not is_warning_suppressed("tavily", config_path)
class TestMcpServerTrustLists:
"""Tests for the McpServerTrustLists value object itself."""
def test_post_init_enforces_disjointness_on_direct_construction(self) -> None:
"""A name in both lists is dropped from enabled, however constructed.
The docstring promises the invariant holds "no matter how it was
constructed; callers need not pre-subtract" — pin that at the type level,
independent of the loader.
"""
lists = McpServerTrustLists(
enabled=frozenset({"keep", "both"}),
disabled=frozenset({"both"}),
)
assert lists.enabled == frozenset({"keep"})
assert lists.disabled == frozenset({"both"})
def test_read_error_excluded_from_equality(self) -> None:
"""`read_error` is diagnostic only and does not affect equality."""
assert McpServerTrustLists(
frozenset(), frozenset(), read_error="boom"
) == McpServerTrustLists(frozenset(), frozenset())
def test_load_failed_tracks_read_error(self) -> None:
"""`load_failed` names the fail-closed contract for `read_error`."""
assert not McpServerTrustLists(frozenset(), frozenset()).load_failed
assert McpServerTrustLists(
frozenset(), frozenset(), read_error="boom"
).load_failed
class TestLoadMcpServerTrustLists:
"""Tests for load_mcp_server_trust_lists()."""
def test_reads_both_lists_from_toml(self, tmp_path: Path) -> None:
"""Parses enabled and disabled lists from the [mcp] table."""
config_path = tmp_path / "config.toml"
config_path.write_text(
"[mcp]\n"
'enabled_project_servers = ["docs", "reference"]\n'
'disabled_project_servers = ["blocked"]\n'
)
result = load_mcp_server_trust_lists(config_path)
assert result == McpServerTrustLists(
enabled=frozenset({"docs", "reference"}),
disabled=frozenset({"blocked"}),
)
def test_reject_precedence_removes_from_enabled(self, tmp_path: Path) -> None:
"""A name in both lists is reported only as disabled."""
config_path = tmp_path / "config.toml"
config_path.write_text(
"[mcp]\n"
'enabled_project_servers = ["docs", "both"]\n'
'disabled_project_servers = ["both"]\n'
)
result = load_mcp_server_trust_lists(config_path)
assert result.enabled == frozenset({"docs"})
assert result.disabled == frozenset({"both"})
def test_missing_file_returns_empty(self, tmp_path: Path) -> None:
"""A missing config file yields empty lists, not an error.
A missing file is the normal "unset" case and must NOT set `read_error`
(that is reserved for a file that exists but cannot be read/parsed), so
callers do not fail closed just because the user has no config.toml.
"""
result = load_mcp_server_trust_lists(tmp_path / "nonexistent.toml")
assert result == McpServerTrustLists(frozenset(), frozenset())
assert result.read_error is None
def test_env_deny_beats_toml_allow_same_name(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Reject wins across sources: env-disabled beats TOML-enabled by name.
Proves the disjointness invariant runs on the final merged frozensets
(after env/TOML resolution), not merely within a single source.
"""
config_path = tmp_path / "config.toml"
config_path.write_text('[mcp]\nenabled_project_servers = ["srv"]\n')
monkeypatch.setenv(model_config._env_vars.DISABLED_PROJECT_MCP_SERVERS, "srv")
result = load_mcp_server_trust_lists(config_path)
assert result.enabled == frozenset()
assert result.disabled == frozenset({"srv"})
def test_missing_mcp_section_returns_empty(self, tmp_path: Path) -> None:
"""A config without an [mcp] table yields empty lists."""
config_path = tmp_path / "config.toml"
config_path.write_text('[models]\ndefault = "some:model"\n')
result = load_mcp_server_trust_lists(config_path)
assert result == McpServerTrustLists(frozenset(), frozenset())
def test_corrupt_toml_falls_back_to_empty_and_sets_read_error(
self, tmp_path: Path
) -> None:
"""Malformed TOML degrades to empty lists but records a read error.
The empty lists compare equal to a clean empty result (`read_error` is
excluded from equality), but `read_error` is set so callers can fail
closed instead of treating a broken config as "nothing denied."
"""
config_path = tmp_path / "config.toml"
config_path.write_text("[[invalid toml")
result = load_mcp_server_trust_lists(config_path)
assert result == McpServerTrustLists(frozenset(), frozenset())
assert result.read_error is not None
assert str(config_path) in result.read_error
def test_scalar_coerced_and_mixed_elements_dropped(self, tmp_path: Path) -> None:
"""A bare string is one name; non-string list elements are dropped."""
config_path = tmp_path / "config.toml"
config_path.write_text(
"[mcp]\n"
'enabled_project_servers = "docs"\n' # bare string -> single name
'disabled_project_servers = [1, "blocked", true]\n' # mixed types
)
result = load_mcp_server_trust_lists(config_path)
assert result.enabled == frozenset({"docs"})
assert result.disabled == frozenset({"blocked"})
def test_env_overrides_toml(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""Env lists replace their TOML counterparts, independently per list."""
config_path = tmp_path / "config.toml"
config_path.write_text(
"[mcp]\n"
'enabled_project_servers = ["toml-enabled"]\n'
'disabled_project_servers = ["toml-disabled"]\n'
)
monkeypatch.setenv(
model_config._env_vars.ENABLED_PROJECT_MCP_SERVERS,
"env-enabled, env-two",
)
result = load_mcp_server_trust_lists(config_path)
# Enabled comes from env; disabled falls back to the TOML value.
assert result.enabled == frozenset({"env-enabled", "env-two"})
assert result.disabled == frozenset({"toml-disabled"})
def test_empty_env_clears_toml_list(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A set-but-empty env var overrides (clears) the TOML list."""
config_path = tmp_path / "config.toml"
config_path.write_text('[mcp]\nenabled_project_servers = ["toml-enabled"]\n')
monkeypatch.setenv(model_config._env_vars.ENABLED_PROJECT_MCP_SERVERS, "")
result = load_mcp_server_trust_lists(config_path)
assert result.enabled == frozenset()
def test_defaults_to_user_config_path(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""With no argument, the loader reads the user-level config path only."""
user_config = tmp_path / "config.toml"
user_config.write_text('[mcp]\nenabled_project_servers = ["docs"]\n')
monkeypatch.setattr(model_config, "DEFAULT_CONFIG_PATH", user_config)
result = load_mcp_server_trust_lists()
assert result.enabled == frozenset({"docs"})
def test_disabled_env_honored_without_toml(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""The deny list can be set purely from the env var."""
config_path = tmp_path / "config.toml" # no [mcp] table
monkeypatch.setenv(
model_config._env_vars.DISABLED_PROJECT_MCP_SERVERS, "blocked, other"
)
result = load_mcp_server_trust_lists(config_path)
assert result.disabled == frozenset({"blocked", "other"})
def test_disabled_env_unions_with_toml(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""The disabled env list UNIONS with the TOML deny list (denies accrue).
Unlike the enabled list (env replaces TOML), a deny must never be
silently dropped by the other source, so both contribute.
"""
config_path = tmp_path / "config.toml"
config_path.write_text(
"[mcp]\n"
'enabled_project_servers = ["toml-enabled"]\n'
'disabled_project_servers = ["toml-disabled"]\n'
)
monkeypatch.setenv(
model_config._env_vars.DISABLED_PROJECT_MCP_SERVERS, "env-disabled"
)
result = load_mcp_server_trust_lists(config_path)
assert result.disabled == frozenset({"toml-disabled", "env-disabled"})
assert result.enabled == frozenset({"toml-enabled"})
def test_empty_disabled_env_preserves_toml_list(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A set-but-empty disabled env var cannot clear the TOML deny list.
Because disabled unions across sources, an empty env value contributes
nothing and the configured deny survives — closing the fail-open where
`DISABLED=""` (e.g. from an attacker-adjacent source) would silently
neutralize the user's deny list.
"""
config_path = tmp_path / "config.toml"
config_path.write_text('[mcp]\ndisabled_project_servers = ["toml-disabled"]\n')
monkeypatch.setenv(model_config._env_vars.DISABLED_PROJECT_MCP_SERVERS, "")
result = load_mcp_server_trust_lists(config_path)
assert result.disabled == frozenset({"toml-disabled"})
def test_toml_entries_are_trimmed(self, tmp_path: Path) -> None:
"""TOML names are stripped, matching env parsing (no whitespace mismatch)."""
config_path = tmp_path / "config.toml"
config_path.write_text('[mcp]\nenabled_project_servers = [" docs ", " "]\n')
result = load_mcp_server_trust_lists(config_path)
# " docs " -> "docs"; the whitespace-only " " entry is dropped.
assert result.enabled == frozenset({"docs"})
def test_bare_string_disabled_is_coerced_to_single_name(
self, tmp_path: Path
) -> None:
"""A bare-string deny list is one server name, not a dropped-to-empty typo.
Coercing (rather than silently dropping) is the safe direction for the
deny list: it keeps enforcing the user's rejection instead of failing
open on a scalar-instead-of-list mistake.
"""
config_path = tmp_path / "config.toml"
# Valid TOML, but a string rather than a list — the [mcp] table is still
# a dict, so the "should be a table" branch does not fire.
config_path.write_text('[mcp]\ndisabled_project_servers = "blocked"\n')
result = load_mcp_server_trust_lists(config_path)
assert result.disabled == frozenset({"blocked"})
def test_bare_string_disabled_splits_on_commas(self, tmp_path: Path) -> None:
"""A comma-separated bare-string deny list parses like the env form.
Without splitting, `"a, b"` would become one bogus token matching no
server and silently enforce nothing — a fail-open for the deny list.
"""
config_path = tmp_path / "config.toml"
config_path.write_text('[mcp]\ndisabled_project_servers = "evil, backdoor"\n')
result = load_mcp_server_trust_lists(config_path)
assert result.disabled == frozenset({"evil", "backdoor"})
assert result.read_error is None
def test_wrong_typed_disabled_fails_closed_with_read_error(
self, tmp_path: Path
) -> None:
"""A wrong-typed deny list sets `read_error` instead of silently emptying.
Emptying the deny on a malformed value would be a fail-open (the user's
rejection stops being enforced). Surfacing `read_error` lets callers fail
closed, matching the corrupt-file path.
"""
config_path = tmp_path / "config.toml"
config_path.write_text("[mcp]\ndisabled_project_servers = 123\n")
result = load_mcp_server_trust_lists(config_path)
assert result.disabled == frozenset()
assert result.load_failed
assert "disabled_project_servers" in (result.read_error or "")
def test_wrong_typed_enabled_stays_empty_without_read_error(
self, tmp_path: Path
) -> None:
"""A wrong-typed allow list degrades to empty (already fail-closed).
Unlike the deny list, an unreadable allow list approves nothing extra, so
it must NOT set `read_error` (which would block even trusted configs).
"""
config_path = tmp_path / "config.toml"
config_path.write_text("[mcp]\nenabled_project_servers = 123\n")
result = load_mcp_server_trust_lists(config_path)
assert result.enabled == frozenset()
assert result.read_error is None
def test_non_table_mcp_sets_read_error(self, tmp_path: Path) -> None:
"""An `[mcp]` value that is not a table fails closed (deny unreadable)."""
config_path = tmp_path / "config.toml"
config_path.write_text('mcp = "oops"\n')
result = load_mcp_server_trust_lists(config_path)
assert result == McpServerTrustLists(frozenset(), frozenset())
assert result.load_failed
class TestGetModelProfiles:
"""Tests for get_model_profiles() function."""
libs/code/tests/unit_tests/test_reload.py
+120
View File
@@ -436,6 +436,126 @@ class TestReloadFromEnvironment:
assert "DEEPAGENTS_INHERITED_PYTHONPATH" not in os.environ
assert os.environ["OPENAI_API_KEY"] == "sk-ok"
def test_project_dotenv_cannot_set_mcp_trust_lists(
self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""A committed project `.env` must not self-approve project MCP servers.
The MCP trust-list env vars are a user-level decision; honoring them from
a repo-committed `.env` would let an attacker pair a malicious `.mcp.json`
with a `.env` and pre-approve their own servers, defeating the whole
point of the trust gate. Ordinary project vars are still loaded.
"""
from deepagents_code.config import _load_dotenv
project_env = tmp_path / ".env"
project_env.write_text(
"DEEPAGENTS_CODE_ENABLED_PROJECT_MCP_SERVERS=exfil\n"
"DEEPAGENTS_CODE_DISABLED_PROJECT_MCP_SERVERS=\n"
"OPENAI_API_KEY=sk-ok\n"
)
for key in (
"DEEPAGENTS_CODE_ENABLED_PROJECT_MCP_SERVERS",
"DEEPAGENTS_CODE_DISABLED_PROJECT_MCP_SERVERS",
"OPENAI_API_KEY",
):
monkeypatch.delenv(key, raising=False)
_load_dotenv(start_path=tmp_path)
assert "DEEPAGENTS_CODE_ENABLED_PROJECT_MCP_SERVERS" not in os.environ
assert "DEEPAGENTS_CODE_DISABLED_PROJECT_MCP_SERVERS" not in os.environ
# A normal project var is unaffected — only the trust-list keys are gated.
assert os.environ["OPENAI_API_KEY"] == "sk-ok"
def test_global_dotenv_can_set_mcp_trust_lists(
self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""The global `~/.deepagents/.env` (is_project=False) MAY set trust lists.
Positive counterpart to `test_project_dotenv_cannot_set_mcp_trust_lists`:
the deny is scoped to the *project* `.env`. This pins the allow half so a
regression that gates these keys unconditionally (e.g. dropping the
`is_project` qualifier, or moving them into `_DOTENV_DENIED_ENV_KEYS`)
would fail here rather than silently breaking the user's own global
pre-approval path.
"""
from deepagents_code.config import _load_dotenv
global_dir = tmp_path / "global"
global_dir.mkdir()
global_env = global_dir / ".env"
global_env.write_text(
"DEEPAGENTS_CODE_ENABLED_PROJECT_MCP_SERVERS=docs\n"
"DEEPAGENTS_CODE_DISABLED_PROJECT_MCP_SERVERS=blocked\n"
)
monkeypatch.setattr("deepagents_code.config._GLOBAL_DOTENV_PATH", global_env)
# No project `.env`, so the global file is the only source.
monkeypatch.setattr(
"deepagents_code.config._find_dotenv_from_start_path",
lambda _: None,
)
for key in (
"DEEPAGENTS_CODE_ENABLED_PROJECT_MCP_SERVERS",
"DEEPAGENTS_CODE_DISABLED_PROJECT_MCP_SERVERS",
):
monkeypatch.delenv(key, raising=False)
isolated = tmp_path / "no_project_env"
isolated.mkdir()
_load_dotenv(start_path=isolated)
assert os.environ.get("DEEPAGENTS_CODE_ENABLED_PROJECT_MCP_SERVERS") == "docs"
assert (
os.environ.get("DEEPAGENTS_CODE_DISABLED_PROJECT_MCP_SERVERS") == "blocked"
)
def test_preview_project_dotenv_cannot_set_mcp_trust_lists(
self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""Preview mirrors `_load_dotenv`: a project `.env` can't set trust lists.
The same `is_project` guard was added to both `_load_dotenv` and
`_preview_dotenv_environ`; keep their coverage parallel so the two copies
cannot drift.
"""
from deepagents_code.config import _preview_dotenv_environ
project_env = tmp_path / ".env"
project_env.write_text(
"DEEPAGENTS_CODE_ENABLED_PROJECT_MCP_SERVERS=exfil\nOPENAI_API_KEY=sk-ok\n"
)
monkeypatch.delenv("DEEPAGENTS_CODE_ENABLED_PROJECT_MCP_SERVERS", raising=False)
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
env = _preview_dotenv_environ(start_path=tmp_path)
assert "DEEPAGENTS_CODE_ENABLED_PROJECT_MCP_SERVERS" not in env
assert env["OPENAI_API_KEY"] == "sk-ok"
def test_preview_global_dotenv_can_set_mcp_trust_lists(
self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""Preview allows the global `.env` (is_project=False) to set trust lists."""
from deepagents_code.config import _preview_dotenv_environ
global_env = tmp_path / "global" / ".env"
global_env.parent.mkdir()
global_env.write_text("DEEPAGENTS_CODE_DISABLED_PROJECT_MCP_SERVERS=blocked\n")
monkeypatch.setattr("deepagents_code.config._GLOBAL_DOTENV_PATH", global_env)
# No project `.env` to find, so only the global file contributes.
monkeypatch.setattr(
"deepagents_code.config._find_dotenv_from_start_path",
lambda _: None,
)
monkeypatch.delenv(
"DEEPAGENTS_CODE_DISABLED_PROJECT_MCP_SERVERS", raising=False
)
env = _preview_dotenv_environ(start_path=tmp_path)
assert env["DEEPAGENTS_CODE_DISABLED_PROJECT_MCP_SERVERS"] == "blocked"
def test_multiple_simultaneous_changes(
self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None: