michael@0: # Copyright 2012, Google Inc.
michael@0: # All rights reserved.
michael@0: #
michael@0: # Redistribution and use in source and binary forms, with or without
michael@0: # modification, are permitted provided that the following conditions are
michael@0: # met:
michael@0: #
michael@0: #     * Redistributions of source code must retain the above copyright
michael@0: # notice, this list of conditions and the following disclaimer.
michael@0: #     * Redistributions in binary form must reproduce the above
michael@0: # copyright notice, this list of conditions and the following disclaimer
michael@0: # in the documentation and/or other materials provided with the
michael@0: # distribution.
michael@0: #     * Neither the name of Google Inc. nor the names of its
michael@0: # contributors may be used to endorse or promote products derived from
michael@0: # this software without specific prior written permission.
michael@0: #
michael@0: # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
michael@0: # "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
michael@0: # LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
michael@0: # A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
michael@0: # OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
michael@0: # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
michael@0: # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
michael@0: # DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
michael@0: # THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
michael@0: # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
michael@0: # OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
michael@0: 
michael@0: 
michael@0: """This file provides classes and helper functions for parsing/building frames
michael@0: of the WebSocket protocol (RFC 6455).
michael@0: 
michael@0: Specification:
michael@0: http://tools.ietf.org/html/rfc6455
michael@0: """
michael@0: 
michael@0: 
michael@0: from collections import deque
michael@0: import os
michael@0: import struct
michael@0: 
michael@0: from mod_pywebsocket import common
michael@0: from mod_pywebsocket import util
michael@0: from mod_pywebsocket._stream_base import BadOperationException
michael@0: from mod_pywebsocket._stream_base import ConnectionTerminatedException
michael@0: from mod_pywebsocket._stream_base import InvalidFrameException
michael@0: from mod_pywebsocket._stream_base import InvalidUTF8Exception
michael@0: from mod_pywebsocket._stream_base import StreamBase
michael@0: from mod_pywebsocket._stream_base import UnsupportedFrameException
michael@0: 
michael@0: 
michael@0: _NOOP_MASKER = util.NoopMasker()
michael@0: 
michael@0: 
michael@0: class Frame(object):
michael@0: 
michael@0:     def __init__(self, fin=1, rsv1=0, rsv2=0, rsv3=0,
michael@0:                  opcode=None, payload=''):
michael@0:         self.fin = fin
michael@0:         self.rsv1 = rsv1
michael@0:         self.rsv2 = rsv2
michael@0:         self.rsv3 = rsv3
michael@0:         self.opcode = opcode
michael@0:         self.payload = payload
michael@0: 
michael@0: 
michael@0: # Helper functions made public to be used for writing unittests for WebSocket
michael@0: # clients.
michael@0: 
michael@0: 
michael@0: def create_length_header(length, mask):
michael@0:     """Creates a length header.
michael@0: 
michael@0:     Args:
michael@0:         length: Frame length. Must be less than 2^63.
michael@0:         mask: Mask bit. Must be boolean.
michael@0: 
michael@0:     Raises:
michael@0:         ValueError: when bad data is given.
michael@0:     """
michael@0: 
michael@0:     if mask:
michael@0:         mask_bit = 1 << 7
michael@0:     else:
michael@0:         mask_bit = 0
michael@0: 
michael@0:     if length < 0:
michael@0:         raise ValueError('length must be non negative integer')
michael@0:     elif length <= 125:
michael@0:         return chr(mask_bit | length)
michael@0:     elif length < (1 << 16):
michael@0:         return chr(mask_bit | 126) + struct.pack('!H', length)
michael@0:     elif length < (1 << 63):
michael@0:         return chr(mask_bit | 127) + struct.pack('!Q', length)
michael@0:     else:
michael@0:         raise ValueError('Payload is too big for one frame')
michael@0: 
michael@0: 
michael@0: def create_header(opcode, payload_length, fin, rsv1, rsv2, rsv3, mask):
michael@0:     """Creates a frame header.
michael@0: 
michael@0:     Raises:
michael@0:         Exception: when bad data is given.
michael@0:     """
michael@0: 
michael@0:     if opcode < 0 or 0xf < opcode:
michael@0:         raise ValueError('Opcode out of range')
michael@0: 
michael@0:     if payload_length < 0 or (1 << 63) <= payload_length:
michael@0:         raise ValueError('payload_length out of range')
michael@0: 
michael@0:     if (fin | rsv1 | rsv2 | rsv3) & ~1:
michael@0:         raise ValueError('FIN bit and Reserved bit parameter must be 0 or 1')
michael@0: 
michael@0:     header = ''
michael@0: 
michael@0:     first_byte = ((fin << 7)
michael@0:                   | (rsv1 << 6) | (rsv2 << 5) | (rsv3 << 4)
michael@0:                   | opcode)
michael@0:     header += chr(first_byte)
michael@0:     header += create_length_header(payload_length, mask)
michael@0: 
michael@0:     return header
michael@0: 
michael@0: 
michael@0: def _build_frame(header, body, mask):
michael@0:     if not mask:
michael@0:         return header + body
michael@0: 
michael@0:     masking_nonce = os.urandom(4)
michael@0:     masker = util.RepeatedXorMasker(masking_nonce)
michael@0: 
michael@0:     return header + masking_nonce + masker.mask(body)
michael@0: 
michael@0: 
michael@0: def _filter_and_format_frame_object(frame, mask, frame_filters):
michael@0:     for frame_filter in frame_filters:
michael@0:         frame_filter.filter(frame)
michael@0: 
michael@0:     header = create_header(
michael@0:         frame.opcode, len(frame.payload), frame.fin,
michael@0:         frame.rsv1, frame.rsv2, frame.rsv3, mask)
michael@0:     return _build_frame(header, frame.payload, mask)
michael@0: 
michael@0: 
michael@0: def create_binary_frame(
michael@0:     message, opcode=common.OPCODE_BINARY, fin=1, mask=False, frame_filters=[]):
michael@0:     """Creates a simple binary frame with no extension, reserved bit."""
michael@0: 
michael@0:     frame = Frame(fin=fin, opcode=opcode, payload=message)
michael@0:     return _filter_and_format_frame_object(frame, mask, frame_filters)
michael@0: 
michael@0: 
michael@0: def create_text_frame(
michael@0:     message, opcode=common.OPCODE_TEXT, fin=1, mask=False, frame_filters=[]):
michael@0:     """Creates a simple text frame with no extension, reserved bit."""
michael@0: 
michael@0:     encoded_message = message.encode('utf-8')
michael@0:     return create_binary_frame(encoded_message, opcode, fin, mask,
michael@0:                                frame_filters)
michael@0: 
michael@0: 
michael@0: class FragmentedFrameBuilder(object):
michael@0:     """A stateful class to send a message as fragments."""
michael@0: 
michael@0:     def __init__(self, mask, frame_filters=[]):
michael@0:         """Constructs an instance."""
michael@0: 
michael@0:         self._mask = mask
michael@0:         self._frame_filters = frame_filters
michael@0: 
michael@0:         self._started = False
michael@0: 
michael@0:         # Hold opcode of the first frame in messages to verify types of other
michael@0:         # frames in the message are all the same.
michael@0:         self._opcode = common.OPCODE_TEXT
michael@0: 
michael@0:     def build(self, message, end, binary):
michael@0:         if binary:
michael@0:             frame_type = common.OPCODE_BINARY
michael@0:         else:
michael@0:             frame_type = common.OPCODE_TEXT
michael@0:         if self._started:
michael@0:             if self._opcode != frame_type:
michael@0:                 raise ValueError('Message types are different in frames for '
michael@0:                                  'the same message')
michael@0:             opcode = common.OPCODE_CONTINUATION
michael@0:         else:
michael@0:             opcode = frame_type
michael@0:             self._opcode = frame_type
michael@0: 
michael@0:         if end:
michael@0:             self._started = False
michael@0:             fin = 1
michael@0:         else:
michael@0:             self._started = True
michael@0:             fin = 0
michael@0: 
michael@0:         if binary:
michael@0:             return create_binary_frame(
michael@0:                 message, opcode, fin, self._mask, self._frame_filters)
michael@0:         else:
michael@0:             return create_text_frame(
michael@0:                 message, opcode, fin, self._mask, self._frame_filters)
michael@0: 
michael@0: 
michael@0: def _create_control_frame(opcode, body, mask, frame_filters):
michael@0:     frame = Frame(opcode=opcode, payload=body)
michael@0: 
michael@0:     return _filter_and_format_frame_object(frame, mask, frame_filters)
michael@0: 
michael@0: 
michael@0: def create_ping_frame(body, mask=False, frame_filters=[]):
michael@0:     return _create_control_frame(common.OPCODE_PING, body, mask, frame_filters)
michael@0: 
michael@0: 
michael@0: def create_pong_frame(body, mask=False, frame_filters=[]):
michael@0:     return _create_control_frame(common.OPCODE_PONG, body, mask, frame_filters)
michael@0: 
michael@0: 
michael@0: def create_close_frame(body, mask=False, frame_filters=[]):
michael@0:     return _create_control_frame(
michael@0:         common.OPCODE_CLOSE, body, mask, frame_filters)
michael@0: 
michael@0: 
michael@0: class StreamOptions(object):
michael@0:     """Holds option values to configure Stream objects."""
michael@0: 
michael@0:     def __init__(self):
michael@0:         """Constructs StreamOptions."""
michael@0: 
michael@0:         # Enables deflate-stream extension.
michael@0:         self.deflate_stream = False
michael@0: 
michael@0:         # Filters applied to frames.
michael@0:         self.outgoing_frame_filters = []
michael@0:         self.incoming_frame_filters = []
michael@0: 
michael@0:         self.mask_send = False
michael@0:         self.unmask_receive = True
michael@0: 
michael@0: 
michael@0: class Stream(StreamBase):
michael@0:     """A class for parsing/building frames of the WebSocket protocol
michael@0:     (RFC 6455).
michael@0:     """
michael@0: 
michael@0:     def __init__(self, request, options):
michael@0:         """Constructs an instance.
michael@0: 
michael@0:         Args:
michael@0:             request: mod_python request.
michael@0:         """
michael@0: 
michael@0:         StreamBase.__init__(self, request)
michael@0: 
michael@0:         self._logger = util.get_class_logger(self)
michael@0: 
michael@0:         self._options = options
michael@0: 
michael@0:         if self._options.deflate_stream:
michael@0:             self._logger.debug('Setup filter for deflate-stream')
michael@0:             self._request = util.DeflateRequest(self._request)
michael@0: 
michael@0:         self._request.client_terminated = False
michael@0:         self._request.server_terminated = False
michael@0: 
michael@0:         # Holds body of received fragments.
michael@0:         self._received_fragments = []
michael@0:         # Holds the opcode of the first fragment.
michael@0:         self._original_opcode = None
michael@0: 
michael@0:         self._writer = FragmentedFrameBuilder(
michael@0:             self._options.mask_send, self._options.outgoing_frame_filters)
michael@0: 
michael@0:         self._ping_queue = deque()
michael@0: 
michael@0:     def _receive_frame(self):
michael@0:         """Receives a frame and return data in the frame as a tuple containing
michael@0:         each header field and payload separately.
michael@0: 
michael@0:         Raises:
michael@0:             ConnectionTerminatedException: when read returns empty
michael@0:                 string.
michael@0:             InvalidFrameException: when the frame contains invalid data.
michael@0:         """
michael@0: 
michael@0:         received = self.receive_bytes(2)
michael@0: 
michael@0:         first_byte = ord(received[0])
michael@0:         fin = (first_byte >> 7) & 1
michael@0:         rsv1 = (first_byte >> 6) & 1
michael@0:         rsv2 = (first_byte >> 5) & 1
michael@0:         rsv3 = (first_byte >> 4) & 1
michael@0:         opcode = first_byte & 0xf
michael@0: 
michael@0:         second_byte = ord(received[1])
michael@0:         mask = (second_byte >> 7) & 1
michael@0:         payload_length = second_byte & 0x7f
michael@0: 
michael@0:         if (mask == 1) != self._options.unmask_receive:
michael@0:             raise InvalidFrameException(
michael@0:                 'Mask bit on the received frame did\'nt match masking '
michael@0:                 'configuration for received frames')
michael@0: 
michael@0:         # The Hybi-13 and later specs disallow putting a value in 0x0-0xFFFF
michael@0:         # into the 8-octet extended payload length field (or 0x0-0xFD in
michael@0:         # 2-octet field).
michael@0:         valid_length_encoding = True
michael@0:         length_encoding_bytes = 1
michael@0:         if payload_length == 127:
michael@0:             extended_payload_length = self.receive_bytes(8)
michael@0:             payload_length = struct.unpack(
michael@0:                 '!Q', extended_payload_length)[0]
michael@0:             if payload_length > 0x7FFFFFFFFFFFFFFF:
michael@0:                 raise InvalidFrameException(
michael@0:                     'Extended payload length >= 2^63')
michael@0:             if self._request.ws_version >= 13 and payload_length < 0x10000:
michael@0:                 valid_length_encoding = False
michael@0:                 length_encoding_bytes = 8
michael@0:         elif payload_length == 126:
michael@0:             extended_payload_length = self.receive_bytes(2)
michael@0:             payload_length = struct.unpack(
michael@0:                 '!H', extended_payload_length)[0]
michael@0:             if self._request.ws_version >= 13 and payload_length < 126:
michael@0:                 valid_length_encoding = False
michael@0:                 length_encoding_bytes = 2
michael@0: 
michael@0:         if not valid_length_encoding:
michael@0:             self._logger.warning(
michael@0:                 'Payload length is not encoded using the minimal number of '
michael@0:                 'bytes (%d is encoded using %d bytes)',
michael@0:                 payload_length,
michael@0:                 length_encoding_bytes)
michael@0: 
michael@0:         if mask == 1:
michael@0:             masking_nonce = self.receive_bytes(4)
michael@0:             masker = util.RepeatedXorMasker(masking_nonce)
michael@0:         else:
michael@0:             masker = _NOOP_MASKER
michael@0: 
michael@0:         bytes = masker.mask(self.receive_bytes(payload_length))
michael@0: 
michael@0:         return opcode, bytes, fin, rsv1, rsv2, rsv3
michael@0: 
michael@0:     def _receive_frame_as_frame_object(self):
michael@0:         opcode, bytes, fin, rsv1, rsv2, rsv3 = self._receive_frame()
michael@0: 
michael@0:         return Frame(fin=fin, rsv1=rsv1, rsv2=rsv2, rsv3=rsv3,
michael@0:                      opcode=opcode, payload=bytes)
michael@0: 
michael@0:     def send_message(self, message, end=True, binary=False):
michael@0:         """Send message.
michael@0: 
michael@0:         Args:
michael@0:             message: text in unicode or binary in str to send.
michael@0:             binary: send message as binary frame.
michael@0: 
michael@0:         Raises:
michael@0:             BadOperationException: when called on a server-terminated
michael@0:                 connection or called with inconsistent message type or
michael@0:                 binary parameter.
michael@0:         """
michael@0: 
michael@0:         if self._request.server_terminated:
michael@0:             raise BadOperationException(
michael@0:                 'Requested send_message after sending out a closing handshake')
michael@0: 
michael@0:         if binary and isinstance(message, unicode):
michael@0:             raise BadOperationException(
michael@0:                 'Message for binary frame must be instance of str')
michael@0: 
michael@0:         try:
michael@0:             self._write(self._writer.build(message, end, binary))
michael@0:         except ValueError, e:
michael@0:             raise BadOperationException(e)
michael@0: 
michael@0:     def receive_message(self):
michael@0:         """Receive a WebSocket frame and return its payload as a text in
michael@0:         unicode or a binary in str.
michael@0: 
michael@0:         Returns:
michael@0:             payload data of the frame
michael@0:             - as unicode instance if received text frame
michael@0:             - as str instance if received binary frame
michael@0:             or None iff received closing handshake.
michael@0:         Raises:
michael@0:             BadOperationException: when called on a client-terminated
michael@0:                 connection.
michael@0:             ConnectionTerminatedException: when read returns empty
michael@0:                 string.
michael@0:             InvalidFrameException: when the frame contains invalid
michael@0:                 data.
michael@0:             UnsupportedFrameException: when the received frame has
michael@0:                 flags, opcode we cannot handle. You can ignore this
michael@0:                 exception and continue receiving the next frame.
michael@0:         """
michael@0: 
michael@0:         if self._request.client_terminated:
michael@0:             raise BadOperationException(
michael@0:                 'Requested receive_message after receiving a closing '
michael@0:                 'handshake')
michael@0: 
michael@0:         while True:
michael@0:             # mp_conn.read will block if no bytes are available.
michael@0:             # Timeout is controlled by TimeOut directive of Apache.
michael@0: 
michael@0:             frame = self._receive_frame_as_frame_object()
michael@0: 
michael@0:             for frame_filter in self._options.incoming_frame_filters:
michael@0:                 frame_filter.filter(frame)
michael@0: 
michael@0:             if frame.rsv1 or frame.rsv2 or frame.rsv3:
michael@0:                 raise UnsupportedFrameException(
michael@0:                     'Unsupported flag is set (rsv = %d%d%d)' %
michael@0:                     (frame.rsv1, frame.rsv2, frame.rsv3))
michael@0: 
michael@0:             if frame.opcode == common.OPCODE_CONTINUATION:
michael@0:                 if not self._received_fragments:
michael@0:                     if frame.fin:
michael@0:                         raise InvalidFrameException(
michael@0:                             'Received a termination frame but fragmentation '
michael@0:                             'not started')
michael@0:                     else:
michael@0:                         raise InvalidFrameException(
michael@0:                             'Received an intermediate frame but '
michael@0:                             'fragmentation not started')
michael@0: 
michael@0:                 if frame.fin:
michael@0:                     # End of fragmentation frame
michael@0:                     self._received_fragments.append(frame.payload)
michael@0:                     message = ''.join(self._received_fragments)
michael@0:                     self._received_fragments = []
michael@0:                 else:
michael@0:                     # Intermediate frame
michael@0:                     self._received_fragments.append(frame.payload)
michael@0:                     continue
michael@0:             else:
michael@0:                 if self._received_fragments:
michael@0:                     if frame.fin:
michael@0:                         raise InvalidFrameException(
michael@0:                             'Received an unfragmented frame without '
michael@0:                             'terminating existing fragmentation')
michael@0:                     else:
michael@0:                         raise InvalidFrameException(
michael@0:                             'New fragmentation started without terminating '
michael@0:                             'existing fragmentation')
michael@0: 
michael@0:                 if frame.fin:
michael@0:                     # Unfragmented frame
michael@0: 
michael@0:                     if (common.is_control_opcode(frame.opcode) and
michael@0:                         len(frame.payload) > 125):
michael@0:                         raise InvalidFrameException(
michael@0:                             'Application data size of control frames must be '
michael@0:                             '125 bytes or less')
michael@0: 
michael@0:                     self._original_opcode = frame.opcode
michael@0:                     message = frame.payload
michael@0:                 else:
michael@0:                     # Start of fragmentation frame
michael@0: 
michael@0:                     if common.is_control_opcode(frame.opcode):
michael@0:                         raise InvalidFrameException(
michael@0:                             'Control frames must not be fragmented')
michael@0: 
michael@0:                     self._original_opcode = frame.opcode
michael@0:                     self._received_fragments.append(frame.payload)
michael@0:                     continue
michael@0: 
michael@0:             if self._original_opcode == common.OPCODE_TEXT:
michael@0:                 # The WebSocket protocol section 4.4 specifies that invalid
michael@0:                 # characters must be replaced with U+fffd REPLACEMENT
michael@0:                 # CHARACTER.
michael@0:                 try:
michael@0:                     return message.decode('utf-8')
michael@0:                 except UnicodeDecodeError, e:
michael@0:                     raise InvalidUTF8Exception(e)
michael@0:             elif self._original_opcode == common.OPCODE_BINARY:
michael@0:                 return message
michael@0:             elif self._original_opcode == common.OPCODE_CLOSE:
michael@0:                 self._request.client_terminated = True
michael@0: 
michael@0:                 # Status code is optional. We can have status reason only if we
michael@0:                 # have status code. Status reason can be empty string. So,
michael@0:                 # allowed cases are
michael@0:                 # - no application data: no code no reason
michael@0:                 # - 2 octet of application data: has code but no reason
michael@0:                 # - 3 or more octet of application data: both code and reason
michael@0:                 if len(message) == 0:
michael@0:                     self._logger.debug('Received close frame (empty body)')
michael@0:                     self._request.ws_close_code = (
michael@0:                         common.STATUS_NO_STATUS_RECEIVED)
michael@0:                 elif len(message) == 1:
michael@0:                     raise InvalidFrameException(
michael@0:                         'If a close frame has status code, the length of '
michael@0:                         'status code must be 2 octet')
michael@0:                 elif len(message) >= 2:
michael@0:                     self._request.ws_close_code = struct.unpack(
michael@0:                         '!H', message[0:2])[0]
michael@0:                     self._request.ws_close_reason = message[2:].decode(
michael@0:                         'utf-8', 'replace')
michael@0:                     self._logger.debug(
michael@0:                         'Received close frame (code=%d, reason=%r)',
michael@0:                         self._request.ws_close_code,
michael@0:                         self._request.ws_close_reason)
michael@0: 
michael@0:                 # Drain junk data after the close frame if necessary.
michael@0:                 self._drain_received_data()
michael@0: 
michael@0:                 if self._request.server_terminated:
michael@0:                     self._logger.debug(
michael@0:                         'Received ack for server-initiated closing handshake')
michael@0:                     return None
michael@0: 
michael@0:                 self._logger.debug(
michael@0:                     'Received client-initiated closing handshake')
michael@0: 
michael@0:                 code = common.STATUS_NORMAL_CLOSURE
michael@0:                 reason = ''
michael@0:                 if hasattr(self._request, '_dispatcher'):
michael@0:                     dispatcher = self._request._dispatcher
michael@0:                     code, reason = dispatcher.passive_closing_handshake(
michael@0:                         self._request)
michael@0:                     if code is None and reason is not None and len(reason) > 0:
michael@0:                         self._logger.warning(
michael@0:                             'Handler specified reason despite code being None')
michael@0:                         reason = ''
michael@0:                     if reason is None:
michael@0:                         reason = ''
michael@0:                 self._send_closing_handshake(code, reason)
michael@0:                 self._logger.debug(
michael@0:                     'Sent ack for client-initiated closing handshake '
michael@0:                     '(code=%r, reason=%r)', code, reason)
michael@0:                 return None
michael@0:             elif self._original_opcode == common.OPCODE_PING:
michael@0:                 try:
michael@0:                     handler = self._request.on_ping_handler
michael@0:                     if handler:
michael@0:                         handler(self._request, message)
michael@0:                         continue
michael@0:                 except AttributeError, e:
michael@0:                     pass
michael@0:                 self._send_pong(message)
michael@0:             elif self._original_opcode == common.OPCODE_PONG:
michael@0:                 # TODO(tyoshino): Add ping timeout handling.
michael@0: 
michael@0:                 inflight_pings = deque()
michael@0: 
michael@0:                 while True:
michael@0:                     try:
michael@0:                         expected_body = self._ping_queue.popleft()
michael@0:                         if expected_body == message:
michael@0:                             # inflight_pings contains pings ignored by the
michael@0:                             # other peer. Just forget them.
michael@0:                             self._logger.debug(
michael@0:                                 'Ping %r is acked (%d pings were ignored)',
michael@0:                                 expected_body, len(inflight_pings))
michael@0:                             break
michael@0:                         else:
michael@0:                             inflight_pings.append(expected_body)
michael@0:                     except IndexError, e:
michael@0:                         # The received pong was unsolicited pong. Keep the
michael@0:                         # ping queue as is.
michael@0:                         self._ping_queue = inflight_pings
michael@0:                         self._logger.debug('Received a unsolicited pong')
michael@0:                         break
michael@0: 
michael@0:                 try:
michael@0:                     handler = self._request.on_pong_handler
michael@0:                     if handler:
michael@0:                         handler(self._request, message)
michael@0:                         continue
michael@0:                 except AttributeError, e:
michael@0:                     pass
michael@0: 
michael@0:                 continue
michael@0:             else:
michael@0:                 raise UnsupportedFrameException(
michael@0:                     'Opcode %d is not supported' % self._original_opcode)
michael@0: 
michael@0:     def _send_closing_handshake(self, code, reason):
michael@0:         body = ''
michael@0:         if code is not None:
michael@0:             if code >= (1 << 16) or code < 0:
michael@0:                 raise BadOperationException('Status code is out of range')
michael@0:             encoded_reason = reason.encode('utf-8')
michael@0:             if len(encoded_reason) + 2 > 125:
michael@0:                 raise BadOperationException(
michael@0:                     'Application data size of close frames must be 125 bytes '
michael@0:                     'or less')
michael@0:             body = struct.pack('!H', code) + encoded_reason
michael@0: 
michael@0:         frame = create_close_frame(
michael@0:             body,
michael@0:             self._options.mask_send,
michael@0:             self._options.outgoing_frame_filters)
michael@0: 
michael@0:         self._request.server_terminated = True
michael@0: 
michael@0:         self._write(frame)
michael@0: 
michael@0:     def close_connection(self, code=common.STATUS_NORMAL_CLOSURE, reason=''):
michael@0:         """Closes a WebSocket connection.
michael@0: 
michael@0:         Args:
michael@0:             code: Status code for close frame. If code is None, a close
michael@0:                 frame with empty body will be sent.
michael@0:             reason: string representing close reason.
michael@0:         Raises:
michael@0:             BadOperationException: when reason is specified with code None
michael@0:             or reason is not an instance of both str and unicode.
michael@0:         """
michael@0: 
michael@0:         if self._request.server_terminated:
michael@0:             self._logger.debug(
michael@0:                 'Requested close_connection but server is already terminated')
michael@0:             return
michael@0: 
michael@0:         if code is None:
michael@0:             if reason is not None and len(reason) > 0:
michael@0:                 raise BadOperationException(
michael@0:                     'close reason must not be specified if code is None')
michael@0:             reason = ''
michael@0:         else:
michael@0:             if not isinstance(reason, str) and not isinstance(reason, unicode):
michael@0:                 raise BadOperationException(
michael@0:                     'close reason must be an instance of str or unicode')
michael@0: 
michael@0:         self._send_closing_handshake(code, reason)
michael@0:         self._logger.debug(
michael@0:             'Sent server-initiated closing handshake (code=%r, reason=%r)',
michael@0:             code, reason)
michael@0: 
michael@0:         if (code == common.STATUS_GOING_AWAY or
michael@0:             code == common.STATUS_PROTOCOL_ERROR):
michael@0:             # It doesn't make sense to wait for a close frame if the reason is
michael@0:             # protocol error or that the server is going away. For some of
michael@0:             # other reasons, it might not make sense to wait for a close frame,
michael@0:             # but it's not clear, yet.
michael@0:             return
michael@0: 
michael@0:         # TODO(ukai): 2. wait until the /client terminated/ flag has been set,
michael@0:         # or until a server-defined timeout expires.
michael@0:         #
michael@0:         # For now, we expect receiving closing handshake right after sending
michael@0:         # out closing handshake.
michael@0:         message = self.receive_message()
michael@0:         if message is not None:
michael@0:             raise ConnectionTerminatedException(
michael@0:                 'Didn\'t receive valid ack for closing handshake')
michael@0:         # TODO: 3. close the WebSocket connection.
michael@0:         # note: mod_python Connection (mp_conn) doesn't have close method.
michael@0: 
michael@0:     def send_ping(self, body=''):
michael@0:         if len(body) > 125:
michael@0:             raise ValueError(
michael@0:                 'Application data size of control frames must be 125 bytes or '
michael@0:                 'less')
michael@0:         frame = create_ping_frame(
michael@0:             body,
michael@0:             self._options.mask_send,
michael@0:             self._options.outgoing_frame_filters)
michael@0:         self._write(frame)
michael@0: 
michael@0:         self._ping_queue.append(body)
michael@0: 
michael@0:     def _send_pong(self, body):
michael@0:         if len(body) > 125:
michael@0:             raise ValueError(
michael@0:                 'Application data size of control frames must be 125 bytes or '
michael@0:                 'less')
michael@0:         frame = create_pong_frame(
michael@0:             body,
michael@0:             self._options.mask_send,
michael@0:             self._options.outgoing_frame_filters)
michael@0:         self._write(frame)
michael@0: 
michael@0:     def _drain_received_data(self):
michael@0:         """Drains unread data in the receive buffer to avoid sending out TCP
michael@0:         RST packet. This is because when deflate-stream is enabled, some
michael@0:         DEFLATE block for flushing data may follow a close frame. If any data
michael@0:         remains in the receive buffer of a socket when the socket is closed,
michael@0:         it sends out TCP RST packet to the other peer.
michael@0: 
michael@0:         Since mod_python's mp_conn object doesn't support non-blocking read,
michael@0:         we perform this only when pywebsocket is running in standalone mode.
michael@0:         """
michael@0: 
michael@0:         # If self._options.deflate_stream is true, self._request is
michael@0:         # DeflateRequest, so we can get wrapped request object by
michael@0:         # self._request._request.
michael@0:         #
michael@0:         # Only _StandaloneRequest has _drain_received_data method.
michael@0:         if (self._options.deflate_stream and
michael@0:             ('_drain_received_data' in dir(self._request._request))):
michael@0:             self._request._request._drain_received_data()
michael@0: 
michael@0: 
michael@0: # vi:sts=4 sw=4 et