diff --git a/devagent/tools/web_tools.py b/devagent/tools/web_tools.py index 71f19c7..22bddc2 100644 --- a/devagent/tools/web_tools.py +++ b/devagent/tools/web_tools.py @@ -10,17 +10,136 @@ from __future__ import annotations +import ipaddress import re +import socket from html.parser import HTMLParser from typing import Any +from urllib.parse import urljoin + +import httpx _DEFAULT_UA = "DevAgent/0.5 (https://github.com/knight22-21/DevAgent)" _BRAVE_SEARCH_URL = "https://api.search.brave.com/res/v1/web/search" _MAX_FETCH_CHARS = 8_000 # cap fetch_url text output +_MAX_FETCH_REDIRECTS = 5 +_FETCH_REDIRECT_STATUSES = frozenset({301, 302, 303, 307, 308}) +FETCH_URL_ALLOWED_SCHEMES = frozenset({"http", "https"}) +_IPV4_COMPATIBLE_PREFIX = ipaddress.IPv6Network("::/96") +_NAT64_WELL_KNOWN_PREFIX = ipaddress.IPv6Network("64:ff9b::/96") _SKIP_TAGS = frozenset({"script", "style", "head", "noscript", "nav", "footer"}) _BLOCK_TAGS = frozenset({"p", "br", "div", "li", "h1", "h2", "h3", "h4", "h5", "tr", "dt", "dd"}) +def _resolve_public_addresses( + url: httpx.URL, +) -> list[ipaddress.IPv4Address | ipaddress.IPv6Address]: + host = url.host + if "%" in host: + raise ValueError("Scoped IP addresses are not allowed") + + try: + addresses = [ipaddress.ip_address(host)] + except ValueError: + port = url.port or (443 if url.scheme == "https" else 80) + records = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM) + addresses = [] + for record in records: + address = record[4][0] + if not isinstance(address, str): + raise TypeError("DNS resolution returned an invalid address") + if "%" in address: + raise ValueError("Scoped IP addresses are not allowed") + parsed_address = ipaddress.ip_address(address) + if parsed_address not in addresses: + addresses.append(parsed_address) + + if not addresses: + raise ValueError(f"URL host did not resolve: {host}") + if any(not _is_public_address(address) for address in addresses): + raise ValueError(f"URL host resolves to a non-public address: {host}") + return addresses + + +def _is_public_address(address: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool: + if isinstance(address, ipaddress.IPv6Address): + if address.ipv4_mapped is not None: + return address.ipv4_mapped.is_global + if ( + address in _IPV4_COMPATIBLE_PREFIX + or address.sixtofour is not None + or address.teredo is not None + ): + return False + if address in _NAT64_WELL_KNOWN_PREFIX: + embedded_ipv4 = ipaddress.IPv4Address(int(address) & 0xFFFFFFFF) + return embedded_ipv4.is_global + return address.is_global + + +def _fetch_url_response( + url: str, + *, + transport: httpx.BaseTransport | None = None, +) -> httpx.Response: + current_url = httpx.URL(url) + redirects = 0 + + with httpx.Client( + transport=transport, + timeout=20.0, + follow_redirects=False, + trust_env=False, + ) as client: + while True: + if current_url.scheme not in FETCH_URL_ALLOWED_SCHEMES: + raise ValueError("Only HTTP and HTTPS URLs can be fetched") + if not current_url.host or current_url.userinfo: + raise ValueError("URL must have a host and cannot include credentials") + + addresses = _resolve_public_addresses(current_url) + extensions: dict[str, Any] = {} + try: + ipaddress.ip_address(current_url.host) + except ValueError: + if current_url.scheme == "https": + extensions["sni_hostname"] = current_url.host + + response: httpx.Response | None = None + last_connect_error: httpx.TransportError | None = None + for address in addresses: + # Pin each attempt to a validated IP while preserving DNS fallback behavior. + request = client.build_request( + "GET", + current_url.copy_with(host=address.compressed), + headers={ + "User-Agent": _DEFAULT_UA, + "Host": current_url.netloc.decode("ascii"), + }, + extensions=extensions, + ) + try: + response = client.send(request, follow_redirects=False) + except (httpx.ConnectError, httpx.ConnectTimeout) as exc: + last_connect_error = exc + continue + break + if response is None: + assert last_connect_error is not None + raise last_connect_error + + location = response.headers.get("location") + if response.status_code not in _FETCH_REDIRECT_STATUSES or location is None: + response.raise_for_status() + return response + + response.close() + if redirects >= _MAX_FETCH_REDIRECTS: + raise ValueError(f"Too many redirects (maximum {_MAX_FETCH_REDIRECTS})") + current_url = httpx.URL(urljoin(str(current_url), location)) + redirects += 1 + + # --------------------------------------------------------------------------- # HTML → plain text # --------------------------------------------------------------------------- @@ -151,14 +270,7 @@ def fetch_url(args: dict[str, Any]) -> str: return "[error] url is required" extract_text = args.get("extract_text", True) try: - import httpx - resp = httpx.get( - url, - headers={"User-Agent": _DEFAULT_UA}, - timeout=20.0, - follow_redirects=True, - ) - resp.raise_for_status() + resp = _fetch_url_response(url) except Exception as exc: return f"[error] fetch failed: {exc}" diff --git a/tests/test_phase11.py b/tests/test_phase11.py index 3097e58..c81d3ab 100644 --- a/tests/test_phase11.py +++ b/tests/test_phase11.py @@ -2,8 +2,12 @@ from __future__ import annotations +import socket from unittest.mock import MagicMock, patch +import httpx +import pytest + # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- @@ -43,6 +47,19 @@ def _text_response(text: str, content_type: str = "text/plain") -> MagicMock: return mock_resp +def _addrinfo(address: str, port: int) -> list[tuple]: + family = socket.AF_INET6 if ":" in address else socket.AF_INET + sockaddr = (address, port, 0, 0) if family == socket.AF_INET6 else (address, port) + return [(family, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", sockaddr)] + + +def _record_request_response( + requests: list[httpx.Request], request: httpx.Request, response: httpx.Response +) -> httpx.Response: + requests.append(request) + return response + + # --------------------------------------------------------------------------- # 11.1 fetch_url — always registered # --------------------------------------------------------------------------- @@ -60,7 +77,9 @@ def test_fetch_url_html_extracts_text(self) -> None: reg = _make_registry() register_web_tools(reg) html = "T

Hello world

" - with patch("httpx.get", return_value=_html_response(html)): + with patch( + "devagent.tools.web_tools._fetch_url_response", return_value=_html_response(html) + ): result = reg.call("fetch_url", {"url": "http://example.com", "extract_text": True}) assert "Hello world" in result assert "

" not in result @@ -70,7 +89,9 @@ def test_fetch_url_strips_script_tags(self) -> None: reg = _make_registry() register_web_tools(reg) html = "

Visible

" - with patch("httpx.get", return_value=_html_response(html)): + with patch( + "devagent.tools.web_tools._fetch_url_response", return_value=_html_response(html) + ): result = reg.call("fetch_url", {"url": "http://example.com"}) assert "var x" not in result assert "Visible" in result @@ -79,7 +100,10 @@ def test_fetch_url_returns_plain_text_unchanged(self) -> None: from devagent.tools.web_tools import register_web_tools reg = _make_registry() register_web_tools(reg) - with patch("httpx.get", return_value=_text_response("raw text content")): + with patch( + "devagent.tools.web_tools._fetch_url_response", + return_value=_text_response("raw text content"), + ): result = reg.call("fetch_url", {"url": "http://example.com/data.txt"}) assert "raw text content" in result @@ -96,10 +120,314 @@ def test_fetch_url_handles_http_error(self) -> None: from devagent.tools.web_tools import register_web_tools reg = _make_registry() register_web_tools(reg) - with patch("httpx.get", side_effect=httpx.ConnectError("refused")): + with patch( + "devagent.tools.web_tools._fetch_url_response", + side_effect=httpx.ConnectError("refused"), + ): result = reg.call("fetch_url", {"url": "http://unreachable.invalid"}) assert "[error]" in result + def test_fetch_url_blocks_cloud_metadata_before_request(self) -> None: + import httpx + + from devagent.tools.web_tools import _fetch_url_response, register_web_tools + + reg = _make_registry() + register_web_tools(reg) + requests: list[httpx.Request] = [] + transport = httpx.MockTransport( + lambda request: _record_request_response( + requests, request, httpx.Response(200, text="metadata") + ) + ) + fetch = lambda url: _fetch_url_response(url, transport=transport) + with patch("devagent.tools.web_tools._fetch_url_response", side_effect=fetch): + result = reg.call("fetch_url", {"url": "http://169.254.169.254/latest/meta-data/"}) + assert "[error]" in result + assert requests == [] + + def test_fetch_url_blocks_localhost_before_request(self) -> None: + import httpx + + from devagent.tools.web_tools import _fetch_url_response, register_web_tools + + reg = _make_registry() + register_web_tools(reg) + requests: list[httpx.Request] = [] + transport = httpx.MockTransport( + lambda request: _record_request_response( + requests, request, httpx.Response(200, text="private") + ) + ) + fetch = lambda url: _fetch_url_response(url, transport=transport) + with ( + patch("devagent.tools.web_tools._fetch_url_response", side_effect=fetch), + patch( + "devagent.tools.web_tools.socket.getaddrinfo", + return_value=_addrinfo("127.0.0.1", 6379), + ), + ): + result = reg.call("fetch_url", {"url": "http://localhost:6379/"}) + + assert "[error]" in result + assert requests == [] + + def test_fetch_url_rejects_non_public_addresses_and_schemes(self) -> None: + import httpx + + from devagent.tools.web_tools import _fetch_url_response + + cases = [ + ("http://169.254.169.254/latest/meta-data/", None), + ("http://127.0.0.1:6379/", None), + ("http://[::1]:8080/admin", None), + ("http://[fd00::1]/", None), + ("http://[::a9fe:a9fe]/", None), + ("http://[::ffff:169.254.169.254]/", None), + ("http://[2002:a9fe:a9fe::]/", None), + ("http://[2001:0000:4136:e378:8000:63bf:3fff:fdd2]/", None), + ("http://[64:ff9b::a9fe:a9fe]/", None), + ("http://localhost:6379/", ["127.0.0.1"]), + ("http://mixed.example/", ["93.184.216.34", "10.0.0.8"]), + ("http://nat64.example/", ["64:ff9b::a9fe:a9fe"]), + ("file:///etc/passwd", None), + ("http://user:password@example.com/", None), + ] + + for url, resolved_addresses in cases: + requests: list[httpx.Request] = [] + + def respond(request: httpx.Request, seen=requests) -> httpx.Response: + seen.append(request) + return httpx.Response(200, text="reached") + + transport = httpx.MockTransport(respond) + + def resolve( + host: str, + port: int, + addresses: list[str] | None = resolved_addresses, + **_kwargs: object, + ) -> list[tuple]: + if addresses is None: + raise AssertionError(f"Unexpected DNS lookup for {host}") + return [record for address in addresses for record in _addrinfo(address, port)] + + with ( + patch("devagent.tools.web_tools.socket.getaddrinfo", side_effect=resolve), + pytest.raises(ValueError), + ): + _fetch_url_response(url, transport=transport) + assert requests == [], url + + def test_fetch_url_rejects_invalid_dns_record_before_request(self) -> None: + import httpx + + from devagent.tools.web_tools import _fetch_url_response + + requests: list[httpx.Request] = [] + transport = httpx.MockTransport( + lambda request: _record_request_response( + requests, request, httpx.Response(200, text="reached") + ) + ) + invalid_record = (socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", (123, 80)) + + with ( + patch("devagent.tools.web_tools.socket.getaddrinfo", return_value=[invalid_record]), + pytest.raises(TypeError, match="invalid address"), + ): + _fetch_url_response("http://invalid-dns.example/", transport=transport) + + assert requests == [] + + def test_fetch_url_pins_public_dns_result_and_preserves_host(self) -> None: + import httpx + + from devagent.tools.web_tools import _fetch_url_response + + requests: list[httpx.Request] = [] + transport = httpx.MockTransport( + lambda request: _record_request_response( + requests, request, httpx.Response(200, text="safe body") + ) + ) + with patch( + "devagent.tools.web_tools.socket.getaddrinfo", + return_value=_addrinfo("93.184.216.34", 8080), + ) as resolve: + response = _fetch_url_response( + "http://example.com:8080/path?query=value", transport=transport + ) + + assert response.text == "safe body" + assert len(requests) == 1 + assert requests[0].url.host == "93.184.216.34" + assert str(requests[0].url) == "http://93.184.216.34:8080/path?query=value" + assert requests[0].headers["host"] == "example.com:8080" + resolve.assert_called_once_with("example.com", 8080, type=socket.SOCK_STREAM) + + @pytest.mark.parametrize("error_type", ["connect", "timeout"]) + def test_fetch_url_falls_back_to_next_public_dns_result(self, error_type: str) -> None: + import httpx + + from devagent.tools.web_tools import _fetch_url_response + + requests: list[httpx.Request] = [] + + def respond(request: httpx.Request) -> httpx.Response: + requests.append(request) + if len(requests) == 1: + error = ( + httpx.ConnectError("connection refused", request=request) + if error_type == "connect" + else httpx.ConnectTimeout("connection timed out", request=request) + ) + raise error + return httpx.Response(200, text="fallback body") + + transport = httpx.MockTransport(respond) + addresses = [_addrinfo("93.184.216.34", 80), _addrinfo("1.1.1.1", 80)] + with patch( + "devagent.tools.web_tools.socket.getaddrinfo", + return_value=[record for result in addresses for record in result], + ) as resolve: + response = _fetch_url_response("http://example.com/path", transport=transport) + + assert response.text == "fallback body" + assert [request.url.host for request in requests] == ["93.184.216.34", "1.1.1.1"] + assert [request.headers["host"] for request in requests] == [ + "example.com", + "example.com", + ] + resolve.assert_called_once_with("example.com", 80, type=socket.SOCK_STREAM) + + def test_fetch_url_ignores_environment_proxies(self) -> None: + import httpx + + from devagent.tools.web_tools import _fetch_url_response + + requests: list[httpx.Request] = [] + mock_transport = httpx.MockTransport( + lambda request: _record_request_response( + requests, request, httpx.Response(200, text="direct") + ) + ) + real_client = httpx.Client + client_options: list[tuple[httpx.BaseTransport | None, bool]] = [] + + def make_client( + *, + transport: httpx.BaseTransport | None, + timeout: float, + follow_redirects: bool, + trust_env: bool, + ) -> httpx.Client: + client_options.append((transport, trust_env)) + return real_client( + transport=mock_transport, + timeout=timeout, + follow_redirects=follow_redirects, + trust_env=trust_env, + ) + + with ( + patch("devagent.tools.web_tools.httpx.Client", side_effect=make_client), + patch( + "devagent.tools.web_tools.socket.getaddrinfo", + return_value=_addrinfo("93.184.216.34", 443), + ), + ): + response = _fetch_url_response("https://example.com/", transport=None) + + assert response.text == "direct" + assert client_options == [(None, False)] + assert len(requests) == 1 + + def test_fetch_url_revalidates_each_redirect_and_keeps_tls_sni(self) -> None: + import httpx + + from devagent.tools.web_tools import _fetch_url_response + + requests: list[httpx.Request] = [] + + def respond(request: httpx.Request) -> httpx.Response: + requests.append(request) + if len(requests) == 1: + return httpx.Response(302, headers={"location": "https://next.example.org/final"}) + return httpx.Response(200, text="redirected body") + + transport = httpx.MockTransport(respond) + addresses = {"example.com": "93.184.216.34", "next.example.org": "1.1.1.1"} + + def resolve(host: str, port: int, **kwargs: object) -> list[tuple]: + return _addrinfo(addresses[host], port) + + with patch("devagent.tools.web_tools.socket.getaddrinfo", side_effect=resolve) as resolver: + response = _fetch_url_response("http://example.com/start", transport=transport) + + assert response.text == "redirected body" + assert [request.url.host for request in requests] == ["93.184.216.34", "1.1.1.1"] + assert [request.headers["host"] for request in requests] == [ + "example.com", + "next.example.org", + ] + assert requests[1].extensions["sni_hostname"] == "next.example.org" + assert [call.args[0] for call in resolver.call_args_list] == [ + "example.com", + "next.example.org", + ] + + def test_fetch_url_stops_redirect_to_private_dns_target(self) -> None: + import httpx + + from devagent.tools.web_tools import _fetch_url_response + + requests: list[httpx.Request] = [] + transport = httpx.MockTransport( + lambda request: ( + _record_request_response( + requests, + request, + httpx.Response(302, headers={"location": "http://internal.example/metadata"}), + ) + ) + ) + + def resolve(host: str, port: int, **kwargs: object) -> list[tuple]: + address = "93.184.216.34" if host == "example.com" else "10.0.0.8" + return _addrinfo(address, port) + + with ( + patch("devagent.tools.web_tools.socket.getaddrinfo", side_effect=resolve), + pytest.raises(ValueError, match="non-public address"), + ): + _fetch_url_response("http://example.com/start", transport=transport) + assert len(requests) == 1 + + def test_fetch_url_limits_redirects(self) -> None: + import httpx + + from devagent.tools.web_tools import _MAX_FETCH_REDIRECTS, _fetch_url_response + + requests: list[httpx.Request] = [] + transport = httpx.MockTransport( + lambda request: ( + _record_request_response( + requests, request, httpx.Response(302, headers={"location": "/again"}) + ) + ) + ) + with ( + patch( + "devagent.tools.web_tools.socket.getaddrinfo", + return_value=_addrinfo("93.184.216.34", 80), + ), + pytest.raises(ValueError, match="Too many redirects"), + ): + _fetch_url_response("http://example.com/start", transport=transport) + assert len(requests) == _MAX_FETCH_REDIRECTS + 1 + # --------------------------------------------------------------------------- # 11.2 web_search — Brave provider