#
# Copyright 2014 Facebook
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""Client and server implementations of HTTP/1.x.
.. versionadded:: 4.0
"""
import asyncio
import logging
import re
import types
import zlib
from collections.abc import Awaitable, Callable
from typing import Optional, Type, cast
from tornado import gen, httputil, iostream
from tornado.concurrent import (
Future,
future_add_done_callback,
future_set_result_unless_cancelled,
)
from tornado.escape import native_str, utf8
from tornado.log import app_log, gen_log
from tornado.util import GzipDecompressor
CR_OR_LF_RE = re.compile(b"\r|\n")
# The maximum number of informational (1xx) responses to accept before the
# real response. Each one is processed with a recursive call to
# _read_message, so an unbounded number of them would exhaust the stack.
# There is no legitimate use for more than a handful.
_MAX_1XX_RESPONSES = 10
# Reads that can be satisfied from data that has already arrived complete
# without returning control to the IOLoop, so a peer that sends a body
# faster than we can process it could otherwise monopolize the IOLoop for
# the duration of the body. While reading a body, we yield to the IOLoop
# after processing this many bytes. Each chunk of a chunked body also counts
# as _CHUNK_YIELD_COST bytes, since the per-chunk overhead dominates when
# the chunks are small.
_YIELD_THRESHOLD = 256 * 1024
_CHUNK_YIELD_COST = 1024
def _is_done(future: Awaitable[object]) -> bool:
# IOStream's read methods are declared to return Awaitables, but they
# are actually Futures which may already be resolved.
return isinstance(future, Future) and future.done()
class _QuietException(Exception):
def __init__(self) -> None:
pass
class _ExceptionLoggingContext:
"""Used with the ``with`` statement when calling delegate methods to
log any exceptions with the given logger. Any exceptions caught are
converted to _QuietException
"""
def __init__(self, logger: logging.Logger) -> None:
self.logger = logger
def __enter__(self) -> None:
pass
def __exit__(
self,
typ: "Optional[Type[BaseException]]",
value: BaseException | None,
tb: types.TracebackType,
) -> None:
if value is not None:
assert typ is not None
# Let HTTPInputError pass through to higher-level handler
if isinstance(value, httputil.HTTPInputError):
return None
self.logger.error("Uncaught exception", exc_info=(typ, value, tb))
raise _QuietException
[docs]
class HTTP1ConnectionParameters:
"""Parameters for `.HTTP1Connection` and `.HTTP1ServerConnection`."""
def __init__(
self,
no_keep_alive: bool = False,
chunk_size: int | None = None,
max_header_size: int | None = None,
header_timeout: float | None = None,
max_body_size: int | None = None,
body_timeout: float | None = None,
decompress: bool = False,
) -> None:
"""
:arg bool no_keep_alive: If true, always close the connection after
one request.
:arg int chunk_size: how much data to read into memory at once
:arg int max_header_size: maximum amount of data for HTTP headers
:arg float header_timeout: how long to wait for all headers (seconds)
:arg int max_body_size: maximum amount of data for body
:arg float body_timeout: how long to wait while reading body (seconds)
:arg bool decompress: if true, decode incoming
``Content-Encoding: gzip``
"""
self.no_keep_alive = no_keep_alive
self.chunk_size = chunk_size or 65536
self.max_header_size = max_header_size or 65536
self.header_timeout = header_timeout
self.max_body_size = max_body_size
self.body_timeout = body_timeout
self.decompress = decompress
[docs]
class HTTP1Connection(httputil.HTTPConnection):
"""Implements the HTTP/1.x protocol.
This class can be on its own for clients, or via `HTTP1ServerConnection`
for servers.
"""
def __init__(
self,
stream: iostream.IOStream,
is_client: bool,
params: HTTP1ConnectionParameters | None = None,
context: object | None = None,
) -> None:
"""
:arg stream: an `.IOStream`
:arg bool is_client: client or server
:arg params: a `.HTTP1ConnectionParameters` instance or ``None``
:arg context: an opaque application-defined object that can be accessed
as ``connection.context``.
"""
self.is_client = is_client
self.stream = stream
if params is None:
params = HTTP1ConnectionParameters()
self.params = params
self.context = context
self.no_keep_alive = params.no_keep_alive
# The body limits can be altered by the delegate, so save them
# here instead of just referencing self.params later.
self._max_body_size = (
self.params.max_body_size
if self.params.max_body_size is not None
else self.stream.max_buffer_size
)
self._body_timeout = self.params.body_timeout
# _write_finished is set to True when finish() has been called,
# i.e. there will be no more data sent. Data may still be in the
# stream's write buffer.
self._write_finished = False
# True when we have read the entire incoming body.
self._read_finished = False
# _finish_future resolves when all data has been written and flushed
# to the IOStream.
self._finish_future: Future[None] = Future()
# If true, the connection should be closed after this request
# (after the response has been written in the server side,
# and after it has been read in the client)
self._disconnect_on_finish = False
self._clear_callbacks()
# Save the start lines after we read or write them; they
# affect later processing (e.g. 304 responses and HEAD methods
# have content-length but no bodies)
self._request_start_line: httputil.RequestStartLine | None = None
self._response_start_line: httputil.ResponseStartLine | None = None
self._request_headers: httputil.HTTPHeaders | None = None
# True if we are writing output with chunked encoding.
self._chunking_output = False
# While reading a body with a content-length, this is the
# amount left to read.
self._expected_content_remaining: int | None = None
# A Future for our outgoing writes, returned by IOStream.write.
self._pending_write: Future[None] | None = None
[docs]
def read_response(self, delegate: httputil.HTTPMessageDelegate) -> Awaitable[bool]:
"""Read a single HTTP response.
Typical client-mode usage is to write a request using `write_headers`,
`write`, and `finish`, and then call ``read_response``.
:arg delegate: a `.HTTPMessageDelegate`
Returns a `.Future` that resolves to a bool after the full response has
been read. The result is true if the stream is still open.
"""
if self.params.decompress:
delegate = _GzipMessageDelegate(
delegate, self.params.chunk_size, self._max_body_size
)
return self._read_message(delegate)
async def _read_message(self, delegate: httputil.HTTPMessageDelegate) -> bool:
try:
return await self._read_message_inner(delegate)
except _QuietException:
# The delegate raised an exception, which has already been logged,
# and the delegate has been told that the connection is closed.
# Close it, instead of letting _QuietException escape to our caller.
self.close()
return False
async def _read_message_inner(
self, delegate: httputil.HTTPMessageDelegate, num_1xx: int = 0
) -> bool:
need_delegate_close = False
try:
header_future = self.stream.read_until_regex(
b"\r?\n\r?\n", max_bytes=self.params.max_header_size
)
if self.params.header_timeout is None:
header_data = await header_future
else:
try:
header_data = await gen.with_timeout(
self.stream.io_loop.time() + self.params.header_timeout,
header_future,
quiet_exceptions=iostream.StreamClosedError,
)
except TimeoutError:
self.close()
return False
start_line_str, headers = self._parse_headers(header_data)
if self.is_client:
resp_start_line = httputil.parse_response_start_line(start_line_str)
self._response_start_line = resp_start_line
start_line: httputil.RequestStartLine | httputil.ResponseStartLine = (
resp_start_line
)
# TODO: this will need to change to support client-side keepalive
self._disconnect_on_finish = False
else:
req_start_line = httputil.parse_request_start_line(start_line_str)
self._request_start_line = req_start_line
self._request_headers = headers
start_line = req_start_line
self._disconnect_on_finish = not self._can_keep_alive(
req_start_line, headers
)
need_delegate_close = True
with _ExceptionLoggingContext(app_log):
header_recv_future = delegate.headers_received(start_line, headers)
if header_recv_future is not None:
await header_recv_future
if self.stream is None:
# We've been detached.
need_delegate_close = False
return False
skip_body = False
if self.is_client:
assert isinstance(start_line, httputil.ResponseStartLine)
if (
self._request_start_line is not None
and self._request_start_line.method == "HEAD"
):
skip_body = True
code = start_line.code
if code == 304:
# 304 responses may include the content-length header
# but do not actually have a body.
# http://tools.ietf.org/html/rfc7230#section-3.3
skip_body = True
if 100 <= code < 200:
# 1xx responses should never indicate the presence of
# a body.
if "Content-Length" in headers or "Transfer-Encoding" in headers:
raise httputil.HTTPInputError(
"Response code %d cannot have body" % code
)
if num_1xx >= _MAX_1XX_RESPONSES:
raise httputil.HTTPInputError("Too many 1xx responses")
# TODO: client delegates will get headers_received twice
# in the case of a 100-continue. Document or change?
#
# The recursive call reads the real response and owns
# the delegate from here on, so there is nothing left
# for this frame to do. Clear need_delegate_close so
# that the finally block does not call
# on_connection_close() on an already-finished
# delegate.
need_delegate_close = False
return await self._read_message_inner(delegate, num_1xx + 1)
else:
if headers.get("Expect") == "100-continue" and not self._write_finished:
self.stream.write(b"HTTP/1.1 100 (Continue)\r\n\r\n")
if not skip_body:
body_future = self._read_body(
resp_start_line.code if self.is_client else 0, headers, delegate
)
if body_future is not None:
if self._body_timeout is None:
await body_future
else:
try:
await gen.with_timeout(
self.stream.io_loop.time() + self._body_timeout,
body_future,
quiet_exceptions=iostream.StreamClosedError,
)
except TimeoutError:
gen_log.info("Timeout reading body from %s", self.context)
self.stream.close()
return False
self._read_finished = True
if not self._write_finished or self.is_client:
with _ExceptionLoggingContext(app_log):
delegate.finish()
# If finish() raised, the delegate is told that the connection
# is closed, as when any of its other methods raise.
need_delegate_close = False
# If we're waiting for the application to produce an asynchronous
# response, and we're not detached, register a close callback
# on the stream (we didn't need one while we were reading)
if (
not self._finish_future.done()
and self.stream is not None
and not self.stream.closed()
):
self.stream.set_close_callback(self._on_connection_close)
await self._finish_future
if self.is_client and self._disconnect_on_finish:
self.close()
if self.stream is None:
return False
except httputil.HTTPInputError as e:
gen_log.info("Malformed HTTP message from %s: %s", self.context, e)
if not self.is_client:
await self.stream.write(b"HTTP/1.1 400 Bad Request\r\n\r\n")
elif not need_delegate_close:
# The error came before headers_received, so the delegate
# doesn't know about this message yet. Tell it the
# connection is closed so a client request doesn't hang.
need_delegate_close = True
self.close()
return False
finally:
if need_delegate_close:
with _ExceptionLoggingContext(app_log):
delegate.on_connection_close()
header_future = None # type: ignore
self._clear_callbacks()
return True
def _clear_callbacks(self) -> None:
"""Clears the callback attributes.
This allows the request handler to be garbage collected more
quickly in CPython by breaking up reference cycles.
"""
self._write_callback = None
self._write_future: Future[None] | None = None
self._close_callback: Callable[[], None] | None = None
if self.stream is not None:
self.stream.set_close_callback(None)
[docs]
def set_close_callback(self, callback: Callable[[], None] | None) -> None:
"""Sets a callback that will be run when the connection is closed.
Note that this callback is slightly different from
`.HTTPMessageDelegate.on_connection_close`: The
`.HTTPMessageDelegate` method is called when the connection is
closed while receiving a message. This callback is used when
there is not an active delegate (for example, on the server
side this callback is used if the client closes the connection
after sending its request but before receiving all the
response.
"""
self._close_callback = callback
def _on_connection_close(self) -> None:
# Note that this callback is only registered on the IOStream
# when we have finished reading the request and are waiting for
# the application to produce its response.
if self._close_callback is not None:
callback = self._close_callback
self._close_callback = None
callback()
if not self._finish_future.done():
future_set_result_unless_cancelled(self._finish_future, None)
self._clear_callbacks()
def close(self) -> None:
if self.stream is not None:
self.stream.close()
self._clear_callbacks()
if not self._finish_future.done():
future_set_result_unless_cancelled(self._finish_future, None)
[docs]
def detach(self) -> iostream.IOStream:
"""Take control of the underlying stream.
Returns the underlying `.IOStream` object and stops all further
HTTP processing. May only be called during
`.HTTPMessageDelegate.headers_received`. Intended for implementing
protocols like websockets that tunnel over an HTTP handshake.
"""
self._clear_callbacks()
stream = self.stream
self.stream = None # type: ignore
if not self._finish_future.done():
future_set_result_unless_cancelled(self._finish_future, None)
return stream
[docs]
def set_body_timeout(self, timeout: float) -> None:
"""Sets the body timeout for a single request.
Overrides the value from `.HTTP1ConnectionParameters`.
"""
self._body_timeout = timeout
[docs]
def set_max_body_size(self, max_body_size: int) -> None:
"""Sets the body size limit for a single request.
Overrides the value from `.HTTP1ConnectionParameters`.
"""
self._max_body_size = max_body_size
def _format_chunk(self, chunk: bytes) -> bytes:
if self._expected_content_remaining is not None:
self._expected_content_remaining -= len(chunk)
if self._expected_content_remaining < 0:
# Close the stream now to stop further framing errors.
self.stream.close()
raise httputil.HTTPOutputError(
"Tried to write more data than Content-Length"
)
if self._chunking_output and chunk:
# Don't write out empty chunks because that means END-OF-STREAM
# with chunked encoding
return utf8("%x" % len(chunk)) + b"\r\n" + chunk + b"\r\n"
else:
return chunk
[docs]
def write(self, chunk: bytes) -> "Future[None]":
"""Implements `.HTTPConnection.write`.
For backwards compatibility it is allowed but deprecated to
skip `write_headers` and instead call `write()` with a
pre-encoded header block.
"""
future = None
if self.stream.closed():
future = self._write_future = Future()
self._write_future.set_exception(iostream.StreamClosedError())
self._write_future.exception()
else:
future = self._write_future = Future()
self._pending_write = self.stream.write(self._format_chunk(chunk))
future_add_done_callback(self._pending_write, self._on_write_complete)
return future
[docs]
def finish(self) -> None:
"""Implements `.HTTPConnection.finish`."""
if (
self._expected_content_remaining is not None
and self._expected_content_remaining != 0
and not self.stream.closed()
):
self.stream.close()
raise httputil.HTTPOutputError(
"Tried to write %d bytes less than Content-Length"
% self._expected_content_remaining
)
if self._chunking_output:
if not self.stream.closed():
self._pending_write = self.stream.write(b"0\r\n\r\n")
self._pending_write.add_done_callback(self._on_write_complete)
self._write_finished = True
# If the app finished the request while we're still reading,
# divert any remaining data away from the delegate and
# close the connection when we're done sending our response.
# Closing the connection is the only way to avoid reading the
# whole input body.
if not self._read_finished:
self._disconnect_on_finish = True
# No more data is coming, so instruct TCP to send any remaining
# data immediately instead of waiting for a full packet or ack.
self.stream.set_nodelay(True)
if self._pending_write is None:
self._finish_request(None)
else:
future_add_done_callback(self._pending_write, self._finish_request)
def _on_write_complete(self, future: "Future[None]") -> None:
exc = future.exception()
if exc is not None and not isinstance(exc, iostream.StreamClosedError):
future.result()
if self._write_callback is not None:
callback = self._write_callback
self._write_callback = None
self.stream.io_loop.add_callback(callback)
if self._write_future is not None:
future = self._write_future
self._write_future = None
future_set_result_unless_cancelled(future, None)
def _can_keep_alive(
self, start_line: httputil.RequestStartLine, headers: httputil.HTTPHeaders
) -> bool:
if self.params.no_keep_alive:
return False
connection_header = headers.get("Connection")
if connection_header is not None:
connection_header = connection_header.lower()
if start_line.version == "HTTP/1.1":
return connection_header != "close"
elif (
"Content-Length" in headers
or is_transfer_encoding_chunked(headers)
or getattr(start_line, "method", None) in ("HEAD", "GET")
):
# start_line may be a request or response start line; only
# the former has a method attribute.
return connection_header == "keep-alive"
return False
def _finish_request(self, future: "Optional[Future[None]]") -> None:
self._clear_callbacks()
if not self.is_client and self._disconnect_on_finish:
self.close()
return
# Turn Nagle's algorithm back on, leaving the stream in its
# default state for the next request.
self.stream.set_nodelay(False)
if not self._finish_future.done():
future_set_result_unless_cancelled(self._finish_future, None)
def _parse_headers(self, data: bytes) -> tuple[str, httputil.HTTPHeaders]:
# The lstrip removes newlines that some implementations sometimes
# insert between messages of a reused connection. Per RFC 7230,
# we SHOULD ignore at least one empty line before the request.
# http://tools.ietf.org/html/rfc7230#section-3.5
data_str = native_str(data.decode("latin1")).lstrip("\r\n")
# RFC 7230 section allows for both CRLF and bare LF.
eol = data_str.find("\n")
start_line = data_str[:eol].rstrip("\r")
headers = httputil.HTTPHeaders.parse(data_str[eol:])
return start_line, headers
def _read_body(
self,
code: int,
headers: httputil.HTTPHeaders,
delegate: httputil.HTTPMessageDelegate,
) -> Awaitable[None] | None:
if "Content-Length" in headers:
if "," in headers["Content-Length"]:
# Proxies sometimes cause Content-Length headers to get
# duplicated. If all the values are identical then we can
# use them but if they differ it's an error.
pieces = re.split(r",\s*", headers["Content-Length"])
if any(i != pieces[0] for i in pieces):
raise httputil.HTTPInputError(
"Multiple unequal Content-Lengths: %r"
% headers["Content-Length"]
)
headers["Content-Length"] = pieces[0]
try:
content_length: int | None = parse_int(headers["Content-Length"])
except ValueError:
# Handles non-integer Content-Length value.
raise httputil.HTTPInputError(
"Only integer Content-Length is allowed: %s"
% headers["Content-Length"]
)
if cast(int, content_length) > self._max_body_size:
raise httputil.HTTPInputError("Content-Length too long")
else:
content_length = None
is_chunked = is_transfer_encoding_chunked(headers)
if code == 204:
# This response code is not allowed to have a non-empty body,
# and has an implicit length of zero instead of read-until-close.
# http://www.w3.org/Protocols/rfc2616/rfc2616-sec4.html#sec4.3
if is_chunked or content_length not in (None, 0):
raise httputil.HTTPInputError(
"Response with code %d should not have body" % code
)
content_length = 0
if is_chunked:
return self._read_chunked_body(delegate)
if content_length is not None:
return self._read_fixed_body(content_length, delegate)
if self.is_client:
return self._read_body_until_close(delegate)
return None
async def _deliver_body(
self, delegate: httputil.HTTPMessageDelegate, data: bytes
) -> None:
if not self._write_finished or self.is_client:
with _ExceptionLoggingContext(app_log):
ret = delegate.data_received(data)
if ret is not None:
await ret
async def _read_fixed_body(
self, content_length: int, delegate: httputil.HTTPMessageDelegate
) -> None:
work = 0 # See _YIELD_THRESHOLD
while content_length > 0:
body = await self.stream.read_bytes(
min(self.params.chunk_size, content_length), partial=True
)
content_length -= len(body)
await self._deliver_body(delegate, body)
work += len(body)
if work >= _YIELD_THRESHOLD:
work = 0
await asyncio.sleep(0)
async def _read_chunked_body(self, delegate: httputil.HTTPMessageDelegate) -> None:
# TODO: "chunk extensions" http://tools.ietf.org/html/rfc2616#section-3.6.1
total_size = 0
# Chunk boundaries are not meaningful (and may be changed by
# intermediaries), so the contents of consecutive small chunks are
# coalesced before being passed to the delegate. This bounds the
# number of data_received calls (and the overhead of delegates that
# buffer each piece they receive) regardless of how small the chunks
# are. Pending data is delivered whenever we would have to wait for
# the network, so this does not add latency for streaming delegates.
pending = bytearray()
work = 0 # See _YIELD_THRESHOLD
def read(future: Awaitable[bytes]) -> Awaitable[bytes]:
if pending and not _is_done(future):
return flush_and_read(future)
return future
async def flush_and_read(future: Awaitable[bytes]) -> bytes:
data = bytes(pending)
pending.clear()
await self._deliver_body(delegate, data)
return await future
while True:
work += _CHUNK_YIELD_COST
if work >= _YIELD_THRESHOLD:
work = 0
await asyncio.sleep(0)
chunk_len_str = await read(self.stream.read_until(b"\r\n", max_bytes=64))
try:
chunk_len = parse_hex_int(native_str(chunk_len_str[:-2]))
except ValueError:
raise httputil.HTTPInputError("invalid chunk size")
if chunk_len == 0:
crlf = await read(self.stream.read_bytes(2))
if crlf != b"\r\n":
raise httputil.HTTPInputError(
"improperly terminated chunked request"
)
if pending:
await self._deliver_body(delegate, bytes(pending))
return
total_size += chunk_len
if total_size > self._max_body_size:
raise httputil.HTTPInputError("chunked body too large")
bytes_to_read = chunk_len
while bytes_to_read:
chunk = await read(
self.stream.read_bytes(
min(bytes_to_read, self.params.chunk_size), partial=True
)
)
bytes_to_read -= len(chunk)
if not pending and len(chunk) == self.params.chunk_size:
# Full-sized reads are passed through without copying.
await self._deliver_body(delegate, chunk)
else:
pending += chunk
if len(pending) >= self.params.chunk_size:
data = bytes(pending[: self.params.chunk_size])
del pending[: self.params.chunk_size]
await self._deliver_body(delegate, data)
work += len(chunk)
if work >= _YIELD_THRESHOLD:
work = 0
await asyncio.sleep(0)
# chunk ends with \r\n
crlf = await read(self.stream.read_bytes(2))
if crlf != b"\r\n":
raise httputil.HTTPInputError("improperly terminated chunk")
async def _read_body_until_close(
self, delegate: httputil.HTTPMessageDelegate
) -> None:
# The body is terminated by the connection closing, so there is no
# length known in advance. Read incrementally so that max_body_size
# is enforced before an over-large body has been buffered, and so
# that the body is not limited by the stream's read buffer size.
total_size = 0
work = 0 # See _YIELD_THRESHOLD
while True:
try:
body = await self.stream.read_bytes(
self.params.chunk_size, partial=True
)
except iostream.StreamClosedError:
# The connection closing is the normal end of this body.
return
total_size += len(body)
if total_size > self._max_body_size:
raise httputil.HTTPInputError("Body too long")
await self._deliver_body(delegate, body)
work += len(body)
if work >= _YIELD_THRESHOLD:
work = 0
await asyncio.sleep(0)
class _GzipMessageDelegate(httputil.HTTPMessageDelegate):
"""Wraps an `HTTPMessageDelegate` to decode ``Content-Encoding: gzip``."""
def __init__(
self,
delegate: httputil.HTTPMessageDelegate,
chunk_size: int,
max_body_size: int,
) -> None:
self._delegate = delegate
self._chunk_size = chunk_size
self._max_body_size = max_body_size
self._decompressed_body_size = 0
self._decompressor: GzipDecompressor | None = None
def headers_received(
self,
start_line: httputil.RequestStartLine | httputil.ResponseStartLine,
headers: httputil.HTTPHeaders,
) -> Awaitable[None] | None:
if headers.get("Content-Encoding", "").lower() == "gzip":
self._decompressor = GzipDecompressor()
# Downstream delegates will only see uncompressed data,
# so rename the content-encoding header.
# (but note that curl_httpclient doesn't do this).
headers.add("X-Consumed-Content-Encoding", headers["Content-Encoding"])
del headers["Content-Encoding"]
return self._delegate.headers_received(start_line, headers)
async def data_received(self, chunk: bytes) -> None:
if self._decompressor:
compressed_data = chunk
while compressed_data:
try:
decompressed = self._decompressor.decompress(
compressed_data, self._chunk_size
)
except zlib.error as e:
raise httputil.HTTPInputError("invalid gzip data: %s" % e)
if decompressed:
self._decompressed_body_size += len(decompressed)
if self._decompressed_body_size > self._max_body_size:
raise httputil.HTTPInputError("decompressed body too large")
ret = self._delegate.data_received(decompressed)
if ret is not None:
await ret
tail = self._decompressor.unconsumed_tail
# A call that finishes a gzip member may return no output,
# but it must consume some input.
if not decompressed and len(tail) >= len(compressed_data):
raise httputil.HTTPInputError(
"encountered unconsumed gzip data without making progress"
)
compressed_data = tail
else:
ret = self._delegate.data_received(chunk)
if ret is not None:
await ret
def finish(self) -> None:
if self._decompressor is not None:
try:
# data_received has passed all the data to decompress, so
# this returns nothing; it checks for truncated input.
self._decompressor.flush()
except zlib.error as e:
# The wrapped delegate will never see finish(), and since this
# delegate has been finished, HTTP1Connection will not call
# our on_connection_close either. Close the wrapped delegate
# here so it gets exactly one of the two.
self._delegate.on_connection_close()
raise httputil.HTTPInputError("invalid gzip data: %s" % e)
return self._delegate.finish()
def on_connection_close(self) -> None:
return self._delegate.on_connection_close()
[docs]
class HTTP1ServerConnection:
"""An HTTP/1.x server."""
def __init__(
self,
stream: iostream.IOStream,
params: HTTP1ConnectionParameters | None = None,
context: object | None = None,
) -> None:
"""
:arg stream: an `.IOStream`
:arg params: a `.HTTP1ConnectionParameters` or None
:arg context: an opaque application-defined object that is accessible
as ``connection.context``
"""
self.stream = stream
if params is None:
params = HTTP1ConnectionParameters()
self.params = params
self.context = context
self._serving_future: Future[None] | None = None
[docs]
async def close(self) -> None:
"""Closes the connection.
Returns a `.Future` that resolves after the serving loop has exited.
"""
self.stream.close()
# Block until the serving loop is done, but ignore any exceptions
# (start_serving is already responsible for logging them).
assert self._serving_future is not None
try:
await self._serving_future
except Exception:
pass
[docs]
def start_serving(self, delegate: httputil.HTTPServerConnectionDelegate) -> None:
"""Starts serving requests on this connection.
:arg delegate: a `.HTTPServerConnectionDelegate`
"""
assert isinstance(delegate, httputil.HTTPServerConnectionDelegate)
fut = gen.convert_yielded(self._server_request_loop(delegate))
self._serving_future = fut
# Register the future on the IOLoop so its errors get logged.
self.stream.io_loop.add_future(fut, lambda f: f.result())
async def _server_request_loop(
self, delegate: httputil.HTTPServerConnectionDelegate
) -> None:
try:
while True:
conn = HTTP1Connection(self.stream, False, self.params, self.context)
request_delegate = delegate.start_request(self, conn)
try:
ret = await conn.read_response(request_delegate)
except (
iostream.StreamClosedError,
iostream.UnsatisfiableReadError,
asyncio.CancelledError,
):
return
except Exception:
gen_log.error("Uncaught exception", exc_info=True)
conn.close()
return
if not ret:
return
await asyncio.sleep(0)
finally:
delegate.on_close(self)
DIGITS = re.compile(r"[0-9]+")
HEXDIGITS = re.compile(r"[0-9a-fA-F]+")
[docs]
def parse_int(s: str) -> int:
"""Parse a non-negative integer from a string."""
if DIGITS.fullmatch(s) is None:
raise ValueError("not an integer: %r" % s)
return int(s)
[docs]
def parse_hex_int(s: str) -> int:
"""Parse a non-negative hexadecimal integer from a string."""
if HEXDIGITS.fullmatch(s) is None:
raise ValueError("not a hexadecimal integer: %r" % s)
return int(s, 16)
[docs]
def is_transfer_encoding_chunked(headers: httputil.HTTPHeaders) -> bool:
"""Returns true if the headers specify Transfer-Encoding: chunked.
Raise httputil.HTTPInputError if any other transfer encoding is used.
"""
# Note that transfer-encoding is an area in which postel's law can lead
# us astray. If a proxy and a backend server are liberal in what they accept,
# but accept slightly different things, this can lead to mismatched framing
# and request smuggling issues. Therefore we are as strict as possible here
# (even technically going beyond the requirements of the RFCs: a value of
# ",chunked" is legal but doesn't appear in practice for legitimate traffic)
if "Transfer-Encoding" not in headers:
return False
if "Content-Length" in headers:
# Message cannot contain both Content-Length and
# Transfer-Encoding headers.
# http://tools.ietf.org/html/rfc7230#section-3.3.3
raise httputil.HTTPInputError(
"Message with both Transfer-Encoding and Content-Length"
)
if headers["Transfer-Encoding"].lower() == "chunked":
return True
# We do not support any transfer-encodings other than chunked, and we do not
# expect to add any support because the concept of transfer-encoding has
# been removed in HTTP/2.
raise httputil.HTTPInputError(
"Unsupported Transfer-Encoding %s" % headers["Transfer-Encoding"]
)