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