path: root/testing/mochitest/pywebsocket3/mod_pywebsocket/stream.py

# Copyright 2011, Google Inc.
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are
# met:
#
#     * Redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer.
#     * Redistributions in binary form must reproduce the above
# copyright notice, this list of conditions and the following disclaimer
# in the documentation and/or other materials provided with the
# distribution.
#     * Neither the name of Google Inc. nor the names of its
# contributors may be used to endorse or promote products derived from
# this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
"""This file provides classes and helper functions for parsing/building frames
of the WebSocket protocol (RFC 6455).

Specification:
http://tools.ietf.org/html/rfc6455
"""

from __future__ import absolute_import, division

from collections import deque
import logging
import os
import struct
import time
import socket
import six

from mod_pywebsocket import common
from mod_pywebsocket import util
from mod_pywebsocket._stream_exceptions import BadOperationException
from mod_pywebsocket._stream_exceptions import ConnectionTerminatedException
from mod_pywebsocket._stream_exceptions import InvalidFrameException
from mod_pywebsocket._stream_exceptions import InvalidUTF8Exception
from mod_pywebsocket._stream_exceptions import UnsupportedFrameException

_NOOP_MASKER = util.NoopMasker()


class Frame(object):
    def __init__(self,
                 fin=1,
                 rsv1=0,
                 rsv2=0,
                 rsv3=0,
                 opcode=None,
                 payload=b''):
        self.fin = fin
        self.rsv1 = rsv1
        self.rsv2 = rsv2
        self.rsv3 = rsv3
        self.opcode = opcode
        self.payload = payload


# Helper functions made public to be used for writing unittests for WebSocket
# clients.


def create_length_header(length, mask):
    """Creates a length header.

    Args:
        length: Frame length. Must be less than 2^63.
        mask: Mask bit. Must be boolean.

    Raises:
        ValueError: when bad data is given.
    """

    if mask:
        mask_bit = 1 << 7
    else:
        mask_bit = 0

    if length < 0:
        raise ValueError('length must be non negative integer')
    elif length <= 125:
        return util.pack_byte(mask_bit | length)
    elif length < (1 << 16):
        return util.pack_byte(mask_bit | 126) + struct.pack('!H', length)
    elif length < (1 << 63):
        return util.pack_byte(mask_bit | 127) + struct.pack('!Q', length)
    else:
        raise ValueError('Payload is too big for one frame')


def create_header(opcode, payload_length, fin, rsv1, rsv2, rsv3, mask):
    """Creates a frame header.

    Raises:
        Exception: when bad data is given.
    """

    if opcode < 0 or 0xf < opcode:
        raise ValueError('Opcode out of range')

    if payload_length < 0 or (1 << 63) <= payload_length:
        raise ValueError('payload_length out of range')

    if (fin | rsv1 | rsv2 | rsv3) & ~1:
        raise ValueError('FIN bit and Reserved bit parameter must be 0 or 1')

    header = b''

    first_byte = ((fin << 7)
                  | (rsv1 << 6) | (rsv2 << 5) | (rsv3 << 4)
                  | opcode)
    header += util.pack_byte(first_byte)
    header += create_length_header(payload_length, mask)

    return header


def _build_frame(header, body, mask):
    if not mask:
        return header + body

    masking_nonce = os.urandom(4)
    masker = util.RepeatedXorMasker(masking_nonce)

    return header + masking_nonce + masker.mask(body)


def _filter_and_format_frame_object(frame, mask, frame_filters):
    for frame_filter in frame_filters:
        frame_filter.filter(frame)

    header = create_header(frame.opcode, len(frame.payload), frame.fin,
                           frame.rsv1, frame.rsv2, frame.rsv3, mask)
    return _build_frame(header, frame.payload, mask)


def create_binary_frame(message,
                        opcode=common.OPCODE_BINARY,
                        fin=1,
                        mask=False,
                        frame_filters=[]):
    """Creates a simple binary frame with no extension, reserved bit."""

    frame = Frame(fin=fin, opcode=opcode, payload=message)
    return _filter_and_format_frame_object(frame, mask, frame_filters)


def create_text_frame(message,
                      opcode=common.OPCODE_TEXT,
                      fin=1,
                      mask=False,
                      frame_filters=[]):
    """Creates a simple text frame with no extension, reserved bit."""

    encoded_message = message.encode('utf-8')
    return create_binary_frame(encoded_message, opcode, fin, mask,
                               frame_filters)


def parse_frame(receive_bytes,
                logger=None,
                ws_version=common.VERSION_HYBI_LATEST,
                unmask_receive=True):
    """Parses a frame. Returns a tuple containing each header field and
    payload.

    Args:
        receive_bytes: a function that reads frame data from a stream or
            something similar. The function takes length of the bytes to be
            read. The function must raise ConnectionTerminatedException if
            there is not enough data to be read.
        logger: a logging object.
        ws_version: the version of WebSocket protocol.
        unmask_receive: unmask received frames. When received unmasked
            frame, raises InvalidFrameException.

    Raises:
        ConnectionTerminatedException: when receive_bytes raises it.
        InvalidFrameException: when the frame contains invalid data.
    """

    if not logger:
        logger = logging.getLogger()

    logger.log(common.LOGLEVEL_FINE, 'Receive the first 2 octets of a frame')

    first_byte = ord(receive_bytes(1))
    fin = (first_byte >> 7) & 1
    rsv1 = (first_byte >> 6) & 1
    rsv2 = (first_byte >> 5) & 1
    rsv3 = (first_byte >> 4) & 1
    opcode = first_byte & 0xf

    second_byte = ord(receive_bytes(1))
    mask = (second_byte >> 7) & 1
    payload_length = second_byte & 0x7f

    logger.log(
        common.LOGLEVEL_FINE, 'FIN=%s, RSV1=%s, RSV2=%s, RSV3=%s, opcode=%s, '
        'Mask=%s, Payload_length=%s', fin, rsv1, rsv2, rsv3, opcode, mask,
        payload_length)

    if (mask == 1) != unmask_receive:
        raise InvalidFrameException(
            'Mask bit on the received frame did\'nt match masking '
            'configuration for received frames')

    # The HyBi and later specs disallow putting a value in 0x0-0xFFFF
    # into the 8-octet extended payload length field (or 0x0-0xFD in
    # 2-octet field).
    valid_length_encoding = True
    length_encoding_bytes = 1
    if payload_length == 127:
        logger.log(common.LOGLEVEL_FINE,
                   'Receive 8-octet extended payload length')

        extended_payload_length = receive_bytes(8)
        payload_length = struct.unpack('!Q', extended_payload_length)[0]
        if payload_length > 0x7FFFFFFFFFFFFFFF:
            raise InvalidFrameException('Extended payload length >= 2^63')
        if ws_version >= 13 and payload_length < 0x10000:
            valid_length_encoding = False
            length_encoding_bytes = 8

        logger.log(common.LOGLEVEL_FINE, 'Decoded_payload_length=%s',
                   payload_length)
    elif payload_length == 126:
        logger.log(common.LOGLEVEL_FINE,
                   'Receive 2-octet extended payload length')

        extended_payload_length = receive_bytes(2)
        payload_length = struct.unpack('!H', extended_payload_length)[0]
        if ws_version >= 13 and payload_length < 126:
            valid_length_encoding = False
            length_encoding_bytes = 2

        logger.log(common.LOGLEVEL_FINE, 'Decoded_payload_length=%s',
                   payload_length)

    if not valid_length_encoding:
        logger.warning(
            'Payload length is not encoded using the minimal number of '
            'bytes (%d is encoded using %d bytes)', payload_length,
            length_encoding_bytes)

    if mask == 1:
        logger.log(common.LOGLEVEL_FINE, 'Receive mask')

        masking_nonce = receive_bytes(4)
        masker = util.RepeatedXorMasker(masking_nonce)

        logger.log(common.LOGLEVEL_FINE, 'Mask=%r', masking_nonce)
    else:
        masker = _NOOP_MASKER

    logger.log(common.LOGLEVEL_FINE, 'Receive payload data')
    if logger.isEnabledFor(common.LOGLEVEL_FINE):
        receive_start = time.time()

    raw_payload_bytes = receive_bytes(payload_length)

    if logger.isEnabledFor(common.LOGLEVEL_FINE):
        # pylint --py3k W1619
        logger.log(
            common.LOGLEVEL_FINE, 'Done receiving payload data at %s MB/s',
            payload_length / (time.time() - receive_start) / 1000 / 1000)
    logger.log(common.LOGLEVEL_FINE, 'Unmask payload data')

    if logger.isEnabledFor(common.LOGLEVEL_FINE):
        unmask_start = time.time()

    unmasked_bytes = masker.mask(raw_payload_bytes)

    if logger.isEnabledFor(common.LOGLEVEL_FINE):
        # pylint --py3k W1619
        logger.log(common.LOGLEVEL_FINE,
                   'Done unmasking payload data at %s MB/s',
                   payload_length / (time.time() - unmask_start) / 1000 / 1000)

    return opcode, unmasked_bytes, fin, rsv1, rsv2, rsv3


class FragmentedFrameBuilder(object):
    """A stateful class to send a message as fragments."""
    def __init__(self, mask, frame_filters=[], encode_utf8=True):
        """Constructs an instance."""

        self._mask = mask
        self._frame_filters = frame_filters
        # This is for skipping UTF-8 encoding when building text type frames
        # from compressed data.
        self._encode_utf8 = encode_utf8

        self._started = False

        # Hold opcode of the first frame in messages to verify types of other
        # frames in the message are all the same.
        self._opcode = common.OPCODE_TEXT

    def build(self, payload_data, end, binary):
        if binary:
            frame_type = common.OPCODE_BINARY
        else:
            frame_type = common.OPCODE_TEXT
        if self._started:
            if self._opcode != frame_type:
                raise ValueError('Message types are different in frames for '
                                 'the same message')
            opcode = common.OPCODE_CONTINUATION
        else:
            opcode = frame_type
            self._opcode = frame_type

        if end:
            self._started = False
            fin = 1
        else:
            self._started = True
            fin = 0

        if binary or not self._encode_utf8:
            return create_binary_frame(payload_data, opcode, fin, self._mask,
                                       self._frame_filters)
        else:
            return create_text_frame(payload_data, opcode, fin, self._mask,
                                     self._frame_filters)


def _create_control_frame(opcode, body, mask, frame_filters):
    frame = Frame(opcode=opcode, payload=body)

    for frame_filter in frame_filters:
        frame_filter.filter(frame)

    if len(frame.payload) > 125:
        raise BadOperationException(
            'Payload data size of control frames must be 125 bytes or less')

    header = create_header(frame.opcode, len(frame.payload), frame.fin,
                           frame.rsv1, frame.rsv2, frame.rsv3, mask)
    return _build_frame(header, frame.payload, mask)


def create_ping_frame(body, mask=False, frame_filters=[]):
    return _create_control_frame(common.OPCODE_PING, body, mask, frame_filters)


def create_pong_frame(body, mask=False, frame_filters=[]):
    return _create_control_frame(common.OPCODE_PONG, body, mask, frame_filters)


def create_close_frame(body, mask=False, frame_filters=[]):
    return _create_control_frame(common.OPCODE_CLOSE, body, mask,
                                 frame_filters)


def create_closing_handshake_body(code, reason):
    body = b''
    if code is not None:
        if (code > common.STATUS_USER_PRIVATE_MAX
                or code < common.STATUS_NORMAL_CLOSURE):
            raise BadOperationException('Status code is out of range')
        if (code == common.STATUS_NO_STATUS_RECEIVED
                or code == common.STATUS_ABNORMAL_CLOSURE
                or code == common.STATUS_TLS_HANDSHAKE):
            raise BadOperationException('Status code is reserved pseudo '
                                        'code')
        encoded_reason = reason.encode('utf-8')
        body = struct.pack('!H', code) + encoded_reason
    return body


class StreamOptions(object):
    """Holds option values to configure Stream objects."""
    def __init__(self):
        """Constructs StreamOptions."""

        # Filters applied to frames.
        self.outgoing_frame_filters = []
        self.incoming_frame_filters = []

        # Filters applied to messages. Control frames are not affected by them.
        self.outgoing_message_filters = []
        self.incoming_message_filters = []

        self.encode_text_message_to_utf8 = True
        self.mask_send = False
        self.unmask_receive = True


class Stream(object):
    """A class for parsing/building frames of the WebSocket protocol
    (RFC 6455).
    """
    def __init__(self, request, options):
        """Constructs an instance.

        Args:
            request: mod_python request.
        """

        self._logger = util.get_class_logger(self)

        self._options = options
        self._request = request

        self._request.client_terminated = False
        self._request.server_terminated = False

        # Holds body of received fragments.
        self._received_fragments = []
        # Holds the opcode of the first fragment.
        self._original_opcode = None

        self._writer = FragmentedFrameBuilder(
            self._options.mask_send, self._options.outgoing_frame_filters,
            self._options.encode_text_message_to_utf8)

        self._ping_queue = deque()

    def _read(self, length):
        """Reads length bytes from connection. In case we catch any exception,
        prepends remote address to the exception message and raise again.

        Raises:
            ConnectionTerminatedException: when read returns empty string.
        """

        try:
            read_bytes = self._request.connection.read(length)
            if not read_bytes:
                raise ConnectionTerminatedException(
                    'Receiving %d byte failed. Peer (%r) closed connection' %
                    (length, (self._request.connection.remote_addr, )))
            return read_bytes
        except IOError as e:
            # Also catch an IOError because mod_python throws it.
            raise ConnectionTerminatedException(
                'Receiving %d byte failed. IOError (%s) occurred' %
                (length, e))

    def _write(self, bytes_to_write):
        """Writes given bytes to connection. In case we catch any exception,
        prepends remote address to the exception message and raise again.
        """

        try:
            self._request.connection.write(bytes_to_write)
        except Exception as e:
            util.prepend_message_to_exception(
                'Failed to send message to %r: ' %
                (self._request.connection.remote_addr, ), e)
            raise

    def receive_bytes(self, length):
        """Receives multiple bytes. Retries read when we couldn't receive the
        specified amount. This method returns byte strings.

        Raises:
            ConnectionTerminatedException: when read returns empty string.
        """

        read_bytes = []
        while length > 0:
            new_read_bytes = self._read(length)
            read_bytes.append(new_read_bytes)
            length -= len(new_read_bytes)
        return b''.join(read_bytes)

    def _read_until(self, delim_char):
        """Reads bytes until we encounter delim_char. The result will not
        contain delim_char.

        Raises:
            ConnectionTerminatedException: when read returns empty string.
        """

        read_bytes = []
        while True:
            ch = self._read(1)
            if ch == delim_char:
                break
            read_bytes.append(ch)
        return b''.join(read_bytes)

    def _receive_frame(self):
        """Receives a frame and return data in the frame as a tuple containing
        each header field and payload separately.

        Raises:
            ConnectionTerminatedException: when read returns empty
                string.
            InvalidFrameException: when the frame contains invalid data.
        """
        def _receive_bytes(length):
            return self.receive_bytes(length)

        return parse_frame(receive_bytes=_receive_bytes,
                           logger=self._logger,
                           ws_version=self._request.ws_version,
                           unmask_receive=self._options.unmask_receive)

    def _receive_frame_as_frame_object(self):
        opcode, unmasked_bytes, fin, rsv1, rsv2, rsv3 = self._receive_frame()

        return Frame(fin=fin,
                     rsv1=rsv1,
                     rsv2=rsv2,
                     rsv3=rsv3,
                     opcode=opcode,
                     payload=unmasked_bytes)

    def receive_filtered_frame(self):
        """Receives a frame and applies frame filters and message filters.
        The frame to be received must satisfy following conditions:
        - The frame is not fragmented.
        - The opcode of the frame is TEXT or BINARY.

        DO NOT USE this method except for testing purpose.
        """

        frame = self._receive_frame_as_frame_object()
        if not frame.fin:
            raise InvalidFrameException(
                'Segmented frames must not be received via '
                'receive_filtered_frame()')
        if (frame.opcode != common.OPCODE_TEXT
                and frame.opcode != common.OPCODE_BINARY):
            raise InvalidFrameException(
                'Control frames must not be received via '
                'receive_filtered_frame()')

        for frame_filter in self._options.incoming_frame_filters:
            frame_filter.filter(frame)
        for message_filter in self._options.incoming_message_filters:
            frame.payload = message_filter.filter(frame.payload)
        return frame

    def send_message(self, message, end=True, binary=False):
        """Send message.

        Args:
            message: text in unicode or binary in str to send.
            binary: send message as binary frame.

        Raises:
            BadOperationException: when called on a server-terminated
                connection or called with inconsistent message type or
                binary parameter.
        """

        if self._request.server_terminated:
            raise BadOperationException(
                'Requested send_message after sending out a closing handshake')

        if binary and isinstance(message, six.text_type):
            raise BadOperationException(
                'Message for binary frame must not be instance of Unicode')

        for message_filter in self._options.outgoing_message_filters:
            message = message_filter.filter(message, end, binary)

        try:
            # Set this to any positive integer to limit maximum size of data in
            # payload data of each frame.
            MAX_PAYLOAD_DATA_SIZE = -1

            if MAX_PAYLOAD_DATA_SIZE <= 0:
                self._write(self._writer.build(message, end, binary))
                return

            bytes_written = 0
            while True:
                end_for_this_frame = end
                bytes_to_write = len(message) - bytes_written
                if (MAX_PAYLOAD_DATA_SIZE > 0
                        and bytes_to_write > MAX_PAYLOAD_DATA_SIZE):
                    end_for_this_frame = False
                    bytes_to_write = MAX_PAYLOAD_DATA_SIZE

                frame = self._writer.build(
                    message[bytes_written:bytes_written + bytes_to_write],
                    end_for_this_frame, binary)
                self._write(frame)

                bytes_written += bytes_to_write

                # This if must be placed here (the end of while block) so that
                # at least one frame is sent.
                if len(message) <= bytes_written:
                    break
        except ValueError as e:
            raise BadOperationException(e)

    def _get_message_from_frame(self, frame):
        """Gets a message from frame. If the message is composed of fragmented
        frames and the frame is not the last fragmented frame, this method
        returns None. The whole message will be returned when the last
        fragmented frame is passed to this method.

        Raises:
            InvalidFrameException: when the frame doesn't match defragmentation
                context, or the frame contains invalid data.
        """

        if frame.opcode == common.OPCODE_CONTINUATION:
            if not self._received_fragments:
                if frame.fin:
                    raise InvalidFrameException(
                        'Received a termination frame but fragmentation '
                        'not started')
                else:
                    raise InvalidFrameException(
                        'Received an intermediate frame but '
                        'fragmentation not started')

            if frame.fin:
                # End of fragmentation frame
                self._received_fragments.append(frame.payload)
                message = b''.join(self._received_fragments)
                self._received_fragments = []
                return message
            else:
                # Intermediate frame
                self._received_fragments.append(frame.payload)
                return None
        else:
            if self._received_fragments:
                if frame.fin:
                    raise InvalidFrameException(
                        'Received an unfragmented frame without '
                        'terminating existing fragmentation')
                else:
                    raise InvalidFrameException(
                        'New fragmentation started without terminating '
                        'existing fragmentation')

            if frame.fin:
                # Unfragmented frame

                self._original_opcode = frame.opcode
                return frame.payload
            else:
                # Start of fragmentation frame

                if common.is_control_opcode(frame.opcode):
                    raise InvalidFrameException(
                        'Control frames must not be fragmented')

                self._original_opcode = frame.opcode
                self._received_fragments.append(frame.payload)
                return None

    def _process_close_message(self, message):
        """Processes close message.

        Args:
            message: close message.

        Raises:
            InvalidFrameException: when the message is invalid.
        """

        self._request.client_terminated = True

        # Status code is optional. We can have status reason only if we
        # have status code. Status reason can be empty string. So,
        # allowed cases are
        # - no application data: no code no reason
        # - 2 octet of application data: has code but no reason
        # - 3 or more octet of application data: both code and reason
        if len(message) == 0:
            self._logger.debug('Received close frame (empty body)')
            self._request.ws_close_code = common.STATUS_NO_STATUS_RECEIVED
        elif len(message) == 1:
            raise InvalidFrameException(
                'If a close frame has status code, the length of '
                'status code must be 2 octet')
        elif len(message) >= 2:
            self._request.ws_close_code = struct.unpack('!H', message[0:2])[0]
            self._request.ws_close_reason = message[2:].decode(
                'utf-8', 'replace')
            self._logger.debug('Received close frame (code=%d, reason=%r)',
                               self._request.ws_close_code,
                               self._request.ws_close_reason)

        # As we've received a close frame, no more data is coming over the
        # socket. We can now safely close the socket without worrying about
        # RST sending.

        if self._request.server_terminated:
            self._logger.debug(
                'Received ack for server-initiated closing handshake')
            return

        self._logger.debug('Received client-initiated closing handshake')

        code = common.STATUS_NORMAL_CLOSURE
        reason = ''
        if hasattr(self._request, '_dispatcher'):
            dispatcher = self._request._dispatcher
            code, reason = dispatcher.passive_closing_handshake(self._request)
            if code is None and reason is not None and len(reason) > 0:
                self._logger.warning(
                    'Handler specified reason despite code being None')
                reason = ''
            if reason is None:
                reason = ''
        self._send_closing_handshake(code, reason)
        self._logger.debug(
            'Acknowledged closing handshake initiated by the peer '
            '(code=%r, reason=%r)', code, reason)

    def _process_ping_message(self, message):
        """Processes ping message.

        Args:
            message: ping message.
        """

        try:
            handler = self._request.on_ping_handler
            if handler:
                handler(self._request, message)
                return
        except AttributeError:
            pass
        self._send_pong(message)

    def _process_pong_message(self, message):
        """Processes pong message.

        Args:
            message: pong message.
        """

        # TODO(tyoshino): Add ping timeout handling.

        inflight_pings = deque()

        while True:
            try:
                expected_body = self._ping_queue.popleft()
                if expected_body == message:
                    # inflight_pings contains pings ignored by the
                    # other peer. Just forget them.
                    self._logger.debug(
                        'Ping %r is acked (%d pings were ignored)',
                        expected_body, len(inflight_pings))
                    break
                else:
                    inflight_pings.append(expected_body)
            except IndexError:
                # The received pong was unsolicited pong. Keep the
                # ping queue as is.
                self._ping_queue = inflight_pings
                self._logger.debug('Received a unsolicited pong')
                break

        try:
            handler = self._request.on_pong_handler
            if handler:
                handler(self._request, message)
        except AttributeError:
            pass

    def receive_message(self):
        """Receive a WebSocket frame and return its payload as a text in
        unicode or a binary in str.

        Returns:
            payload data of the frame
            - as unicode instance if received text frame
            - as str instance if received binary frame
            or None iff received closing handshake.
        Raises:
            BadOperationException: when called on a client-terminated
                connection.
            ConnectionTerminatedException: when read returns empty
                string.
            InvalidFrameException: when the frame contains invalid
                data.
            UnsupportedFrameException: when the received frame has
                flags, opcode we cannot handle. You can ignore this
                exception and continue receiving the next frame.
        """

        if self._request.client_terminated:
            raise BadOperationException(
                'Requested receive_message after receiving a closing '
                'handshake')

        while True:
            # mp_conn.read will block if no bytes are available.

            frame = self._receive_frame_as_frame_object()

            # Check the constraint on the payload size for control frames
            # before extension processes the frame.
            # See also http://tools.ietf.org/html/rfc6455#section-5.5
            if (common.is_control_opcode(frame.opcode)
                    and len(frame.payload) > 125):
                raise InvalidFrameException(
                    'Payload data size of control frames must be 125 bytes or '
                    'less')

            for frame_filter in self._options.incoming_frame_filters:
                frame_filter.filter(frame)

            if frame.rsv1 or frame.rsv2 or frame.rsv3:
                raise UnsupportedFrameException(
                    'Unsupported flag is set (rsv = %d%d%d)' %
                    (frame.rsv1, frame.rsv2, frame.rsv3))

            message = self._get_message_from_frame(frame)
            if message is None:
                continue

            for message_filter in self._options.incoming_message_filters:
                message = message_filter.filter(message)

            if self._original_opcode == common.OPCODE_TEXT:
                # The WebSocket protocol section 4.4 specifies that invalid
                # characters must be replaced with U+fffd REPLACEMENT
                # CHARACTER.
                try:
                    return message.decode('utf-8')
                except UnicodeDecodeError as e:
                    raise InvalidUTF8Exception(e)
            elif self._original_opcode == common.OPCODE_BINARY:
                return message
            elif self._original_opcode == common.OPCODE_CLOSE:
                self._process_close_message(message)
                return None
            elif self._original_opcode == common.OPCODE_PING:
                self._process_ping_message(message)
            elif self._original_opcode == common.OPCODE_PONG:
                self._process_pong_message(message)
            else:
                raise UnsupportedFrameException('Opcode %d is not supported' %
                                                self._original_opcode)

    def _send_closing_handshake(self, code, reason):
        body = create_closing_handshake_body(code, reason)
        frame = create_close_frame(
            body,
            mask=self._options.mask_send,
            frame_filters=self._options.outgoing_frame_filters)

        self._request.server_terminated = True

        self._write(frame)

    def close_connection(self,
                         code=common.STATUS_NORMAL_CLOSURE,
                         reason='',
                         wait_response=True):
        """Closes a WebSocket connection. Note that this method blocks until
        it receives acknowledgement to the closing handshake.

        Args:
            code: Status code for close frame. If code is None, a close
                frame with empty body will be sent.
            reason: string representing close reason.
            wait_response: True when caller want to wait the response.
        Raises:
            BadOperationException: when reason is specified with code None
            or reason is not an instance of both str and unicode.
        """

        if self._request.server_terminated:
            self._logger.debug(
                'Requested close_connection but server is already terminated')
            return

        # When we receive a close frame, we call _process_close_message().
        # _process_close_message() immediately acknowledges to the
        # server-initiated closing handshake and sets server_terminated to
        # True. So, here we can assume that we haven't received any close
        # frame. We're initiating a closing handshake.

        if code is None:
            if reason is not None and len(reason) > 0:
                raise BadOperationException(
                    'close reason must not be specified if code is None')
            reason = ''
        else:
            if not isinstance(reason, bytes) and not isinstance(
                    reason, six.text_type):
                raise BadOperationException(
                    'close reason must be an instance of bytes or unicode')

        self._send_closing_handshake(code, reason)
        self._logger.debug('Initiated closing handshake (code=%r, reason=%r)',
                           code, reason)

        if (code == common.STATUS_GOING_AWAY
                or code == common.STATUS_PROTOCOL_ERROR) or not wait_response:
            # It doesn't make sense to wait for a close frame if the reason is
            # protocol error or that the server is going away. For some of
            # other reasons, it might not make sense to wait for a close frame,
            # but it's not clear, yet.
            return

        # TODO(ukai): 2. wait until the /client terminated/ flag has been set,
        # or until a server-defined timeout expires.
        #
        # For now, we expect receiving closing handshake right after sending
        # out closing handshake.
        message = self.receive_message()
        if message is not None:
            raise ConnectionTerminatedException(
                'Didn\'t receive valid ack for closing handshake')
        # TODO: 3. close the WebSocket connection.
        # note: mod_python Connection (mp_conn) doesn't have close method.

    def send_ping(self, body, binary=False):
        if not binary and isinstance(body, six.text_type):
            body = body.encode('UTF-8')
        frame = create_ping_frame(body, self._options.mask_send,
                                  self._options.outgoing_frame_filters)
        self._write(frame)

        self._ping_queue.append(body)

    def _send_pong(self, body):
        frame = create_pong_frame(body, self._options.mask_send,
                                  self._options.outgoing_frame_filters)
        self._write(frame)

    def get_last_received_opcode(self):
        """Returns the opcode of the WebSocket message which the last received
        frame belongs to. The return value is valid iff immediately after
        receive_message call.
        """

        return self._original_opcode


# vi:sts=4 sw=4 et