HTTP/1.1¶

async def read_http_request(
    reader: asyncio.StreamReader,
    *,
    max_head_size: int = 1_048_576,
    body_limit: int = 4_194_304,
    parser_mode: str = "auto",
    stream_body: bool = False,
) -> HTTPRequest | None:
    """
    Reads and parses a full HTTP request (head and body) from the reader.

    Args:
        reader (asyncio.StreamReader): The client stream reader.
        max_head_size (int): Max allowed bytes for the request line and headers.
        body_limit (int): Max allowed bytes for the body.
        parser_mode (str): Choice of parser ('httptools', 'h11', or 'auto').

    Returns:
        HTTPRequest | None: The parsed request, or None if the client disconnected.
    """
    try:
        head = await reader.readuntil(b"\r\n\r\n")
    except asyncio.LimitOverrunError as exc:
        raise ValueError("HTTP head exceeds configured limit") from exc
    except asyncio.IncompleteReadError:
        return None

    if len(head) > max_head_size:
        raise ValueError("HTTP head exceeds configured limit")

    if parser_mode == "auto":
        try:
            (
                method_raw,
                target_raw,
                version_raw,
                headers,
                content_length_raw,
                transfer_encoding,
                connection_header,
                expect_header,
                websocket_upgrade,
            ) = parse_request_head_normalized(head)
            method = method_raw.decode("latin-1")
            target = target_raw.decode("latin-1")
            version = version_raw.decode("latin-1")
        except ValueError:
            parsed_with_fallback = False
            with suppress(ValueError):
                method, target, version, raw_headers = _parse_request_head_httptools(head)
                (
                    headers,
                    content_length_raw,
                    transfer_encoding,
                    connection_header,
                    expect_header,
                    websocket_upgrade,
                ) = _prepare_request_headers(raw_headers, already_normalized=True)
                parsed_with_fallback = True
            if not parsed_with_fallback:
                method, target, version, raw_headers = _parse_request_head_h11(head)
                (
                    headers,
                    content_length_raw,
                    transfer_encoding,
                    connection_header,
                    expect_header,
                    websocket_upgrade,
                ) = _prepare_request_headers(raw_headers, already_normalized=True)
    else:
        method, target, version, raw_headers = _parse_request_head(head, parser_mode)
        (
            headers,
            content_length_raw,
            transfer_encoding,
            connection_header,
            expect_header,
            websocket_upgrade,
        ) = _prepare_request_headers(
            raw_headers,
            already_normalized=parser_mode in {"h11", "httptools"},
        )

    content_length = 0
    if content_length_raw is not None:
        try:
            content_length = int(content_length_raw)
        except ValueError as exc:
            raise ValueError("Invalid Content-Length header") from exc
        if content_length < 0:
            raise ValueError("Invalid Content-Length header")

    if content_length > body_limit:
        raise ValueError("HTTP body exceeds configured limit")

    body = b""
    body_chunks: list[bytes] = []
    body_stream: HTTPBodyStream | None = None
    if b"chunked" in transfer_encoding:
        body_chunks = await _read_chunked_body_chunks(reader, body_limit)
        body = body_chunks[0] if len(body_chunks) == 1 else b"".join(body_chunks)
    elif stream_body and content_length > 0:
        body_stream = HTTPBodyStream(reader, content_length, body_limit)
    elif content_length > 0:
        body_chunks = await _read_content_length_body_chunks(reader, content_length, body_limit)
        body = body_chunks[0] if len(body_chunks) == 1 else b"".join(body_chunks)

    if websocket_upgrade:
        request_headers: list[tuple[str, str] | tuple[bytes, bytes]] = [
            (name.decode("latin-1"), value.decode("latin-1")) for name, value in headers
        ]
        headers_normalized = False
    else:
        request_headers = cast(list[tuple[str, str] | tuple[bytes, bytes]], headers)
        headers_normalized = True

    return HTTPRequest(
        method=method,
        target=target,
        http_version=version,
        headers=request_headers,
        body=body,
        body_chunks=body_chunks,
        body_stream=body_stream,
        headers_normalized=headers_normalized,
        connection_header=connection_header,
        expect_header=expect_header,
        is_websocket_upgrade=websocket_upgrade,
    )

def build_http_scope(
    request: HTTPRequest,
    *,
    client: ClientAddress,
    server: ServerAddress,
    root_path: str,
    is_tls: bool,
    app_state: dict[str, Any] | None = None,
    asgi_version: str = "3.0",
) -> Scope:
    """
    Converts an internal HTTPRequest into an ASGI 3.0 scope dictionary.

    Args:
        request (HTTPRequest): Parsed request data.
        client (ClientAddress): IP and port of the client.
        server (ServerAddress): IP and port of the server.
        root_path (str): The mounting point of the application.
        is_tls (bool): True if connection is encrypted.
        app_state (dict[str, Any] | None): Lifespan state to shallow-copy into
            the per-request scope.
        asgi_version (str): ASGI callable version reported in the scope.

    Returns:
        Scope: A dictionary conforming to the ASGI HTTP specification.
    """
    path, _, query = request.target.partition("?")
    decoded_path = unquote(path) if "%" in path else path
    raw_path = path.encode("latin-1")
    if root_path:
        root_path_bytes = root_path.encode("latin-1")
        full_path = root_path + decoded_path
        full_raw_path = root_path_bytes + raw_path
    else:
        full_path = decoded_path
        full_raw_path = raw_path

    headers = request.headers
    scope_headers = (
        cast(list[tuple[bytes, bytes]], headers)
        if request.headers_normalized
        or all(
            isinstance(name, bytes) and isinstance(value, bytes) and name == name.lower()
            for name, value in headers
        )
        else [
            (_coerce_header_bytes(name).lower(), _coerce_header_bytes(value))
            for name, value in headers
        ]
    )

    return {
        "type": "http",
        "asgi": {"version": asgi_version, "spec_version": "2.3"},
        "http_version": request.http_version.removeprefix("HTTP/"),
        "method": request.method,
        "scheme": "https" if is_tls else "http",
        "path": full_path,
        "raw_path": full_raw_path,
        "query_string": query.encode("latin-1"),
        "root_path": root_path,
        "headers": scope_headers,
        "client": client,
        "server": server,
        "state": app_state.copy() if app_state else {},
    }

async def run_http_asgi(
    app: ASGIApplication,
    scope: Scope,
    request_body: bytes | list[bytes] | HTTPBodyStream,
    *,
    expect_100_continue: bool = False,
    on_100_continue: Callable[[], Awaitable[None]] | None = None,
    on_response_start: Callable[[HTTPResponse], Awaitable[None]] | None = None,
    on_response_body: Callable[[HTTPResponse, bytes, bool], Awaitable[None]] | None = None,
) -> HTTPResponse:
    """
    Orchestrates the ASGI request/response lifecycle with high-performance formatting.

    This function manages the strict state machine required by the ASGI HTTP specification.
    It incorporates several critical performance optimizations:

    1. **Single-Chunk Optimization:** Defers finalizing headers until the first body chunk arrives.
       If the app yields a single chunk (`more_body=False`) without a `Content-Length`, it calculates
       it instantly to avoid the overhead of HTTP chunked transfer framing.
    2. **In-Flight Chunk Framing:** If `Transfer-Encoding: chunked` is actually required, it formats
       the hex framing continuously during the app's `send()` loop rather than post-processing.



    Args:
        app (ASGIApplication): The user-provided ASGI application callable.
        scope (Scope): The ASGI connection scope dictionary.
        request_body (bytes | list[bytes]): The pre-buffered input body chunks.
        expect_100_continue (bool): If the client expects a 100-Continue response.
        on_100_continue (Callable | None): Callback to trigger the 100 status code transmission.

    Returns:
        HTTPResponse: The resulting parsed response state ready for wire encoding.

    Raises:
        RuntimeError: If the ASGI application violates the protocol sequence or crashes.
    """
    response = HTTPResponse()
    body_stream = request_body if isinstance(request_body, HTTPBodyStream) else None
    body_chunks = (
        []
        if body_stream is not None
        else request_body
        if isinstance(request_body, list)
        else [request_body]
    )
    if body_stream is None and not body_chunks:
        body_chunks = [b""]

    response_started = False
    response_complete = False
    waiting_for_100_continue = expect_100_continue
    body_index = 0
    message_complete: asyncio.Event | None = None

    chunked_encoding: bool | None = None
    expected_content_length = 0

    def mark_message_complete() -> None:
        if message_complete is not None:
            message_complete.set()

    async def wait_for_message_complete() -> None:
        nonlocal message_complete
        if response_complete:
            return
        if message_complete is None:
            message_complete = asyncio.Event()
        await message_complete.wait()

    async def _send_internal_server_error() -> None:
        """Sends a standard 500 Internal Server Error response.

        This is used as a safety mechanism when an ASGI application fails
        to start a response or crashes during its initial execution.
        """
        nonlocal response_started, response_complete, chunked_encoding, expected_content_length
        response_started = True
        response_complete = True
        chunked_encoding = False
        expected_content_length = 0
        response.status = 500
        response.headers = [
            (b"content-type", b"text/plain; charset=utf-8"),
            (b"content-length", b"21"),
            (b"connection", b"close"),
        ]
        response.headers_metadata_trusted = True
        response.has_content_length = True
        response.has_transfer_encoding = False
        response.has_connection_header = True
        response.connection_header = b"close"
        response.body_chunks = [] if scope.get("method") == "HEAD" else [b"Internal Server Error"]
        response.chunked_encoding = False
        response.suppress_body = scope.get("method") == "HEAD"
        mark_message_complete()

    async def receive() -> Message:
        """The ASGI receive channel for the application.

        Returns:
            Message: An 'http.request' or 'http.disconnect' message.
        """
        nonlocal waiting_for_100_continue, body_index
        if response_complete:
            return {"type": "http.disconnect"}

        if waiting_for_100_continue:
            waiting_for_100_continue = False
            if on_100_continue is not None:
                await on_100_continue()

        if body_stream is not None:
            if body_stream.disconnected:
                return {"type": "http.disconnect"}
            if not body_stream.complete:
                body, more_body = await body_stream.receive()
                if body_stream.disconnected and not body:
                    return {"type": "http.disconnect"}
                return {
                    "type": "http.request",
                    "body": body,
                    "more_body": more_body,
                }
            await wait_for_message_complete()
            return {"type": "http.disconnect"}

        if body_index < len(body_chunks):
            body = body_chunks[body_index]
            body_index += 1
            return {
                "type": "http.request",
                "body": body,
                "more_body": body_index < len(body_chunks),
            }

        await wait_for_message_complete()
        return {"type": "http.disconnect"}

    async def send(message: Message) -> None:
        """The ASGI send channel for the application.

        Args:
            message (Message): The message sent by the ASGI application.

        Raises:
            RuntimeError: If the application sends invalid message sequences
                or violates the ASGI HTTP specification.
        """
        nonlocal response_started, response_complete, waiting_for_100_continue
        nonlocal chunked_encoding, expected_content_length

        msg_type = message["type"]

        if msg_type == "http.response.start":
            if response_started:
                raise RuntimeError("ASGI message 'http.response.start' sent more than once.")

            response_started = True
            waiting_for_100_continue = False

            response.status = int(message.get("status", 200))
            _validate_response_status(response.status)
            response.headers = []
            response.headers_metadata_trusted = True
            response.has_content_length = False
            response.has_transfer_encoding = False
            response.has_connection_header = False
            response.connection_header = b""
            for raw_name, raw_value in message.get("headers", []):
                name = raw_name if isinstance(raw_name, bytes) else _coerce_header_bytes(raw_name)
                value = (
                    raw_value if isinstance(raw_value, bytes) else _coerce_header_bytes(raw_value)
                )
                _validate_response_header(name, value)
                response.headers.append((name, value))
                lowered_name = name if name.islower() else name.lower()
                if lowered_name == b"content-length":
                    response.has_content_length = True
                    expected_content_length = int(value.decode("latin-1"))
                    chunked_encoding = False
                elif lowered_name == b"transfer-encoding":
                    response.has_transfer_encoding = True
                    expected_content_length = 0
                    if value.lower() == b"chunked":
                        chunked_encoding = True
                elif lowered_name == b"connection":
                    response.has_connection_header = True
                    response.connection_header = value.lower()
            response.suppress_body = scope.get("method") == "HEAD"

            # Default to chunked if no explicit length/encoding (allows apps to stream without Content-Length)
            if (
                chunked_encoding is None
                and not response.suppress_body
                and response.status not in {204, 304}
            ):
                chunked_encoding = True
                response.headers.append((b"transfer-encoding", b"chunked"))
                response.has_transfer_encoding = True

            response.chunked_encoding = bool(chunked_encoding)

            # HEAD request overrides validation
            if response.suppress_body:
                expected_content_length = 0

            if on_response_start is not None:
                await on_response_start(response)
                response.streamed = True
            elif on_response_body is not None:
                response.streamed = True
            return

        elif msg_type == "http.response.body":
            if not response_started:
                raise RuntimeError(
                    "ASGI message 'http.response.body' sent before 'http.response.start'."
                )
            if response_complete:
                raise RuntimeError(
                    "ASGI message 'http.response.body' sent after response already completed."
                )

            body = message.get("body", b"")
            if not isinstance(body, bytes):
                body = bytes(body)
            more_body = bool(message.get("more_body", False))

            if response.suppress_body:
                pass  # HEAD requests intentionally drop the body
            elif chunked_encoding:
                if not response.streamed:
                    response.body_chunks.append(body)
            else:
                body_size = len(body)
                if body_size > expected_content_length:
                    raise RuntimeError("Response content longer than Content-Length")
                expected_content_length -= body_size
                if not response.streamed:
                    response.body_chunks.append(body)

            if response.streamed and on_response_body is not None:
                await on_response_body(response, body, more_body)

            if not more_body:
                if not chunked_encoding and expected_content_length != 0:
                    raise RuntimeError("Response content shorter than Content-Length")
                response_complete = True
                mark_message_complete()
        else:
            raise RuntimeError(f"Unexpected ASGI message type: '{msg_type}'.")

    try:
        result = await app(scope, receive, send)
    except BaseException as exc:
        if not response_started:
            await _send_internal_server_error()
        elif isinstance(exc, HTTPResponseStartedError):
            raise
        elif isinstance(exc, RuntimeError):
            raise HTTPResponseStartedError(str(exc), response) from exc
        else:
            raise HTTPResponseStartedError("Exception in ASGI application", response) from exc
    else:
        if result is not None:
            message = f"ASGI callable should return None, but returned '{result}'."
            if response_started:
                raise HTTPResponseStartedError(message, response)
            raise RuntimeError(message)
        if not response_started:
            await _send_internal_server_error()
        elif not response_complete:
            raise HTTPResponseStartedError(
                "ASGI callable returned without completing response.",
                response,
            )

    return response

def append_default_response_headers(
    response: HTTPResponse,
    config: PalfreyConfig,
    *,
    default_headers: list[tuple[bytes, bytes]] | None = None,
) -> None:
    """Applies configured default headers to the response.

    Injects headers like 'Server' and 'Date' if enabled and not already present.

    Args:
        response (HTTPResponse): The response object to update.
        config (PalfreyConfig): Application configuration.
        default_headers (list | None): Cached list of headers for fast-path insertion.
    """
    if default_headers is not None:
        if not response.headers:
            for name, value in default_headers:
                _append_response_header(response, name, value)
            return

        if len(default_headers) == 2:
            first_name, first_value = default_headers[0]
            second_name, second_value = default_headers[1]
            has_first = False
            has_second = False
            for name, _value in response.headers:
                if name == first_name:
                    has_first = True
                elif name == second_name:
                    has_second = True
                elif not name.islower():
                    lowered_name = name.lower()
                    if lowered_name == first_name:
                        has_first = True
                    elif lowered_name == second_name:
                        has_second = True
                if has_first and has_second:
                    break
            if not has_first:
                _append_response_header(response, first_name, first_value)
            if not has_second:
                _append_response_header(response, second_name, second_value)
            return

        for name, value in default_headers:
            if _has_header(response.headers, name):
                continue
            _append_response_header(response, name, value)
        return

    existing_headers = {name.lower() for name, _ in response.headers}

    configured_headers = config.normalized_headers
    configured_header_names = {name.lower() for name, _ in configured_headers}

    if (
        config.server_header
        and b"server" not in existing_headers
        and "server" not in configured_header_names
    ):
        _append_response_header(response, b"server", _SERVER_HEADER_VALUE)

    if (
        config.date_header
        and b"date" not in existing_headers
        and "date" not in configured_header_names
    ):
        _append_response_header(response, b"date", _http_date_header())

    for name, value in configured_headers:
        _append_response_header(response, name.encode("latin-1"), value.encode("latin-1"))

def encode_http_response_head(response: HTTPResponse, keep_alive: bool) -> Iterable[bytes]:
    """Serializes the HTTP response status line and headers.

    Handles status line and headers while preserving chunked transfer metadata
    for callers that stream body chunks separately.

    Args:
        response (HTTPResponse): The response to serialize.
        keep_alive (bool): Whether the connection should be kept alive.

    Yields:
        bytes: Serialized status and header fragments.
    """
    if response.status in _STATUS_LINES:
        yield _STATUS_LINES[response.status]
    else:
        try:
            reason = http.HTTPStatus(response.status).phrase
        except ValueError:
            reason = ""
        yield f"HTTP/1.1 {response.status} {reason}\r\n".encode("ascii")

    metadata_trusted = response.headers_metadata_trusted
    has_content_length = response.has_content_length if metadata_trusted else False
    has_transfer_encoding = response.has_transfer_encoding if metadata_trusted else False
    has_connection = response.has_connection_header if metadata_trusted else False

    for name, value in response.headers:
        if not metadata_trusted:
            if name == b"content-length":
                has_content_length = True
            elif name == b"transfer-encoding":
                has_transfer_encoding = True
            elif name == b"connection":
                has_connection = True
            elif not name.islower():
                lowered_name = name.lower()
                if lowered_name == b"content-length":
                    has_content_length = True
                elif lowered_name == b"transfer-encoding":
                    has_transfer_encoding = True
                elif lowered_name == b"connection":
                    has_connection = True

        yield name
        yield _HEADER_SEPARATOR
        yield value
        yield _CRLF

    if not has_content_length and not has_transfer_encoding:
        payload_len = 0 if response.suppress_body else sum(len(c) for c in response.body_chunks)
        yield b"content-length: "
        yield str(payload_len).encode("ascii")
        yield _CRLF

    if not has_connection and not keep_alive:
        yield _CONNECTION_CLOSE

    yield _CRLF

def encode_http_response_body_chunk(
    response: HTTPResponse,
    body: bytes,
    *,
    more_body: bool,
) -> Iterable[bytes]:
    """Serializes a single ASGI response body event into wire bytes."""
    if response.suppress_body:
        return
    if response.chunked_encoding:
        if body:
            yield f"{len(body):x}\r\n".encode("ascii")
            yield body
            yield _CRLF
        if not more_body:
            yield b"0\r\n\r\n"
        return
    if body:
        yield body