fix(websearch): block non-public URLs in page fetching

ms_agent/tools/fetch_playwright_fallback.py

@@ -18,6 +18,7 @@
 import threading
 from typing import Dict, List, Tuple
 
+from ms_agent.tools.url_safety import is_safe_public_url
 from ms_agent.utils.logger import get_logger
 
 logger = get_logger()
@@ -130,6 +131,8 @@ def try_playwright_inner_text(
     """
     if not url.startswith(('http://', 'https://')):
         return ''
+    if not is_safe_public_url(url):
+        return ''
     settle_ms = max(0, int(settle_ms))
     try:
         from playwright.sync_api import sync_playwright  # noqa: F401
@@ -143,7 +146,25 @@ def try_playwright_inner_text(
     text = ''
     try:
         browser = _thread_browser()
-        page = browser.new_page()
+        context = browser.new_context()
+        page = context.new_page()
+
+        def _guard_navigation(route) -> None:
+            # Redirects are followed by the browser; re-validate every document
+            # navigation so a public URL cannot pivot to an internal one.
+            try:
+                if (route.request.resource_type == 'document'
+                        and not is_safe_public_url(route.request.url)):
+                    route.abort()
+                else:
+                    route.continue_()
+            except Exception:
+                try:
+                    route.continue_()
+                except Exception:
+                    pass
+
+        context.route('**/*', _guard_navigation)
         try:
             page.set_default_timeout(timeout_ms)
             page.goto(url, wait_until='domcontentloaded', timeout=timeout_ms)
@@ -161,6 +182,10 @@ def try_playwright_inner_text(
                 page.close()
             except Exception:
                 pass
+            try:
+                context.close()
+            except Exception:
+                pass
     except RuntimeError:
         return ''
     except Exception as e:

ms_agent/tools/jina_reader.py

@@ -7,11 +7,12 @@
 from dataclasses import dataclass, field
 from typing import Any, Dict, List, Optional, Tuple
 from urllib.error import HTTPError, URLError
-from urllib.parse import quote, urlparse
-from urllib.request import Request, urlopen
+from urllib.parse import quote
+from urllib.request import Request
 
 from ms_agent.tools.fetch_playwright_fallback import (
     looks_like_spa_shell_html, try_playwright_inner_text)
+from ms_agent.tools.url_safety import is_safe_public_url, safe_urlopen
 from ms_agent.utils.logger import get_logger
 
 logger = get_logger()
@@ -79,15 +80,7 @@ def _postprocess_text(raw_text: str) -> str:
 
 
 def _is_direct_http_allowed(url: str) -> bool:
-    try:
-        parsed = urlparse(url)
-        if parsed.scheme not in ('http', 'https'):
-            return False
-        if not parsed.netloc:
-            return False
-        return True
-    except Exception:
-        return False
+    return is_safe_public_url(url)
 
 
 def _html_to_plaintext(html: str) -> str:
@@ -124,7 +117,7 @@ def _fetch_direct_http_pair(url: str, timeout: float) -> Tuple[str, str]:
         return '', ''
     try:
         req = Request(url, headers=_DIRECT_FETCH_HEADERS)
-        with urlopen(req, timeout=timeout) as resp:
+        with safe_urlopen(req, timeout=timeout) as resp:
             raw = resp.read(_MAX_DIRECT_RESPONSE_BYTES + 1)
             if len(raw) > _MAX_DIRECT_RESPONSE_BYTES:
                 raw = raw[:_MAX_DIRECT_RESPONSE_BYTES]
@@ -201,8 +194,14 @@ def fetch_single_text_with_meta(
 
     Returns:
         (text, meta) where ``meta['content_source']`` is one of:
-        ``jina_reader`` | ``direct_http_fallback`` | ``playwright_fallback`` | ``none``.
+        ``jina_reader`` | ``direct_http_fallback`` | ``playwright_fallback`` | ``none`` | ``blocked``.
     """
+    if not is_safe_public_url(url):
+        logger.warning('Blocked fetch of non-public URL: %s', url[:80])
+        return '', {
+            'content_source': 'blocked',
+            'fetch_error': 'blocked_non_public_url',
+        }
     jina_raw = _fetch_via_jina(url, config)
     jina_text = _postprocess_text(jina_raw)
     if jina_text:

ms_agent/tools/search/websearch_tool.py

@@ -162,7 +162,12 @@ def chunk_text_simple(text: str,
 
 
 class ContentFetcher:
-    """Base interface for content fetching."""
+    """Base interface for content fetching.
+
+    Implementations that issue requests from the local process must reject
+    non-public destinations, for example with
+    :func:`ms_agent.tools.url_safety.is_safe_public_url`.
+    """
 
     def fetch(self, url: str) -> Tuple[str, Dict[str, Any]]:
         """

ms_agent/tools/url_safety.py

@@ -0,0 +1,108 @@
+# Copyright (c) ModelScope Contributors. All rights reserved.
+"""Helpers for restricting outbound fetches to public HTTP(S) destinations."""
+
+from __future__ import annotations
+
+import ipaddress
+import socket
+from typing import Any
+from urllib.error import HTTPError
+from urllib.parse import urlparse
+from urllib.request import (HTTPRedirectHandler, Request, build_opener)
+
+_BLOCKED_HOSTNAMES = frozenset({
+    'localhost',
+    'localhost.localdomain',
+    'metadata',
+    'metadata.google.internal',
+})
+_BLOCKED_HOST_SUFFIXES = ('.localhost', '.local', '.internal')
+
+
+def _is_public_ip(address: str) -> bool:
+    try:
+        ip = ipaddress.ip_address(address.split('%', 1)[0])
+    except ValueError:
+        return False
+    if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None:
+        ip = ip.ipv4_mapped
+    return bool(
+        ip.is_global
+        and not (ip.is_private or ip.is_loopback or ip.is_link_local
+                 or ip.is_multicast or ip.is_reserved or ip.is_unspecified))
+
+
+def _resolve_hostname(hostname: str) -> list[str]:
+    infos = socket.getaddrinfo(hostname, None, type=socket.SOCK_STREAM)
+    return [info[4][0] for info in infos]
+
+
+def validate_public_url(url: str) -> str:
+    """Return ``url`` when it targets a public HTTP(S) host.
+
+    Raises:
+        ValueError: if the scheme, hostname, or resolved addresses are not
+            public HTTP(S). Hostnames are resolved here so a name pointing at
+            loopback, RFC1918, link-local, or cloud-metadata space is rejected.
+    """
+    try:
+        parsed = urlparse(url)
+    except ValueError as exc:
+        raise ValueError(f'invalid URL: {url!r}') from exc
+
+    if parsed.scheme.lower() not in {'http', 'https'}:
+        raise ValueError(f'unsupported URL scheme: {parsed.scheme!r}')
+    hostname = (parsed.hostname or '').strip().lower()
+    if not hostname:
+        raise ValueError('URL must include a hostname')
+    if hostname in _BLOCKED_HOSTNAMES or hostname.endswith(
+            _BLOCKED_HOST_SUFFIXES):
+        raise ValueError(f'blocked hostname: {hostname!r}')
+    try:
+        parsed.port
+    except ValueError as exc:
+        raise ValueError(f'invalid port in URL: {url!r}') from exc
+
+    try:
+        addresses = _resolve_hostname(hostname)
+    except OSError as exc:
+        raise ValueError(f'could not resolve hostname: {hostname!r}') from exc
+    if not addresses:
+        raise ValueError(f'could not resolve hostname: {hostname!r}')
+    if not all(_is_public_ip(address) for address in addresses):
+        raise ValueError(f'URL resolves to a non-public address: {hostname!r}')
+    return url
+
+
+def is_safe_public_url(url: str) -> bool:
+    """Return whether ``url`` may be fetched from the local host."""
+    try:
+        validate_public_url(url)
+    except ValueError:
+        return False
+    return True
+
+
+class _SafeRedirectHandler(HTTPRedirectHandler):
+    """Reject redirects that leave the public HTTP(S) address space."""
+
+    def redirect_request(self, req, fp, code, msg, headers, newurl):
+        if not is_safe_public_url(newurl):
+            raise HTTPError(
+                newurl,
+                code,
+                f'blocked redirect to a non-public URL: {newurl}',
+                headers,
+                fp,
+            )
+        return super().redirect_request(req, fp, code, msg, headers, newurl)
+
+
+_SAFE_OPENER = build_opener(_SafeRedirectHandler())
+
+
+def safe_urlopen(request: Request, timeout: float) -> Any:
+    """Open ``request`` and validate every redirect target before following it."""
+    if not is_safe_public_url(request.full_url):
+        raise ValueError(f'blocked non-public URL: {request.full_url}')
+    return _SAFE_OPENER.open(request, timeout=timeout)

tests/tools/test_url_safety.py

@@ -0,0 +1,150 @@
+# Copyright (c) ModelScope Contributors. All rights reserved.
+import socket
+from email.message import Message
+from unittest.mock import patch
+from urllib.error import HTTPError
+from urllib.request import Request
+
+import pytest
+
+from ms_agent.tools import jina_reader
+from ms_agent.tools.search.websearch_tool import JinaContentFetcher
+from ms_agent.tools.url_safety import (_SafeRedirectHandler,
+                                       is_safe_public_url, safe_urlopen)
+
+
+def _addrinfo(*addresses):
+    return [(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, '',
+             (address, 0)) for address in addresses]
+
+
+@pytest.mark.parametrize(
+    'url',
+    [
+        'http://127.0.0.1/',
+        'http://10.0.0.1/',
+        'http://192.168.1.1/',
+        'http://169.254.169.254/latest/meta-data/',
+        'http://100.100.100.200/',
+        'http://0.0.0.0/',
+        'http://[::1]/',
+        'http://[fe80::1]/',
+        'http://[::ffff:127.0.0.1]/',
+        'http://224.0.0.1/',
+        'http://localhost/',
+        'http://service.internal/',
+        'ftp://example.com/file',
+        'file:///etc/passwd',
+        'http:///missing-host',
+        'http://example.com:not-a-port/',
+    ],
+)
+def test_rejects_non_public_literal_urls(url):
+    assert not is_safe_public_url(url)
+
+
+@pytest.mark.parametrize(
+    'address',
+    [
+        '10.0.0.1',
+        '192.168.1.1',
+        '169.254.169.254',
+        '100.100.100.200',
+        '127.0.0.1',
+        '::1',
+        'fe80::1',
+        '::ffff:127.0.0.1',
+        '224.0.0.1',
+        '0.0.0.0',
+    ],
+)
+def test_rejects_hostnames_resolving_to_non_public(address):
+    with patch(
+            'ms_agent.tools.url_safety.socket.getaddrinfo',
+            return_value=_addrinfo(address)):
+        assert not is_safe_public_url('https://example.com/page')
+
+
+def test_rejects_mixed_public_and_private_dns_answers():
+    with patch(
+            'ms_agent.tools.url_safety.socket.getaddrinfo',
+            return_value=_addrinfo('93.184.216.34', '10.0.0.1')):
+        assert not is_safe_public_url('https://example.com/')
+
+
+def test_allows_public_hostname():
+    with patch(
+            'ms_agent.tools.url_safety.socket.getaddrinfo',
+            return_value=_addrinfo('93.184.216.34')):
+        assert is_safe_public_url('https://example.com/path?q=1')
+
+
+def test_rejects_dns_failure():
+    with patch(
+            'ms_agent.tools.url_safety.socket.getaddrinfo',
+            side_effect=socket.gaierror):
+        assert not is_safe_public_url('https://example.com/')
+
+
+def test_safe_urlopen_blocks_non_public_initial_url():
+    with pytest.raises(ValueError, match='non-public URL'):
+        safe_urlopen(Request('http://127.0.0.1/'), timeout=1)
+
+
+def test_redirect_handler_blocks_non_public_target():
+    handler = _SafeRedirectHandler()
+    with patch(
+            'ms_agent.tools.url_safety.is_safe_public_url',
+            return_value=False):
+        with pytest.raises(HTTPError):
+            handler.redirect_request(
+                Request('https://example.com/'),
+                None,
+                302,
+                'Found',
+                Message(),
+                'http://169.254.169.254/',
+            )
+
+
+def test_redirect_handler_allows_public_target():
+    handler = _SafeRedirectHandler()
+    with patch(
+            'ms_agent.tools.url_safety.is_safe_public_url',
+            return_value=True):
+        redirected = handler.redirect_request(
+            Request('https://example.com/'),
+            None,
+            302,
+            'Found',
+            Message(),
+            'https://example.org/',
+        )
+    assert redirected.full_url == 'https://example.org/'
+
+
+def test_fetch_single_text_blocks_before_jina(monkeypatch):
+    def fail(*args, **kwargs):
+        raise AssertionError('Jina must not be called for a non-public URL')
+
+    monkeypatch.setattr(jina_reader, '_fetch_via_jina', fail)
+    text, meta = jina_reader.fetch_single_text_with_meta(
+        'http://127.0.0.1/',
+        jina_reader.JinaReaderConfig(),
+    )
+    assert text == ''
+    assert meta['content_source'] == 'blocked'
+    assert meta['fetch_error'] == 'blocked_non_public_url'
+
+
+def test_jina_content_fetcher_blocks_non_public():
+    content, meta = JinaContentFetcher().fetch('http://127.0.0.1/')
+    assert content == ''
+    assert meta['content_source'] == 'blocked'
+    assert meta['fetch_error'] == 'blocked_non_public_url'
+
+
+def test_playwright_blocks_non_public_before_import():
+    from ms_agent.tools.fetch_playwright_fallback import try_playwright_inner_text
+
+    assert try_playwright_inner_text('http://127.0.0.1/', 1000) == ''