rob/CmdForge
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Harden M7.3 server: typed schemas, depth tracking, fnmatch, config preservation, and collision detection

Browse Source
This commit is contained in:
rob
2026-07-20 01:14:04 -03:00
parent 3525e2eeae
commit fe53721725
6 changed files with 502 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_mcp.py
+232 -7
View File
@@ -5,7 +5,8 @@ import time
from pathlib import Path
import pytest
from unittest.mock import patch, MagicMock
import yaml
from unittest.mock import patch
from cmdforge.mcp_client import (
DEFAULT_INHERITED_ENV,
McpServerConfig,
@@ -126,6 +127,17 @@ class TestMcpConfigPersistence:
perms = oct(temp_mcp_file.stat().st_mode & 0o777)
assert perms == "0o600"
def test_save_preserves_server_exposure_policy(self, temp_mcp_file):
temp_mcp_file.write_text(
"version: 1\nserver:\n expose: [summarize]\n deny: [dangerous]\n"
)
save_mcp_config([McpServerConfig(name="test", command="echo")])
data = yaml.safe_load(temp_mcp_file.read_text())
assert data["server"] == {
"expose": ["summarize"],
"deny": ["dangerous"],
}
class TestResultNormalization:
def test_auto_prefers_structured_content(self):
@@ -432,6 +444,50 @@ class TestMcpServeConfig:
assert config.is_exposed("extract")
assert not config.is_exposed("other")
@pytest.mark.parametrize(
"field, value",
[("expose", "*"), ("deny", "dangerous"), ("expose", [""])],
)
def test_rejects_malformed_policy_lists(self, field, value):
from cmdforge.mcp_client import McpServeConfig
values = {"expose": [], "deny": [], field: value}
with pytest.raises(ValueError, match="list of non-empty strings"):
McpServeConfig(**values)
def test_general_wildcard_matching(self):
from cmdforge.mcp_client import McpServeConfig
config = McpServeConfig(expose=["official/*"], deny=["*/dangerous"])
assert config.is_exposed("official/summarize")
assert not config.is_exposed("official/dangerous")
def test_private_tool_still_honors_deny(self):
from cmdforge.mcp_client import McpServeConfig
from cmdforge.mcp_server import _is_exposable
from cmdforge.tool import Tool
tool = Tool(name="dangerous", visibility="private")
config = McpServeConfig(expose=["official/dangerous"], deny=["official/*"])
assert not _is_exposable(tool, config, qualified_name="official/dangerous")
def test_private_qualified_tool_requires_qualified_expose(self):
from cmdforge.mcp_client import McpServeConfig
from cmdforge.mcp_server import _is_exposable
from cmdforge.tool import Tool
tool = Tool(name="extract", visibility="private")
assert not _is_exposable(
tool,
McpServeConfig(expose=["extract"]),
qualified_name="official/extract",
)
assert _is_exposable(
tool,
McpServeConfig(expose=["official/extract"]),
qualified_name="official/extract",
)
class TestToolNameMapping:
def test_simple_name_passes_through(self):
@@ -455,24 +511,26 @@ class TestArgumentToJsonSchema:
def test_integer_arg_with_default(self):
from cmdforge.tool import ToolArgument
from cmdforge.mcp_server import _argument_to_json_schema
arg = ToolArgument(flag="--limit", variable="limit", default="10")
setattr(arg, "type", "integer")
arg = ToolArgument(
flag="--limit", variable="limit", default="10", type="integer"
)
schema = _argument_to_json_schema(arg)
assert schema["type"] == "integer"
assert schema["default"] == "10"
assert schema["default"] == 10
def test_arg_with_enum(self):
from cmdforge.tool import ToolArgument
from cmdforge.mcp_server import _argument_to_json_schema
arg = ToolArgument(flag="--mode", variable="mode")
setattr(arg, "enum", ["fast", "accurate"])
arg = ToolArgument(
flag="--mode", variable="mode", enum=["fast", "accurate"]
)
schema = _argument_to_json_schema(arg)
assert schema["enum"] == ["fast", "accurate"]
class TestBuildToolSchema:
def test_builds_input_schema(self, tmp_path):
from cmdforge.tool import Tool, ToolArgument, save_tool
from cmdforge.tool import Tool, ToolArgument
from cmdforge.mcp_server import _build_tool_schema
with patch('cmdforge.tool.TOOLS_DIR', tmp_path / ".cmdforge"):
@@ -494,3 +552,170 @@ class TestBuildToolSchema:
assert schema["inputSchema"]["type"] == "object"
assert "name" in schema["inputSchema"]["properties"]
assert "count" in schema["inputSchema"]["properties"]
assert "input" in schema["inputSchema"]["properties"]
class TestRegisteredMcpTool:
def test_fastmcp_uses_flat_typed_schema_and_invokes_tool(self):
import asyncio
from mcp.server.fastmcp import FastMCP
from cmdforge.mcp_server import _build_tool_schema, _register_tool
from cmdforge.tool import Tool, ToolArgument
server = FastMCP("test")
tool = Tool(
name="greet",
arguments=[
ToolArgument(
flag="--name",
variable="name",
type="string",
required=True,
description="Who to greet",
),
ToolArgument(
flag="--count",
variable="count",
type="integer",
default=1,
enum=[1, 2, 3],
),
],
output="Hello {name} x{count}: {input}",
)
_register_tool(server, "greet", "greet", _build_tool_schema(tool), tool)
async def exercise():
listed = (await server.list_tools())[0]
result = await server.call_tool(
"greet", {"name": "Alice", "count": 2, "input": "welcome"}
)
return listed, result
listed, result = asyncio.run(exercise())
assert listed.inputSchema["required"] == ["name"]
assert listed.inputSchema["properties"]["count"]["type"] == "integer"
assert listed.inputSchema["properties"]["count"]["enum"] == [1, 2, 3]
assert "kwargs" not in listed.inputSchema["properties"]
assert result[0].text == "Hello Alice x2: welcome"
def test_handler_starts_at_inherited_depth(self):
import asyncio
from mcp.server.fastmcp import FastMCP
from cmdforge.mcp_server import _build_tool_schema, _register_tool
from cmdforge.tool import Tool
server = FastMCP("test")
tool = Tool(name="depth", output="ok")
with patch("cmdforge.runner.run_tool", return_value=("ok", 0)) as run:
_register_tool(server, "depth", "depth", _build_tool_schema(tool), tool)
asyncio.run(server.call_tool("depth", {}))
assert run.call_args.kwargs["_depth"] == 0
def test_handler_rejects_exhausted_mcp_depth(self, monkeypatch):
import asyncio
from mcp.server.fastmcp import FastMCP
from cmdforge.mcp_server import (
MAX_TOOL_DEPTH,
_build_tool_schema,
_register_tool,
)
from cmdforge.tool import Tool
monkeypatch.setenv("CMDFORGE_MCP_DEPTH", str(MAX_TOOL_DEPTH))
server = FastMCP("test")
tool = Tool(name="depth", output="ok")
_register_tool(server, "depth", "depth", _build_tool_schema(tool), tool)
with pytest.raises(Exception, match="Maximum MCP nesting depth"):
asyncio.run(server.call_tool("depth", {}))
def test_serve_fails_closed_on_mapped_name_collision(self):
from cmdforge.mcp_client import McpServeConfig
from cmdforge.mcp_server import serve
from cmdforge.tool import Tool
tools = {
"owner/tool": Tool(name="tool"),
"owner__tool": Tool(name="owner__tool"),
}
with (
patch(
"cmdforge.mcp_server.load_mcp_serve_config",
return_value=McpServeConfig(expose=["*"]),
),
patch("cmdforge.tool.list_tools", return_value=list(tools)),
patch("cmdforge.tool.load_tool", side_effect=tools.get),
patch("mcp.server.fastmcp.FastMCP.run"),
):
with pytest.raises(RuntimeError, match="name collision"):
serve()
class TestCmdForgeMcpServerEndToEnd:
def test_stdio_discovery_and_invocation(self, tmp_path, monkeypatch):
pytest.importorskip("mcp")
home = tmp_path / "home"
cmdforge_dir = home / ".cmdforge"
cmdforge_dir.mkdir(parents=True)
(cmdforge_dir / "mcp.yaml").write_text(
yaml.safe_dump({
"version": 1,
"server": {
"expose": ["greet", "echo-input", "hidden"],
"deny": ["hidden"],
},
})
)
tools = {
"greet": {
"name": "greet",
"arguments": [{
"flag": "--name",
"variable": "name",
"type": "string",
"required": True,
}],
"steps": [],
"output": "Hello {name}",
},
"echo-input": {
"name": "echo-input",
"steps": [],
"output": "{input}",
},
"hidden": {
"name": "hidden",
"visibility": "private",
"steps": [],
"output": "secret",
},
}
for name, data in tools.items():
tool_dir = cmdforge_dir / name
tool_dir.mkdir()
(tool_dir / "config.yaml").write_text(yaml.safe_dump(data))
client_config = tmp_path / "client-mcp.yaml"
monkeypatch.setattr("cmdforge.mcp_client.MCP_CONFIG_FILE", client_config)
save_mcp_config([
McpServerConfig(
name="cmdforge",
command=sys.executable,
args=["-m", "cmdforge.cli", "mcp", "serve"],
env={"HOME": str(home)},
timeout=10,
approved=True,
)
])
manager = McpClientManager()
discovered = {tool["name"]: tool for tool in manager.discover("cmdforge")}
assert set(discovered) == {"greet", "echo-input"}
assert discovered["greet"]["inputSchema"]["required"] == ["name"]
assert manager.call_tool(
"cmdforge", "greet", {"name": "Alice"}, "text"
) == "Hello Alice"
assert manager.call_tool(
"cmdforge", "echo-input", {"input": "through stdin"}, "text"
) == "through stdin"
tests/test_tool.py
+25
View File
@@ -80,6 +80,31 @@ class TestToolArgument:
assert restored.default == original.default
assert restored.description == original.description
def test_typed_metadata_roundtrip(self):
original = ToolArgument(
flag="--limit",
variable="limit",
default=10,
description="Maximum results",
type="integer",
enum=[10, 25, 50],
required=True,
)
assert ToolArgument.from_dict(original.to_dict()) == original
@pytest.mark.parametrize(
"values",
[
{"type": "object"},
{"enum": []},
{"enum": "fast"},
{"required": "yes"},
],
)
def test_rejects_invalid_typed_metadata(self, values):
with pytest.raises(ValueError):
ToolArgument(flag="--value", variable="value", **values)
class TestPromptStep:
"""Tests for PromptStep dataclass."""