diff --git a/scrapy/core/downloader/handlers/http11.py b/scrapy/core/downloader/handlers/http11.py index c288ae792..3108c5b7e 100644 --- a/scrapy/core/downloader/handlers/http11.py +++ b/scrapy/core/downloader/handlers/http11.py @@ -196,7 +196,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 @@ -219,6 +221,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 @@ -258,9 +261,24 @@ 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() + else: + # Still connecting to the proxy; stop the connection attempt. This + # errbacks _tunnelReadyDeferred through connectFailed(). + assert self._connectDeferred is not None + 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 diff --git a/tests/mockserver/proxy_stalling.py b/tests/mockserver/proxy_stalling.py new file mode 100644 index 000000000..692a20cb4 --- /dev/null +++ b/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 scrapy.utils.asyncio import 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 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 sleep(0.05) + raise AssertionError(f"No request reached the proxy in {timeout} seconds") diff --git a/tests/test_downloader_handler_twisted_http11.py b/tests/test_downloader_handler_twisted_http11.py index db353750d..752337875 100644 --- a/tests/test_downloader_handler_twisted_http11.py +++ b/tests/test_downloader_handler_twisted_http11.py @@ -3,15 +3,24 @@ 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.misc import build_from_crawler +from scrapy.utils.test import get_crawler from tests.utils.bases.download_handlers_http import ( TestHttpBase, TestHttpProxyBase, @@ -28,6 +37,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 @@ -55,6 +66,35 @@ def test_not_configured_without_reactor() -> None: build_from_crawler(HTTP11DownloadHandler, 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 diff --git a/tests/utils/bases/download_handlers_http.py b/tests/utils/bases/download_handlers_http.py index abf153817..474c6ac8e 100644 --- a/tests/utils/bases/download_handlers_http.py +++ b/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, @@ -1354,6 +1355,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