forward-port draft76 websockets

from tornado-3.2.2

only enabled during testing with phantomjs

IPython/html/allow76.py

@@ -0,0 +1,312 @@
+"""WebsocketProtocol76 from tornado 3.2.2 for tornado >= 4.0
+
+The contents of this file are Copyright (c) Tornado
+Used under the Apache 2.0 license
+"""
+
+
+from __future__ import absolute_import, division, print_function, with_statement
+# Author: Jacob Kristhammar, 2010
+
+import functools
+import hashlib
+import struct
+import time
+import tornado.escape
+import tornado.web
+
+from tornado.log import gen_log, app_log
+from tornado.util import bytes_type, unicode_type
+
+from tornado.websocket import WebSocketHandler, WebSocketProtocol13
+
+class AllowDraftWebSocketHandler(WebSocketHandler):
+    """Restore Draft76 support for tornado 4
+    
+    Remove when we can run tests without phantomjs + qt4
+    """
+    
+    # get is unmodified except between the BEGIN/END PATCH lines
+    @tornado.web.asynchronous
+    def get(self, *args, **kwargs):
+        self.open_args = args
+        self.open_kwargs = kwargs
+
+        # Upgrade header should be present and should be equal to WebSocket
+        if self.request.headers.get("Upgrade", "").lower() != 'websocket':
+            self.set_status(400)
+            self.finish("Can \"Upgrade\" only to \"WebSocket\".")
+            return
+
+        # Connection header should be upgrade. Some proxy servers/load balancers
+        # might mess with it.
+        headers = self.request.headers
+        connection = map(lambda s: s.strip().lower(), headers.get("Connection", "").split(","))
+        if 'upgrade' not in connection:
+            self.set_status(400)
+            self.finish("\"Connection\" must be \"Upgrade\".")
+            return
+
+        # Handle WebSocket Origin naming convention differences
+        # The difference between version 8 and 13 is that in 8 the
+        # client sends a "Sec-Websocket-Origin" header and in 13 it's
+        # simply "Origin".
+        if "Origin" in self.request.headers:
+            origin = self.request.headers.get("Origin")
+        else:
+            origin = self.request.headers.get("Sec-Websocket-Origin", None)
+
+
+        # If there was an origin header, check to make sure it matches
+        # according to check_origin. When the origin is None, we assume it
+        # did not come from a browser and that it can be passed on.
+        if origin is not None and not self.check_origin(origin):
+            self.set_status(403)
+            self.finish("Cross origin websockets not allowed")
+            return
+
+        self.stream = self.request.connection.detach()
+        self.stream.set_close_callback(self.on_connection_close)
+
+        if self.request.headers.get("Sec-WebSocket-Version") in ("7", "8", "13"):
+            self.ws_connection = WebSocketProtocol13(
+                self, compression_options=self.get_compression_options())
+            self.ws_connection.accept_connection()
+        #--------------- BEGIN PATCH ----------------
+        elif (self.allow_draft76() and
+              "Sec-WebSocket-Version" not in self.request.headers):
+            self.ws_connection = WebSocketProtocol76(self)
+            self.ws_connection.accept_connection()
+        #--------------- END PATCH ----------------
+        else:
+            if not self.stream.closed():
+                self.stream.write(tornado.escape.utf8(
+                    "HTTP/1.1 426 Upgrade Required\r\n"
+                    "Sec-WebSocket-Version: 8\r\n\r\n"))
+                self.stream.close()
+    
+    # 3.2 methods removed in 4.0:
+    def allow_draft76(self):
+        """Using this class allows draft76 connections by default"""
+        return True
+    
+    def get_websocket_scheme(self):
+        """Return the url scheme used for this request, either "ws" or "wss".
+        This is normally decided by HTTPServer, but applications
+        may wish to override this if they are using an SSL proxy
+        that does not provide the X-Scheme header as understood
+        by HTTPServer.
+        Note that this is only used by the draft76 protocol.
+        """
+        return "wss" if self.request.protocol == "https" else "ws"
+    
+
+
+# No modifications from tornado-3.2.2 below this line
+
+class WebSocketProtocol(object):
+    """Base class for WebSocket protocol versions.
+    """
+    def __init__(self, handler):
+        self.handler = handler
+        self.request = handler.request
+        self.stream = handler.stream
+        self.client_terminated = False
+        self.server_terminated = False
+
+    def async_callback(self, callback, *args, **kwargs):
+        """Wrap callbacks with this if they are used on asynchronous requests.
+
+        Catches exceptions properly and closes this WebSocket if an exception
+        is uncaught.
+        """
+        if args or kwargs:
+            callback = functools.partial(callback, *args, **kwargs)
+
+        def wrapper(*args, **kwargs):
+            try:
+                return callback(*args, **kwargs)
+            except Exception:
+                app_log.error("Uncaught exception in %s",
+                              self.request.path, exc_info=True)
+                self._abort()
+        return wrapper
+
+    def on_connection_close(self):
+        self._abort()
+
+    def _abort(self):
+        """Instantly aborts the WebSocket connection by closing the socket"""
+        self.client_terminated = True
+        self.server_terminated = True
+        self.stream.close()  # forcibly tear down the connection
+        self.close()  # let the subclass cleanup
+
+
+class WebSocketProtocol76(WebSocketProtocol):
+    """Implementation of the WebSockets protocol, version hixie-76.
+
+    This class provides basic functionality to process WebSockets requests as
+    specified in
+    http://tools.ietf.org/html/draft-hixie-thewebsocketprotocol-76
+    """
+    def __init__(self, handler):
+        WebSocketProtocol.__init__(self, handler)
+        self.challenge = None
+        self._waiting = None
+
+    def accept_connection(self):
+        try:
+            self._handle_websocket_headers()
+        except ValueError:
+            gen_log.debug("Malformed WebSocket request received")
+            self._abort()
+            return
+
+        scheme = self.handler.get_websocket_scheme()
+
+        # draft76 only allows a single subprotocol
+        subprotocol_header = ''
+        subprotocol = self.request.headers.get("Sec-WebSocket-Protocol", None)
+        if subprotocol:
+            selected = self.handler.select_subprotocol([subprotocol])
+            if selected:
+                assert selected == subprotocol
+                subprotocol_header = "Sec-WebSocket-Protocol: %s\r\n" % selected
+
+        # Write the initial headers before attempting to read the challenge.
+        # This is necessary when using proxies (such as HAProxy), which
+        # need to see the Upgrade headers before passing through the
+        # non-HTTP traffic that follows.
+        self.stream.write(tornado.escape.utf8(
+            "HTTP/1.1 101 WebSocket Protocol Handshake\r\n"
+            "Upgrade: WebSocket\r\n"
+            "Connection: Upgrade\r\n"
+            "Server: TornadoServer/%(version)s\r\n"
+            "Sec-WebSocket-Origin: %(origin)s\r\n"
+            "Sec-WebSocket-Location: %(scheme)s://%(host)s%(uri)s\r\n"
+            "%(subprotocol)s"
+            "\r\n" % (dict(
+                version=tornado.version,
+                origin=self.request.headers["Origin"],
+                scheme=scheme,
+                host=self.request.host,
+                uri=self.request.uri,
+                subprotocol=subprotocol_header))))
+        self.stream.read_bytes(8, self._handle_challenge)
+
+    def challenge_response(self, challenge):
+        """Generates the challenge response that's needed in the handshake
+
+        The challenge parameter should be the raw bytes as sent from the
+        client.
+        """
+        key_1 = self.request.headers.get("Sec-Websocket-Key1")
+        key_2 = self.request.headers.get("Sec-Websocket-Key2")
+        try:
+            part_1 = self._calculate_part(key_1)
+            part_2 = self._calculate_part(key_2)
+        except ValueError:
+            raise ValueError("Invalid Keys/Challenge")
+        return self._generate_challenge_response(part_1, part_2, challenge)
+
+    def _handle_challenge(self, challenge):
+        try:
+            challenge_response = self.challenge_response(challenge)
+        except ValueError:
+            gen_log.debug("Malformed key data in WebSocket request")
+            self._abort()
+            return
+        self._write_response(challenge_response)
+
+    def _write_response(self, challenge):
+        self.stream.write(challenge)
+        self.async_callback(self.handler.open)(*self.handler.open_args, **self.handler.open_kwargs)
+        self._receive_message()
+
+    def _handle_websocket_headers(self):
+        """Verifies all invariant- and required headers
+
+        If a header is missing or have an incorrect value ValueError will be
+        raised
+        """
+        fields = ("Origin", "Host", "Sec-Websocket-Key1",
+                  "Sec-Websocket-Key2")
+        if not all(map(lambda f: self.request.headers.get(f), fields)):
+            raise ValueError("Missing/Invalid WebSocket headers")
+
+    def _calculate_part(self, key):
+        """Processes the key headers and calculates their key value.
+
+        Raises ValueError when feed invalid key."""
+        # pyflakes complains about variable reuse if both of these lines use 'c'
+        number = int(''.join(c for c in key if c.isdigit()))
+        spaces = len([c2 for c2 in key if c2.isspace()])
+        try:
+            key_number = number // spaces
+        except (ValueError, ZeroDivisionError):
+            raise ValueError
+        return struct.pack(">I", key_number)
+
+    def _generate_challenge_response(self, part_1, part_2, part_3):
+        m = hashlib.md5()
+        m.update(part_1)
+        m.update(part_2)
+        m.update(part_3)
+        return m.digest()
+
+    def _receive_message(self):
+        self.stream.read_bytes(1, self._on_frame_type)
+
+    def _on_frame_type(self, byte):
+        frame_type = ord(byte)
+        if frame_type == 0x00:
+            self.stream.read_until(b"\xff", self._on_end_delimiter)
+        elif frame_type == 0xff:
+            self.stream.read_bytes(1, self._on_length_indicator)
+        else:
+            self._abort()
+
+    def _on_end_delimiter(self, frame):
+        if not self.client_terminated:
+            self.async_callback(self.handler.on_message)(
+                frame[:-1].decode("utf-8", "replace"))
+        if not self.client_terminated:
+            self._receive_message()
+
+    def _on_length_indicator(self, byte):
+        if ord(byte) != 0x00:
+            self._abort()
+            return
+        self.client_terminated = True
+        self.close()
+
+    def write_message(self, message, binary=False):
+        """Sends the given message to the client of this Web Socket."""
+        if binary:
+            raise ValueError(
+                "Binary messages not supported by this version of websockets")
+        if isinstance(message, unicode_type):
+            message = message.encode("utf-8")
+        assert isinstance(message, bytes_type)
+        self.stream.write(b"\x00" + message + b"\xff")
+
+    def write_ping(self, data):
+        """Send ping frame."""
+        raise ValueError("Ping messages not supported by this version of websockets")
+
+    def close(self):
+        """Closes the WebSocket connection."""
+        if not self.server_terminated:
+            if not self.stream.closed():
+                self.stream.write("\xff\x00")
+            self.server_terminated = True
+        if self.client_terminated:
+            if self._waiting is not None:
+                self.stream.io_loop.remove_timeout(self._waiting)
+            self._waiting = None
+            self.stream.close()
+        elif self._waiting is None:
+            self._waiting = self.stream.io_loop.add_timeout(
+                time.time() + 5, self._abort)
+

IPython/html/base/zmqhandlers.py

@@ -4,8 +4,10 @@
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
+import os
 import json
 import struct
+import warnings
 
 try:
     from urllib.parse import urlparse # Py 3
@@ -13,7 +15,8 @@ except ImportError:
     from urlparse import urlparse # Py 2
 
 import tornado
-from tornado import gen, ioloop, web, websocket
+from tornado import gen, ioloop, web
+from tornado.websocket import WebSocketHandler
 
 from IPython.kernel.zmq.session import Session
 from IPython.utils.jsonutil import date_default, extract_dates
@@ -21,7 +24,6 @@ from IPython.utils.py3compat import cast_unicode
 
 from .handlers import IPythonHandler
 
-
 def serialize_binary_message(msg):
     """serialize a message as a binary blob
 
@@ -79,8 +81,18 @@ def deserialize_binary_message(bmsg):
     msg['buffers'] = bufs[1:]
     return msg
 
+# ping interval for keeping websockets alive (30 seconds)
+WS_PING_INTERVAL = 30000
 
-class ZMQStreamHandler(websocket.WebSocketHandler):
+if os.environ.get('IPYTHON_ALLOW_DRAFT_WEBSOCKETS_FOR_PHANTOMJS', False):
+    warnings.warn("""Allowing draft76 websocket connections!
+    This should only be done for testing with phantomjs!""")
+    from IPython.html import allow76
+    WebSocketHandler = allow76.AllowDraftWebSocketHandler
+    # draft 76 doesn't support ping
+    WS_PING_INTERVAL = 0
+
+class ZMQStreamHandler(WebSocketHandler):
     
     def check_origin(self, origin):
         """Check Origin == Host or Access-Control-Allow-Origin.
@@ -154,17 +166,6 @@ class ZMQStreamHandler(websocket.WebSocketHandler):
         else:
             self.write_message(msg, binary=isinstance(msg, bytes))
 
-    def allow_draft76(self):
-        """Allow draft 76, until browsers such as Safari update to RFC 6455.
-        
-        This has been disabled by default in tornado in release 2.2.0, and
-        support will be removed in later versions.
-        """
-        return True
-
-# ping interval for keeping websockets alive (30 seconds)
-WS_PING_INTERVAL = 30000
-
 class AuthenticatedZMQStreamHandler(ZMQStreamHandler, IPythonHandler):
     ping_callback = None
     last_ping = 0

IPython/testing/iptestcontroller.py

@@ -325,7 +325,15 @@ class JSController(TestController):
             command.append('--KernelManager.transport=ipc')
         self.stream_capturer = c = StreamCapturer()
         c.start()
-        self.server = subprocess.Popen(command, stdout=c.writefd, stderr=subprocess.STDOUT, cwd=self.nbdir.name)
+        env = os.environ.copy()
+        if self.engine == 'phantomjs':
+            env['IPYTHON_ALLOW_DRAFT_WEBSOCKETS_FOR_PHANTOMJS'] = '1'
+        self.server = subprocess.Popen(command,
+            stdout=c.writefd,
+            stderr=subprocess.STDOUT,
+            cwd=self.nbdir.name,
+            env=env,
+        )
         self.server_info_file = os.path.join(self.ipydir.name,
             'profile_default', 'security', 'nbserver-%i.json' % self.server.pid
         )