From 7e95bef2e73162306fd5eeb54ab928364497ecd8 Mon Sep 17 00:00:00 2001 From: ajayr Date: Sun, 23 Aug 2026 08:16:04 +0100 Subject: [PATCH] feat(mcp): per-user credential injection for shared MCP servers (#4868) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(mcp): per-user credential injection for shared MCP servers A single HTTP/SSE MCP server entry can now serve several users, each authenticated to the remote service with their own credential. A server opts in with a user_auth block mapping user ids to credential header values ($ENV_VAR references supported): "user_auth": { "header": "Authorization", "users": { "": "$SERVICE_TOKEN_ALICE" } } The built-in user-scoped auth interceptor resolves the authenticated runtime user on every tool call (request runtime -> ambient LangGraph runtime -> auth config -> request-scoped user ContextVar) and rewrites the configured header via request.override(), the same per-call mechanism as the OAuth interceptor. It registers after OAuth in the shared assembly so its per-user value wins the header when a server declares both. The entry's static headers are used only for startup tool discovery. Fail-closed by default: an unmapped user - including the anonymous default-user fallback - or a credential whose env reference resolved empty gets an actionable ToolException instead of another user's credential; on_missing: "passthrough" opts out per server. Combined with the existing per-(user, thread) MCP session scoping this gives credential isolation on shared servers. Gateway API: user_auth.users values are masked in GET responses, and PUT round-trips preserve stored credentials for masked values (same contract as env/headers/oauth secrets); a masked value for a user id not already stored is rejected. Co-Authored-By: Claude Fable 5 * fix(mcp): address review — preserve stored user_auth sub-fields on partial PUT, warn-and-skip stdio, allow extras Review findings from #4868: 1. A partial user_auth payload (e.g. {"enabled": false}) merged to users={} and default on_missing, irreversibly wiping stored credentials on PUT. The merge is now sub-field-aware via model_fields_set — omitted sub-fields carry the stored values, an explicitly sent users map still replaces (so full-round-trip removal works), masked values still swap back for stored credentials. 2. user_auth on a stdio server was a silent no-op: rewritten headers go to call meta, never a transport header, while deny errors still fired. The interceptor builder now warns and skips non-sse/http servers, matching the tool_call_timeout transport-mismatch convention. 3. McpUserScopedAuthConfigResponse now allows extra keys like the harness-side model, and extras survive masking and merge, matching the server-level model_extra handling. Adds four regression tests (partial-PUT preservation, explicit-map replacement, extras round-trip, stdio warn-and-skip). Co-Authored-By: Claude Fable 5 * fix(mcp): reject blank user_auth.header at the gateway A blank header passed the gateway response model, was persisted, then failed the harness-side ExtensionsConfig validator on reload — the PUT returned 500 after the write and every later config load/startup failed until the file was hand-edited. Mirror the harness non-blank validator on McpUserScopedAuthConfigResponse so the PUT fails with 422 before anything is written. Co-Authored-By: Claude Fable 5 * style: ruff format extensions_config.py Co-Authored-By: Claude Fable 5 * fix(mcp): harden the trust chain and masking around user-scoped credential selection Address review round 4: - Scrub client-supplied user_id from run context/configurable for external callers in inject_authenticated_user_context, before every early return, and restamp only from request.state.user. Now that user_id selects which user's credential user-scoped MCP auth injects, a forged value must not survive any future path that skips the restamp. Internal callers (IM channels, scheduler) keep supplying end-user identity as before (PR #3294 contract). Regression tests pin both the scrub and that a forged body.context.user_id can never resolve as another user through merge + inject ordering. - Include the caller's own resolved user id in the fail-closed deny message so operators can copy the exact users key (it differs by deployment path), and document the key formats in the mcp.mdx doc. - Mask sensitive extra keys inside user_auth on GET like server-level extras, and swap masked sentinels back for stored values on PUT via _merge_extra_value_preserving_masked. - Extract the interceptor wrap loop into compose_tool_interceptors and pin the security property functionally: an OAuth interceptor that actually sets Authorization loses the final header to the per-user credential through the same composition the session-pool path uses. Co-Authored-By: Claude Fable 5 --------- Co-authored-by: Claude Fable 5 --- backend/app/gateway/routers/mcp.py | 86 +++- backend/app/gateway/services.py | 12 + .../deerflow/config/extensions_config.py | 38 ++ .../packages/harness/deerflow/mcp/AGENTS.md | 1 + .../harness/deerflow/mcp/interceptors.py | 31 +- .../packages/harness/deerflow/mcp/tools.py | 11 +- .../harness/deerflow/mcp/user_scoped_auth.py | 117 ++++++ backend/tests/test_gateway_services.py | 37 ++ backend/tests/test_mcp_user_scoped_auth.py | 397 ++++++++++++++++++ frontend/src/content/en/harness/mcp.mdx | 49 +++ 10 files changed, 768 insertions(+), 11 deletions(-) create mode 100644 backend/packages/harness/deerflow/mcp/user_scoped_auth.py create mode 100644 backend/tests/test_mcp_user_scoped_auth.py diff --git a/backend/app/gateway/routers/mcp.py b/backend/app/gateway/routers/mcp.py index 5999741b0..3308c980a 100644 --- a/backend/app/gateway/routers/mcp.py +++ b/backend/app/gateway/routers/mcp.py @@ -7,7 +7,7 @@ from pathlib import Path from typing import Any, Literal, NamedTuple from fastapi import APIRouter, HTTPException, Request, status -from pydantic import BaseModel, ConfigDict, Field, model_validator +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator from app.gateway.deps import require_admin_user from deerflow.config.extensions_config import ( @@ -347,6 +347,29 @@ _CODE_INJECTING_ENV_VARS = frozenset( ) +class McpUserScopedAuthConfigResponse(BaseModel): + """Per-user credential injection configuration for an MCP server.""" + + enabled: bool = Field(default=True, description="Whether user-scoped credential injection is enabled") + header: str = Field(default="Authorization", description="HTTP header to set with the resolved user credential") + users: dict[str, str] = Field(default_factory=dict, description="Map of DeerFlow user id to credential header value") + on_missing: Literal["deny", "passthrough"] = Field(default="deny", description="Behavior when the calling user has no mapped credential") + # Mirror the harness-side McpUserScopedAuthConfig (extra="allow"): without + # this, an operator's unknown key inside user_auth would be silently + # stripped by the next admin PUT, while server-level extras are preserved. + model_config = ConfigDict(extra="allow") + + # Mirror the harness-side non-blank check: a blank header accepted here + # would be persisted, then fail ExtensionsConfig validation on reload — + # wedging every subsequent config load until the file is hand-edited. + @field_validator("header") + @classmethod + def _validate_header_not_blank(cls, value: str) -> str: + if not value.strip(): + raise ValueError("user_auth.header must not be empty") + return value + + class McpOAuthConfigResponse(BaseModel): """OAuth configuration for an MCP server.""" @@ -377,6 +400,7 @@ class McpServerConfigResponse(BaseModel): url: str | None = Field(default=None, description="URL of the MCP server (for sse or http type)") headers: dict[str, str] = Field(default_factory=dict, description="HTTP headers to send (for sse or http type)") oauth: McpOAuthConfigResponse | None = Field(default=None, description="OAuth configuration for MCP HTTP/SSE servers") + user_auth: McpUserScopedAuthConfigResponse | None = Field(default=None, description="Per-user credential injection for MCP HTTP/SSE servers") description: str = Field(default="", description="Human-readable description of what this MCP server provides") routing: McpRoutingConfig = Field(default_factory=McpRoutingConfig, description="Soft routing hints for tools from this MCP server") tools: dict[str, McpToolOverride] = Field(default_factory=dict, description="Per-original-tool MCP configuration overrides") @@ -653,12 +677,21 @@ def _mask_server_config(server: McpServerConfigResponse) -> McpServerConfigRespo "refresh_token": None, } ) + masked_user_auth = None + if server.user_auth is not None: + # Extras inside user_auth get the same sensitive-key masking as + # server-level extras: they round-trip through PUT (extra="allow"), so + # an operator-stored secret-bearing key must not come back in + # cleartext from GET while the identical key at server level is masked. + masked_ua_extra = {key: _MASKED_VALUE if _is_sensitive_extra_key(key) else _mask_sensitive_extra_value(value) for key, value in (server.user_auth.model_extra or {}).items()} + masked_user_auth = server.user_auth.model_copy(update={"users": {k: _MASKED_VALUE for k in server.user_auth.users}, **masked_ua_extra}) masked_extra = {key: _MASKED_VALUE if _is_sensitive_extra_key(key) else _mask_sensitive_extra_value(value) for key, value in (server.model_extra or {}).items()} return server.model_copy( update={ "env": masked_env, "headers": masked_headers, "oauth": masked_oauth, + "user_auth": masked_user_auth, **masked_extra, } ) @@ -720,11 +753,62 @@ def _merge_preserving_secrets( "refresh_token": merged_refresh_token, } ) + merged_user_auth = incoming.user_auth + if incoming.user_auth is not None: + # Sub-field-aware merge: a partial user_auth payload (e.g. just + # {"enabled": false}) must not wipe the stored credential map or reset + # other stored sub-fields. Only sub-fields the request explicitly set + # replace stored values; the rest carry over — the same contract the + # block-level `model_fields_set` check below applies one level up. + incoming_ua = incoming.user_auth + base = existing.user_auth + set_fields = incoming_ua.model_fields_set + effective: dict[str, Any] = {} + if base is not None: + effective.update({name: getattr(base, name) for name in ("enabled", "header", "users", "on_missing")}) + effective.update(base.model_extra or {}) + for name in ("enabled", "header", "on_missing"): + if name in set_fields: + effective[name] = getattr(incoming_ua, name) + # Extras are masked by GET (see _mask_server_config), so a round-trip + # PUT must swap masked sentinel values back for the stored ones — + # the same contract server-level extras get below. + base_extra = (base.model_extra or {}) if base is not None else {} + for key, value in (incoming_ua.model_extra or {}).items(): + effective[key] = _merge_extra_value_preserving_masked( + key, + value, + base_extra.get(key), + existing_present=key in base_extra, + ) + if "users" in set_fields: + # An explicitly sent map replaces the stored one (so a full + # round-trip can remove a user), with masked values swapped back + # for the stored credentials. + existing_users = base.users if base is not None else {} + merged_users = {} + for k, v in incoming_ua.users.items(): + if v == _MASKED_VALUE: + if k in existing_users: + merged_users[k] = existing_users[k] + else: + raise HTTPException( + status_code=400, + detail=f"Cannot set user_auth credential for '{k}' to masked value '***'; provide a real value.", + ) + else: + merged_users[k] = v + effective["users"] = merged_users + merged_user_auth = McpUserScopedAuthConfigResponse(**effective) + update = { "env": merged_env, "headers": merged_headers, "oauth": merged_oauth, + "user_auth": merged_user_auth, } + if "user_auth" not in incoming.model_fields_set: + update["user_auth"] = existing.user_auth if "routing" not in incoming.model_fields_set: update["routing"] = existing.routing if "tools" not in incoming.model_fields_set: diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py index f7335f2b4..2671152ae 100644 --- a/backend/app/gateway/services.py +++ b/backend/app/gateway/services.py @@ -499,6 +499,18 @@ def inject_authenticated_user_context( for key in _SERVER_OWNED_AUTHZ_CONTEXT_KEYS: configurable.pop(key, None) auth_source = getattr(getattr(request, "state", None), "auth_source", None) + # ``user_id`` is server-owned for EXTERNAL callers: it now selects which + # user's credential user-scoped MCP auth injects, so a client-forged value + # must never survive any early return below — scrub it here and restamp it + # only from ``request.state.user``. Internal callers (IM channels, the + # scheduler) are the deliberate exception: they authenticate their own end + # users and supply that identity in run context (PR #3294), which the + # internal-role branch below preserves. + user = getattr(getattr(request, "state", None), "user", None) + if auth_source != AUTH_SOURCE_INTERNAL and getattr(user, "system_role", None) != INTERNAL_SYSTEM_ROLE: + runtime_context.pop("user_id", None) + if isinstance(configurable, dict): + configurable.pop("user_id", None) runtime_context["is_internal"] = auth_source == AUTH_SOURCE_INTERNAL if auth_source == AUTH_SOURCE_INTERNAL and request_context is not None: channel_user_id = request_context.get("channel_user_id") diff --git a/backend/packages/harness/deerflow/config/extensions_config.py b/backend/packages/harness/deerflow/config/extensions_config.py index 4c8b248c0..ae1987ed1 100644 --- a/backend/packages/harness/deerflow/config/extensions_config.py +++ b/backend/packages/harness/deerflow/config/extensions_config.py @@ -98,6 +98,40 @@ class McpTaskToolsetConfig(BaseModel): return value +class McpUserScopedAuthConfig(BaseModel): + """Per-user credential injection for a shared MCP server (HTTP/SSE transports). + + Maps DeerFlow user ids to credential header values so that one configured + MCP server can serve several users, each authenticated to the remote + service with their own credential. The credential for the authenticated + user is injected into every tool call by the built-in user-scoped auth + interceptor; the server entry's static ``headers`` are only used for + startup tool discovery. + + Values support the same ``$ENV_VAR`` resolution as the rest of this file, + so raw secrets can stay in the process environment. + """ + + enabled: bool = Field(default=True, description="Whether user-scoped credential injection is enabled") + header: str = Field(default="Authorization", description="HTTP header to set with the resolved user credential") + users: dict[str, str] = Field( + default_factory=dict, + description="Map of DeerFlow user id to full credential header value (e.g. 'Bearer '); values support $ENV_VAR references", + ) + on_missing: Literal["deny", "passthrough"] = Field( + default="deny", + description=("Behavior when the calling user has no mapped credential (or the mapped value resolved empty): 'deny' fails the tool call with an actionable error; 'passthrough' forwards the request with the server's static headers"), + ) + model_config = ConfigDict(extra="allow") + + @field_validator("header") + @classmethod + def _validate_header_not_blank(cls, value: str) -> str: + if not value.strip(): + raise ValueError("user_auth.header must not be empty") + return value + + class McpOAuthConfig(BaseModel): """OAuth configuration for an MCP server (HTTP/SSE transports).""" @@ -132,6 +166,10 @@ class McpServerConfig(BaseModel): url: str | None = Field(default=None, description="URL of the MCP server (for sse or http type)") headers: dict[str, str] = Field(default_factory=dict, description="HTTP headers to send (for sse or http type)") oauth: McpOAuthConfig | None = Field(default=None, description="OAuth configuration (for sse or http type)") + user_auth: McpUserScopedAuthConfig | None = Field( + default=None, + description="Per-user credential injection (for sse or http type): map DeerFlow user ids to per-user credential header values", + ) description: str = Field(default="", description="Human-readable description of what this MCP server provides") routing: McpRoutingConfig = Field(default_factory=McpRoutingConfig, description="Soft routing hints for tools from this MCP server") tools: dict[str, McpToolOverride] = Field(default_factory=dict, description="Per-original-tool MCP configuration overrides") diff --git a/backend/packages/harness/deerflow/mcp/AGENTS.md b/backend/packages/harness/deerflow/mcp/AGENTS.md index 00f60a72f..36b665646 100644 --- a/backend/packages/harness/deerflow/mcp/AGENTS.md +++ b/backend/packages/harness/deerflow/mcp/AGENTS.md @@ -10,6 +10,7 @@ - **Transports**: stdio (command-based), SSE, HTTP - **Per-server tool-name prefixing**: `mcpServers..tool_name_prefix` defaults to `true`, preserving the collision-safe `_` prefix. Servers whose tools already carry a stable namespace may set it to `false`; discovery then calls `langchain_mcp_adapters.tools.load_mcp_tools` with that server's flag. Source routing and stdio session-pool wrapping are based on the producing server and transport, never on whether the visible tool name starts with the server prefix. - **OAuth (HTTP/SSE)**: Supports token endpoint flows (`client_credentials`, `refresh_token`) with automatic token refresh + Authorization header injection +- **Per-user credentials (HTTP/SSE)**: `mcpServers..user_auth` maps DeerFlow user ids to credential header values (`$ENV_VAR` references supported). `mcp/user_scoped_auth.py::build_user_scoped_auth_interceptor` rewrites the configured header on every tool call from the authenticated runtime user (registered after OAuth in `mcp/interceptors.py`, so its per-user value wins the header for servers declaring both). Fail-closed: an unmapped user or an empty resolved credential raises a `ToolException` unless `on_missing: "passthrough"` is set. The server's static `headers` serve startup tool discovery only. Gateway GET masks `user_auth.users` values; PUT round-trips masked values by preserving stored credentials. - **Routing hints**: `extensions_config.json -> mcpServers..routing` and `tools..routing` are soft preference metadata. The effective routing is resolved while `mcp/tools.py::get_mcp_tools()` still has both diff --git a/backend/packages/harness/deerflow/mcp/interceptors.py b/backend/packages/harness/deerflow/mcp/interceptors.py index e283db6ae..65578ed75 100644 --- a/backend/packages/harness/deerflow/mcp/interceptors.py +++ b/backend/packages/harness/deerflow/mcp/interceptors.py @@ -7,6 +7,7 @@ from typing import Any from deerflow.config.extensions_config import ExtensionsConfig from deerflow.mcp.oauth import build_oauth_tool_interceptor +from deerflow.mcp.user_scoped_auth import build_user_scoped_auth_interceptor from deerflow.reflection import resolve_variable logger = logging.getLogger(__name__) @@ -16,15 +17,23 @@ def build_mcp_tool_interceptors( extensions_config: ExtensionsConfig, *, oauth_builder: Any = build_oauth_tool_interceptor, + user_auth_builder: Any = build_user_scoped_auth_interceptor, resolver: Any = resolve_variable, target_logger: logging.Logger = logger, ) -> list[Any]: - """Build OAuth followed by configured custom MCP interceptors.""" + """Build OAuth, user-scoped auth, then configured custom MCP interceptors.""" interceptors: list[Any] = [] oauth_interceptor = oauth_builder(extensions_config) if oauth_interceptor is not None: interceptors.append(oauth_interceptor) + # After OAuth so a server declaring both gets the per-user credential: + # interceptors wrap outermost-first, so the later-registered user-scoped + # override runs closer to the transport and wins the final header value. + user_auth_interceptor = user_auth_builder(extensions_config) + if user_auth_interceptor is not None: + interceptors.append(user_auth_interceptor) + raw_paths = (extensions_config.model_extra or {}).get("mcpInterceptors") if isinstance(raw_paths, str): raw_paths = [raw_paths] @@ -55,3 +64,23 @@ def build_mcp_tool_interceptors( exc_info=True, ) return interceptors + + +def compose_tool_interceptors(interceptors: list[Any], base_handler: Any) -> Any: + """Compose interceptors onion-style around ``base_handler``: first = outermost. + + The later-registered interceptor runs closer to the transport, so its + header writes win over earlier ones — the property user-scoped auth relies + on to override an OAuth-injected credential. This is the single wrap + convention; the session-pool tool path composes through here so tests that + pin the override property exercise the production composition. + """ + handler = base_handler + for interceptor in reversed(interceptors): + outer = handler + + async def wrapped(req: Any, _i: Any = interceptor, _h: Any = outer) -> Any: + return await _i(req, _h) + + handler = wrapped + return handler diff --git a/backend/packages/harness/deerflow/mcp/tools.py b/backend/packages/harness/deerflow/mcp/tools.py index 0a2af312b..73de63161 100644 --- a/backend/packages/harness/deerflow/mcp/tools.py +++ b/backend/packages/harness/deerflow/mcp/tools.py @@ -18,7 +18,7 @@ from deerflow.config.extensions_config import ExtensionsConfig, McpServerConfig, from deerflow.config.paths import VIRTUAL_PATH_PREFIX, Paths, get_paths from deerflow.constants import DEFAULT_MCP_SESSION_INIT_TIMEOUT, MCP_TMP_SUBDIR from deerflow.mcp.client import build_servers_config -from deerflow.mcp.interceptors import build_mcp_tool_interceptors +from deerflow.mcp.interceptors import build_mcp_tool_interceptors, compose_tool_interceptors from deerflow.mcp.oauth import build_oauth_tool_interceptor, get_initial_oauth_headers from deerflow.mcp.session_pool import get_session_pool from deerflow.mcp.tasks import ORDINARY_MCP_TASK_DRIVER, TaskSubmitRequest @@ -562,14 +562,7 @@ def _make_session_pool_tool( **kwargs, ) - handler = base_handler - for interceptor in reversed(tool_interceptors): - outer = handler - - async def wrapped(req: Any, _i: Any = interceptor, _h: Any = outer) -> Any: - return await _i(req, _h) - - handler = wrapped + handler = compose_tool_interceptors(tool_interceptors, base_handler) request = MCPToolCallRequest( name=original_name, diff --git a/backend/packages/harness/deerflow/mcp/user_scoped_auth.py b/backend/packages/harness/deerflow/mcp/user_scoped_auth.py new file mode 100644 index 000000000..5c2ba4f62 --- /dev/null +++ b/backend/packages/harness/deerflow/mcp/user_scoped_auth.py @@ -0,0 +1,117 @@ +"""Per-user credential injection for shared MCP servers. + +One configured HTTP/SSE MCP server can serve several DeerFlow users, each +authenticated to the remote service with their own credential. A server opts in +by declaring a ``user_auth`` block (:class:`McpUserScopedAuthConfig`) mapping +DeerFlow user ids to credential header values. On every tool call the +interceptor resolves the authenticated user and rewrites the configured header +via ``request.override(headers=...)`` — the same per-call mechanism the OAuth +interceptor uses. + +The server entry's static ``headers`` are used only for startup tool discovery +(``tools/list``); they never authenticate a user's tool call when ``user_auth`` +is enabled for that server, except under an explicit ``on_missing: +"passthrough"`` opt-out. + +Fail-closed by default: an unmapped user (including the anonymous +``DEFAULT_USER_ID`` fallback), or a mapped credential whose ``$ENV_VAR`` +reference resolved to an empty string, gets an actionable ``ToolException`` +instead of another user's credential or the discovery credential. +""" + +from __future__ import annotations + +import logging +from typing import Any + +from langchain_core.tools import ToolException + +from deerflow.config.extensions_config import ExtensionsConfig, McpUserScopedAuthConfig +from deerflow.runtime.user_context import resolve_runtime_user_id + +logger = logging.getLogger(__name__) + + +def _current_runtime() -> Any | None: + """Best-effort access to the LangGraph runtime for the current tool call. + + ``get_runtime()`` raises outside a runtime context (embedded clients, unit + tests, discovery paths); ``resolve_runtime_user_id`` accepts ``None`` and + falls back to LangGraph auth config and the request-scoped user + ContextVar, so failures here reduce accuracy but never crash the call. + """ + try: + from langgraph.runtime import get_runtime + + return get_runtime() + except Exception: + return None + + +def build_user_scoped_auth_interceptor(extensions_config: ExtensionsConfig) -> Any | None: + """Build a tool interceptor injecting per-user credentials, or ``None``. + + Returns ``None`` when no enabled server declares an enabled ``user_auth`` + block, so callers can skip registration entirely (mirrors + ``build_oauth_tool_interceptor``). + """ + user_auth_by_server: dict[str, McpUserScopedAuthConfig] = {} + for server_name, server_config in extensions_config.get_enabled_mcp_servers().items(): + if server_config.user_auth is None or not server_config.user_auth.enabled: + continue + if server_config.type not in ("sse", "http"): + # A stdio server has no HTTP headers: the pooled stdio path forwards + # rewritten headers as call meta, never a transport header, so the + # credential would go nowhere while deny errors still fired for + # unmapped users. Warn-and-skip matches the existing convention for + # transport/config mismatches (e.g. tool_call_timeout on non-stdio). + logger.warning( + "MCP server '%s' declares user_auth but uses the '%s' transport; user-scoped credentials only apply to 'sse'/'http' servers — ignoring user_auth for this server", + server_name, + server_config.type, + ) + continue + user_auth_by_server[server_name] = server_config.user_auth + + if not user_auth_by_server: + return None + + async def user_scoped_auth_interceptor(request: Any, handler: Any) -> Any: + user_auth = user_auth_by_server.get(request.server_name) + if user_auth is None: + return await handler(request) + + # Prefer the runtime attached to the request (set by the adapter when + # the call originates inside a graph); fall back to the ambient + # LangGraph runtime, then to resolve_runtime_user_id's own chain + # (LangGraph auth config → request-scoped user ContextVar → default). + runtime = getattr(request, "runtime", None) + if runtime is None: + runtime = _current_runtime() + user_id = resolve_runtime_user_id(runtime) + # Empty string covers a `$ENV_VAR` reference whose variable was unset: + # ExtensionsConfig.resolve_env_variables stores "" for those, and an + # empty credential must fail closed rather than send an empty header. + credential = user_auth.users.get(user_id, "") + if not credential: + if user_auth.on_missing == "passthrough": + return await handler(request) + logger.warning( + "Denied MCP tool call to server '%s': no user-scoped credential for user '%s'", + request.server_name, + user_id, + ) + # The resolved id is included so the operator can copy the exact + # ``users`` key: it differs by deployment path (a safe-slug like + # ``alice-example-com-ab12cd34`` via LangGraph auth, a raw user + # UUID via the embedded Gateway). It is the caller's own id, so + # surfacing it leaks nothing across users. + raise ToolException( + f"No credential is configured for your account (user id '{user_id}') on MCP server '{request.server_name}'. Ask the operator to add this exact id to that server's user_auth.users map (or set its environment variable)." + ) + + updated_headers = dict(request.headers or {}) + updated_headers[user_auth.header] = credential + return await handler(request.override(headers=updated_headers)) + + return user_scoped_auth_interceptor diff --git a/backend/tests/test_gateway_services.py b/backend/tests/test_gateway_services.py index 062b98e86..ad73333a8 100644 --- a/backend/tests/test_gateway_services.py +++ b/backend/tests/test_gateway_services.py @@ -3015,3 +3015,40 @@ async def test_start_run_rejects_invalid_thread_id_before_resolving_dependencies assert exc_info.value.status_code == 422 assert "Invalid thread_id" in exc_info.value.detail + + +def test_client_forged_user_id_is_scrubbed_for_external_callers(): + """user_id now selects which credential user-scoped MCP auth injects, so a + client-forged value must never survive merge + inject on any external path + — including ones that end in an early return (no authenticated user).""" + from types import SimpleNamespace + + from app.gateway.services import build_run_config, inject_authenticated_user_context, merge_run_context_overrides + + # Forged via body.config (copied verbatim) AND body.context (merged). + config = build_run_config("thread-1", {"context": {"user_id": "victim"}, "configurable": {"user_id": "victim"}}, None) + merge_run_context_overrides(config, {"user_id": "victim"}) + + # External caller with no authenticated user: scrub, never restamp. + request = SimpleNamespace(state=SimpleNamespace(user=None, auth_source=None)) + inject_authenticated_user_context(config, request) + assert "user_id" not in config["context"] + assert "user_id" not in config["configurable"] + + +def test_client_forged_user_id_never_selects_another_users_credential(): + """End-to-end pin through merge + inject ordering: the id user-scoped MCP + auth resolves from runtime context is the authenticated user, regardless of + what the client put in body.context/config.""" + from types import SimpleNamespace + + from app.gateway.services import build_run_config, inject_authenticated_user_context, merge_run_context_overrides + from deerflow.runtime.user_context import resolve_runtime_user_id + + config = build_run_config("thread-1", {"context": {"user_id": "victim"}}, None) + merge_run_context_overrides(config, {"user_id": "victim"}) + request = SimpleNamespace(state=SimpleNamespace(user=SimpleNamespace(id="attacker-own-id", system_role=None, oauth_provider=None, oauth_id=None), auth_source=None)) + inject_authenticated_user_context(config, request) + + runtime = SimpleNamespace(server_info=None, context=config["context"]) + assert resolve_runtime_user_id(runtime) == "attacker-own-id" diff --git a/backend/tests/test_mcp_user_scoped_auth.py b/backend/tests/test_mcp_user_scoped_auth.py new file mode 100644 index 000000000..e42477f73 --- /dev/null +++ b/backend/tests/test_mcp_user_scoped_auth.py @@ -0,0 +1,397 @@ +"""Tests for per-user credential injection on shared MCP servers.""" + +import asyncio +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +import pytest +from langchain_core.tools import ToolException +from langchain_mcp_adapters.interceptors import MCPToolCallRequest + +from deerflow.config.extensions_config import ( + ExtensionsConfig, + McpServerConfig, + McpUserScopedAuthConfig, +) +from deerflow.mcp.interceptors import build_mcp_tool_interceptors +from deerflow.mcp.user_scoped_auth import build_user_scoped_auth_interceptor + + +def _config(**user_auth_kwargs) -> ExtensionsConfig: + return ExtensionsConfig( + mcp_servers={ + "shared-http": McpServerConfig( + enabled=True, + type="http", + url="https://mcp.example.com/mcp", + headers={"Authorization": "Bearer discovery-token"}, + user_auth=McpUserScopedAuthConfig(**user_auth_kwargs), + ), + "other": McpServerConfig(enabled=True, type="http", url="https://other.example.com/mcp"), + }, + skills={}, + ) + + +def _request(server_name: str = "shared-http", headers: dict | None = None, runtime: object | None = None) -> MCPToolCallRequest: + return MCPToolCallRequest( + name="act", + args={}, + server_name=server_name, + headers=headers, + runtime=runtime, + ) + + +def _runtime_for_user(user_id: str) -> object: + return SimpleNamespace(server_info=None, context={"user_id": user_id}) + + +async def _echo_handler(request: MCPToolCallRequest) -> MCPToolCallRequest: + return request + + +def test_no_user_auth_servers_returns_none(): + config = ExtensionsConfig( + mcp_servers={"plain": McpServerConfig(enabled=True, type="http", url="https://x.example.com")}, + skills={}, + ) + assert build_user_scoped_auth_interceptor(config) is None + + +def test_disabled_user_auth_returns_none(): + config = _config(users={"u1": "Bearer t1"}, enabled=False) + assert build_user_scoped_auth_interceptor(config) is None + + +def test_disabled_server_is_ignored(): + config = _config(users={"u1": "Bearer t1"}) + config.mcp_servers["shared-http"].enabled = False + assert build_user_scoped_auth_interceptor(config) is None + + +def test_mapped_user_gets_own_credential(): + interceptor = build_user_scoped_auth_interceptor(_config(users={"u1": "Bearer t1", "u2": "Bearer t2"})) + result = asyncio.run(interceptor(_request(headers={"Authorization": "Bearer discovery-token"}, runtime=_runtime_for_user("u2")), _echo_handler)) + assert result.headers["Authorization"] == "Bearer t2" + + +def test_custom_header_and_other_headers_preserved(): + interceptor = build_user_scoped_auth_interceptor(_config(header="X-Api-Key", users={"u1": "k1"})) + result = asyncio.run(interceptor(_request(headers={"Accept": "application/json"}, runtime=_runtime_for_user("u1")), _echo_handler)) + assert result.headers == {"Accept": "application/json", "X-Api-Key": "k1"} + + +def test_other_server_passes_through_untouched(): + interceptor = build_user_scoped_auth_interceptor(_config(users={"u1": "Bearer t1"})) + request = _request(server_name="other", headers={"Authorization": "Bearer static"}, runtime=_runtime_for_user("u1")) + result = asyncio.run(interceptor(request, _echo_handler)) + assert result is request + + +def test_unmapped_user_denied_without_calling_handler(): + interceptor = build_user_scoped_auth_interceptor(_config(users={"u1": "Bearer t1"})) + handler = AsyncMock() + with pytest.raises(ToolException, match="No credential is configured"): + asyncio.run(interceptor(_request(runtime=_runtime_for_user("stranger")), handler)) + handler.assert_not_awaited() + + +def test_empty_resolved_credential_is_denied(): + """An unset $ENV_VAR reference resolves to "" and must fail closed.""" + interceptor = build_user_scoped_auth_interceptor(_config(users={"u1": ""})) + with pytest.raises(ToolException, match="No credential is configured"): + asyncio.run(interceptor(_request(runtime=_runtime_for_user("u1")), AsyncMock())) + + +def test_on_missing_passthrough_keeps_static_headers(): + interceptor = build_user_scoped_auth_interceptor(_config(users={"u1": "Bearer t1"}, on_missing="passthrough")) + request = _request(headers={"Authorization": "Bearer discovery-token"}, runtime=_runtime_for_user("stranger")) + result = asyncio.run(interceptor(request, _echo_handler)) + assert result.headers["Authorization"] == "Bearer discovery-token" + + +def test_default_user_fallback_is_denied_when_unmapped(): + """Without any resolvable identity the DEFAULT_USER_ID fallback must not inherit a credential.""" + interceptor = build_user_scoped_auth_interceptor(_config(users={"u1": "Bearer t1"})) + with patch("deerflow.mcp.user_scoped_auth._current_runtime", return_value=None), pytest.raises(ToolException): + asyncio.run(interceptor(_request(runtime=None), AsyncMock())) + + +def test_env_var_reference_resolution(tmp_path, monkeypatch): + monkeypatch.setenv("TEST_USER_CRED", "Bearer from-env") + config_file = tmp_path / "extensions_config.json" + config_file.write_text( + """ + { + "mcpServers": { + "shared-http": { + "enabled": true, + "type": "http", + "url": "https://mcp.example.com/mcp", + "user_auth": {"users": {"u1": "$TEST_USER_CRED", "u2": "$TEST_USER_CRED_UNSET"}} + } + } + } + """ + ) + config = ExtensionsConfig.from_file(str(config_file)) + user_auth = config.mcp_servers["shared-http"].user_auth + assert user_auth.users["u1"] == "Bearer from-env" + assert user_auth.users["u2"] == "" + + +def test_registered_after_oauth_in_shared_assembly(): + config = _config(users={"u1": "Bearer t1"}) + + async def oauth(request, handler): # pragma: no cover - identity only + return await handler(request) + + interceptors = build_mcp_tool_interceptors(config, oauth_builder=lambda _cfg: oauth) + assert len(interceptors) == 2 + assert interceptors[0] is oauth + assert interceptors[1].__name__ == "user_scoped_auth_interceptor" + + +def test_shared_assembly_skips_when_no_user_auth(): + config = ExtensionsConfig( + mcp_servers={"plain": McpServerConfig(enabled=True, type="http", url="https://x.example.com")}, + skills={}, + ) + interceptors = build_mcp_tool_interceptors(config, oauth_builder=lambda _cfg: None) + assert interceptors == [] + + +def test_gateway_masks_user_auth_credentials(): + from app.gateway.routers.mcp import ( + McpServerConfigResponse, + McpUserScopedAuthConfigResponse, + _mask_server_config, + ) + + server = McpServerConfigResponse( + type="http", + url="https://mcp.example.com/mcp", + user_auth=McpUserScopedAuthConfigResponse(users={"u1": "Bearer real-secret"}), + ) + masked = _mask_server_config(server) + assert masked.user_auth.users == {"u1": "***"} + assert masked.user_auth.header == "Authorization" + + +def test_gateway_merge_preserves_masked_user_auth_values(): + from app.gateway.routers.mcp import ( + McpServerConfigResponse, + McpUserScopedAuthConfigResponse, + _merge_preserving_secrets, + ) + + existing = McpServerConfigResponse( + type="http", + url="https://mcp.example.com/mcp", + user_auth=McpUserScopedAuthConfigResponse(users={"u1": "Bearer real-secret"}), + ) + incoming = McpServerConfigResponse( + type="http", + url="https://mcp.example.com/mcp", + user_auth=McpUserScopedAuthConfigResponse(users={"u1": "***", "u2": "Bearer new-secret"}), + ) + merged = _merge_preserving_secrets(incoming, existing) + assert merged.user_auth.users == {"u1": "Bearer real-secret", "u2": "Bearer new-secret"} + + +def test_gateway_merge_rejects_masked_value_for_new_user(): + from fastapi import HTTPException + + from app.gateway.routers.mcp import ( + McpServerConfigResponse, + McpUserScopedAuthConfigResponse, + _merge_preserving_secrets, + ) + + existing = McpServerConfigResponse(type="http", url="https://mcp.example.com/mcp") + incoming = McpServerConfigResponse( + type="http", + url="https://mcp.example.com/mcp", + user_auth=McpUserScopedAuthConfigResponse(users={"new-user": "***"}), + ) + with pytest.raises(HTTPException): + _merge_preserving_secrets(incoming, existing) + + +def test_gateway_merge_preserves_user_auth_when_field_omitted(): + from app.gateway.routers.mcp import ( + McpServerConfigResponse, + McpUserScopedAuthConfigResponse, + _merge_preserving_secrets, + ) + + existing = McpServerConfigResponse( + type="http", + url="https://mcp.example.com/mcp", + user_auth=McpUserScopedAuthConfigResponse(users={"u1": "Bearer real-secret"}), + ) + incoming = McpServerConfigResponse(type="http", url="https://mcp.example.com/mcp") + merged = _merge_preserving_secrets(incoming, existing) + assert merged.user_auth is not None + assert merged.user_auth.users == {"u1": "Bearer real-secret"} + + +def test_partial_user_auth_put_preserves_stored_subfields(): + """A payload like {"enabled": false} must not wipe users or reset on_missing.""" + from app.gateway.routers.mcp import ( + McpServerConfigResponse, + McpUserScopedAuthConfigResponse, + _merge_preserving_secrets, + ) + + existing = McpServerConfigResponse( + type="http", + url="https://mcp.example.com/mcp", + user_auth=McpUserScopedAuthConfigResponse(users={"u1": "Bearer real-secret"}, on_missing="passthrough", header="X-Api-Key"), + ) + incoming = McpServerConfigResponse( + type="http", + url="https://mcp.example.com/mcp", + user_auth=McpUserScopedAuthConfigResponse(enabled=False), + ) + merged = _merge_preserving_secrets(incoming, existing) + assert merged.user_auth.enabled is False + assert merged.user_auth.users == {"u1": "Bearer real-secret"} + assert merged.user_auth.on_missing == "passthrough" + assert merged.user_auth.header == "X-Api-Key" + + +def test_explicit_users_map_still_replaces_and_can_remove(): + """An explicitly sent map replaces the stored one, so removal via full round-trip works.""" + from app.gateway.routers.mcp import ( + McpServerConfigResponse, + McpUserScopedAuthConfigResponse, + _merge_preserving_secrets, + ) + + existing = McpServerConfigResponse( + type="http", + url="https://mcp.example.com/mcp", + user_auth=McpUserScopedAuthConfigResponse(users={"u1": "Bearer s1", "u2": "Bearer s2"}), + ) + incoming = McpServerConfigResponse( + type="http", + url="https://mcp.example.com/mcp", + user_auth=McpUserScopedAuthConfigResponse(users={"u1": "***"}), + ) + merged = _merge_preserving_secrets(incoming, existing) + assert merged.user_auth.users == {"u1": "Bearer s1"} # u2 removed, u1 preserved through mask + + +def test_user_auth_extra_keys_survive_parse_mask_and_merge(): + from app.gateway.routers.mcp import ( + McpServerConfigResponse, + McpUserScopedAuthConfigResponse, + _mask_server_config, + _merge_preserving_secrets, + ) + + ua = McpUserScopedAuthConfigResponse(**{"users": {"u1": "Bearer s"}, "custom_note": "keep-me"}) + assert (ua.model_extra or {}).get("custom_note") == "keep-me" + server = McpServerConfigResponse(type="http", url="https://x", user_auth=ua) + masked = _mask_server_config(server) + assert (masked.user_auth.model_extra or {}).get("custom_note") == "keep-me" + merged = _merge_preserving_secrets( + McpServerConfigResponse(type="http", url="https://x", user_auth=McpUserScopedAuthConfigResponse(enabled=False)), + server, + ) + assert (merged.user_auth.model_extra or {}).get("custom_note") == "keep-me" + + +def test_stdio_server_user_auth_is_skipped_with_warning(caplog): + import logging + + config = ExtensionsConfig( + mcp_servers={ + "local-stdio": McpServerConfig( + enabled=True, + type="stdio", + command="npx", + args=["-y", "some-server"], + user_auth=McpUserScopedAuthConfig(users={"u1": "Bearer t1"}), + ), + }, + skills={}, + ) + with caplog.at_level(logging.WARNING, logger="deerflow.mcp.user_scoped_auth"): + interceptor = build_user_scoped_auth_interceptor(config) + assert interceptor is None # no eligible servers -> nothing registered, no deny errors + assert any("user_auth" in r.message and "stdio" in r.message for r in caplog.records) + + +def test_gateway_rejects_blank_user_auth_header(): + """A blank header must be rejected at the gateway, not persisted and left to + wedge extensions_config.json on reload (harness-side validator would raise).""" + import pydantic + import pytest + + from app.gateway.routers.mcp import McpUserScopedAuthConfigResponse + + for blank in ("", " ", "\t"): + with pytest.raises(pydantic.ValidationError, match="must not be empty"): + McpUserScopedAuthConfigResponse(header=blank) + # Non-blank still fine, and default untouched. + assert McpUserScopedAuthConfigResponse(header="X-Api-Key").header == "X-Api-Key" + assert McpUserScopedAuthConfigResponse().header == "Authorization" + + +def test_deny_error_includes_the_callers_resolved_user_id(): + """The users key format differs by deployment path; the fail-closed error + must show the caller's own resolved id so the operator can copy the exact key.""" + interceptor = build_user_scoped_auth_interceptor(_config(users={"u1": "Bearer t1"})) + with pytest.raises(ToolException, match="user id 'stranger-uuid'"): + asyncio.run(interceptor(_request(runtime=_runtime_for_user("stranger-uuid")), AsyncMock())) + + +def test_user_credential_wins_over_oauth_set_header_through_real_composition(): + """Pin the wrap-order property functionally, not just list order: an OAuth + interceptor that actually sets Authorization must lose the final header to + the per-user credential, through the same composition the session-pool + tool path uses.""" + from deerflow.mcp.interceptors import compose_tool_interceptors + + config = _config(users={"u1": "Bearer user-cred"}) + + async def oauth(request, handler): + headers = dict(request.headers or {}) + headers["Authorization"] = "Bearer oauth-token" + return await handler(request.override(headers=headers)) + + interceptors = build_mcp_tool_interceptors(config, oauth_builder=lambda _cfg: oauth) + handler = compose_tool_interceptors(interceptors, _echo_handler) + final = asyncio.run(handler(_request(runtime=_runtime_for_user("u1")))) + assert final.headers["Authorization"] == "Bearer user-cred" + # And on a server without user_auth the OAuth header must survive untouched. + final_other = asyncio.run(handler(_request(server_name="other", runtime=_runtime_for_user("u1")))) + assert final_other.headers["Authorization"] == "Bearer oauth-token" + + +def test_gateway_masks_sensitive_user_auth_extra_keys(): + """Secret-bearing extras inside user_auth must be masked by GET like the + identical keys at server level, and a masked round-trip must preserve them.""" + from app.gateway.routers.mcp import ( + McpServerConfigResponse, + McpUserScopedAuthConfigResponse, + _mask_server_config, + _merge_preserving_secrets, + ) + + server = McpServerConfigResponse( + type="http", + url="https://mcp.example.com/mcp", + user_auth=McpUserScopedAuthConfigResponse(users={"u1": "Bearer s1"}, client_secret="super-secret", custom_note="keep-me"), + ) + masked = _mask_server_config(server) + assert masked.user_auth.model_extra["client_secret"] == "***" + assert masked.user_auth.model_extra["custom_note"] == "keep-me" + + # Round-trip: PUT of the masked GET payload keeps the stored secret. + merged = _merge_preserving_secrets(masked, server) + assert merged.user_auth.model_extra["client_secret"] == "super-secret" + assert merged.user_auth.users == {"u1": "Bearer s1"} diff --git a/frontend/src/content/en/harness/mcp.mdx b/frontend/src/content/en/harness/mcp.mdx index 07e37bbbb..c51f34e41 100644 --- a/frontend/src/content/en/harness/mcp.mdx +++ b/frontend/src/content/en/harness/mcp.mdx @@ -105,6 +105,55 @@ When an OAuth-protected MCP server is connected, DeerFlow will: The OAuth flow is transparent to the Lead Agent — it simply calls the tool, and DeerFlow handles the authentication. +## Per-user credentials + +A single HTTP/SSE MCP server entry can serve several DeerFlow users, each +authenticated to the remote service with their own credential. Declare a +`user_auth` block mapping DeerFlow user ids to credential header values: + +```json +{ + "mcpServers": { + "shared-service": { + "type": "http", + "url": "https://api.example.com/mcp", + "headers": { "Authorization": "$SERVICE_DISCOVERY_TOKEN" }, + "user_auth": { + "header": "Authorization", + "users": { + "alice-user-id": "$SERVICE_TOKEN_ALICE", + "bob-user-id": "$SERVICE_TOKEN_BOB" + } + } + } + } +} +``` + +On every tool call, the built-in user-scoped auth interceptor resolves the +authenticated DeerFlow user and injects that user's credential into the +configured header. The entry's static `headers` are used only for startup tool +discovery. + +- Values support the same `$ENV_VAR` resolution as the rest of the file, so raw + secrets can stay in the process environment. +- **Fail-closed by default**: a user with no mapped credential — or a mapped + `$ENV_VAR` that is unset — gets a clear error instead of another user's + credential. Set `"on_missing": "passthrough"` to instead forward such calls + with the server's static headers. +- Combined with per-`(user, thread)` MCP session scoping, users cannot reach + each other's authenticated sessions or credentials. +- The Gateway API masks `user_auth.users` values in GET responses and preserves + stored credentials when masked values are round-tripped through PUT. + +**Finding the right `users` key.** The resolved user id depends on how the +deployment authenticates: with LangGraph Server auth it is the safe-slug form +of the authenticated identity (e.g. `alice-example-com-ab12cd34`), while the +embedded Gateway resolves the DeerFlow user's raw id (a UUID). The simplest way +to get the exact key is to have the user attempt a call before being mapped: +the fail-closed error message includes their resolved id verbatim, ready to +copy into `user_auth.users`. + ## Managing MCP servers MCP servers can be managed in several ways: