| 1 | #!/usr/bin/env python3 |
| 2 | """Offline conversion contracts; only the synthetic Node fixture is executed.""" |
| 3 | |
| 4 | import argparse |
| 5 | import importlib.util |
| 6 | import json |
| 7 | import os |
| 8 | from pathlib import Path |
| 9 | import shutil |
| 10 | import subprocess |
| 11 | import sys |
| 12 | import tempfile |
| 13 | import unittest |
| 14 | from unittest import mock |
| 15 | |
| 16 | import yaml |
| 17 | |
| 18 | |
| 19 | SCRIPT = Path(__file__).resolve().with_name("convert-plugin.py") |
| 20 | SPEC = importlib.util.spec_from_file_location("convert_plugin", SCRIPT) |
| 21 | assert SPEC and SPEC.loader |
| 22 | converter = importlib.util.module_from_spec(SPEC) |
| 23 | SPEC.loader.exec_module(converter) |
| 24 | CANARY = "conversion-secret-canary-do-not-emit-7391" |
| 25 | |
| 26 | |
| 27 | class ConversionTests(unittest.TestCase): |
| 28 | def setUp(self): |
| 29 | self.temp = tempfile.TemporaryDirectory(prefix="plugin-conversion-test-") |
| 30 | self.addCleanup(self.temp.cleanup) |
| 31 | # macOS /tmp and /var are links; the converter intentionally rejects them. |
| 32 | self.root = Path(self.temp.name).resolve() |
| 33 | self.sequence = 0 |
| 34 | |
| 35 | def fresh(self, prefix): |
| 36 | self.sequence += 1 |
| 37 | return self.root / f"{prefix}-{self.sequence}" |
| 38 | |
| 39 | def write(self, content, suffix=".json"): |
| 40 | path = self.fresh("source").with_suffix(suffix) |
| 41 | path.write_text(content, encoding="utf-8") |
| 42 | return path |
| 43 | |
| 44 | def config(self, document): |
| 45 | return self.write(json.dumps(document)) |
| 46 | |
| 47 | def remote(self, **changes): |
| 48 | return {"type": "remote", "url": "https://docs.example.invalid/mcp", "oauth": False, **changes} |
| 49 | |
| 50 | def v1(self, **changes): |
| 51 | return {"mcp": {"docs": self.remote(**changes)}} |
| 52 | |
| 53 | def dsh(self, **changes): |
| 54 | return [{"id": "docs-entry", "name": "@deepseek-ai/dsh-mcp-client", "config": { |
| 55 | "serverName": "docs", "transport": "streamable-http", |
| 56 | "url": "https://docs.example.invalid/mcp", **changes}}] |
| 57 | |
| 58 | def skill(self, name="safe-skill", *, metadata=None, body="Read the local reference.\n", directory=True): |
| 59 | meta = {"name": name, "description": "Local reference guidance", **(metadata or {})} |
| 60 | text = "---\n" + yaml.safe_dump(meta, sort_keys=False) + "---\n" + body |
| 61 | if not directory: |
| 62 | return self.write(text, ".md") |
| 63 | path = self.fresh("skill") |
| 64 | path.mkdir() |
| 65 | (path / "SKILL.md").write_text(text, encoding="utf-8") |
| 66 | return path |
| 67 | |
| 68 | def args(self, *, config=None, bundle=None, skills=(), dialect="opencode-v1", output=None, |
| 69 | name="converted-demo", stdio_roots=()): |
| 70 | return argparse.Namespace(config=config, bundle=bundle, skill=list(skills), format=dialect, |
| 71 | output=output or self.fresh("output"), name=name, stdio_root=list(stdio_roots)) |
| 72 | |
| 73 | def cli(self, args, *, env=None): |
| 74 | command = [sys.executable, "-B", str(SCRIPT), "--format", args.format, |
| 75 | "--name", args.name, "--output", str(args.output)] |
| 76 | if args.config: |
| 77 | command += ["--config", str(args.config)] |
| 78 | if args.bundle: |
| 79 | command += ["--bundle", str(args.bundle)] |
| 80 | for skill in args.skill: |
| 81 | command += ["--skill", str(skill)] |
| 82 | for root in args.stdio_root: |
| 83 | command += ["--stdio-root", root] |
| 84 | environment = dict(os.environ) |
| 85 | environment.update(env or {}) |
| 86 | environment["PYTHONDONTWRITEBYTECODE"] = "1" |
| 87 | return subprocess.run(command, capture_output=True, text=True, timeout=15, |
| 88 | cwd=self.root, env=environment, check=False) |
| 89 | |
| 90 | def refuse(self, args, *, message=None): |
| 91 | with self.assertRaises(converter.ConversionError) as failure: |
| 92 | converter.convert(args) |
| 93 | if message: |
| 94 | self.assertIn(message, str(failure.exception)) |
| 95 | self.assertFalse(args.output.exists(), "A rejected conversion must not publish a partial bundle") |
| 96 | |
| 97 | def servers(self, output): |
| 98 | return json.loads((output / "mcp.json").read_text())["mcpServers"] |
| 99 | |
| 100 | def assert_no_canary(self, result, output): |
| 101 | self.assertNotIn(CANARY, result.stdout) |
| 102 | self.assertNotIn(CANARY, result.stderr) |
| 103 | if output.exists(): |
| 104 | for path in output.rglob("*"): |
| 105 | if path.is_file(): |
| 106 | self.assertNotIn(CANARY.encode(), path.read_bytes()) |
| 107 | |
| 108 | def test_cli_v1_preserves_disable_request_timeout_and_env_name(self): |
| 109 | args = self.args(config=self.config(self.v1( |
| 110 | enabled=False, timeout=7000, headers={"Authorization": "{env:CONVERSION_TEST_TOKEN}"}))) |
| 111 | result = self.cli(args, env={"CONVERSION_TEST_TOKEN": CANARY}) |
| 112 | self.assertEqual(result.returncode, 0, result.stderr) |
| 113 | self.assertEqual(self.servers(args.output), {"docs": { |
| 114 | "type": "streamable-http", "url": "https://docs.example.invalid/mcp", |
| 115 | "extensions": {"net.codewhale": {"disabled": True, "connect_timeout": 7, |
| 116 | "execute_timeout": 7, "env_headers": {"Authorization": "CONVERSION_TEST_TOKEN"}}}}}) |
| 117 | manifest = json.loads((args.output / "plugin.json").read_text()) |
| 118 | self.assertEqual(manifest["extensions"]["net.codewhale"]["capabilities"]["network_hosts"], |
| 119 | ["docs.example.invalid"]) |
| 120 | self.assert_no_canary(result, args.output) |
| 121 | |
| 122 | def test_v2_preserves_global_timeout_and_per_server_override(self): |
| 123 | document = {"mcp": {"timeout": {"startup": 4000, "request": 11000}, "servers": { |
| 124 | "docs": self.remote(disabled=True, timeout={"request": 23000}), |
| 125 | "other": self.remote(disabled=False)}}} |
| 126 | args = self.args(config=self.config(document), dialect="opencode-v2") |
| 127 | self.assertEqual(converter.convert(args), (0, 2, 0)) |
| 128 | servers = self.servers(args.output) |
| 129 | self.assertEqual(servers["docs"]["extensions"]["net.codewhale"], |
| 130 | {"disabled": True, "connect_timeout": 4, "execute_timeout": 23}) |
| 131 | self.assertEqual(servers["other"]["extensions"]["net.codewhale"], |
| 132 | {"disabled": False, "connect_timeout": 4, "execute_timeout": 11}) |
| 133 | |
| 134 | def test_cli_dsh_preserves_disabled_and_tool_timeout(self): |
| 135 | document = self.dsh(toolCallTimeoutMs=19000) |
| 136 | document[0]["disabled"] = True |
| 137 | args = self.args(config=self.write(yaml.safe_dump(document), ".yml"), dialect="dsh") |
| 138 | result = self.cli(args) |
| 139 | self.assertEqual(result.returncode, 0, result.stderr) |
| 140 | self.assertEqual(self.servers(args.output)["docs"], { |
| 141 | "type": "streamable-http", "url": "https://docs.example.invalid/mcp", |
| 142 | "extensions": {"net.codewhale": {"disabled": True, "execute_timeout": 19}}}) |
| 143 | |
| 144 | def node_source(self): |
| 145 | root = self.fresh("packaged-node") |
| 146 | root.mkdir() |
| 147 | (root / "server.mjs").write_text("throw new Error('converter must never execute this');\n") |
| 148 | return root |
| 149 | |
| 150 | def local(self, **changes): |
| 151 | return {"type": "local", "command": ["node", "server.mjs"], **changes} |
| 152 | |
| 153 | def test_cli_local_node_copies_source_without_execution_or_credential_lookup(self): |
| 154 | root = self.node_source() |
| 155 | (root / "resource.json").write_text('{"answer":42}') |
| 156 | args = self.args(config=self.config({"mcp": {"localdocs": self.local( |
| 157 | enabled=False, timeout=7000, environment={"API_TOKEN": "{env:CONVERSION_TEST_TOKEN}"})}}), |
| 158 | stdio_roots=[f"localdocs={root}"]) |
| 159 | result = self.cli(args, env={"CONVERSION_TEST_TOKEN": CANARY}) |
| 160 | self.assertEqual(result.returncode, 0, result.stderr) |
| 161 | self.assertEqual(self.servers(args.output)["localdocs"], { |
| 162 | "type": "stdio", "command": "node", "args": ["server.mjs"], "cwd": "mcp/localdocs", |
| 163 | "env": {"API_TOKEN": "${CONVERSION_TEST_TOKEN}"}, |
| 164 | "extensions": {"net.codewhale": {"disabled": True, "connect_timeout": 7, "execute_timeout": 7}}}) |
| 165 | self.assertEqual((args.output / "mcp/localdocs/server.mjs").read_bytes(), (root / "server.mjs").read_bytes()) |
| 166 | self.assertEqual((args.output / "mcp/localdocs/resource.json").read_bytes(), (root / "resource.json").read_bytes()) |
| 167 | extension = json.loads((args.output / "plugin.json").read_text())["extensions"]["net.codewhale"] |
| 168 | self.assertEqual(extension, {"when": {"binaries": ["node"]}}) |
| 169 | self.assert_no_canary(result, args.output) |
| 170 | |
| 171 | def test_local_node_v2_and_dsh_preserve_disable_cwd_and_timeouts(self): |
| 172 | root = self.node_source() |
| 173 | cases = [("opencode-v2", {"mcp": {"timeout": {"startup": 4000, "request": 11000}, "servers": { |
| 174 | "docs": self.local(cwd=".", disabled=True, timeout={"request": 23000})}}}, |
| 175 | {"disabled": True, "connect_timeout": 4, "execute_timeout": 23}), |
| 176 | ("dsh", [{"name": "@deepseek-ai/dsh-mcp-client", "disabled": True, "config": { |
| 177 | "serverName": "docs", "transport": "stdio", "command": "node", "args": ["server.mjs"], |
| 178 | "cwd": ".", "env": {}, "toolCallTimeoutMs": 19000}}], |
| 179 | {"disabled": True, "execute_timeout": 19})] |
| 180 | for dialect, document, extension in cases: |
| 181 | with self.subTest(dialect=dialect): |
| 182 | args = self.args(config=self.config(document), dialect=dialect, stdio_roots=[f"docs={root}"]) |
| 183 | self.assertEqual(converter.convert(args), (0, 1, 0)) |
| 184 | server = self.servers(args.output)["docs"] |
| 185 | self.assertEqual(server["cwd"], "mcp/docs") |
| 186 | self.assertEqual(server["extensions"]["net.codewhale"], extension) |
| 187 | |
| 188 | @unittest.skipUnless(shutil.which("node"), "Node is needed for the synthetic MCP fixture") |
| 189 | def test_packaged_node_mcp_discovers_and_calls_tool_with_sibling_and_cwd_resource(self): |
| 190 | root = self.node_source() |
| 191 | (root / "resource.json").write_text('{"answer":42}') |
| 192 | (root / "sibling.mjs").write_text("export const name = 'fixture_answer';\n") |
| 193 | (root / "server.mjs").write_text('''import readline from 'node:readline'; |
| 194 | import { readFileSync } from 'node:fs'; |
| 195 | import { name } from './sibling.mjs'; |
| 196 | for await (const line of readline.createInterface({ input: process.stdin })) { |
| 197 | const request = JSON.parse(line); |
| 198 | if (request.id === undefined) continue; |
| 199 | const result = request.method === 'initialize' |
| 200 | ? { protocolVersion: '2024-11-05', capabilities: { tools: {} }, serverInfo: { name: 'fixture', version: '1' } } |
| 201 | : request.method === 'tools/list' |
| 202 | ? { tools: [{ name, description: 'Read the packaged answer', inputSchema: { type: 'object' } }] } |
| 203 | : { content: [{ type: 'text', text: String(JSON.parse(readFileSync('resource.json', 'utf8')).answer) }] }; |
| 204 | process.stdout.write(JSON.stringify({ jsonrpc: '2.0', id: request.id, result }) + '\\n'); |
| 205 | } |
| 206 | ''') |
| 207 | args = self.args(config=self.config({"mcp": {"docs": self.local()}}), stdio_roots=[f"docs={root}"]) |
| 208 | converter.convert(args) |
| 209 | # Mutating original resources cannot change the converted package. |
| 210 | (root / "resource.json").write_text('{"answer":99}') |
| 211 | server = self.servers(args.output)["docs"] |
| 212 | requests = [ |
| 213 | {"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {}}, |
| 214 | {"jsonrpc": "2.0", "method": "notifications/initialized"}, |
| 215 | {"jsonrpc": "2.0", "id": 2, "method": "tools/list"}, |
| 216 | {"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": {"name": "fixture_answer", "arguments": {}}}, |
| 217 | ] |
| 218 | result = subprocess.run([shutil.which("node"), *server["args"]], cwd=args.output / server["cwd"], |
| 219 | input="".join(json.dumps(request) + "\n" for request in requests), text=True, capture_output=True, |
| 220 | timeout=10, check=False, env={"PATH": str(Path(shutil.which("node")).parent)}) |
| 221 | self.assertEqual(result.returncode, 0, result.stderr) |
| 222 | responses = [json.loads(line) for line in result.stdout.splitlines()] |
| 223 | self.assertEqual(responses[1]["result"]["tools"][0]["name"], "fixture_answer") |
| 224 | self.assertEqual(responses[2]["result"]["content"], [{"type": "text", "text": "42"}]) |
| 225 | |
| 226 | @unittest.skipUnless(shutil.which("node"), "Node is needed for the synthetic module fixture") |
| 227 | def test_packaged_js_and_cjs_preserve_node_module_context(self): |
| 228 | for entry, package_type, esm in (("server.js", "module", True), |
| 229 | ("server.js", "commonjs", False), |
| 230 | ("server.cjs", "module", False)): |
| 231 | with self.subTest(entry=entry, package_type=package_type): |
| 232 | root = self.fresh("module-context") |
| 233 | root.mkdir() |
| 234 | (root / "package.json").write_text(json.dumps({"type": package_type})) |
| 235 | (root / "resource.json").write_text('{"answer":42}') |
| 236 | (root / "helper.cjs").write_text("exports.answer = 7;\n") |
| 237 | imports = ("import fs from 'node:fs'; import helper from './helper.cjs';\n" if esm else |
| 238 | "const fs = require('node:fs'); const helper = require('./helper.cjs');\n") |
| 239 | (root / entry).write_text(imports + |
| 240 | "console.log(JSON.stringify([helper.answer, JSON.parse(fs.readFileSync('resource.json', 'utf8')).answer]));\n") |
| 241 | args = self.args(config=self.config({"mcp": {"docs": self.local(command=["node", entry])}}), |
| 242 | stdio_roots=[f"docs={root}"]) |
| 243 | self.assertEqual(converter.convert(args), (0, 1, 0)) |
| 244 | (root / "resource.json").write_text('{"answer":99}') |
| 245 | server = self.servers(args.output)["docs"] |
| 246 | result = subprocess.run([shutil.which("node"), *server["args"]], |
| 247 | cwd=args.output / server["cwd"], text=True, capture_output=True, timeout=10, |
| 248 | check=False, env={"PATH": str(Path(shutil.which("node")).parent)}) |
| 249 | self.assertEqual(result.returncode, 0, result.stderr) |
| 250 | self.assertEqual(json.loads(result.stdout), [7, 42]) |
| 251 | |
| 252 | def test_local_node_requires_matching_explicit_roots_and_safe_launcher(self): |
| 253 | root = self.node_source() |
| 254 | for command in (["node", "../server.mjs"], ["node", "/server.mjs"], ["node", "C:\\server.mjs"], |
| 255 | ["node", "server.ts"], ["node", "server.py"], |
| 256 | ["node", "--eval", "process.exit()"], ["node", "server.mjs", CANARY], |
| 257 | ["npx", "some-server"], ["sh", "server.mjs"], ["node", "https://example.invalid/server.mjs"], |
| 258 | ["node", "{env:ENTRY}.mjs"], ["node", 1]): |
| 259 | with self.subTest(command=command): |
| 260 | self.refuse(self.args(config=self.config({"mcp": {"docs": self.local(command=command)}}), |
| 261 | stdio_roots=[f"docs={root}"])) |
| 262 | config = self.config({"mcp": {"docs": self.local()}}) |
| 263 | for roots in ([], [f"other={root}"], [f"docs={root}", f"docs={root}"], [f"docs={root}", f"other={root}"]): |
| 264 | self.refuse(self.args(config=config, stdio_roots=roots)) |
| 265 | self.refuse(self.args(skills=[self.skill()], stdio_roots=[f"docs={root}"])) |
| 266 | self.refuse(self.args(config=self.config(self.v1()), stdio_roots=[f"docs={root}"])) |
| 267 | self.refuse(self.args(config=config, stdio_roots=[f"docs={root}"], output=root / "output")) |
| 268 | |
| 269 | def test_local_node_rejects_literal_environment_loader_overrides_and_nonportable_cwd(self): |
| 270 | root = self.node_source() |
| 271 | for environment in ({"TOKEN": CANARY}, {"TOKEN": "{file:/private/key}"}, {"PLUGIN_ROOT": "{env:TOKEN}"}, |
| 272 | {"node_options": "{env:OPTIONS}"}, {"NODE_PATH": "{env:IMPORTS}"}, {"PATH": "{env:PATH}"}, |
| 273 | {"DYLD_INSERT_LIBRARIES": "{env:LIBRARY}"}, {"LD_PRELOAD": "{env:LIBRARY}"}): |
| 274 | args = self.args(config=self.config({"mcp": {"docs": self.local(environment=environment)}}), |
| 275 | stdio_roots=[f"docs={root}"]) |
| 276 | result = self.cli(args) |
| 277 | self.assertEqual(result.returncode, 1) |
| 278 | self.assertFalse(args.output.exists()) |
| 279 | self.assert_no_canary(result, args.output) |
| 280 | for cwd in ("..", "./workspace", "/workspace", ["."]): |
| 281 | self.refuse(self.args(config=self.config({"mcp": {"servers": {"docs": self.local(cwd=cwd)}}}), |
| 282 | dialect="opencode-v2", stdio_roots=[f"docs={root}"])) |
| 283 | for extra in ({"env": {"TOKEN": "{env:TOKEN}"}}, {"failOnStartupError": True}, {"reconnect": {}}): |
| 284 | config = {"serverName": "docs", "transport": "stdio", "command": "node", "args": ["server.mjs"], **extra} |
| 285 | self.refuse(self.args(config=self.config([{"name": "@deepseek-ai/dsh-mcp-client", "config": config}]), |
| 286 | dialect="dsh", stdio_roots=[f"docs={root}"])) |
| 287 | |
| 288 | def test_local_node_refuses_secret_and_linked_dependencies_atomically(self): |
| 289 | for name in (".env.local", ".gitignore", ".npmrc", "credentials.json", "server.key", "identity.pem"): |
| 290 | root = self.node_source() |
| 291 | (root / name).write_text(CANARY) |
| 292 | args = self.args(config=self.config({"mcp": {"docs": self.local()}}), stdio_roots=[f"docs={root}"]) |
| 293 | result = self.cli(args) |
| 294 | self.assertEqual(result.returncode, 1) |
| 295 | self.assert_no_canary(result, args.output) |
| 296 | self.assertFalse(args.output.exists()) |
| 297 | for kind in ("symlink", "hardlink", "directory-link"): |
| 298 | root = self.node_source() |
| 299 | outside = self.write("outside", ".mjs") |
| 300 | if kind == "hardlink": |
| 301 | os.link(outside, root / "dependency.mjs") |
| 302 | elif kind == "directory-link": |
| 303 | (root / "node_modules").symlink_to(self.root, target_is_directory=True) |
| 304 | else: |
| 305 | (root / "dependency.mjs").symlink_to(outside) |
| 306 | self.refuse(self.args(config=self.config({"mcp": {"docs": self.local()}}), stdio_roots=[f"docs={root}"])) |
| 307 | self.assertEqual(outside.read_text(), "outside") |
| 308 | |
| 309 | def test_local_node_shares_aggregate_copy_budget_with_skills_and_servers(self): |
| 310 | root = self.node_source() |
| 311 | (root / "resource.bin").write_bytes(b"a" * 2300) |
| 312 | config = self.config({"mcp": {"docs": self.local(), "second": self.local()}}) |
| 313 | with mock.patch.object(converter, "MAX_BYTES", 4000): |
| 314 | self.refuse(self.args(config=config, stdio_roots=[f"docs={root}", f"second={root}"])) |
| 315 | with mock.patch.object(converter, "MAX_FILES", 4): |
| 316 | self.refuse(self.args(config=config, stdio_roots=[f"docs={root}", f"second={root}"])) |
| 317 | |
| 318 | def test_skill_keeps_explicit_invocation_and_metadata_out_of_frontmatter(self): |
| 319 | extra = {"name": "wrong-name", "invocation": "automatic", "description": "wrong description"} |
| 320 | skill = self.skill(metadata={"disable-model-invocation": True, "metadata": extra, |
| 321 | "license": "MIT", "description": "Line one\ninvocation: automatic"}) |
| 322 | sentinel = self.root / "executed" |
| 323 | code = f"from pathlib import Path\nPath({str(sentinel)!r}).write_text('executed')\n" |
| 324 | (skill / "helper.py").write_text(code) |
| 325 | args = self.args(skills=[skill]) |
| 326 | result = self.cli(args) |
| 327 | self.assertEqual(result.returncode, 0, result.stderr) |
| 328 | generated = args.output / "skills/safe-skill" |
| 329 | text = (generated / "SKILL.md").read_text() |
| 330 | front = yaml.safe_load(text.split("---", 2)[1]) |
| 331 | self.assertEqual(front, {"name": "safe-skill", "description": "Line one\ninvocation: automatic", |
| 332 | "invocation": "explicit-only"}) |
| 333 | self.assertEqual(json.loads((generated / "SOURCE_SKILL_METADATA.json").read_text()), |
| 334 | {"metadata": extra, "license": "MIT"}) |
| 335 | self.assertEqual((generated / "helper.py").read_text(), code) |
| 336 | self.assertFalse(sentinel.exists()) |
| 337 | self.assertIn("Read the local reference.", text) |
| 338 | |
| 339 | def test_flat_markdown_skill_uses_declared_name(self): |
| 340 | skill = self.skill("flat-guide", directory=False) |
| 341 | args = self.args(skills=[skill], dialect="opencode-v2") |
| 342 | self.assertEqual(converter.convert(args), (1, 0, 0)) |
| 343 | self.assertTrue((args.output / "skills/flat-guide/SKILL.md").is_file()) |
| 344 | |
| 345 | def test_skill_delimiter_and_unrepresentable_invocation_are_refused(self): |
| 346 | for metadata in ({"description": "Review --- carefully", "disable-model-invocation": True}, |
| 347 | {"user-invocable": False}, {"disable-model-invocation": "false"}): |
| 348 | with self.subTest(metadata=metadata): |
| 349 | self.refuse(self.args(skills=[self.skill(metadata=metadata)])) |
| 350 | |
| 351 | def test_executable_plugin_declarations_never_run(self): |
| 352 | sentinel = self.root / "foreign-code-ran" |
| 353 | plugin = self.root / "plugin.py" |
| 354 | plugin.write_text(f"from pathlib import Path\nPath({str(sentinel)!r}).touch()\n") |
| 355 | cases = [("opencode-v1", {"plugin": [str(plugin)], **self.v1()}), |
| 356 | ("opencode-v2", {"plugins": [{"package": str(plugin)}], "mcp": {"servers": {"docs": self.remote()}}}), |
| 357 | ("dsh", [{"id": "foreign", "name": str(plugin), "config": {}}])] |
| 358 | for dialect, document in cases: |
| 359 | with self.subTest(dialect=dialect): |
| 360 | args = self.args(config=self.config(document), dialect=dialect) |
| 361 | result = self.cli(args) |
| 362 | self.assertEqual(result.returncode, 1) |
| 363 | self.assertFalse(args.output.exists()) |
| 364 | self.assertFalse(sentinel.exists()) |
| 365 | |
| 366 | def test_cli_global_tool_disable_is_not_dropped_during_mcp_conversion(self): |
| 367 | document = {**self.v1(), "tools": {"docs*": False}} |
| 368 | source = self.config(document) |
| 369 | original = source.read_bytes() |
| 370 | args = self.args(config=source) |
| 371 | result = self.cli(args) |
| 372 | self.assertEqual(result.returncode, 1) |
| 373 | self.assertIn("preserve their restrictions in Codewhale", result.stderr) |
| 374 | self.assertFalse(args.output.exists()) |
| 375 | self.assertEqual(source.read_bytes(), original) |
| 376 | |
| 377 | def test_opencode_policy_fields_require_a_manual_port_even_for_disabled_servers(self): |
| 378 | rules = [{"action": "docs_*", "resource": "*", "effect": "deny"}] |
| 379 | policies = {"tools": {"docs*": False}, "permission": {"docs_*": "deny"}, |
| 380 | "permissions": rules, |
| 381 | "agent": {"reviewer": {"permission": {"docs_*": "deny"}}}, |
| 382 | "agents": {"reviewer": {"permissions": rules}}, |
| 383 | "mode": {"plan": {"tools": {"docs*": False}}}, "default_agent": "plan"} |
| 384 | for dialect in ("opencode-v1", "opencode-v2"): |
| 385 | for disabled in (False, True): |
| 386 | for field, value in policies.items(): |
| 387 | with self.subTest(dialect=dialect, disabled=disabled, field=field): |
| 388 | document = self.v1(enabled=not disabled) if dialect == "opencode-v1" else { |
| 389 | "mcp": {"servers": {"docs": self.remote(disabled=disabled)}}} |
| 390 | document[field] = value |
| 391 | self.refuse(self.args(config=self.config(document), dialect=dialect), |
| 392 | message="require a manual port") |
| 393 | |
| 394 | def test_dsh_tag_and_plain_expression_are_rejected_even_when_disabled(self): |
| 395 | sentinel = self.root / "expression-ran" |
| 396 | expression = f"require('node:fs').writeFileSync({json.dumps(str(sentinel))}, 'ran')" |
| 397 | document = self.dsh() |
| 398 | document[0]["disabled"] = True |
| 399 | document[0]["config"]["headers"] = {"Authorization": {"__jsExpr": expression}} |
| 400 | tagged = yaml.safe_dump(self.dsh()).replace("transport: streamable-http", "transport: !!js " + expression) |
| 401 | for text in (json.dumps(document), tagged): |
| 402 | with self.subTest(text=text): |
| 403 | self.refuse(self.args(config=self.write(text, ".yml"), dialect="dsh")) |
| 404 | self.assertFalse(sentinel.exists()) |
| 405 | |
| 406 | def test_dsh_patch_followed_by_disable_is_not_partially_imported(self): |
| 407 | patches = [{"insert": self.dsh()}, {"id": "docs-entry", "disabled": True, |
| 408 | "config": {"url": "https://replacement.example.invalid/mcp"}}] |
| 409 | self.refuse(self.args(config=self.config(patches), dialect="dsh")) |
| 410 | |
| 411 | def bundle(self, patch_text=None, manifest=None, patch_name="cordis.patch.yml"): |
| 412 | directory = self.fresh("dsh-bundle") |
| 413 | directory.mkdir() |
| 414 | package = {"name": "@demo/tools-dsh", "version": "1.2.3", |
| 415 | "dsh": {"bundle": {"patch": f"./{patch_name}"}}, **(manifest or {})} |
| 416 | (directory / "package.json").write_text(json.dumps(package)) |
| 417 | if patch_text is not None: |
| 418 | (directory / patch_name).write_text(patch_text) |
| 419 | return directory |
| 420 | |
| 421 | def test_dsh_bundle_evaluates_patches_and_skips_foreign_rows(self): |
| 422 | bundle = self.bundle(yaml.safe_dump([ |
| 423 | {"insert": [ |
| 424 | {"id": "docs-entry", "name": "@deepseek-ai/dsh-mcp-client", "config": { |
| 425 | "serverName": "docs", "transport": "streamable-http", |
| 426 | "url": "https://docs.example.invalid/mcp", "toolCallTimeoutMs": 19000}}, |
| 427 | {"id": "skin", "name": "@deepseek-ai/dsh-client-ui-theme", "config": {"hue": 4}}, |
| 428 | ]}, |
| 429 | {"id": "docs-entry", "disabled": True}, |
| 430 | {"id": "ghost", "disabled": True}, |
| 431 | ])) |
| 432 | args = self.args(bundle=bundle, dialect="dsh") |
| 433 | result = self.cli(args) |
| 434 | self.assertEqual(result.returncode, 0, result.stderr) |
| 435 | self.assertEqual(self.servers(args.output)["docs"], { |
| 436 | "type": "streamable-http", "url": "https://docs.example.invalid/mcp", |
| 437 | "extensions": {"net.codewhale": {"disabled": True, "execute_timeout": 19}}}) |
| 438 | receipt = (args.output / "CONVERSION.md").read_text() |
| 439 | self.assertIn("@demo/tools-dsh@1.2.3", receipt) |
| 440 | self.assertIn("skin", receipt) |
| 441 | self.assertIn("ghost", receipt) |
| 442 | |
| 443 | def test_dsh_bundle_lowers_js_command_and_host_path_arg(self): |
| 444 | bundle = self.bundle() |
| 445 | server_dir = self.node_source() |
| 446 | patch = ("- insert:\n - id: local-entry\n name: '@deepseek-ai/dsh-mcp-client'\n" |
| 447 | " config:\n serverName: localdocs\n transport: stdio\n" |
| 448 | " command: !!js process.execPath\n" |
| 449 | " args:\n - !!js process.env.CONVERT_TEST_UNSET_ENTRY_7391 || '" |
| 450 | + str(server_dir / "server.mjs") + "'\n") |
| 451 | (bundle / "cordis.patch.yml").write_text(patch) |
| 452 | args = self.args(bundle=bundle, dialect="dsh") |
| 453 | result = self.cli(args) |
| 454 | self.assertEqual(result.returncode, 0, result.stderr) |
| 455 | self.assertEqual(self.servers(args.output)["localdocs"], { |
| 456 | "type": "stdio", "command": "node", "args": ["server.mjs"], "cwd": "mcp/localdocs", |
| 457 | "env": {}, "extensions": {"net.codewhale": {}}}) |
| 458 | self.assertEqual((args.output / "mcp/localdocs/server.mjs").read_bytes(), |
| 459 | (server_dir / "server.mjs").read_bytes()) |
| 460 | |
| 461 | def test_dsh_bundle_relative_entry_and_group_children_convert(self): |
| 462 | bundle = self.bundle() |
| 463 | (bundle / "mcp").mkdir() |
| 464 | (bundle / "mcp" / "server.mjs").write_text("// fixture\n") |
| 465 | patch = yaml.safe_dump([ |
| 466 | {"insert": [{"id": "grouped", "group": True, "config": []}]}, |
| 467 | {"id": "grouped", "insert": [ |
| 468 | {"id": "in-group", "name": "@deepseek-ai/dsh-mcp-client", "config": { |
| 469 | "serverName": "inner", "transport": "stdio", "command": "node", |
| 470 | "args": ["server.mjs"], "cwd": "mcp"}}]}, |
| 471 | ]) |
| 472 | (bundle / "cordis.patch.yml").write_text(patch) |
| 473 | args = self.args(bundle=bundle, dialect="dsh") |
| 474 | result = self.cli(args) |
| 475 | self.assertEqual(result.returncode, 0, result.stderr) |
| 476 | self.assertEqual(self.servers(args.output)["inner"]["cwd"], "mcp/inner") |
| 477 | self.assertTrue((args.output / "mcp/inner/server.mjs").is_file()) |
| 478 | |
| 479 | def test_dsh_bundle_imports_custom_skill_dirs(self): |
| 480 | bundle = self.bundle() |
| 481 | skill = bundle / "pack-skills" / "guide" |
| 482 | skill.mkdir(parents=True) |
| 483 | (skill / "SKILL.md").write_text("---\nname: guide\ndescription: Bundled skill\n---\nBody.\n") |
| 484 | patch = yaml.safe_dump([ |
| 485 | {"insert": [ |
| 486 | {"id": "skills-row", "name": "@deepseek-ai/dsh-skill-filesystem", |
| 487 | "config": {"customSkillDirs": ["pack-skills"]}}, |
| 488 | {"id": "mcp", "name": "@deepseek-ai/dsh-mcp-client", "config": { |
| 489 | "serverName": "docs", "transport": "streamable-http", |
| 490 | "url": "https://docs.example.invalid/mcp"}}, |
| 491 | ]}, |
| 492 | ]) |
| 493 | (bundle / "cordis.patch.yml").write_text(patch) |
| 494 | args = self.args(bundle=bundle, dialect="dsh") |
| 495 | self.assertEqual(converter.convert(args), (1, 1, 0)) |
| 496 | self.assertTrue((args.output / "skills/guide/SKILL.md").is_file()) |
| 497 | |
| 498 | def test_dsh_bundle_never_executes_js_and_records_unlowerable_rows(self): |
| 499 | sentinel = self.root / "expression-ran" |
| 500 | bundle = self.bundle() |
| 501 | patch = ("- insert:\n - id: bad\n name: '@deepseek-ai/dsh-mcp-client'\n" |
| 502 | " config:\n serverName: bad\n transport: streamable-http\n" |
| 503 | " url: !!js require('node:fs').writeFileSync('" + str(sentinel) + "', 'ran')\n" |
| 504 | " - id: ok\n name: '@deepseek-ai/dsh-mcp-client'\n" |
| 505 | " config:\n serverName: ok\n transport: streamable-http\n" |
| 506 | " url: https://ok.example.invalid/mcp\n") |
| 507 | (bundle / "cordis.patch.yml").write_text(patch) |
| 508 | args = self.args(bundle=bundle, dialect="dsh") |
| 509 | result = self.cli(args) |
| 510 | self.assertEqual(result.returncode, 0, result.stderr) |
| 511 | self.assertFalse(sentinel.exists()) |
| 512 | self.assertEqual(sorted(self.servers(args.output)), ["ok"]) |
| 513 | self.assertIn("bad", (args.output / "CONVERSION.md").read_text()) |
| 514 | |
| 515 | def test_dsh_bundle_requires_manifest_patch_inside_package(self): |
| 516 | bundle = self.bundle() |
| 517 | self.refuse(self.args(bundle=bundle, dialect="dsh"), message="patch") |
| 518 | escaping = self.bundle(manifest={"dsh": {"bundle": {"patch": "../outside.yml"}}}) |
| 519 | self.refuse(self.args(bundle=escaping, dialect="dsh")) |
| 520 | plain = self.fresh("not-a-bundle") |
| 521 | plain.mkdir() |
| 522 | (plain / "package.json").write_text(json.dumps({"name": "plain"})) |
| 523 | self.refuse(self.args(bundle=plain, dialect="dsh"), message="dsh.bundle.patch") |
| 524 | self.refuse(self.args(bundle=self.bundle(), dialect="opencode-v1"), message="--format dsh") |
| 525 | combined = self.bundle() |
| 526 | (combined / "cordis.patch.yml").write_text("[]") |
| 527 | self.refuse(self.args(bundle=combined, config=self.config(self.dsh()), dialect="dsh"), |
| 528 | message="not both") |
| 529 | |
| 530 | def test_cli_literal_credentials_and_interpolation_never_echo_or_publish(self): |
| 531 | secret_file = self.write(CANARY, ".txt") |
| 532 | cases = [self.v1(headers={"Authorization": "Bearer " + CANARY}), |
| 533 | self.v1(headers={"Authorization": "{file:" + str(secret_file) + "}"}), |
| 534 | self.v1(url="https://user:" + CANARY + "@docs.example.invalid/mcp"), |
| 535 | self.v1(url="https://docs.example.invalid/mcp?token=" + CANARY), |
| 536 | self.v1(url="https://docs.example.invalid/mcp#" + CANARY)] |
| 537 | for document in cases: |
| 538 | with self.subTest(document=document): |
| 539 | args = self.args(config=self.config(document)) |
| 540 | result = self.cli(args) |
| 541 | self.assertEqual(result.returncode, 1) |
| 542 | self.assert_no_canary(result, args.output) |
| 543 | self.assertFalse(args.output.exists()) |
| 544 | self.assertEqual(secret_file.read_text(), CANARY) |
| 545 | |
| 546 | def test_opposite_dialect_enablement_is_not_silently_ignored(self): |
| 547 | cases = [("opencode-v1", self.v1(disabled=True)), |
| 548 | ("opencode-v2", {"mcp": {"servers": {"docs": self.remote(enabled=False)}}}), |
| 549 | ("opencode-v1", {"mcp": {"servers": {"docs": self.remote()}}}), |
| 550 | ("opencode-v2", self.v1())] |
| 551 | for dialect, document in cases: |
| 552 | with self.subTest(dialect=dialect, document=document): |
| 553 | self.refuse(self.args(config=self.config(document), dialect=dialect)) |
| 554 | |
| 555 | def test_duplicate_json_and_yaml_keys_are_rejected(self): |
| 556 | cases = [("opencode-v1", '{"mcp": {}, "mcp": {}}'), |
| 557 | ("opencode-v1", '{"mcp":{"docs":{"type":"remote","url":"https://a.invalid","url":"https://b.invalid","oauth":false}}}'), |
| 558 | ("dsh", '- id: one\n name: x\n name: y\n config: {}\n')] |
| 559 | for dialect, text in cases: |
| 560 | with self.subTest(dialect=dialect): |
| 561 | self.refuse(self.args(config=self.write(text), dialect=dialect)) |
| 562 | |
| 563 | def test_case_duplicate_and_reserved_headers_are_refused(self): |
| 564 | for headers in ({"Authorization": "{env:A}", "authorization": "{env:B}"}, |
| 565 | {"Accept": "{env:A}"}, {"content-TYPE": "{env:A}"}): |
| 566 | with self.subTest(headers=headers): |
| 567 | self.refuse(self.args(config=self.config(self.v1(headers=headers)))) |
| 568 | |
| 569 | def test_duplicate_server_and_skill_names_do_not_overwrite(self): |
| 570 | duplicate = self.dsh() + self.dsh() |
| 571 | duplicate[1]["id"] = "other-entry" |
| 572 | self.refuse(self.args(config=self.config(duplicate), dialect="dsh")) |
| 573 | self.refuse(self.args(skills=[self.skill(), self.skill()])) |
| 574 | |
| 575 | def test_unsupported_oauth_stdio_and_lifecycle_fields_are_refused(self): |
| 576 | cases = [("opencode-v1", self.v1(oauth={})), |
| 577 | ("opencode-v1", {"mcp": {"docs": {"type": "remote", "url": "https://docs.example.invalid/mcp"}}}), |
| 578 | ("opencode-v1", {"mcp": {"docs": {"type": "local", "command": ["python", "plugin.py"]}}}), |
| 579 | ("dsh", self.dsh(transport="stdio", command="python")), |
| 580 | ("dsh", self.dsh(reconnect={"enabled": False})), |
| 581 | ("dsh", self.dsh(failOnStartupError=True))] |
| 582 | for dialect, document in cases: |
| 583 | with self.subTest(dialect=dialect, document=document): |
| 584 | self.refuse(self.args(config=self.config(document), dialect=dialect)) |
| 585 | |
| 586 | def test_timeout_values_must_preserve_exact_native_units(self): |
| 587 | for value in (True, 0, 999, 1500, 3600001, 4000000, "7000"): |
| 588 | with self.subTest(value=value): |
| 589 | self.refuse(self.args(config=self.config(self.v1(timeout=value)))) |
| 590 | |
| 591 | def test_canonical_loopback_hosts_and_ambiguous_numeric_addresses(self): |
| 592 | for url, host in (("http://127.0.0.1:4312/mcp", "127.0.0.1"), |
| 593 | ("http://[::1]:4312/mcp", "[::1]")): |
| 594 | with self.subTest(url=url): |
| 595 | args = self.args(config=self.config(self.v1(url=url))) |
| 596 | converter.convert(args) |
| 597 | self.assertEqual(self.servers(args.output)["docs"]["url"], url) |
| 598 | manifest = json.loads((args.output / "plugin.json").read_text()) |
| 599 | self.assertEqual(manifest["extensions"]["net.codewhale"]["capabilities"]["network_hosts"], [host]) |
| 600 | for url in ("http://public.example.invalid/mcp", "https://127.1/mcp", "https://0x7f000001/mcp", |
| 601 | "https://127.000.000.001/mcp", "https://docs.example.invalid:0/mcp"): |
| 602 | with self.subTest(url=url): |
| 603 | self.refuse(self.args(config=self.config(self.v1(url=url)))) |
| 604 | |
| 605 | def test_source_and_ancestor_symlinks_are_refused(self): |
| 606 | source = self.skill() |
| 607 | link = self.root / "linked-skill" |
| 608 | link.symlink_to(source, target_is_directory=True) |
| 609 | self.refuse(self.args(skills=[link])) |
| 610 | parent_link = self.root / "linked-parent" |
| 611 | parent_link.symlink_to(self.root, target_is_directory=True) |
| 612 | self.refuse(self.args(skills=[parent_link / source.name])) |
| 613 | (source / "linked-companion").symlink_to(self.write("private data", ".txt")) |
| 614 | self.refuse(self.args(skills=[source])) |
| 615 | |
| 616 | def test_hardlinked_source_is_refused_without_altering_either_name(self): |
| 617 | source = self.skill(directory=False) |
| 618 | before = source.read_bytes() |
| 619 | other = self.root / "hardlinked.md" |
| 620 | os.link(source, other) |
| 621 | self.refuse(self.args(skills=[source])) |
| 622 | self.assertEqual(source.read_bytes(), before) |
| 623 | self.assertEqual(other.read_bytes(), before) |
| 624 | |
| 625 | def test_output_link_is_refused_and_its_target_is_untouched(self): |
| 626 | target = self.root / "existing-target" |
| 627 | target.mkdir() |
| 628 | marker = target / "keep" |
| 629 | marker.write_text("keep") |
| 630 | output = self.root / "output-link" |
| 631 | output.symlink_to(target, target_is_directory=True) |
| 632 | args = self.args(skills=[self.skill()], output=output) |
| 633 | with self.assertRaises(converter.ConversionError): |
| 634 | converter.convert(args) |
| 635 | self.assertTrue(output.is_symlink()) |
| 636 | self.assertEqual(list(target.iterdir()), [marker]) |
| 637 | self.assertEqual(marker.read_text(), "keep") |
| 638 | |
| 639 | def test_traversal_names_and_output_inside_source_are_refused(self): |
| 640 | skill = self.skill() |
| 641 | self.refuse(self.args(skills=[skill], name="../escaped")) |
| 642 | self.refuse(self.args(skills=[self.skill("../escaped")])) |
| 643 | self.refuse(self.args(skills=[skill], output=skill / "generated")) |
| 644 | self.assertFalse((self.root.parent / "escaped").exists()) |
| 645 | |
| 646 | def test_generated_metadata_collision_never_overwrites_source(self): |
| 647 | skill = self.skill(metadata={"metadata": {"license-owner": "fixture"}}) |
| 648 | companion = skill / "SOURCE_SKILL_METADATA.json" |
| 649 | companion.write_text("original") |
| 650 | self.refuse(self.args(skills=[skill]), message="collides") |
| 651 | self.assertEqual(companion.read_text(), "original") |
| 652 | |
| 653 | def test_existing_output_is_unchanged_by_cli_refusal(self): |
| 654 | output = self.fresh("existing-output") |
| 655 | output.mkdir() |
| 656 | (output / "plugin.json").write_text("original manifest") |
| 657 | (output / "other.txt").write_bytes(b"untouched") |
| 658 | result = self.cli(self.args(skills=[self.skill()], output=output)) |
| 659 | self.assertEqual(result.returncode, 1) |
| 660 | self.assertEqual({p.name: p.read_bytes() for p in output.iterdir()}, |
| 661 | {"plugin.json": b"original manifest", "other.txt": b"untouched"}) |
| 662 | |
| 663 | def test_aggregate_byte_budget_rejects_before_opening_excess_companion(self): |
| 664 | first, second = self.skill("first-guide"), self.skill("second-guide") |
| 665 | (first / "asset.bin").write_bytes(b"a" * 2300) |
| 666 | (second / "asset.bin").write_bytes(b"b" * 2300) |
| 667 | with mock.patch.object(converter, "MAX_BYTES", 4000): |
| 668 | # Either selection fits; their combination exceeds the same limit. |
| 669 | converter.convert(self.args(skills=[first])) |
| 670 | converter.convert(self.args(skills=[second])) |
| 671 | with mock.patch.object(converter.os, "open", wraps=os.open) as opened: |
| 672 | self.refuse(self.args(skills=[first, second])) |
| 673 | paths = [Path(call.args[0]) for call in opened.call_args_list] |
| 674 | self.assertIn(first / "asset.bin", paths) |
| 675 | self.assertNotIn(second / "asset.bin", paths) |
| 676 | |
| 677 | def test_file_budget_has_a_success_control_and_preserves_atomic_rejection(self): |
| 678 | skill = self.skill() |
| 679 | (skill / "one.txt").write_text("one") |
| 680 | (skill / "two.txt").write_text("two") |
| 681 | with mock.patch.object(converter, "MAX_FILES", 5): |
| 682 | converter.convert(self.args(skills=[skill])) |
| 683 | (skill / "three.txt").write_text("three") |
| 684 | self.refuse(self.args(skills=[skill]), message="budget") |
| 685 | |
| 686 | def test_oversize_document_is_rejected_before_open(self): |
| 687 | source = self.write(" " * (1024 * 1024 + 1)) |
| 688 | with mock.patch.object(converter.os, "open", wraps=os.open) as opened: |
| 689 | self.refuse(self.args(config=source), message="size limit") |
| 690 | opened.assert_not_called() |
| 691 | |
| 692 | def test_depth_alias_and_server_count_bounds(self): |
| 693 | for text, dialect in (("[" * 34 + "0" + "]" * 34, "dsh"), |
| 694 | ("- &entry {name: x}\n- *entry\n", "dsh"), |
| 695 | (json.dumps({"mcp": {f"server{i}": self.remote() for i in range(65)}}), "opencode-v1")): |
| 696 | with self.subTest(dialect=dialect, text=text[:40]): |
| 697 | self.refuse(self.args(config=self.write(text), dialect=dialect)) |
| 698 | |
| 699 | def test_write_failure_cleans_only_new_output_and_preserves_source(self): |
| 700 | skill = self.skill() |
| 701 | original = (skill / "SKILL.md").read_bytes() |
| 702 | unrelated = self.write("keep", ".txt") |
| 703 | args = self.args(skills=[skill]) |
| 704 | real_open = Path.open |
| 705 | |
| 706 | def failing_open(path, *positional, **keywords): |
| 707 | if path == args.output / "plugin.json": |
| 708 | raise OSError("simulated disk write failure") |
| 709 | return real_open(path, *positional, **keywords) |
| 710 | |
| 711 | with mock.patch.object(Path, "open", failing_open): |
| 712 | with self.assertRaises(OSError): |
| 713 | converter.convert(args) |
| 714 | self.assertFalse(args.output.exists()) |
| 715 | self.assertEqual((skill / "SKILL.md").read_bytes(), original) |
| 716 | self.assertEqual(unrelated.read_text(), "keep") |
| 717 | |
| 718 | def test_cli_jsonc_and_malformed_secret_input_are_safely_refused(self): |
| 719 | for text in ('// comment\n' + json.dumps(self.v1()), |
| 720 | '{"mcp": {},}', '{"mcp": "' + CANARY): |
| 721 | with self.subTest(text=text[:30]): |
| 722 | args = self.args(config=self.write(text, ".jsonc")) |
| 723 | result = self.cli(args) |
| 724 | self.assertEqual(result.returncode, 1) |
| 725 | self.assert_no_canary(result, args.output) |
| 726 | self.assertFalse(args.output.exists()) |
| 727 | |
| 728 | def test_local_secret_companion_is_refused_without_publication(self): |
| 729 | skill = self.skill() |
| 730 | (skill / ".env").write_text("TOKEN=" + CANARY) |
| 731 | args = self.args(skills=[skill]) |
| 732 | result = self.cli(args) |
| 733 | self.assertEqual(result.returncode, 1) |
| 734 | self.assert_no_canary(result, args.output) |
| 735 | self.assertFalse(args.output.exists()) |
| 736 | |
| 737 | |
| 738 | if __name__ == "__main__": |
| 739 | unittest.main() |
| 740 |