From 7464deb19af99f9ff8726b746d5cbe514e51ca13 Mon Sep 17 00:00:00 2001 From: gyixuan Date: Mon, 7 Sep 2026 02:00:37 +0800 Subject: [PATCH] feat(fetch): add opt-in host allowlisting via --allowed-hosts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a host allowlist for the fetch server, addressing the SSRF surface tracked in #2317 without changing default behavior: - New --allowed-hosts flag (exact hosts, case-insensitive; *.example.com wildcards covering the bare domain and any subdomain; IP literals). - Enforced on the initial request, the robots.txt pre-check, and every redirect hop: redirects are now followed manually (bounded at 20, same as httpx's default) and each hop is re-validated, so a redirect from an allowed host can no longer bounce a fetch to a disallowed host. - URL validation parses with httpx.URL — the same parser used to connect — so the validated host is always the host being connected. - Denials fail closed with an error that does not echo the allowlist. Default-IP-blocking is intentionally left to the separate default-deny proposal; with no flag set, behavior is identical to before. --- src/fetch/README.md | 24 ++ src/fetch/src/mcp_server_fetch/__init__.py | 9 +- src/fetch/src/mcp_server_fetch/server.py | 158 ++++++++- src/fetch/tests/test_server.py | 388 +++++++++++++++++++++ 4 files changed, 565 insertions(+), 14 deletions(-) diff --git a/src/fetch/README.md b/src/fetch/README.md index 7e2869983f..03974aec08 100644 --- a/src/fetch/README.md +++ b/src/fetch/README.md @@ -174,6 +174,30 @@ 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 - Allowed hosts + +By default the server can fetch any host (see the security caution above). To restrict which hosts the server may connect to, add the `--allowed-hosts` argument with one or more entries to the `args` list in the configuration: + +- `example.com` allows exactly `example.com` (matching is case-insensitive) +- `*.example.com` allows `example.com` itself and any subdomain, e.g. `api.example.com` + +Matching is by hostname only: an entry allows that host on any port, and IPv6 addresses are listed without brackets (e.g. `--allowed-hosts ::1`). + +The allowlist is enforced on the initial request, on the `robots.txt` pre-check, and on every redirect hop, so a redirect from an allowed host cannot bounce the fetch to a disallowed host. Requests to any other host fail with an error explaining that the host is not allowlisted. + +Note that the allowlist matches hostnames, not the IP addresses they resolve to: an allowlisted domain whose DNS records point at internal addresses can still be fetched (this is what makes it possible to deliberately allowlist internal hosts). + +```json +{ + "mcpServers": { + "fetch": { + "command": "uvx", + "args": ["mcp-server-fetch", "--allowed-hosts", "example.com", "*.github.com"] + } + } +} +``` + ## Windows Configuration If you're experiencing timeout issues on Windows, you may need to set the `PYTHONIOENCODING` environment variable to ensure proper character encoding: diff --git a/src/fetch/src/mcp_server_fetch/__init__.py b/src/fetch/src/mcp_server_fetch/__init__.py index 09744ce319..f4b4b20ffd 100644 --- a/src/fetch/src/mcp_server_fetch/__init__.py +++ b/src/fetch/src/mcp_server_fetch/__init__.py @@ -16,9 +16,16 @@ def main(): help="Ignore robots.txt restrictions", ) parser.add_argument("--proxy-url", type=str, help="Proxy URL to use for requests") + parser.add_argument( + "--allowed-hosts", + type=str, + nargs="+", + metavar="HOST", + help="Only allow fetching these hosts (exact names like example.com or wildcards like *.example.com, which also covers example.com itself). Applies to the initial URL and every redirect hop. If omitted, all hosts are allowed.", + ) args = parser.parse_args() - asyncio.run(serve(args.user_agent, args.ignore_robots_txt, args.proxy_url)) + asyncio.run(serve(args.user_agent, args.ignore_robots_txt, args.proxy_url, allowed_hosts=args.allowed_hosts)) if __name__ == "__main__": diff --git a/src/fetch/src/mcp_server_fetch/server.py b/src/fetch/src/mcp_server_fetch/server.py index b42c7b1f6b..ea0cc417bf 100644 --- a/src/fetch/src/mcp_server_fetch/server.py +++ b/src/fetch/src/mcp_server_fetch/server.py @@ -1,5 +1,5 @@ -from typing import Annotated, Tuple -from urllib.parse import urlparse, urlunparse +from typing import TYPE_CHECKING, Annotated, Any, Tuple +from urllib.parse import urljoin, urlparse, urlunparse import markdownify import readabilipy.simple_json @@ -20,9 +20,133 @@ from protego import Protego from pydantic import BaseModel, Field, AnyUrl +if TYPE_CHECKING: + from httpx import AsyncClient, Response + 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)" +REDIRECT_STATUS_CODES = frozenset({301, 302, 303, 307, 308}) +MAX_REDIRECTS = 20 + + +def is_host_allowed(hostname: str | None, allowed_hosts: list[str] | None) -> bool: + """Check whether a hostname is permitted by the configured allowlist. + + Args: + hostname: Hostname taken from the request URL + allowed_hosts: Allowlist entries. ``None`` disables the allowlist (every + host is allowed). An entry is either an exact host (``example.com``) + or a wildcard (``*.example.com``, which matches ``example.com`` itself + and any subdomain). Matching is case-insensitive. + + Returns: + True if the host may be fetched, False otherwise + """ + if allowed_hosts is None: + return True + if not hostname: + return False + hostname = hostname.lower().rstrip(".") + for entry in allowed_hosts: + entry = entry.strip().lower().rstrip(".").strip("[]") + if entry.startswith("*."): + suffix = entry[2:] + if hostname == suffix or hostname.endswith("." + suffix): + return True + elif hostname == entry: + return True + return False + + +def validate_url_allowed(url: str, allowed_hosts: list[str] | None) -> None: + """Validate a URL's host against the allowlist. + + The URL is parsed with httpx — the same parser that will be used to + connect — so the validated host is always the host that gets connected. + + Raises: + McpError: If the URL has no hostname or its host is not allowlisted + """ + if allowed_hosts is None: + return + from httpx import URL, InvalidURL + + try: + hostname = URL(url).host + except InvalidURL: + hostname = None + if not hostname: + raise McpError(ErrorData( + code=INVALID_PARAMS, + message=f"Invalid URL: could not determine a hostname for {url}", + )) + if not is_host_allowed(hostname, allowed_hosts): + raise McpError(ErrorData( + code=INTERNAL_ERROR, + message=f"Fetching '{hostname}' is not allowed: this server is configured with a host allowlist (--allowed-hosts) and this host is not on it. The user can adjust the server configuration if this host should be accessible.", + )) + + +async def _get_following_redirects( + client: "AsyncClient", + url: str, + *, + user_agent: str, + allowed_hosts: list[str] | None, + timeout: float | None = None, +) -> "Response": + """GET a URL, following redirects manually and re-validating every hop. + + Redirects are followed by hand (instead of httpx's follow_redirects) so that + each redirect target is checked against the allowlist before connecting; + otherwise a 302 from an allowed host could bounce the fetch to any host. + + Args: + client: httpx.AsyncClient to use + url: Initial URL to fetch + user_agent: User-Agent header value + allowed_hosts: Allowlist applied to the initial URL and every redirect hop + timeout: Optional per-request timeout in seconds (httpx default if None) + + Returns: + The final (non-redirect) httpx.Response + + Raises: + McpError: If a hop is not allowlisted or the redirect limit is exceeded + """ + request_kwargs: dict[str, Any] = {"follow_redirects": False, "headers": {"User-Agent": user_agent}} + if timeout is not None: + request_kwargs["timeout"] = timeout + + current_url = url + redirects_remaining = MAX_REDIRECTS + while True: + validate_url_allowed(current_url, allowed_hosts) + response = await client.get(current_url, **request_kwargs) + if response.status_code not in REDIRECT_STATUS_CODES: + return response + location = response.headers.get("location") + if location is None: + # A redirect status without a Location header is not followable; + # the response is used as-is (matches httpx's follow_redirects). + return response + if redirects_remaining <= 0: + raise McpError(ErrorData( + code=INTERNAL_ERROR, + message=f"Failed to fetch {url}: exceeded the limit of {MAX_REDIRECTS} redirects", + )) + redirects_remaining -= 1 + # An empty Location redirects to the same URL (matching httpx), so a + # redirect loop — self-inflicted or otherwise — hits the limit above. + try: + current_url = urljoin(str(response.url), location) + except ValueError: + raise McpError(ErrorData( + code=INTERNAL_ERROR, + message=f"Failed to fetch {url}: redirect target {location!r} is not a valid URL", + )) + def extract_content_from_html(html: str) -> str: """Extract and convert HTML content to Markdown format. @@ -63,21 +187,23 @@ def get_robots_txt_url(url: str) -> str: return robots_url -async def check_may_autonomously_fetch_url(url: str, user_agent: str, proxy_url: str | None = None) -> None: +async def check_may_autonomously_fetch_url(url: str, user_agent: str, proxy_url: str | None = None, allowed_hosts: list[str] | None = None) -> None: """ 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 + validate_url_allowed(url, allowed_hosts) robot_txt_url = get_robots_txt_url(url) async with AsyncClient(proxy=proxy_url) as client: try: - response = await client.get( + response = await _get_following_redirects( + client, robot_txt_url, - follow_redirects=True, - headers={"User-Agent": user_agent}, + user_agent=user_agent, + allowed_hosts=allowed_hosts, ) except HTTPError: raise McpError(ErrorData( @@ -109,19 +235,22 @@ async def check_may_autonomously_fetch_url(url: str, user_agent: str, proxy_url: async def fetch_url( - url: str, user_agent: str, force_raw: bool = False, proxy_url: str | None = None + url: str, user_agent: str, force_raw: bool = False, proxy_url: str | None = None, allowed_hosts: list[str] | None = None ) -> Tuple[str, str]: """ 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 + validate_url_allowed(url, allowed_hosts) + async with AsyncClient(proxy=proxy_url) as client: try: - response = await client.get( + response = await _get_following_redirects( + client, url, - follow_redirects=True, - headers={"User-Agent": user_agent}, + user_agent=user_agent, + allowed_hosts=allowed_hosts, timeout=30, ) except HTTPError as e: @@ -182,6 +311,7 @@ async def serve( custom_user_agent: str | None = None, ignore_robots_txt: bool = False, proxy_url: str | None = None, + allowed_hosts: list[str] | None = None, ) -> None: """Run the fetch MCP server. @@ -189,6 +319,8 @@ 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: Optional host allowlist; when set, only these hosts + (exact names or *.example.com wildcards) may be fetched """ server = Server("mcp-fetch") user_agent_autonomous = custom_user_agent or DEFAULT_USER_AGENT_AUTONOMOUS @@ -232,10 +364,10 @@ async def call_tool(name, arguments: dict) -> list[TextContent]: raise McpError(ErrorData(code=INVALID_PARAMS, message="URL is required")) if not ignore_robots_txt: - await check_may_autonomously_fetch_url(url, user_agent_autonomous, proxy_url) + await check_may_autonomously_fetch_url(url, user_agent_autonomous, proxy_url, allowed_hosts=allowed_hosts) content, prefix = await fetch_url( - url, user_agent_autonomous, force_raw=args.raw, proxy_url=proxy_url + url, user_agent_autonomous, force_raw=args.raw, proxy_url=proxy_url, allowed_hosts=allowed_hosts ) original_length = len(content) if args.start_index >= original_length: @@ -262,7 +394,7 @@ async def get_prompt(name: str, arguments: dict | None) -> GetPromptResult: url = arguments["url"] try: - content, prefix = await fetch_url(url, user_agent_manual, proxy_url=proxy_url) + content, prefix = await fetch_url(url, user_agent_manual, proxy_url=proxy_url, allowed_hosts=allowed_hosts) # TODO: after SDK bug is addressed, don't catch the exception except McpError as e: return GetPromptResult( diff --git a/src/fetch/tests/test_server.py b/src/fetch/tests/test_server.py index 96c1cb38c7..dcfaeec6a4 100644 --- a/src/fetch/tests/test_server.py +++ b/src/fetch/tests/test_server.py @@ -7,9 +7,12 @@ from mcp_server_fetch.server import ( extract_content_from_html, get_robots_txt_url, + is_host_allowed, + validate_url_allowed, check_may_autonomously_fetch_url, fetch_url, DEFAULT_USER_AGENT_AUTONOMOUS, + MAX_REDIRECTS, ) @@ -324,3 +327,388 @@ 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") + + +def _make_mock_client(*responses): + """Build a mock httpx.AsyncClient whose get() returns the given responses in order.""" + mock_client = AsyncMock() + mock_client.get = AsyncMock(side_effect=list(responses)) + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=None) + return mock_client + + +def _redirect_response(location: str, url: str = "https://example.com/start"): + mock_response = MagicMock() + mock_response.status_code = 302 + mock_response.headers = {"location": location} + mock_response.url = url + return mock_response + + +def _text_response(text: str = "hello", content_type: str = "text/plain"): + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.text = text + mock_response.headers = {"content-type": content_type} + return mock_response + + +class TestIsHostAllowed: + """Tests for is_host_allowed matching rules.""" + + def test_no_allowlist_allows_everything(self): + """Test that a None allowlist permits any host.""" + assert is_host_allowed("anything.example", None) is True + assert is_host_allowed(None, None) is True + + def test_exact_match(self): + """Test that an exact entry allows the same host.""" + assert is_host_allowed("example.com", ["example.com"]) is True + + def test_non_listed_host_denied(self): + """Test that a host absent from the list is denied.""" + assert is_host_allowed("evil.com", ["example.com"]) is False + + def test_case_insensitive(self): + """Test that matching ignores case on both sides.""" + assert is_host_allowed("ExAmPlE.CoM", ["example.com"]) is True + assert is_host_allowed("example.com", ["EXAMPLE.COM"]) is True + + def test_trailing_dot_normalized(self): + """Test that a trailing root-label dot is ignored.""" + assert is_host_allowed("example.com.", ["example.com"]) is True + + def test_subdomain_of_exact_entry_denied(self): + """Test that an exact entry does not cover subdomains.""" + assert is_host_allowed("api.example.com", ["example.com"]) is False + + def test_wildcard_matches_subdomain(self): + """Test that a wildcard entry matches a subdomain.""" + assert is_host_allowed("api.example.com", ["*.example.com"]) is True + + def test_wildcard_matches_bare_domain(self): + """Test that a wildcard entry also matches the bare domain.""" + assert is_host_allowed("example.com", ["*.example.com"]) is True + + def test_wildcard_matches_deep_subdomain(self): + """Test that a wildcard entry matches multi-level subdomains.""" + assert is_host_allowed("a.b.example.com", ["*.example.com"]) is True + + def test_wildcard_does_not_match_partial_suffix(self): + """Test that lookalike domains sharing a suffix are denied.""" + assert is_host_allowed("notexample.com", ["*.example.com"]) is False + assert is_host_allowed("example.com.evil.com", ["*.example.com"]) is False + assert is_host_allowed("example.com.evil.com", ["example.com"]) is False + + def test_ip_literal_exact_match(self): + """Test that IP literals match literally.""" + assert is_host_allowed("127.0.0.1", ["127.0.0.1"]) is True + assert is_host_allowed("127.0.0.1", ["example.com"]) is False + + def test_ipv6_entry_with_or_without_brackets(self): + """Test that IPv6 entries are accepted in both bracketed and bare form.""" + assert is_host_allowed("::1", ["::1"]) is True + assert is_host_allowed("::1", ["[::1]"]) is True + + def test_empty_list_denies_everything(self): + """Test that an empty allowlist denies all hosts.""" + assert is_host_allowed("example.com", []) is False + + def test_multiple_entries(self): + """Test matching against several entries.""" + allowed = ["example.com", "*.github.com"] + assert is_host_allowed("example.com", allowed) is True + assert is_host_allowed("api.github.com", allowed) is True + assert is_host_allowed("example.org", allowed) is False + + def test_entries_are_stripped(self): + """Test that whitespace around entries is ignored.""" + assert is_host_allowed("example.com", [" example.com "]) is True + + +class TestValidateUrlAllowed: + """Tests for validate_url_allowed.""" + + def test_no_allowlist_never_raises(self): + """Test that a None allowlist permits any URL.""" + validate_url_allowed("https://anything.example/page", None) + + def test_allowed_host_passes(self): + """Test that an allowlisted host passes validation.""" + validate_url_allowed("https://api.example.com/page", ["*.example.com"]) + + def test_denied_host_raises(self): + """Test that a non-allowlisted host raises McpError.""" + with pytest.raises(McpError) as exc_info: + validate_url_allowed("https://evil.com/page", ["example.com"]) + assert "not allowed" in str(exc_info.value) + + def test_userinfo_does_not_bypass_allowlist(self): + """Test that https://allowed.com@evil.com/ connects to evil.com and is denied.""" + with pytest.raises(McpError) as exc_info: + validate_url_allowed("https://example.com@evil.com/page", ["example.com"]) + assert "evil.com" in str(exc_info.value) + + def test_port_is_not_part_of_matching(self): + """Test that an entry allows the host on any port.""" + validate_url_allowed("https://example.com:8443/page", ["example.com"]) + + def test_ipv6_url_validated_without_brackets(self): + """Test that the host of an IPv6 URL matches a bare IPv6 entry.""" + validate_url_allowed("https://[::1]/page", ["::1"]) + + def test_url_without_hostname_raises(self): + """Test that a URL without a hostname raises McpError.""" + with pytest.raises(McpError): + validate_url_allowed("file:///etc/passwd", ["example.com"]) + + def test_malformed_url_raises_mcp_error(self): + """Test that malformed URLs (e.g. invalid IPv6 literal) fail closed with McpError.""" + with pytest.raises(McpError): + validate_url_allowed("https://[example.com]/", ["example.com"]) + + def test_url_without_hostname_allowed_when_no_allowlist(self): + """Test that hostless URLs pass when no allowlist is set.""" + validate_url_allowed("file:///etc/passwd", None) + + +class TestFetchUrlWithAllowlist: + """Tests for allowlist enforcement in fetch_url.""" + + @pytest.mark.asyncio + async def test_allowed_host_fetches(self): + """Test that an allowlisted host is fetched normally.""" + mock_client = _make_mock_client(_text_response('{"ok": true}', "application/json")) + with patch("httpx.AsyncClient", return_value=mock_client): + content, _ = await fetch_url( + "https://example.com/data", + DEFAULT_USER_AGENT_AUTONOMOUS, + allowed_hosts=["example.com"], + ) + assert content == '{"ok": true}' + + @pytest.mark.asyncio + async def test_denied_host_raises_before_request(self): + """Test that a denied host raises before any request is sent.""" + mock_client = _make_mock_client(_text_response()) + with patch("httpx.AsyncClient", return_value=mock_client): + with pytest.raises(McpError) as exc_info: + await fetch_url( + "https://evil.com/data", + DEFAULT_USER_AGENT_AUTONOMOUS, + allowed_hosts=["example.com"], + ) + assert "not allowed" in str(exc_info.value) + mock_client.get.assert_not_called() + + @pytest.mark.asyncio + async def test_wildcard_allowlist_permits_subdomain(self): + """Test fetching a subdomain allowed via a wildcard entry.""" + mock_client = _make_mock_client(_text_response("sub", "text/plain")) + with patch("httpx.AsyncClient", return_value=mock_client): + content, _ = await fetch_url( + "https://api.example.com/data", + DEFAULT_USER_AGENT_AUTONOMOUS, + allowed_hosts=["*.example.com"], + ) + assert content == "sub" + + @pytest.mark.asyncio + async def test_redirect_to_allowed_host_followed(self): + """Test that a redirect to an allowlisted host is followed.""" + mock_client = _make_mock_client( + _redirect_response("https://cdn.example.com/final"), + _text_response("redirected", "text/plain"), + ) + with patch("httpx.AsyncClient", return_value=mock_client): + content, _ = await fetch_url( + "https://example.com/start", + DEFAULT_USER_AGENT_AUTONOMOUS, + allowed_hosts=["*.example.com"], + ) + assert content == "redirected" + assert mock_client.get.call_count == 2 + + @pytest.mark.asyncio + async def test_redirect_to_denied_host_blocked(self): + """Test that a redirect to a denied host raises before the second request.""" + mock_client = _make_mock_client( + _redirect_response("https://evil.com/steal"), + _text_response(), + ) + with patch("httpx.AsyncClient", return_value=mock_client): + with pytest.raises(McpError) as exc_info: + await fetch_url( + "https://example.com/start", + DEFAULT_USER_AGENT_AUTONOMOUS, + allowed_hosts=["example.com"], + ) + assert "not allowed" in str(exc_info.value) + # Only the first request may have gone out + assert mock_client.get.call_count == 1 + + @pytest.mark.asyncio + async def test_redirect_loop_raises(self): + """Test that an endless redirect loop raises after the redirect limit.""" + mock_client = _make_mock_client( + *[_redirect_response("https://example.com/loop", url="https://example.com/loop")] + * (MAX_REDIRECTS + 1), + ) + with patch("httpx.AsyncClient", return_value=mock_client): + with pytest.raises(McpError) as exc_info: + await fetch_url( + "https://example.com/loop", + DEFAULT_USER_AGENT_AUTONOMOUS, + allowed_hosts=["example.com"], + ) + assert "redirect" in str(exc_info.value).lower() + + @pytest.mark.asyncio + async def test_relative_redirect_resolved(self): + """Test that a relative Location header is resolved against the hop URL.""" + mock_client = _make_mock_client( + _redirect_response("/final", url="https://example.com/start"), + _text_response("relative ok", "text/plain"), + ) + with patch("httpx.AsyncClient", return_value=mock_client): + content, _ = await fetch_url( + "https://example.com/start", + DEFAULT_USER_AGENT_AUTONOMOUS, + allowed_hosts=["example.com"], + ) + assert content == "relative ok" + assert mock_client.get.call_args_list[1].args[0] == "https://example.com/final" + + @pytest.mark.asyncio + async def test_empty_location_redirect_loops_until_limit(self): + """Test that an empty Location header self-redirects (like httpx) until the limit.""" + mock_client = _make_mock_client( + *[_redirect_response("", url="https://example.com/start")] * (MAX_REDIRECTS + 1), + ) + with patch("httpx.AsyncClient", return_value=mock_client): + with pytest.raises(McpError) as exc_info: + await fetch_url( + "https://example.com/start", + DEFAULT_USER_AGENT_AUTONOMOUS, + ) + assert "redirect" in str(exc_info.value).lower() + + @pytest.mark.asyncio + async def test_malformed_location_raises_mcp_error(self): + """Test that an unparseable Location header raises McpError, not a raw exception.""" + mock_client = _make_mock_client( + _redirect_response("http://[::1", url="https://example.com/start"), + ) + with patch("httpx.AsyncClient", return_value=mock_client): + with pytest.raises(McpError): + await fetch_url( + "https://example.com/start", + DEFAULT_USER_AGENT_AUTONOMOUS, + ) + + @pytest.mark.asyncio + async def test_redirect_without_location_returned_as_final(self): + """Test that a redirect status without a Location header is returned as-is.""" + response = _text_response("placeholder", "text/plain") + response.status_code = 302 + mock_client = _make_mock_client(response) + with patch("httpx.AsyncClient", return_value=mock_client): + content, _ = await fetch_url( + "https://example.com/start", + DEFAULT_USER_AGENT_AUTONOMOUS, + ) + assert content == "placeholder" + assert mock_client.get.call_count == 1 + + @pytest.mark.asyncio + async def test_no_allowlist_redirect_anywhere_still_works(self): + """Test that without an allowlist, redirects to any host behave as before.""" + mock_client = _make_mock_client( + _redirect_response("https://other.example/final"), + _text_response("free", "text/plain"), + ) + with patch("httpx.AsyncClient", return_value=mock_client): + content, _ = await fetch_url( + "https://example.com/start", + DEFAULT_USER_AGENT_AUTONOMOUS, + ) + assert content == "free" + + @pytest.mark.asyncio + @pytest.mark.parametrize("status_code", [301, 302, 303, 307, 308]) + async def test_all_redirect_statuses_followed_and_validated(self, status_code): + """Test that every redirect status triggers re-validation of the target.""" + redirect = _redirect_response("https://evil.com/final") + redirect.status_code = status_code + mock_client = _make_mock_client(redirect, _text_response()) + with patch("httpx.AsyncClient", return_value=mock_client): + with pytest.raises(McpError) as exc_info: + await fetch_url( + "https://example.com/start", + DEFAULT_USER_AGENT_AUTONOMOUS, + allowed_hosts=["example.com"], + ) + assert "not allowed" in str(exc_info.value) + + @pytest.mark.asyncio + async def test_per_request_timeout_preserved(self): + """Test that fetch requests keep their 30 second per-request timeout.""" + mock_client = _make_mock_client(_text_response("ok", "text/plain")) + with patch("httpx.AsyncClient", return_value=mock_client): + await fetch_url( + "https://example.com/data", + DEFAULT_USER_AGENT_AUTONOMOUS, + allowed_hosts=["example.com"], + ) + assert mock_client.get.call_args.kwargs["timeout"] == 30 + + +class TestCheckMayAutonomouslyFetchUrlWithAllowlist: + """Tests for allowlist enforcement in the robots.txt pre-check.""" + + @pytest.mark.asyncio + async def test_denied_host_raises_before_request(self): + """Test that a denied host raises before the robots.txt request is sent.""" + mock_client = _make_mock_client(_text_response()) + with patch("httpx.AsyncClient", return_value=mock_client): + with pytest.raises(McpError) as exc_info: + await check_may_autonomously_fetch_url( + "https://evil.com/page", + DEFAULT_USER_AGENT_AUTONOMOUS, + allowed_hosts=["example.com"], + ) + assert "not allowed" in str(exc_info.value) + mock_client.get.assert_not_called() + + @pytest.mark.asyncio + async def test_allowed_host_checks_robots(self): + """Test that an allowlisted host has its robots.txt fetched.""" + mock_response = MagicMock() + mock_response.status_code = 404 + mock_client = _make_mock_client(mock_response) + with patch("httpx.AsyncClient", return_value=mock_client): + # Should not raise + await check_may_autonomously_fetch_url( + "https://example.com/page", + DEFAULT_USER_AGENT_AUTONOMOUS, + allowed_hosts=["*.example.com"], + ) + assert mock_client.get.call_args_list[0].args[0] == "https://example.com/robots.txt" + + @pytest.mark.asyncio + async def test_robots_redirect_to_denied_host_blocked(self): + """Test that a robots.txt redirect to a denied host is blocked.""" + mock_client = _make_mock_client( + _redirect_response("https://evil.com/robots.txt", url="https://example.com/robots.txt"), + _text_response(), + ) + with patch("httpx.AsyncClient", return_value=mock_client): + with pytest.raises(McpError) as exc_info: + await check_may_autonomously_fetch_url( + "https://example.com/page", + DEFAULT_USER_AGENT_AUTONOMOUS, + allowed_hosts=["example.com"], + ) + assert "not allowed" in str(exc_info.value) + assert mock_client.get.call_count == 1