fix(fetch): keep scheme lock under --allow-internal-ips; INVALID_PARAMS

Addresses Copilot review feedback on #4497:
- The scheme lock (http/https only) is now always enforced. Previously
  --allow-internal-ips skipped _validate_url_is_safe entirely, which
  also disabled the scheme check; the flag now only relaxes the
  private-IP check (check_private_ips=False) and never unlocks file://
  or other non-http(s) schemes. Added a regression test asserting
  file:// is refused (and the HTTP client is never called) under the
  bypass.
- _resolve_host_ips now raises INVALID_PARAMS (not INTERNAL_ERROR) on
  DNS failure / no parseable IPs, since these are caller-supplied
  input errors, matching the other URL-validation errors.
- Redirect loop enforces the cap before following the next hop, so it
  no longer processes MAX_REDIRECTS+1 redirect responses.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: olaservo

src/fetch/src/mcp_server_fetch/server.py

@@ -133,7 +133,7 @@ async def _resolve_host_ips(
         )
     except socket.gaierror as e:
         raise McpError(ErrorData(
-            code=INTERNAL_ERROR,
+            code=INVALID_PARAMS,
             message=f"Failed to resolve host {host}: {e}",
         ))
 
@@ -151,18 +151,19 @@ async def _resolve_host_ips(
     # classify, refuse rather than fall through to an empty (allow-all) check.
     if not ips:
         raise McpError(ErrorData(
-            code=INTERNAL_ERROR,
+            code=INVALID_PARAMS,
             message=f"Failed to resolve host {host} to a usable IP address.",
         ))
     return ips
 
 
-async def _validate_url_is_safe(url: str) -> None:
+async def _validate_url_is_safe(url: str, *, check_private_ips: bool = True) -> None:
     """Guard a URL against SSRF before it is fetched.
 
     Rejects non-http(s) schemes and any URL whose host resolves to a
-    non-public IP address. Servers started with --allow-internal-ips skip
-    this check entirely.
+    non-public IP address. The scheme (and host-presence) check is always
+    enforced; --allow-internal-ips only relaxes the private-IP check
+    (check_private_ips=False), so it never unlocks non-http(s) schemes.
     """
     parsed = urlparse(url)
     if parsed.scheme not in ("http", "https"):
@@ -178,6 +179,9 @@ async def _validate_url_is_safe(url: str) -> None:
             message=f"Cannot fetch {url}: URL has no host.",
         ))
 
+    if not check_private_ips:
+        return
+
     for ip in await _resolve_host_ips(host, parsed.port, parsed.scheme):
         if _is_blocked_ip(ip):
             raise McpError(ErrorData(
@@ -208,27 +212,32 @@ async def _get_following_redirects(
     from httpx import URL
 
     current_url = url
-    for _ in range(MAX_REDIRECTS + 1):
-        if not allow_internal_ips:
-            await _validate_url_is_safe(current_url)
+    redirects = 0
+    while True:
+        # The scheme lock is always enforced; --allow-internal-ips only relaxes
+        # the private-IP check, so it can never unlock file:// et al.
+        await _validate_url_is_safe(
+            current_url, check_private_ips=not allow_internal_ips
+        )
         response = await client.get(
             current_url,
             follow_redirects=False,
             headers=headers,
             timeout=timeout,
         )
-        if response.status_code in REDIRECT_STATUS_CODES:
-            location = response.headers.get("location")
-            if not location:
-                return response
-            current_url = str(URL(current_url).join(location))
-            continue
-        return response
-
-    raise McpError(ErrorData(
-        code=INTERNAL_ERROR,
-        message=f"Cannot fetch {url}: exceeded the maximum of {MAX_REDIRECTS} redirects.",
-    ))
+        if response.status_code not in REDIRECT_STATUS_CODES:
+            return response
+        location = response.headers.get("location")
+        if not location:
+            return response
+        # Enforce the cap before following (and re-validating) the next hop.
+        if redirects >= MAX_REDIRECTS:
+            raise McpError(ErrorData(
+                code=INTERNAL_ERROR,
+                message=f"Cannot fetch {url}: exceeded the maximum of {MAX_REDIRECTS} redirects.",
+            ))
+        redirects += 1
+        current_url = str(URL(current_url).join(location))
 
 
 async def check_may_autonomously_fetch_url(url: str, user_agent: str, proxy_url: str | None = None, allow_internal_ips: bool = False) -> None:

src/fetch/tests/test_server.py

@@ -478,3 +478,22 @@ async def test_fails_closed_when_no_resolved_ip_parses(self):
         with patch.object(loop, "getaddrinfo", new=AsyncMock(return_value=fake)):
             with pytest.raises(McpError):
                 await _validate_url_is_safe("http://weird.example/")
+
+    @pytest.mark.asyncio
+    async def test_allow_internal_ips_still_blocks_non_http_scheme(self):
+        """--allow-internal-ips relaxes only the private-IP check; the scheme
+        lock stays on, so file:// is still refused and no request is made."""
+        with patch("httpx.AsyncClient") as mock_client_class:
+            mock_client = AsyncMock()
+            mock_client.get = AsyncMock()
+            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):
+                await fetch_url(
+                    "file:///etc/passwd",
+                    DEFAULT_USER_AGENT_AUTONOMOUS,
+                    allow_internal_ips=True,
+                )
+
+            mock_client.get.assert_not_called()