Acceleration¶

def parse_header_items(headers: Sequence[str]) -> list[tuple[str, str]]:
    """
    Parses a sequence of raw header strings into a list of key-value tuples.

    Each string in the input sequence is expected to be in the format 'name:value'. The
    function trims whitespace from the name and leading whitespace from the value.

    Args:
        headers (Sequence[str]): A sequence of header strings, such as those provided via
            command-line interfaces.

    Returns:
        list[tuple[str, str]]: A list of tuples where the first element is the stripped
            header name and the second is the left-stripped header value.

    Raises:
        HeaderParseError: If a header string does not contain the mandatory colon separator.
    """
    if not headers:
        return []

    # Attempt to use the Rust-accelerated implementation if available for performance
    if HAS_RUST_EXTENSION and _parse_header_items is not None:
        try:
            return list(_parse_header_items(list(headers)))
        except ValueError as exc:
            raise HeaderParseError(str(exc)) from exc

    # Pure-Python fallback when Rust extension is unavailable or disabled
    parsed: list[tuple[str, str]] = []
    for item in headers:
        # Partition splits the string at the first occurrence of the separator
        name, separator, value = item.partition(":")
        if not separator:
            raise HeaderParseError(f"Invalid header '{item}'. Expected 'name:value'.")
        parsed.append((name.strip(), value.lstrip()))
    return parsed

def split_csv_values(value: str) -> list[str]:
    """
    Splits a comma-separated string into a list of individual, normalized values.

    This function removes surrounding whitespace from each segment and filters out any
    resulting empty strings to ensure a clean list of values.

    Args:
        value (str): The raw comma-separated string to be processed.

    Returns:
        list[str]: A list of non-empty, trimmed strings extracted from the input.
    """
    if HAS_RUST_EXTENSION and _split_csv_values is not None:
        return list(_split_csv_values(value))

    # Fallback to pure-Python list comprehension for splitting and stripping
    return [segment.strip() for segment in value.split(",") if segment.strip()]

def parse_request_head(data: bytes) -> tuple[str, str, str, list[tuple[str, str]]]:
    """
    Parses raw HTTP request head bytes into structured components.

    Processes the initial request line and subsequent headers. The input data should
    ideally contain the full header block ending with the standard CRLFCRLF delimiter.

    Args:
        data (bytes): The raw byte sequence representing the HTTP request head.

    Returns:
        tuple[str, str, str, list[tuple[str, str]]]: A four-element tuple containing
            (method, target, version, headers), where headers is a list of name-value pairs.

    Raises:
        ValueError: If the request line is missing, malformed, or if any header line is
            not correctly formatted.
    """
    if HAS_RUST_EXTENSION and _parse_request_head is not None:
        rust_result = _parse_request_head(data)

        if isinstance(rust_result[0], bytes):
            rust_method, rust_target, rust_version, rust_headers = cast(
                "tuple[bytes, bytes, bytes, list[tuple[bytes, bytes]]]", rust_result
            )
            method = rust_method.decode("latin-1")
            target = rust_target.decode("latin-1")
            version = rust_version.decode("latin-1")
            headers = [
                (name.decode("latin-1"), value.decode("latin-1")) for name, value in rust_headers
            ]
            return method, target, version, headers

        return cast("tuple[str, str, str, list[tuple[str, str]]]", rust_result)

    # Use latin-1 decoding to preserve the original byte values as per HTTP specs
    decoded = data.decode("latin-1")

    lines = decoded.split("\r\n")
    if not lines or not lines[0]:
        raise ValueError("Missing request line")

    request_line_parts = lines[0].split(" ")
    if len(request_line_parts) != 3:
        raise ValueError("Invalid request line")

    method, target, version = request_line_parts
    headers: list[tuple[str, str]] = []

    # Iterate through lines following the request line
    for line in lines[1:]:
        if not line:
            break
        name, separator, value = line.partition(":")
        if not separator:
            raise ValueError(f"Malformed header line: {line!r}")
        headers.append((name.strip(), value.lstrip()))

    return method, target, version, headers

def unmask_websocket_payload(payload: WebSocketPayloadBuffer, masking_key: bytes) -> bytes:
    """
    Applies the WebSocket XOR masking algorithm to a payload.

    WebSockets require client-to-server frames to be masked using a 4-byte key. This
    function reverses that mask to retrieve the original data (or applies it).

    Args:
        payload (WebSocketPayloadBuffer): The masked (or unmasked) payload data buffer.
        masking_key (bytes): A 4-byte byte string used as the XOR masking key.

    Returns:
        bytes: The resulting transformed byte string.

    Raises:
        ValueError: If the provided masking_key is not exactly 4 bytes in length.
    """
    if len(masking_key) != 4:
        raise ValueError("WebSocket masking key must be exactly 4 bytes")

    # Convert memoryview to bytes for Rust compatibility (PyO3 &[u8] doesn't auto-convert)
    if isinstance(payload, memoryview):
        payload = bytes(payload)

    if HAS_RUST_EXTENSION and _unmask_websocket_payload is not None:
        return _unmask_websocket_payload(payload, masking_key)

    # Manual XOR application for pure-Python fallback
    m0, m1, m2, m3 = masking_key
    output = bytearray(payload)
    length = len(output)
    index = 0

    # Process bytes in 4-byte chunks for a slight performance boost in Python
    while index + 4 <= length:
        output[index] ^= m0
        output[index + 1] ^= m1
        output[index + 2] ^= m2
        output[index + 3] ^= m3
        index += 4

    # Handle remaining bytes (less than 4) if the payload length is not a multiple of 4
    if index < length:
        output[index] ^= m0
        index += 1
    if index < length:
        output[index] ^= m1
        index += 1
    if index < length:
        output[index] ^= m2

    return bytes(output)