diff --git a/tests/test_mcp_registry.py b/tests/test_mcp_registry.py index 610713e8..609d3cee 100644 --- a/tests/test_mcp_registry.py +++ b/tests/test_mcp_registry.py @@ -398,6 +398,72 @@ class TestResolveInstallConfig: config = resolve_install_config(server, "remote", 0) assert config["url"] == "https://us-east.example.com/mcp" + def test_remote_variable_substitution_invalid_scheme(self) -> None: + server = RegistryServer( + name="io.example/test", + version="1.0.0", + remotes=[ + RegistryRemote( + type="streamable-http", + url="{scheme}://evil.example.com/mcp", + variables={ + "scheme": RegistryRemoteVariable(is_required=True), + }, + ) + ], + ) + with pytest.raises(MCPRegistryError, match="Invalid URL scheme"): + resolve_install_config(server, "remote", 0, variables={"scheme": "file"}) + + def test_remote_variable_substitution_preserves_valid_scheme(self) -> None: + server = RegistryServer( + name="io.example/test", + version="1.0.0", + remotes=[ + RegistryRemote( + type="streamable-http", + url="https://{host}.example.com/mcp", + variables={ + "host": RegistryRemoteVariable(is_required=True), + }, + ) + ], + ) + config = resolve_install_config(server, "remote", 0, variables={"host": "api"}) + assert config["url"] == "https://api.example.com/mcp" + + def test_remote_variable_substitution_missing_hostname(self) -> None: + """URL like https:///mcp has valid scheme but no hostname.""" + server = RegistryServer( + name="io.example/test", + version="1.0.0", + remotes=[ + RegistryRemote( + type="streamable-http", + url="https:///mcp", + ) + ], + ) + with pytest.raises(MCPRegistryError, match="hostname is missing"): + resolve_install_config(server, "remote", 0) + + def test_remote_variable_substitution_embedded_credentials(self) -> None: + server = RegistryServer( + name="io.example/test", + version="1.0.0", + remotes=[ + RegistryRemote( + type="streamable-http", + url="https://{creds}@example.com/mcp", + variables={ + "creds": RegistryRemoteVariable(is_required=True), + }, + ) + ], + ) + with pytest.raises(MCPRegistryError, match="embedded credentials"): + resolve_install_config(server, "remote", 0, variables={"creds": "user:pass"}) + def test_remote_no_remotes(self) -> None: server = RegistryServer(name="io.example/test", version="1.0.0") with pytest.raises(MCPRegistryError, match="no remote"): diff --git a/turnstone/core/mcp_registry.py b/turnstone/core/mcp_registry.py index d10d7a55..ada38bc6 100644 --- a/turnstone/core/mcp_registry.py +++ b/turnstone/core/mcp_registry.py @@ -11,6 +11,7 @@ from __future__ import annotations import re from dataclasses import dataclass, field from typing import Any +from urllib.parse import urlparse import httpx @@ -400,6 +401,17 @@ def resolve_install_config( raise MCPRegistryError(f"Required URL variable '{var_name}' not provided") url = url.replace(placeholder, value) + # Validate URL after substitution to prevent SSRF-style redirection + parsed = urlparse(url) + if parsed.scheme not in ("http", "https"): + raise MCPRegistryError( + f"Invalid URL scheme '{parsed.scheme}' after variable substitution" + ) + if not parsed.hostname: + raise MCPRegistryError("Invalid URL (hostname is missing) after variable substitution") + if parsed.username is not None or parsed.password is not None: + raise MCPRegistryError("URLs with embedded credentials are not allowed in MCP remotes") + # Build headers dict (required keys only — values provided by user at install time) headers: dict[str, str] = {} for h in remote.headers: