Zhipeng Zheng 53a80d3ad1
feat(skills): per-user custom skill isolation with sandbox mounting (#3889)
* feat(skills): per-user skill isolation (#2905)

Implement user-scoped skill storage that isolates custom skills between
users while sharing public skills globally.

Key changes:
- Add UserScopedSkillStorage class for per-user custom skill directories
- Introduce get_or_new_user_skill_storage() factory with user_id context
- Auth middleware sets effective_user_id for request-scoped storage
- Agent/prompt/middleware now use user-scoped storage and prompt cache
- Sandbox mounts user-scoped skill directories for search/read tools
- Add validate_skill_file_path() to SkillStorage for path security
- Migration script supports --all-users bulk migration
- Frontend: add editable field to Skill type, error check in enableSkill
- All skill categories can be toggled (custom skills default to enabled)
- Update skill-creator SKILL.md with isolation-aware instructions

Tests:
- Add test_user_scoped_skill_storage.py (new)
- Update all existing skill tests for user-scoped storage
- Update sandbox, client, and router tests

* fix(skills): address second-round PR review feedback (#3889)

- P1-1: restrict legacy skill mount to users without custom skills
- P1-2: fail-closed for _is_disabled_skill_path (OSError → return True)
- P2-1: AND-merge global extensions_config skill disabled state
- P2-2: atomic write for _skill_states.json (mkstemp + replace)
- P2-3: normalize X-DeerFlow-Owner-User-Id in trusted boundary
- P2-4: LRU-bounded _enabled_skills_by_config_cache (OrderedDict, maxsize=256)
- P2-5: clear global prompt cache on PUBLIC skill toggle
- P2-6: invalidate skill caches on client.update_skill

* fix(tests): correct tool policy test after merge

* fix(skills): use DEFAULT_SKILLS_CONTAINER_PATH in UserScopedSkillStorage

The "/mnt/skills" literal in UserScopedSkillStorage.__init__ triggers
test_skill_container_path_defaults::test_mnt_skills_literal_is_owned_by_skill_constants_module
on CI. Migrate the default to the existing deerflow.constants constant,
matching the pattern already used by LocalSkillStorage, SkillStorage, and
the durable/tool_error middlewares.

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
2026-07-04 13:54:04 +08:00

294 lines
12 KiB
Python

"""Unified extensions configuration for MCP servers and skills."""
import json
import os
from pathlib import Path
from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, Field, model_validator
from deerflow.config.runtime_paths import existing_project_file
class McpOAuthConfig(BaseModel):
"""OAuth configuration for an MCP server (HTTP/SSE transports)."""
enabled: bool = Field(default=True, description="Whether OAuth token injection is enabled")
token_url: str = Field(description="OAuth token endpoint URL")
grant_type: Literal["client_credentials", "refresh_token"] = Field(
default="client_credentials",
description="OAuth grant type",
)
client_id: str | None = Field(default=None, description="OAuth client ID")
client_secret: str | None = Field(default=None, description="OAuth client secret")
refresh_token: str | None = Field(default=None, description="OAuth refresh token (for refresh_token grant)")
scope: str | None = Field(default=None, description="OAuth scope")
audience: str | None = Field(default=None, description="OAuth audience (provider-specific)")
token_field: str = Field(default="access_token", description="Field name containing access token in token response")
token_type_field: str = Field(default="token_type", description="Field name containing token type in token response")
expires_in_field: str = Field(default="expires_in", description="Field name containing expiry (seconds) in token response")
default_token_type: str = Field(default="Bearer", description="Default token type when missing in token response")
refresh_skew_seconds: int = Field(default=60, description="Refresh token this many seconds before expiry")
extra_token_params: dict[str, str] = Field(default_factory=dict, description="Additional form params sent to token endpoint")
model_config = ConfigDict(extra="allow")
class McpServerConfig(BaseModel):
"""Configuration for a single MCP server."""
enabled: bool = Field(default=True, description="Whether this MCP server is enabled")
type: str = Field(default="stdio", description="Transport type: 'stdio', 'sse', or 'http'")
command: str | None = Field(default=None, description="Command to execute to start the MCP server (for stdio type)")
args: list[str] = Field(default_factory=list, description="Arguments to pass to the command (for stdio type)")
env: dict[str, str] = Field(default_factory=dict, description="Environment variables for the MCP server")
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)")
description: str = Field(default="", description="Human-readable description of what this MCP server provides")
tool_call_timeout: float | None = Field(
default=None,
description="Timeout in seconds for individual stdio MCP tool calls. HTTP/SSE servers use transport-level timeouts. None means no timeout.",
)
model_config = ConfigDict(extra="allow")
@model_validator(mode="before")
@classmethod
def _accept_transport_alias(cls, data: Any) -> Any:
"""Accept the MCP-spec ``transport`` field as an alias for ``type``.
The official MCP configuration schema uses ``transport`` to indicate
the transport mechanism (``stdio``/``sse``/``http``). Earlier versions
of this project only honored ``type``, which caused remote SSE/HTTP
servers configured with just ``transport`` to be incorrectly treated as
``stdio`` (the default). This validator normalizes the two so either
spelling works, with ``type`` taking precedence when both are provided.
"""
if isinstance(data, dict):
transport = data.get("transport")
if transport and not data.get("type"):
data = {**data, "type": transport}
return data
class SkillStateConfig(BaseModel):
"""Configuration for a single skill's state."""
enabled: bool = Field(default=True, description="Whether this skill is enabled")
class ExtensionsConfig(BaseModel):
"""Unified configuration for MCP servers and skills."""
mcp_servers: dict[str, McpServerConfig] = Field(
default_factory=dict,
description="Map of MCP server name to configuration",
alias="mcpServers",
)
skills: dict[str, SkillStateConfig] = Field(
default_factory=dict,
description="Map of skill name to state configuration",
)
model_config = ConfigDict(extra="allow", populate_by_name=True)
@classmethod
def resolve_config_path(cls, config_path: str | None = None) -> Path | None:
"""Resolve the extensions config file path.
Priority:
1. If provided `config_path` argument, use it.
2. If provided `DEER_FLOW_EXTENSIONS_CONFIG_PATH` environment variable, use it.
3. Otherwise, search the caller project root for `extensions_config.json`, then `mcp_config.json`.
4. For backward compatibility, also search legacy backend/repository-root defaults.
5. If not found, return None (extensions are optional).
Args:
config_path: Optional path to extensions config file.
Resolution order:
1. If provided `config_path` argument, use it.
2. If provided `DEER_FLOW_EXTENSIONS_CONFIG_PATH` environment variable, use it.
3. Otherwise, search the caller project root for
`extensions_config.json`, then legacy `mcp_config.json`.
4. Finally, search backend/repository-root defaults for monorepo compatibility.
Returns:
Path to the extensions config file if found, otherwise None.
"""
if config_path:
path = Path(config_path)
if not path.exists():
raise FileNotFoundError(f"Extensions config file specified by param `config_path` not found at {path}")
return path
elif os.getenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH"):
path = Path(os.getenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH"))
if not path.exists():
raise FileNotFoundError(f"Extensions config file specified by environment variable `DEER_FLOW_EXTENSIONS_CONFIG_PATH` not found at {path}")
return path
else:
project_config = existing_project_file(("extensions_config.json", "mcp_config.json"))
if project_config is not None:
return project_config
backend_dir = Path(__file__).resolve().parents[4]
repo_root = backend_dir.parent
for path in (
backend_dir / "extensions_config.json",
repo_root / "extensions_config.json",
backend_dir / "mcp_config.json",
repo_root / "mcp_config.json",
):
if path.exists():
return path
# Extensions are optional, so return None if not found
return None
@classmethod
def from_file(cls, config_path: str | None = None) -> "ExtensionsConfig":
"""Load extensions config from JSON file.
See `resolve_config_path` for more details.
Args:
config_path: Path to the extensions config file.
Returns:
ExtensionsConfig: The loaded config, or empty config if file not found.
"""
resolved_path = cls.resolve_config_path(config_path)
if resolved_path is None:
# Return empty config if extensions config file is not found
return cls(mcp_servers={}, skills={})
try:
with open(resolved_path, encoding="utf-8") as f:
config_data = json.load(f)
config_data = cls.resolve_env_variables(config_data)
return cls.model_validate(config_data)
except json.JSONDecodeError as e:
raise ValueError(f"Extensions config file at {resolved_path} is not valid JSON: {e}") from e
except Exception as e:
raise RuntimeError(f"Failed to load extensions config from {resolved_path}: {e}") from e
@classmethod
def resolve_env_variables(cls, config: Any) -> Any:
"""Recursively resolve environment variables in the config.
Environment variables are resolved using the `os.getenv` function. Example: $OPENAI_API_KEY
Args:
config: The config to resolve environment variables in.
Returns:
The config with environment variables resolved.
"""
if isinstance(config, str):
if not config.startswith("$"):
return config
env_value = os.getenv(config[1:])
if env_value is None:
# Unresolved placeholder — store empty string so downstream
# consumers (e.g. MCP servers) don't receive the literal "$VAR"
# token as an actual environment value.
return ""
return env_value
if isinstance(config, dict):
return {key: cls.resolve_env_variables(value) for key, value in config.items()}
if isinstance(config, list):
return [cls.resolve_env_variables(item) for item in config]
if isinstance(config, tuple):
return tuple(cls.resolve_env_variables(item) for item in config)
return config
def get_enabled_mcp_servers(self) -> dict[str, McpServerConfig]:
"""Get only the enabled MCP servers.
Returns:
Dictionary of enabled MCP servers.
"""
return {name: config for name, config in self.mcp_servers.items() if config.enabled}
def is_skill_enabled(self, skill_name: str, skill_category: str) -> bool:
"""Check if a skill is enabled.
Args:
skill_name: Name of the skill
skill_category: Category of the skill (public, custom, or legacy)
Returns:
True if enabled, False otherwise.
Note:
All skill categories (public, custom, legacy) respect the
extensions_config enabled/disabled state. When no explicit
entry exists, skills default to enabled.
"""
skill_config = self.skills.get(skill_name)
if skill_config is None:
# Default to enabled for all skill categories
return skill_category in ("public", "custom", "legacy")
return skill_config.enabled
_extensions_config: ExtensionsConfig | None = None
def get_extensions_config() -> ExtensionsConfig:
"""Get the extensions config instance.
Returns a cached singleton instance. Use `reload_extensions_config()` to reload
from file, or `reset_extensions_config()` to clear the cache.
Returns:
The cached ExtensionsConfig instance.
"""
global _extensions_config
if _extensions_config is None:
_extensions_config = ExtensionsConfig.from_file()
return _extensions_config
def reload_extensions_config(config_path: str | None = None) -> ExtensionsConfig:
"""Reload the extensions config from file and update the cached instance.
This is useful when the config file has been modified and you want
to pick up the changes without restarting the application.
Args:
config_path: Optional path to extensions config file. If not provided,
uses the default resolution strategy.
Returns:
The newly loaded ExtensionsConfig instance.
"""
global _extensions_config
_extensions_config = ExtensionsConfig.from_file(config_path)
return _extensions_config
def reset_extensions_config() -> None:
"""Reset the cached extensions config instance.
This clears the singleton cache, causing the next call to
`get_extensions_config()` to reload from file. Useful for testing
or when switching between different configurations.
"""
global _extensions_config
_extensions_config = None
def set_extensions_config(config: ExtensionsConfig) -> None:
"""Set a custom extensions config instance.
This allows injecting a custom or mock config for testing purposes.
Args:
config: The ExtensionsConfig instance to use.
"""
global _extensions_config
_extensions_config = config