Method decode_header

gunicorn/dirty/protocol.py:131–177

Decode the 16-byte message header. Args: data: 16 bytes of header data Returns: tuple: (msg_type, request_id, payload_length) Raises: DirtyProtocolError: If header is invalid

(data: bytes)

    @staticmethod
    def decode_header(data: bytes) -> tuple:
        """
        Decode the 16-byte message header.

        Args:
            data: 16 bytes of header data

        Returns:
            tuple: (msg_type, request_id, payload_length)

        Raises:
            DirtyProtocolError: If header is invalid
        """
        if len(data) < HEADER_SIZE:
            raise DirtyProtocolError(
                f"Header too short: {len(data)} bytes, expected {HEADER_SIZE}",
                raw_data=data
            )

        magic, version, msg_type, length, request_id = struct.unpack(
            HEADER_FORMAT, data[:HEADER_SIZE]
        )

        if magic != MAGIC:
            raise DirtyProtocolError(
                f"Invalid magic: {magic!r}, expected {MAGIC!r}",
                raw_data=data[:20]
            )

        if version != VERSION:
            raise DirtyProtocolError(
                f"Unsupported protocol version: {version}, expected {VERSION}",
                raw_data=data[:20]
            )

        if msg_type not in MSG_TYPE_TO_STR:
            raise DirtyProtocolError(
                f"Unknown message type: 0x{msg_type:02x}",
                raw_data=data[:20]
            )

        if length > MAX_MESSAGE_SIZE:
            raise DirtyProtocolError(
                f"Message too large: {length} bytes (max: {MAX_MESSAGE_SIZE})"
            )

        return msg_type, request_id, length

    @staticmethod
    def encode_request(request_id: int, app_path: str, action: str,