Close proxy connections after timeout

scrapy/core/downloader/handlers/http11.py

@@ -188,7 +188,9 @@ class _TunnelingTCP4ClientEndpoint(TCP4ClientEndpoint):
     ):
         proxyHost, proxyPort, self._proxyAuthHeader = proxyConf
         super().__init__(reactor, proxyHost, proxyPort, timeout, bindAddress)
-        self._tunnelReadyDeferred: Deferred[Protocol] = Deferred()
+        self._tunnelReadyDeferred: Deferred[Protocol] = Deferred(self._cancelTunnel)
+        self._connectDeferred: Deferred[Protocol] | None = None
+        self._protocol: Protocol | None = None
         self._tunneledHost: str = host
         self._tunneledPort: int = port
         self._contextFactory: IPolicyForHTTPS = contextFactory
@@ -211,6 +213,7 @@ class _TunnelingTCP4ClientEndpoint(TCP4ClientEndpoint):
         created, notifies the client that we are ready to send requests. If not
         raises a TunnelError.
         """
+        assert self._protocol
         assert self._protocol.transport
         self._connectBuffer += data
         # make sure that enough (all) bytes are consumed
@@ -250,9 +253,23 @@ class _TunnelingTCP4ClientEndpoint(TCP4ClientEndpoint):
         """Propagates the errback to the appropriate deferred."""
         self._tunnelReadyDeferred.errback(reason)
 
+    def _cancelTunnel(self, deferred: Deferred[Protocol]) -> None:
+        """Tear down the connection to the proxy when the deferred returned by
+        :meth:`connect` is cancelled, e.g. on a download timeout."""
+        if self._protocol is not None:
+            # The proxy connection is up; drop it, and stop intercepting data
+            # in case the proxy answers the CONNECT after this point.
+            self._protocol.dataReceived = self._protocolDataReceived  # type: ignore[method-assign]
+            assert self._protocol.transport
+            self._protocol.transport.loseConnection()
+        elif self._connectDeferred is not None:
+            # Still connecting to the proxy; stop the connection attempt. This
+            # errbacks _tunnelReadyDeferred through connectFailed().
+            self._connectDeferred.cancel()
+
     def connect(self, protocolFactory: Factory) -> Deferred[Protocol]:
         self._protocolFactory = protocolFactory
-        connectDeferred = super().connect(protocolFactory)
+        self._connectDeferred = connectDeferred = super().connect(protocolFactory)
         connectDeferred.addCallback(self.requestTunnel)
         connectDeferred.addErrback(self.connectFailed)
         return self._tunnelReadyDeferred

tests/mockserver/proxy_stalling.py

@@ -0,0 +1,110 @@
+"""An HTTP proxy that never answers ``CONNECT`` requests."""
+
+from __future__ import annotations
+
+import socket
+import threading
+from typing import TYPE_CHECKING
+
+from tests.utils import async_sleep
+
+if TYPE_CHECKING:
+    # typing.Self requires Python 3.11
+    from typing_extensions import Self
+
+
+class StallingProxyConnection:
+    """A connection received by :class:`StallingProxy`."""
+
+    def __init__(self, sock: socket.socket) -> None:
+        self._socket = sock
+        #: Bytes read from the client, i.e. its ``CONNECT`` request.
+        self.request: bytes = b""
+        #: Whether the client has closed the connection.
+        self.closed: bool = False
+
+    def close(self) -> None:
+        self._socket.close()
+
+    def _read(self) -> None:
+        while True:
+            try:
+                data = self._socket.recv(4096)
+            except OSError:
+                break
+            if not data:  # the client closed the connection
+                break
+            self.request += data
+        self.closed = True
+
+    async def wait_closed(self, timeout: float = 10.0) -> bool:
+        """Wait for the client to close the connection, and return whether it
+        did so within *timeout* seconds."""
+        for _ in range(int(timeout / 0.05)):
+            if self.closed:
+                return True
+            await async_sleep(0.05)
+        return self.closed
+
+
+class StallingProxy:
+    """An HTTP proxy that reads requests and never answers them, so that
+    downloads through it can only finish with a timeout.
+
+    It is meant to be used with HTTPS targets, to stall the ``CONNECT``
+    handshake, and it can tell whether the client closed the connection::
+
+        with StallingProxy() as proxy:
+            request = Request(
+                "https://example.com", meta={"proxy": proxy.url, ...}
+            )
+            ...
+            connection = await proxy.wait_for_connection()
+            assert await connection.wait_closed()
+
+    It uses blocking sockets on separate threads, instead of the Twisted-based
+    approach of the other mock servers, so that it works with any reactor, and
+    so that a client connection being closed is detected even while the reactor
+    is busy.
+    """
+
+    def __init__(self) -> None:
+        self._socket = socket.socket()
+        self._socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        self._socket.bind(("127.0.0.1", 0))
+        self._socket.listen(10)
+        self.host, self.port = self._socket.getsockname()
+        self.connections: list[StallingProxyConnection] = []
+
+    def __enter__(self) -> Self:
+        threading.Thread(target=self._accept, daemon=True).start()
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback) -> None:
+        self._socket.close()
+        for connection in self.connections:
+            connection.close()
+
+    @property
+    def url(self) -> str:
+        return f"http://{self.host}:{self.port}"
+
+    def _accept(self) -> None:
+        while True:
+            try:
+                sock, _ = self._socket.accept()
+            except OSError:  # the proxy was stopped
+                return
+            connection = StallingProxyConnection(sock)
+            self.connections.append(connection)
+            threading.Thread(target=connection._read, daemon=True).start()
+
+    async def wait_for_connection(
+        self, timeout: float = 10.0
+    ) -> StallingProxyConnection:
+        """Wait for a client connection with a complete request, and return it."""
+        for _ in range(int(timeout / 0.05)):
+            if self.connections and b"\r\n\r\n" in self.connections[0].request:
+                return self.connections[0]
+            await async_sleep(0.05)
+        raise AssertionError(f"No request reached the proxy in {timeout} seconds")

tests/test_downloader_handler_twisted_http11.py

@@ -3,14 +3,23 @@
 from __future__ import annotations
 
 import sys
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, cast
 
 import pytest
+from twisted.internet.error import ConnectingCancelledError
+from twisted.internet.protocol import Factory, Protocol
+from twisted.internet.testing import MemoryReactorClock
+from twisted.python.failure import Failure
 
 from scrapy import Spider
-from scrapy.core.downloader.handlers.http11 import HTTP11DownloadHandler
+from scrapy.core.downloader.contextfactory import _load_context_factory_from_settings
+from scrapy.core.downloader.handlers.http11 import (
+    HTTP11DownloadHandler,
+    _TunnelingTCP4ClientEndpoint,
+)
 from scrapy.crawler import Crawler
 from scrapy.exceptions import NotConfigured
+from scrapy.utils.test import get_crawler
 from tests.utils.bases.download_handlers_http import (
     TestHttpBase,
     TestHttpProxyBase,
@@ -27,6 +36,8 @@ from tests.utils.bases.download_handlers_http import (
 )
 
 if TYPE_CHECKING:
+    from twisted.internet.base import ReactorBase
+
     from scrapy.core.downloader.handlers import DownloadHandlerProtocol
 
 
@@ -54,6 +65,35 @@ def test_not_configured_without_reactor() -> None:
         HTTP11DownloadHandler.from_crawler(crawler)
 
 
+def test_tunneling_cancelled_before_connected() -> None:
+    """Cancelling a tunneled download before the connection to the proxy is
+    established stops the connection attempt.
+
+    A download cannot reach this scenario, because the connection timeout of
+    the endpoint, which uses the same value as the download timeout, always
+    fires earlier than the cancellation. See also
+    ``test_download_with_proxy_stalled_connect``, which covers cancelling once
+    the connection to the proxy is established.
+    """
+    reactor = MemoryReactorClock()
+    endpoint = _TunnelingTCP4ClientEndpoint(
+        reactor=cast("ReactorBase", reactor),
+        host="example.com",
+        port=443,
+        proxyConf=("127.0.0.1", 8080, None),
+        contextFactory=_load_context_factory_from_settings(get_crawler()),
+        timeout=30,
+    )
+    results: list[Protocol | Failure] = []
+    endpoint.connect(Factory()).addBoth(results.append)
+
+    endpoint._tunnelReadyDeferred.cancel()
+
+    assert reactor.connectors[0].stoppedConnecting
+    assert isinstance(results[0], Failure)
+    assert isinstance(results[0].value, ConnectingCancelledError)
+
+
 class TestHttp(HTTP11DownloadHandlerMixin, TestHttpBase):
     pass
 

tests/utils/bases/download_handlers_http.py

@@ -42,6 +42,7 @@ from scrapy.utils.test import get_crawler
 from tests import NON_EXISTING_RESOLVABLE
 from tests.mockserver.mitm_proxy import wrong_credentials
 from tests.mockserver.proxy_echo import ProxyEchoMockServer
+from tests.mockserver.proxy_stalling import StallingProxy
 from tests.mockserver.simple_https import SimpleMockServer
 from tests.spiders import (
     BytesReceivedCallbackSpider,
@@ -1316,6 +1317,26 @@ class TestHttpProxyBase(ABC):
                 await download_handler.download_request(request)
         assert domain in str(exc_info.value)
 
+    @coroutine_test
+    async def test_download_with_proxy_stalled_connect(self) -> None:
+        """A download that times out while the proxy is being asked to open a
+        tunnel must not leave the connection to the proxy open."""
+        if self.is_secure:
+            pytest.skip("The stalling proxy only speaks plain HTTP")
+        with StallingProxy() as proxy:
+            request = Request(
+                "https://example.com",
+                meta={"proxy": proxy.url, "download_timeout": 0.2},
+            )
+            async with self.get_dh() as download_handler:
+                with pytest.raises(DownloadTimeoutError):
+                    await download_handler.download_request(request)
+                connection = await proxy.wait_for_connection()
+                assert connection.request.startswith(b"CONNECT example.com:443")
+                assert await connection.wait_closed(), (
+                    "The connection to the proxy was left open"
+                )
+
     @coroutine_test
     async def test_download_with_proxy_without_http_scheme(
         self, proxy_mockserver: ProxyEchoMockServer