from __future__ import annotations import logging from enum import Enum from io import BytesIO from typing import TYPE_CHECKING, Any from h2.errors import ErrorCodes from h2.exceptions import H2Error, ProtocolError, StreamClosedError from twisted.internet.defer import CancelledError, Deferred from twisted.internet.error import ConnectionClosed from twisted.python.failure import Failure from twisted.web.client import ResponseFailed from scrapy.http.headers import Headers from scrapy.responsetypes import responsetypes from scrapy.utils.httpobj import urlparse_cached if TYPE_CHECKING: from hpack import HeaderTuple from scrapy.core.http2.protocol import H2ClientProtocol from scrapy.http import Request, Response logger = logging.getLogger(__name__) class InactiveStreamClosed(ConnectionClosed): """Connection was closed without sending request headers of the stream. This happens when a stream is waiting for other streams to close and connection is lost.""" def __init__(self, request: Request) -> None: self.request = request def __str__(self) -> str: return f"InactiveStreamClosed: Connection was closed without sending the request {self.request!r}" class InvalidHostname(H2Error): def __init__( self, request: Request, expected_hostname: str, expected_netloc: str ) -> None: self.request = request self.expected_hostname = expected_hostname self.expected_netloc = expected_netloc def __str__(self) -> str: return f"InvalidHostname: Expected {self.expected_hostname} or {self.expected_netloc} in {self.request}" class StreamCloseReason(Enum): # Received a StreamEnded event from the remote ENDED = 1 # Received a StreamReset event -- ended abruptly RESET = 2 # Transport connection was lost CONNECTION_LOST = 3 # Expected response body size is more than allowed limit MAXSIZE_EXCEEDED = 4 # Response deferred is cancelled by the client # (happens when client called response_deferred.cancel()) CANCELLED = 5 # Connection lost and the stream was not initiated INACTIVE = 6 # The hostname of the request is not same as of connected peer hostname # As a result sending this request will the end the connection INVALID_HOSTNAME = 7 class Stream: """Represents a single HTTP/2 Stream. Stream is a bidirectional flow of bytes within an established connection, which may carry one or more messages. Handles the transfer of HTTP Headers and Data frames. Role of this class is to 1. Combine all the data frames """ def __init__( self, stream_id: int, request: Request, protocol: H2ClientProtocol, download_maxsize: int = 0, download_warnsize: int = 0, ) -> None: """ Arguments: stream_id -- Unique identifier for the stream within a single HTTP/2 connection request -- The HTTP request associated to the stream protocol -- Parent H2ClientProtocol instance """ self.stream_id: int = stream_id self._request: Request = request self._protocol: H2ClientProtocol = protocol self._download_maxsize = self._request.meta.get( "download_maxsize", download_maxsize ) self._download_warnsize = self._request.meta.get( "download_warnsize", download_warnsize ) # Metadata of an HTTP/2 connection stream # initialized when stream is instantiated self.metadata: dict[str, Any] = { "request_content_length": ( 0 if self._request.body is None else len(self._request.body) ), # Flag to keep track whether the stream has initiated the request "request_sent": False, # Flag to track whether we have logged about exceeding download warnsize "reached_warnsize": False, # Each time we send a data frame, we will decrease value by the amount send. "remaining_content_length": ( 0 if self._request.body is None else len(self._request.body) ), # Flag to keep track whether client (self) have closed this stream "stream_closed_local": False, # Flag to keep track whether the server has closed the stream "stream_closed_server": False, } # Private variable used to build the response # this response is then converted to appropriate Response class # passed to the response deferred callback self._response: dict[str, Any] = { # Data received frame by frame from the server is appended # and passed to the response Deferred when completely received. "body": BytesIO(), # The amount of data received that counts against the # flow control window "flow_controlled_size": 0, # Headers received after sending the request "headers": Headers({}), } def _cancel(_: Any) -> None: # Close this stream as gracefully as possible # If the associated request is initiated we reset this stream # else we directly call close() method if self.metadata["request_sent"]: self.reset_stream(StreamCloseReason.CANCELLED) else: self.close(StreamCloseReason.CANCELLED) self._deferred_response: Deferred[Response] = Deferred(_cancel) def __repr__(self) -> str: return f"Stream(id={self.stream_id!r})" @property def _log_warnsize(self) -> bool: """Checks if we have received data which exceeds the download warnsize and whether we have not already logged about it. Returns: True if both the above conditions hold true False if any of the conditions is false """ content_length_header = int( self._response["headers"].get(b"Content-Length", -1) ) return ( self._download_warnsize and ( self._response["flow_controlled_size"] > self._download_warnsize or content_length_header > self._download_warnsize ) and not self.metadata["reached_warnsize"] ) def get_response(self) -> Deferred[Response]: """Simply return a Deferred which fires when response from the asynchronous request is available """ return self._deferred_response def check_request_url(self) -> bool: # Make sure that we are sending the request to the correct URL url = urlparse_cached(self._request) return ( url.netloc == str(self._protocol.metadata["uri"].host, "utf-8") or url.netloc == str(self._protocol.metadata["uri"].netloc, "utf-8") or url.netloc == f"{self._protocol.metadata['ip_address']}:{self._protocol.metadata['uri'].port}" ) def _get_request_headers(self) -> list[tuple[str, str]]: url = urlparse_cached(self._request) path = url.path if url.query: path += "?" + url.query # This pseudo-header field MUST NOT be empty for "http" or "https" # URIs; "http" or "https" URIs that do not contain a path component # MUST include a value of '/'. The exception to this rule is an # OPTIONS request for an "http" or "https" URI that does not include # a path component; these MUST include a ":path" pseudo-header field # with a value of '*' (refer RFC 7540 - Section 8.1.2.3) if not path: path = "*" if self._request.method == "OPTIONS" else "/" # Make sure pseudo-headers comes before all the other headers headers = [ (":method", self._request.method), (":authority", url.netloc), ] # The ":scheme" and ":path" pseudo-header fields MUST # be omitted for CONNECT method (refer RFC 7540 - Section 8.3) if self._request.method != "CONNECT": headers += [ (":scheme", self._protocol.metadata["uri"].scheme), (":path", path), ] content_length = str(len(self._request.body)) headers.append(("Content-Length", content_length)) content_length_name = self._request.headers.normkey(b"Content-Length") for name, values in self._request.headers.items(): for value_bytes in values: value = str(value_bytes, "utf-8") if name == content_length_name: if value != content_length: logger.warning( "Ignoring bad Content-Length header %r of request %r, " "sending %r instead", value, self._request, content_length, ) continue headers.append((str(name, "utf-8"), value)) return headers def initiate_request(self) -> None: if self.check_request_url(): headers = self._get_request_headers() self._protocol.conn.send_headers(self.stream_id, headers, end_stream=False) self.metadata["request_sent"] = True self.send_data() else: # Close this stream calling the response errback # Note that we have not sent any headers self.close(StreamCloseReason.INVALID_HOSTNAME) def send_data(self) -> None: """Called immediately after the headers are sent. Here we send all the data as part of the request. If the content length is 0 initially then we end the stream immediately and wait for response data. Warning: Only call this method when stream not closed from client side and has initiated request already by sending HEADER frame. If not then stream will raise ProtocolError (raise by h2 state machine). """ if self.metadata["stream_closed_local"]: raise StreamClosedError(self.stream_id) # Firstly, check what the flow control window is for current stream. window_size = self._protocol.conn.local_flow_control_window( stream_id=self.stream_id ) # Next, check what the maximum frame size is. max_frame_size = self._protocol.conn.max_outbound_frame_size # We will send no more than the window size or the remaining file size # of data in this call, whichever is smaller. bytes_to_send_size = min(window_size, self.metadata["remaining_content_length"]) # We now need to send a number of data frames. while bytes_to_send_size > 0: chunk_size = min(bytes_to_send_size, max_frame_size) data_chunk_start_id = ( self.metadata["request_content_length"] - self.metadata["remaining_content_length"] ) data_chunk = self._request.body[ data_chunk_start_id : data_chunk_start_id + chunk_size ] self._protocol.conn.send_data(self.stream_id, data_chunk, end_stream=False) bytes_to_send_size -= chunk_size self.metadata["remaining_content_length"] -= chunk_size self.metadata["remaining_content_length"] = max( 0, self.metadata["remaining_content_length"] ) # End the stream if no more data needs to be send if self.metadata["remaining_content_length"] == 0: self._protocol.conn.end_stream(self.stream_id) # Q. What about the rest of the data? # Ans: Remaining Data frames will be sent when we get a WindowUpdate frame def receive_window_update(self) -> None: """Flow control window size was changed. Send data that earlier could not be sent as we were blocked behind the flow control. """ if ( self.metadata["remaining_content_length"] and not self.metadata["stream_closed_server"] and self.metadata["request_sent"] ): self.send_data() def receive_data(self, data: bytes, flow_controlled_length: int) -> None: self._response["body"].write(data) self._response["flow_controlled_size"] += flow_controlled_length # We check maxsize here in case the Content-Length header was not received if ( self._download_maxsize and self._response["flow_controlled_size"] > self._download_maxsize ): self.reset_stream(StreamCloseReason.MAXSIZE_EXCEEDED) return if self._log_warnsize: self.metadata["reached_warnsize"] = True warning_msg = ( f"Received more ({self._response['flow_controlled_size']}) bytes than download " f"warn size ({self._download_warnsize}) in request {self._request}" ) logger.warning(warning_msg) # Acknowledge the data received self._protocol.conn.acknowledge_received_data( self._response["flow_controlled_size"], self.stream_id ) def receive_headers(self, headers: list[HeaderTuple]) -> None: for name, value in headers: self._response["headers"].appendlist(name, value) # Check if we exceed the allowed max data size which can be received expected_size = int(self._response["headers"].get(b"Content-Length", -1)) if self._download_maxsize and expected_size > self._download_maxsize: self.reset_stream(StreamCloseReason.MAXSIZE_EXCEEDED) return if self._log_warnsize: self.metadata["reached_warnsize"] = True warning_msg = ( f"Expected response size ({expected_size}) larger than " f"download warn size ({self._download_warnsize}) in request {self._request}" ) logger.warning(warning_msg) def reset_stream(self, reason: StreamCloseReason = StreamCloseReason.RESET) -> None: """Close this stream by sending a RST_FRAME to the remote peer""" if self.metadata["stream_closed_local"]: raise StreamClosedError(self.stream_id) # Clear buffer earlier to avoid keeping data in memory for a long time self._response["body"].truncate(0) self.metadata["stream_closed_local"] = True self._protocol.conn.reset_stream(self.stream_id, ErrorCodes.REFUSED_STREAM) self.close(reason) def close( self, reason: StreamCloseReason, errors: list[BaseException] | None = None, from_protocol: bool = False, ) -> None: """Based on the reason sent we will handle each case.""" if self.metadata["stream_closed_server"]: raise StreamClosedError(self.stream_id) if not isinstance(reason, StreamCloseReason): raise TypeError( f"Expected StreamCloseReason, received {reason.__class__.__qualname__}" ) # Have default value of errors as an empty list as # some cases can add a list of exceptions errors = errors or [] if not from_protocol: self._protocol.pop_stream(self.stream_id) self.metadata["stream_closed_server"] = True # We do not check for Content-Length or Transfer-Encoding in response headers # and add `partial` flag as in HTTP/1.1 as 'A request or response that includes # a payload body can include a content-length header field' (RFC 7540 - Section 8.1.2.6) # NOTE: Order of handling the events is important here # As we immediately cancel the request when maxsize is exceeded while # receiving DATA_FRAME's when we have received the headers (not # having Content-Length) if reason is StreamCloseReason.MAXSIZE_EXCEEDED: expected_size = int( self._response["headers"].get( b"Content-Length", self._response["flow_controlled_size"] ) ) error_msg = ( f"Cancelling download of {self._request.url}: received response " f"size ({expected_size}) larger than download max size ({self._download_maxsize})" ) logger.error(error_msg) self._deferred_response.errback(CancelledError(error_msg)) elif reason is StreamCloseReason.ENDED: self._fire_response_deferred() # Stream was abruptly ended here elif reason is StreamCloseReason.CANCELLED: # Client has cancelled the request. Remove all the data # received and fire the response deferred with no flags set # NOTE: The data is already flushed in Stream.reset_stream() called # immediately when the stream needs to be cancelled # There maybe no :status in headers, we make # HTTP Status Code: 499 - Client Closed Request self._response["headers"][":status"] = "499" self._fire_response_deferred() elif reason is StreamCloseReason.RESET: self._deferred_response.errback( ResponseFailed( [ Failure( f"Remote peer {self._protocol.metadata['ip_address']} sent RST_STREAM", ProtocolError, ) ] ) ) elif reason is StreamCloseReason.CONNECTION_LOST: self._deferred_response.errback(ResponseFailed(errors)) elif reason is StreamCloseReason.INACTIVE: errors.insert(0, InactiveStreamClosed(self._request)) self._deferred_response.errback(ResponseFailed(errors)) else: assert reason is StreamCloseReason.INVALID_HOSTNAME self._deferred_response.errback( InvalidHostname( self._request, str(self._protocol.metadata["uri"].host, "utf-8"), f"{self._protocol.metadata['ip_address']}:{self._protocol.metadata['uri'].port}", ) ) def _fire_response_deferred(self) -> None: """Builds response from the self._response dict and fires the response deferred callback with the generated response instance""" body = self._response["body"].getvalue() response_cls = responsetypes.from_args( headers=self._response["headers"], url=self._request.url, body=body, ) response = response_cls( url=self._request.url, status=int(self._response["headers"][":status"]), headers=self._response["headers"], body=body, request=self._request, certificate=self._protocol.metadata["certificate"], ip_address=self._protocol.metadata["ip_address"], protocol="h2", ) self._deferred_response.callback(response)