From aeaeb7385b7d1e6570bad38da4db492de8fb4206 Mon Sep 17 00:00:00 2001 From: Aditya Date: Tue, 14 Jul 2020 03:55:14 +0530 Subject: [PATCH] feat: assert negotiated protocol as h2 - implement IHandshakeListener in H2ClientProtocol to know when handshake is completed - implement IProtocolNegotiationFactory in H2ClientFactory to provide information about the acceptableProtols (h2) during NPN or ALPN protocol --- scrapy/core/http2/protocol.py | 69 ++++++++++++++++++++++------- scrapy/core/http2/stream.py | 4 +- tests/test_http2_client_protocol.py | 2 +- 3 files changed, 56 insertions(+), 19 deletions(-) diff --git a/scrapy/core/http2/protocol.py b/scrapy/core/http2/protocol.py index ee51300cc..5d859d475 100644 --- a/scrapy/core/http2/protocol.py +++ b/scrapy/core/http2/protocol.py @@ -12,10 +12,12 @@ from h2.events import ( ) from h2.exceptions import ProtocolError from twisted.internet.defer import Deferred +from twisted.internet.interfaces import IHandshakeListener, IProtocolNegotiationFactory from twisted.internet.protocol import connectionDone, Factory, Protocol from twisted.internet.ssl import Certificate from twisted.python.failure import Failure from twisted.web.client import URI +from zope.interface import implementer from scrapy.core.http2.stream import Stream, StreamCloseReason from scrapy.core.http2.types import H2ConnectionMetadataDict @@ -25,15 +27,29 @@ from scrapy.settings import Settings logger = logging.getLogger(__name__) +class InvalidNegotiatedProtocol(ProtocolError): + + def __init__(self, negotiated_protocol: str) -> None: + self.negotiated_protocol = negotiated_protocol + + def __str__(self) -> str: + return f'InvalidHostname: Expected h2 as negotiated protocol, received {self.negotiated_protocol}' + + +@implementer(IHandshakeListener) class H2ClientProtocol(Protocol): - def __init__(self, uri: URI, settings: Settings) -> None: + def __init__(self, uri: URI, settings: Settings, conn_lost_deferred: Optional[Deferred] = None) -> 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 = conn_lost_deferred + config = H2Configuration(client_side=True, header_encoding='utf-8') self.conn = H2Connection(config=config) @@ -55,9 +71,9 @@ class H2ClientProtocol(Protocol): # We use simple FIFO policy to handle pending requests self._active_streams = 0 - # Save an instance of ProtocolError raised by hyper-h2 - # We pass this instance to the streams ResponseFailed() failure - self._protocol_error: Optional[ProtocolError] = None + # 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] = [] self.metadata: H2ConnectionMetadataDict = { 'certificate': None, @@ -136,6 +152,12 @@ class H2ClientProtocol(Protocol): # Add the stream to the request pool self._pending_request_stream_pool.append(stream) + + # If we are connection and receive a request + # There is a good chance that the connection was IDLE + # Hence, we need to initiate pending requests + if self.is_connected: + self._send_pending_requests() return d def connectionMade(self) -> None: @@ -149,6 +171,19 @@ class H2ClientProtocol(Protocol): self.conn.initiate_connection() self._write_to_transport() + def _lose_connection_with_error(self, errors: List[BaseException]): + """Helper function to lose the connection with the error sent as a + reason""" + self._conn_lost_errors += errors + self.transport.loseConnection() + + def handshakeCompleted(self): + """We close the connection with InvalidNegotiatedProtocol exception + when the connection was not made via h2 protocol""" + negotiated_protocol = str(self.transport.negotiatedProtocol, 'utf-8') + if negotiated_protocol != 'h2': + self._lose_connection_with_error([InvalidNegotiatedProtocol(negotiated_protocol)]) + def dataReceived(self, data: bytes) -> None: try: events = self.conn.receive_data(data) @@ -157,10 +192,7 @@ class H2ClientProtocol(Protocol): # 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._protocol_error = e - - # We lose the transport connection here - self.transport.loseConnection() + self._lose_connection_with_error([e]) finally: self._write_to_transport() @@ -168,17 +200,17 @@ class H2ClientProtocol(Protocol): """Called by Twisted when the transport connection is lost. No need to write anything to transport here. """ - errors = [] + # Notify the connection pool instance such that no new requests are + # sent over current connection if not reason.check(connectionDone): - logger.warning("Connection lost with reason " + str(reason)) - errors.append(reason) + self._conn_lost_errors.append(reason) - if self._protocol_error: - errors.append(self._protocol_error) + if self._conn_lost_deferred: + self._conn_lost_deferred.callback(self._conn_lost_errors) for stream in self.streams.values(): if stream.request_sent: - stream.close(StreamCloseReason.CONNECTION_LOST, errors, from_protocol=True) + stream.close(StreamCloseReason.CONNECTION_LOST, self._conn_lost_errors, from_protocol=True) else: stream.close(StreamCloseReason.INACTIVE, from_protocol=True) @@ -242,10 +274,15 @@ class H2ClientProtocol(Protocol): stream.receive_window_update() +@implementer(IProtocolNegotiationFactory) class H2ClientFactory(Factory): - def __init__(self, uri: URI, settings: Settings) -> None: + def __init__(self, uri: URI, settings: Settings, conn_lost_deferred: Optional[Deferred] = None) -> None: self.uri = uri self.settings = settings + self.conn_lost_deferred = conn_lost_deferred def buildProtocol(self, addr) -> H2ClientProtocol: - return H2ClientProtocol(self.uri, self.settings) + return H2ClientProtocol(self.uri, self.settings, self.conn_lost_deferred) + + def acceptableProtocols(self) -> List[bytes]: + return [b'h2'] diff --git a/scrapy/core/http2/stream.py b/scrapy/core/http2/stream.py index a2f0e2aa4..16d54b2fc 100644 --- a/scrapy/core/http2/stream.py +++ b/scrapy/core/http2/stream.py @@ -44,7 +44,7 @@ class InvalidHostname(Exception): class StreamCloseReason(Enum): - # Received a StreamEnded event + # Received a StreamEnded event from the remote ENDED = 1 # Received a StreamReset event -- ended abruptly @@ -329,7 +329,7 @@ class Stream: def close( self, reason: StreamCloseReason, - errors: Optional[List[Exception]] = None, + errors: Optional[List[BaseException]] = None, from_protocol: bool = False ) -> None: """Based on the reason sent we will handle each case. diff --git a/tests/test_http2_client_protocol.py b/tests/test_http2_client_protocol.py index 05f4889ef..7fcea58c5 100644 --- a/tests/test_http2_client_protocol.py +++ b/tests/test_http2_client_protocol.py @@ -9,7 +9,7 @@ from urllib.parse import urlencode from h2.exceptions import InvalidBodyLengthError from twisted.internet import reactor -from twisted.internet.defer import inlineCallbacks, DeferredList, CancelledError +from twisted.internet.defer import CancelledError, DeferredList, inlineCallbacks from twisted.internet.endpoints import SSL4ClientEndpoint, SSL4ServerEndpoint from twisted.internet.ssl import optionsForClientTLS, PrivateCertificate, Certificate from twisted.python.failure import Failure