Function parse_headers

src/websockets/http11.py:329–377

Parse HTTP headers. Non-ASCII characters are represented with surrogate escapes. Args: read_line: Generator-based coroutine that reads a LF-terminated line or raises an exception if there isn't enough data. Raises: EOFError: If the connection is closed

(
    read_line: Callable[[int], Generator[None, None, bytes | bytearray]],
)

def parse_headers(
    read_line: Callable[[int], Generator[None, None, bytes | bytearray]],
) -> Generator[None, None, Headers]:
    """
    Parse HTTP headers.

    Non-ASCII characters are represented with surrogate escapes.

    Args:
        read_line: Generator-based coroutine that reads a LF-terminated line
            or raises an exception if there isn't enough data.

    Raises:
        EOFError: If the connection is closed without complete headers.
        SecurityError: If the request exceeds a security limit.
        ValueError: If the request isn't well formatted.

    """
    # https://datatracker.ietf.org/doc/html/rfc7230#section-3.2

    # We don't attempt to support obsolete line folding.

    headers = Headers()
    for _ in range(MAX_NUM_HEADERS + 1):
        try:
            line = yield from parse_line(read_line)
        except EOFError as exc:
            raise EOFError("connection closed while reading HTTP headers") from exc
        if line == b"":
            break

        try:
            raw_name, raw_value = line.split(b":", 1)
        except ValueError:  # not enough values to unpack (expected 2, got 1)
            raise ValueError(f"invalid HTTP header line: {d(line)}") from None
        if not _token_re.fullmatch(raw_name):
            raise ValueError(f"invalid HTTP header name: {d(raw_name)}")
        raw_value = raw_value.strip(b" \t")
        if not _value_re.fullmatch(raw_value):
            raise ValueError(f"invalid HTTP header value: {d(raw_value)}")

        name = raw_name.decode("ascii")  # guaranteed to be ASCII at this point
        value = raw_value.decode("ascii", "surrogateescape")
        headers[name] = value

    else:
        raise SecurityError("too many HTTP headers")

    return headers


def read_body(