mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-12 23:12:23 -06:00
fix: validate URL scheme after MCP registry template substitution (#133)
* fix: validate URL scheme after MCP registry template substitution resolve_install_config() substitutes user-provided values into URL templates via string replacement without validating the resulting URL. Add urlparse check after substitution to reject non-HTTP(S) schemes, preventing SSRF-style redirection through crafted template variables. * fix: address review — reject empty hostname and embedded credentials Add hostname presence check and userinfo rejection after URL scheme validation. Prevents URLs like https:///path (no host) and https://user:pass@host (credential leakage in config). Two new tests.
This commit is contained in:
@@ -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"):
|
||||
|
||||
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user