scrapy/scrapy/core/http2/protocol.py

from __future__ import annotations

import ipaddress
import itertools
import logging
from collections import deque
from typing import TYPE_CHECKING, Any, Optional, Union

from h2.config import H2Configuration
from h2.connection import H2Connection
from h2.errors import ErrorCodes
from h2.events import (
    ConnectionTerminated,
    DataReceived,
    Event,
    ResponseReceived,
    SettingsAcknowledged,
    StreamEnded,
    StreamReset,
    UnknownFrameReceived,
    WindowUpdated,
)
from h2.exceptions import FrameTooLargeError, H2Error
from twisted.internet.error import TimeoutError
from twisted.internet.interfaces import (
    IAddress,
    IHandshakeListener,
    IProtocolNegotiationFactory,
)
from twisted.internet.protocol import Factory, Protocol, connectionDone
from twisted.internet.ssl import Certificate
from twisted.protocols.policies import TimeoutMixin
from zope.interface import implementer

from scrapy.core.http2.stream import Stream, StreamCloseReason
from scrapy.http import Request, Response

if TYPE_CHECKING:
    from ipaddress import IPv4Address, IPv6Address

    from twisted.internet.defer import Deferred
    from twisted.python.failure import Failure
    from twisted.web.client import URI

    from scrapy.settings import Settings
    from scrapy.spiders import Spider


logger = logging.getLogger(__name__)


PROTOCOL_NAME = b"h2"


class InvalidNegotiatedProtocol(H2Error):
    def __init__(self, negotiated_protocol: bytes) -> None:
        self.negotiated_protocol = negotiated_protocol

    def __str__(self) -> str:
        return f"Expected {PROTOCOL_NAME!r}, received {self.negotiated_protocol!r}"


class RemoteTerminatedConnection(H2Error):
    def __init__(
        self,
        remote_ip_address: Optional[Union[IPv4Address, IPv6Address]],
        event: ConnectionTerminated,
    ) -> None:
        self.remote_ip_address = remote_ip_address
        self.terminate_event = event

    def __str__(self) -> str:
        return f"Received GOAWAY frame from {self.remote_ip_address!r}"


class MethodNotAllowed405(H2Error):
    def __init__(
        self, remote_ip_address: Optional[Union[IPv4Address, IPv6Address]]
    ) -> None:
        self.remote_ip_address = remote_ip_address

    def __str__(self) -> str:
        return f"Received 'HTTP/2.0 405 Method Not Allowed' from {self.remote_ip_address!r}"


@implementer(IHandshakeListener)
class H2ClientProtocol(Protocol, TimeoutMixin):
    IDLE_TIMEOUT = 240

    def __init__(
        self,
        uri: URI,
        settings: Settings,
        conn_lost_deferred: Deferred[list[BaseException]],
    ) -> None:
        """
        Arguments:
            uri -- URI of the base url to which HTTP/2 Connection will be made.
                uri is used to verify that incoming client requests have correct
                base URL.
            settings -- Scrapy project settings
            conn_lost_deferred -- Deferred fires with the reason: Failure to notify
                that connection was lost
        """
        self._conn_lost_deferred: Deferred[list[BaseException]] = conn_lost_deferred

        config = H2Configuration(client_side=True, header_encoding="utf-8")
        self.conn = H2Connection(config=config)

        # ID of the next request stream
        # Following the convention - 'Streams initiated by a client MUST
        # use odd-numbered stream identifiers' (RFC 7540 - Section 5.1.1)
        self._stream_id_generator = itertools.count(start=1, step=2)

        # Streams are stored in a dictionary keyed off their stream IDs
        self.streams: dict[int, Stream] = {}

        # If requests are received before connection is made we keep
        # all requests in a pool and send them as the connection is made
        self._pending_request_stream_pool: deque[Stream] = deque()

        # Save an instance of errors raised which lead to losing the connection
        # We pass these instances to the streams ResponseFailed() failure
        self._conn_lost_errors: list[BaseException] = []

        # Some meta data of this connection
        # initialized when connection is successfully made
        self.metadata: dict[str, Any] = {
            # Peer certificate instance
            "certificate": None,
            # Address of the server we are connected to which
            # is updated when HTTP/2 connection is  made successfully
            "ip_address": None,
            # URI of the peer HTTP/2 connection is made
            "uri": uri,
            # Both ip_address and uri are used by the Stream before
            # initiating the request to verify that the base address
            # Variables taken from Project Settings
            "default_download_maxsize": settings.getint("DOWNLOAD_MAXSIZE"),
            "default_download_warnsize": settings.getint("DOWNLOAD_WARNSIZE"),
            # Counter to keep track of opened streams. This counter
            # is used to make sure that not more than MAX_CONCURRENT_STREAMS
            # streams are opened which leads to ProtocolError
            # We use simple FIFO policy to handle pending requests
            "active_streams": 0,
            # Flag to keep track if settings were acknowledged by the remote
            # This ensures that we have established a HTTP/2 connection
            "settings_acknowledged": False,
        }

    @property
    def h2_connected(self) -> bool:
        """Boolean to keep track of the connection status.
        This is used while initiating pending streams to make sure
        that we initiate stream only during active HTTP/2 Connection
        """
        assert self.transport is not None  # typing
        return bool(self.transport.connected) and self.metadata["settings_acknowledged"]

    @property
    def allowed_max_concurrent_streams(self) -> int:
        """We keep total two streams for client (sending data) and
        server side (receiving data) for a single request. To be safe
        we choose the minimum. Since this value can change in event
        RemoteSettingsChanged we make variable a property.
        """
        return min(
            self.conn.local_settings.max_concurrent_streams,
            self.conn.remote_settings.max_concurrent_streams,
        )

    def _send_pending_requests(self) -> None:
        """Initiate all pending requests from the deque following FIFO
        We make sure that at any time {allowed_max_concurrent_streams}
        streams are active.
        """
        while (
            self._pending_request_stream_pool
            and self.metadata["active_streams"] < self.allowed_max_concurrent_streams
            and self.h2_connected
        ):
            self.metadata["active_streams"] += 1
            stream = self._pending_request_stream_pool.popleft()
            stream.initiate_request()
            self._write_to_transport()

    def pop_stream(self, stream_id: int) -> Stream:
        """Perform cleanup when a stream is closed"""
        stream = self.streams.pop(stream_id)
        self.metadata["active_streams"] -= 1
        self._send_pending_requests()
        return stream

    def _new_stream(self, request: Request, spider: Spider) -> Stream:
        """Instantiates a new Stream object"""
        stream = Stream(
            stream_id=next(self._stream_id_generator),
            request=request,
            protocol=self,
            download_maxsize=getattr(
                spider, "download_maxsize", self.metadata["default_download_maxsize"]
            ),
            download_warnsize=getattr(
                spider, "download_warnsize", self.metadata["default_download_warnsize"]
            ),
        )
        self.streams[stream.stream_id] = stream
        return stream

    def _write_to_transport(self) -> None:
        """Write data to the underlying transport connection
        from the HTTP2 connection instance if any
        """
        assert self.transport is not None  # typing
        # Reset the idle timeout as connection is still actively sending data
        self.resetTimeout()

        data = self.conn.data_to_send()
        self.transport.write(data)

    def request(self, request: Request, spider: Spider) -> Deferred[Response]:
        if not isinstance(request, Request):
            raise TypeError(
                f"Expected scrapy.http.Request, received {request.__class__.__qualname__}"
            )

        stream = self._new_stream(request, spider)
        d: Deferred[Response] = stream.get_response()

        # Add the stream to the request pool
        self._pending_request_stream_pool.append(stream)

        # If we receive a request when connection is idle
        # We need to initiate pending requests
        self._send_pending_requests()
        return d

    def connectionMade(self) -> None:
        """Called by Twisted when the connection is established. We can start
        sending some data now: we should open with the connection preamble.
        """
        # Initialize the timeout
        self.setTimeout(self.IDLE_TIMEOUT)

        assert self.transport is not None  # typing
        destination = self.transport.getPeer()
        self.metadata["ip_address"] = ipaddress.ip_address(destination.host)

        # Initiate H2 Connection
        self.conn.initiate_connection()
        self._write_to_transport()

    def _lose_connection_with_error(self, errors: list[BaseException]) -> None:
        """Helper function to lose the connection with the error sent as a
        reason"""
        self._conn_lost_errors += errors
        assert self.transport is not None  # typing
        self.transport.loseConnection()

    def handshakeCompleted(self) -> None:
        """
        Close the connection if it's not made via the expected protocol
        """
        assert self.transport is not None  # typing
        if (
            self.transport.negotiatedProtocol is not None
            and self.transport.negotiatedProtocol != PROTOCOL_NAME
        ):
            # we have not initiated the connection yet, no need to send a GOAWAY frame to the remote peer
            self._lose_connection_with_error(
                [InvalidNegotiatedProtocol(self.transport.negotiatedProtocol)]
            )

    def _check_received_data(self, data: bytes) -> None:
        """Checks for edge cases where the connection to remote fails
        without raising an appropriate H2Error

        Arguments:
            data -- Data received from the remote
        """
        if data.startswith(b"HTTP/2.0 405 Method Not Allowed"):
            raise MethodNotAllowed405(self.metadata["ip_address"])

    def dataReceived(self, data: bytes) -> None:
        # Reset the idle timeout as connection is still actively receiving data
        self.resetTimeout()

        try:
            self._check_received_data(data)
            events = self.conn.receive_data(data)
            self._handle_events(events)
        except H2Error as e:
            if isinstance(e, FrameTooLargeError):
                # hyper-h2 does not drop the connection in this scenario, we
                # need to abort the connection manually.
                self._conn_lost_errors += [e]
                assert self.transport is not None  # typing
                self.transport.abortConnection()
                return

            # Save this error as ultimately the connection will be dropped
            # internally by hyper-h2. Saved error will be passed to all the streams
            # closed with the connection.
            self._lose_connection_with_error([e])
        finally:
            self._write_to_transport()

    def timeoutConnection(self) -> None:
        """Called when the connection times out.
        We lose the connection with TimeoutError"""

        # Check whether there are open streams. If there are, we're going to
        # want to use the error code PROTOCOL_ERROR. If there aren't, use
        # NO_ERROR.
        if (
            self.conn.open_outbound_streams > 0
            or self.conn.open_inbound_streams > 0
            or self.metadata["active_streams"] > 0
        ):
            error_code = ErrorCodes.PROTOCOL_ERROR
        else:
            error_code = ErrorCodes.NO_ERROR
        self.conn.close_connection(error_code=error_code)
        self._write_to_transport()

        self._lose_connection_with_error(
            [TimeoutError(f"Connection was IDLE for more than {self.IDLE_TIMEOUT}s")]
        )

    def connectionLost(self, reason: Failure = connectionDone) -> None:
        """Called by Twisted when the transport connection is lost.
        No need to write anything to transport here.
        """
        # Cancel the timeout if not done yet
        self.setTimeout(None)

        # Notify the connection pool instance such that no new requests are
        # sent over current connection
        if not reason.check(connectionDone):
            self._conn_lost_errors.append(reason)

        self._conn_lost_deferred.callback(self._conn_lost_errors)

        for stream in self.streams.values():
            if stream.metadata["request_sent"]:
                close_reason = StreamCloseReason.CONNECTION_LOST
            else:
                close_reason = StreamCloseReason.INACTIVE
            stream.close(close_reason, self._conn_lost_errors, from_protocol=True)

        self.metadata["active_streams"] -= len(self.streams)
        self.streams.clear()
        self._pending_request_stream_pool.clear()
        self.conn.close_connection()

    def _handle_events(self, events: list[Event]) -> None:
        """Private method which acts as a bridge between the events
        received from the HTTP/2 data and IH2EventsHandler

        Arguments:
            events -- A list of events that the remote peer triggered by sending data
        """
        for event in events:
            if isinstance(event, ConnectionTerminated):
                self.connection_terminated(event)
            elif isinstance(event, DataReceived):
                self.data_received(event)
            elif isinstance(event, ResponseReceived):
                self.response_received(event)
            elif isinstance(event, StreamEnded):
                self.stream_ended(event)
            elif isinstance(event, StreamReset):
                self.stream_reset(event)
            elif isinstance(event, WindowUpdated):
                self.window_updated(event)
            elif isinstance(event, SettingsAcknowledged):
                self.settings_acknowledged(event)
            elif isinstance(event, UnknownFrameReceived):
                logger.warning("Unknown frame received: %s", event.frame)

    # Event handler functions starts here
    def connection_terminated(self, event: ConnectionTerminated) -> None:
        self._lose_connection_with_error(
            [RemoteTerminatedConnection(self.metadata["ip_address"], event)]
        )

    def data_received(self, event: DataReceived) -> None:
        try:
            stream = self.streams[event.stream_id]
        except KeyError:
            pass  # We ignore server-initiated events
        else:
            stream.receive_data(event.data, event.flow_controlled_length)

    def response_received(self, event: ResponseReceived) -> None:
        try:
            stream = self.streams[event.stream_id]
        except KeyError:
            pass  # We ignore server-initiated events
        else:
            stream.receive_headers(event.headers)

    def settings_acknowledged(self, event: SettingsAcknowledged) -> None:
        self.metadata["settings_acknowledged"] = True

        # Send off all the pending requests as now we have
        # established a proper HTTP/2 connection
        self._send_pending_requests()

        # Update certificate when our HTTP/2 connection is established
        assert self.transport is not None  # typing
        self.metadata["certificate"] = Certificate(self.transport.getPeerCertificate())

    def stream_ended(self, event: StreamEnded) -> None:
        try:
            stream = self.pop_stream(event.stream_id)
        except KeyError:
            pass  # We ignore server-initiated events
        else:
            stream.close(StreamCloseReason.ENDED, from_protocol=True)

    def stream_reset(self, event: StreamReset) -> None:
        try:
            stream = self.pop_stream(event.stream_id)
        except KeyError:
            pass  # We ignore server-initiated events
        else:
            stream.close(StreamCloseReason.RESET, from_protocol=True)

    def window_updated(self, event: WindowUpdated) -> None:
        if event.stream_id != 0:
            self.streams[event.stream_id].receive_window_update()
        else:
            # Send leftover data for all the streams
            for stream in self.streams.values():
                stream.receive_window_update()


@implementer(IProtocolNegotiationFactory)
class H2ClientFactory(Factory):
    def __init__(
        self,
        uri: URI,
        settings: Settings,
        conn_lost_deferred: Deferred[list[BaseException]],
    ) -> None:
        self.uri = uri
        self.settings = settings
        self.conn_lost_deferred = conn_lost_deferred

    def buildProtocol(self, addr: IAddress) -> H2ClientProtocol:
        return H2ClientProtocol(self.uri, self.settings, self.conn_lost_deferred)

    def acceptableProtocols(self) -> list[bytes]:
        return [PROTOCOL_NAME]