diff options
| author |  RaNaN <Mast3rRaNaN@hotmail.de>
| 2013-06-09 18:10:22 +0200 |
|---|---|---|
| committer | RaNaN <Mast3rRaNaN@hotmail.de>
| 2013-06-09 18:10:23 +0200 |
| commit | 16af85004c84d0d6c626b4f8424ce9647669a0c1 (patch) | |
| tree | 025d479862d376dbc17e934f4ed20031c8cd97d1 /module/lib/mod_pywebsocket | |
| parent | adapted to jshint config (diff) | |
| download | pyload-16af85004c84d0d6c626b4f8424ce9647669a0c1.tar.xz | |
moved everything from module to pyload
Diffstat (limited to 'module/lib/mod_pywebsocket')
20 files changed, 0 insertions, 7968 deletions
diff --git a/module/lib/mod_pywebsocket/COPYING b/module/lib/mod_pywebsocket/COPYING deleted file mode 100644 index 989d02e4c..000000000 --- a/module/lib/mod_pywebsocket/COPYING +++ /dev/null @@ -1,28 +0,0 @@ -Copyright 2012, 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. diff --git a/module/lib/mod_pywebsocket/__init__.py b/module/lib/mod_pywebsocket/__init__.py deleted file mode 100644 index 454ae0c45..000000000 --- a/module/lib/mod_pywebsocket/__init__.py +++ /dev/null @@ -1,197 +0,0 @@ -# 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. - - -"""WebSocket extension for Apache HTTP Server. - -mod_pywebsocket is a WebSocket extension for Apache HTTP Server -intended for testing or experimental purposes. mod_python is required. - - -Installation -============ - -0. Prepare an Apache HTTP Server for which mod_python is enabled. - -1. Specify the following Apache HTTP Server directives to suit your - configuration. - - If mod_pywebsocket is not in the Python path, specify the following. - <websock_lib> is the directory where mod_pywebsocket is installed. - - PythonPath "sys.path+['<websock_lib>']" - - Always specify the following. <websock_handlers> is the directory where - user-written WebSocket handlers are placed. - - PythonOption mod_pywebsocket.handler_root <websock_handlers> - PythonHeaderParserHandler mod_pywebsocket.headerparserhandler - - To limit the search for WebSocket handlers to a directory <scan_dir> - under <websock_handlers>, configure as follows: - - PythonOption mod_pywebsocket.handler_scan <scan_dir> - - <scan_dir> is useful in saving scan time when <websock_handlers> - contains many non-WebSocket handler files. - - If you want to allow handlers whose canonical path is not under the root - directory (i.e. symbolic link is in root directory but its target is not), - configure as follows: - - PythonOption mod_pywebsocket.allow_handlers_outside_root_dir On - - Example snippet of httpd.conf: - (mod_pywebsocket is in /websock_lib, WebSocket handlers are in - /websock_handlers, port is 80 for ws, 443 for wss.) - - <IfModule python_module> - PythonPath "sys.path+['/websock_lib']" - PythonOption mod_pywebsocket.handler_root /websock_handlers - PythonHeaderParserHandler mod_pywebsocket.headerparserhandler - </IfModule> - -2. Tune Apache parameters for serving WebSocket. We'd like to note that at - least TimeOut directive from core features and RequestReadTimeout - directive from mod_reqtimeout should be modified not to kill connections - in only a few seconds of idle time. - -3. Verify installation. You can use example/console.html to poke the server. - - -Writing WebSocket handlers -========================== - -When a WebSocket request comes in, the resource name -specified in the handshake is considered as if it is a file path under -<websock_handlers> and the handler defined in -<websock_handlers>/<resource_name>_wsh.py is invoked. - -For example, if the resource name is /example/chat, the handler defined in -<websock_handlers>/example/chat_wsh.py is invoked. - -A WebSocket handler is composed of the following three functions: - - web_socket_do_extra_handshake(request) - web_socket_transfer_data(request) - web_socket_passive_closing_handshake(request) - -where: - request: mod_python request. - -web_socket_do_extra_handshake is called during the handshake after the -headers are successfully parsed and WebSocket properties (ws_location, -ws_origin, and ws_resource) are added to request. A handler -can reject the request by raising an exception. - -A request object has the following properties that you can use during the -extra handshake (web_socket_do_extra_handshake): -- ws_resource -- ws_origin -- ws_version -- ws_location (HyBi 00 only) -- ws_extensions (HyBi 06 and later) -- ws_deflate (HyBi 06 and later) -- ws_protocol -- ws_requested_protocols (HyBi 06 and later) - -The last two are a bit tricky. See the next subsection. - - -Subprotocol Negotiation ------------------------ - -For HyBi 06 and later, ws_protocol is always set to None when -web_socket_do_extra_handshake is called. If ws_requested_protocols is not -None, you must choose one subprotocol from this list and set it to -ws_protocol. - -For HyBi 00, when web_socket_do_extra_handshake is called, -ws_protocol is set to the value given by the client in -Sec-WebSocket-Protocol header or None if -such header was not found in the opening handshake request. Finish extra -handshake with ws_protocol untouched to accept the request subprotocol. -Then, Sec-WebSocket-Protocol header will be sent to -the client in response with the same value as requested. Raise an exception -in web_socket_do_extra_handshake to reject the requested subprotocol. - - -Data Transfer -------------- - -web_socket_transfer_data is called after the handshake completed -successfully. A handler can receive/send messages from/to the client -using request. mod_pywebsocket.msgutil module provides utilities -for data transfer. - -You can receive a message by the following statement. - - message = request.ws_stream.receive_message() - -This call blocks until any complete text frame arrives, and the payload data -of the incoming frame will be stored into message. When you're using IETF -HyBi 00 or later protocol, receive_message() will return None on receiving -client-initiated closing handshake. When any error occurs, receive_message() -will raise some exception. - -You can send a message by the following statement. - - request.ws_stream.send_message(message) - - -Closing Connection ------------------- - -Executing the following statement or just return-ing from -web_socket_transfer_data cause connection close. - - request.ws_stream.close_connection() - -close_connection will wait -for closing handshake acknowledgement coming from the client. When it -couldn't receive a valid acknowledgement, raises an exception. - -web_socket_passive_closing_handshake is called after the server receives -incoming closing frame from the client peer immediately. You can specify -code and reason by return values. They are sent as a outgoing closing frame -from the server. A request object has the following properties that you can -use in web_socket_passive_closing_handshake. -- ws_close_code -- ws_close_reason - - -Threading ---------- - -A WebSocket handler must be thread-safe if the server (Apache or -standalone.py) is configured to use threads. -""" - - -# vi:sts=4 sw=4 et tw=72 diff --git a/module/lib/mod_pywebsocket/_stream_base.py b/module/lib/mod_pywebsocket/_stream_base.py deleted file mode 100644 index 60fb33d2c..000000000 --- a/module/lib/mod_pywebsocket/_stream_base.py +++ /dev/null @@ -1,165 +0,0 @@ -# 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. - - -"""Base stream class. -""" - - -# Note: request.connection.write/read are used in this module, even though -# mod_python document says that they should be used only in connection -# handlers. Unfortunately, we have no other options. For example, -# request.write/read are not suitable because they don't allow direct raw bytes -# writing/reading. - - -from mod_pywebsocket import util - - -# Exceptions - - -class ConnectionTerminatedException(Exception): - """This exception will be raised when a connection is terminated - unexpectedly. - """ - - pass - - -class InvalidFrameException(ConnectionTerminatedException): - """This exception will be raised when we received an invalid frame we - cannot parse. - """ - - pass - - -class BadOperationException(Exception): - """This exception will be raised when send_message() is called on - server-terminated connection or receive_message() is called on - client-terminated connection. - """ - - pass - - -class UnsupportedFrameException(Exception): - """This exception will be raised when we receive a frame with flag, opcode - we cannot handle. Handlers can just catch and ignore this exception and - call receive_message() again to continue processing the next frame. - """ - - pass - - -class InvalidUTF8Exception(Exception): - """This exception will be raised when we receive a text frame which - contains invalid UTF-8 strings. - """ - - pass - - -class StreamBase(object): - """Base stream class.""" - - def __init__(self, request): - """Construct an instance. - - Args: - request: mod_python request. - """ - - self._logger = util.get_class_logger(self) - - self._request = request - - 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. - """ - - bytes = self._request.connection.read(length) - if not bytes: - raise ConnectionTerminatedException( - 'Receiving %d byte failed. Peer (%r) closed connection' % - (length, (self._request.connection.remote_addr,))) - return bytes - - def _write(self, bytes): - """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) - except Exception, 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. - - Raises: - ConnectionTerminatedException: when read returns empty string. - """ - - bytes = [] - while length > 0: - new_bytes = self._read(length) - bytes.append(new_bytes) - length -= len(new_bytes) - return ''.join(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. - """ - - bytes = [] - while True: - ch = self._read(1) - if ch == delim_char: - break - bytes.append(ch) - return ''.join(bytes) - - -# vi:sts=4 sw=4 et diff --git a/module/lib/mod_pywebsocket/_stream_hixie75.py b/module/lib/mod_pywebsocket/_stream_hixie75.py deleted file mode 100644 index 94cf5b31b..000000000 --- a/module/lib/mod_pywebsocket/_stream_hixie75.py +++ /dev/null @@ -1,229 +0,0 @@ -# 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 a class for parsing/building frames of the WebSocket -protocol version HyBi 00 and Hixie 75. - -Specification: -- HyBi 00 http://tools.ietf.org/html/draft-ietf-hybi-thewebsocketprotocol-00 -- Hixie 75 http://tools.ietf.org/html/draft-hixie-thewebsocketprotocol-75 -""" - - -from mod_pywebsocket import common -from mod_pywebsocket._stream_base import BadOperationException -from mod_pywebsocket._stream_base import ConnectionTerminatedException -from mod_pywebsocket._stream_base import InvalidFrameException -from mod_pywebsocket._stream_base import StreamBase -from mod_pywebsocket._stream_base import UnsupportedFrameException -from mod_pywebsocket import util - - -class StreamHixie75(StreamBase): - """A class for parsing/building frames of the WebSocket protocol version - HyBi 00 and Hixie 75. - """ - - def __init__(self, request, enable_closing_handshake=False): - """Construct an instance. - - Args: - request: mod_python request. - enable_closing_handshake: to let StreamHixie75 perform closing - handshake as specified in HyBi 00, set - this option to True. - """ - - StreamBase.__init__(self, request) - - self._logger = util.get_class_logger(self) - - self._enable_closing_handshake = enable_closing_handshake - - self._request.client_terminated = False - self._request.server_terminated = False - - def send_message(self, message, end=True, binary=False): - """Send message. - - Args: - message: unicode string to send. - binary: not used in hixie75. - - Raises: - BadOperationException: when called on a server-terminated - connection. - """ - - if not end: - raise BadOperationException( - 'StreamHixie75 doesn\'t support send_message with end=False') - - if binary: - raise BadOperationException( - 'StreamHixie75 doesn\'t support send_message with binary=True') - - if self._request.server_terminated: - raise BadOperationException( - 'Requested send_message after sending out a closing handshake') - - self._write(''.join(['\x00', message.encode('utf-8'), '\xff'])) - - def _read_payload_length_hixie75(self): - """Reads a length header in a Hixie75 version frame with length. - - Raises: - ConnectionTerminatedException: when read returns empty string. - """ - - length = 0 - while True: - b_str = self._read(1) - b = ord(b_str) - length = length * 128 + (b & 0x7f) - if (b & 0x80) == 0: - break - return length - - def receive_message(self): - """Receive a WebSocket frame and return its payload an unicode string. - - Returns: - payload unicode string in a WebSocket frame. - - Raises: - ConnectionTerminatedException: when read returns empty - string. - BadOperationException: when called on a client-terminated - connection. - """ - - if self._request.client_terminated: - raise BadOperationException( - 'Requested receive_message after receiving a closing ' - 'handshake') - - while True: - # Read 1 byte. - # mp_conn.read will block if no bytes are available. - # Timeout is controlled by TimeOut directive of Apache. - frame_type_str = self.receive_bytes(1) - frame_type = ord(frame_type_str) - if (frame_type & 0x80) == 0x80: - # The payload length is specified in the frame. - # Read and discard. - length = self._read_payload_length_hixie75() - if length > 0: - _ = self.receive_bytes(length) - # 5.3 3. 12. if /type/ is 0xFF and /length/ is 0, then set the - # /client terminated/ flag and abort these steps. - if not self._enable_closing_handshake: - continue - - if frame_type == 0xFF and length == 0: - self._request.client_terminated = True - - if self._request.server_terminated: - self._logger.debug( - 'Received ack for server-initiated closing ' - 'handshake') - return None - - self._logger.debug( - 'Received client-initiated closing handshake') - - self._send_closing_handshake() - self._logger.debug( - 'Sent ack for client-initiated closing handshake') - return None - else: - # The payload is delimited with \xff. - bytes = self._read_until('\xff') - # The WebSocket protocol section 4.4 specifies that invalid - # characters must be replaced with U+fffd REPLACEMENT - # CHARACTER. - message = bytes.decode('utf-8', 'replace') - if frame_type == 0x00: - return message - # Discard data of other types. - - def _send_closing_handshake(self): - if not self._enable_closing_handshake: - raise BadOperationException( - 'Closing handshake is not supported in Hixie 75 protocol') - - self._request.server_terminated = True - - # 5.3 the server may decide to terminate the WebSocket connection by - # running through the following steps: - # 1. send a 0xFF byte and a 0x00 byte to the client to indicate the - # start of the closing handshake. - self._write('\xff\x00') - - def close_connection(self, unused_code='', unused_reason=''): - """Closes a WebSocket connection. - - Raises: - ConnectionTerminatedException: when closing handshake was - not successfull. - """ - - if self._request.server_terminated: - self._logger.debug( - 'Requested close_connection but server is already terminated') - return - - if not self._enable_closing_handshake: - self._request.server_terminated = True - self._logger.debug('Connection closed') - return - - self._send_closing_handshake() - self._logger.debug('Sent server-initiated closing handshake') - - # 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, and if we couldn't receive non-handshake - # frame, we take it as ConnectionTerminatedException. - 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): - raise BadOperationException( - 'StreamHixie75 doesn\'t support send_ping') - - -# vi:sts=4 sw=4 et diff --git a/module/lib/mod_pywebsocket/_stream_hybi.py b/module/lib/mod_pywebsocket/_stream_hybi.py deleted file mode 100644 index bd158fa6b..000000000 --- a/module/lib/mod_pywebsocket/_stream_hybi.py +++ /dev/null @@ -1,915 +0,0 @@ -# Copyright 2012, 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 collections import deque -import logging -import os -import struct -import time - -from mod_pywebsocket import common -from mod_pywebsocket import util -from mod_pywebsocket._stream_base import BadOperationException -from mod_pywebsocket._stream_base import ConnectionTerminatedException -from mod_pywebsocket._stream_base import InvalidFrameException -from mod_pywebsocket._stream_base import InvalidUTF8Exception -from mod_pywebsocket._stream_base import StreamBase -from mod_pywebsocket._stream_base import UnsupportedFrameException - - -_NOOP_MASKER = util.NoopMasker() - - -class Frame(object): - - def __init__(self, fin=1, rsv1=0, rsv2=0, rsv3=0, - opcode=None, payload=''): - 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 chr(mask_bit | length) - elif length < (1 << 16): - return chr(mask_bit | 126) + struct.pack('!H', length) - elif length < (1 << 63): - return chr(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 = '' - - first_byte = ((fin << 7) - | (rsv1 << 6) | (rsv2 << 5) | (rsv3 << 4) - | opcode) - header += chr(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') - - received = receive_bytes(2) - - first_byte = ord(received[0]) - 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(received[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): - 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() - - bytes = masker.mask(raw_payload_bytes) - - if logger.isEnabledFor(common.LOGLEVEL_FINE): - logger.log( - common.LOGLEVEL_FINE, - 'Done unmasking payload data at %s MB/s', - payload_length / (time.time() - unmask_start) / 1000 / 1000) - - return opcode, 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 = '' - 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.""" - - # Enables deflate-stream extension. - self.deflate_stream = False - - # 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 - # RFC6455 disallows fragmented control frames, but mux extension - # relaxes the restriction. - self.allow_fragmented_control_frame = False - - -class Stream(StreamBase): - """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. - """ - - StreamBase.__init__(self, request) - - self._logger = util.get_class_logger(self) - - self._options = options - - if self._options.deflate_stream: - self._logger.debug('Setup filter for deflate-stream') - self._request = util.DeflateRequest(self._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 _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, bytes, fin, rsv1, rsv2, rsv3 = self._receive_frame() - - return Frame(fin=fin, rsv1=rsv1, rsv2=rsv2, rsv3=rsv3, - opcode=opcode, payload=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, unicode): - raise BadOperationException( - 'Message for binary frame must be instance of str') - - 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, 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 = ''.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 (not self._options.allow_fragmented_control_frame and - 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) - - # Drain junk data after the close frame if necessary. - self._drain_received_data() - - 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( - 'Sent ack for client-initiated closing handshake ' - '(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, e: - 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, e: - # 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, e: - 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. - # Timeout is controlled by TimeOut directive of Apache. - - 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, 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=''): - """Closes a WebSocket connection. - - 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. - 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 - - 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, str) and not isinstance(reason, unicode): - raise BadOperationException( - 'close reason must be an instance of str or unicode') - - self._send_closing_handshake(code, reason) - self._logger.debug( - 'Sent server-initiated closing handshake (code=%r, reason=%r)', - code, reason) - - if (code == common.STATUS_GOING_AWAY or - code == common.STATUS_PROTOCOL_ERROR): - # 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=''): - 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 - - def _drain_received_data(self): - """Drains unread data in the receive buffer to avoid sending out TCP - RST packet. This is because when deflate-stream is enabled, some - DEFLATE block for flushing data may follow a close frame. If any data - remains in the receive buffer of a socket when the socket is closed, - it sends out TCP RST packet to the other peer. - - Since mod_python's mp_conn object doesn't support non-blocking read, - we perform this only when pywebsocket is running in standalone mode. - """ - - # If self._options.deflate_stream is true, self._request is - # DeflateRequest, so we can get wrapped request object by - # self._request._request. - # - # Only _StandaloneRequest has _drain_received_data method. - if (self._options.deflate_stream and - ('_drain_received_data' in dir(self._request._request))): - self._request._request._drain_received_data() - - -# vi:sts=4 sw=4 et diff --git a/module/lib/mod_pywebsocket/common.py b/module/lib/mod_pywebsocket/common.py deleted file mode 100644 index 2388379c0..000000000 --- a/module/lib/mod_pywebsocket/common.py +++ /dev/null @@ -1,307 +0,0 @@ -# Copyright 2012, 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 must not depend on any module specific to the WebSocket protocol. -""" - - -from mod_pywebsocket import http_header_util - - -# Additional log level definitions. -LOGLEVEL_FINE = 9 - -# Constants indicating WebSocket protocol version. -VERSION_HIXIE75 = -1 -VERSION_HYBI00 = 0 -VERSION_HYBI01 = 1 -VERSION_HYBI02 = 2 -VERSION_HYBI03 = 2 -VERSION_HYBI04 = 4 -VERSION_HYBI05 = 5 -VERSION_HYBI06 = 6 -VERSION_HYBI07 = 7 -VERSION_HYBI08 = 8 -VERSION_HYBI09 = 8 -VERSION_HYBI10 = 8 -VERSION_HYBI11 = 8 -VERSION_HYBI12 = 8 -VERSION_HYBI13 = 13 -VERSION_HYBI14 = 13 -VERSION_HYBI15 = 13 -VERSION_HYBI16 = 13 -VERSION_HYBI17 = 13 - -# Constants indicating WebSocket protocol latest version. -VERSION_HYBI_LATEST = VERSION_HYBI13 - -# Port numbers -DEFAULT_WEB_SOCKET_PORT = 80 -DEFAULT_WEB_SOCKET_SECURE_PORT = 443 - -# Schemes -WEB_SOCKET_SCHEME = 'ws' -WEB_SOCKET_SECURE_SCHEME = 'wss' - -# Frame opcodes defined in the spec. -OPCODE_CONTINUATION = 0x0 -OPCODE_TEXT = 0x1 -OPCODE_BINARY = 0x2 -OPCODE_CLOSE = 0x8 -OPCODE_PING = 0x9 -OPCODE_PONG = 0xa - -# UUIDs used by HyBi 04 and later opening handshake and frame masking. -WEBSOCKET_ACCEPT_UUID = '258EAFA5-E914-47DA-95CA-C5AB0DC85B11' - -# Opening handshake header names and expected values. -UPGRADE_HEADER = 'Upgrade' -WEBSOCKET_UPGRADE_TYPE = 'websocket' -WEBSOCKET_UPGRADE_TYPE_HIXIE75 = 'WebSocket' -CONNECTION_HEADER = 'Connection' -UPGRADE_CONNECTION_TYPE = 'Upgrade' -HOST_HEADER = 'Host' -ORIGIN_HEADER = 'Origin' -SEC_WEBSOCKET_ORIGIN_HEADER = 'Sec-WebSocket-Origin' -SEC_WEBSOCKET_KEY_HEADER = 'Sec-WebSocket-Key' -SEC_WEBSOCKET_ACCEPT_HEADER = 'Sec-WebSocket-Accept' -SEC_WEBSOCKET_VERSION_HEADER = 'Sec-WebSocket-Version' -SEC_WEBSOCKET_PROTOCOL_HEADER = 'Sec-WebSocket-Protocol' -SEC_WEBSOCKET_EXTENSIONS_HEADER = 'Sec-WebSocket-Extensions' -SEC_WEBSOCKET_DRAFT_HEADER = 'Sec-WebSocket-Draft' -SEC_WEBSOCKET_KEY1_HEADER = 'Sec-WebSocket-Key1' -SEC_WEBSOCKET_KEY2_HEADER = 'Sec-WebSocket-Key2' -SEC_WEBSOCKET_LOCATION_HEADER = 'Sec-WebSocket-Location' - -# Extensions -DEFLATE_STREAM_EXTENSION = 'deflate-stream' -DEFLATE_FRAME_EXTENSION = 'deflate-frame' -PERFRAME_COMPRESSION_EXTENSION = 'perframe-compress' -PERMESSAGE_COMPRESSION_EXTENSION = 'permessage-compress' -X_WEBKIT_DEFLATE_FRAME_EXTENSION = 'x-webkit-deflate-frame' -X_WEBKIT_PERMESSAGE_COMPRESSION_EXTENSION = 'x-webkit-permessage-compress' -MUX_EXTENSION = 'mux_DO_NOT_USE' - -# Status codes -# Code STATUS_NO_STATUS_RECEIVED, STATUS_ABNORMAL_CLOSURE, and -# STATUS_TLS_HANDSHAKE are pseudo codes to indicate specific error cases. -# Could not be used for codes in actual closing frames. -# Application level errors must use codes in the range -# STATUS_USER_REGISTERED_BASE to STATUS_USER_PRIVATE_MAX. The codes in the -# range STATUS_USER_REGISTERED_BASE to STATUS_USER_REGISTERED_MAX are managed -# by IANA. Usually application must define user protocol level errors in the -# range STATUS_USER_PRIVATE_BASE to STATUS_USER_PRIVATE_MAX. -STATUS_NORMAL_CLOSURE = 1000 -STATUS_GOING_AWAY = 1001 -STATUS_PROTOCOL_ERROR = 1002 -STATUS_UNSUPPORTED_DATA = 1003 -STATUS_NO_STATUS_RECEIVED = 1005 -STATUS_ABNORMAL_CLOSURE = 1006 -STATUS_INVALID_FRAME_PAYLOAD_DATA = 1007 -STATUS_POLICY_VIOLATION = 1008 -STATUS_MESSAGE_TOO_BIG = 1009 -STATUS_MANDATORY_EXTENSION = 1010 -STATUS_INTERNAL_ENDPOINT_ERROR = 1011 -STATUS_TLS_HANDSHAKE = 1015 -STATUS_USER_REGISTERED_BASE = 3000 -STATUS_USER_REGISTERED_MAX = 3999 -STATUS_USER_PRIVATE_BASE = 4000 -STATUS_USER_PRIVATE_MAX = 4999 -# Following definitions are aliases to keep compatibility. Applications must -# not use these obsoleted definitions anymore. -STATUS_NORMAL = STATUS_NORMAL_CLOSURE -STATUS_UNSUPPORTED = STATUS_UNSUPPORTED_DATA -STATUS_CODE_NOT_AVAILABLE = STATUS_NO_STATUS_RECEIVED -STATUS_ABNORMAL_CLOSE = STATUS_ABNORMAL_CLOSURE -STATUS_INVALID_FRAME_PAYLOAD = STATUS_INVALID_FRAME_PAYLOAD_DATA -STATUS_MANDATORY_EXT = STATUS_MANDATORY_EXTENSION - -# HTTP status codes -HTTP_STATUS_BAD_REQUEST = 400 -HTTP_STATUS_FORBIDDEN = 403 -HTTP_STATUS_NOT_FOUND = 404 - - -def is_control_opcode(opcode): - return (opcode >> 3) == 1 - - -class ExtensionParameter(object): - """Holds information about an extension which is exchanged on extension - negotiation in opening handshake. - """ - - def __init__(self, name): - self._name = name - # TODO(tyoshino): Change the data structure to more efficient one such - # as dict when the spec changes to say like - # - Parameter names must be unique - # - The order of parameters is not significant - self._parameters = [] - - def name(self): - return self._name - - def add_parameter(self, name, value): - self._parameters.append((name, value)) - - def get_parameters(self): - return self._parameters - - def get_parameter_names(self): - return [name for name, unused_value in self._parameters] - - def has_parameter(self, name): - for param_name, param_value in self._parameters: - if param_name == name: - return True - return False - - def get_parameter_value(self, name): - for param_name, param_value in self._parameters: - if param_name == name: - return param_value - - -class ExtensionParsingException(Exception): - def __init__(self, name): - super(ExtensionParsingException, self).__init__(name) - - -def _parse_extension_param(state, definition, allow_quoted_string): - param_name = http_header_util.consume_token(state) - - if param_name is None: - raise ExtensionParsingException('No valid parameter name found') - - http_header_util.consume_lwses(state) - - if not http_header_util.consume_string(state, '='): - definition.add_parameter(param_name, None) - return - - http_header_util.consume_lwses(state) - - if allow_quoted_string: - # TODO(toyoshim): Add code to validate that parsed param_value is token - param_value = http_header_util.consume_token_or_quoted_string(state) - else: - param_value = http_header_util.consume_token(state) - if param_value is None: - raise ExtensionParsingException( - 'No valid parameter value found on the right-hand side of ' - 'parameter %r' % param_name) - - definition.add_parameter(param_name, param_value) - - -def _parse_extension(state, allow_quoted_string): - extension_token = http_header_util.consume_token(state) - if extension_token is None: - return None - - extension = ExtensionParameter(extension_token) - - while True: - http_header_util.consume_lwses(state) - - if not http_header_util.consume_string(state, ';'): - break - - http_header_util.consume_lwses(state) - - try: - _parse_extension_param(state, extension, allow_quoted_string) - except ExtensionParsingException, e: - raise ExtensionParsingException( - 'Failed to parse parameter for %r (%r)' % - (extension_token, e)) - - return extension - - -def parse_extensions(data, allow_quoted_string=False): - """Parses Sec-WebSocket-Extensions header value returns a list of - ExtensionParameter objects. - - Leading LWSes must be trimmed. - """ - - state = http_header_util.ParsingState(data) - - extension_list = [] - while True: - extension = _parse_extension(state, allow_quoted_string) - if extension is not None: - extension_list.append(extension) - - http_header_util.consume_lwses(state) - - if http_header_util.peek(state) is None: - break - - if not http_header_util.consume_string(state, ','): - raise ExtensionParsingException( - 'Failed to parse Sec-WebSocket-Extensions header: ' - 'Expected a comma but found %r' % - http_header_util.peek(state)) - - http_header_util.consume_lwses(state) - - if len(extension_list) == 0: - raise ExtensionParsingException( - 'No valid extension entry found') - - return extension_list - - -def format_extension(extension): - """Formats an ExtensionParameter object.""" - - formatted_params = [extension.name()] - for param_name, param_value in extension.get_parameters(): - if param_value is None: - formatted_params.append(param_name) - else: - quoted_value = http_header_util.quote_if_necessary(param_value) - formatted_params.append('%s=%s' % (param_name, quoted_value)) - return '; '.join(formatted_params) - - -def format_extensions(extension_list): - """Formats a list of ExtensionParameter objects.""" - - formatted_extension_list = [] - for extension in extension_list: - formatted_extension_list.append(format_extension(extension)) - return ', '.join(formatted_extension_list) - - -# vi:sts=4 sw=4 et diff --git a/module/lib/mod_pywebsocket/dispatch.py b/module/lib/mod_pywebsocket/dispatch.py deleted file mode 100644 index 25905f180..000000000 --- a/module/lib/mod_pywebsocket/dispatch.py +++ /dev/null @@ -1,387 +0,0 @@ -# Copyright 2012, 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. - - -"""Dispatch WebSocket request. -""" - - -import logging -import os -import re - -from mod_pywebsocket import common -from mod_pywebsocket import handshake -from mod_pywebsocket import msgutil -from mod_pywebsocket import mux -from mod_pywebsocket import stream -from mod_pywebsocket import util - - -_SOURCE_PATH_PATTERN = re.compile(r'(?i)_wsh\.py$') -_SOURCE_SUFFIX = '_wsh.py' -_DO_EXTRA_HANDSHAKE_HANDLER_NAME = 'web_socket_do_extra_handshake' -_TRANSFER_DATA_HANDLER_NAME = 'web_socket_transfer_data' -_PASSIVE_CLOSING_HANDSHAKE_HANDLER_NAME = ( - 'web_socket_passive_closing_handshake') - - -class DispatchException(Exception): - """Exception in dispatching WebSocket request.""" - - def __init__(self, name, status=common.HTTP_STATUS_NOT_FOUND): - super(DispatchException, self).__init__(name) - self.status = status - - -def _default_passive_closing_handshake_handler(request): - """Default web_socket_passive_closing_handshake handler.""" - - return common.STATUS_NORMAL_CLOSURE, '' - - -def _normalize_path(path): - """Normalize path. - - Args: - path: the path to normalize. - - Path is converted to the absolute path. - The input path can use either '\\' or '/' as the separator. - The normalized path always uses '/' regardless of the platform. - """ - - path = path.replace('\\', os.path.sep) - path = os.path.realpath(path) - path = path.replace('\\', '/') - return path - - -def _create_path_to_resource_converter(base_dir): - """Returns a function that converts the path of a WebSocket handler source - file to a resource string by removing the path to the base directory from - its head, removing _SOURCE_SUFFIX from its tail, and replacing path - separators in it with '/'. - - Args: - base_dir: the path to the base directory. - """ - - base_dir = _normalize_path(base_dir) - - base_len = len(base_dir) - suffix_len = len(_SOURCE_SUFFIX) - - def converter(path): - if not path.endswith(_SOURCE_SUFFIX): - return None - # _normalize_path must not be used because resolving symlink breaks - # following path check. - path = path.replace('\\', '/') - if not path.startswith(base_dir): - return None - return path[base_len:-suffix_len] - - return converter - - -def _enumerate_handler_file_paths(directory): - """Returns a generator that enumerates WebSocket Handler source file names - in the given directory. - """ - - for root, unused_dirs, files in os.walk(directory): - for base in files: - path = os.path.join(root, base) - if _SOURCE_PATH_PATTERN.search(path): - yield path - - -class _HandlerSuite(object): - """A handler suite holder class.""" - - def __init__(self, do_extra_handshake, transfer_data, - passive_closing_handshake): - self.do_extra_handshake = do_extra_handshake - self.transfer_data = transfer_data - self.passive_closing_handshake = passive_closing_handshake - - -def _source_handler_file(handler_definition): - """Source a handler definition string. - - Args: - handler_definition: a string containing Python statements that define - handler functions. - """ - - global_dic = {} - try: - exec handler_definition in global_dic - except Exception: - raise DispatchException('Error in sourcing handler:' + - util.get_stack_trace()) - passive_closing_handshake_handler = None - try: - passive_closing_handshake_handler = _extract_handler( - global_dic, _PASSIVE_CLOSING_HANDSHAKE_HANDLER_NAME) - except Exception: - passive_closing_handshake_handler = ( - _default_passive_closing_handshake_handler) - return _HandlerSuite( - _extract_handler(global_dic, _DO_EXTRA_HANDSHAKE_HANDLER_NAME), - _extract_handler(global_dic, _TRANSFER_DATA_HANDLER_NAME), - passive_closing_handshake_handler) - - -def _extract_handler(dic, name): - """Extracts a callable with the specified name from the given dictionary - dic. - """ - - if name not in dic: - raise DispatchException('%s is not defined.' % name) - handler = dic[name] - if not callable(handler): - raise DispatchException('%s is not callable.' % name) - return handler - - -class Dispatcher(object): - """Dispatches WebSocket requests. - - This class maintains a map from resource name to handlers. - """ - - def __init__( - self, root_dir, scan_dir=None, - allow_handlers_outside_root_dir=True): - """Construct an instance. - - Args: - root_dir: The directory where handler definition files are - placed. - scan_dir: The directory where handler definition files are - searched. scan_dir must be a directory under root_dir, - including root_dir itself. If scan_dir is None, - root_dir is used as scan_dir. scan_dir can be useful - in saving scan time when root_dir contains many - subdirectories. - allow_handlers_outside_root_dir: Scans handler files even if their - canonical path is not under root_dir. - """ - - self._logger = util.get_class_logger(self) - - self._handler_suite_map = {} - self._source_warnings = [] - if scan_dir is None: - scan_dir = root_dir - if not os.path.realpath(scan_dir).startswith( - os.path.realpath(root_dir)): - raise DispatchException('scan_dir:%s must be a directory under ' - 'root_dir:%s.' % (scan_dir, root_dir)) - self._source_handler_files_in_dir( - root_dir, scan_dir, allow_handlers_outside_root_dir) - - def add_resource_path_alias(self, - alias_resource_path, existing_resource_path): - """Add resource path alias. - - Once added, request to alias_resource_path would be handled by - handler registered for existing_resource_path. - - Args: - alias_resource_path: alias resource path - existing_resource_path: existing resource path - """ - try: - handler_suite = self._handler_suite_map[existing_resource_path] - self._handler_suite_map[alias_resource_path] = handler_suite - except KeyError: - raise DispatchException('No handler for: %r' % - existing_resource_path) - - def source_warnings(self): - """Return warnings in sourcing handlers.""" - - return self._source_warnings - - def do_extra_handshake(self, request): - """Do extra checking in WebSocket handshake. - - Select a handler based on request.uri and call its - web_socket_do_extra_handshake function. - - Args: - request: mod_python request. - - Raises: - DispatchException: when handler was not found - AbortedByUserException: when user handler abort connection - HandshakeException: when opening handshake failed - """ - - handler_suite = self.get_handler_suite(request.ws_resource) - if handler_suite is None: - raise DispatchException('No handler for: %r' % request.ws_resource) - do_extra_handshake_ = handler_suite.do_extra_handshake - try: - do_extra_handshake_(request) - except handshake.AbortedByUserException, e: - raise - except Exception, e: - util.prepend_message_to_exception( - '%s raised exception for %s: ' % ( - _DO_EXTRA_HANDSHAKE_HANDLER_NAME, - request.ws_resource), - e) - raise handshake.HandshakeException(e, common.HTTP_STATUS_FORBIDDEN) - - def transfer_data(self, request): - """Let a handler transfer_data with a WebSocket client. - - Select a handler based on request.ws_resource and call its - web_socket_transfer_data function. - - Args: - request: mod_python request. - - Raises: - DispatchException: when handler was not found - AbortedByUserException: when user handler abort connection - """ - - # TODO(tyoshino): Terminate underlying TCP connection if possible. - try: - if mux.use_mux(request): - mux.start(request, self) - else: - handler_suite = self.get_handler_suite(request.ws_resource) - if handler_suite is None: - raise DispatchException('No handler for: %r' % - request.ws_resource) - transfer_data_ = handler_suite.transfer_data - transfer_data_(request) - - if not request.server_terminated: - request.ws_stream.close_connection() - # Catch non-critical exceptions the handler didn't handle. - except handshake.AbortedByUserException, e: - self._logger.debug('%s', e) - raise - except msgutil.BadOperationException, e: - self._logger.debug('%s', e) - request.ws_stream.close_connection(common.STATUS_ABNORMAL_CLOSURE) - except msgutil.InvalidFrameException, e: - # InvalidFrameException must be caught before - # ConnectionTerminatedException that catches InvalidFrameException. - self._logger.debug('%s', e) - request.ws_stream.close_connection(common.STATUS_PROTOCOL_ERROR) - except msgutil.UnsupportedFrameException, e: - self._logger.debug('%s', e) - request.ws_stream.close_connection(common.STATUS_UNSUPPORTED_DATA) - except stream.InvalidUTF8Exception, e: - self._logger.debug('%s', e) - request.ws_stream.close_connection( - common.STATUS_INVALID_FRAME_PAYLOAD_DATA) - except msgutil.ConnectionTerminatedException, e: - self._logger.debug('%s', e) - except Exception, e: - util.prepend_message_to_exception( - '%s raised exception for %s: ' % ( - _TRANSFER_DATA_HANDLER_NAME, request.ws_resource), - e) - raise - - def passive_closing_handshake(self, request): - """Prepare code and reason for responding client initiated closing - handshake. - """ - - handler_suite = self.get_handler_suite(request.ws_resource) - if handler_suite is None: - return _default_passive_closing_handshake_handler(request) - return handler_suite.passive_closing_handshake(request) - - def get_handler_suite(self, resource): - """Retrieves two handlers (one for extra handshake processing, and one - for data transfer) for the given request as a HandlerSuite object. - """ - - fragment = None - if '#' in resource: - resource, fragment = resource.split('#', 1) - if '?' in resource: - resource = resource.split('?', 1)[0] - handler_suite = self._handler_suite_map.get(resource) - if handler_suite and fragment: - raise DispatchException('Fragment identifiers MUST NOT be used on ' - 'WebSocket URIs', - common.HTTP_STATUS_BAD_REQUEST) - return handler_suite - - def _source_handler_files_in_dir( - self, root_dir, scan_dir, allow_handlers_outside_root_dir): - """Source all the handler source files in the scan_dir directory. - - The resource path is determined relative to root_dir. - """ - - # We build a map from resource to handler code assuming that there's - # only one path from root_dir to scan_dir and it can be obtained by - # comparing realpath of them. - - # Here we cannot use abspath. See - # https://bugs.webkit.org/show_bug.cgi?id=31603 - - convert = _create_path_to_resource_converter(root_dir) - scan_realpath = os.path.realpath(scan_dir) - root_realpath = os.path.realpath(root_dir) - for path in _enumerate_handler_file_paths(scan_realpath): - if (not allow_handlers_outside_root_dir and - (not os.path.realpath(path).startswith(root_realpath))): - self._logger.debug( - 'Canonical path of %s is not under root directory' % - path) - continue - try: - handler_suite = _source_handler_file(open(path).read()) - except DispatchException, e: - self._source_warnings.append('%s: %s' % (path, e)) - continue - resource = convert(path) - if resource is None: - self._logger.debug( - 'Path to resource conversion on %s failed' % path) - else: - self._handler_suite_map[convert(path)] = handler_suite - - -# vi:sts=4 sw=4 et diff --git a/module/lib/mod_pywebsocket/extensions.py b/module/lib/mod_pywebsocket/extensions.py deleted file mode 100644 index 03dbf9ee1..000000000 --- a/module/lib/mod_pywebsocket/extensions.py +++ /dev/null @@ -1,727 +0,0 @@ -# Copyright 2012, 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. - - -from mod_pywebsocket import common -from mod_pywebsocket import util -from mod_pywebsocket.http_header_util import quote_if_necessary - - -_available_processors = {} - - -class ExtensionProcessorInterface(object): - - def name(self): - return None - - def get_extension_response(self): - return None - - def setup_stream_options(self, stream_options): - pass - - -class DeflateStreamExtensionProcessor(ExtensionProcessorInterface): - """WebSocket DEFLATE stream extension processor. - - Specification: - Section 9.2.1 in - http://tools.ietf.org/html/draft-ietf-hybi-thewebsocketprotocol-10 - """ - - def __init__(self, request): - self._logger = util.get_class_logger(self) - - self._request = request - - def name(self): - return common.DEFLATE_STREAM_EXTENSION - - def get_extension_response(self): - if len(self._request.get_parameter_names()) != 0: - return None - - self._logger.debug( - 'Enable %s extension', common.DEFLATE_STREAM_EXTENSION) - - return common.ExtensionParameter(common.DEFLATE_STREAM_EXTENSION) - - def setup_stream_options(self, stream_options): - stream_options.deflate_stream = True - - -_available_processors[common.DEFLATE_STREAM_EXTENSION] = ( - DeflateStreamExtensionProcessor) - - -def _log_compression_ratio(logger, original_bytes, total_original_bytes, - filtered_bytes, total_filtered_bytes): - # Print inf when ratio is not available. - ratio = float('inf') - average_ratio = float('inf') - if original_bytes != 0: - ratio = float(filtered_bytes) / original_bytes - if total_original_bytes != 0: - average_ratio = ( - float(total_filtered_bytes) / total_original_bytes) - logger.debug('Outgoing compress ratio: %f (average: %f)' % - (ratio, average_ratio)) - - -def _log_decompression_ratio(logger, received_bytes, total_received_bytes, - filtered_bytes, total_filtered_bytes): - # Print inf when ratio is not available. - ratio = float('inf') - average_ratio = float('inf') - if received_bytes != 0: - ratio = float(received_bytes) / filtered_bytes - if total_filtered_bytes != 0: - average_ratio = ( - float(total_received_bytes) / total_filtered_bytes) - logger.debug('Incoming compress ratio: %f (average: %f)' % - (ratio, average_ratio)) - - -class DeflateFrameExtensionProcessor(ExtensionProcessorInterface): - """WebSocket Per-frame DEFLATE extension processor. - - Specification: - http://tools.ietf.org/html/draft-tyoshino-hybi-websocket-perframe-deflate - """ - - _WINDOW_BITS_PARAM = 'max_window_bits' - _NO_CONTEXT_TAKEOVER_PARAM = 'no_context_takeover' - - def __init__(self, request): - self._logger = util.get_class_logger(self) - - self._request = request - - self._response_window_bits = None - self._response_no_context_takeover = False - self._bfinal = False - - # Counters for statistics. - - # Total number of outgoing bytes supplied to this filter. - self._total_outgoing_payload_bytes = 0 - # Total number of bytes sent to the network after applying this filter. - self._total_filtered_outgoing_payload_bytes = 0 - - # Total number of bytes received from the network. - self._total_incoming_payload_bytes = 0 - # Total number of incoming bytes obtained after applying this filter. - self._total_filtered_incoming_payload_bytes = 0 - - def name(self): - return common.DEFLATE_FRAME_EXTENSION - - def get_extension_response(self): - # Any unknown parameter will be just ignored. - - window_bits = self._request.get_parameter_value( - self._WINDOW_BITS_PARAM) - no_context_takeover = self._request.has_parameter( - self._NO_CONTEXT_TAKEOVER_PARAM) - if (no_context_takeover and - self._request.get_parameter_value( - self._NO_CONTEXT_TAKEOVER_PARAM) is not None): - return None - - if window_bits is not None: - try: - window_bits = int(window_bits) - except ValueError, e: - return None - if window_bits < 8 or window_bits > 15: - return None - - self._deflater = util._RFC1979Deflater( - window_bits, no_context_takeover) - - self._inflater = util._RFC1979Inflater() - - self._compress_outgoing = True - - response = common.ExtensionParameter(self._request.name()) - - if self._response_window_bits is not None: - response.add_parameter( - self._WINDOW_BITS_PARAM, str(self._response_window_bits)) - if self._response_no_context_takeover: - response.add_parameter( - self._NO_CONTEXT_TAKEOVER_PARAM, None) - - self._logger.debug( - 'Enable %s extension (' - 'request: window_bits=%s; no_context_takeover=%r, ' - 'response: window_wbits=%s; no_context_takeover=%r)' % - (self._request.name(), - window_bits, - no_context_takeover, - self._response_window_bits, - self._response_no_context_takeover)) - - return response - - def setup_stream_options(self, stream_options): - - class _OutgoingFilter(object): - - def __init__(self, parent): - self._parent = parent - - def filter(self, frame): - self._parent._outgoing_filter(frame) - - class _IncomingFilter(object): - - def __init__(self, parent): - self._parent = parent - - def filter(self, frame): - self._parent._incoming_filter(frame) - - stream_options.outgoing_frame_filters.append( - _OutgoingFilter(self)) - stream_options.incoming_frame_filters.insert( - 0, _IncomingFilter(self)) - - def set_response_window_bits(self, value): - self._response_window_bits = value - - def set_response_no_context_takeover(self, value): - self._response_no_context_takeover = value - - def set_bfinal(self, value): - self._bfinal = value - - def enable_outgoing_compression(self): - self._compress_outgoing = True - - def disable_outgoing_compression(self): - self._compress_outgoing = False - - def _outgoing_filter(self, frame): - """Transform outgoing frames. This method is called only by - an _OutgoingFilter instance. - """ - - original_payload_size = len(frame.payload) - self._total_outgoing_payload_bytes += original_payload_size - - if (not self._compress_outgoing or - common.is_control_opcode(frame.opcode)): - self._total_filtered_outgoing_payload_bytes += ( - original_payload_size) - return - - frame.payload = self._deflater.filter( - frame.payload, bfinal=self._bfinal) - frame.rsv1 = 1 - - filtered_payload_size = len(frame.payload) - self._total_filtered_outgoing_payload_bytes += filtered_payload_size - - _log_compression_ratio(self._logger, original_payload_size, - self._total_outgoing_payload_bytes, - filtered_payload_size, - self._total_filtered_outgoing_payload_bytes) - - def _incoming_filter(self, frame): - """Transform incoming frames. This method is called only by - an _IncomingFilter instance. - """ - - received_payload_size = len(frame.payload) - self._total_incoming_payload_bytes += received_payload_size - - if frame.rsv1 != 1 or common.is_control_opcode(frame.opcode): - self._total_filtered_incoming_payload_bytes += ( - received_payload_size) - return - - frame.payload = self._inflater.filter(frame.payload) - frame.rsv1 = 0 - - filtered_payload_size = len(frame.payload) - self._total_filtered_incoming_payload_bytes += filtered_payload_size - - _log_decompression_ratio(self._logger, received_payload_size, - self._total_incoming_payload_bytes, - filtered_payload_size, - self._total_filtered_incoming_payload_bytes) - - -_available_processors[common.DEFLATE_FRAME_EXTENSION] = ( - DeflateFrameExtensionProcessor) - - -# Adding vendor-prefixed deflate-frame extension. -# TODO(bashi): Remove this after WebKit stops using vendor prefix. -_available_processors[common.X_WEBKIT_DEFLATE_FRAME_EXTENSION] = ( - DeflateFrameExtensionProcessor) - - -def _parse_compression_method(data): - """Parses the value of "method" extension parameter.""" - - return common.parse_extensions(data, allow_quoted_string=True) - - -def _create_accepted_method_desc(method_name, method_params): - """Creates accepted-method-desc from given method name and parameters""" - - extension = common.ExtensionParameter(method_name) - for name, value in method_params: - extension.add_parameter(name, value) - return common.format_extension(extension) - - -class CompressionExtensionProcessorBase(ExtensionProcessorInterface): - """Base class for Per-frame and Per-message compression extension.""" - - _METHOD_PARAM = 'method' - - def __init__(self, request): - self._logger = util.get_class_logger(self) - self._request = request - self._compression_method_name = None - self._compression_processor = None - self._compression_processor_hook = None - - def name(self): - return '' - - def _lookup_compression_processor(self, method_desc): - return None - - def _get_compression_processor_response(self): - """Looks up the compression processor based on the self._request and - returns the compression processor's response. - """ - - method_list = self._request.get_parameter_value(self._METHOD_PARAM) - if method_list is None: - return None - methods = _parse_compression_method(method_list) - if methods is None: - return None - comression_processor = None - # The current implementation tries only the first method that matches - # supported algorithm. Following methods aren't tried even if the - # first one is rejected. - # TODO(bashi): Need to clarify this behavior. - for method_desc in methods: - compression_processor = self._lookup_compression_processor( - method_desc) - if compression_processor is not None: - self._compression_method_name = method_desc.name() - break - if compression_processor is None: - return None - - if self._compression_processor_hook: - self._compression_processor_hook(compression_processor) - - processor_response = compression_processor.get_extension_response() - if processor_response is None: - return None - self._compression_processor = compression_processor - return processor_response - - def get_extension_response(self): - processor_response = self._get_compression_processor_response() - if processor_response is None: - return None - - response = common.ExtensionParameter(self._request.name()) - accepted_method_desc = _create_accepted_method_desc( - self._compression_method_name, - processor_response.get_parameters()) - response.add_parameter(self._METHOD_PARAM, accepted_method_desc) - self._logger.debug( - 'Enable %s extension (method: %s)' % - (self._request.name(), self._compression_method_name)) - return response - - def setup_stream_options(self, stream_options): - if self._compression_processor is None: - return - self._compression_processor.setup_stream_options(stream_options) - - def set_compression_processor_hook(self, hook): - self._compression_processor_hook = hook - - def get_compression_processor(self): - return self._compression_processor - - -class PerFrameCompressionExtensionProcessor(CompressionExtensionProcessorBase): - """WebSocket Per-frame compression extension processor. - - Specification: - http://tools.ietf.org/html/draft-ietf-hybi-websocket-perframe-compression - """ - - _DEFLATE_METHOD = 'deflate' - - def __init__(self, request): - CompressionExtensionProcessorBase.__init__(self, request) - - def name(self): - return common.PERFRAME_COMPRESSION_EXTENSION - - def _lookup_compression_processor(self, method_desc): - if method_desc.name() == self._DEFLATE_METHOD: - return DeflateFrameExtensionProcessor(method_desc) - return None - - -_available_processors[common.PERFRAME_COMPRESSION_EXTENSION] = ( - PerFrameCompressionExtensionProcessor) - - -class DeflateMessageProcessor(ExtensionProcessorInterface): - """Per-message deflate processor.""" - - _S2C_MAX_WINDOW_BITS_PARAM = 's2c_max_window_bits' - _S2C_NO_CONTEXT_TAKEOVER_PARAM = 's2c_no_context_takeover' - _C2S_MAX_WINDOW_BITS_PARAM = 'c2s_max_window_bits' - _C2S_NO_CONTEXT_TAKEOVER_PARAM = 'c2s_no_context_takeover' - - def __init__(self, request): - self._request = request - self._logger = util.get_class_logger(self) - - self._c2s_max_window_bits = None - self._c2s_no_context_takeover = False - self._bfinal = False - - self._compress_outgoing_enabled = False - - # True if a message is fragmented and compression is ongoing. - self._compress_ongoing = False - - # Counters for statistics. - - # Total number of outgoing bytes supplied to this filter. - self._total_outgoing_payload_bytes = 0 - # Total number of bytes sent to the network after applying this filter. - self._total_filtered_outgoing_payload_bytes = 0 - - # Total number of bytes received from the network. - self._total_incoming_payload_bytes = 0 - # Total number of incoming bytes obtained after applying this filter. - self._total_filtered_incoming_payload_bytes = 0 - - def name(self): - return 'deflate' - - def get_extension_response(self): - # Any unknown parameter will be just ignored. - - s2c_max_window_bits = self._request.get_parameter_value( - self._S2C_MAX_WINDOW_BITS_PARAM) - if s2c_max_window_bits is not None: - try: - s2c_max_window_bits = int(s2c_max_window_bits) - except ValueError, e: - return None - if s2c_max_window_bits < 8 or s2c_max_window_bits > 15: - return None - - s2c_no_context_takeover = self._request.has_parameter( - self._S2C_NO_CONTEXT_TAKEOVER_PARAM) - if (s2c_no_context_takeover and - self._request.get_parameter_value( - self._S2C_NO_CONTEXT_TAKEOVER_PARAM) is not None): - return None - - self._deflater = util._RFC1979Deflater( - s2c_max_window_bits, s2c_no_context_takeover) - - self._inflater = util._RFC1979Inflater() - - self._compress_outgoing_enabled = True - - response = common.ExtensionParameter(self._request.name()) - - if s2c_max_window_bits is not None: - response.add_parameter( - self._S2C_MAX_WINDOW_BITS_PARAM, str(s2c_max_window_bits)) - - if s2c_no_context_takeover: - response.add_parameter( - self._S2C_NO_CONTEXT_TAKEOVER_PARAM, None) - - if self._c2s_max_window_bits is not None: - response.add_parameter( - self._C2S_MAX_WINDOW_BITS_PARAM, - str(self._c2s_max_window_bits)) - if self._c2s_no_context_takeover: - response.add_parameter( - self._C2S_NO_CONTEXT_TAKEOVER_PARAM, None) - - self._logger.debug( - 'Enable %s extension (' - 'request: s2c_max_window_bits=%s; s2c_no_context_takeover=%r, ' - 'response: c2s_max_window_bits=%s; c2s_no_context_takeover=%r)' % - (self._request.name(), - s2c_max_window_bits, - s2c_no_context_takeover, - self._c2s_max_window_bits, - self._c2s_no_context_takeover)) - - return response - - def setup_stream_options(self, stream_options): - class _OutgoingMessageFilter(object): - - def __init__(self, parent): - self._parent = parent - - def filter(self, message, end=True, binary=False): - return self._parent._process_outgoing_message( - message, end, binary) - - class _IncomingMessageFilter(object): - - def __init__(self, parent): - self._parent = parent - self._decompress_next_message = False - - def decompress_next_message(self): - self._decompress_next_message = True - - def filter(self, message): - message = self._parent._process_incoming_message( - message, self._decompress_next_message) - self._decompress_next_message = False - return message - - self._outgoing_message_filter = _OutgoingMessageFilter(self) - self._incoming_message_filter = _IncomingMessageFilter(self) - stream_options.outgoing_message_filters.append( - self._outgoing_message_filter) - stream_options.incoming_message_filters.append( - self._incoming_message_filter) - - class _OutgoingFrameFilter(object): - - def __init__(self, parent): - self._parent = parent - self._set_compression_bit = False - - def set_compression_bit(self): - self._set_compression_bit = True - - def filter(self, frame): - self._parent._process_outgoing_frame( - frame, self._set_compression_bit) - self._set_compression_bit = False - - class _IncomingFrameFilter(object): - - def __init__(self, parent): - self._parent = parent - - def filter(self, frame): - self._parent._process_incoming_frame(frame) - - self._outgoing_frame_filter = _OutgoingFrameFilter(self) - self._incoming_frame_filter = _IncomingFrameFilter(self) - stream_options.outgoing_frame_filters.append( - self._outgoing_frame_filter) - stream_options.incoming_frame_filters.append( - self._incoming_frame_filter) - - stream_options.encode_text_message_to_utf8 = False - - def set_c2s_max_window_bits(self, value): - self._c2s_max_window_bits = value - - def set_c2s_no_context_takeover(self, value): - self._c2s_no_context_takeover = value - - def set_bfinal(self, value): - self._bfinal = value - - def enable_outgoing_compression(self): - self._compress_outgoing_enabled = True - - def disable_outgoing_compression(self): - self._compress_outgoing_enabled = False - - def _process_incoming_message(self, message, decompress): - if not decompress: - return message - - received_payload_size = len(message) - self._total_incoming_payload_bytes += received_payload_size - - message = self._inflater.filter(message) - - filtered_payload_size = len(message) - self._total_filtered_incoming_payload_bytes += filtered_payload_size - - _log_decompression_ratio(self._logger, received_payload_size, - self._total_incoming_payload_bytes, - filtered_payload_size, - self._total_filtered_incoming_payload_bytes) - - return message - - def _process_outgoing_message(self, message, end, binary): - if not binary: - message = message.encode('utf-8') - - if not self._compress_outgoing_enabled: - return message - - original_payload_size = len(message) - self._total_outgoing_payload_bytes += original_payload_size - - message = self._deflater.filter( - message, flush=end, bfinal=self._bfinal) - - filtered_payload_size = len(message) - self._total_filtered_outgoing_payload_bytes += filtered_payload_size - - _log_compression_ratio(self._logger, original_payload_size, - self._total_outgoing_payload_bytes, - filtered_payload_size, - self._total_filtered_outgoing_payload_bytes) - - if not self._compress_ongoing: - self._outgoing_frame_filter.set_compression_bit() - self._compress_ongoing = not end - return message - - def _process_incoming_frame(self, frame): - if frame.rsv1 == 1 and not common.is_control_opcode(frame.opcode): - self._incoming_message_filter.decompress_next_message() - frame.rsv1 = 0 - - def _process_outgoing_frame(self, frame, compression_bit): - if (not compression_bit or - common.is_control_opcode(frame.opcode)): - return - - frame.rsv1 = 1 - - -class PerMessageCompressionExtensionProcessor( - CompressionExtensionProcessorBase): - """WebSocket Per-message compression extension processor. - - Specification: - http://tools.ietf.org/html/draft-ietf-hybi-permessage-compression - """ - - _DEFLATE_METHOD = 'deflate' - - def __init__(self, request): - CompressionExtensionProcessorBase.__init__(self, request) - - def name(self): - return common.PERMESSAGE_COMPRESSION_EXTENSION - - def _lookup_compression_processor(self, method_desc): - if method_desc.name() == self._DEFLATE_METHOD: - return DeflateMessageProcessor(method_desc) - return None - - -_available_processors[common.PERMESSAGE_COMPRESSION_EXTENSION] = ( - PerMessageCompressionExtensionProcessor) - - -# Adding vendor-prefixed permessage-compress extension. -# TODO(bashi): Remove this after WebKit stops using vendor prefix. -_available_processors[common.X_WEBKIT_PERMESSAGE_COMPRESSION_EXTENSION] = ( - PerMessageCompressionExtensionProcessor) - - -class MuxExtensionProcessor(ExtensionProcessorInterface): - """WebSocket multiplexing extension processor.""" - - _QUOTA_PARAM = 'quota' - - def __init__(self, request): - self._request = request - - def name(self): - return common.MUX_EXTENSION - - def get_extension_response(self, ws_request, - logical_channel_extensions): - # Mux extension cannot be used after extensions that depend on - # frame boundary, extension data field, or any reserved bits - # which are attributed to each frame. - for extension in logical_channel_extensions: - name = extension.name() - if (name == common.PERFRAME_COMPRESSION_EXTENSION or - name == common.DEFLATE_FRAME_EXTENSION or - name == common.X_WEBKIT_DEFLATE_FRAME_EXTENSION): - return None - - quota = self._request.get_parameter_value(self._QUOTA_PARAM) - if quota is None: - ws_request.mux_quota = 0 - else: - try: - quota = int(quota) - except ValueError, e: - return None - if quota < 0 or quota >= 2 ** 32: - return None - ws_request.mux_quota = quota - - ws_request.mux = True - ws_request.mux_extensions = logical_channel_extensions - return common.ExtensionParameter(common.MUX_EXTENSION) - - def setup_stream_options(self, stream_options): - pass - - -_available_processors[common.MUX_EXTENSION] = MuxExtensionProcessor - - -def get_extension_processor(extension_request): - global _available_processors - processor_class = _available_processors.get(extension_request.name()) - if processor_class is None: - return None - return processor_class(extension_request) - - -# vi:sts=4 sw=4 et diff --git a/module/lib/mod_pywebsocket/handshake/__init__.py b/module/lib/mod_pywebsocket/handshake/__init__.py deleted file mode 100644 index 194f6b395..000000000 --- a/module/lib/mod_pywebsocket/handshake/__init__.py +++ /dev/null @@ -1,110 +0,0 @@ -# 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. - - -"""WebSocket opening handshake processor. This class try to apply available -opening handshake processors for each protocol version until a connection is -successfully established. -""" - - -import logging - -from mod_pywebsocket import common -from mod_pywebsocket.handshake import hybi00 -from mod_pywebsocket.handshake import hybi -# Export AbortedByUserException, HandshakeException, and VersionException -# symbol from this module. -from mod_pywebsocket.handshake._base import AbortedByUserException -from mod_pywebsocket.handshake._base import HandshakeException -from mod_pywebsocket.handshake._base import VersionException - - -_LOGGER = logging.getLogger(__name__) - - -def do_handshake(request, dispatcher, allowDraft75=False, strict=False): - """Performs WebSocket handshake. - - Args: - request: mod_python request. - dispatcher: Dispatcher (dispatch.Dispatcher). - allowDraft75: obsolete argument. ignored. - strict: obsolete argument. ignored. - - Handshaker will add attributes such as ws_resource in performing - handshake. - """ - - _LOGGER.debug('Client\'s opening handshake resource: %r', request.uri) - # To print mimetools.Message as escaped one-line string, we converts - # headers_in to dict object. Without conversion, if we use %r, it just - # prints the type and address, and if we use %s, it prints the original - # header string as multiple lines. - # - # Both mimetools.Message and MpTable_Type of mod_python can be - # converted to dict. - # - # mimetools.Message.__str__ returns the original header string. - # dict(mimetools.Message object) returns the map from header names to - # header values. While MpTable_Type doesn't have such __str__ but just - # __repr__ which formats itself as well as dictionary object. - _LOGGER.debug( - 'Client\'s opening handshake headers: %r', dict(request.headers_in)) - - handshakers = [] - handshakers.append( - ('RFC 6455', hybi.Handshaker(request, dispatcher))) - handshakers.append( - ('HyBi 00', hybi00.Handshaker(request, dispatcher))) - - for name, handshaker in handshakers: - _LOGGER.debug('Trying protocol version %s', name) - try: - handshaker.do_handshake() - _LOGGER.info('Established (%s protocol)', name) - return - except HandshakeException, e: - _LOGGER.debug( - 'Failed to complete opening handshake as %s protocol: %r', - name, e) - if e.status: - raise e - except AbortedByUserException, e: - raise - except VersionException, e: - raise - - # TODO(toyoshim): Add a test to cover the case all handshakers fail. - raise HandshakeException( - 'Failed to complete opening handshake for all available protocols', - status=common.HTTP_STATUS_BAD_REQUEST) - - -# vi:sts=4 sw=4 et diff --git a/module/lib/mod_pywebsocket/handshake/_base.py b/module/lib/mod_pywebsocket/handshake/_base.py deleted file mode 100644 index e5c94ca90..000000000 --- a/module/lib/mod_pywebsocket/handshake/_base.py +++ /dev/null @@ -1,226 +0,0 @@ -# Copyright 2012, 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. - - -"""Common functions and exceptions used by WebSocket opening handshake -processors. -""" - - -from mod_pywebsocket import common -from mod_pywebsocket import http_header_util - - -class AbortedByUserException(Exception): - """Exception for aborting a connection intentionally. - - If this exception is raised in do_extra_handshake handler, the connection - will be abandoned. No other WebSocket or HTTP(S) handler will be invoked. - - If this exception is raised in transfer_data_handler, the connection will - be closed without closing handshake. No other WebSocket or HTTP(S) handler - will be invoked. - """ - - pass - - -class HandshakeException(Exception): - """This exception will be raised when an error occurred while processing - WebSocket initial handshake. - """ - - def __init__(self, name, status=None): - super(HandshakeException, self).__init__(name) - self.status = status - - -class VersionException(Exception): - """This exception will be raised when a version of client request does not - match with version the server supports. - """ - - def __init__(self, name, supported_versions=''): - """Construct an instance. - - Args: - supported_version: a str object to show supported hybi versions. - (e.g. '8, 13') - """ - super(VersionException, self).__init__(name) - self.supported_versions = supported_versions - - -def get_default_port(is_secure): - if is_secure: - return common.DEFAULT_WEB_SOCKET_SECURE_PORT - else: - return common.DEFAULT_WEB_SOCKET_PORT - - -def validate_subprotocol(subprotocol, hixie): - """Validate a value in the Sec-WebSocket-Protocol field. - - See - - RFC 6455: Section 4.1., 4.2.2., and 4.3. - - HyBi 00: Section 4.1. Opening handshake - - Args: - hixie: if True, checks if characters in subprotocol are in range - between U+0020 and U+007E. It's required by HyBi 00 but not by - RFC 6455. - """ - - if not subprotocol: - raise HandshakeException('Invalid subprotocol name: empty') - if hixie: - # Parameter should be in the range U+0020 to U+007E. - for c in subprotocol: - if not 0x20 <= ord(c) <= 0x7e: - raise HandshakeException( - 'Illegal character in subprotocol name: %r' % c) - else: - # Parameter should be encoded HTTP token. - state = http_header_util.ParsingState(subprotocol) - token = http_header_util.consume_token(state) - rest = http_header_util.peek(state) - # If |rest| is not None, |subprotocol| is not one token or invalid. If - # |rest| is None, |token| must not be None because |subprotocol| is - # concatenation of |token| and |rest| and is not None. - if rest is not None: - raise HandshakeException('Invalid non-token string in subprotocol ' - 'name: %r' % rest) - - -def parse_host_header(request): - fields = request.headers_in['Host'].split(':', 1) - if len(fields) == 1: - return fields[0], get_default_port(request.is_https()) - try: - return fields[0], int(fields[1]) - except ValueError, e: - raise HandshakeException('Invalid port number format: %r' % e) - - -def format_header(name, value): - return '%s: %s\r\n' % (name, value) - - -def build_location(request): - """Build WebSocket location for request.""" - location_parts = [] - if request.is_https(): - location_parts.append(common.WEB_SOCKET_SECURE_SCHEME) - else: - location_parts.append(common.WEB_SOCKET_SCHEME) - location_parts.append('://') - host, port = parse_host_header(request) - connection_port = request.connection.local_addr[1] - if port != connection_port: - raise HandshakeException('Header/connection port mismatch: %d/%d' % - (port, connection_port)) - location_parts.append(host) - if (port != get_default_port(request.is_https())): - location_parts.append(':') - location_parts.append(str(port)) - location_parts.append(request.uri) - return ''.join(location_parts) - - -def get_mandatory_header(request, key): - value = request.headers_in.get(key) - if value is None: - raise HandshakeException('Header %s is not defined' % key) - return value - - -def validate_mandatory_header(request, key, expected_value, fail_status=None): - value = get_mandatory_header(request, key) - - if value.lower() != expected_value.lower(): - raise HandshakeException( - 'Expected %r for header %s but found %r (case-insensitive)' % - (expected_value, key, value), status=fail_status) - - -def check_request_line(request): - # 5.1 1. The three character UTF-8 string "GET". - # 5.1 2. A UTF-8-encoded U+0020 SPACE character (0x20 byte). - if request.method != 'GET': - raise HandshakeException('Method is not GET: %r' % request.method) - - if request.protocol != 'HTTP/1.1': - raise HandshakeException('Version is not HTTP/1.1: %r' % - request.protocol) - - -def check_header_lines(request, mandatory_headers): - check_request_line(request) - - # The expected field names, and the meaning of their corresponding - # values, are as follows. - # |Upgrade| and |Connection| - for key, expected_value in mandatory_headers: - validate_mandatory_header(request, key, expected_value) - - -def parse_token_list(data): - """Parses a header value which follows 1#token and returns parsed elements - as a list of strings. - - Leading LWSes must be trimmed. - """ - - state = http_header_util.ParsingState(data) - - token_list = [] - - while True: - token = http_header_util.consume_token(state) - if token is not None: - token_list.append(token) - - http_header_util.consume_lwses(state) - - if http_header_util.peek(state) is None: - break - - if not http_header_util.consume_string(state, ','): - raise HandshakeException( - 'Expected a comma but found %r' % http_header_util.peek(state)) - - http_header_util.consume_lwses(state) - - if len(token_list) == 0: - raise HandshakeException('No valid token found') - - return token_list - - -# vi:sts=4 sw=4 et diff --git a/module/lib/mod_pywebsocket/handshake/hybi.py b/module/lib/mod_pywebsocket/handshake/hybi.py deleted file mode 100644 index fc0e2a096..000000000 --- a/module/lib/mod_pywebsocket/handshake/hybi.py +++ /dev/null @@ -1,404 +0,0 @@ -# Copyright 2012, 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 the opening handshake processor for the WebSocket -protocol (RFC 6455). - -Specification: -http://tools.ietf.org/html/rfc6455 -""" - - -# Note: request.connection.write is used in this module, even though mod_python -# document says that it should be used only in connection handlers. -# Unfortunately, we have no other options. For example, request.write is not -# suitable because it doesn't allow direct raw bytes writing. - - -import base64 -import logging -import os -import re - -from mod_pywebsocket import common -from mod_pywebsocket.extensions import get_extension_processor -from mod_pywebsocket.handshake._base import check_request_line -from mod_pywebsocket.handshake._base import format_header -from mod_pywebsocket.handshake._base import get_mandatory_header -from mod_pywebsocket.handshake._base import HandshakeException -from mod_pywebsocket.handshake._base import parse_token_list -from mod_pywebsocket.handshake._base import validate_mandatory_header -from mod_pywebsocket.handshake._base import validate_subprotocol -from mod_pywebsocket.handshake._base import VersionException -from mod_pywebsocket.stream import Stream -from mod_pywebsocket.stream import StreamOptions -from mod_pywebsocket import util - - -# Used to validate the value in the Sec-WebSocket-Key header strictly. RFC 4648 -# disallows non-zero padding, so the character right before == must be any of -# A, Q, g and w. -_SEC_WEBSOCKET_KEY_REGEX = re.compile('^[+/0-9A-Za-z]{21}[AQgw]==$') - -# Defining aliases for values used frequently. -_VERSION_HYBI08 = common.VERSION_HYBI08 -_VERSION_HYBI08_STRING = str(_VERSION_HYBI08) -_VERSION_LATEST = common.VERSION_HYBI_LATEST -_VERSION_LATEST_STRING = str(_VERSION_LATEST) -_SUPPORTED_VERSIONS = [ - _VERSION_LATEST, - _VERSION_HYBI08, -] - - -def compute_accept(key): - """Computes value for the Sec-WebSocket-Accept header from value of the - Sec-WebSocket-Key header. - """ - - accept_binary = util.sha1_hash( - key + common.WEBSOCKET_ACCEPT_UUID).digest() - accept = base64.b64encode(accept_binary) - - return (accept, accept_binary) - - -class Handshaker(object): - """Opening handshake processor for the WebSocket protocol (RFC 6455).""" - - def __init__(self, request, dispatcher): - """Construct an instance. - - Args: - request: mod_python request. - dispatcher: Dispatcher (dispatch.Dispatcher). - - Handshaker will add attributes such as ws_resource during handshake. - """ - - self._logger = util.get_class_logger(self) - - self._request = request - self._dispatcher = dispatcher - - def _validate_connection_header(self): - connection = get_mandatory_header( - self._request, common.CONNECTION_HEADER) - - try: - connection_tokens = parse_token_list(connection) - except HandshakeException, e: - raise HandshakeException( - 'Failed to parse %s: %s' % (common.CONNECTION_HEADER, e)) - - connection_is_valid = False - for token in connection_tokens: - if token.lower() == common.UPGRADE_CONNECTION_TYPE.lower(): - connection_is_valid = True - break - if not connection_is_valid: - raise HandshakeException( - '%s header doesn\'t contain "%s"' % - (common.CONNECTION_HEADER, common.UPGRADE_CONNECTION_TYPE)) - - def do_handshake(self): - self._request.ws_close_code = None - self._request.ws_close_reason = None - - # Parsing. - - check_request_line(self._request) - - validate_mandatory_header( - self._request, - common.UPGRADE_HEADER, - common.WEBSOCKET_UPGRADE_TYPE) - - self._validate_connection_header() - - self._request.ws_resource = self._request.uri - - unused_host = get_mandatory_header(self._request, common.HOST_HEADER) - - self._request.ws_version = self._check_version() - - # This handshake must be based on latest hybi. We are responsible to - # fallback to HTTP on handshake failure as latest hybi handshake - # specifies. - try: - self._get_origin() - self._set_protocol() - self._parse_extensions() - - # Key validation, response generation. - - key = self._get_key() - (accept, accept_binary) = compute_accept(key) - self._logger.debug( - '%s: %r (%s)', - common.SEC_WEBSOCKET_ACCEPT_HEADER, - accept, - util.hexify(accept_binary)) - - self._logger.debug('Protocol version is RFC 6455') - - # Setup extension processors. - - processors = [] - if self._request.ws_requested_extensions is not None: - for extension_request in self._request.ws_requested_extensions: - processor = get_extension_processor(extension_request) - # Unknown extension requests are just ignored. - if processor is not None: - processors.append(processor) - self._request.ws_extension_processors = processors - - # Extra handshake handler may modify/remove processors. - self._dispatcher.do_extra_handshake(self._request) - processors = filter(lambda processor: processor is not None, - self._request.ws_extension_processors) - - accepted_extensions = [] - - # We need to take care of mux extension here. Extensions that - # are placed before mux should be applied to logical channels. - mux_index = -1 - for i, processor in enumerate(processors): - if processor.name() == common.MUX_EXTENSION: - mux_index = i - break - if mux_index >= 0: - mux_processor = processors[mux_index] - logical_channel_processors = processors[:mux_index] - processors = processors[mux_index+1:] - - for processor in logical_channel_processors: - extension_response = processor.get_extension_response() - if extension_response is None: - # Rejected. - continue - accepted_extensions.append(extension_response) - # Pass a shallow copy of accepted_extensions as extensions for - # logical channels. - mux_response = mux_processor.get_extension_response( - self._request, accepted_extensions[:]) - if mux_response is not None: - accepted_extensions.append(mux_response) - - stream_options = StreamOptions() - - # When there is mux extension, here, |processors| contain only - # prosessors for extensions placed after mux. - for processor in processors: - - extension_response = processor.get_extension_response() - if extension_response is None: - # Rejected. - continue - - accepted_extensions.append(extension_response) - - processor.setup_stream_options(stream_options) - - if len(accepted_extensions) > 0: - self._request.ws_extensions = accepted_extensions - self._logger.debug( - 'Extensions accepted: %r', - map(common.ExtensionParameter.name, accepted_extensions)) - else: - self._request.ws_extensions = None - - self._request.ws_stream = self._create_stream(stream_options) - - if self._request.ws_requested_protocols is not None: - if self._request.ws_protocol is None: - raise HandshakeException( - 'do_extra_handshake must choose one subprotocol from ' - 'ws_requested_protocols and set it to ws_protocol') - validate_subprotocol(self._request.ws_protocol, hixie=False) - - self._logger.debug( - 'Subprotocol accepted: %r', - self._request.ws_protocol) - else: - if self._request.ws_protocol is not None: - raise HandshakeException( - 'ws_protocol must be None when the client didn\'t ' - 'request any subprotocol') - - self._send_handshake(accept) - except HandshakeException, e: - if not e.status: - # Fallback to 400 bad request by default. - e.status = common.HTTP_STATUS_BAD_REQUEST - raise e - - def _get_origin(self): - if self._request.ws_version is _VERSION_HYBI08: - origin_header = common.SEC_WEBSOCKET_ORIGIN_HEADER - else: - origin_header = common.ORIGIN_HEADER - origin = self._request.headers_in.get(origin_header) - if origin is None: - self._logger.debug('Client request does not have origin header') - self._request.ws_origin = origin - - def _check_version(self): - version = get_mandatory_header(self._request, - common.SEC_WEBSOCKET_VERSION_HEADER) - if version == _VERSION_HYBI08_STRING: - return _VERSION_HYBI08 - if version == _VERSION_LATEST_STRING: - return _VERSION_LATEST - - if version.find(',') >= 0: - raise HandshakeException( - 'Multiple versions (%r) are not allowed for header %s' % - (version, common.SEC_WEBSOCKET_VERSION_HEADER), - status=common.HTTP_STATUS_BAD_REQUEST) - raise VersionException( - 'Unsupported version %r for header %s' % - (version, common.SEC_WEBSOCKET_VERSION_HEADER), - supported_versions=', '.join(map(str, _SUPPORTED_VERSIONS))) - - def _set_protocol(self): - self._request.ws_protocol = None - - protocol_header = self._request.headers_in.get( - common.SEC_WEBSOCKET_PROTOCOL_HEADER) - - if protocol_header is None: - self._request.ws_requested_protocols = None - return - - self._request.ws_requested_protocols = parse_token_list( - protocol_header) - self._logger.debug('Subprotocols requested: %r', - self._request.ws_requested_protocols) - - def _parse_extensions(self): - extensions_header = self._request.headers_in.get( - common.SEC_WEBSOCKET_EXTENSIONS_HEADER) - if not extensions_header: - self._request.ws_requested_extensions = None - return - - if self._request.ws_version is common.VERSION_HYBI08: - allow_quoted_string=False - else: - allow_quoted_string=True - try: - self._request.ws_requested_extensions = common.parse_extensions( - extensions_header, allow_quoted_string=allow_quoted_string) - except common.ExtensionParsingException, e: - raise HandshakeException( - 'Failed to parse Sec-WebSocket-Extensions header: %r' % e) - - self._logger.debug( - 'Extensions requested: %r', - map(common.ExtensionParameter.name, - self._request.ws_requested_extensions)) - - def _validate_key(self, key): - if key.find(',') >= 0: - raise HandshakeException('Request has multiple %s header lines or ' - 'contains illegal character \',\': %r' % - (common.SEC_WEBSOCKET_KEY_HEADER, key)) - - # Validate - key_is_valid = False - try: - # Validate key by quick regex match before parsing by base64 - # module. Because base64 module skips invalid characters, we have - # to do this in advance to make this server strictly reject illegal - # keys. - if _SEC_WEBSOCKET_KEY_REGEX.match(key): - decoded_key = base64.b64decode(key) - if len(decoded_key) == 16: - key_is_valid = True - except TypeError, e: - pass - - if not key_is_valid: - raise HandshakeException( - 'Illegal value for header %s: %r' % - (common.SEC_WEBSOCKET_KEY_HEADER, key)) - - return decoded_key - - def _get_key(self): - key = get_mandatory_header( - self._request, common.SEC_WEBSOCKET_KEY_HEADER) - - decoded_key = self._validate_key(key) - - self._logger.debug( - '%s: %r (%s)', - common.SEC_WEBSOCKET_KEY_HEADER, - key, - util.hexify(decoded_key)) - - return key - - def _create_stream(self, stream_options): - return Stream(self._request, stream_options) - - def _create_handshake_response(self, accept): - response = [] - - response.append('HTTP/1.1 101 Switching Protocols\r\n') - - response.append(format_header( - common.UPGRADE_HEADER, common.WEBSOCKET_UPGRADE_TYPE)) - response.append(format_header( - common.CONNECTION_HEADER, common.UPGRADE_CONNECTION_TYPE)) - response.append(format_header( - common.SEC_WEBSOCKET_ACCEPT_HEADER, accept)) - if self._request.ws_protocol is not None: - response.append(format_header( - common.SEC_WEBSOCKET_PROTOCOL_HEADER, - self._request.ws_protocol)) - if (self._request.ws_extensions is not None and - len(self._request.ws_extensions) != 0): - response.append(format_header( - common.SEC_WEBSOCKET_EXTENSIONS_HEADER, - common.format_extensions(self._request.ws_extensions))) - response.append('\r\n') - - return ''.join(response) - - def _send_handshake(self, accept): - raw_response = self._create_handshake_response(accept) - self._request.connection.write(raw_response) - self._logger.debug('Sent server\'s opening handshake: %r', - raw_response) - - -# vi:sts=4 sw=4 et diff --git a/module/lib/mod_pywebsocket/handshake/hybi00.py b/module/lib/mod_pywebsocket/handshake/hybi00.py deleted file mode 100644 index cc6f8dc43..000000000 --- a/module/lib/mod_pywebsocket/handshake/hybi00.py +++ /dev/null @@ -1,242 +0,0 @@ -# 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 the opening handshake processor for the WebSocket -protocol version HyBi 00. - -Specification: -http://tools.ietf.org/html/draft-ietf-hybi-thewebsocketprotocol-00 -""" - - -# Note: request.connection.write/read are used in this module, even though -# mod_python document says that they should be used only in connection -# handlers. Unfortunately, we have no other options. For example, -# request.write/read are not suitable because they don't allow direct raw bytes -# writing/reading. - - -import logging -import re -import struct - -from mod_pywebsocket import common -from mod_pywebsocket.stream import StreamHixie75 -from mod_pywebsocket import util -from mod_pywebsocket.handshake._base import HandshakeException -from mod_pywebsocket.handshake._base import build_location -from mod_pywebsocket.handshake._base import check_header_lines -from mod_pywebsocket.handshake._base import format_header -from mod_pywebsocket.handshake._base import get_mandatory_header -from mod_pywebsocket.handshake._base import validate_subprotocol - - -_MANDATORY_HEADERS = [ - # key, expected value or None - [common.UPGRADE_HEADER, common.WEBSOCKET_UPGRADE_TYPE_HIXIE75], - [common.CONNECTION_HEADER, common.UPGRADE_CONNECTION_TYPE], -] - - -class Handshaker(object): - """Opening handshake processor for the WebSocket protocol version HyBi 00. - """ - - def __init__(self, request, dispatcher): - """Construct an instance. - - Args: - request: mod_python request. - dispatcher: Dispatcher (dispatch.Dispatcher). - - Handshaker will add attributes such as ws_resource in performing - handshake. - """ - - self._logger = util.get_class_logger(self) - - self._request = request - self._dispatcher = dispatcher - - def do_handshake(self): - """Perform WebSocket Handshake. - - On _request, we set - ws_resource, ws_protocol, ws_location, ws_origin, ws_challenge, - ws_challenge_md5: WebSocket handshake information. - ws_stream: Frame generation/parsing class. - ws_version: Protocol version. - - Raises: - HandshakeException: when any error happened in parsing the opening - handshake request. - """ - - # 5.1 Reading the client's opening handshake. - # dispatcher sets it in self._request. - check_header_lines(self._request, _MANDATORY_HEADERS) - self._set_resource() - self._set_subprotocol() - self._set_location() - self._set_origin() - self._set_challenge_response() - self._set_protocol_version() - - self._dispatcher.do_extra_handshake(self._request) - - self._send_handshake() - - def _set_resource(self): - self._request.ws_resource = self._request.uri - - def _set_subprotocol(self): - # |Sec-WebSocket-Protocol| - subprotocol = self._request.headers_in.get( - common.SEC_WEBSOCKET_PROTOCOL_HEADER) - if subprotocol is not None: - validate_subprotocol(subprotocol, hixie=True) - self._request.ws_protocol = subprotocol - - def _set_location(self): - # |Host| - host = self._request.headers_in.get(common.HOST_HEADER) - if host is not None: - self._request.ws_location = build_location(self._request) - # TODO(ukai): check host is this host. - - def _set_origin(self): - # |Origin| - origin = self._request.headers_in.get(common.ORIGIN_HEADER) - if origin is not None: - self._request.ws_origin = origin - - def _set_protocol_version(self): - # |Sec-WebSocket-Draft| - draft = self._request.headers_in.get(common.SEC_WEBSOCKET_DRAFT_HEADER) - if draft is not None and draft != '0': - raise HandshakeException('Illegal value for %s: %s' % - (common.SEC_WEBSOCKET_DRAFT_HEADER, - draft)) - - self._logger.debug('Protocol version is HyBi 00') - self._request.ws_version = common.VERSION_HYBI00 - self._request.ws_stream = StreamHixie75(self._request, True) - - def _set_challenge_response(self): - # 5.2 4-8. - self._request.ws_challenge = self._get_challenge() - # 5.2 9. let /response/ be the MD5 finterprint of /challenge/ - self._request.ws_challenge_md5 = util.md5_hash( - self._request.ws_challenge).digest() - self._logger.debug( - 'Challenge: %r (%s)', - self._request.ws_challenge, - util.hexify(self._request.ws_challenge)) - self._logger.debug( - 'Challenge response: %r (%s)', - self._request.ws_challenge_md5, - util.hexify(self._request.ws_challenge_md5)) - - def _get_key_value(self, key_field): - key_value = get_mandatory_header(self._request, key_field) - - self._logger.debug('%s: %r', key_field, key_value) - - # 5.2 4. let /key-number_n/ be the digits (characters in the range - # U+0030 DIGIT ZERO (0) to U+0039 DIGIT NINE (9)) in /key_n/, - # interpreted as a base ten integer, ignoring all other characters - # in /key_n/. - try: - key_number = int(re.sub("\\D", "", key_value)) - except: - raise HandshakeException('%s field contains no digit' % key_field) - # 5.2 5. let /spaces_n/ be the number of U+0020 SPACE characters - # in /key_n/. - spaces = re.subn(" ", "", key_value)[1] - if spaces == 0: - raise HandshakeException('%s field contains no space' % key_field) - - self._logger.debug( - '%s: Key-number is %d and number of spaces is %d', - key_field, key_number, spaces) - - # 5.2 6. if /key-number_n/ is not an integral multiple of /spaces_n/ - # then abort the WebSocket connection. - if key_number % spaces != 0: - raise HandshakeException( - '%s: Key-number (%d) is not an integral multiple of spaces ' - '(%d)' % (key_field, key_number, spaces)) - # 5.2 7. let /part_n/ be /key-number_n/ divided by /spaces_n/. - part = key_number / spaces - self._logger.debug('%s: Part is %d', key_field, part) - return part - - def _get_challenge(self): - # 5.2 4-7. - key1 = self._get_key_value(common.SEC_WEBSOCKET_KEY1_HEADER) - key2 = self._get_key_value(common.SEC_WEBSOCKET_KEY2_HEADER) - # 5.2 8. let /challenge/ be the concatenation of /part_1/, - challenge = '' - challenge += struct.pack('!I', key1) # network byteorder int - challenge += struct.pack('!I', key2) # network byteorder int - challenge += self._request.connection.read(8) - return challenge - - def _send_handshake(self): - response = [] - - # 5.2 10. send the following line. - response.append('HTTP/1.1 101 WebSocket Protocol Handshake\r\n') - - # 5.2 11. send the following fields to the client. - response.append(format_header( - common.UPGRADE_HEADER, common.WEBSOCKET_UPGRADE_TYPE_HIXIE75)) - response.append(format_header( - common.CONNECTION_HEADER, common.UPGRADE_CONNECTION_TYPE)) - response.append(format_header( - common.SEC_WEBSOCKET_LOCATION_HEADER, self._request.ws_location)) - response.append(format_header( - common.SEC_WEBSOCKET_ORIGIN_HEADER, self._request.ws_origin)) - if self._request.ws_protocol: - response.append(format_header( - common.SEC_WEBSOCKET_PROTOCOL_HEADER, - self._request.ws_protocol)) - # 5.2 12. send two bytes 0x0D 0x0A. - response.append('\r\n') - # 5.2 13. send /response/ - response.append(self._request.ws_challenge_md5) - - raw_response = ''.join(response) - self._request.connection.write(raw_response) - self._logger.debug('Sent server\'s opening handshake: %r', - raw_response) - - -# vi:sts=4 sw=4 et diff --git a/module/lib/mod_pywebsocket/headerparserhandler.py b/module/lib/mod_pywebsocket/headerparserhandler.py deleted file mode 100644 index 2cc62de04..000000000 --- a/module/lib/mod_pywebsocket/headerparserhandler.py +++ /dev/null @@ -1,244 +0,0 @@ -# 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. - - -"""PythonHeaderParserHandler for mod_pywebsocket. - -Apache HTTP Server and mod_python must be configured such that this -function is called to handle WebSocket request. -""" - - -import logging - -from mod_python import apache - -from mod_pywebsocket import common -from mod_pywebsocket import dispatch -from mod_pywebsocket import handshake -from mod_pywebsocket import util - - -# PythonOption to specify the handler root directory. -_PYOPT_HANDLER_ROOT = 'mod_pywebsocket.handler_root' - -# PythonOption to specify the handler scan directory. -# This must be a directory under the root directory. -# The default is the root directory. -_PYOPT_HANDLER_SCAN = 'mod_pywebsocket.handler_scan' - -# PythonOption to allow handlers whose canonical path is -# not under the root directory. It's disallowed by default. -# Set this option with value of 'yes' to allow. -_PYOPT_ALLOW_HANDLERS_OUTSIDE_ROOT = ( - 'mod_pywebsocket.allow_handlers_outside_root_dir') -# Map from values to their meanings. 'Yes' and 'No' are allowed just for -# compatibility. -_PYOPT_ALLOW_HANDLERS_OUTSIDE_ROOT_DEFINITION = { - 'off': False, 'no': False, 'on': True, 'yes': True} - -# (Obsolete option. Ignored.) -# PythonOption to specify to allow handshake defined in Hixie 75 version -# protocol. The default is None (Off) -_PYOPT_ALLOW_DRAFT75 = 'mod_pywebsocket.allow_draft75' -# Map from values to their meanings. -_PYOPT_ALLOW_DRAFT75_DEFINITION = {'off': False, 'on': True} - - -class ApacheLogHandler(logging.Handler): - """Wrapper logging.Handler to emit log message to apache's error.log.""" - - _LEVELS = { - logging.DEBUG: apache.APLOG_DEBUG, - logging.INFO: apache.APLOG_INFO, - logging.WARNING: apache.APLOG_WARNING, - logging.ERROR: apache.APLOG_ERR, - logging.CRITICAL: apache.APLOG_CRIT, - } - - def __init__(self, request=None): - logging.Handler.__init__(self) - self._log_error = apache.log_error - if request is not None: - self._log_error = request.log_error - - # Time and level will be printed by Apache. - self._formatter = logging.Formatter('%(name)s: %(message)s') - - def emit(self, record): - apache_level = apache.APLOG_DEBUG - if record.levelno in ApacheLogHandler._LEVELS: - apache_level = ApacheLogHandler._LEVELS[record.levelno] - - msg = self._formatter.format(record) - - # "server" parameter must be passed to have "level" parameter work. - # If only "level" parameter is passed, nothing shows up on Apache's - # log. However, at this point, we cannot get the server object of the - # virtual host which will process WebSocket requests. The only server - # object we can get here is apache.main_server. But Wherever (server - # configuration context or virtual host context) we put - # PythonHeaderParserHandler directive, apache.main_server just points - # the main server instance (not any of virtual server instance). Then, - # Apache follows LogLevel directive in the server configuration context - # to filter logs. So, we need to specify LogLevel in the server - # configuration context. Even if we specify "LogLevel debug" in the - # virtual host context which actually handles WebSocket connections, - # DEBUG level logs never show up unless "LogLevel debug" is specified - # in the server configuration context. - # - # TODO(tyoshino): Provide logging methods on request object. When - # request is mp_request object (when used together with Apache), the - # methods call request.log_error indirectly. When request is - # _StandaloneRequest, the methods call Python's logging facility which - # we create in standalone.py. - self._log_error(msg, apache_level, apache.main_server) - - -def _configure_logging(): - logger = logging.getLogger() - # Logs are filtered by Apache based on LogLevel directive in Apache - # configuration file. We must just pass logs for all levels to - # ApacheLogHandler. - logger.setLevel(logging.DEBUG) - logger.addHandler(ApacheLogHandler()) - - -_configure_logging() - -_LOGGER = logging.getLogger(__name__) - - -def _parse_option(name, value, definition): - if value is None: - return False - - meaning = definition.get(value.lower()) - if meaning is None: - raise Exception('Invalid value for PythonOption %s: %r' % - (name, value)) - return meaning - - -def _create_dispatcher(): - _LOGGER.info('Initializing Dispatcher') - - options = apache.main_server.get_options() - - handler_root = options.get(_PYOPT_HANDLER_ROOT, None) - if not handler_root: - raise Exception('PythonOption %s is not defined' % _PYOPT_HANDLER_ROOT, - apache.APLOG_ERR) - - handler_scan = options.get(_PYOPT_HANDLER_SCAN, handler_root) - - allow_handlers_outside_root = _parse_option( - _PYOPT_ALLOW_HANDLERS_OUTSIDE_ROOT, - options.get(_PYOPT_ALLOW_HANDLERS_OUTSIDE_ROOT), - _PYOPT_ALLOW_HANDLERS_OUTSIDE_ROOT_DEFINITION) - - dispatcher = dispatch.Dispatcher( - handler_root, handler_scan, allow_handlers_outside_root) - - for warning in dispatcher.source_warnings(): - apache.log_error('mod_pywebsocket: %s' % warning, apache.APLOG_WARNING) - - return dispatcher - - -# Initialize -_dispatcher = _create_dispatcher() - - -def headerparserhandler(request): - """Handle request. - - Args: - request: mod_python request. - - This function is named headerparserhandler because it is the default - name for a PythonHeaderParserHandler. - """ - - handshake_is_done = False - try: - # Fallback to default http handler for request paths for which - # we don't have request handlers. - if not _dispatcher.get_handler_suite(request.uri): - request.log_error('No handler for resource: %r' % request.uri, - apache.APLOG_INFO) - request.log_error('Fallback to Apache', apache.APLOG_INFO) - return apache.DECLINED - except dispatch.DispatchException, e: - request.log_error('mod_pywebsocket: %s' % e, apache.APLOG_INFO) - if not handshake_is_done: - return e.status - - try: - allow_draft75 = _parse_option( - _PYOPT_ALLOW_DRAFT75, - apache.main_server.get_options().get(_PYOPT_ALLOW_DRAFT75), - _PYOPT_ALLOW_DRAFT75_DEFINITION) - - try: - handshake.do_handshake( - request, _dispatcher, allowDraft75=allow_draft75) - except handshake.VersionException, e: - request.log_error('mod_pywebsocket: %s' % e, apache.APLOG_INFO) - request.err_headers_out.add(common.SEC_WEBSOCKET_VERSION_HEADER, - e.supported_versions) - return apache.HTTP_BAD_REQUEST - except handshake.HandshakeException, e: - # Handshake for ws/wss failed. - # Send http response with error status. - request.log_error('mod_pywebsocket: %s' % e, apache.APLOG_INFO) - return e.status - - handshake_is_done = True - request._dispatcher = _dispatcher - _dispatcher.transfer_data(request) - except handshake.AbortedByUserException, e: - request.log_error('mod_pywebsocket: %s' % e, apache.APLOG_INFO) - except Exception, e: - # DispatchException can also be thrown if something is wrong in - # pywebsocket code. It's caught here, then. - - request.log_error('mod_pywebsocket: %s\n%s' % - (e, util.get_stack_trace()), - apache.APLOG_ERR) - # Unknown exceptions before handshake mean Apache must handle its - # request with another handler. - if not handshake_is_done: - return apache.DECLINED - # Set assbackwards to suppress response header generation by Apache. - request.assbackwards = 1 - return apache.DONE # Return DONE such that no other handlers are invoked. - - -# vi:sts=4 sw=4 et diff --git a/module/lib/mod_pywebsocket/http_header_util.py b/module/lib/mod_pywebsocket/http_header_util.py deleted file mode 100644 index b77465393..000000000 --- a/module/lib/mod_pywebsocket/http_header_util.py +++ /dev/null @@ -1,263 +0,0 @@ -# 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. - - -"""Utilities for parsing and formatting headers that follow the grammar defined -in HTTP RFC http://www.ietf.org/rfc/rfc2616.txt. -""" - - -import urlparse - - -_SEPARATORS = '()<>@,;:\\"/[]?={} \t' - - -def _is_char(c): - """Returns true iff c is in CHAR as specified in HTTP RFC.""" - - return ord(c) <= 127 - - -def _is_ctl(c): - """Returns true iff c is in CTL as specified in HTTP RFC.""" - - return ord(c) <= 31 or ord(c) == 127 - - -class ParsingState(object): - - def __init__(self, data): - self.data = data - self.head = 0 - - -def peek(state, pos=0): - """Peeks the character at pos from the head of data.""" - - if state.head + pos >= len(state.data): - return None - - return state.data[state.head + pos] - - -def consume(state, amount=1): - """Consumes specified amount of bytes from the head and returns the - consumed bytes. If there's not enough bytes to consume, returns None. - """ - - if state.head + amount > len(state.data): - return None - - result = state.data[state.head:state.head + amount] - state.head = state.head + amount - return result - - -def consume_string(state, expected): - """Given a parsing state and a expected string, consumes the string from - the head. Returns True if consumed successfully. Otherwise, returns - False. - """ - - pos = 0 - - for c in expected: - if c != peek(state, pos): - return False - pos += 1 - - consume(state, pos) - return True - - -def consume_lws(state): - """Consumes a LWS from the head. Returns True if any LWS is consumed. - Otherwise, returns False. - - LWS = [CRLF] 1*( SP | HT ) - """ - - original_head = state.head - - consume_string(state, '\r\n') - - pos = 0 - - while True: - c = peek(state, pos) - if c == ' ' or c == '\t': - pos += 1 - else: - if pos == 0: - state.head = original_head - return False - else: - consume(state, pos) - return True - - -def consume_lwses(state): - """Consumes *LWS from the head.""" - - while consume_lws(state): - pass - - -def consume_token(state): - """Consumes a token from the head. Returns the token or None if no token - was found. - """ - - pos = 0 - - while True: - c = peek(state, pos) - if c is None or c in _SEPARATORS or _is_ctl(c) or not _is_char(c): - if pos == 0: - return None - - return consume(state, pos) - else: - pos += 1 - - -def consume_token_or_quoted_string(state): - """Consumes a token or a quoted-string, and returns the token or unquoted - string. If no token or quoted-string was found, returns None. - """ - - original_head = state.head - - if not consume_string(state, '"'): - return consume_token(state) - - result = [] - - expect_quoted_pair = False - - while True: - if not expect_quoted_pair and consume_lws(state): - result.append(' ') - continue - - c = consume(state) - if c is None: - # quoted-string is not enclosed with double quotation - state.head = original_head - return None - elif expect_quoted_pair: - expect_quoted_pair = False - if _is_char(c): - result.append(c) - else: - # Non CHAR character found in quoted-pair - state.head = original_head - return None - elif c == '\\': - expect_quoted_pair = True - elif c == '"': - return ''.join(result) - elif _is_ctl(c): - # Invalid character %r found in qdtext - state.head = original_head - return None - else: - result.append(c) - - -def quote_if_necessary(s): - """Quotes arbitrary string into quoted-string.""" - - quote = False - if s == '': - return '""' - - result = [] - for c in s: - if c == '"' or c in _SEPARATORS or _is_ctl(c) or not _is_char(c): - quote = True - - if c == '"' or _is_ctl(c): - result.append('\\' + c) - else: - result.append(c) - - if quote: - return '"' + ''.join(result) + '"' - else: - return ''.join(result) - - -def parse_uri(uri): - """Parse absolute URI then return host, port and resource.""" - - parsed = urlparse.urlsplit(uri) - if parsed.scheme != 'wss' and parsed.scheme != 'ws': - # |uri| must be a relative URI. - # TODO(toyoshim): Should validate |uri|. - return None, None, uri - - if parsed.hostname is None: - return None, None, None - - port = None - try: - port = parsed.port - except ValueError, e: - # port property cause ValueError on invalid null port description like - # 'ws://host:/path'. - return None, None, None - - if port is None: - if parsed.scheme == 'ws': - port = 80 - else: - port = 443 - - path = parsed.path - if not path: - path += '/' - if parsed.query: - path += '?' + parsed.query - if parsed.fragment: - path += '#' + parsed.fragment - - return parsed.hostname, port, path - - -try: - urlparse.uses_netloc.index('ws') -except ValueError, e: - # urlparse in Python2.5.1 doesn't have 'ws' and 'wss' entries. - urlparse.uses_netloc.append('ws') - urlparse.uses_netloc.append('wss') - - -# vi:sts=4 sw=4 et diff --git a/module/lib/mod_pywebsocket/memorizingfile.py b/module/lib/mod_pywebsocket/memorizingfile.py deleted file mode 100644 index 4d4cd9585..000000000 --- a/module/lib/mod_pywebsocket/memorizingfile.py +++ /dev/null @@ -1,99 +0,0 @@ -#!/usr/bin/env python -# -# 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. - - -"""Memorizing file. - -A memorizing file wraps a file and memorizes lines read by readline. -""" - - -import sys - - -class MemorizingFile(object): - """MemorizingFile wraps a file and memorizes lines read by readline. - - Note that data read by other methods are not memorized. This behavior - is good enough for memorizing lines SimpleHTTPServer reads before - the control reaches WebSocketRequestHandler. - """ - - def __init__(self, file_, max_memorized_lines=sys.maxint): - """Construct an instance. - - Args: - file_: the file object to wrap. - max_memorized_lines: the maximum number of lines to memorize. - Only the first max_memorized_lines are memorized. - Default: sys.maxint. - """ - - self._file = file_ - self._memorized_lines = [] - self._max_memorized_lines = max_memorized_lines - self._buffered = False - self._buffered_line = None - - def __getattribute__(self, name): - if name in ('_file', '_memorized_lines', '_max_memorized_lines', - '_buffered', '_buffered_line', 'readline', - 'get_memorized_lines'): - return object.__getattribute__(self, name) - return self._file.__getattribute__(name) - - def readline(self, size=-1): - """Override file.readline and memorize the line read. - - Note that even if size is specified and smaller than actual size, - the whole line will be read out from underlying file object by - subsequent readline calls. - """ - - if self._buffered: - line = self._buffered_line - self._buffered = False - else: - line = self._file.readline() - if line and len(self._memorized_lines) < self._max_memorized_lines: - self._memorized_lines.append(line) - if size >= 0 and size < len(line): - self._buffered = True - self._buffered_line = line[size:] - return line[:size] - return line - - def get_memorized_lines(self): - """Get lines memorized so far.""" - return self._memorized_lines - - -# vi:sts=4 sw=4 et diff --git a/module/lib/mod_pywebsocket/msgutil.py b/module/lib/mod_pywebsocket/msgutil.py deleted file mode 100644 index 4c1a0114b..000000000 --- a/module/lib/mod_pywebsocket/msgutil.py +++ /dev/null @@ -1,219 +0,0 @@ -# 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. - - -"""Message related utilities. - -Note: request.connection.write/read are used in this module, even though -mod_python document says that they should be used only in connection -handlers. Unfortunately, we have no other options. For example, -request.write/read are not suitable because they don't allow direct raw -bytes writing/reading. -""" - - -import Queue -import threading - - -# Export Exception symbols from msgutil for backward compatibility -from mod_pywebsocket._stream_base import ConnectionTerminatedException -from mod_pywebsocket._stream_base import InvalidFrameException -from mod_pywebsocket._stream_base import BadOperationException -from mod_pywebsocket._stream_base import UnsupportedFrameException - - -# An API for handler to send/receive WebSocket messages. -def close_connection(request): - """Close connection. - - Args: - request: mod_python request. - """ - request.ws_stream.close_connection() - - -def send_message(request, payload_data, end=True, binary=False): - """Send a message (or part of a message). - - Args: - request: mod_python request. - payload_data: unicode text or str binary to send. - end: True to terminate a message. - False to send payload_data as part of a message that is to be - terminated by next or later send_message call with end=True. - binary: send payload_data as binary frame(s). - Raises: - BadOperationException: when server already terminated. - """ - request.ws_stream.send_message(payload_data, end, binary) - - -def receive_message(request): - """Receive a WebSocket frame and return its payload as a text in - unicode or a binary in str. - - Args: - request: mod_python request. - Raises: - InvalidFrameException: when client send invalid frame. - UnsupportedFrameException: when client send unsupported frame e.g. some - of reserved bit is set but no extension can - recognize it. - InvalidUTF8Exception: when client send a text frame containing any - invalid UTF-8 string. - ConnectionTerminatedException: when the connection is closed - unexpectedly. - BadOperationException: when client already terminated. - """ - return request.ws_stream.receive_message() - - -def send_ping(request, body=''): - request.ws_stream.send_ping(body) - - -class MessageReceiver(threading.Thread): - """This class receives messages from the client. - - This class provides three ways to receive messages: blocking, - non-blocking, and via callback. Callback has the highest precedence. - - Note: This class should not be used with the standalone server for wss - because pyOpenSSL used by the server raises a fatal error if the socket - is accessed from multiple threads. - """ - - def __init__(self, request, onmessage=None): - """Construct an instance. - - Args: - request: mod_python request. - onmessage: a function to be called when a message is received. - May be None. If not None, the function is called on - another thread. In that case, MessageReceiver.receive - and MessageReceiver.receive_nowait are useless - because they will never return any messages. - """ - - threading.Thread.__init__(self) - self._request = request - self._queue = Queue.Queue() - self._onmessage = onmessage - self._stop_requested = False - self.setDaemon(True) - self.start() - - def run(self): - try: - while not self._stop_requested: - message = receive_message(self._request) - if self._onmessage: - self._onmessage(message) - else: - self._queue.put(message) - finally: - close_connection(self._request) - - def receive(self): - """ Receive a message from the channel, blocking. - - Returns: - message as a unicode string. - """ - return self._queue.get() - - def receive_nowait(self): - """ Receive a message from the channel, non-blocking. - - Returns: - message as a unicode string if available. None otherwise. - """ - try: - message = self._queue.get_nowait() - except Queue.Empty: - message = None - return message - - def stop(self): - """Request to stop this instance. - - The instance will be stopped after receiving the next message. - This method may not be very useful, but there is no clean way - in Python to forcefully stop a running thread. - """ - self._stop_requested = True - - -class MessageSender(threading.Thread): - """This class sends messages to the client. - - This class provides both synchronous and asynchronous ways to send - messages. - - Note: This class should not be used with the standalone server for wss - because pyOpenSSL used by the server raises a fatal error if the socket - is accessed from multiple threads. - """ - - def __init__(self, request): - """Construct an instance. - - Args: - request: mod_python request. - """ - threading.Thread.__init__(self) - self._request = request - self._queue = Queue.Queue() - self.setDaemon(True) - self.start() - - def run(self): - while True: - message, condition = self._queue.get() - condition.acquire() - send_message(self._request, message) - condition.notify() - condition.release() - - def send(self, message): - """Send a message, blocking.""" - - condition = threading.Condition() - condition.acquire() - self._queue.put((message, condition)) - condition.wait() - - def send_nowait(self, message): - """Send a message, non-blocking.""" - - self._queue.put((message, threading.Condition())) - - -# vi:sts=4 sw=4 et diff --git a/module/lib/mod_pywebsocket/mux.py b/module/lib/mod_pywebsocket/mux.py deleted file mode 100644 index f0bdd2461..000000000 --- a/module/lib/mod_pywebsocket/mux.py +++ /dev/null @@ -1,1636 +0,0 @@ -# Copyright 2012, 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 multiplexing extension. - -Specification: -http://tools.ietf.org/html/draft-ietf-hybi-websocket-multiplexing-06 -""" - - -import collections -import copy -import email -import email.parser -import logging -import math -import struct -import threading -import traceback - -from mod_pywebsocket import common -from mod_pywebsocket import handshake -from mod_pywebsocket import util -from mod_pywebsocket._stream_base import BadOperationException -from mod_pywebsocket._stream_base import ConnectionTerminatedException -from mod_pywebsocket._stream_hybi import Frame -from mod_pywebsocket._stream_hybi import Stream -from mod_pywebsocket._stream_hybi import StreamOptions -from mod_pywebsocket._stream_hybi import create_binary_frame -from mod_pywebsocket._stream_hybi import create_closing_handshake_body -from mod_pywebsocket._stream_hybi import create_header -from mod_pywebsocket._stream_hybi import create_length_header -from mod_pywebsocket._stream_hybi import parse_frame -from mod_pywebsocket.handshake import hybi - - -_CONTROL_CHANNEL_ID = 0 -_DEFAULT_CHANNEL_ID = 1 - -_MUX_OPCODE_ADD_CHANNEL_REQUEST = 0 -_MUX_OPCODE_ADD_CHANNEL_RESPONSE = 1 -_MUX_OPCODE_FLOW_CONTROL = 2 -_MUX_OPCODE_DROP_CHANNEL = 3 -_MUX_OPCODE_NEW_CHANNEL_SLOT = 4 - -_MAX_CHANNEL_ID = 2 ** 29 - 1 - -_INITIAL_NUMBER_OF_CHANNEL_SLOTS = 64 -_INITIAL_QUOTA_FOR_CLIENT = 8 * 1024 - -_HANDSHAKE_ENCODING_IDENTITY = 0 -_HANDSHAKE_ENCODING_DELTA = 1 - -# We need only these status code for now. -_HTTP_BAD_RESPONSE_MESSAGES = { - common.HTTP_STATUS_BAD_REQUEST: 'Bad Request', -} - -# DropChannel reason code -# TODO(bashi): Define all reason code defined in -05 draft. -_DROP_CODE_NORMAL_CLOSURE = 1000 - -_DROP_CODE_INVALID_ENCAPSULATING_MESSAGE = 2001 -_DROP_CODE_CHANNEL_ID_TRUNCATED = 2002 -_DROP_CODE_ENCAPSULATED_FRAME_IS_TRUNCATED = 2003 -_DROP_CODE_UNKNOWN_MUX_OPCODE = 2004 -_DROP_CODE_INVALID_MUX_CONTROL_BLOCK = 2005 -_DROP_CODE_CHANNEL_ALREADY_EXISTS = 2006 -_DROP_CODE_NEW_CHANNEL_SLOT_VIOLATION = 2007 - -_DROP_CODE_UNKNOWN_REQUEST_ENCODING = 3002 -_DROP_CODE_SEND_QUOTA_VIOLATION = 3005 -_DROP_CODE_ACKNOWLEDGED = 3008 - - -class MuxUnexpectedException(Exception): - """Exception in handling multiplexing extension.""" - pass - - -# Temporary -class MuxNotImplementedException(Exception): - """Raised when a flow enters unimplemented code path.""" - pass - - -class LogicalConnectionClosedException(Exception): - """Raised when logical connection is gracefully closed.""" - pass - - -class PhysicalConnectionError(Exception): - """Raised when there is a physical connection error.""" - def __init__(self, drop_code, message=''): - super(PhysicalConnectionError, self).__init__( - 'code=%d, message=%r' % (drop_code, message)) - self.drop_code = drop_code - self.message = message - - -class LogicalChannelError(Exception): - """Raised when there is a logical channel error.""" - def __init__(self, channel_id, drop_code, message=''): - super(LogicalChannelError, self).__init__( - 'channel_id=%d, code=%d, message=%r' % ( - channel_id, drop_code, message)) - self.channel_id = channel_id - self.drop_code = drop_code - self.message = message - - -def _encode_channel_id(channel_id): - if channel_id < 0: - raise ValueError('Channel id %d must not be negative' % channel_id) - - if channel_id < 2 ** 7: - return chr(channel_id) - if channel_id < 2 ** 14: - return struct.pack('!H', 0x8000 + channel_id) - if channel_id < 2 ** 21: - first = chr(0xc0 + (channel_id >> 16)) - return first + struct.pack('!H', channel_id & 0xffff) - if channel_id < 2 ** 29: - return struct.pack('!L', 0xe0000000 + channel_id) - - raise ValueError('Channel id %d is too large' % channel_id) - - -def _encode_number(number): - return create_length_header(number, False) - - -def _create_add_channel_response(channel_id, encoded_handshake, - encoding=0, rejected=False, - outer_frame_mask=False): - if encoding != 0 and encoding != 1: - raise ValueError('Invalid encoding %d' % encoding) - - first_byte = ((_MUX_OPCODE_ADD_CHANNEL_RESPONSE << 5) | - (rejected << 4) | encoding) - block = (chr(first_byte) + - _encode_channel_id(channel_id) + - _encode_number(len(encoded_handshake)) + - encoded_handshake) - payload = _encode_channel_id(_CONTROL_CHANNEL_ID) + block - return create_binary_frame(payload, mask=outer_frame_mask) - - -def _create_drop_channel(channel_id, code=None, message='', - outer_frame_mask=False): - if len(message) > 0 and code is None: - raise ValueError('Code must be specified if message is specified') - - first_byte = _MUX_OPCODE_DROP_CHANNEL << 5 - block = chr(first_byte) + _encode_channel_id(channel_id) - if code is None: - block += _encode_number(0) # Reason size - else: - reason = struct.pack('!H', code) + message - reason_size = _encode_number(len(reason)) - block += reason_size + reason - - payload = _encode_channel_id(_CONTROL_CHANNEL_ID) + block - return create_binary_frame(payload, mask=outer_frame_mask) - - -def _create_flow_control(channel_id, replenished_quota, - outer_frame_mask=False): - first_byte = _MUX_OPCODE_FLOW_CONTROL << 5 - block = (chr(first_byte) + - _encode_channel_id(channel_id) + - _encode_number(replenished_quota)) - payload = _encode_channel_id(_CONTROL_CHANNEL_ID) + block - return create_binary_frame(payload, mask=outer_frame_mask) - - -def _create_new_channel_slot(slots, send_quota, outer_frame_mask=False): - if slots < 0 or send_quota < 0: - raise ValueError('slots and send_quota must be non-negative.') - first_byte = _MUX_OPCODE_NEW_CHANNEL_SLOT << 5 - block = (chr(first_byte) + - _encode_number(slots) + - _encode_number(send_quota)) - payload = _encode_channel_id(_CONTROL_CHANNEL_ID) + block - return create_binary_frame(payload, mask=outer_frame_mask) - - -def _create_fallback_new_channel_slot(outer_frame_mask=False): - first_byte = (_MUX_OPCODE_NEW_CHANNEL_SLOT << 5) | 1 # Set the F flag - block = (chr(first_byte) + _encode_number(0) + _encode_number(0)) - payload = _encode_channel_id(_CONTROL_CHANNEL_ID) + block - return create_binary_frame(payload, mask=outer_frame_mask) - - -def _parse_request_text(request_text): - request_line, header_lines = request_text.split('\r\n', 1) - - words = request_line.split(' ') - if len(words) != 3: - raise ValueError('Bad Request-Line syntax %r' % request_line) - [command, path, version] = words - if version != 'HTTP/1.1': - raise ValueError('Bad request version %r' % version) - - # email.parser.Parser() parses RFC 2822 (RFC 822) style headers. - # RFC 6455 refers RFC 2616 for handshake parsing, and RFC 2616 refers - # RFC 822. - headers = email.parser.Parser().parsestr(header_lines) - return command, path, version, headers - - -class _ControlBlock(object): - """A structure that holds parsing result of multiplexing control block. - Control block specific attributes will be added by _MuxFramePayloadParser. - (e.g. encoded_handshake will be added for AddChannelRequest and - AddChannelResponse) - """ - - def __init__(self, opcode): - self.opcode = opcode - - -class _MuxFramePayloadParser(object): - """A class that parses multiplexed frame payload.""" - - def __init__(self, payload): - self._data = payload - self._read_position = 0 - self._logger = util.get_class_logger(self) - - def read_channel_id(self): - """Reads channel id. - - Raises: - ValueError: when the payload doesn't contain - valid channel id. - """ - - remaining_length = len(self._data) - self._read_position - pos = self._read_position - if remaining_length == 0: - raise ValueError('Invalid channel id format') - - channel_id = ord(self._data[pos]) - channel_id_length = 1 - if channel_id & 0xe0 == 0xe0: - if remaining_length < 4: - raise ValueError('Invalid channel id format') - channel_id = struct.unpack('!L', - self._data[pos:pos+4])[0] & 0x1fffffff - channel_id_length = 4 - elif channel_id & 0xc0 == 0xc0: - if remaining_length < 3: - raise ValueError('Invalid channel id format') - channel_id = (((channel_id & 0x1f) << 16) + - struct.unpack('!H', self._data[pos+1:pos+3])[0]) - channel_id_length = 3 - elif channel_id & 0x80 == 0x80: - if remaining_length < 2: - raise ValueError('Invalid channel id format') - channel_id = struct.unpack('!H', - self._data[pos:pos+2])[0] & 0x3fff - channel_id_length = 2 - self._read_position += channel_id_length - - return channel_id - - def read_inner_frame(self): - """Reads an inner frame. - - Raises: - PhysicalConnectionError: when the inner frame is invalid. - """ - - if len(self._data) == self._read_position: - raise PhysicalConnectionError( - _DROP_CODE_ENCAPSULATED_FRAME_IS_TRUNCATED) - - bits = ord(self._data[self._read_position]) - self._read_position += 1 - fin = (bits & 0x80) == 0x80 - rsv1 = (bits & 0x40) == 0x40 - rsv2 = (bits & 0x20) == 0x20 - rsv3 = (bits & 0x10) == 0x10 - opcode = bits & 0xf - payload = self.remaining_data() - # Consume rest of the message which is payload data of the original - # frame. - self._read_position = len(self._data) - return fin, rsv1, rsv2, rsv3, opcode, payload - - def _read_number(self): - if self._read_position + 1 > len(self._data): - raise PhysicalConnectionError( - _DROP_CODE_INVALID_MUX_CONTROL_BLOCK, - 'Cannot read the first byte of number field') - - number = ord(self._data[self._read_position]) - if number & 0x80 == 0x80: - raise PhysicalConnectionError( - _DROP_CODE_INVALID_MUX_CONTROL_BLOCK, - 'The most significant bit of the first byte of number should ' - 'be unset') - self._read_position += 1 - pos = self._read_position - if number == 127: - if pos + 8 > len(self._data): - raise PhysicalConnectionError( - _DROP_CODE_INVALID_MUX_CONTROL_BLOCK, - 'Invalid number field') - self._read_position += 8 - number = struct.unpack('!Q', self._data[pos:pos+8])[0] - if number > 0x7FFFFFFFFFFFFFFF: - raise PhysicalConnectionError( - _DROP_CODE_INVALID_MUX_CONTROL_BLOCK, - 'Encoded number >= 2^63') - if number <= 0xFFFF: - raise PhysicalConnectionError( - _DROP_CODE_INVALID_MUX_CONTROL_BLOCK, - '%d should not be encoded by 9 bytes encoding' % number) - return number - if number == 126: - if pos + 2 > len(self._data): - raise PhysicalConnectionError( - _DROP_CODE_INVALID_MUX_CONTROL_BLOCK, - 'Invalid number field') - self._read_position += 2 - number = struct.unpack('!H', self._data[pos:pos+2])[0] - if number <= 125: - raise PhysicalConnectionError( - _DROP_CODE_INVALID_MUX_CONTROL_BLOCK, - '%d should not be encoded by 3 bytes encoding' % number) - return number - - def _read_size_and_contents(self): - """Reads data that consists of followings: - - the size of the contents encoded the same way as payload length - of the WebSocket Protocol with 1 bit padding at the head. - - the contents. - """ - - size = self._read_number() - pos = self._read_position - if pos + size > len(self._data): - raise PhysicalConnectionError( - _DROP_CODE_INVALID_MUX_CONTROL_BLOCK, - 'Cannot read %d bytes data' % size) - - self._read_position += size - return self._data[pos:pos+size] - - def _read_add_channel_request(self, first_byte, control_block): - reserved = (first_byte >> 2) & 0x7 - if reserved != 0: - raise PhysicalConnectionError( - _DROP_CODE_INVALID_MUX_CONTROL_BLOCK, - 'Reserved bits must be unset') - - # Invalid encoding will be handled by MuxHandler. - encoding = first_byte & 0x3 - try: - control_block.channel_id = self.read_channel_id() - except ValueError, e: - raise PhysicalConnectionError(_DROP_CODE_INVALID_MUX_CONTROL_BLOCK) - control_block.encoding = encoding - encoded_handshake = self._read_size_and_contents() - control_block.encoded_handshake = encoded_handshake - return control_block - - def _read_add_channel_response(self, first_byte, control_block): - reserved = (first_byte >> 2) & 0x3 - if reserved != 0: - raise PhysicalConnectionError( - _DROP_CODE_INVALID_MUX_CONTROL_BLOCK, - 'Reserved bits must be unset') - - control_block.accepted = (first_byte >> 4) & 1 - control_block.encoding = first_byte & 0x3 - try: - control_block.channel_id = self.read_channel_id() - except ValueError, e: - raise PhysicalConnectionError(_DROP_CODE_INVALID_MUX_CONTROL_BLOCK) - control_block.encoded_handshake = self._read_size_and_contents() - return control_block - - def _read_flow_control(self, first_byte, control_block): - reserved = first_byte & 0x1f - if reserved != 0: - raise PhysicalConnectionError( - _DROP_CODE_INVALID_MUX_CONTROL_BLOCK, - 'Reserved bits must be unset') - - try: - control_block.channel_id = self.read_channel_id() - except ValueError, e: - raise PhysicalConnectionError(_DROP_CODE_INVALID_MUX_CONTROL_BLOCK) - control_block.send_quota = self._read_number() - return control_block - - def _read_drop_channel(self, first_byte, control_block): - reserved = first_byte & 0x1f - if reserved != 0: - raise PhysicalConnectionError( - _DROP_CODE_INVALID_MUX_CONTROL_BLOCK, - 'Reserved bits must be unset') - - try: - control_block.channel_id = self.read_channel_id() - except ValueError, e: - raise PhysicalConnectionError(_DROP_CODE_INVALID_MUX_CONTROL_BLOCK) - reason = self._read_size_and_contents() - if len(reason) == 0: - control_block.drop_code = None - control_block.drop_message = '' - elif len(reason) >= 2: - control_block.drop_code = struct.unpack('!H', reason[:2])[0] - control_block.drop_message = reason[2:] - else: - raise PhysicalConnectionError( - _DROP_CODE_INVALID_MUX_CONTROL_BLOCK, - 'Received DropChannel that conains only 1-byte reason') - return control_block - - def _read_new_channel_slot(self, first_byte, control_block): - reserved = first_byte & 0x1e - if reserved != 0: - raise PhysicalConnectionError( - _DROP_CODE_INVALID_MUX_CONTROL_BLOCK, - 'Reserved bits must be unset') - control_block.fallback = first_byte & 1 - control_block.slots = self._read_number() - control_block.send_quota = self._read_number() - return control_block - - def read_control_blocks(self): - """Reads control block(s). - - Raises: - PhysicalConnectionError: when the payload contains invalid control - block(s). - StopIteration: when no control blocks left. - """ - - while self._read_position < len(self._data): - first_byte = ord(self._data[self._read_position]) - self._read_position += 1 - opcode = (first_byte >> 5) & 0x7 - control_block = _ControlBlock(opcode=opcode) - if opcode == _MUX_OPCODE_ADD_CHANNEL_REQUEST: - yield self._read_add_channel_request(first_byte, control_block) - elif opcode == _MUX_OPCODE_ADD_CHANNEL_RESPONSE: - yield self._read_add_channel_response( - first_byte, control_block) - elif opcode == _MUX_OPCODE_FLOW_CONTROL: - yield self._read_flow_control(first_byte, control_block) - elif opcode == _MUX_OPCODE_DROP_CHANNEL: - yield self._read_drop_channel(first_byte, control_block) - elif opcode == _MUX_OPCODE_NEW_CHANNEL_SLOT: - yield self._read_new_channel_slot(first_byte, control_block) - else: - raise PhysicalConnectionError( - _DROP_CODE_UNKNOWN_MUX_OPCODE, - 'Invalid opcode %d' % opcode) - - assert self._read_position == len(self._data) - raise StopIteration - - def remaining_data(self): - """Returns remaining data.""" - - return self._data[self._read_position:] - - -class _LogicalRequest(object): - """Mimics mod_python request.""" - - def __init__(self, channel_id, command, path, protocol, headers, - connection): - """Constructs an instance. - - Args: - channel_id: the channel id of the logical channel. - command: HTTP request command. - path: HTTP request path. - headers: HTTP headers. - connection: _LogicalConnection instance. - """ - - self.channel_id = channel_id - self.method = command - self.uri = path - self.protocol = protocol - self.headers_in = headers - self.connection = connection - self.server_terminated = False - self.client_terminated = False - - def is_https(self): - """Mimics request.is_https(). Returns False because this method is - used only by old protocols (hixie and hybi00). - """ - - return False - - -class _LogicalConnection(object): - """Mimics mod_python mp_conn.""" - - # For details, see the comment of set_read_state(). - STATE_ACTIVE = 1 - STATE_GRACEFULLY_CLOSED = 2 - STATE_TERMINATED = 3 - - def __init__(self, mux_handler, channel_id): - """Constructs an instance. - - Args: - mux_handler: _MuxHandler instance. - channel_id: channel id of this connection. - """ - - self._mux_handler = mux_handler - self._channel_id = channel_id - self._incoming_data = '' - self._write_condition = threading.Condition() - self._waiting_write_completion = False - self._read_condition = threading.Condition() - self._read_state = self.STATE_ACTIVE - - def get_local_addr(self): - """Getter to mimic mp_conn.local_addr.""" - - return self._mux_handler.physical_connection.get_local_addr() - local_addr = property(get_local_addr) - - def get_remote_addr(self): - """Getter to mimic mp_conn.remote_addr.""" - - return self._mux_handler.physical_connection.get_remote_addr() - remote_addr = property(get_remote_addr) - - def get_memorized_lines(self): - """Gets memorized lines. Not supported.""" - - raise MuxUnexpectedException('_LogicalConnection does not support ' - 'get_memorized_lines') - - def write(self, data): - """Writes data. mux_handler sends data asynchronously. The caller will - be suspended until write done. - - Args: - data: data to be written. - - Raises: - MuxUnexpectedException: when called before finishing the previous - write. - """ - - try: - self._write_condition.acquire() - if self._waiting_write_completion: - raise MuxUnexpectedException( - 'Logical connection %d is already waiting the completion ' - 'of write' % self._channel_id) - - self._waiting_write_completion = True - self._mux_handler.send_data(self._channel_id, data) - self._write_condition.wait() - finally: - self._write_condition.release() - - def write_control_data(self, data): - """Writes data via the control channel. Don't wait finishing write - because this method can be called by mux dispatcher. - - Args: - data: data to be written. - """ - - self._mux_handler.send_control_data(data) - - def notify_write_done(self): - """Called when sending data is completed.""" - - try: - self._write_condition.acquire() - if not self._waiting_write_completion: - raise MuxUnexpectedException( - 'Invalid call of notify_write_done for logical connection' - ' %d' % self._channel_id) - self._waiting_write_completion = False - self._write_condition.notify() - finally: - self._write_condition.release() - - def append_frame_data(self, frame_data): - """Appends incoming frame data. Called when mux_handler dispatches - frame data to the corresponding application. - - Args: - frame_data: incoming frame data. - """ - - self._read_condition.acquire() - self._incoming_data += frame_data - self._read_condition.notify() - self._read_condition.release() - - def read(self, length): - """Reads data. Blocks until enough data has arrived via physical - connection. - - Args: - length: length of data to be read. - Raises: - LogicalConnectionClosedException: when closing handshake for this - logical channel has been received. - ConnectionTerminatedException: when the physical connection has - closed, or an error is caused on the reader thread. - """ - - self._read_condition.acquire() - while (self._read_state == self.STATE_ACTIVE and - len(self._incoming_data) < length): - self._read_condition.wait() - - try: - if self._read_state == self.STATE_GRACEFULLY_CLOSED: - raise LogicalConnectionClosedException( - 'Logical channel %d has closed.' % self._channel_id) - elif self._read_state == self.STATE_TERMINATED: - raise ConnectionTerminatedException( - 'Receiving %d byte failed. Logical channel (%d) closed' % - (length, self._channel_id)) - - value = self._incoming_data[:length] - self._incoming_data = self._incoming_data[length:] - finally: - self._read_condition.release() - - return value - - def set_read_state(self, new_state): - """Sets the state of this connection. Called when an event for this - connection has occurred. - - Args: - new_state: state to be set. new_state must be one of followings: - - STATE_GRACEFULLY_CLOSED: when closing handshake for this - connection has been received. - - STATE_TERMINATED: when the physical connection has closed or - DropChannel of this connection has received. - """ - - self._read_condition.acquire() - self._read_state = new_state - self._read_condition.notify() - self._read_condition.release() - - -class _LogicalStream(Stream): - """Mimics the Stream class. This class interprets multiplexed WebSocket - frames. - """ - - def __init__(self, request, send_quota, receive_quota): - """Constructs an instance. - - Args: - request: _LogicalRequest instance. - send_quota: Initial send quota. - receive_quota: Initial receive quota. - """ - - # TODO(bashi): Support frame filters. - stream_options = StreamOptions() - # Physical stream is responsible for masking. - stream_options.unmask_receive = False - # Control frames can be fragmented on logical channel. - stream_options.allow_fragmented_control_frame = True - Stream.__init__(self, request, stream_options) - self._send_quota = send_quota - self._send_quota_condition = threading.Condition() - self._receive_quota = receive_quota - self._write_inner_frame_semaphore = threading.Semaphore() - - def _create_inner_frame(self, opcode, payload, end=True): - # TODO(bashi): Support extensions that use reserved bits. - first_byte = (end << 7) | opcode - return (_encode_channel_id(self._request.channel_id) + - chr(first_byte) + payload) - - def _write_inner_frame(self, opcode, payload, end=True): - payload_length = len(payload) - write_position = 0 - - try: - # An inner frame will be fragmented if there is no enough send - # quota. This semaphore ensures that fragmented inner frames are - # sent in order on the logical channel. - # Note that frames that come from other logical channels or - # multiplexing control blocks can be inserted between fragmented - # inner frames on the physical channel. - self._write_inner_frame_semaphore.acquire() - while write_position < payload_length: - try: - self._send_quota_condition.acquire() - while self._send_quota == 0: - self._logger.debug( - 'No quota. Waiting FlowControl message for %d.' % - self._request.channel_id) - self._send_quota_condition.wait() - - remaining = payload_length - write_position - write_length = min(self._send_quota, remaining) - inner_frame_end = ( - end and - (write_position + write_length == payload_length)) - - inner_frame = self._create_inner_frame( - opcode, - payload[write_position:write_position+write_length], - inner_frame_end) - frame_data = self._writer.build( - inner_frame, end=True, binary=True) - self._send_quota -= write_length - self._logger.debug('Consumed quota=%d, remaining=%d' % - (write_length, self._send_quota)) - finally: - self._send_quota_condition.release() - - # Writing data will block the worker so we need to release - # _send_quota_condition before writing. - self._logger.debug('Sending inner frame: %r' % frame_data) - self._request.connection.write(frame_data) - write_position += write_length - - opcode = common.OPCODE_CONTINUATION - - except ValueError, e: - raise BadOperationException(e) - finally: - self._write_inner_frame_semaphore.release() - - def replenish_send_quota(self, send_quota): - """Replenish send quota.""" - - self._send_quota_condition.acquire() - self._send_quota += send_quota - self._logger.debug('Replenished send quota for channel id %d: %d' % - (self._request.channel_id, self._send_quota)) - self._send_quota_condition.notify() - self._send_quota_condition.release() - - def consume_receive_quota(self, amount): - """Consumes receive quota. Returns False on failure.""" - - if self._receive_quota < amount: - self._logger.debug('Violate quota on channel id %d: %d < %d' % - (self._request.channel_id, - self._receive_quota, amount)) - return False - self._receive_quota -= amount - return True - - def send_message(self, message, end=True, binary=False): - """Override Stream.send_message.""" - - if self._request.server_terminated: - raise BadOperationException( - 'Requested send_message after sending out a closing handshake') - - if binary and isinstance(message, unicode): - raise BadOperationException( - 'Message for binary frame must be instance of str') - - if binary: - opcode = common.OPCODE_BINARY - else: - opcode = common.OPCODE_TEXT - message = message.encode('utf-8') - - self._write_inner_frame(opcode, message, end) - - def _receive_frame(self): - """Overrides Stream._receive_frame. - - In addition to call Stream._receive_frame, this method adds the amount - of payload to receiving quota and sends FlowControl to the client. - We need to do it here because Stream.receive_message() handles - control frames internally. - """ - - opcode, payload, fin, rsv1, rsv2, rsv3 = Stream._receive_frame(self) - amount = len(payload) - self._receive_quota += amount - frame_data = _create_flow_control(self._request.channel_id, - amount) - self._logger.debug('Sending flow control for %d, replenished=%d' % - (self._request.channel_id, amount)) - self._request.connection.write_control_data(frame_data) - return opcode, payload, fin, rsv1, rsv2, rsv3 - - def receive_message(self): - """Overrides Stream.receive_message.""" - - # Just call Stream.receive_message(), but catch - # LogicalConnectionClosedException, which is raised when the logical - # connection has closed gracefully. - try: - return Stream.receive_message(self) - except LogicalConnectionClosedException, e: - self._logger.debug('%s', e) - return None - - def _send_closing_handshake(self, code, reason): - """Overrides Stream._send_closing_handshake.""" - - body = create_closing_handshake_body(code, reason) - self._logger.debug('Sending closing handshake for %d: (%r, %r)' % - (self._request.channel_id, code, reason)) - self._write_inner_frame(common.OPCODE_CLOSE, body, end=True) - - self._request.server_terminated = True - - def send_ping(self, body=''): - """Overrides Stream.send_ping""" - - self._logger.debug('Sending ping on logical channel %d: %r' % - (self._request.channel_id, body)) - self._write_inner_frame(common.OPCODE_PING, body, end=True) - - self._ping_queue.append(body) - - def _send_pong(self, body): - """Overrides Stream._send_pong""" - - self._logger.debug('Sending pong on logical channel %d: %r' % - (self._request.channel_id, body)) - self._write_inner_frame(common.OPCODE_PONG, body, end=True) - - def close_connection(self, code=common.STATUS_NORMAL_CLOSURE, reason=''): - """Overrides Stream.close_connection.""" - - # TODO(bashi): Implement - self._logger.debug('Closing logical connection %d' % - self._request.channel_id) - self._request.server_terminated = True - - def _drain_received_data(self): - """Overrides Stream._drain_received_data. Nothing need to be done for - logical channel. - """ - - pass - - -class _OutgoingData(object): - """A structure that holds data to be sent via physical connection and - origin of the data. - """ - - def __init__(self, channel_id, data): - self.channel_id = channel_id - self.data = data - - -class _PhysicalConnectionWriter(threading.Thread): - """A thread that is responsible for writing data to physical connection. - - TODO(bashi): Make sure there is no thread-safety problem when the reader - thread reads data from the same socket at a time. - """ - - def __init__(self, mux_handler): - """Constructs an instance. - - Args: - mux_handler: _MuxHandler instance. - """ - - threading.Thread.__init__(self) - self._logger = util.get_class_logger(self) - self._mux_handler = mux_handler - self.setDaemon(True) - self._stop_requested = False - self._deque = collections.deque() - self._deque_condition = threading.Condition() - - def put_outgoing_data(self, data): - """Puts outgoing data. - - Args: - data: _OutgoingData instance. - - Raises: - BadOperationException: when the thread has been requested to - terminate. - """ - - try: - self._deque_condition.acquire() - if self._stop_requested: - raise BadOperationException('Cannot write data anymore') - - self._deque.append(data) - self._deque_condition.notify() - finally: - self._deque_condition.release() - - def _write_data(self, outgoing_data): - try: - self._mux_handler.physical_connection.write(outgoing_data.data) - except Exception, e: - util.prepend_message_to_exception( - 'Failed to send message to %r: ' % - (self._mux_handler.physical_connection.remote_addr,), e) - raise - - # TODO(bashi): It would be better to block the thread that sends - # control data as well. - if outgoing_data.channel_id != _CONTROL_CHANNEL_ID: - self._mux_handler.notify_write_done(outgoing_data.channel_id) - - def run(self): - self._deque_condition.acquire() - while not self._stop_requested: - if len(self._deque) == 0: - self._deque_condition.wait() - continue - - outgoing_data = self._deque.popleft() - self._deque_condition.release() - self._write_data(outgoing_data) - self._deque_condition.acquire() - - # Flush deque - try: - while len(self._deque) > 0: - outgoing_data = self._deque.popleft() - self._write_data(outgoing_data) - finally: - self._deque_condition.release() - - def stop(self): - """Stops the writer thread.""" - - self._deque_condition.acquire() - self._stop_requested = True - self._deque_condition.notify() - self._deque_condition.release() - - -class _PhysicalConnectionReader(threading.Thread): - """A thread that is responsible for reading data from physical connection. - """ - - def __init__(self, mux_handler): - """Constructs an instance. - - Args: - mux_handler: _MuxHandler instance. - """ - - threading.Thread.__init__(self) - self._logger = util.get_class_logger(self) - self._mux_handler = mux_handler - self.setDaemon(True) - - def run(self): - while True: - try: - physical_stream = self._mux_handler.physical_stream - message = physical_stream.receive_message() - if message is None: - break - # Below happens only when a data message is received. - opcode = physical_stream.get_last_received_opcode() - if opcode != common.OPCODE_BINARY: - self._mux_handler.fail_physical_connection( - _DROP_CODE_INVALID_ENCAPSULATING_MESSAGE, - 'Received a text message on physical connection') - break - - except ConnectionTerminatedException, e: - self._logger.debug('%s', e) - break - - try: - self._mux_handler.dispatch_message(message) - except PhysicalConnectionError, e: - self._mux_handler.fail_physical_connection( - e.drop_code, e.message) - break - except LogicalChannelError, e: - self._mux_handler.fail_logical_channel( - e.channel_id, e.drop_code, e.message) - except Exception, e: - self._logger.debug(traceback.format_exc()) - break - - self._mux_handler.notify_reader_done() - - -class _Worker(threading.Thread): - """A thread that is responsible for running the corresponding application - handler. - """ - - def __init__(self, mux_handler, request): - """Constructs an instance. - - Args: - mux_handler: _MuxHandler instance. - request: _LogicalRequest instance. - """ - - threading.Thread.__init__(self) - self._logger = util.get_class_logger(self) - self._mux_handler = mux_handler - self._request = request - self.setDaemon(True) - - def run(self): - self._logger.debug('Logical channel worker started. (id=%d)' % - self._request.channel_id) - try: - # Non-critical exceptions will be handled by dispatcher. - self._mux_handler.dispatcher.transfer_data(self._request) - finally: - self._mux_handler.notify_worker_done(self._request.channel_id) - - -class _MuxHandshaker(hybi.Handshaker): - """Opening handshake processor for multiplexing.""" - - _DUMMY_WEBSOCKET_KEY = 'dGhlIHNhbXBsZSBub25jZQ==' - - def __init__(self, request, dispatcher, send_quota, receive_quota): - """Constructs an instance. - Args: - request: _LogicalRequest instance. - dispatcher: Dispatcher instance (dispatch.Dispatcher). - send_quota: Initial send quota. - receive_quota: Initial receive quota. - """ - - hybi.Handshaker.__init__(self, request, dispatcher) - self._send_quota = send_quota - self._receive_quota = receive_quota - - # Append headers which should not be included in handshake field of - # AddChannelRequest. - # TODO(bashi): Make sure whether we should raise exception when - # these headers are included already. - request.headers_in[common.UPGRADE_HEADER] = ( - common.WEBSOCKET_UPGRADE_TYPE) - request.headers_in[common.CONNECTION_HEADER] = ( - common.UPGRADE_CONNECTION_TYPE) - request.headers_in[common.SEC_WEBSOCKET_VERSION_HEADER] = ( - str(common.VERSION_HYBI_LATEST)) - request.headers_in[common.SEC_WEBSOCKET_KEY_HEADER] = ( - self._DUMMY_WEBSOCKET_KEY) - - def _create_stream(self, stream_options): - """Override hybi.Handshaker._create_stream.""" - - self._logger.debug('Creating logical stream for %d' % - self._request.channel_id) - return _LogicalStream(self._request, self._send_quota, - self._receive_quota) - - def _create_handshake_response(self, accept): - """Override hybi._create_handshake_response.""" - - response = [] - - response.append('HTTP/1.1 101 Switching Protocols\r\n') - - # Upgrade, Connection and Sec-WebSocket-Accept should be excluded. - if self._request.ws_protocol is not None: - response.append('%s: %s\r\n' % ( - common.SEC_WEBSOCKET_PROTOCOL_HEADER, - self._request.ws_protocol)) - if (self._request.ws_extensions is not None and - len(self._request.ws_extensions) != 0): - response.append('%s: %s\r\n' % ( - common.SEC_WEBSOCKET_EXTENSIONS_HEADER, - common.format_extensions(self._request.ws_extensions))) - response.append('\r\n') - - return ''.join(response) - - def _send_handshake(self, accept): - """Override hybi.Handshaker._send_handshake.""" - - # Don't send handshake response for the default channel - if self._request.channel_id == _DEFAULT_CHANNEL_ID: - return - - handshake_response = self._create_handshake_response(accept) - frame_data = _create_add_channel_response( - self._request.channel_id, - handshake_response) - self._logger.debug('Sending handshake response for %d: %r' % - (self._request.channel_id, frame_data)) - self._request.connection.write_control_data(frame_data) - - -class _LogicalChannelData(object): - """A structure that holds information about logical channel. - """ - - def __init__(self, request, worker): - self.request = request - self.worker = worker - self.drop_code = _DROP_CODE_NORMAL_CLOSURE - self.drop_message = '' - - -class _HandshakeDeltaBase(object): - """A class that holds information for delta-encoded handshake.""" - - def __init__(self, headers): - self._headers = headers - - def create_headers(self, delta=None): - """Creates request headers for an AddChannelRequest that has - delta-encoded handshake. - - Args: - delta: headers should be overridden. - """ - - headers = copy.copy(self._headers) - if delta: - for key, value in delta.items(): - # The spec requires that a header with an empty value is - # removed from the delta base. - if len(value) == 0 and headers.has_key(key): - del headers[key] - else: - headers[key] = value - # TODO(bashi): Support extensions - headers['Sec-WebSocket-Extensions'] = '' - return headers - - -class _MuxHandler(object): - """Multiplexing handler. When a handler starts, it launches three - threads; the reader thread, the writer thread, and a worker thread. - - The reader thread reads data from the physical stream, i.e., the - ws_stream object of the underlying websocket connection. The reader - thread interprets multiplexed frames and dispatches them to logical - channels. Methods of this class are mostly called by the reader thread. - - The writer thread sends multiplexed frames which are created by - logical channels via the physical connection. - - The worker thread launched at the starting point handles the - "Implicitly Opened Connection". If multiplexing handler receives - an AddChannelRequest and accepts it, the handler will launch a new worker - thread and dispatch the request to it. - """ - - def __init__(self, request, dispatcher): - """Constructs an instance. - - Args: - request: mod_python request of the physical connection. - dispatcher: Dispatcher instance (dispatch.Dispatcher). - """ - - self.original_request = request - self.dispatcher = dispatcher - self.physical_connection = request.connection - self.physical_stream = request.ws_stream - self._logger = util.get_class_logger(self) - self._logical_channels = {} - self._logical_channels_condition = threading.Condition() - # Holds client's initial quota - self._channel_slots = collections.deque() - self._handshake_base = None - self._worker_done_notify_received = False - self._reader = None - self._writer = None - - def start(self): - """Starts the handler. - - Raises: - MuxUnexpectedException: when the handler already started, or when - opening handshake of the default channel fails. - """ - - if self._reader or self._writer: - raise MuxUnexpectedException('MuxHandler already started') - - self._reader = _PhysicalConnectionReader(self) - self._writer = _PhysicalConnectionWriter(self) - self._reader.start() - self._writer.start() - - # Create "Implicitly Opened Connection". - logical_connection = _LogicalConnection(self, _DEFAULT_CHANNEL_ID) - self._handshake_base = _HandshakeDeltaBase( - self.original_request.headers_in) - logical_request = _LogicalRequest( - _DEFAULT_CHANNEL_ID, - self.original_request.method, - self.original_request.uri, - self.original_request.protocol, - self._handshake_base.create_headers(), - logical_connection) - # Client's send quota for the implicitly opened connection is zero, - # but we will send FlowControl later so set the initial quota to - # _INITIAL_QUOTA_FOR_CLIENT. - self._channel_slots.append(_INITIAL_QUOTA_FOR_CLIENT) - if not self._do_handshake_for_logical_request( - logical_request, send_quota=self.original_request.mux_quota): - raise MuxUnexpectedException( - 'Failed handshake on the default channel id') - self._add_logical_channel(logical_request) - - # Send FlowControl for the implicitly opened connection. - frame_data = _create_flow_control(_DEFAULT_CHANNEL_ID, - _INITIAL_QUOTA_FOR_CLIENT) - logical_request.connection.write_control_data(frame_data) - - def add_channel_slots(self, slots, send_quota): - """Adds channel slots. - - Args: - slots: number of slots to be added. - send_quota: initial send quota for slots. - """ - - self._channel_slots.extend([send_quota] * slots) - # Send NewChannelSlot to client. - frame_data = _create_new_channel_slot(slots, send_quota) - self.send_control_data(frame_data) - - def wait_until_done(self, timeout=None): - """Waits until all workers are done. Returns False when timeout has - occurred. Returns True on success. - - Args: - timeout: timeout in sec. - """ - - self._logical_channels_condition.acquire() - try: - while len(self._logical_channels) > 0: - self._logger.debug('Waiting workers(%d)...' % - len(self._logical_channels)) - self._worker_done_notify_received = False - self._logical_channels_condition.wait(timeout) - if not self._worker_done_notify_received: - self._logger.debug('Waiting worker(s) timed out') - return False - - finally: - self._logical_channels_condition.release() - - # Flush pending outgoing data - self._writer.stop() - self._writer.join() - - return True - - def notify_write_done(self, channel_id): - """Called by the writer thread when a write operation has done. - - Args: - channel_id: objective channel id. - """ - - try: - self._logical_channels_condition.acquire() - if channel_id in self._logical_channels: - channel_data = self._logical_channels[channel_id] - channel_data.request.connection.notify_write_done() - else: - self._logger.debug('Seems that logical channel for %d has gone' - % channel_id) - finally: - self._logical_channels_condition.release() - - def send_control_data(self, data): - """Sends data via the control channel. - - Args: - data: data to be sent. - """ - - self._writer.put_outgoing_data(_OutgoingData( - channel_id=_CONTROL_CHANNEL_ID, data=data)) - - def send_data(self, channel_id, data): - """Sends data via given logical channel. This method is called by - worker threads. - - Args: - channel_id: objective channel id. - data: data to be sent. - """ - - self._writer.put_outgoing_data(_OutgoingData( - channel_id=channel_id, data=data)) - - def _send_drop_channel(self, channel_id, code=None, message=''): - frame_data = _create_drop_channel(channel_id, code, message) - self._logger.debug( - 'Sending drop channel for channel id %d' % channel_id) - self.send_control_data(frame_data) - - def _send_error_add_channel_response(self, channel_id, status=None): - if status is None: - status = common.HTTP_STATUS_BAD_REQUEST - - if status in _HTTP_BAD_RESPONSE_MESSAGES: - message = _HTTP_BAD_RESPONSE_MESSAGES[status] - else: - self._logger.debug('Response message for %d is not found' % status) - message = '???' - - response = 'HTTP/1.1 %d %s\r\n\r\n' % (status, message) - frame_data = _create_add_channel_response(channel_id, - encoded_handshake=response, - encoding=0, rejected=True) - self.send_control_data(frame_data) - - def _create_logical_request(self, block): - if block.channel_id == _CONTROL_CHANNEL_ID: - # TODO(bashi): Raise PhysicalConnectionError with code 2006 - # instead of MuxUnexpectedException. - raise MuxUnexpectedException( - 'Received the control channel id (0) as objective channel ' - 'id for AddChannel') - - if block.encoding > _HANDSHAKE_ENCODING_DELTA: - raise PhysicalConnectionError( - _DROP_CODE_UNKNOWN_REQUEST_ENCODING) - - method, path, version, headers = _parse_request_text( - block.encoded_handshake) - if block.encoding == _HANDSHAKE_ENCODING_DELTA: - headers = self._handshake_base.create_headers(headers) - - connection = _LogicalConnection(self, block.channel_id) - request = _LogicalRequest(block.channel_id, method, path, version, - headers, connection) - return request - - def _do_handshake_for_logical_request(self, request, send_quota=0): - try: - receive_quota = self._channel_slots.popleft() - except IndexError: - raise LogicalChannelError( - request.channel_id, _DROP_CODE_NEW_CHANNEL_SLOT_VIOLATION) - - handshaker = _MuxHandshaker(request, self.dispatcher, - send_quota, receive_quota) - try: - handshaker.do_handshake() - except handshake.VersionException, e: - self._logger.info('%s', e) - self._send_error_add_channel_response( - request.channel_id, status=common.HTTP_STATUS_BAD_REQUEST) - return False - except handshake.HandshakeException, e: - # TODO(bashi): Should we _Fail the Logical Channel_ with 3001 - # instead? - self._logger.info('%s', e) - self._send_error_add_channel_response(request.channel_id, - status=e.status) - return False - except handshake.AbortedByUserException, e: - self._logger.info('%s', e) - self._send_error_add_channel_response(request.channel_id) - return False - - return True - - def _add_logical_channel(self, logical_request): - try: - self._logical_channels_condition.acquire() - if logical_request.channel_id in self._logical_channels: - self._logger.debug('Channel id %d already exists' % - logical_request.channel_id) - raise PhysicalConnectionError( - _DROP_CODE_CHANNEL_ALREADY_EXISTS, - 'Channel id %d already exists' % - logical_request.channel_id) - worker = _Worker(self, logical_request) - channel_data = _LogicalChannelData(logical_request, worker) - self._logical_channels[logical_request.channel_id] = channel_data - worker.start() - finally: - self._logical_channels_condition.release() - - def _process_add_channel_request(self, block): - try: - logical_request = self._create_logical_request(block) - except ValueError, e: - self._logger.debug('Failed to create logical request: %r' % e) - self._send_error_add_channel_response( - block.channel_id, status=common.HTTP_STATUS_BAD_REQUEST) - return - if self._do_handshake_for_logical_request(logical_request): - if block.encoding == _HANDSHAKE_ENCODING_IDENTITY: - # Update handshake base. - # TODO(bashi): Make sure this is the right place to update - # handshake base. - self._handshake_base = _HandshakeDeltaBase( - logical_request.headers_in) - self._add_logical_channel(logical_request) - else: - self._send_error_add_channel_response( - block.channel_id, status=common.HTTP_STATUS_BAD_REQUEST) - - def _process_flow_control(self, block): - try: - self._logical_channels_condition.acquire() - if not block.channel_id in self._logical_channels: - return - channel_data = self._logical_channels[block.channel_id] - channel_data.request.ws_stream.replenish_send_quota( - block.send_quota) - finally: - self._logical_channels_condition.release() - - def _process_drop_channel(self, block): - self._logger.debug( - 'DropChannel received for %d: code=%r, reason=%r' % - (block.channel_id, block.drop_code, block.drop_message)) - try: - self._logical_channels_condition.acquire() - if not block.channel_id in self._logical_channels: - return - channel_data = self._logical_channels[block.channel_id] - channel_data.drop_code = _DROP_CODE_ACKNOWLEDGED - # Close the logical channel - channel_data.request.connection.set_read_state( - _LogicalConnection.STATE_TERMINATED) - finally: - self._logical_channels_condition.release() - - def _process_control_blocks(self, parser): - for control_block in parser.read_control_blocks(): - opcode = control_block.opcode - self._logger.debug('control block received, opcode: %d' % opcode) - if opcode == _MUX_OPCODE_ADD_CHANNEL_REQUEST: - self._process_add_channel_request(control_block) - elif opcode == _MUX_OPCODE_ADD_CHANNEL_RESPONSE: - raise PhysicalConnectionError( - _DROP_CODE_INVALID_MUX_CONTROL_BLOCK, - 'Received AddChannelResponse') - elif opcode == _MUX_OPCODE_FLOW_CONTROL: - self._process_flow_control(control_block) - elif opcode == _MUX_OPCODE_DROP_CHANNEL: - self._process_drop_channel(control_block) - elif opcode == _MUX_OPCODE_NEW_CHANNEL_SLOT: - raise PhysicalConnectionError( - _DROP_CODE_INVALID_MUX_CONTROL_BLOCK, - 'Received NewChannelSlot') - else: - raise MuxUnexpectedException( - 'Unexpected opcode %r' % opcode) - - def _process_logical_frame(self, channel_id, parser): - self._logger.debug('Received a frame. channel id=%d' % channel_id) - try: - self._logical_channels_condition.acquire() - if not channel_id in self._logical_channels: - # We must ignore the message for an inactive channel. - return - channel_data = self._logical_channels[channel_id] - fin, rsv1, rsv2, rsv3, opcode, payload = parser.read_inner_frame() - if not channel_data.request.ws_stream.consume_receive_quota( - len(payload)): - # The client violates quota. Close logical channel. - raise LogicalChannelError( - channel_id, _DROP_CODE_SEND_QUOTA_VIOLATION) - header = create_header(opcode, len(payload), fin, rsv1, rsv2, rsv3, - mask=False) - frame_data = header + payload - channel_data.request.connection.append_frame_data(frame_data) - finally: - self._logical_channels_condition.release() - - def dispatch_message(self, message): - """Dispatches message. The reader thread calls this method. - - Args: - message: a message that contains encapsulated frame. - Raises: - PhysicalConnectionError: if the message contains physical - connection level errors. - LogicalChannelError: if the message contains logical channel - level errors. - """ - - parser = _MuxFramePayloadParser(message) - try: - channel_id = parser.read_channel_id() - except ValueError, e: - raise PhysicalConnectionError(_DROP_CODE_CHANNEL_ID_TRUNCATED) - if channel_id == _CONTROL_CHANNEL_ID: - self._process_control_blocks(parser) - else: - self._process_logical_frame(channel_id, parser) - - def notify_worker_done(self, channel_id): - """Called when a worker has finished. - - Args: - channel_id: channel id corresponded with the worker. - """ - - self._logger.debug('Worker for channel id %d terminated' % channel_id) - try: - self._logical_channels_condition.acquire() - if not channel_id in self._logical_channels: - raise MuxUnexpectedException( - 'Channel id %d not found' % channel_id) - channel_data = self._logical_channels.pop(channel_id) - finally: - self._worker_done_notify_received = True - self._logical_channels_condition.notify() - self._logical_channels_condition.release() - - if not channel_data.request.server_terminated: - self._send_drop_channel( - channel_id, code=channel_data.drop_code, - message=channel_data.drop_message) - - def notify_reader_done(self): - """This method is called by the reader thread when the reader has - finished. - """ - - # Terminate all logical connections - self._logger.debug('termiating all logical connections...') - self._logical_channels_condition.acquire() - for channel_data in self._logical_channels.values(): - try: - channel_data.request.connection.set_read_state( - _LogicalConnection.STATE_TERMINATED) - except Exception: - pass - self._logical_channels_condition.release() - - def fail_physical_connection(self, code, message): - """Fail the physical connection. - - Args: - code: drop reason code. - message: drop message. - """ - - self._logger.debug('Failing the physical connection...') - self._send_drop_channel(_CONTROL_CHANNEL_ID, code, message) - self.physical_stream.close_connection( - common.STATUS_INTERNAL_ENDPOINT_ERROR) - - def fail_logical_channel(self, channel_id, code, message): - """Fail a logical channel. - - Args: - channel_id: channel id. - code: drop reason code. - message: drop message. - """ - - self._logger.debug('Failing logical channel %d...' % channel_id) - try: - self._logical_channels_condition.acquire() - if channel_id in self._logical_channels: - channel_data = self._logical_channels[channel_id] - # Close the logical channel. notify_worker_done() will be - # called later and it will send DropChannel. - channel_data.drop_code = code - channel_data.drop_message = message - channel_data.request.connection.set_read_state( - _LogicalConnection.STATE_TERMINATED) - else: - self._send_drop_channel(channel_id, code, message) - finally: - self._logical_channels_condition.release() - - -def use_mux(request): - return hasattr(request, 'mux') and request.mux - - -def start(request, dispatcher): - mux_handler = _MuxHandler(request, dispatcher) - mux_handler.start() - - mux_handler.add_channel_slots(_INITIAL_NUMBER_OF_CHANNEL_SLOTS, - _INITIAL_QUOTA_FOR_CLIENT) - - mux_handler.wait_until_done() - - -# vi:sts=4 sw=4 et diff --git a/module/lib/mod_pywebsocket/standalone.py b/module/lib/mod_pywebsocket/standalone.py deleted file mode 100755 index 07a33d9c9..000000000 --- a/module/lib/mod_pywebsocket/standalone.py +++ /dev/null @@ -1,998 +0,0 @@ -#!/usr/bin/env python -# -# Copyright 2012, 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. - - -"""Standalone WebSocket server. - -Use this file to launch pywebsocket without Apache HTTP Server. - - -BASIC USAGE - -Go to the src directory and run - - $ python mod_pywebsocket/standalone.py [-p <ws_port>] - [-w <websock_handlers>] - [-d <document_root>] - -<ws_port> is the port number to use for ws:// connection. - -<document_root> is the path to the root directory of HTML files. - -<websock_handlers> is the path to the root directory of WebSocket handlers. -If not specified, <document_root> will be used. See __init__.py (or -run $ pydoc mod_pywebsocket) for how to write WebSocket handlers. - -For more detail and other options, run - - $ python mod_pywebsocket/standalone.py --help - -or see _build_option_parser method below. - -For trouble shooting, adding "--log_level debug" might help you. - - -TRY DEMO - -Go to the src directory and run - - $ python standalone.py -d example - -to launch pywebsocket with the sample handler and html on port 80. Open -http://localhost/console.html, click the connect button, type something into -the text box next to the send button and click the send button. If everything -is working, you'll see the message you typed echoed by the server. - - -SUPPORTING TLS - -To support TLS, run standalone.py with -t, -k, and -c options. - - -SUPPORTING CLIENT AUTHENTICATION - -To support client authentication with TLS, run standalone.py with -t, -k, -c, -and --tls-client-auth, and --tls-client-ca options. - -E.g., $./standalone.py -d ../example -p 10443 -t -c ../test/cert/cert.pem -k -../test/cert/key.pem --tls-client-auth --tls-client-ca=../test/cert/cacert.pem - - -CONFIGURATION FILE - -You can also write a configuration file and use it by specifying the path to -the configuration file by --config option. Please write a configuration file -following the documentation of the Python ConfigParser library. Name of each -entry must be the long version argument name. E.g. to set log level to debug, -add the following line: - -log_level=debug - -For options which doesn't take value, please add some fake value. E.g. for ---tls option, add the following line: - -tls=True - -Note that tls will be enabled even if you write tls=False as the value part is -fake. - -When both a command line argument and a configuration file entry are set for -the same configuration item, the command line value will override one in the -configuration file. - - -THREADING - -This server is derived from SocketServer.ThreadingMixIn. Hence a thread is -used for each request. - - -SECURITY WARNING - -This uses CGIHTTPServer and CGIHTTPServer is not secure. -It may execute arbitrary Python code or external programs. It should not be -used outside a firewall. -""" - -import BaseHTTPServer -import CGIHTTPServer -import SimpleHTTPServer -import SocketServer -import ConfigParser -import base64 -import httplib -import logging -import logging.handlers -import optparse -import os -import re -import select -import socket -import sys -import threading -import time - -_HAS_SSL = False -_HAS_OPEN_SSL = False -try: - import ssl - _HAS_SSL = True -except ImportError: - try: - import OpenSSL.SSL - _HAS_OPEN_SSL = True - except ImportError: - pass - -from mod_pywebsocket import common -from mod_pywebsocket import dispatch -from mod_pywebsocket import handshake -from mod_pywebsocket import http_header_util -from mod_pywebsocket import memorizingfile -from mod_pywebsocket import util - - -_DEFAULT_LOG_MAX_BYTES = 1024 * 256 -_DEFAULT_LOG_BACKUP_COUNT = 5 - -_DEFAULT_REQUEST_QUEUE_SIZE = 128 - -# 1024 is practically large enough to contain WebSocket handshake lines. -_MAX_MEMORIZED_LINES = 1024 - - -class _StandaloneConnection(object): - """Mimic mod_python mp_conn.""" - - def __init__(self, request_handler): - """Construct an instance. - - Args: - request_handler: A WebSocketRequestHandler instance. - """ - - self._request_handler = request_handler - - def get_local_addr(self): - """Getter to mimic mp_conn.local_addr.""" - - return (self._request_handler.server.server_name, - self._request_handler.server.server_port) - local_addr = property(get_local_addr) - - def get_remote_addr(self): - """Getter to mimic mp_conn.remote_addr. - - Setting the property in __init__ won't work because the request - handler is not initialized yet there.""" - - return self._request_handler.client_address - remote_addr = property(get_remote_addr) - - def write(self, data): - """Mimic mp_conn.write().""" - - return self._request_handler.wfile.write(data) - - def read(self, length): - """Mimic mp_conn.read().""" - - return self._request_handler.rfile.read(length) - - def get_memorized_lines(self): - """Get memorized lines.""" - - return self._request_handler.rfile.get_memorized_lines() - - -class _StandaloneRequest(object): - """Mimic mod_python request.""" - - def __init__(self, request_handler, use_tls): - """Construct an instance. - - Args: - request_handler: A WebSocketRequestHandler instance. - """ - - self._logger = util.get_class_logger(self) - - self._request_handler = request_handler - self.connection = _StandaloneConnection(request_handler) - self._use_tls = use_tls - self.headers_in = request_handler.headers - - def get_uri(self): - """Getter to mimic request.uri.""" - - return self._request_handler.path - uri = property(get_uri) - - def get_method(self): - """Getter to mimic request.method.""" - - return self._request_handler.command - method = property(get_method) - - def get_protocol(self): - """Getter to mimic request.protocol.""" - - return self._request_handler.request_version - protocol = property(get_protocol) - - def is_https(self): - """Mimic request.is_https().""" - - return self._use_tls - - def _drain_received_data(self): - """Don't use this method from WebSocket handler. Drains unread data - in the receive buffer. - """ - - raw_socket = self._request_handler.connection - drained_data = util.drain_received_data(raw_socket) - - if drained_data: - self._logger.debug( - 'Drained data following close frame: %r', drained_data) - - -class _StandaloneSSLConnection(object): - """A wrapper class for OpenSSL.SSL.Connection to provide makefile method - which is not supported by the class. - """ - - def __init__(self, connection): - self._connection = connection - - def __getattribute__(self, name): - if name in ('_connection', 'makefile'): - return object.__getattribute__(self, name) - return self._connection.__getattribute__(name) - - def __setattr__(self, name, value): - if name in ('_connection', 'makefile'): - return object.__setattr__(self, name, value) - return self._connection.__setattr__(name, value) - - def makefile(self, mode='r', bufsize=-1): - return socket._fileobject(self._connection, mode, bufsize) - - -def _alias_handlers(dispatcher, websock_handlers_map_file): - """Set aliases specified in websock_handler_map_file in dispatcher. - - Args: - dispatcher: dispatch.Dispatcher instance - websock_handler_map_file: alias map file - """ - - fp = open(websock_handlers_map_file) - try: - for line in fp: - if line[0] == '#' or line.isspace(): - continue - m = re.match('(\S+)\s+(\S+)', line) - if not m: - logging.warning('Wrong format in map file:' + line) - continue - try: - dispatcher.add_resource_path_alias( - m.group(1), m.group(2)) - except dispatch.DispatchException, e: - logging.error(str(e)) - finally: - fp.close() - - -class WebSocketServer(SocketServer.ThreadingMixIn, BaseHTTPServer.HTTPServer): - """HTTPServer specialized for WebSocket.""" - - # Overrides SocketServer.ThreadingMixIn.daemon_threads - daemon_threads = True - # Overrides BaseHTTPServer.HTTPServer.allow_reuse_address - allow_reuse_address = True - - def __init__(self, options): - """Override SocketServer.TCPServer.__init__ to set SSL enabled - socket object to self.socket before server_bind and server_activate, - if necessary. - """ - - # Share a Dispatcher among request handlers to save time for - # instantiation. Dispatcher can be shared because it is thread-safe. - options.dispatcher = dispatch.Dispatcher( - options.websock_handlers, - options.scan_dir, - options.allow_handlers_outside_root_dir) - if options.websock_handlers_map_file: - _alias_handlers(options.dispatcher, - options.websock_handlers_map_file) - warnings = options.dispatcher.source_warnings() - if warnings: - for warning in warnings: - logging.warning('mod_pywebsocket: %s' % warning) - - self._logger = util.get_class_logger(self) - - self.request_queue_size = options.request_queue_size - self.__ws_is_shut_down = threading.Event() - self.__ws_serving = False - - SocketServer.BaseServer.__init__( - self, (options.server_host, options.port), WebSocketRequestHandler) - - # Expose the options object to allow handler objects access it. We name - # it with websocket_ prefix to avoid conflict. - self.websocket_server_options = options - - self._create_sockets() - self.server_bind() - self.server_activate() - - def _create_sockets(self): - self.server_name, self.server_port = self.server_address - self._sockets = [] - if not self.server_name: - # On platforms that doesn't support IPv6, the first bind fails. - # On platforms that supports IPv6 - # - If it binds both IPv4 and IPv6 on call with AF_INET6, the - # first bind succeeds and the second fails (we'll see 'Address - # already in use' error). - # - If it binds only IPv6 on call with AF_INET6, both call are - # expected to succeed to listen both protocol. - addrinfo_array = [ - (socket.AF_INET6, socket.SOCK_STREAM, '', '', ''), - (socket.AF_INET, socket.SOCK_STREAM, '', '', '')] - else: - addrinfo_array = socket.getaddrinfo(self.server_name, - self.server_port, - socket.AF_UNSPEC, - socket.SOCK_STREAM, - socket.IPPROTO_TCP) - for addrinfo in addrinfo_array: - self._logger.info('Create socket on: %r', addrinfo) - family, socktype, proto, canonname, sockaddr = addrinfo - try: - socket_ = socket.socket(family, socktype) - except Exception, e: - self._logger.info('Skip by failure: %r', e) - continue - if self.websocket_server_options.use_tls: - if _HAS_SSL: - if self.websocket_server_options.tls_client_auth: - client_cert_ = ssl.CERT_REQUIRED - else: - client_cert_ = ssl.CERT_NONE - socket_ = ssl.wrap_socket(socket_, - keyfile=self.websocket_server_options.private_key, - certfile=self.websocket_server_options.certificate, - ssl_version=ssl.PROTOCOL_SSLv23, - ca_certs=self.websocket_server_options.tls_client_ca, - cert_reqs=client_cert_) - if _HAS_OPEN_SSL: - ctx = OpenSSL.SSL.Context(OpenSSL.SSL.SSLv23_METHOD) - ctx.use_privatekey_file( - self.websocket_server_options.private_key) - ctx.use_certificate_file( - self.websocket_server_options.certificate) - socket_ = OpenSSL.SSL.Connection(ctx, socket_) - self._sockets.append((socket_, addrinfo)) - - def server_bind(self): - """Override SocketServer.TCPServer.server_bind to enable multiple - sockets bind. - """ - - failed_sockets = [] - - for socketinfo in self._sockets: - socket_, addrinfo = socketinfo - self._logger.info('Bind on: %r', addrinfo) - if self.allow_reuse_address: - socket_.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) - try: - socket_.bind(self.server_address) - except Exception, e: - self._logger.info('Skip by failure: %r', e) - socket_.close() - failed_sockets.append(socketinfo) - if self.server_address[1] == 0: - # The operating system assigns the actual port number for port - # number 0. This case, the second and later sockets should use - # the same port number. Also self.server_port is rewritten - # because it is exported, and will be used by external code. - self.server_address = ( - self.server_name, socket_.getsockname()[1]) - self.server_port = self.server_address[1] - self._logger.info('Port %r is assigned', self.server_port) - - for socketinfo in failed_sockets: - self._sockets.remove(socketinfo) - - def server_activate(self): - """Override SocketServer.TCPServer.server_activate to enable multiple - sockets listen. - """ - - failed_sockets = [] - - for socketinfo in self._sockets: - socket_, addrinfo = socketinfo - self._logger.info('Listen on: %r', addrinfo) - try: - socket_.listen(self.request_queue_size) - except Exception, e: - self._logger.info('Skip by failure: %r', e) - socket_.close() - failed_sockets.append(socketinfo) - - for socketinfo in failed_sockets: - self._sockets.remove(socketinfo) - - if len(self._sockets) == 0: - self._logger.critical( - 'No sockets activated. Use info log level to see the reason.') - - def server_close(self): - """Override SocketServer.TCPServer.server_close to enable multiple - sockets close. - """ - - for socketinfo in self._sockets: - socket_, addrinfo = socketinfo - self._logger.info('Close on: %r', addrinfo) - socket_.close() - - def fileno(self): - """Override SocketServer.TCPServer.fileno.""" - - self._logger.critical('Not supported: fileno') - return self._sockets[0][0].fileno() - - def handle_error(self, rquest, client_address): - """Override SocketServer.handle_error.""" - - self._logger.error( - 'Exception in processing request from: %r\n%s', - client_address, - util.get_stack_trace()) - # Note: client_address is a tuple. - - def get_request(self): - """Override TCPServer.get_request to wrap OpenSSL.SSL.Connection - object with _StandaloneSSLConnection to provide makefile method. We - cannot substitute OpenSSL.SSL.Connection.makefile since it's readonly - attribute. - """ - - accepted_socket, client_address = self.socket.accept() - if self.websocket_server_options.use_tls and _HAS_OPEN_SSL: - accepted_socket = _StandaloneSSLConnection(accepted_socket) - return accepted_socket, client_address - - def serve_forever(self, poll_interval=0.5): - """Override SocketServer.BaseServer.serve_forever.""" - - self.__ws_serving = True - self.__ws_is_shut_down.clear() - handle_request = self.handle_request - if hasattr(self, '_handle_request_noblock'): - handle_request = self._handle_request_noblock - else: - self._logger.warning('Fallback to blocking request handler') - try: - while self.__ws_serving: - r, w, e = select.select( - [socket_[0] for socket_ in self._sockets], - [], [], poll_interval) - for socket_ in r: - self.socket = socket_ - handle_request() - self.socket = None - finally: - self.__ws_is_shut_down.set() - - def shutdown(self): - """Override SocketServer.BaseServer.shutdown.""" - - self.__ws_serving = False - self.__ws_is_shut_down.wait() - - -class WebSocketRequestHandler(CGIHTTPServer.CGIHTTPRequestHandler): - """CGIHTTPRequestHandler specialized for WebSocket.""" - - # Use httplib.HTTPMessage instead of mimetools.Message. - MessageClass = httplib.HTTPMessage - - def setup(self): - """Override SocketServer.StreamRequestHandler.setup to wrap rfile - with MemorizingFile. - - This method will be called by BaseRequestHandler's constructor - before calling BaseHTTPRequestHandler.handle. - BaseHTTPRequestHandler.handle will call - BaseHTTPRequestHandler.handle_one_request and it will call - WebSocketRequestHandler.parse_request. - """ - - # Call superclass's setup to prepare rfile, wfile, etc. See setup - # definition on the root class SocketServer.StreamRequestHandler to - # understand what this does. - CGIHTTPServer.CGIHTTPRequestHandler.setup(self) - - self.rfile = memorizingfile.MemorizingFile( - self.rfile, - max_memorized_lines=_MAX_MEMORIZED_LINES) - - def __init__(self, request, client_address, server): - self._logger = util.get_class_logger(self) - - self._options = server.websocket_server_options - - # Overrides CGIHTTPServerRequestHandler.cgi_directories. - self.cgi_directories = self._options.cgi_directories - # Replace CGIHTTPRequestHandler.is_executable method. - if self._options.is_executable_method is not None: - self.is_executable = self._options.is_executable_method - - # This actually calls BaseRequestHandler.__init__. - CGIHTTPServer.CGIHTTPRequestHandler.__init__( - self, request, client_address, server) - - def parse_request(self): - """Override BaseHTTPServer.BaseHTTPRequestHandler.parse_request. - - Return True to continue processing for HTTP(S), False otherwise. - - See BaseHTTPRequestHandler.handle_one_request method which calls - this method to understand how the return value will be handled. - """ - - # We hook parse_request method, but also call the original - # CGIHTTPRequestHandler.parse_request since when we return False, - # CGIHTTPRequestHandler.handle_one_request continues processing and - # it needs variables set by CGIHTTPRequestHandler.parse_request. - # - # Variables set by this method will be also used by WebSocket request - # handling (self.path, self.command, self.requestline, etc. See also - # how _StandaloneRequest's members are implemented using these - # attributes). - if not CGIHTTPServer.CGIHTTPRequestHandler.parse_request(self): - return False - - if self._options.use_basic_auth: - auth = self.headers.getheader('Authorization') - if auth != self._options.basic_auth_credential: - self.send_response(401) - self.send_header('WWW-Authenticate', - 'Basic realm="Pywebsocket"') - self.end_headers() - self._logger.info('Request basic authentication') - return True - - host, port, resource = http_header_util.parse_uri(self.path) - if resource is None: - self._logger.info('Invalid URI: %r', self.path) - self._logger.info('Fallback to CGIHTTPRequestHandler') - return True - server_options = self.server.websocket_server_options - if host is not None: - validation_host = server_options.validation_host - if validation_host is not None and host != validation_host: - self._logger.info('Invalid host: %r (expected: %r)', - host, - validation_host) - self._logger.info('Fallback to CGIHTTPRequestHandler') - return True - if port is not None: - validation_port = server_options.validation_port - if validation_port is not None and port != validation_port: - self._logger.info('Invalid port: %r (expected: %r)', - port, - validation_port) - self._logger.info('Fallback to CGIHTTPRequestHandler') - return True - self.path = resource - - request = _StandaloneRequest(self, self._options.use_tls) - - try: - # Fallback to default http handler for request paths for which - # we don't have request handlers. - if not self._options.dispatcher.get_handler_suite(self.path): - self._logger.info('No handler for resource: %r', - self.path) - self._logger.info('Fallback to CGIHTTPRequestHandler') - return True - except dispatch.DispatchException, e: - self._logger.info('%s', e) - self.send_error(e.status) - return False - - # If any Exceptions without except clause setup (including - # DispatchException) is raised below this point, it will be caught - # and logged by WebSocketServer. - - try: - try: - handshake.do_handshake( - request, - self._options.dispatcher, - allowDraft75=self._options.allow_draft75, - strict=self._options.strict) - except handshake.VersionException, e: - self._logger.info('%s', e) - self.send_response(common.HTTP_STATUS_BAD_REQUEST) - self.send_header(common.SEC_WEBSOCKET_VERSION_HEADER, - e.supported_versions) - self.end_headers() - return False - except handshake.HandshakeException, e: - # Handshake for ws(s) failed. - self._logger.info('%s', e) - self.send_error(e.status) - return False - - request._dispatcher = self._options.dispatcher - self._options.dispatcher.transfer_data(request) - except handshake.AbortedByUserException, e: - self._logger.info('%s', e) - return False - - def log_request(self, code='-', size='-'): - """Override BaseHTTPServer.log_request.""" - - self._logger.info('"%s" %s %s', - self.requestline, str(code), str(size)) - - def log_error(self, *args): - """Override BaseHTTPServer.log_error.""" - - # Despite the name, this method is for warnings than for errors. - # For example, HTTP status code is logged by this method. - self._logger.warning('%s - %s', - self.address_string(), - args[0] % args[1:]) - - def is_cgi(self): - """Test whether self.path corresponds to a CGI script. - - Add extra check that self.path doesn't contains .. - Also check if the file is a executable file or not. - If the file is not executable, it is handled as static file or dir - rather than a CGI script. - """ - - if CGIHTTPServer.CGIHTTPRequestHandler.is_cgi(self): - if '..' in self.path: - return False - # strip query parameter from request path - resource_name = self.path.split('?', 2)[0] - # convert resource_name into real path name in filesystem. - scriptfile = self.translate_path(resource_name) - if not os.path.isfile(scriptfile): - return False - if not self.is_executable(scriptfile): - return False - return True - return False - - -def _get_logger_from_class(c): - return logging.getLogger('%s.%s' % (c.__module__, c.__name__)) - - -def _configure_logging(options): - logging.addLevelName(common.LOGLEVEL_FINE, 'FINE') - - logger = logging.getLogger() - logger.setLevel(logging.getLevelName(options.log_level.upper())) - if options.log_file: - handler = logging.handlers.RotatingFileHandler( - options.log_file, 'a', options.log_max, options.log_count) - else: - handler = logging.StreamHandler() - formatter = logging.Formatter( - '[%(asctime)s] [%(levelname)s] %(name)s: %(message)s') - handler.setFormatter(formatter) - logger.addHandler(handler) - - deflate_log_level_name = logging.getLevelName( - options.deflate_log_level.upper()) - _get_logger_from_class(util._Deflater).setLevel( - deflate_log_level_name) - _get_logger_from_class(util._Inflater).setLevel( - deflate_log_level_name) - - -def _build_option_parser(): - parser = optparse.OptionParser() - - parser.add_option('--config', dest='config_file', type='string', - default=None, - help=('Path to configuration file. See the file comment ' - 'at the top of this file for the configuration ' - 'file format')) - parser.add_option('-H', '--server-host', '--server_host', - dest='server_host', - default='', - help='server hostname to listen to') - parser.add_option('-V', '--validation-host', '--validation_host', - dest='validation_host', - default=None, - help='server hostname to validate in absolute path.') - parser.add_option('-p', '--port', dest='port', type='int', - default=common.DEFAULT_WEB_SOCKET_PORT, - help='port to listen to') - parser.add_option('-P', '--validation-port', '--validation_port', - dest='validation_port', type='int', - default=None, - help='server port to validate in absolute path.') - parser.add_option('-w', '--websock-handlers', '--websock_handlers', - dest='websock_handlers', - default='.', - help=('The root directory of WebSocket handler files. ' - 'If the path is relative, --document-root is used ' - 'as the base.')) - parser.add_option('-m', '--websock-handlers-map-file', - '--websock_handlers_map_file', - dest='websock_handlers_map_file', - default=None, - help=('WebSocket handlers map file. ' - 'Each line consists of alias_resource_path and ' - 'existing_resource_path, separated by spaces.')) - parser.add_option('-s', '--scan-dir', '--scan_dir', dest='scan_dir', - default=None, - help=('Must be a directory under --websock-handlers. ' - 'Only handlers under this directory are scanned ' - 'and registered to the server. ' - 'Useful for saving scan time when the handler ' - 'root directory contains lots of files that are ' - 'not handler file or are handler files but you ' - 'don\'t want them to be registered. ')) - parser.add_option('--allow-handlers-outside-root-dir', - '--allow_handlers_outside_root_dir', - dest='allow_handlers_outside_root_dir', - action='store_true', - default=False, - help=('Scans WebSocket handlers even if their canonical ' - 'path is not under --websock-handlers.')) - parser.add_option('-d', '--document-root', '--document_root', - dest='document_root', default='.', - help='Document root directory.') - parser.add_option('-x', '--cgi-paths', '--cgi_paths', dest='cgi_paths', - default=None, - help=('CGI paths relative to document_root.' - 'Comma-separated. (e.g -x /cgi,/htbin) ' - 'Files under document_root/cgi_path are handled ' - 'as CGI programs. Must be executable.')) - parser.add_option('-t', '--tls', dest='use_tls', action='store_true', - default=False, help='use TLS (wss://)') - parser.add_option('-k', '--private-key', '--private_key', - dest='private_key', - default='', help='TLS private key file.') - parser.add_option('-c', '--certificate', dest='certificate', - default='', help='TLS certificate file.') - parser.add_option('--tls-client-auth', dest='tls_client_auth', - action='store_true', default=False, - help='Requires TLS client auth on every connection.') - parser.add_option('--tls-client-ca', dest='tls_client_ca', default='', - help=('Specifies a pem file which contains a set of ' - 'concatenated CA certificates which are used to ' - 'validate certificates passed from clients')) - parser.add_option('--basic-auth', dest='use_basic_auth', - action='store_true', default=False, - help='Requires Basic authentication.') - parser.add_option('--basic-auth-credential', - dest='basic_auth_credential', default='test:test', - help='Specifies the credential of basic authentication ' - 'by username:password pair (e.g. test:test).') - parser.add_option('-l', '--log-file', '--log_file', dest='log_file', - default='', help='Log file.') - # Custom log level: - # - FINE: Prints status of each frame processing step - parser.add_option('--log-level', '--log_level', type='choice', - dest='log_level', default='warn', - choices=['fine', - 'debug', 'info', 'warning', 'warn', 'error', - 'critical'], - help='Log level.') - parser.add_option('--deflate-log-level', '--deflate_log_level', - type='choice', - dest='deflate_log_level', default='warn', - choices=['debug', 'info', 'warning', 'warn', 'error', - 'critical'], - help='Log level for _Deflater and _Inflater.') - parser.add_option('--thread-monitor-interval-in-sec', - '--thread_monitor_interval_in_sec', - dest='thread_monitor_interval_in_sec', - type='int', default=-1, - help=('If positive integer is specified, run a thread ' - 'monitor to show the status of server threads ' - 'periodically in the specified inteval in ' - 'second. If non-positive integer is specified, ' - 'disable the thread monitor.')) - parser.add_option('--log-max', '--log_max', dest='log_max', type='int', - default=_DEFAULT_LOG_MAX_BYTES, - help='Log maximum bytes') - parser.add_option('--log-count', '--log_count', dest='log_count', - type='int', default=_DEFAULT_LOG_BACKUP_COUNT, - help='Log backup count') - parser.add_option('--allow-draft75', dest='allow_draft75', - action='store_true', default=False, - help='Obsolete option. Ignored.') - parser.add_option('--strict', dest='strict', action='store_true', - default=False, help='Obsolete option. Ignored.') - parser.add_option('-q', '--queue', dest='request_queue_size', type='int', - default=_DEFAULT_REQUEST_QUEUE_SIZE, - help='request queue size') - - return parser - - -class ThreadMonitor(threading.Thread): - daemon = True - - def __init__(self, interval_in_sec): - threading.Thread.__init__(self, name='ThreadMonitor') - - self._logger = util.get_class_logger(self) - - self._interval_in_sec = interval_in_sec - - def run(self): - while True: - thread_name_list = [] - for thread in threading.enumerate(): - thread_name_list.append(thread.name) - self._logger.info( - "%d active threads: %s", - threading.active_count(), - ', '.join(thread_name_list)) - time.sleep(self._interval_in_sec) - - -def _parse_args_and_config(args): - parser = _build_option_parser() - - # First, parse options without configuration file. - temporary_options, temporary_args = parser.parse_args(args=args) - if temporary_args: - logging.critical( - 'Unrecognized positional arguments: %r', temporary_args) - sys.exit(1) - - if temporary_options.config_file: - try: - config_fp = open(temporary_options.config_file, 'r') - except IOError, e: - logging.critical( - 'Failed to open configuration file %r: %r', - temporary_options.config_file, - e) - sys.exit(1) - - config_parser = ConfigParser.SafeConfigParser() - config_parser.readfp(config_fp) - config_fp.close() - - args_from_config = [] - for name, value in config_parser.items('pywebsocket'): - args_from_config.append('--' + name) - args_from_config.append(value) - if args is None: - args = args_from_config - else: - args = args_from_config + args - return parser.parse_args(args=args) - else: - return temporary_options, temporary_args - - -def _main(args=None): - """You can call this function from your own program, but please note that - this function has some side-effects that might affect your program. For - example, util.wrap_popen3_for_win use in this method replaces implementation - of os.popen3. - """ - - options, args = _parse_args_and_config(args=args) - - os.chdir(options.document_root) - - _configure_logging(options) - - # TODO(tyoshino): Clean up initialization of CGI related values. Move some - # of code here to WebSocketRequestHandler class if it's better. - options.cgi_directories = [] - options.is_executable_method = None - if options.cgi_paths: - options.cgi_directories = options.cgi_paths.split(',') - if sys.platform in ('cygwin', 'win32'): - cygwin_path = None - # For Win32 Python, it is expected that CYGWIN_PATH - # is set to a directory of cygwin binaries. - # For example, websocket_server.py in Chromium sets CYGWIN_PATH to - # full path of third_party/cygwin/bin. - if 'CYGWIN_PATH' in os.environ: - cygwin_path = os.environ['CYGWIN_PATH'] - util.wrap_popen3_for_win(cygwin_path) - - def __check_script(scriptpath): - return util.get_script_interp(scriptpath, cygwin_path) - - options.is_executable_method = __check_script - - if options.use_tls: - if not (_HAS_SSL or _HAS_OPEN_SSL): - logging.critical('TLS support requires ssl or pyOpenSSL module.') - sys.exit(1) - if not options.private_key or not options.certificate: - logging.critical( - 'To use TLS, specify private_key and certificate.') - sys.exit(1) - - if options.tls_client_auth: - if not options.use_tls: - logging.critical('TLS must be enabled for client authentication.') - sys.exit(1) - if not _HAS_SSL: - logging.critical('Client authentication requires ssl module.') - - if not options.scan_dir: - options.scan_dir = options.websock_handlers - - if options.use_basic_auth: - options.basic_auth_credential = 'Basic ' + base64.b64encode( - options.basic_auth_credential) - - try: - if options.thread_monitor_interval_in_sec > 0: - # Run a thread monitor to show the status of server threads for - # debugging. - ThreadMonitor(options.thread_monitor_interval_in_sec).start() - - server = WebSocketServer(options) - server.serve_forever() - except Exception, e: - logging.critical('mod_pywebsocket: %s' % e) - logging.critical('mod_pywebsocket: %s' % util.get_stack_trace()) - sys.exit(1) - - -if __name__ == '__main__': - _main(sys.argv[1:]) - - -# vi:sts=4 sw=4 et diff --git a/module/lib/mod_pywebsocket/stream.py b/module/lib/mod_pywebsocket/stream.py deleted file mode 100644 index edc533279..000000000 --- a/module/lib/mod_pywebsocket/stream.py +++ /dev/null @@ -1,57 +0,0 @@ -# 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 exports public symbols. -""" - - -from mod_pywebsocket._stream_base import BadOperationException -from mod_pywebsocket._stream_base import ConnectionTerminatedException -from mod_pywebsocket._stream_base import InvalidFrameException -from mod_pywebsocket._stream_base import InvalidUTF8Exception -from mod_pywebsocket._stream_base import UnsupportedFrameException -from mod_pywebsocket._stream_hixie75 import StreamHixie75 -from mod_pywebsocket._stream_hybi import Frame -from mod_pywebsocket._stream_hybi import Stream -from mod_pywebsocket._stream_hybi import StreamOptions - -# These methods are intended to be used by WebSocket client developers to have -# their implementations receive broken data in tests. -from mod_pywebsocket._stream_hybi import create_close_frame -from mod_pywebsocket._stream_hybi import create_header -from mod_pywebsocket._stream_hybi import create_length_header -from mod_pywebsocket._stream_hybi import create_ping_frame -from mod_pywebsocket._stream_hybi import create_pong_frame -from mod_pywebsocket._stream_hybi import create_binary_frame -from mod_pywebsocket._stream_hybi import create_text_frame -from mod_pywebsocket._stream_hybi import create_closing_handshake_body - - -# vi:sts=4 sw=4 et diff --git a/module/lib/mod_pywebsocket/util.py b/module/lib/mod_pywebsocket/util.py deleted file mode 100644 index 7bb0b5d9e..000000000 --- a/module/lib/mod_pywebsocket/util.py +++ /dev/null @@ -1,515 +0,0 @@ -# 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. - - -"""WebSocket utilities. -""" - - -import array -import errno - -# Import hash classes from a module available and recommended for each Python -# version and re-export those symbol. Use sha and md5 module in Python 2.4, and -# hashlib module in Python 2.6. -try: - import hashlib - md5_hash = hashlib.md5 - sha1_hash = hashlib.sha1 -except ImportError: - import md5 - import sha - md5_hash = md5.md5 - sha1_hash = sha.sha - -import StringIO -import logging -import os -import re -import socket -import traceback -import zlib - - -def get_stack_trace(): - """Get the current stack trace as string. - - This is needed to support Python 2.3. - TODO: Remove this when we only support Python 2.4 and above. - Use traceback.format_exc instead. - """ - - out = StringIO.StringIO() - traceback.print_exc(file=out) - return out.getvalue() - - -def prepend_message_to_exception(message, exc): - """Prepend message to the exception.""" - - exc.args = (message + str(exc),) - return - - -def __translate_interp(interp, cygwin_path): - """Translate interp program path for Win32 python to run cygwin program - (e.g. perl). Note that it doesn't support path that contains space, - which is typically true for Unix, where #!-script is written. - For Win32 python, cygwin_path is a directory of cygwin binaries. - - Args: - interp: interp command line - cygwin_path: directory name of cygwin binary, or None - Returns: - translated interp command line. - """ - if not cygwin_path: - return interp - m = re.match('^[^ ]*/([^ ]+)( .*)?', interp) - if m: - cmd = os.path.join(cygwin_path, m.group(1)) - return cmd + m.group(2) - return interp - - -def get_script_interp(script_path, cygwin_path=None): - """Gets #!-interpreter command line from the script. - - It also fixes command path. When Cygwin Python is used, e.g. in WebKit, - it could run "/usr/bin/perl -wT hello.pl". - When Win32 Python is used, e.g. in Chromium, it couldn't. So, fix - "/usr/bin/perl" to "<cygwin_path>\perl.exe". - - Args: - script_path: pathname of the script - cygwin_path: directory name of cygwin binary, or None - Returns: - #!-interpreter command line, or None if it is not #!-script. - """ - fp = open(script_path) - line = fp.readline() - fp.close() - m = re.match('^#!(.*)', line) - if m: - return __translate_interp(m.group(1), cygwin_path) - return None - - -def wrap_popen3_for_win(cygwin_path): - """Wrap popen3 to support #!-script on Windows. - - Args: - cygwin_path: path for cygwin binary if command path is needed to be - translated. None if no translation required. - """ - - __orig_popen3 = os.popen3 - - def __wrap_popen3(cmd, mode='t', bufsize=-1): - cmdline = cmd.split(' ') - interp = get_script_interp(cmdline[0], cygwin_path) - if interp: - cmd = interp + ' ' + cmd - return __orig_popen3(cmd, mode, bufsize) - - os.popen3 = __wrap_popen3 - - -def hexify(s): - return ' '.join(map(lambda x: '%02x' % ord(x), s)) - - -def get_class_logger(o): - return logging.getLogger( - '%s.%s' % (o.__class__.__module__, o.__class__.__name__)) - - -class NoopMasker(object): - """A masking object that has the same interface as RepeatedXorMasker but - just returns the string passed in without making any change. - """ - - def __init__(self): - pass - - def mask(self, s): - return s - - -class RepeatedXorMasker(object): - """A masking object that applies XOR on the string given to mask method - with the masking bytes given to the constructor repeatedly. This object - remembers the position in the masking bytes the last mask method call - ended and resumes from that point on the next mask method call. - """ - - def __init__(self, mask): - self._mask = map(ord, mask) - self._mask_size = len(self._mask) - self._count = 0 - - def mask(self, s): - result = array.array('B') - result.fromstring(s) - # Use temporary local variables to eliminate the cost to access - # attributes - count = self._count - mask = self._mask - mask_size = self._mask_size - for i in xrange(len(result)): - result[i] ^= mask[count] - count = (count + 1) % mask_size - self._count = count - - return result.tostring() - - -class DeflateRequest(object): - """A wrapper class for request object to intercept send and recv to perform - deflate compression and decompression transparently. - """ - - def __init__(self, request): - self._request = request - self.connection = DeflateConnection(request.connection) - - def __getattribute__(self, name): - if name in ('_request', 'connection'): - return object.__getattribute__(self, name) - return self._request.__getattribute__(name) - - def __setattr__(self, name, value): - if name in ('_request', 'connection'): - return object.__setattr__(self, name, value) - return self._request.__setattr__(name, value) - - -# By making wbits option negative, we can suppress CMF/FLG (2 octet) and -# ADLER32 (4 octet) fields of zlib so that we can use zlib module just as -# deflate library. DICTID won't be added as far as we don't set dictionary. -# LZ77 window of 32K will be used for both compression and decompression. -# For decompression, we can just use 32K to cover any windows size. For -# compression, we use 32K so receivers must use 32K. -# -# Compression level is Z_DEFAULT_COMPRESSION. We don't have to match level -# to decode. -# -# See zconf.h, deflate.cc, inflate.cc of zlib library, and zlibmodule.c of -# Python. See also RFC1950 (ZLIB 3.3). - - -class _Deflater(object): - - def __init__(self, window_bits): - self._logger = get_class_logger(self) - - self._compress = zlib.compressobj( - zlib.Z_DEFAULT_COMPRESSION, zlib.DEFLATED, -window_bits) - - def compress(self, bytes): - compressed_bytes = self._compress.compress(bytes) - self._logger.debug('Compress input %r', bytes) - self._logger.debug('Compress result %r', compressed_bytes) - return compressed_bytes - - def compress_and_flush(self, bytes): - compressed_bytes = self._compress.compress(bytes) - compressed_bytes += self._compress.flush(zlib.Z_SYNC_FLUSH) - self._logger.debug('Compress input %r', bytes) - self._logger.debug('Compress result %r', compressed_bytes) - return compressed_bytes - - def compress_and_finish(self, bytes): - compressed_bytes = self._compress.compress(bytes) - compressed_bytes += self._compress.flush(zlib.Z_FINISH) - self._logger.debug('Compress input %r', bytes) - self._logger.debug('Compress result %r', compressed_bytes) - return compressed_bytes - -class _Inflater(object): - - def __init__(self): - self._logger = get_class_logger(self) - - self._unconsumed = '' - - self.reset() - - def decompress(self, size): - if not (size == -1 or size > 0): - raise Exception('size must be -1 or positive') - - data = '' - - while True: - if size == -1: - data += self._decompress.decompress(self._unconsumed) - # See Python bug http://bugs.python.org/issue12050 to - # understand why the same code cannot be used for updating - # self._unconsumed for here and else block. - self._unconsumed = '' - else: - data += self._decompress.decompress( - self._unconsumed, size - len(data)) - self._unconsumed = self._decompress.unconsumed_tail - if self._decompress.unused_data: - # Encountered a last block (i.e. a block with BFINAL = 1) and - # found a new stream (unused_data). We cannot use the same - # zlib.Decompress object for the new stream. Create a new - # Decompress object to decompress the new one. - # - # It's fine to ignore unconsumed_tail if unused_data is not - # empty. - self._unconsumed = self._decompress.unused_data - self.reset() - if size >= 0 and len(data) == size: - # data is filled. Don't call decompress again. - break - else: - # Re-invoke Decompress.decompress to try to decompress all - # available bytes before invoking read which blocks until - # any new byte is available. - continue - else: - # Here, since unused_data is empty, even if unconsumed_tail is - # not empty, bytes of requested length are already in data. We - # don't have to "continue" here. - break - - if data: - self._logger.debug('Decompressed %r', data) - return data - - def append(self, data): - self._logger.debug('Appended %r', data) - self._unconsumed += data - - def reset(self): - self._logger.debug('Reset') - self._decompress = zlib.decompressobj(-zlib.MAX_WBITS) - - -# Compresses/decompresses given octets using the method introduced in RFC1979. - - -class _RFC1979Deflater(object): - """A compressor class that applies DEFLATE to given byte sequence and - flushes using the algorithm described in the RFC1979 section 2.1. - """ - - def __init__(self, window_bits, no_context_takeover): - self._deflater = None - if window_bits is None: - window_bits = zlib.MAX_WBITS - self._window_bits = window_bits - self._no_context_takeover = no_context_takeover - - def filter(self, bytes, flush=True, bfinal=False): - if self._deflater is None or (self._no_context_takeover and flush): - self._deflater = _Deflater(self._window_bits) - - if bfinal: - result = self._deflater.compress_and_finish(bytes) - # Add a padding block with BFINAL = 0 and BTYPE = 0. - result = result + chr(0) - self._deflater = None - return result - if flush: - # Strip last 4 octets which is LEN and NLEN field of a - # non-compressed block added for Z_SYNC_FLUSH. - return self._deflater.compress_and_flush(bytes)[:-4] - return self._deflater.compress(bytes) - -class _RFC1979Inflater(object): - """A decompressor class for byte sequence compressed and flushed following - the algorithm described in the RFC1979 section 2.1. - """ - - def __init__(self): - self._inflater = _Inflater() - - def filter(self, bytes): - # Restore stripped LEN and NLEN field of a non-compressed block added - # for Z_SYNC_FLUSH. - self._inflater.append(bytes + '\x00\x00\xff\xff') - return self._inflater.decompress(-1) - - -class DeflateSocket(object): - """A wrapper class for socket object to intercept send and recv to perform - deflate compression and decompression transparently. - """ - - # Size of the buffer passed to recv to receive compressed data. - _RECV_SIZE = 4096 - - def __init__(self, socket): - self._socket = socket - - self._logger = get_class_logger(self) - - self._deflater = _Deflater(zlib.MAX_WBITS) - self._inflater = _Inflater() - - def recv(self, size): - """Receives data from the socket specified on the construction up - to the specified size. Once any data is available, returns it even - if it's smaller than the specified size. - """ - - # TODO(tyoshino): Allow call with size=0. It should block until any - # decompressed data is available. - if size <= 0: - raise Exception('Non-positive size passed') - while True: - data = self._inflater.decompress(size) - if len(data) != 0: - return data - - read_data = self._socket.recv(DeflateSocket._RECV_SIZE) - if not read_data: - return '' - self._inflater.append(read_data) - - def sendall(self, bytes): - self.send(bytes) - - def send(self, bytes): - self._socket.sendall(self._deflater.compress_and_flush(bytes)) - return len(bytes) - - -class DeflateConnection(object): - """A wrapper class for request object to intercept write and read to - perform deflate compression and decompression transparently. - """ - - def __init__(self, connection): - self._connection = connection - - self._logger = get_class_logger(self) - - self._deflater = _Deflater(zlib.MAX_WBITS) - self._inflater = _Inflater() - - def get_remote_addr(self): - return self._connection.remote_addr - remote_addr = property(get_remote_addr) - - def put_bytes(self, bytes): - self.write(bytes) - - def read(self, size=-1): - """Reads at most size bytes. Blocks until there's at least one byte - available. - """ - - # TODO(tyoshino): Allow call with size=0. - if not (size == -1 or size > 0): - raise Exception('size must be -1 or positive') - - data = '' - while True: - if size == -1: - data += self._inflater.decompress(-1) - else: - data += self._inflater.decompress(size - len(data)) - - if size >= 0 and len(data) != 0: - break - - # TODO(tyoshino): Make this read efficient by some workaround. - # - # In 3.0.3 and prior of mod_python, read blocks until length bytes - # was read. We don't know the exact size to read while using - # deflate, so read byte-by-byte. - # - # _StandaloneRequest.read that ultimately performs - # socket._fileobject.read also blocks until length bytes was read - read_data = self._connection.read(1) - if not read_data: - break - self._inflater.append(read_data) - return data - - def write(self, bytes): - self._connection.write(self._deflater.compress_and_flush(bytes)) - - -def _is_ewouldblock_errno(error_number): - """Returns True iff error_number indicates that receive operation would - block. To make this portable, we check availability of errno and then - compare them. - """ - - for error_name in ['WSAEWOULDBLOCK', 'EWOULDBLOCK', 'EAGAIN']: - if (error_name in dir(errno) and - error_number == getattr(errno, error_name)): - return True - return False - - -def drain_received_data(raw_socket): - # Set the socket non-blocking. - original_timeout = raw_socket.gettimeout() - raw_socket.settimeout(0.0) - - drained_data = [] - - # Drain until the socket is closed or no data is immediately - # available for read. - while True: - try: - data = raw_socket.recv(1) - if not data: - break - drained_data.append(data) - except socket.error, e: - # e can be either a pair (errno, string) or just a string (or - # something else) telling what went wrong. We suppress only - # the errors that indicates that the socket blocks. Those - # exceptions can be parsed as a pair (errno, string). - try: - error_number, message = e - except: - # Failed to parse socket.error. - raise e - - if _is_ewouldblock_errno(error_number): - break - else: - raise e - - # Rollback timeout value. - raw_socket.settimeout(original_timeout) - - return ''.join(drained_data) - - -# vi:sts=4 sw=4 et |