Function read_body

src/websockets/http11.py:380–438

(
    status_code: int,
    headers: Headers,
    read_line: Callable[[int], Generator[None, None, bytes | bytearray]],
    read_exact: Callable[[int], Generator[None, None, bytes | bytearray]],
    read_to_eof: Callable[[int], Generator[None, None, bytes | bytearray]],
)

def read_body(
    status_code: int,
    headers: Headers,
    read_line: Callable[[int], Generator[None, None, bytes | bytearray]],
    read_exact: Callable[[int], Generator[None, None, bytes | bytearray]],
    read_to_eof: Callable[[int], Generator[None, None, bytes | bytearray]],
) -> Generator[None, None, bytes | bytearray]:
    # https://datatracker.ietf.org/doc/html/rfc7230#section-3.3.3

    # Since websockets only does GET requests (no HEAD, no CONNECT), all
    # responses except 1xx, 204, and 304 include a message body.
    if 100 <= status_code < 200 or status_code == 204 or status_code == 304:
        return b""

    # MultipleValuesError is sufficiently unlikely that we don't attempt to
    # handle it when accessing headers. Instead we document that its parent
    # class, LookupError, may be raised.
    # Conversions from str to int are protected by sys.set_int_max_str_digits..

    elif (coding := headers.get("Transfer-Encoding")) is not None:
        if coding != "chunked":
            raise NotImplementedError(f"transfer coding {coding} isn't supported")

        body = b""
        while True:
            chunk_size_line = yield from parse_line(read_line)
            raw_chunk_size = chunk_size_line.split(b";", 1)[0]
            # Set a lower limit than default_max_str_digits; 1 EB is plenty.
            if len(raw_chunk_size) > 15:
                str_chunk_size = raw_chunk_size.decode(errors="backslashreplace")
                raise SecurityError(f"chunk too large: 0x{str_chunk_size} bytes")
            chunk_size = int(raw_chunk_size, 16)
            if chunk_size == 0:
                break
            if len(body) + chunk_size > MAX_BODY_SIZE:
                raise SecurityError(
                    f"chunk too large: {chunk_size} bytes after {len(body)} bytes"
                )
            body += yield from read_exact(chunk_size)
            if (yield from read_exact(2)) != b"\r\n":
                raise ValueError("chunk without CRLF")
        # Read the trailer.
        yield from parse_headers(read_line)
        return body

    elif (raw_content_length := headers.get("Content-Length")) is not None:
        # Set a lower limit than default_max_str_digits; 1 EiB is plenty.
        if len(raw_content_length) > 18:
            raise SecurityError(f"body too large: {raw_content_length} bytes")
        content_length = int(raw_content_length)
        if content_length > MAX_BODY_SIZE:
            raise SecurityError(f"body too large: {content_length} bytes")
        return (yield from read_exact(content_length))

    else:
        try:
            return (yield from read_to_eof(MAX_BODY_SIZE))
        except RuntimeError: