Merge pull request #5392 from modelcontextprotocol/v2/feat/2317-fetch-allowed-hosts

feat(fetch): add an operator host allowlist, --allowed-hosts

Author: cliffhall

src/fetch/README.md

@@ -191,6 +191,38 @@ The client receives an error naming the refused address.
 To fetch from local or internal hosts (a development server, an intranet page), add the argument
 `--allow-private-ips` to the `args` list in the configuration.
 
+### Customization - Allowed hosts
+
+By default the server may fetch from any host (subject to the private-address guard above). To restrict it to the
+hosts you name, add `--allowed-hosts` followed by one or more entries to the `args` list (the argument can also be
+repeated). Each entry is one of:
+
+- an exact host name, such as `docs.python.org`, which matches that host only, not its subdomains;
+- `*.` followed by a domain, such as `*.github.com`, which matches every subdomain of it at any depth
+  (`api.github.com`, `a.b.github.com`) but not `github.com` itself: list both to allow both;
+- an IP address, such as `203.0.113.7` or `2001:db8::7`, which matches a URL that uses that address literally.
+
+Matching ignores case and a trailing dot, and an internationalized name may be given in either its Unicode or its
+`xn--` form. An entry names a host only, so it carries no scheme, port or path. The list is checked before every request
+the server sends, by the host name in the URL rather than by the addresses it resolves to: the requested page, its
+robots.txt check, and every redirect hop of either, from the tool and from the prompt alike. A request to any other host
+is refused before it is sent, and the client receives an error naming the refused host.
+
+The allowlist and the private-address guard are separate checks, and a request must pass both. Listing an internal
+host, such as `localhost` or `10.0.0.5`, does not by itself let the server reach it: that also needs
+`--allow-private-ips`, which in turn does not lift the allowlist.
+
+```json
+{
+  "mcpServers": {
+    "fetch": {
+      "command": "uvx",
+      "args": ["mcp-server-fetch", "--allowed-hosts", "docs.python.org", "*.wikipedia.org"]
+    }
+  }
+}
+```
+
 ### Customization - User-agent
 
 By default, depending on if the request came from the model (via a tool), or was user initiated (via a prompt), the

src/fetch/src/mcp_server_fetch/__init__.py

@@ -1,3 +1,4 @@
+from .allowlist import parse_allowed_host
 from .server import serve
 
 
@@ -21,6 +22,15 @@ def main():
         action="store_true",
         help="Allow fetching private, loopback and link-local addresses (refused by default)",
     )
+    parser.add_argument(
+        "--allowed-hosts",
+        nargs="+",
+        action="extend",
+        type=parse_allowed_host,
+        metavar="HOST",
+        help="Only fetch from these hosts: an exact host name or IP address, or "
+        "*.example.com for any subdomain of example.com (all hosts by default)",
+    )
     parser.add_argument(
         "--no-readability-js",
         action="store_true",
@@ -35,6 +45,7 @@ def main():
             args.ignore_robots_txt,
             args.proxy_url,
             args.allow_private_ips,
+            args.allowed_hosts,
             not args.no_readability_js,
         )
     )

src/fetch/src/mcp_server_fetch/allowlist.py

@@ -0,0 +1,101 @@
+# The fetch server's operator-configured host allowlist (#2317). With
+# `--allowed-hosts`, every request the server sends (the tool's page, the
+# prompt's page, each robots.txt check and every redirect hop of any of them)
+# must name a listed host, or it is refused before it is sent. Without the flag
+# nothing is restricted, so existing configurations keep working.
+#
+# The allowlist names hosts, and it is checked against the host the request
+# connects to (`raw_host`, the ASCII form), never against resolved addresses:
+# a name the operator chose is trusted by name. It is a separate check from
+# the private-address guard in server.py, and both apply: listing `localhost`
+# or an intranet name does not, by itself, let the server reach a private
+# address; that still needs `--allow-private-ips`. Two entry forms only, to
+# keep the semantics obvious: an exact host (a name or an IP literal), and
+# `*.example.com`, which matches any subdomain at any depth of example.com but
+# not example.com itself. Ports and schemes are not part of an entry.
+
+import argparse
+import ipaddress
+from collections.abc import Awaitable, Callable, Iterable
+
+import httpx
+from mcp.server.mcpserver.exceptions import ToolError
+
+
+def normalize_allowed_host(entry: str) -> str:
+    """The canonical form of one ``--allowed-hosts`` entry, or ``ValueError``.
+
+    A name is lowercased, its trailing dot dropped, and IDNA-encoded to the
+    ASCII form requests are checked in (``bücher.example`` becomes
+    ``xn--bcher-kva.example``). An IP literal is written in its compressed
+    form, an IPv6 one with or without brackets. A wildcard is ``*.`` followed
+    by a name.
+    """
+    value = entry.strip().rstrip(".").lower()
+    wildcard = value.startswith("*.")
+    name = value[2:] if wildcard else value
+    if not wildcard:
+        try:
+            return ipaddress.ip_address(name.strip("[]")).compressed
+        except ValueError:
+            pass
+    if not name or any(c in name for c in "*/\\:@[]?#% ") or ".." in name:
+        raise ValueError(
+            f"{entry!r} is not a host name, an IP address or *.<domain> "
+            "(no scheme, port, path or other wildcard)"
+        )
+    try:
+        ascii_name = httpx.URL(f"http://{name}/").raw_host.decode("ascii")
+    except httpx.InvalidURL as e:
+        raise ValueError(f"{entry!r} is not a valid host name: {e}") from None
+    return f"*.{ascii_name}" if wildcard else ascii_name
+
+
+def parse_allowed_host(entry: str) -> str:
+    """``normalize_allowed_host`` as an argparse ``type``, with its message."""
+    try:
+        return normalize_allowed_host(entry)
+    except ValueError as e:
+        raise argparse.ArgumentTypeError(str(e)) from None
+
+
+def normalize_allowed_hosts(entries: Iterable[str] | None) -> tuple[str, ...]:
+    """Every entry normalized; an empty tuple means no allowlist."""
+    return tuple(normalize_allowed_host(entry) for entry in entries or ())
+
+
+def host_is_allowed(host: str, allowed_hosts: tuple[str, ...]) -> bool:
+    """Whether ``host`` (a request's ``raw_host``) matches a normalized entry."""
+    host = host.rstrip(".").lower()
+    try:
+        host = ipaddress.ip_address(host).compressed
+    except ValueError:
+        pass
+    for entry in allowed_hosts:
+        if entry.startswith("*."):
+            if host.endswith(entry[1:]):
+                return True
+        elif host == entry:
+            return True
+    return False
+
+
+def refuse_unlisted_destination(
+    allowed_hosts: tuple[str, ...],
+) -> Callable[[httpx.Request], Awaitable[None]]:
+    """An httpx request hook refusing any request to a host not on the list.
+
+    httpx runs request hooks before every request it sends, redirect hops
+    included, so a listed host cannot redirect the fetch to an unlisted one.
+    The error does not repeat the list, which is the operator's configuration.
+    """
+
+    async def hook(request: httpx.Request) -> None:
+        host = request.url.raw_host.decode("ascii")
+        if not host_is_allowed(host, allowed_hosts):
+            raise ToolError(
+                f"Refused to fetch {request.url}: {host} is not one of the "
+                "hosts this server is allowed to fetch from (--allowed-hosts)."
+            )
+
+    return hook

src/fetch/src/mcp_server_fetch/server.py

@@ -23,7 +23,7 @@
 import ipaddress
 import os
 import socket
-from collections.abc import Awaitable, Callable, Mapping
+from collections.abc import Awaitable, Callable, Mapping, Sequence
 from typing import Annotated, Any, Tuple
 from importlib.metadata import version
 from urllib.parse import urlparse, urlunparse
@@ -33,6 +33,7 @@
 import readabilipy.simple_json
 from .readability_js import extract_article
 from bs4 import BeautifulSoup, Tag
+from .allowlist import normalize_allowed_hosts, refuse_unlisted_destination
 from mcp.server import MCPServer, ServerRequestContext
 from mcp.server.context import CallNext, HandlerResult, ServerMiddleware
 from mcp.server.mcpserver.exceptions import ToolError
@@ -479,6 +480,7 @@ async def check_may_autonomously_fetch_url(
     user_agent: str,
     proxy_url: str | None = None,
     allow_private_ips: bool = False,
+    allowed_hosts: tuple[str, ...] = (),
 ) -> None:
     """
     Check if the URL can be fetched by the user agent according to the robots.txt file.
@@ -489,8 +491,12 @@ async def check_may_autonomously_fetch_url(
     robot_txt_url = get_robots_txt_url(url)
 
     async with make_client(proxy_url) as client:
+        hooks: list[Callable[[httpx.Request], Awaitable[None]]] = []
+        if allowed_hosts:
+            hooks.append(refuse_unlisted_destination(allowed_hosts))
         if not allow_private_ips:
-            client.event_hooks = {"request": [_refuse_private_destination]}
+            hooks.append(_refuse_private_destination)
+        client.event_hooks = {"request": hooks}
         try:
             response = await client.get(
                 robot_txt_url,
@@ -543,6 +549,7 @@ async def fetch_url(
     force_raw: bool = False,
     proxy_url: str | None = None,
     allow_private_ips: bool = False,
+    allowed_hosts: tuple[str, ...] = (),
     check_robots_txt: bool = False,
     use_readability_js: bool = True,
 ) -> Tuple[str, str]:
@@ -564,11 +571,13 @@ async def _check_redirect_robots_txt(request: httpx.Request) -> None:
             seen_first_request = True
             return
         await check_may_autonomously_fetch_url(
-            str(request.url), user_agent, proxy_url, allow_private_ips
+            str(request.url), user_agent, proxy_url, allow_private_ips, allowed_hosts
         )
 
     async with make_client(proxy_url) as client:
         hooks: list[Callable[[httpx.Request], Awaitable[None]]] = []
+        if allowed_hosts:
+            hooks.append(refuse_unlisted_destination(allowed_hosts))
         if not allow_private_ips:
             hooks.append(_refuse_private_destination)
         if check_robots_txt:
@@ -660,6 +669,7 @@ def create_server(
     ignore_robots_txt: bool = False,
     proxy_url: str | None = None,
     allow_private_ips: bool = False,
+    allowed_hosts: Sequence[str] | None = None,
     use_readability_js: bool = True,
 ) -> MCPServer:
     """Build the fetch server: the `fetch` tool, the `fetch` prompt, and the guard.
@@ -669,6 +679,7 @@ def create_server(
     if proxy_url:
         proxy_url = normalize_proxy_url(proxy_url)
     normalize_proxy_env()
+    allowed = normalize_allowed_hosts(allowed_hosts)
     user_agent_autonomous = custom_user_agent or DEFAULT_USER_AGENT_AUTONOMOUS
     user_agent_manual = custom_user_agent or DEFAULT_USER_AGENT_MANUAL
 
@@ -729,7 +740,7 @@ async def fetch(
 
         if not ignore_robots_txt:
             await check_may_autonomously_fetch_url(
-                page_url, user_agent_autonomous, proxy_url, allow_private_ips
+                page_url, user_agent_autonomous, proxy_url, allow_private_ips, allowed
             )
 
         content, prefix = await fetch_url(
@@ -738,6 +749,7 @@ async def fetch(
             force_raw=raw,
             proxy_url=proxy_url,
             allow_private_ips=allow_private_ips,
+            allowed_hosts=allowed,
             check_robots_txt=not ignore_robots_txt,
             use_readability_js=use_readability_js,
         )
@@ -771,6 +783,7 @@ async def fetch_prompt(
                 user_agent_manual,
                 proxy_url=proxy_url,
                 allow_private_ips=allow_private_ips,
+                allowed_hosts=allowed,
                 use_readability_js=use_readability_js,
             )
         except ToolError as e:
@@ -785,6 +798,7 @@ async def serve(
     ignore_robots_txt: bool = False,
     proxy_url: str | None = None,
     allow_private_ips: bool = False,
+    allowed_hosts: Sequence[str] | None = None,
     use_readability_js: bool = True,
 ) -> None:
     """Run the fetch MCP server over stdio.
@@ -793,6 +807,10 @@ async def serve(
         custom_user_agent: Optional custom User-Agent string to use for requests
         ignore_robots_txt: Whether to ignore robots.txt restrictions
         proxy_url: Optional proxy URL to use for requests
+        allowed_hosts: When given, the only hosts the server may fetch from:
+            exact hosts, or ``*.example.com`` for any subdomain of it, checked
+            in addition to the private-address guard. Empty or None allows
+            every host
         allow_private_ips: Allow fetching private, loopback and link-local
             addresses, which are refused by default
         use_readability_js: Extract HTML with Readability.js under Node.js
@@ -804,6 +822,7 @@ async def serve(
         ignore_robots_txt,
         proxy_url,
         allow_private_ips,
+        allowed_hosts,
         use_readability_js,
     )
     # Both eras (#4853): the SDK's dual-era loop lets the client's first