diff --git a/docs/deployment.md b/docs/deployment.md index a251f4a1c..21423a2e2 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -44,7 +44,7 @@ Options: $WEB_CONCURRENCY environment variable if available, or 1. Not valid with --reload. --loop [auto|asyncio|uvloop] Event loop implementation. [default: auto] - --http [auto|h11|httptools] HTTP protocol implementation. [default: + --http [auto|h11|httptools|h2] HTTP protocol implementation. [default: auto] --ws [auto|none|websockets|wsproto] WebSocket protocol implementation. diff --git a/docs/index.md b/docs/index.md index 957882c72..f96173c77 100644 --- a/docs/index.md +++ b/docs/index.md @@ -117,7 +117,7 @@ Options: $WEB_CONCURRENCY environment variable if available, or 1. Not valid with --reload. --loop [auto|asyncio|uvloop] Event loop implementation. [default: auto] - --http [auto|h11|httptools] HTTP protocol implementation. [default: + --http [auto|h11|httptools|h2] HTTP protocol implementation. [default: auto] --ws [auto|none|websockets|wsproto] WebSocket protocol implementation. diff --git a/requirements.txt b/requirements.txt index 77c9e2d49..b53928270 100644 --- a/requirements.txt +++ b/requirements.txt @@ -20,6 +20,7 @@ types-pyyaml trustme cryptography coverage +starlette httpx==0.16.* pytest-asyncio==0.14.* async_generator; python_version < '3.7' diff --git a/setup.py b/setup.py index 8d8721351..c8356f1a2 100755 --- a/setup.py +++ b/setup.py @@ -47,6 +47,7 @@ def get_packages(package): "asgiref>=3.3.4", "click>=7.*", "h11>=0.8", + "h2>=4.0.0", "typing-extensions;" + env_marker_below_38, ] diff --git a/tests/protocols/test_h2.py b/tests/protocols/test_h2.py new file mode 100644 index 000000000..94da70553 --- /dev/null +++ b/tests/protocols/test_h2.py @@ -0,0 +1,41 @@ +import httpx +import pytest +from starlette.applications import Starlette +from starlette.responses import JSONResponse +from starlette.routing import Route + +from tests.utils import run_server +from uvicorn.config import Config + + +async def homepage(request): + return JSONResponse({"hello": "world"}) + + +app = Starlette( + routes=[ + Route("/", homepage, methods=["GET", "POST"]), + ], +) + + +@pytest.mark.asyncio +async def test_run( + tls_ca_ssl_context, tls_ca_certificate_pem_path, tls_ca_certificate_private_key_path +): + config = Config( + app=app, + http="h2", + loop="asyncio", + limit_max_requests=1, + ssl_keyfile=tls_ca_certificate_private_key_path, + ssl_certfile=tls_ca_certificate_pem_path, + ssl_ca_certs=tls_ca_certificate_pem_path, + ) + async with run_server(config): + async with httpx.AsyncClient(verify=tls_ca_ssl_context, http2=True) as client: + response = await client.post( + "https://127.0.0.1:8000", data={"hello": "world"} + ) + assert response.status_code == 200 + assert response.http_version == "HTTP/2" diff --git a/tests/protocols/test_http.py b/tests/protocols/test_http.py index 577f70edb..55c473012 100644 --- a/tests/protocols/test_http.py +++ b/tests/protocols/test_http.py @@ -73,6 +73,26 @@ ] ) +UPGRADE_REQUEST_h2c = b"\r\n".join( + [ + b"GET / HTTP/1.1", + b"Host: example.org", + b"Upgrade: h2c", + b"HTTP2-Settings: SomeHTTP2Setting", + b"", + b"", + ] +) + +UPGRADE_REQUEST_HTTP2_PRIOR = b"\r\n".join( + [ + b"PRI * HTTP/2.0", + b"", + b"", + ] +) + + INVALID_REQUEST_TEMPLATE = b"\r\n".join( [ b"%s", @@ -683,6 +703,24 @@ def test_supported_upgrade_request(protocol_cls, event_loop): assert b"HTTP/1.1 426 " in protocol.transport.buffer +# @pytest.mark.parametrize("protocol_cls", [H11Protocol]) +# def test_h2c_upgrade_request(protocol_cls, event_loop): +# app = Response("Hello, world", media_type="text/plain") +# +# with get_connected_protocol(app, protocol_cls, event_loop) as protocol: +# protocol.data_received(UPGRADE_REQUEST_h2c) +# # TODO: check h2c_upgrade_request response + + +@pytest.mark.parametrize("protocol_cls", [H11Protocol]) +def test_h2_prior_upgrade_request(protocol_cls, event_loop): + app = Response("Hello, world", media_type="text/plain") + + with get_connected_protocol(app, protocol_cls, event_loop) as protocol: + protocol.data_received(UPGRADE_REQUEST_HTTP2_PRIOR) + # TODO: check h2_prior_upgrade_request response + + async def asgi3app(scope, receive, send): pass diff --git a/tests/test_config.py b/tests/test_config.py index 1e26c4cc5..917570afb 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -185,6 +185,18 @@ def test_ssl_config_combined(tls_certificate_pem_path: str) -> None: assert config.is_ssl is True +def test_ssl_config_h2(tls_certificate_pem_path): + config = Config( + app=asgi_app, + http="h2", + ssl_certfile=tls_certificate_pem_path, + ) + config.load() + + assert config.is_ssl is True + # TODO: Should we also check HTTP/2-Specific 'ciphers' and 'options' here? + + def asgi2_app(scope: Scope) -> typing.Callable: async def asgi( receive: ASGIReceiveCallable, send: ASGISendCallable diff --git a/uvicorn/_handlers/http.py b/uvicorn/_handlers/http.py index e810f6626..43249751e 100644 --- a/uvicorn/_handlers/http.py +++ b/uvicorn/_handlers/http.py @@ -35,6 +35,12 @@ async def handle_http( loop = asyncio.get_event_loop() connection_lost = loop.create_future() + ssl_object = writer.get_extra_info("ssl_object") + if ssl_object is not None: + alpn_protocol = ssl_object.selected_alpn_protocol() + if alpn_protocol == "h2": + config.http_protocol_class = config.h2_protocol_class + # Switch the protocol from the stream reader to our own HTTP protocol class. protocol = config.http_protocol_class( # type: ignore[call-arg, operator] config=config, diff --git a/uvicorn/config.py b/uvicorn/config.py index 2846d104e..b1da418ff 100644 --- a/uvicorn/config.py +++ b/uvicorn/config.py @@ -52,6 +52,7 @@ "auto": "uvicorn.protocols.http.auto:AutoHTTPProtocol", "h11": "uvicorn.protocols.http.h11_impl:H11Protocol", "httptools": "uvicorn.protocols.http.httptools_impl:HttpToolsProtocol", + "h2": "uvicorn.protocols.http.h2_impl:H2Protocol", } WS_PROTOCOLS: Dict[WSProtocolType, Optional[str]] = { "auto": "uvicorn.protocols.websockets.auto:AutoWebSocketsProtocol", @@ -121,6 +122,7 @@ def create_ssl_context( cert_reqs: int, ca_certs: Optional[Union[str, os.PathLike]], ciphers: Optional[str], + enable_h2: Optional[bool], ) -> ssl.SSLContext: ctx = ssl.SSLContext(ssl_version) get_password = (lambda: password) if password else None @@ -130,6 +132,20 @@ def create_ssl_context( ctx.load_verify_locations(ca_certs) if ciphers: ctx.set_ciphers(ciphers) + if enable_h2: + ctx.options |= ( + ssl.OP_NO_SSLv2 + | ssl.OP_NO_SSLv3 + | ssl.OP_NO_TLSv1 + | ssl.OP_NO_TLSv1_1 + | ssl.OP_NO_COMPRESSION + | ssl.OP_CIPHER_SERVER_PREFERENCE + ) + ctx.set_alpn_protocols(["h2", "http/1.1"]) + try: + ctx.set_npn_protocols(["h2", "http/1.1"]) + except NotImplementedError: + pass return ctx @@ -221,6 +237,11 @@ def __init__( self.headers: List[List[str]] = headers or [] self.encoded_headers: Optional[List[Tuple[bytes, bytes]]] = None self.factory = factory + self.h2_protocol_class = None + self.h2_max_concurrent_streams = 100 + self.h2_max_header_list_size = 2 ** 16 + self.h2_max_inbound_frame_size = 2 ** 14 + self.h2_ssl_ciphers = "ECDHE+AESGCM" self.loaded = False self.configure_logging() @@ -305,6 +326,12 @@ def configure_logging(self) -> None: def load(self) -> None: assert not self.loaded + enable_h2 = self.http in ("h2",) + ciphers = ( + self.h2_ssl_ciphers + if (self.ssl_ciphers == "TLSv1" and enable_h2) + else self.ssl_ciphers + ) if self.is_ssl: assert self.ssl_certfile self.ssl: Optional[ssl.SSLContext] = create_ssl_context( @@ -314,7 +341,8 @@ def load(self) -> None: ssl_version=self.ssl_version, cert_reqs=self.ssl_cert_reqs, ca_certs=self.ssl_ca_certs, - ciphers=self.ssl_ciphers, + ciphers=ciphers, + enable_h2=enable_h2, ) else: self.ssl = None @@ -343,6 +371,8 @@ def load(self) -> None: self.lifespan_class = import_from_string(LIFESPAN[self.lifespan]) + self.h2_protocol_class = import_from_string(HTTP_PROTOCOLS["h2"]) + try: self.loaded_app = import_from_string(self.app) except ImportFromStringError as exc: diff --git a/uvicorn/protocols/http/h11_impl.py b/uvicorn/protocols/http/h11_impl.py index 04b997ffe..b77ae8aab 100644 --- a/uvicorn/protocols/http/h11_impl.py +++ b/uvicorn/protocols/http/h11_impl.py @@ -153,6 +153,10 @@ def handle_events(self): break elif event_type is h11.Request: + + if self.check_connection_preface(event): + return + self.headers = [(key.lower(), value) for key, value in event.headers] raw_path, _, query_string = event.target.partition(b"?") self.scope = { @@ -179,6 +183,9 @@ def handle_events(self): if b"upgrade" in tokens: self.handle_upgrade(event) return + elif name == b"upgrade" and value.lower() == b"h2c": + self.handle_upgrade(event) + return # Handle 503 responses when 'limit_concurrency' is exceeded. if self.limit_concurrency is not None and ( @@ -223,13 +230,76 @@ def handle_events(self): self.cycle.more_body = False self.cycle.message_event.set() + def check_connection_preface(self, event): + if ( + event.method == b"PRI" + and event.target == b"*" + and event.http_version == b"2.0" + ): + # https://tools.ietf.org/html/rfc7540#section-3.5 + self.connections.discard(self) + + protocol = self.config.h2_protocol_class( + config=self.config, + server_state=self.server_state, + on_connection_lost=self.on_connection_lost, + ) + protocol.connection_made(self.transport) + self.transport.set_protocol(protocol) + protocol.data_received( + b"PRI * HTTP/2.0\r\n\r\n" + self.conn.trailing_data[0] + ) + return True + return False + def handle_upgrade(self, event): upgrade_value = None + has_body = False for name, value in self.headers: if name == b"upgrade": upgrade_value = value.lower() + elif name in {"content-length", "transfer-encoding"}: + has_body = True + + if upgrade_value == b"h2c" and not has_body: + self.connections.discard(self) + + headers = ((b"upgrade", b"h2c"), *self.headers) + self.transport.write( + self.conn.send( + h11.InformationalResponse(status_code=101, headers=headers) + ) + ) + protocol = self.config.h2_protocol_class( + config=self.config, + server_state=self.server_state, + on_connection_lost=self.on_connection_lost, + ) + protocol.connection_made(self.transport, upgrade_request=event) + self.transport.set_protocol(protocol) + request_data = self.conn.trailing_data[0] + if request_data != b"": + protocol.data_received(request_data) + + elif upgrade_value == b"websocket" and self.ws_protocol_class is not None: + if self.logger.level <= TRACE_LOG_LEVEL: + prefix = "%s:%d - " % tuple(self.client) if self.client else "" + self.logger.log(TRACE_LOG_LEVEL, "%sUpgrading to WebSocket", prefix) + self.connections.discard(self) + output = [event.method, b" ", event.target, b" HTTP/1.1\r\n"] + for name, value in self.headers: + output += [name, b": ", value, b"\r\n"] + output.append(b"\r\n") + protocol = self.ws_protocol_class( + config=self.config, + server_state=self.server_state, + on_connection_lost=self.on_connection_lost, + ) + protocol.connection_made(self.transport) + protocol.data_received(b"".join(output)) + self.transport.set_protocol(protocol) - if upgrade_value != b"websocket" or self.ws_protocol_class is None: + else: msg = "Unsupported upgrade request." self.logger.warning(msg) @@ -254,25 +324,6 @@ def handle_upgrade(self, event): output = self.conn.send(event) self.transport.write(output) self.transport.close() - return - - if self.logger.level <= TRACE_LOG_LEVEL: - prefix = "%s:%d - " % tuple(self.client) if self.client else "" - self.logger.log(TRACE_LOG_LEVEL, "%sUpgrading to WebSocket", prefix) - - self.connections.discard(self) - output = [event.method, b" ", event.target, b" HTTP/1.1\r\n"] - for name, value in self.headers: - output += [name, b": ", value, b"\r\n"] - output.append(b"\r\n") - protocol = self.ws_protocol_class( - config=self.config, - server_state=self.server_state, - on_connection_lost=self.on_connection_lost, - ) - protocol.connection_made(self.transport) - protocol.data_received(b"".join(output)) - self.transport.set_protocol(protocol) def on_response_complete(self): self.server_state.total_requests += 1 diff --git a/uvicorn/protocols/http/h2_impl.py b/uvicorn/protocols/http/h2_impl.py new file mode 100644 index 000000000..6ab534e7b --- /dev/null +++ b/uvicorn/protocols/http/h2_impl.py @@ -0,0 +1,588 @@ +import asyncio +import collections +import http +import logging +from urllib.parse import unquote + +import h2.config +import h2.connection +import h2.errors +import h2.events +import h2.exceptions +import h2.settings + +from uvicorn.protocols.http.flow_control import ( + CLOSE_HEADER, + HIGH_WATER_LIMIT, + TRACE_LOG_LEVEL, + FlowControl, + service_unavailable, +) +from uvicorn.protocols.utils import ( + get_client_addr, + get_local_addr, + get_path_with_query_string, + get_remote_addr, + is_ssl, +) + + +def _get_status_phrase(status_code): + try: + return http.HTTPStatus(status_code).phrase.encode() + except ValueError: + return b"" + + +STATUS_PHRASES = { + status_code: _get_status_phrase(status_code) for status_code in range(100, 600) +} + +_StreamRequest = collections.namedtuple("_StreamRequest", ("headers", "scope", "cycle")) + + +class H2Protocol(asyncio.Protocol): + def __init__(self, config, server_state, on_connection_lost=None, _loop=None): + if not config.loaded: + config.load() + + self.config = config + self.app = config.loaded_app + self.on_connection_lost = on_connection_lost + self.loop = _loop or asyncio.get_event_loop() + self.logger = logging.getLogger("uvicorn.error") + self.access_logger = logging.getLogger("uvicorn.access") + self.access_log = self.access_logger.hasHandlers() + + self.conn = h2.connection.H2Connection( + config=h2.config.H2Configuration(client_side=False, header_encoding=None) + ) + self.conn.DEFAULT_MAX_INBOUND_FRAME_SIZE = config.h2_max_inbound_frame_size + self.conn.local_settings = h2.settings.Settings( + client=False, + initial_values={ + h2.settings.SettingCodes.MAX_CONCURRENT_STREAMS: config.h2_max_concurrent_streams, # noqa: E501 + h2.settings.SettingCodes.MAX_HEADER_LIST_SIZE: config.h2_max_header_list_size, # noqa: E501 + h2.settings.SettingCodes.ENABLE_CONNECT_PROTOCOL: 1, + }, + ) + + self.ws_protocol_class = config.ws_protocol_class + self.root_path = config.root_path + self.limit_concurrency = config.limit_concurrency + + # Timeouts + self.timeout_keep_alive_task = None + self.timeout_keep_alive = config.timeout_keep_alive + + # Shared server state + self.server_state = server_state + self.connections = server_state.connections + self.tasks = server_state.tasks + self.default_headers = server_state.default_headers + + # Per-connection state + self.transport = None + self.flow = None + self.server = None + self.client = None + self.scheme = None + + # Per-request state + self.scope = None + self.headers = None + self.streams = {} + + # Protocol interface + def connection_made(self, transport: asyncio.Transport, upgrade_request=None): + self.connections.add(self) + + self.transport = transport + self.flow = FlowControl(transport) + self.server = get_local_addr(transport) + self.client = get_remote_addr(transport) + self.scheme = "https" if is_ssl(transport) else "http" + + if upgrade_request is None: + self.conn.initiate_connection() + self.transport.write(self.conn.data_to_send()) + else: + # TODO: Implementations for httptools_impl to handle h2c + # For now, only h11_impl will handle h2c + + # if type(upgrade_request) == h11.Request + settings = "" + headers = [] + for name, value in upgrade_request.headers: + if name.lower() == b"http2-settings": + settings = value.decode() + elif name.lower() == b"host": + headers.append((b":authority", value)) + headers.append((name, value)) + headers.append((b":method", upgrade_request.method)) + headers.append((b":path", upgrade_request.target)) + headers.append((b":scheme", self.scheme.encode("ascii"))) + self.conn.initiate_upgrade_connection(settings) + self.transport.write(self.conn.data_to_send()) + event = h2.events.RequestReceived() + event.stream_id = 1 + event.headers = headers + self.on_request_received(event) + self.on_stream_ended(event) + + if self.logger.level <= TRACE_LOG_LEVEL: + prefix = "%s:%d - " % tuple(self.client) if self.client else "" + self.logger.log(TRACE_LOG_LEVEL, "%sConnection made", prefix) + + def connection_lost(self, exc): + self.connections.discard(self) + + self.logger.debug("%s - Disconnected", self.client[0]) + + for stream_id, stream in self.streams.items(): + if stream.cycle: + if not stream.cycle.response_complete: + stream.cycle.disconnected = True + try: + self.conn.close_connection(last_stream_id=stream_id) + self.transport.write(self.conn.data_to_send()) + except h2.exceptions.ProtocolError as err: + self.logger.debug( + "connection lost, failed to close connection.", exc_info=err + ) + stream.cycle.message_event.set() + + self.logger.debug( + "Disconnected, current streams: %s", list(self.streams.keys()), exc_info=exc + ) + self.streams = {} + if self.flow is not None: + self.flow.resume_writing() + + if exc is None: + # Ref: https://github.com/encode/uvicorn/pull/929#discussion_r582739354 + self.transport.close() + + if self.on_connection_lost is not None: + self.on_connection_lost() + + def _unset_keepalive_if_required(self): + if self.timeout_keep_alive_task is not None: + self.timeout_keep_alive_task.cancel() + self.timeout_keep_alive_task = None + + def eof_received(self): + self.logger.debug( + "eof received, current streams: %s", list(self.streams.keys()) + ) + self.streams = {} + + def data_received(self, data): + self._unset_keepalive_if_required() + try: + events = self.conn.receive_data(data) + except h2.exceptions.ProtocolError: + self.transport.write(self.conn.data_to_send()) + self.transport.close() + else: + self.transport.write(self.conn.data_to_send()) + self.handle_events(events) + + def handle_events(self, events): + for event in events: + event_type = type(event) + if event_type is h2.events.RequestReceived: + self.on_request_received(event) + elif event_type is h2.events.DataReceived: + self.on_data_received(event) + elif event_type is h2.events.StreamEnded: + self.on_stream_ended(event) + elif event_type is h2.events.StreamReset: + self.on_stream_reset(event) + elif event_type is h2.events.WindowUpdated: + pass + elif event_type is h2.events.PriorityUpdated: + pass + elif event_type is h2.events.RemoteSettingsChanged: + pass + elif event_type is h2.events.ConnectionTerminated: + self.on_connection_terminated(event) + self.transport.write(self.conn.data_to_send()) + + def on_request_received(self, event: h2.events.RequestReceived): + self.scope = { + "type": "http", + "asgi": { + "version": self.config.asgi_version, + "spec_version": "2.1", + }, + "http_version": "2", + "server": self.server, + "client": self.client, + "root_path": self.root_path, + "extensions": {"http.response.push": {}}, + "headers": [], + } + scope_mapping = { + b":scheme": "scheme", + b":authority": "authority", + b":method": "method", + b":path": "raw_path", + } + + websocket_protocol = False + for key, value in event.headers: + if key in scope_mapping: + self.scope[scope_mapping[key]] = value.decode("ascii") + elif key == b":protocol" and value == b"websocket": + websocket_protocol = True + else: + self.scope["headers"].append((key.lower(), value)) + path, _, query_string = self.scope["raw_path"].partition("?") + self.scope["path"], self.scope["query_string"] = ( + unquote(path), + query_string.encode("ascii"), + ) + + if self.scope["method"] == "CONNECT" and websocket_protocol: + # TODO: Websocket Extended CONNECT Implementation + pass + + # Handle 503 responses when 'limit_concurrency' is exceeded. + if self.limit_concurrency is not None and ( + len(self.connections) >= self.limit_concurrency + or len(self.tasks) >= self.limit_concurrency + ): + app = service_unavailable + message = "Exceeded concurrency limit." + self.logger.warning(message) + else: + app = self.app + + stream_id = event.stream_id + + cycle = RequestResponseCycle( + stream_id=stream_id, + scope=self.scope, + conn=self.conn, + transport=self.transport, + flow=self.flow, + logger=self.logger, + access_logger=self.access_logger, + access_log=self.access_log, + default_headers=self.default_headers, + message_event=asyncio.Event(), + on_response=self.on_response_complete, + ) + self.streams[stream_id] = _StreamRequest( + headers=self.scope["headers"], scope=self.scope, cycle=cycle + ) + self.logger.debug( + "New request received, current stream(%s), all streams: %s", + stream_id, + list(self.streams.keys()), + ) + task = self.loop.create_task(self.streams[stream_id].cycle.run_asgi(app)) + task.add_done_callback(self.tasks.discard) + task.add_done_callback( + lambda t: self.logger.debug( + "stream(%s) done, path(%s)", stream_id, self.scope["path"] + ) + ) + self.tasks.add(task) + + def on_data_received(self, event: h2.events.DataReceived): + stream_id = event.stream_id + self.logger.debug( + "On data received, current %s, streams: %s", stream_id, self.streams.keys() + ) + try: + self.streams[stream_id].cycle.body += event.data + except KeyError: + self.conn.reset_stream( + stream_id, error_code=h2.errors.ErrorCodes.PROTOCOL_ERROR + ) + else: + # In Hypercorn: + # https://gitlab.com/pgjones/hypercorn/-/blob/0.11.2/src/hypercorn/protocol/h2.py#L233-235 + # self.conn.acknowledge_received_data( + # event.flow_controlled_length, event.stream_id + # ) + # TODO: To be done here, or in RequestResponseCycle's `receive()`? 😕 + + body_size = len(self.streams[stream_id].cycle.body) + if body_size > HIGH_WATER_LIMIT: + self.flow.pause_reading() + self.streams[stream_id].cycle.message_event.set() + + def on_stream_ended(self, event: h2.events.StreamEnded): + stream_id = event.stream_id + self.logger.debug( + "On stream ended, current %s, streams: %s", stream_id, self.streams.keys() + ) + try: + stream = self.streams[stream_id] + except KeyError: + self.conn.reset_stream( + stream_id, error_code=h2.errors.ErrorCodes.STREAM_CLOSED + ) + else: + self.flow.resume_reading() + stream.cycle.more_body = False + self.streams[stream_id].cycle.message_event.set() + + def on_stream_reset(self, event: h2.events.StreamReset): + self.logger.debug( + "stream(%s) reset by %s with error_code %s", + event.stream_id, + "server" if event.remote_reset else "remote peer", + event.error_code, + ) + self.streams.pop(event.stream_id, None) + + # TODO: To be done or not? + # In Hypercorn: + # app_put({"type": "http.disconnect"}) + + def on_connection_terminated(self, event: h2.events.ConnectionTerminated): + stream_id = event.last_stream_id + self.logger.debug( + "H2Connection terminated, additional_data(%s), " + "error_code(%s), last_stream(%s), streams: %s", + event.additional_data, + event.error_code, + stream_id, + list(self.streams.keys()), + ) + stream = self.streams.pop(stream_id) + if stream: + stream.cycle.disconnected = True + self.conn.close_connection(last_stream_id=stream_id) + self.transport.write(self.conn.data_to_send()) + self.transport.close() + + def on_response_complete(self): + self.server_state.total_requests += 1 + + if self.transport.is_closing(): + return + + # Set a short Keep-Alive timeout. + self._unset_keepalive_if_required() + + self.timeout_keep_alive_task = self.loop.call_later( + self.timeout_keep_alive, self.timeout_keep_alive_handler + ) + + # Unpause data reads if needed. + self.flow.resume_reading() + + def pause_writing(self): + """ + Called by the transport when the write buffer exceeds the high water mark. + """ + self.flow.pause_writing() + + def resume_writing(self): + """ + Called by the transport when the write buffer drops below the low water mark. + """ + self.flow.resume_writing() + + def shutdown(self): + self.logger.debug( + "Shutdown. streams: %s, tasks: %s", self.streams.keys(), self.tasks + ) + for stream_id, stream in self.streams.items(): + if stream.cycle is None or stream.cycle.response_complete: + self.conn.close_connection(last_stream_id=stream_id) + self.transport.write(self.conn.data_to_send()) + else: + stream.cycle.keep_alive = False + self.streams = {} + self.transport.close() + + def timeout_keep_alive_handler(self): + """ + Called on a keep-alive connection if no new data is received after a short + delay. + """ + if not self.transport.is_closing(): + for stream_id, stream in self.streams.items(): + self.conn.close_connection(last_stream_id=stream_id) + self.transport.write(self.conn.data_to_send()) + self.transport.close() + + +class RequestResponseCycle: + def __init__( + self, + stream_id, + scope, + conn, + transport, + flow, + logger, + access_logger, + access_log, + default_headers, + message_event, + on_response, + ): + self.stream_id = stream_id + self.scope = scope + self.conn = conn + self.transport = transport + self.flow = flow + self.logger = logger + self.access_logger = access_logger + self.access_log = access_log + self.default_headers = default_headers + self.message_event = message_event + self.on_response = on_response + + # Connection state + self.disconnected = False + self.keep_alive = True + + # Request state + self.body = b"" + self.more_body = True + + # Response state + self.response_started = False + self.response_complete = False + + # ASGI exception wrapper + async def run_asgi(self, app): + try: + result = await app(self.scope, self.receive, self.send) + except BaseException as exc: + msg = "Exception in ASGI application\n" + self.logger.error(msg, exc_info=exc) + if not self.response_started: + await self.send_500_response() + else: + self.transport.close() + else: + if result is not None: + msg = "ASGI callable should return None, but returned '%s'." + self.logger.error(msg, result) + self.transport.close() + elif not self.response_started and not self.disconnected: + msg = "ASGI callable returned without starting response." + self.logger.error(msg) + await self.send_500_response() + elif not self.response_complete and not self.disconnected: + msg = "ASGI callable returned without completing response." + self.logger.error(msg) + self.transport.close() + finally: + self.on_response = None + + async def send_500_response(self): + await self.send( + { + "type": "http.response.start", + "status": 500, + "headers": [ + (b"content-type", b"text/plain; charset=utf-8"), + (b"connection", b"close"), + ], + } + ) + await self.send( + {"type": "http.response.body", "body": b"Internal Server Error"} + ) + + # ASGI interface + async def send(self, message): + message_type = message["type"] + + if self.disconnected: + return + + if self.flow.write_paused: + await self.flow.drain() + + if not self.response_started: + # Sending response status line and headers + if message_type != "http.response.start": + msg = "Expected ASGI message 'http.response.start', but got '%s'." + raise RuntimeError(msg % message_type) + + self.response_started = True + + status_code = message["status"] + + headers = self.default_headers + message.get("headers", []) + + if CLOSE_HEADER in self.scope["headers"] and CLOSE_HEADER not in headers: + headers = headers + [CLOSE_HEADER] + + if self.access_log: + self.access_logger.info( + '%s - "%s %s HTTP/%s" %d', + get_client_addr(self.scope), + self.scope["method"], + get_path_with_query_string(self.scope), + self.scope["http_version"], + status_code, + ) + + # Write response status line and headers + headers = ((":status", str(status_code)), *headers) + self.logger.debug("response start, message %s", message) + self.conn.send_headers(self.stream_id, headers, end_stream=False) + self.transport.write(self.conn.data_to_send()) + elif not self.response_complete: + # Sending response body + if message_type == "http.response.body": + more_body = message.get("more_body", False) + + # Write response body + if self.scope["method"] == "HEAD": + body = b"" + else: + body = message.get("body", b"") + self.conn.send_data(self.stream_id, body, end_stream=(not more_body)) + self.transport.write(self.conn.data_to_send()) + + # Handle response completion + if not more_body: + self.response_complete = True + elif message_type == "http.response.push": + # TODO: Implement or Not? + # https://groups.google.com/a/chromium.org/g/blink-dev/c/K3rYLvmQUBY/m/vOWBKZGoAQAJ 😕 # noqa: E501 + pass + else: + msg = ( + "Expected ASGI message 'http.response.body' " + "or 'http.response.push', but got '%s'." + ) + raise RuntimeError(msg % message_type) + else: + # Response already sent + msg = "Unexpected ASGI message '%s' sent, after response already completed." + raise RuntimeError(msg % message_type) + + if self.response_complete: + self.on_response() + + async def receive(self): + if not self.disconnected and not self.response_complete: + self.flow.resume_reading() + await self.message_event.wait() + self.message_event.clear() + + if self.disconnected or self.response_complete: + message = {"type": "http.disconnect"} + else: + message = { + "type": "http.request", + "body": self.body, + "more_body": self.more_body, + } + self.body = b"" + + return message