##// END OF EJS Templates
on("destroy",...) -> once("destroy",...) so we don't keep a reference to it, preventing gc...
on("destroy",...) -> once("destroy",...) so we don't keep a reference to it, preventing gc Thanks to Sylvain Corlay for the suggestion.

File last commit:

r17881:a122b461
r18058:c7253b21
Show More
zmqhandlers.py
214 lines | 7.3 KiB | text/x-python | PythonLexer
"""Tornado handlers for WebSocket <-> ZMQ sockets."""
# Copyright (c) IPython Development Team.
# Distributed under the terms of the Modified BSD License.
import json
try:
from urllib.parse import urlparse # Py 3
except ImportError:
from urlparse import urlparse # Py 2
try:
from http.cookies import SimpleCookie # Py 3
except ImportError:
from Cookie import SimpleCookie # Py 2
import logging
import tornado
from tornado import ioloop
from tornado import web
from tornado import websocket
from IPython.kernel.zmq.session import Session
from IPython.utils.jsonutil import date_default
from IPython.utils.py3compat import PY3, cast_unicode
from .handlers import IPythonHandler
class ZMQStreamHandler(websocket.WebSocketHandler):
def check_origin(self, origin):
"""Check Origin == Host or Access-Control-Allow-Origin.
Tornado >= 4 calls this method automatically, raising 403 if it returns False.
We call it explicitly in `open` on Tornado < 4.
"""
if self.allow_origin == '*':
return True
host = self.request.headers.get("Host")
# If no header is provided, assume we can't verify origin
if origin is None:
self.log.warn("Missing Origin header, rejecting WebSocket connection.")
return False
if host is None:
self.log.warn("Missing Host header, rejecting WebSocket connection.")
return False
origin = origin.lower()
origin_host = urlparse(origin).netloc
# OK if origin matches host
if origin_host == host:
return True
# Check CORS headers
if self.allow_origin:
allow = self.allow_origin == origin
elif self.allow_origin_pat:
allow = bool(self.allow_origin_pat.match(origin))
else:
# No CORS headers deny the request
allow = False
if not allow:
self.log.warn("Blocking Cross Origin WebSocket Attempt. Origin: %s, Host: %s",
origin, host,
)
return allow
def clear_cookie(self, *args, **kwargs):
"""meaningless for websockets"""
pass
def _reserialize_reply(self, msg_list):
"""Reserialize a reply message using JSON.
This takes the msg list from the ZMQ socket, unserializes it using
self.session and then serializes the result using JSON. This method
should be used by self._on_zmq_reply to build messages that can
be sent back to the browser.
"""
idents, msg_list = self.session.feed_identities(msg_list)
msg = self.session.unserialize(msg_list)
try:
msg['header'].pop('date')
except KeyError:
pass
try:
msg['parent_header'].pop('date')
except KeyError:
pass
msg.pop('buffers')
return json.dumps(msg, default=date_default)
def _on_zmq_reply(self, msg_list):
# Sometimes this gets triggered when the on_close method is scheduled in the
# eventloop but hasn't been called.
if self.stream.closed(): return
try:
msg = self._reserialize_reply(msg_list)
except Exception:
self.log.critical("Malformed message: %r" % msg_list, exc_info=True)
else:
self.write_message(msg)
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
last_pong = 0
@property
def ping_interval(self):
"""The interval for websocket keep-alive pings.
Set ws_ping_interval = 0 to disable pings.
"""
return self.settings.get('ws_ping_interval', WS_PING_INTERVAL)
@property
def ping_timeout(self):
"""If no ping is received in this many milliseconds,
close the websocket connection (VPNs, etc. can fail to cleanly close ws connections).
Default is max of 3 pings or 30 seconds.
"""
return self.settings.get('ws_ping_timeout',
max(3 * self.ping_interval, WS_PING_INTERVAL)
)
def set_default_headers(self):
"""Undo the set_default_headers in IPythonHandler
which doesn't make sense for websockets
"""
pass
def open(self, kernel_id):
self.kernel_id = cast_unicode(kernel_id, 'ascii')
# Check to see that origin matches host directly, including ports
# Tornado 4 already does CORS checking
if tornado.version_info[0] < 4:
if not self.check_origin(self.get_origin()):
raise web.HTTPError(403)
self.session = Session(config=self.config)
self.save_on_message = self.on_message
self.on_message = self.on_first_message
# start the pinging
if self.ping_interval > 0:
self.last_ping = ioloop.IOLoop.instance().time() # Remember time of last ping
self.last_pong = self.last_ping
self.ping_callback = ioloop.PeriodicCallback(self.send_ping, self.ping_interval)
self.ping_callback.start()
def send_ping(self):
"""send a ping to keep the websocket alive"""
if self.stream.closed() and self.ping_callback is not None:
self.ping_callback.stop()
return
# check for timeout on pong. Make sure that we really have sent a recent ping in
# case the machine with both server and client has been suspended since the last ping.
now = ioloop.IOLoop.instance().time()
since_last_pong = 1e3 * (now - self.last_pong)
since_last_ping = 1e3 * (now - self.last_ping)
if since_last_ping < 2*self.ping_interval and since_last_pong > self.ping_timeout:
self.log.warn("WebSocket ping timeout after %i ms.", since_last_pong)
self.close()
return
self.ping(b'')
self.last_ping = now
def on_pong(self, data):
self.last_pong = ioloop.IOLoop.instance().time()
def _inject_cookie_message(self, msg):
"""Inject the first message, which is the document cookie,
for authentication."""
if not PY3 and isinstance(msg, unicode):
# Cookie constructor doesn't accept unicode strings
# under Python 2.x for some reason
msg = msg.encode('utf8', 'replace')
try:
identity, msg = msg.split(':', 1)
self.session.session = cast_unicode(identity, 'ascii')
except Exception:
logging.error("First ws message didn't have the form 'identity:[cookie]' - %r", msg)
try:
self.request._cookies = SimpleCookie(msg)
except:
self.log.warn("couldn't parse cookie string: %s",msg, exc_info=True)
def on_first_message(self, msg):
self._inject_cookie_message(msg)
if self.get_current_user() is None:
self.log.warn("Couldn't authenticate WebSocket connection")
raise web.HTTPError(403)
self.on_message = self.save_on_message