fix(fetch): add configurable SSL certificate verification

src/fetch/README.md

@@ -174,6 +174,29 @@ This can be customized by adding the argument `--user-agent=YourUserAgent` to th
 
 The server can be configured to use a proxy by using the `--proxy-url` argument.
 
+### Customization - SSL Verification
+
+By default, SSL certificate verification is enabled. If you need to fetch content from internal servers that use self-signed certificates, you can disable verification by setting the `MCP_FETCH_SSL_VERIFY` environment variable to `false`:
+
+```json
+{
+  "mcpServers": {
+    "fetch": {
+      "command": "uvx",
+      "args": ["mcp-server-fetch"],
+      "env": {
+        "MCP_FETCH_SSL_VERIFY": "false"
+      }
+    }
+  }
+}
+```
+
+Parsing is fail-secure: any value other than `false` keeps SSL verification enabled.
+
+> [!WARNING]
+> Disabling SSL verification reduces security. Only use this in trusted environments with internal servers that have self-signed certificates.
+
 ## Windows Configuration
 
 If you're experiencing timeout issues on Windows, you may need to set the `PYTHONIOENCODING` environment variable to ensure proper character encoding:

src/fetch/src/mcp_server_fetch/server.py

@@ -1,3 +1,5 @@
+import os
+import ssl
 from typing import Annotated, Tuple
 from urllib.parse import urlparse, urlunparse
 
@@ -23,6 +25,10 @@
 DEFAULT_USER_AGENT_AUTONOMOUS = "ModelContextProtocol/1.0 (Autonomous; +https://github.com/modelcontextprotocol/servers)"
 DEFAULT_USER_AGENT_MANUAL = "ModelContextProtocol/1.0 (User-Specified; +https://github.com/modelcontextprotocol/servers)"
 
+# Set MCP_FETCH_SSL_VERIFY=false to disable SSL verification for internal/self-signed certificates.
+# Fail-secure: any value other than "false" keeps SSL verification enabled.
+SSL_VERIFY = os.getenv("MCP_FETCH_SSL_VERIFY", "true").strip().lower() != "false"
+
 
 def extract_content_from_html(html: str) -> str:
     """Extract and convert HTML content to Markdown format.
@@ -68,17 +74,40 @@ async def check_may_autonomously_fetch_url(url: str, user_agent: str, proxy_url:
     Check if the URL can be fetched by the user agent according to the robots.txt file.
     Raises a McpError if not.
     """
-    from httpx import AsyncClient, HTTPError
+    from httpx import AsyncClient, ConnectError, HTTPError
 
     robot_txt_url = get_robots_txt_url(url)
 
-    async with AsyncClient(proxy=proxy_url) as client:
+    async with AsyncClient(proxy=proxy_url, verify=SSL_VERIFY) as client:
         try:
             response = await client.get(
                 robot_txt_url,
                 follow_redirects=True,
                 headers={"User-Agent": user_agent},
             )
+        except ssl.SSLError as e:
+            raise McpError(ErrorData(
+                code=INTERNAL_ERROR,
+                message=f"SSL certificate verification failed for {robot_txt_url}. "
+                f"If this is an internal server with a self-signed certificate, "
+                f"set MCP_FETCH_SSL_VERIFY=false in your environment. "
+                f"Error details: {str(e)}",
+            ))
+        except ConnectError as e:
+            # httpx wraps SSL errors in ConnectError in some cases
+            error_str = str(e).lower()
+            if "ssl" in error_str or "certificate" in error_str or "verify" in error_str:
+                raise McpError(ErrorData(
+                    code=INTERNAL_ERROR,
+                    message=f"SSL certificate verification failed for {robot_txt_url}. "
+                    f"If this is an internal server with a self-signed certificate, "
+                    f"set MCP_FETCH_SSL_VERIFY=false in your environment. "
+                    f"Error details: {str(e)}",
+                ))
+            raise McpError(ErrorData(
+                code=INTERNAL_ERROR,
+                message=f"Failed to fetch robots.txt {robot_txt_url} due to a connection issue",
+            ))
         except HTTPError:
             raise McpError(ErrorData(
                 code=INTERNAL_ERROR,
@@ -114,16 +143,36 @@ async def fetch_url(
     """
     Fetch the URL and return the content in a form ready for the LLM, as well as a prefix string with status information.
     """
-    from httpx import AsyncClient, HTTPError
+    from httpx import AsyncClient, ConnectError, HTTPError
 
-    async with AsyncClient(proxy=proxy_url) as client:
+    async with AsyncClient(proxy=proxy_url, verify=SSL_VERIFY) as client:
         try:
             response = await client.get(
                 url,
                 follow_redirects=True,
                 headers={"User-Agent": user_agent},
                 timeout=30,
             )
+        except ssl.SSLError as e:
+            raise McpError(ErrorData(
+                code=INTERNAL_ERROR,
+                message=f"SSL certificate verification failed for {url}. "
+                f"If this is an internal server with a self-signed certificate, "
+                f"set MCP_FETCH_SSL_VERIFY=false in your environment. "
+                f"Error details: {str(e)}",
+            ))
+        except ConnectError as e:
+            # httpx wraps SSL errors in ConnectError in some cases
+            error_str = str(e).lower()
+            if "ssl" in error_str or "certificate" in error_str or "verify" in error_str:
+                raise McpError(ErrorData(
+                    code=INTERNAL_ERROR,
+                    message=f"SSL certificate verification failed for {url}. "
+                    f"If this is an internal server with a self-signed certificate, "
+                    f"set MCP_FETCH_SSL_VERIFY=false in your environment. "
+                    f"Error details: {str(e)}",
+                ))
+            raise McpError(ErrorData(code=INTERNAL_ERROR, message=f"Failed to fetch {url}: {e!r}"))
         except HTTPError as e:
             raise McpError(ErrorData(code=INTERNAL_ERROR, message=f"Failed to fetch {url}: {e!r}"))
         if response.status_code >= 400:

src/fetch/tests/test_server.py

@@ -323,4 +323,94 @@ async def test_fetch_with_proxy(self):
             )
 
             # Verify AsyncClient was called with proxy
-            mock_client_class.assert_called_once_with(proxy="http://proxy.example.com:8080")
+            mock_client_class.assert_called_once_with(proxy="http://proxy.example.com:8080", verify=True)
+
+
+class TestSSLVerifyConfiguration:
+    """Tests for the MCP_FETCH_SSL_VERIFY environment variable."""
+
+    def _reload_ssl_verify(self, value):
+        """Reload the server module with MCP_FETCH_SSL_VERIFY set to value."""
+        import importlib
+        import os
+        import mcp_server_fetch.server as server_module
+
+        env = {"MCP_FETCH_SSL_VERIFY": value} if value is not None else {}
+        with patch.dict(os.environ, env, clear=False):
+            if value is None:
+                os.environ.pop("MCP_FETCH_SSL_VERIFY", None)
+            importlib.reload(server_module)
+            result = server_module.SSL_VERIFY
+        importlib.reload(server_module)
+        return result
+
+    def test_ssl_verify_enabled_by_default(self):
+        """SSL verification is enabled when the env var is not set."""
+        assert self._reload_ssl_verify(None) is True
+
+    def test_ssl_verify_disabled_by_false(self):
+        """Setting MCP_FETCH_SSL_VERIFY=false disables verification."""
+        assert self._reload_ssl_verify("false") is False
+
+    def test_ssl_verify_disabled_case_insensitive(self):
+        """Value parsing is case-insensitive."""
+        assert self._reload_ssl_verify("FALSE") is False
+
+    def test_ssl_verify_stays_enabled_for_other_values(self):
+        """Fail-secure: any value other than 'false' keeps verification enabled."""
+        for value in ("true", "0", "no", "off", "", "garbage"):
+            assert self._reload_ssl_verify(value) is True
+
+
+class TestSSLErrorHandling:
+    """Tests for SSL error handling in fetch_url."""
+
+    @pytest.mark.asyncio
+    async def test_ssl_error_message_mentions_env_var(self):
+        """SSL errors produce a helpful message pointing to MCP_FETCH_SSL_VERIFY."""
+        import ssl
+
+        with patch("httpx.AsyncClient") as mock_client_class:
+            mock_client = AsyncMock()
+            mock_client.get = AsyncMock(side_effect=ssl.SSLError("certificate verify failed"))
+            mock_client_class.return_value.__aenter__ = AsyncMock(return_value=mock_client)
+            mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None)
+
+            with pytest.raises(McpError) as exc_info:
+                await fetch_url("https://example.com", DEFAULT_USER_AGENT_AUTONOMOUS)
+
+            assert "MCP_FETCH_SSL_VERIFY" in str(exc_info.value)
+
+    @pytest.mark.asyncio
+    async def test_connect_error_wrapping_ssl_mentions_env_var(self):
+        """SSL errors wrapped in ConnectError are detected and explained."""
+        from httpx import ConnectError
+
+        with patch("httpx.AsyncClient") as mock_client_class:
+            mock_client = AsyncMock()
+            mock_client.get = AsyncMock(
+                side_effect=ConnectError("certificate verify failed: self-signed certificate")
+            )
+            mock_client_class.return_value.__aenter__ = AsyncMock(return_value=mock_client)
+            mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None)
+
+            with pytest.raises(McpError) as exc_info:
+                await fetch_url("https://example.com", DEFAULT_USER_AGENT_AUTONOMOUS)
+
+            assert "MCP_FETCH_SSL_VERIFY" in str(exc_info.value)
+
+    @pytest.mark.asyncio
+    async def test_non_ssl_connect_error_not_misreported(self):
+        """Non-SSL connection errors are not reported as SSL failures."""
+        from httpx import ConnectError
+
+        with patch("httpx.AsyncClient") as mock_client_class:
+            mock_client = AsyncMock()
+            mock_client.get = AsyncMock(side_effect=ConnectError("connection refused"))
+            mock_client_class.return_value.__aenter__ = AsyncMock(return_value=mock_client)
+            mock_client_class.return_value.__aexit__ = AsyncMock(return_value=None)
+
+            with pytest.raises(McpError) as exc_info:
+                await fetch_url("https://example.com", DEFAULT_USER_AGENT_AUTONOMOUS)
+
+            assert "MCP_FETCH_SSL_VERIFY" not in str(exc_info.value)