from __future__ import annotations import asyncio import contextvars import logging import sys from collections.abc import Callable, Generator from typing import Any, ClassVar, Literal from urllib.parse import unquote import zttp from zttp import Event from uvicorn._types import ( ASGI3Application, ASGIReceiveEvent, ASGISendEvent, HTTPRequestEvent, HTTPResponseBodyEvent, HTTPResponseStartEvent, HTTPScope, ) from uvicorn.config import Config from uvicorn.logging import TRACE_LOG_LEVEL from uvicorn.protocols.http.flow_control import HIGH_WATER_LIMIT, FlowControl, service_unavailable from uvicorn.protocols.utils import get_client_addr, get_local_addr, get_path_with_query_string, get_remote_addr, is_ssl from uvicorn.server import ServerState # RFC 9113 section 8.2.2: connection-specific headers MUST NOT appear in # HTTP/2 messages. zttp rejects them with LocalProtocolError, and an ASGI app # or middleware tuned for HTTP/1.1 may still emit them, so strip them on the # way out. FORBIDDEN_HEADERS = frozenset({b"connection", b"keep-alive", b"proxy-connection", b"transfer-encoding", b"upgrade"}) class ZttpH2Protocol(asyncio.Protocol): alpn_protocols: ClassVar[list[str]] = ["h2"] def __init__( self, config: Config, server_state: ServerState, app_state: dict[str, Any], _loop: asyncio.AbstractEventLoop | None = None, ) -> None: if not config.loaded: config.load() self.config = config self.app = config.loaded_app 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: zttp.H2Connection = zttp.Connection(zttp.SERVER, protocol=zttp.HTTP2) self.root_path = config.root_path self.asgi_version = config.asgi_version self.limit_concurrency = config.limit_concurrency self.app_state = app_state # Timeouts self.timeout_keep_alive_task: asyncio.TimerHandle | None = 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 # Per-connection state self.transport: asyncio.Transport = None # type: ignore[assignment] self.flow: FlowControl = None # type: ignore[assignment] self.server: tuple[str, int | None] | None = None self.client: tuple[str, int] | None = None self.scheme: Literal["http", "https"] | None = None self.shutdown_requested = False # Per-stream state, keyed by HTTP/2 stream id self.cycles: dict[int, RequestResponseCycle] = {} # Protocol interface def connection_made( # type: ignore[override] self, transport: asyncio.Transport ) -> 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" self.conn.initiate_connection() self.flush() if self.logger.level <= TRACE_LOG_LEVEL: prefix = "%s:%d - " % self.client if self.client else "" self.logger.log(TRACE_LOG_LEVEL, "%sHTTP/2 connection made", prefix) def connection_lost(self, exc: Exception | None) -> None: self.connections.discard(self) if self.logger.level <= TRACE_LOG_LEVEL: prefix = "%s:%d - " % self.client if self.client else "" self.logger.log(TRACE_LOG_LEVEL, "%sHTTP/2 connection lost", prefix) for cycle in self.cycles.values(): if not cycle.response_complete: cycle.disconnected = True cycle.message_event.set() if self.flow is not None: self.flow.resume_writing() if exc is None: self.transport.close() self._unset_keepalive_if_required() def eof_received(self) -> None: pass def _unset_keepalive_if_required(self) -> None: if self.timeout_keep_alive_task is not None: self.timeout_keep_alive_task.cancel() self.timeout_keep_alive_task = None def data_received(self, data: bytes) -> None: self._unset_keepalive_if_required() try: self.conn.receive_data(data) self.handle_events() except zttp.RemoteProtocolError as exc: msg = "Invalid HTTP/2 frame received: %s" self.logger.warning(msg, exc) self.flush() self.transport.close() return self.flush() # Frames that carry no stream-level event (SETTINGS, PING, # WINDOW_UPDATE) cancelled the keep-alive timer above but never re-arm # it, so re-arm now if the connection is idle. if not self.cycles and self.timeout_keep_alive_task is None and not self.transport.is_closing(): self.timeout_keep_alive_task = self.loop.call_later( self.timeout_keep_alive, self.timeout_keep_alive_handler ) def flush(self) -> None: """Write out bytes zttp queued outside a cycle's send path - the connection preface and DATA released by a WINDOW_UPDATE credit.""" data = self.conn.data_to_send() if data: self.transport.write(data) def events(self) -> Generator[Event]: """Yield every complete event currently available.""" while True: event = self.conn.next_event() if event is zttp.NEED_DATA: return yield event def handle_events(self) -> None: for event in self.events(): if isinstance(event, zttp.Request): self.handle_request(event) elif isinstance(event, zttp.Data): cycle = self.cycles.get(event.stream_id) if cycle is None or cycle.response_complete: continue cycle.body += event.data # Unreachable until zttp grows its 64 KiB inbound flow-control # windows, which cap how much body can buffer per stream. if len(cycle.body) > HIGH_WATER_LIMIT: # pragma: no cover self.flow.pause_reading() cycle.message_event.set() elif isinstance(event, zttp.EndOfMessage): cycle = self.cycles.get(event.stream_id) if cycle is None or cycle.response_complete: continue cycle.more_body = False cycle.message_event.set() elif isinstance(event, zttp.RstStream): self.handle_rst_stream(event) elif isinstance(event, zttp.GoAway): self.shutdown_requested = True if not self.cycles: self._close_connection() # Settings, Ping and WindowUpdate need no action here: zttp tracks # the send windows internally, and `flush` writes whatever bytes # the new credit released. def handle_request(self, event: zttp.Request) -> None: headers = ( event.headers.to_list(lowercase_names=True) if isinstance(event.headers, zttp.HeaderBlock) else [(name.lower(), value) for name, value in event.headers] ) path = unquote(event.path.decode("ascii")) full_path = self.root_path + path full_raw_path = self.root_path.encode("ascii") + event.path scope: HTTPScope = { "type": "http", "asgi": {"version": self.asgi_version, "spec_version": "2.3"}, "http_version": "2", "server": self.server, "client": self.client, "scheme": self.scheme, # type: ignore[typeddict-item] "method": event.method.decode("ascii"), "root_path": self.root_path, "path": full_path, "raw_path": full_raw_path, "query_string": event.query, "headers": headers, "state": self.app_state.copy(), } # Refuse new streams once a shutdown began, and handle 503 responses # when 'limit_concurrency' is exceeded. if self.shutdown_requested: app = service_unavailable elif 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 cycle = RequestResponseCycle( scope=scope, conn=self.conn, stream=self.conn.stream(event.stream_id), transport=self.transport, flow=self.flow, logger=self.logger, access_logger=self.access_logger, access_log=self.access_log, default_headers=self.server_state.default_headers, message_event=asyncio.Event(), on_response=self.on_response_complete, resume_reading=self.resume_reading_if_idle, ) self.cycles[event.stream_id] = cycle if self.config.reset_contextvars: if sys.version_info >= (3, 11): # pragma: py-lt-311 task = self.loop.create_task(cycle.run_asgi(app), context=contextvars.Context()) else: # pragma: py-gte-311 task = contextvars.Context().run(self.loop.create_task, cycle.run_asgi(app)) else: task = self.loop.create_task(cycle.run_asgi(app)) task.add_done_callback(self.tasks.discard) self.tasks.add(task) def handle_rst_stream(self, event: zttp.RstStream) -> None: cycle = self.cycles.pop(event.stream_id, None) if cycle is None: return if not cycle.response_complete: cycle.disconnected = True cycle.message_event.set() self.on_stream_closed() def resume_reading_if_idle(self) -> None: """Resume reads only if no stream is over the body buffer high-water mark; otherwise a slow stream's buffer would keep growing while a fast peer ships frames on a sibling stream.""" for cycle in self.cycles.values(): if len(cycle.body) > HIGH_WATER_LIMIT: return self.flow.resume_reading() def on_response_complete(self, stream_id: int) -> None: self.server_state.total_requests += 1 self.cycles.pop(stream_id, None) self.on_stream_closed() def on_stream_closed(self) -> None: if self.transport.is_closing(): return # The stream's body buffer is gone, so it can no longer hold back the # read side of the transport. self.resume_reading_if_idle() self._unset_keepalive_if_required() if not self.cycles: if self.shutdown_requested: self._close_connection() return self.timeout_keep_alive_task = self.loop.call_later( self.timeout_keep_alive, self.timeout_keep_alive_handler ) def _close_connection(self) -> None: """Send GOAWAY and close the transport.""" if self.transport.is_closing(): return self.conn.close() self.flush() self.transport.close() def shutdown(self) -> None: """ Called by the server to commence a graceful shutdown. Closes immediately when idle; otherwise new streams are refused with 503 and the connection closes once the last in-flight stream finishes. """ self.shutdown_requested = True if not self.cycles: self._close_connection() def pause_writing(self) -> None: """ Called by the transport when the write buffer exceeds the high water mark. """ self.flow.pause_writing() # pragma: no cover def resume_writing(self) -> None: """ Called by the transport when the write buffer drops below the low water mark. """ self.flow.resume_writing() # pragma: no cover def timeout_keep_alive_handler(self) -> None: """ Called on a keep-alive connection if no new data is received after a short delay. """ self._close_connection() class RequestResponseCycle: def __init__( self, scope: HTTPScope, conn: zttp.H2Connection, stream: zttp.Stream, transport: asyncio.Transport, flow: FlowControl, logger: logging.Logger, access_logger: logging.Logger, access_log: bool, default_headers: list[tuple[bytes, bytes]], message_event: asyncio.Event, on_response: Callable[[int], None], resume_reading: Callable[[], None], ) -> None: self.scope = scope self.conn = conn self.stream = stream 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 self.resume_reading = resume_reading # Connection state self.disconnected = False # Request state self.body = bytearray() self.more_body = True # Response state self.response_started = False self.response_complete = False self.bodyless = False self.expected_content_length: int | None = None # ASGI exception wrapper async def run_asgi(self, app: ASGI3Application) -> None: try: result = await app( # type: ignore[func-returns-value] 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.abort_stream() else: if result is not None: msg = "ASGI callable should return None, but returned '%s'." self.logger.error(msg, result) self.abort_stream() 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.abort_stream() finally: self.on_response(self.stream.stream_id) self.on_response = lambda stream_id: None def abort_stream(self) -> None: """Reset only this stream so sibling streams on the connection survive.""" if self.transport.is_closing(): return self.stream.reset() self.transport.write(self.conn.data_to_send()) self.disconnected = True async def send_500_response(self) -> None: response_start_event: HTTPResponseStartEvent = { "type": "http.response.start", "status": 500, "headers": [(b"content-type", b"text/plain; charset=utf-8")], } await self.send(response_start_event) response_body_event: HTTPResponseBodyEvent = { "type": "http.response.body", "body": b"Internal Server Error", "more_body": False, } await self.send(response_body_event) # ASGI interface async def send(self, message: ASGISendEvent) -> None: if self.flow.write_paused and not self.disconnected: await self.flow.drain() # pragma: no cover if self.disconnected: return if not self.response_started: # Sending response status line and headers if message["type"] != "http.response.start": raise RuntimeError(f"Expected ASGI message 'http.response.start', but got '{message['type']}'.") self.response_started = True status = message["status"] headers: list[tuple[bytes, bytes]] = [] for name, value in list(self.default_headers) + list(message.get("headers", [])): name = name.lower() if name in FORBIDDEN_HEADERS: continue if name == b"te" and value.lower().strip() != b"trailers": continue if name == b"content-length": self.expected_content_length = int(value.decode()) headers.append((name, value)) self.bodyless = self.scope["method"] == "HEAD" or status in (204, 304) or status < 200 if self.bodyless: self.expected_content_length = None 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, ) # Write response headers self.stream.send_response(status, headers) self.transport.write(self.conn.data_to_send()) elif not self.response_complete: # Sending response body if message["type"] != "http.response.body": raise RuntimeError(f"Expected ASGI message 'http.response.body', but got '{message['type']}'.") body = message.get("body", b"") more_body = message.get("more_body", False) # Write response body if self.bodyless: body = b"" elif self.expected_content_length is not None: if len(body) > self.expected_content_length: raise RuntimeError("Response content longer than Content-Length") self.expected_content_length -= len(body) if body: self.stream.send_data(body) self.transport.write(self.conn.data_to_send()) # Handle response completion if not more_body: if self.expected_content_length not in (None, 0): raise RuntimeError("Response content shorter than Content-Length") self.response_complete = True self.message_event.set() self.stream.end_message() self.transport.write(self.conn.data_to_send()) self.on_response(self.stream.stream_id) self.on_response = lambda stream_id: None else: # Response already sent raise RuntimeError(f"Unexpected ASGI message '{message['type']}' sent, after response already completed.") async def receive(self) -> ASGIReceiveEvent: if not self.disconnected and not self.response_complete: self.resume_reading() await self.message_event.wait() self.message_event.clear() if self.disconnected or self.response_complete: return {"type": "http.disconnect"} message: HTTPRequestEvent = {"type": "http.request", "body": bytes(self.body), "more_body": self.more_body} self.body = bytearray() return message