WebSocket¶

def build_websocket_scope(
    *,
    target: str,
    headers: list[tuple[str, str]],
    client: ClientAddress,
    server: ServerAddress,
    root_path: str,
    is_tls: bool,
    protocol_header: str | None = None,
) -> Scope:
    """
    Construct an ASGI scope dictionary for a WebSocket connection.

    This scope includes connection metadata such as path, headers, and
    negotiated subprotocols required by the ASGI 3.0 specification.

    Args:
        target (str): The raw request target string.
        headers (list[tuple[str, str]]): Initial HTTP upgrade headers.
        client (ClientAddress): IP and port of the connecting client.
        server (ServerAddress): IP and port of the server.
        root_path (str): The ASGI root path.
        is_tls (bool): True if the connection is encrypted.
        protocol_header (str | None): Optional pre-extracted protocol header.

    Returns:
        Scope: A dictionary representing the connection scope.
    """
    path, _, query = target.partition("?")
    decoded_path = unquote(path)
    raw_path = path.encode("latin-1")
    root_path_bytes = root_path.encode("latin-1")
    full_path = root_path + decoded_path
    full_raw_path = root_path_bytes + raw_path

    if protocol_header is None:
        protocol_header = _header_value(headers, "sec-websocket-protocol")
    subprotocols = []
    if protocol_header:
        subprotocols = [item.strip() for item in protocol_header.split(",") if item.strip()]

    return {
        "type": "websocket",
        "asgi": {"version": "3.0", "spec_version": "2.3"},
        "http_version": "1.1",
        "scheme": "wss" if is_tls else "ws",
        "path": full_path,
        "raw_path": full_raw_path,
        "query_string": query.encode("latin-1"),
        "root_path": root_path,
        "headers": [
            (name.lower().encode("latin-1"), value.encode("latin-1")) for name, value in headers
        ],
        "client": client,
        "server": server,
        "subprotocols": subprotocols,
        "state": {},
        "extensions": {"websocket.http.response": {}},
    }

def build_handshake_response(
    headers: list[tuple[str, str]],
    *,
    subprotocol: str | None,
    extra_headers: list[tuple[bytes, bytes]] | None = None,
) -> bytes:
    """
    Extract the client key and build a full Switching Protocols handshake response.

    Args:
        headers (list[tuple[str, str]]): Incoming HTTP request headers.
        subprotocol (str | None): The negotiated subprotocol to include in headers.
        extra_headers (list[tuple[bytes, bytes]] | None): Additional server headers.

    Returns:
        bytes: Serialized HTTP 101 response bytes.

    Raises:
        ValueError: If the 'Sec-WebSocket-Key' header is missing from the request.
    """
    client_key = _header_value(headers, "sec-websocket-key")
    if not client_key:
        raise ValueError("Missing Sec-WebSocket-Key")

    return _build_handshake_response_for_key(
        client_key,
        subprotocol=subprotocol,
        extra_headers=extra_headers,
    )

async def handle_websocket(
    app: ASGIApplication,
    config: PalfreyConfig,
    *,
    reader: asyncio.StreamReader,
    writer: asyncio.StreamWriter,
    headers: list[tuple[str, str]],
    target: str,
    client: ClientAddress,
    server: ServerAddress,
    is_tls: bool,
) -> None:
    """
    Handle the ASGI WebSocket flow for a connection, dispatching to the configured backend.

    Backend dispatching follows Uvicorn semantics:
    - 'wsproto' -> Use the wsproto engine.
    - 'websockets-sansio' -> Use the websockets sans-io implementation.
    - 'websockets' -> Use the full websockets library backend.
    - 'none' -> Use the core manual framing backend.
    """

    selected_ws = config.effective_ws
    if selected_ws == "wsproto":
        await _handle_websocket_wsproto_backend(
            app,
            config,
            reader=reader,
            writer=writer,
            headers=headers,
            target=target,
            client=client,
            server=server,
            is_tls=is_tls,
        )
        return

    if selected_ws == "websockets-sansio":
        await _handle_websocket_websockets_sansio_backend(
            app,
            config,
            reader=reader,
            writer=writer,
            headers=headers,
            target=target,
            client=client,
            server=server,
            is_tls=is_tls,
        )
        return

    if selected_ws == "none":
        await _handle_websocket_core(
            app,
            config,
            reader=reader,
            writer=writer,
            headers=headers,
            target=target,
            client=client,
            server=server,
            is_tls=is_tls,
        )
        return

    await _handle_websocket_websockets_backend(
        app,
        config,
        reader=reader,
        writer=writer,
        headers=headers,
        target=target,
        client=client,
        server=server,
        is_tls=is_tls,
    )