CVE-2026-10177

@@ -1,7 +1,10 @@
 #!/usr/bin/env python
 
+import ipaddress
 import re
+import socket
 import sys
+from urllib.parse import urljoin, urlsplit
 
 import pypandoc
 
@@ -10,6 +13,72 @@
 
 aider_user_agent = f"Aider/{__version__} +{urls.website}"
 
+blocked_scrape_hostnames = {
+    "metadata.google.internal",
+}
+
+
+class ScrapeNetworkBackend:
+    def __init__(self, scraper):
+        self.scraper = scraper
+
+    def connect_tcp(
+        self,
+        host,
+        port,
+        timeout=None,
+        local_address=None,
+        socket_options=None,
+    ):
+        from httpcore import ConnectError, ConnectTimeout
+        from httpcore._backends.sync import SyncStream
+
+        addresses, error = self.scraper.get_safe_addresses(
+            host, port, "http", f"{host}:{port}"
+        )
+        if error:
+            raise ConnectError(error)
+
+        last_error = None
+        for family, socktype, proto, sockaddr, _ip in addresses:
+            sock = None
+            try:
+                sock = socket.socket(family, socktype, proto)
+                sock.settimeout(timeout)
+                if local_address is not None:
+                    if family == socket.AF_INET6:
+                        sock.bind((local_address, 0, 0, 0))
+                    else:
+                        sock.bind((local_address, 0))
+                if socket_options:
+                    for option in socket_options:
+                        sock.setsockopt(*option)
+                sock.connect(sockaddr)
+                sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
+                return SyncStream(sock)
+            except socket.timeout as err:
+                last_error = err
+            except OSError as err:
+                last_error = err
+
+            if sock is not None:
+                sock.close()
+
+        if isinstance(last_error, socket.timeout):
+            raise ConnectTimeout(str(last_error))
+        raise ConnectError(str(last_error or f"Unable to connect to {host}:{port}"))
+
+    def connect_unix_socket(self, path, timeout=None, socket_options=None):
+        from httpcore._backends.sync import SyncBackend
+
+        return SyncBackend().connect_unix_socket(path, timeout, socket_options)
+
+    def sleep(self, seconds):
+        from time import sleep
+
+        sleep(seconds)
+
+
 # Playwright is nice because it has a simple way to install dependencies on most
 # platforms.
 
@@ -94,6 +163,7 @@ def __init__(self, print_error=None, playwright_available=None, verify_ssl=True)
 
         self.playwright_available = playwright_available
         self.verify_ssl = verify_ssl
+        self._blocked_playwright_request_error = None
 
     def scrape(self, url):
         """
@@ -103,7 +173,12 @@ def scrape(self, url):
         `url` - the URL to scrape.
         """
 
-        if self.playwright_available:
+        error = self.validate_url(url)
+        if error:
+            self.print_error(error)
+            return None
+
+        if self.playwright_available and self.can_use_playwright_for_url(url):
             content, mime_type = self.scrape_with_playwright(url)
         else:
             content, mime_type = self.scrape_with_httpx(url)
@@ -121,6 +196,120 @@ def scrape(self, url):
 
         return content
 
+    def validate_url(self, url):
+        """
+        Return an error if the URL targets a private or otherwise non-public network.
+        """
+        try:
+            parsed_url = urlsplit(url)
+        except ValueError as err:
+            return f"Invalid URL {url}: {err}"
+
+        if parsed_url.scheme not in ("http", "https"):
+            return None
+
+        hostname = parsed_url.hostname
+        if not hostname:
+            return f"Invalid URL {url}: missing hostname"
+
+        if hostname.lower().rstrip(".") in blocked_scrape_hostnames:
+            return f"Blocked scraping private or metadata URL: {url}"
+
+        try:
+            port = parsed_url.port
+        except ValueError as err:
+            return f"Invalid URL {url}: {err}"
+
+        _addresses, error = self.get_safe_addresses(hostname, port, parsed_url.scheme, url)
+        if error:
+            return error
+
+    def get_safe_addresses(self, hostname, port, scheme, display_url):
+        if hostname.lower().rstrip(".") in blocked_scrape_hostnames:
+            return [], f"Blocked scraping private or metadata URL: {display_url}"
+
+        addresses, error = self.resolve_url_addresses(hostname, port, scheme)
+        if error:
+            return [], error
+
+        if not addresses:
+            return [], f"Unable to resolve URL host {hostname}: no addresses found"
+
+        for _family, _socktype, _proto, _sockaddr, ip in addresses:
+            if not self.is_public_unicast_address(ip):
+                return [], f"Blocked scraping private or metadata URL: {display_url}"
+
+        return addresses, None
+
+    def is_public_unicast_address(self, ip):
+        return (
+            ip.is_global
+            and not ip.is_multicast
+            and not ip.is_reserved
+            and not ip.is_unspecified
+        )
+
+    def resolve_url_addresses(self, hostname, port, scheme):
+        if port is None:
+            port = 443 if scheme == "https" else 80
+
+        try:
+            infos = socket.getaddrinfo(hostname, port, type=socket.SOCK_STREAM)
+        except OSError as err:
+            return [], f"Unable to resolve URL host {hostname}: {err}"
+
+        addresses = []
+        for info in infos:
+            address = info[4][0]
+            try:
+                ip = ipaddress.ip_address(address)
+            except ValueError:
+                continue
+            addresses.append((info[0], info[1], info[2], info[4], ip))
+
+        return addresses, None
+
+    def can_use_playwright_for_url(self, url):
+        try:
+            parsed_url = urlsplit(url)
+        except ValueError:
+            return False
+
+        if parsed_url.scheme not in ("http", "https") or not parsed_url.hostname:
+            return False
+
+        try:
+            ipaddress.ip_address(parsed_url.hostname)
+        except ValueError:
+            return False
+        return True
+
+    def validate_playwright_url(self, url):
+        error = self.validate_url(url)
+        if error:
+            return error
+
+        try:
+            parsed_url = urlsplit(url)
+        except ValueError as err:
+            return f"Invalid URL {url}: {err}"
+
+        if parsed_url.scheme in ("http", "https"):
+            try:
+                ipaddress.ip_address(parsed_url.hostname)
+            except ValueError:
+                return f"Blocked scraping browser URL that requires DNS resolution: {url}"
+
+    def route_scrape_request(self, route):
+        error = self.validate_playwright_url(route.request.url)
+        if error:
+            if not self._blocked_playwright_request_error:
+                self._blocked_playwright_request_error = error
+            route.abort()
+            return
+
+        route.continue_()
+
     def looks_like_html(self, content):
         """
         Check if the content looks like HTML.
@@ -157,6 +346,8 @@ def scrape_with_playwright(self, url):
             try:
                 context = browser.new_context(ignore_https_errors=not self.verify_ssl)
                 page = context.new_page()
+                self._blocked_playwright_request_error = None
+                page.route("**/*", self.route_scrape_request)
 
                 user_agent = page.evaluate("navigator.userAgent")
                 user_agent = user_agent.replace("Headless", "")
@@ -172,7 +363,10 @@ def scrape_with_playwright(self, url):
                     print(f"Page didn't quiesce, scraping content anyway: {url}")
                     response = None
                 except PlaywrightError as e:
-                    self.print_error(f"Error navigating to {url}: {str(e)}")
+                    if self._blocked_playwright_request_error:
+                        self.print_error(self._blocked_playwright_request_error)
+                    else:
+                        self.print_error(f"Error navigating to {url}: {str(e)}")
                     return None, None
 
                 try:
@@ -196,12 +390,32 @@ def scrape_with_httpx(self, url):
 
         headers = {"User-Agent": f"Mozilla./5.0 ({aider_user_agent})"}
         try:
+            transport = httpx.HTTPTransport(verify=self.verify_ssl, trust_env=False)
+            transport._pool._network_backend = ScrapeNetworkBackend(self)
             with httpx.Client(
-                headers=headers, verify=self.verify_ssl, follow_redirects=True
+                headers=headers,
+                follow_redirects=False,
+                transport=transport,
+                trust_env=False,
             ) as client:
-                response = client.get(url)
-                response.raise_for_status()
-                return response.text, response.headers.get("content-type", "").split(";")[0]
+                for _ in range(20):
+                    error = self.validate_url(url)
+                    if error:
+                        self.print_error(error)
+                        return None, None
+
+                    response = client.get(url)
+                    if response.is_redirect:
+                        location = response.headers.get("location")
+                        if not location:
+                            response.raise_for_status()
+                        url = urljoin(str(response.url), location)
+                        continue
+
+                    response.raise_for_status()
+                    return response.text, response.headers.get("content-type", "").split(";")[0]
+
+                self.print_error(f"Too many redirects while scraping {url}")
         except httpx.HTTPError as http_err:
             self.print_error(f"HTTP error occurred: {http_err}")
         except Exception as err:

@@ -30,7 +30,7 @@ def fake_httpx(url):
         return "plain text", "text/plain"
 
     scraper.scrape_with_httpx = fake_httpx
-    content = scraper.scrape("http://example.com")
+    content = scraper.scrape("http://93.184.216.34")
     assert content == "plain text"
     assert called["called"]
 
@@ -47,7 +47,7 @@ def fake_playwright(url):
         return "<html>hi</html>", "text/html"
 
     scraper.scrape_with_playwright = fake_playwright
-    content = scraper.scrape("http://example.com")
+    content = scraper.scrape("http://93.184.216.34")
     assert content.startswith("hi") or "<html>" in content
     assert called["called"]
 

@@ -1,10 +1,14 @@
+import socket
 import time
 import unittest
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
+
+import httpcore
+import httpx
 
 from aider.commands import Commands
 from aider.io import InputOutput
-from aider.scrape import Scraper
+from aider.scrape import ScrapeNetworkBackend, Scraper
 
 
 class TestScrape(unittest.TestCase):
@@ -112,22 +116,22 @@ def mock_content():
         scraper.scrape_with_playwright.return_value = (None, None)
 
         # Call the scrape method
-        result = scraper.scrape("https://example.com")
+        result = scraper.scrape("https://93.184.216.34")
 
         # Assert that the result is None
         self.assertIsNone(result)
 
         # Assert that print_error was called with the expected error message
         mock_print_error.assert_called_once_with(
-            "Failed to retrieve content from https://example.com"
+            "Failed to retrieve content from https://93.184.216.34"
         )
 
         # Reset the mock
         mock_print_error.reset_mock()
 
         # Test with a different return value
         scraper.scrape_with_playwright.return_value = ("Some content", "text/html")
-        result = scraper.scrape("https://example.com")
+        result = scraper.scrape("https://93.184.216.34")
 
         # Assert that the result is not None
         self.assertIsNotNone(result)
@@ -144,7 +148,7 @@ def test_scrape_text_plain(self):
         scraper.scrape_with_playwright = MagicMock(return_value=(plain_text, "text/plain"))
 
         # Call the scrape method
-        result = scraper.scrape("https://example.com")
+        result = scraper.scrape("https://93.184.216.34")
 
         # Assert that the result is the same as the input plain text
         self.assertEqual(result, plain_text)
@@ -162,14 +166,272 @@ def test_scrape_text_html(self):
         scraper.html_to_markdown = MagicMock(return_value=expected_markdown)
 
         # Call the scrape method
-        result = scraper.scrape("https://example.com")
+        result = scraper.scrape("https://93.184.216.34")
 
         # Assert that the result is the expected markdown
         self.assertEqual(result, expected_markdown)
 
         # Assert that html_to_markdown was called with the HTML content
         scraper.html_to_markdown.assert_called_once_with(html_content)
 
+    def test_scrape_blocks_private_urls_before_httpx(self):
+        private_urls = [
+            "http://10.0.0.1/",
+            "http://127.0.0.1/",
+            "http://169.254.169.254/latest/meta-data/iam/security-credentials/",
+            "http://[::1]/",
+            "http://[fd00:ec2::254]/",
+        ]
+
+        for url in private_urls:
+            with self.subTest(url=url):
+                mock_print_error = MagicMock()
+                scraper = Scraper(print_error=mock_print_error, playwright_available=False)
+                scraper.scrape_with_httpx = MagicMock()
+
+                result = scraper.scrape(url)
+
+                self.assertIsNone(result)
+                scraper.scrape_with_httpx.assert_not_called()
+                mock_print_error.assert_called_once_with(
+                    f"Blocked scraping private or metadata URL: {url}"
+                )
+
+    def test_scrape_blocks_private_urls_before_playwright(self):
+        mock_print_error = MagicMock()
+        scraper = Scraper(print_error=mock_print_error, playwright_available=True)
+        scraper.scrape_with_playwright = MagicMock()
+
+        result = scraper.scrape("http://192.168.1.10/docs")
+
+        self.assertIsNone(result)
+        scraper.scrape_with_playwright.assert_not_called()
+        mock_print_error.assert_called_once_with(
+            "Blocked scraping private or metadata URL: http://192.168.1.10/docs"
+        )
+
+    @patch("aider.scrape.socket.getaddrinfo")
+    def test_scrape_blocks_metadata_hostname_without_dns(self, mock_getaddrinfo):
+        mock_print_error = MagicMock()
+        scraper = Scraper(print_error=mock_print_error, playwright_available=False)
+        scraper.scrape_with_httpx = MagicMock()
+
+        result = scraper.scrape("http://metadata.google.internal/computeMetadata/v1/")
+
+        self.assertIsNone(result)
+        mock_getaddrinfo.assert_not_called()
+        scraper.scrape_with_httpx.assert_not_called()
+        mock_print_error.assert_called_once_with(
+            "Blocked scraping private or metadata URL: "
+            "http://metadata.google.internal/computeMetadata/v1/"
+        )
+
+    @patch(
+        "aider.scrape.socket.getaddrinfo",
+        return_value=[(None, None, None, None, ("10.0.0.5", 80))],
+    )
+    def test_scrape_blocks_hostnames_that_resolve_private(self, mock_getaddrinfo):
+        mock_print_error = MagicMock()
+        scraper = Scraper(print_error=mock_print_error, playwright_available=False)
+        scraper.scrape_with_httpx = MagicMock()
+
+        result = scraper.scrape("https://docs.example.test/path")
+
+        self.assertIsNone(result)
+        mock_getaddrinfo.assert_called_once_with(
+            "docs.example.test", 443, type=socket.SOCK_STREAM
+        )
+        scraper.scrape_with_httpx.assert_not_called()
+        mock_print_error.assert_called_once_with(
+            "Blocked scraping private or metadata URL: https://docs.example.test/path"
+        )
+
+    @patch(
+        "aider.scrape.socket.getaddrinfo",
+        return_value=[(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("224.0.0.1", 443))],
+    )
+    def test_scrape_blocks_hostnames_that_resolve_multicast(self, mock_getaddrinfo):
+        mock_print_error = MagicMock()
+        scraper = Scraper(print_error=mock_print_error, playwright_available=False)
+        scraper.scrape_with_httpx = MagicMock()
+
+        result = scraper.scrape("https://docs.example.test/path")
+
+        self.assertIsNone(result)
+        mock_getaddrinfo.assert_called_once_with(
+            "docs.example.test", 443, type=socket.SOCK_STREAM
+        )
+        scraper.scrape_with_httpx.assert_not_called()
+        mock_print_error.assert_called_once_with(
+            "Blocked scraping private or metadata URL: https://docs.example.test/path"
+        )
+
+    @patch("aider.scrape.socket.getaddrinfo", side_effect=socket.gaierror("mock failure"))
+    def test_scrape_blocks_unresolved_hostname_before_httpx(self, mock_getaddrinfo):
+        mock_print_error = MagicMock()
+        scraper = Scraper(print_error=mock_print_error, playwright_available=False)
+        scraper.scrape_with_httpx = MagicMock()
+
+        result = scraper.scrape("https://docs.example.test/path")
+
+        self.assertIsNone(result)
+        mock_getaddrinfo.assert_called_once_with(
+            "docs.example.test", 443, type=socket.SOCK_STREAM
+        )
+        scraper.scrape_with_httpx.assert_not_called()
+        mock_print_error.assert_called_once_with(
+            "Unable to resolve URL host docs.example.test: mock failure"
+        )
+
+    def test_scrape_allows_public_ip_before_httpx(self):
+        mock_print_error = MagicMock()
+        scraper = Scraper(print_error=mock_print_error, playwright_available=False)
+        scraper.scrape_with_httpx = MagicMock(return_value=("public content", "text/plain"))
+
+        result = scraper.scrape("https://93.184.216.34/")
+
+        self.assertEqual(result, "public content")
+        scraper.scrape_with_httpx.assert_called_once_with("https://93.184.216.34/")
+        mock_print_error.assert_not_called()
+
+    def test_scrape_uses_httpx_for_hostnames_even_with_playwright(self):
+        mock_print_error = MagicMock()
+        scraper = Scraper(print_error=mock_print_error, playwright_available=True)
+        scraper.scrape_with_playwright = MagicMock()
+        scraper.scrape_with_httpx = MagicMock(return_value=("public content", "text/plain"))
+
+        with patch(
+            "aider.scrape.socket.getaddrinfo",
+            return_value=[(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 443))],
+        ):
+            result = scraper.scrape("https://docs.example.test/")
+
+        self.assertEqual(result, "public content")
+        scraper.scrape_with_playwright.assert_not_called()
+        scraper.scrape_with_httpx.assert_called_once_with("https://docs.example.test/")
+        mock_print_error.assert_not_called()
+
+    def test_httpx_backend_blocks_dns_rebinding_at_connect_time(self):
+        mock_print_error = MagicMock()
+        scraper = Scraper(print_error=mock_print_error, playwright_available=False)
+        safe_addresses = [
+            (socket.AF_INET, socket.SOCK_STREAM, 6, ("93.184.216.34", 443), "93.184.216.34")
+        ]
+        scraper.get_safe_addresses = MagicMock(
+            side_effect=[
+                (safe_addresses, None),
+                ([], "Blocked scraping private or metadata URL: docs.example.test:443"),
+            ]
+        )
+
+        result = scraper.scrape_with_httpx("https://docs.example.test/")
+
+        self.assertEqual(scraper.get_safe_addresses.call_count, 2)
+        self.assertIsNone(result[0])
+        mock_print_error.assert_called_once_with(
+            "HTTP error occurred: Blocked scraping private or metadata URL: "
+            "docs.example.test:443"
+        )
+
+    def test_httpx_network_backend_connects_to_validated_address(self):
+        scraper = Scraper(print_error=MagicMock(), playwright_available=False)
+        backend = ScrapeNetworkBackend(scraper)
+        sock = MagicMock()
+        getaddrinfo_result = [
+            (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 443))
+        ]
+
+        with patch("aider.scrape.socket.getaddrinfo", return_value=getaddrinfo_result):
+            with patch("aider.scrape.socket.socket", return_value=sock):
+                backend.connect_tcp("docs.example.test", 443)
+
+        sock.connect.assert_called_once_with(("93.184.216.34", 443))
+
+    @patch(
+        "aider.scrape.socket.getaddrinfo",
+        return_value=[(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("10.0.0.5", 443))],
+    )
+    def test_httpx_network_backend_rejects_private_connect_address(self, mock_getaddrinfo):
+        scraper = Scraper(print_error=MagicMock(), playwright_available=False)
+        backend = ScrapeNetworkBackend(scraper)
+
+        with patch("aider.scrape.socket.socket") as mock_socket:
+            with self.assertRaises(httpcore.ConnectError):
+                backend.connect_tcp("docs.example.test", 443)
+
+        mock_getaddrinfo.assert_called_once_with(
+            "docs.example.test", 443, type=socket.SOCK_STREAM
+        )
+        mock_socket.assert_not_called()
+
+    @patch("httpx.Client")
+    def test_scrape_blocks_private_redirect_before_following(self, mock_client_class):
+        mock_print_error = MagicMock()
+        client = MagicMock()
+        client.__enter__.return_value = client
+        client.__exit__.return_value = None
+        mock_client_class.return_value = client
+        client.get.return_value = httpx.Response(
+            302,
+            headers={"location": "http://169.254.169.254/latest/meta-data/"},
+            request=httpx.Request("GET", "https://93.184.216.34/start"),
+        )
+        scraper = Scraper(print_error=mock_print_error, playwright_available=False)
+
+        result = scraper.scrape("https://93.184.216.34/start")
+
+        self.assertIsNone(result)
+        client.get.assert_called_once_with("https://93.184.216.34/start")
+        mock_print_error.assert_any_call(
+            "Blocked scraping private or metadata URL: "
+            "http://169.254.169.254/latest/meta-data/"
+        )
+
+    def test_playwright_route_blocks_private_request(self):
+        scraper = Scraper(print_error=MagicMock(), playwright_available=True)
+        route = MagicMock()
+        route.request.url = "http://169.254.169.254/latest/meta-data/"
+
+        scraper.route_scrape_request(route)
+
+        route.abort.assert_called_once()
+        route.continue_.assert_not_called()
+        self.assertEqual(
+            scraper._blocked_playwright_request_error,
+            "Blocked scraping private or metadata URL: "
+            "http://169.254.169.254/latest/meta-data/",
+        )
+
+    def test_playwright_route_allows_public_request(self):
+        scraper = Scraper(print_error=MagicMock(), playwright_available=True)
+        route = MagicMock()
+        route.request.url = "https://93.184.216.34/"
+
+        scraper.route_scrape_request(route)
+
+        route.continue_.assert_called_once()
+        route.abort.assert_not_called()
+        self.assertIsNone(scraper._blocked_playwright_request_error)
+
+    def test_playwright_route_blocks_hostname_request(self):
+        scraper = Scraper(print_error=MagicMock(), playwright_available=True)
+        route = MagicMock()
+        route.request.url = "https://docs.example.test/"
+
+        with patch(
+            "aider.scrape.socket.getaddrinfo",
+            return_value=[(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 443))],
+        ):
+            scraper.route_scrape_request(route)
+
+        route.abort.assert_called_once()
+        route.continue_.assert_not_called()
+        self.assertEqual(
+            scraper._blocked_playwright_request_error,
+            "Blocked scraping browser URL that requires DNS resolution: "
+            "https://docs.example.test/",
+        )
+
 
 if __name__ == "__main__":
     unittest.main()