use HMAC digest to sign messages instead of cleartext key

also some cleanup of Session code

security doc updated as well.

Buffers do not get digested, so large (non-copying) messages should not
cause performance to suffer too greatly.

IPython/parallel/apps/ipcontrollerapp.py

@@ -1,398 +1,402 b''
 #!/usr/bin/env python
 # encoding: utf-8
 """
 The IPython controller application.
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008-2009  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 from __future__ import with_statement
 
 import os
 import socket
 import stat
 import sys
 import uuid
 
 from multiprocessing import Process
 
 import zmq
 from zmq.devices import ProcessMonitoredQueue
 from zmq.log.handlers import PUBHandler
 from zmq.utils import jsonapi as json
 
+from IPython.config.application import boolean_flag
 from IPython.core.newapplication import ProfileDir
 
 from IPython.parallel.apps.baseapp import (
     BaseParallelApplication,
     base_flags
 )
 from IPython.utils.importstring import import_item
 from IPython.utils.traitlets import Instance, Unicode, Bool, List, Dict
 
 # from IPython.parallel.controller.controller import ControllerFactory
 from IPython.parallel.streamsession import StreamSession
 from IPython.parallel.controller.heartmonitor import HeartMonitor
 from IPython.parallel.controller.hub import HubFactory
 from IPython.parallel.controller.scheduler import TaskScheduler,launch_scheduler
 from IPython.parallel.controller.sqlitedb import SQLiteDB
 
 from IPython.parallel.util import signal_children, split_url
 
 # conditional import of MongoDB backend class
 
 try:
     from IPython.parallel.controller.mongodb import MongoDB
 except ImportError:
     maybe_mongo = []
 else:
     maybe_mongo = [MongoDB]
 
 
 #-----------------------------------------------------------------------------
 # Module level variables
 #-----------------------------------------------------------------------------
 
 
 #: The default config file name for this application
 default_config_file_name = u'ipcontroller_config.py'
 
 
 _description = """Start the IPython controller for parallel computing.
 
 The IPython controller provides a gateway between the IPython engines and
 clients. The controller needs to be started before the engines and can be
 configured using command line options or using a cluster directory. Cluster
 directories contain config, log and security files and are usually located in
 your ipython directory and named as "cluster_<profile>". See the `profile`
 and `profile_dir` options for details.
 """
 
 
 
 
 #-----------------------------------------------------------------------------
 # The main application
 #-----------------------------------------------------------------------------
 flags = {}
 flags.update(base_flags)
 flags.update({
     'usethreads' : ( {'IPControllerApp' : {'use_threads' : True}},
                     'Use threads instead of processes for the schedulers'),
     'sqlitedb' : ({'HubFactory' : {'db_class' : 'IPython.parallel.controller.sqlitedb.SQLiteDB'}},
                     'use the SQLiteDB backend'),
     'mongodb' : ({'HubFactory' : {'db_class' : 'IPython.parallel.controller.mongodb.MongoDB'}},
                     'use the MongoDB backend'),
     'dictdb' : ({'HubFactory' : {'db_class' : 'IPython.parallel.controller.dictdb.DictDB'}},
                     'use the in-memory DictDB backend'),
     'reuse' : ({'IPControllerApp' : {'reuse_files' : True}},
                     'reuse existing json connection files')
 })
 
-flags.update()
+flags.update(boolean_flag('secure', 'IPControllerApp.secure',
+    "Use HMAC digests for authentication of messages.",
+    "Don't authenticate messages."
+))
 
 class IPControllerApp(BaseParallelApplication):
 
     name = u'ipcontroller'
     description = _description
     config_file_name = Unicode(default_config_file_name)
     classes = [ProfileDir, StreamSession, HubFactory, TaskScheduler, HeartMonitor, SQLiteDB] + maybe_mongo
     
     # change default to True
     auto_create = Bool(True, config=True,
-        help="""Whether to create profile dir if it doesn't exist""")
+        help="""Whether to create profile dir if it doesn't exist.""")
     
     reuse_files = Bool(False, config=True,
-        help='Whether to reuse existing json connection files [default: False]'
+        help='Whether to reuse existing json connection files.'
     )
     secure = Bool(True, config=True,
-        help='Whether to use exec_keys for extra authentication [default: True]'
+        help='Whether to use HMAC digests for extra message authentication.'
     )
     ssh_server = Unicode(u'', config=True,
         help="""ssh url for clients to use when connecting to the Controller
         processes. It should be of the form: [user@]server[:port]. The
-        Controller\'s listening addresses must be accessible from the ssh server""",
+        Controller's listening addresses must be accessible from the ssh server""",
     )
     location = Unicode(u'', config=True,
         help="""The external IP or domain name of the Controller, used for disambiguating
         engine and client connections.""",
     )
     import_statements = List([], config=True,
         help="import statements to be run at startup.  Necessary in some environments"
     )
 
     use_threads = Bool(False, config=True,
         help='Use threads instead of processes for the schedulers',
         )
 
     # internal
     children = List()
     mq_class = Unicode('zmq.devices.ProcessMonitoredQueue')
 
     def _use_threads_changed(self, name, old, new):
         self.mq_class = 'zmq.devices.%sMonitoredQueue'%('Thread' if new else 'Process')
 
     aliases = Dict(dict(
         log_level = 'IPControllerApp.log_level',
         log_url = 'IPControllerApp.log_url',
         reuse_files = 'IPControllerApp.reuse_files',
         secure = 'IPControllerApp.secure',
         ssh = 'IPControllerApp.ssh_server',
         use_threads = 'IPControllerApp.use_threads',
         import_statements = 'IPControllerApp.import_statements',
         location = 'IPControllerApp.location',
 
         ident = 'StreamSession.session',
         user = 'StreamSession.username',
         exec_key = 'StreamSession.keyfile',
 
         url = 'HubFactory.url',
         ip = 'HubFactory.ip',
         transport = 'HubFactory.transport',
         port = 'HubFactory.regport',
 
         ping = 'HeartMonitor.period',
 
         scheme = 'TaskScheduler.scheme_name',
         hwm = 'TaskScheduler.hwm',
 
 
         profile = "BaseIPythonApplication.profile",
         profile_dir = 'ProfileDir.location',
         
     ))
     flags = Dict(flags)
     
 
     def save_connection_dict(self, fname, cdict):
         """save a connection dict to json file."""
         c = self.config
         url = cdict['url']
         location = cdict['location']
         if not location:
             try:
                 proto,ip,port = split_url(url)
             except AssertionError:
                 pass
             else:
                 location = socket.gethostbyname_ex(socket.gethostname())[2][-1]
             cdict['location'] = location
         fname = os.path.join(self.profile_dir.security_dir, fname)
         with open(fname, 'w') as f:
             f.write(json.dumps(cdict, indent=2))
         os.chmod(fname, stat.S_IRUSR|stat.S_IWUSR)
     
     def load_config_from_json(self):
         """load config from existing json connector files."""
         c = self.config
         # load from engine config
         with open(os.path.join(self.profile_dir.security_dir, 'ipcontroller-engine.json')) as f:
             cfg = json.loads(f.read())
         key = c.StreamSession.key = cfg['exec_key']
         xport,addr = cfg['url'].split('://')
         c.HubFactory.engine_transport = xport
         ip,ports = addr.split(':')
         c.HubFactory.engine_ip = ip
         c.HubFactory.regport = int(ports)
         self.location = cfg['location']
         
         # load client config
         with open(os.path.join(self.profile_dir.security_dir, 'ipcontroller-client.json')) as f:
             cfg = json.loads(f.read())
         assert key == cfg['exec_key'], "exec_key mismatch between engine and client keys"
         xport,addr = cfg['url'].split('://')
         c.HubFactory.client_transport = xport
         ip,ports = addr.split(':')
         c.HubFactory.client_ip = ip
         self.ssh_server = cfg['ssh']
         assert int(ports) == c.HubFactory.regport, "regport mismatch"
     
     def init_hub(self):
         c = self.config
         
         self.do_import_statements()
         reusing = self.reuse_files
         if reusing:
             try:
                 self.load_config_from_json()
             except (AssertionError,IOError):
                 reusing=False
         # check again, because reusing may have failed:
         if reusing:
             pass
         elif self.secure:
             key = str(uuid.uuid4())
             # keyfile = os.path.join(self.profile_dir.security_dir, self.exec_key)
             # with open(keyfile, 'w') as f:
             #     f.write(key)
             # os.chmod(keyfile, stat.S_IRUSR|stat.S_IWUSR)
             c.StreamSession.key = key
         else:
             key = c.StreamSession.key = ''
         
         try:
             self.factory = HubFactory(config=c, log=self.log)
             # self.start_logging()
             self.factory.init_hub()
         except:
             self.log.error("Couldn't construct the Controller", exc_info=True)
             self.exit(1)
         
         if not reusing:
             # save to new json config files
             f = self.factory
             cdict = {'exec_key' : key,
                     'ssh' : self.ssh_server,
                     'url' : "%s://%s:%s"%(f.client_transport, f.client_ip, f.regport),
                     'location' : self.location
                     }
             self.save_connection_dict('ipcontroller-client.json', cdict)
             edict = cdict
             edict['url']="%s://%s:%s"%((f.client_transport, f.client_ip, f.regport))
             self.save_connection_dict('ipcontroller-engine.json', edict)
 
     #
     def init_schedulers(self):
         children = self.children
         mq = import_item(str(self.mq_class))
         
         hub = self.factory
         # maybe_inproc = 'inproc://monitor' if self.use_threads else self.monitor_url
         # IOPub relay (in a Process)
         q = mq(zmq.PUB, zmq.SUB, zmq.PUB, 'N/A','iopub')
         q.bind_in(hub.client_info['iopub'])
         q.bind_out(hub.engine_info['iopub'])
         q.setsockopt_out(zmq.SUBSCRIBE, '')
         q.connect_mon(hub.monitor_url)
         q.daemon=True
         children.append(q)
 
         # Multiplexer Queue (in a Process)
         q = mq(zmq.XREP, zmq.XREP, zmq.PUB, 'in', 'out')
         q.bind_in(hub.client_info['mux'])
         q.setsockopt_in(zmq.IDENTITY, 'mux')
         q.bind_out(hub.engine_info['mux'])
         q.connect_mon(hub.monitor_url)
         q.daemon=True
         children.append(q)
 
         # Control Queue (in a Process)
         q = mq(zmq.XREP, zmq.XREP, zmq.PUB, 'incontrol', 'outcontrol')
         q.bind_in(hub.client_info['control'])
         q.setsockopt_in(zmq.IDENTITY, 'control')
         q.bind_out(hub.engine_info['control'])
         q.connect_mon(hub.monitor_url)
         q.daemon=True
         children.append(q)
         try:
             scheme = self.config.TaskScheduler.scheme_name
         except AttributeError:
             scheme = TaskScheduler.scheme_name.get_default_value()
         # Task Queue (in a Process)
         if scheme == 'pure':
             self.log.warn("task::using pure XREQ Task scheduler")
             q = mq(zmq.XREP, zmq.XREQ, zmq.PUB, 'intask', 'outtask')
             # q.setsockopt_out(zmq.HWM, hub.hwm)
             q.bind_in(hub.client_info['task'][1])
             q.setsockopt_in(zmq.IDENTITY, 'task')
             q.bind_out(hub.engine_info['task'])
             q.connect_mon(hub.monitor_url)
             q.daemon=True
             children.append(q)
         elif scheme == 'none':
             self.log.warn("task::using no Task scheduler")
 
         else:
             self.log.info("task::using Python %s Task scheduler"%scheme)
             sargs = (hub.client_info['task'][1], hub.engine_info['task'],
                                 hub.monitor_url, hub.client_info['notification'])
             kwargs = dict(logname='scheduler', loglevel=self.log_level,
                             log_url = self.log_url, config=dict(self.config))
             q = Process(target=launch_scheduler, args=sargs, kwargs=kwargs)
             q.daemon=True
             children.append(q)
 
     
     def save_urls(self):
         """save the registration urls to files."""
         c = self.config
         
         sec_dir = self.profile_dir.security_dir
         cf = self.factory
         
         with open(os.path.join(sec_dir, 'ipcontroller-engine.url'), 'w') as f:
             f.write("%s://%s:%s"%(cf.engine_transport, cf.engine_ip, cf.regport))
         
         with open(os.path.join(sec_dir, 'ipcontroller-client.url'), 'w') as f:
             f.write("%s://%s:%s"%(cf.client_transport, cf.client_ip, cf.regport))
         
     
     def do_import_statements(self):
         statements = self.import_statements
         for s in statements:
             try:
                 self.log.msg("Executing statement: '%s'" % s)
                 exec s in globals(), locals()
             except:
                 self.log.msg("Error running statement: %s" % s)
 
     def forward_logging(self):
         if self.log_url:
             self.log.info("Forwarding logging to %s"%self.log_url)
             context = zmq.Context.instance()
             lsock = context.socket(zmq.PUB)
             lsock.connect(self.log_url)
             handler = PUBHandler(lsock)
             self.log.removeHandler(self._log_handler)
             handler.root_topic = 'controller'
             handler.setLevel(self.log_level)
             self.log.addHandler(handler)
             self._log_handler = handler
     # #
     
     def initialize(self, argv=None):
         super(IPControllerApp, self).initialize(argv)
         self.forward_logging()
         self.init_hub()
         self.init_schedulers()
     
     def start(self):
         # Start the subprocesses:
         self.factory.start()
         child_procs = []
         for child in self.children:
             child.start()
             if isinstance(child, ProcessMonitoredQueue):
                 child_procs.append(child.launcher)
             elif isinstance(child, Process):
                 child_procs.append(child)
         if child_procs:
             signal_children(child_procs)
 
         self.write_pid_file(overwrite=True)
 
         try:
             self.factory.loop.start()
         except KeyboardInterrupt:
             self.log.critical("Interrupted, Exiting...\n")
             
 
 
 def launch_new_instance():
     """Create and run the IPython controller"""
     app = IPControllerApp.instance()
     app.initialize()
     app.start()
 
 
 if __name__ == '__main__':
     launch_new_instance()

IPython/parallel/controller/hub.py

@@ -1,1270 +1,1277 b''
 #!/usr/bin/env python
 """The IPython Controller Hub with 0MQ
 This is the master object that handles connections from engines and clients,
 and monitors traffic through the various queues.
 """
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 from __future__ import print_function
 
 import sys
 import time
 from datetime import datetime
 
 import zmq
 from zmq.eventloop import ioloop
 from zmq.eventloop.zmqstream import ZMQStream
 
 # internal:
 from IPython.utils.importstring import import_item
 from IPython.utils.traitlets import (
         HasTraits, Instance, Int, Unicode, Dict, Set, Tuple, CStr
         )
 
 from IPython.parallel import error, util
 from IPython.parallel.factory import RegistrationFactory, LoggingFactory
 
 from .heartmonitor import HeartMonitor
 
 #-----------------------------------------------------------------------------
 # Code
 #-----------------------------------------------------------------------------
 
 def _passer(*args, **kwargs):
     return
 
 def _printer(*args, **kwargs):
     print (args)
     print (kwargs)
 
 def empty_record():
     """Return an empty dict with all record keys."""
     return {
         'msg_id' : None,
         'header' : None,
         'content': None,
         'buffers': None,
         'submitted': None,
         'client_uuid' : None,
         'engine_uuid' : None,
         'started': None,
         'completed': None,
         'resubmitted': None,
         'result_header' : None,
         'result_content' : None,
         'result_buffers' : None,
         'queue' : None,
         'pyin' : None,
         'pyout': None,
         'pyerr': None,
         'stdout': '',
         'stderr': '',
     }
     
 def init_record(msg):
     """Initialize a TaskRecord based on a request."""
     header = msg['header']
     return {
         'msg_id' : header['msg_id'],
         'header' : header,
         'content': msg['content'],
         'buffers': msg['buffers'],
         'submitted': datetime.strptime(header['date'], util.ISO8601),
         'client_uuid' : None,
         'engine_uuid' : None,
         'started': None,
         'completed': None,
         'resubmitted': None,
         'result_header' : None,
         'result_content' : None,
         'result_buffers' : None,
         'queue' : None,
         'pyin' : None,
         'pyout': None,
         'pyerr': None,
         'stdout': '',
         'stderr': '',
     }
 
 
 class EngineConnector(HasTraits):
     """A simple object for accessing the various zmq connections of an object.
     Attributes are:
     id (int): engine ID
     uuid (str): uuid (unused?)
     queue (str): identity of queue's XREQ socket
     registration (str): identity of registration XREQ socket
     heartbeat (str): identity of heartbeat XREQ socket
     """
     id=Int(0)
     queue=CStr()
     control=CStr()
     registration=CStr()
     heartbeat=CStr()
     pending=Set()
 
 class HubFactory(RegistrationFactory):
     """The Configurable for setting up a Hub."""
     
     # port-pairs for monitoredqueues:
     hb = Tuple(Int,Int,config=True,
         help="""XREQ/SUB Port pair for Engine heartbeats""")
     def _hb_default(self):
         return tuple(util.select_random_ports(2))
 
     mux = Tuple(Int,Int,config=True,
         help="""Engine/Client Port pair for MUX queue""")
     
     def _mux_default(self):
         return tuple(util.select_random_ports(2))
     
     task = Tuple(Int,Int,config=True,
         help="""Engine/Client Port pair for Task queue""")
     def _task_default(self):
         return tuple(util.select_random_ports(2))
 
     control = Tuple(Int,Int,config=True,
         help="""Engine/Client Port pair for Control queue""")
     
     def _control_default(self):
         return tuple(util.select_random_ports(2))
 
     iopub = Tuple(Int,Int,config=True,
         help="""Engine/Client Port pair for IOPub relay""")
     
     def _iopub_default(self):
         return tuple(util.select_random_ports(2))
     
     # single ports:
     mon_port = Int(config=True,
         help="""Monitor (SUB) port for queue traffic""")
 
     def _mon_port_default(self):
         return util.select_random_ports(1)[0]
     
     notifier_port = Int(config=True,
         help="""PUB port for sending engine status notifications""")
 
     def _notifier_port_default(self):
         return util.select_random_ports(1)[0]
     
     engine_ip = Unicode('127.0.0.1', config=True,
         help="IP on which to listen for engine connections. [default: loopback]")
     engine_transport = Unicode('tcp', config=True,
         help="0MQ transport for engine connections. [default: tcp]")
     
     client_ip = Unicode('127.0.0.1', config=True,
         help="IP on which to listen for client connections. [default: loopback]")
     client_transport = Unicode('tcp', config=True,
         help="0MQ transport for client connections. [default : tcp]")
     
     monitor_ip = Unicode('127.0.0.1', config=True,
         help="IP on which to listen for monitor messages. [default: loopback]")
     monitor_transport = Unicode('tcp', config=True,
         help="0MQ transport for monitor messages. [default : tcp]")
     
     monitor_url = Unicode('')
     
     db_class = Unicode('IPython.parallel.controller.dictdb.DictDB', config=True,
         help="""The class to use for the DB backend""")
     
     # not configurable
     db = Instance('IPython.parallel.controller.dictdb.BaseDB')
     heartmonitor = Instance('IPython.parallel.controller.heartmonitor.HeartMonitor')
     
     def _ip_changed(self, name, old, new):
         self.engine_ip = new
         self.client_ip = new
         self.monitor_ip = new
         self._update_monitor_url()
     
     def _update_monitor_url(self):
         self.monitor_url = "%s://%s:%i"%(self.monitor_transport, self.monitor_ip, self.mon_port)
     
     def _transport_changed(self, name, old, new):
         self.engine_transport = new
         self.client_transport = new
         self.monitor_transport = new
         self._update_monitor_url()
         
     def __init__(self, **kwargs):
         super(HubFactory, self).__init__(**kwargs)
         self._update_monitor_url()
     
     
     def construct(self):
         self.init_hub()
     
     def start(self):
         self.heartmonitor.start()
         self.log.info("Heartmonitor started")
     
     def init_hub(self):
         """construct"""
         client_iface = "%s://%s:"%(self.client_transport, self.client_ip) + "%i"
         engine_iface = "%s://%s:"%(self.engine_transport, self.engine_ip) + "%i"
         
         ctx = self.context
         loop = self.loop
         
         # Registrar socket
         q = ZMQStream(ctx.socket(zmq.XREP), loop)
         q.bind(client_iface % self.regport)
         self.log.info("Hub listening on %s for registration."%(client_iface%self.regport))
         if self.client_ip != self.engine_ip:
             q.bind(engine_iface % self.regport)
             self.log.info("Hub listening on %s for registration."%(engine_iface%self.regport))
         
         ### Engine connections ###
 
         # heartbeat
         hpub = ctx.socket(zmq.PUB)
         hpub.bind(engine_iface % self.hb[0])
         hrep = ctx.socket(zmq.XREP)
         hrep.bind(engine_iface % self.hb[1])
         self.heartmonitor = HeartMonitor(loop=loop, pingstream=ZMQStream(hpub,loop), pongstream=ZMQStream(hrep,loop), 
                                 config=self.config)
 
         ### Client connections ###
         # Notifier socket
         n = ZMQStream(ctx.socket(zmq.PUB), loop)
         n.bind(client_iface%self.notifier_port)
 
         ### build and launch the queues ###
 
         # monitor socket
         sub = ctx.socket(zmq.SUB)
         sub.setsockopt(zmq.SUBSCRIBE, "")
         sub.bind(self.monitor_url)
         sub.bind('inproc://monitor')
         sub = ZMQStream(sub, loop)
         
         # connect the db
         self.log.info('Hub using DB backend: %r'%(self.db_class.split()[-1]))
         # cdir = self.config.Global.cluster_dir
         self.db = import_item(str(self.db_class))(session=self.session.session, config=self.config)
         time.sleep(.25)
         try:
             scheme = self.config.TaskScheduler.scheme_name
         except AttributeError:
             from .scheduler import TaskScheduler
             scheme = TaskScheduler.scheme_name.get_default_value()
         # build connection dicts
         self.engine_info = {
             'control' : engine_iface%self.control[1],
             'mux': engine_iface%self.mux[1],
             'heartbeat': (engine_iface%self.hb[0], engine_iface%self.hb[1]),
             'task' : engine_iface%self.task[1],
             'iopub' : engine_iface%self.iopub[1],
             # 'monitor' : engine_iface%self.mon_port,
             }
 
         self.client_info = {
             'control' : client_iface%self.control[0],
             'mux': client_iface%self.mux[0],
             'task' : (scheme, client_iface%self.task[0]),
             'iopub' : client_iface%self.iopub[0],
             'notification': client_iface%self.notifier_port
             }
         self.log.debug("Hub engine addrs: %s"%self.engine_info)
         self.log.debug("Hub client addrs: %s"%self.client_info)
 
         # resubmit stream
         r = ZMQStream(ctx.socket(zmq.XREQ), loop)
         url = util.disambiguate_url(self.client_info['task'][-1])
         r.setsockopt(zmq.IDENTITY, self.session.session)
         r.connect(url)
 
         self.hub = Hub(loop=loop, session=self.session, monitor=sub, heartmonitor=self.heartmonitor,
                 query=q, notifier=n, resubmit=r, db=self.db,
                 engine_info=self.engine_info, client_info=self.client_info,
                 logname=self.log.name)
     
 
 class Hub(LoggingFactory):
     """The IPython Controller Hub with 0MQ connections
     
     Parameters
     ==========
     loop: zmq IOLoop instance
     session: StreamSession object
     <removed> context: zmq context for creating new connections (?)
     queue: ZMQStream for monitoring the command queue (SUB)
     query: ZMQStream for engine registration and client queries requests (XREP)
     heartbeat: HeartMonitor object checking the pulse of the engines
     notifier: ZMQStream for broadcasting engine registration changes (PUB)
     db: connection to db for out of memory logging of commands
                 NotImplemented
     engine_info: dict of zmq connection information for engines to connect
                 to the queues.
     client_info: dict of zmq connection information for engines to connect
                 to the queues.
     """
     # internal data structures:
     ids=Set() # engine IDs
     keytable=Dict()
     by_ident=Dict()
     engines=Dict()
     clients=Dict()
     hearts=Dict()
     pending=Set()
     queues=Dict()  # pending msg_ids keyed by engine_id
     tasks=Dict() # pending msg_ids submitted as tasks, keyed by client_id
     completed=Dict() # completed msg_ids keyed by engine_id
     all_completed=Set() # completed msg_ids keyed by engine_id
     dead_engines=Set() # completed msg_ids keyed by engine_id
     unassigned=Set() # set of task msg_ds not yet assigned a destination
     incoming_registrations=Dict()
     registration_timeout=Int()
     _idcounter=Int(0)
     
     # objects from constructor:
     loop=Instance(ioloop.IOLoop)
     query=Instance(ZMQStream)
     monitor=Instance(ZMQStream)
     notifier=Instance(ZMQStream)
     resubmit=Instance(ZMQStream)
     heartmonitor=Instance(HeartMonitor)
     db=Instance(object)
     client_info=Dict()
     engine_info=Dict()
     
     
     def __init__(self, **kwargs):
         """
         # universal:
         loop: IOLoop for creating future connections
         session: streamsession for sending serialized data
         # engine:
         queue: ZMQStream for monitoring queue messages
         query: ZMQStream for engine+client registration and client requests
         heartbeat: HeartMonitor object for tracking engines
         # extra:
         db: ZMQStream for db connection (NotImplemented)
         engine_info: zmq address/protocol dict for engine connections
         client_info: zmq address/protocol dict for client connections
         """
         
         super(Hub, self).__init__(**kwargs)
         self.registration_timeout = max(5000, 2*self.heartmonitor.period)
         
         # validate connection dicts:
         for k,v in self.client_info.iteritems():
             if k == 'task':
                 util.validate_url_container(v[1])
             else:
                 util.validate_url_container(v)
         # util.validate_url_container(self.client_info)
         util.validate_url_container(self.engine_info)
         
         # register our callbacks
         self.query.on_recv(self.dispatch_query)
         self.monitor.on_recv(self.dispatch_monitor_traffic)
         
         self.heartmonitor.add_heart_failure_handler(self.handle_heart_failure)
         self.heartmonitor.add_new_heart_handler(self.handle_new_heart)
         
         self.monitor_handlers = { 'in' : self.save_queue_request,
                                 'out': self.save_queue_result,
                                 'intask': self.save_task_request,
                                 'outtask': self.save_task_result,
                                 'tracktask': self.save_task_destination,
                                 'incontrol': _passer,
                                 'outcontrol': _passer,
                                 'iopub': self.save_iopub_message,
         }
         
         self.query_handlers = {'queue_request': self.queue_status,
                                 'result_request': self.get_results,
                                 'history_request': self.get_history,
                                 'db_request': self.db_query,
                                 'purge_request': self.purge_results,
                                 'load_request': self.check_load,
                                 'resubmit_request': self.resubmit_task,
                                 'shutdown_request': self.shutdown_request,
                                 'registration_request' : self.register_engine,
                                 'unregistration_request' : self.unregister_engine,
                                 'connection_request': self.connection_request,
         }
         
         # ignore resubmit replies
         self.resubmit.on_recv(lambda msg: None, copy=False)
 
         self.log.info("hub::created hub")
     
     @property
     def _next_id(self):
         """gemerate a new ID.
         
         No longer reuse old ids, just count from 0."""
         newid = self._idcounter
         self._idcounter += 1
         return newid
         # newid = 0
         # incoming = [id[0] for id in self.incoming_registrations.itervalues()]
         # # print newid, self.ids, self.incoming_registrations
         # while newid in self.ids or newid in incoming:
         #     newid += 1
         # return newid
     
     #-----------------------------------------------------------------------------
     # message validation
     #-----------------------------------------------------------------------------
     
     def _validate_targets(self, targets):
         """turn any valid targets argument into a list of integer ids"""
         if targets is None:
             # default to all
             targets = self.ids
             
         if isinstance(targets, (int,str,unicode)):
             # only one target specified
             targets = [targets]
         _targets = []
         for t in targets:
             # map raw identities to ids
             if isinstance(t, (str,unicode)):
                 t = self.by_ident.get(t, t)
             _targets.append(t)
         targets = _targets
         bad_targets = [ t for t in targets if t not in self.ids ]
         if bad_targets:
             raise IndexError("No Such Engine: %r"%bad_targets)
         if not targets:
             raise IndexError("No Engines Registered")
         return targets
     
     #-----------------------------------------------------------------------------
     # dispatch methods (1 per stream)
     #-----------------------------------------------------------------------------
     
     
     def dispatch_monitor_traffic(self, msg):
         """all ME and Task queue messages come through here, as well as
         IOPub traffic."""
         self.log.debug("monitor traffic: %r"%msg[:2])
         switch = msg[0]
         try:
             idents, msg = self.session.feed_identities(msg[1:])
         except ValueError:
             idents=[]
         if not idents:
             self.log.error("Bad Monitor Message: %r"%msg)
             return
         handler = self.monitor_handlers.get(switch, None)
         if handler is not None:
             handler(idents, msg)
         else:
             self.log.error("Invalid monitor topic: %r"%switch)
         
     
     def dispatch_query(self, msg):
         """Route registration requests and queries from clients."""
-        idents, msg = self.session.feed_identities(msg)
+        try:
+            idents, msg = self.session.feed_identities(msg)
+        except ValueError:
+            idents = []
         if not idents:
             self.log.error("Bad Query Message: %r"%msg)
             return
         client_id = idents[0]
         try:
             msg = self.session.unpack_message(msg, content=True)
-        except:
+        except Exception:
             content = error.wrap_exception()
             self.log.error("Bad Query Message: %r"%msg, exc_info=True)
             self.session.send(self.query, "hub_error", ident=client_id, 
                     content=content)
             return
-        
+        print( idents, msg)
         # print client_id, header, parent, content
         #switch on message type:
         msg_type = msg['msg_type']
         self.log.info("client::client %r requested %r"%(client_id, msg_type))
         handler = self.query_handlers.get(msg_type, None)
         try:
             assert handler is not None, "Bad Message Type: %r"%msg_type
         except:
             content = error.wrap_exception()
             self.log.error("Bad Message Type: %r"%msg_type, exc_info=True)
             self.session.send(self.query, "hub_error", ident=client_id, 
                     content=content)
             return
 
         else:
             handler(idents, msg)
             
     def dispatch_db(self, msg):
         """"""
         raise NotImplementedError
     
     #---------------------------------------------------------------------------
     # handler methods (1 per event)
     #---------------------------------------------------------------------------
     
     #----------------------- Heartbeat --------------------------------------
     
     def handle_new_heart(self, heart):
         """handler to attach to heartbeater.
         Called when a new heart starts to beat.
         Triggers completion of registration."""
         self.log.debug("heartbeat::handle_new_heart(%r)"%heart)
         if heart not in self.incoming_registrations:
             self.log.info("heartbeat::ignoring new heart: %r"%heart)
         else:
             self.finish_registration(heart)
         
     
     def handle_heart_failure(self, heart):
         """handler to attach to heartbeater.
         called when a previously registered heart fails to respond to beat request.
         triggers unregistration"""
         self.log.debug("heartbeat::handle_heart_failure(%r)"%heart)
         eid = self.hearts.get(heart, None)
         queue = self.engines[eid].queue
         if eid is None:
             self.log.info("heartbeat::ignoring heart failure %r"%heart)
         else:
             self.unregister_engine(heart, dict(content=dict(id=eid, queue=queue)))
     
     #----------------------- MUX Queue Traffic ------------------------------
     
     def save_queue_request(self, idents, msg):
         if len(idents) < 2:
             self.log.error("invalid identity prefix: %r"%idents)
             return
         queue_id, client_id = idents[:2]
         try:
             msg = self.session.unpack_message(msg, content=False)
         except Exception:
             self.log.error("queue::client %r sent invalid message to %r: %r"%(client_id, queue_id, msg), exc_info=True)
             return
         
         eid = self.by_ident.get(queue_id, None)
         if eid is None:
             self.log.error("queue::target %r not registered"%queue_id)
             self.log.debug("queue::    valid are: %r"%(self.by_ident.keys()))
             return
             
         header = msg['header']
         msg_id = header['msg_id']
         record = init_record(msg)
         record['engine_uuid'] = queue_id
         record['client_uuid'] = client_id
         record['queue'] = 'mux'
 
         try:
             # it's posible iopub arrived first:
             existing = self.db.get_record(msg_id)
             for key,evalue in existing.iteritems():
                 rvalue = record.get(key, None)
                 if evalue and rvalue and evalue != rvalue:
                     self.log.warn("conflicting initial state for record: %r:%r <%r> %r"%(msg_id, rvalue, key, evalue))
                 elif evalue and not rvalue:
                     record[key] = evalue
             self.db.update_record(msg_id, record)
         except KeyError:
             self.db.add_record(msg_id, record)
         
         self.pending.add(msg_id)
         self.queues[eid].append(msg_id)
     
     def save_queue_result(self, idents, msg):
         if len(idents) < 2:
             self.log.error("invalid identity prefix: %r"%idents)
             return
             
         client_id, queue_id = idents[:2]
         try:
             msg = self.session.unpack_message(msg, content=False)
         except Exception:
             self.log.error("queue::engine %r sent invalid message to %r: %r"%(
                     queue_id,client_id, msg), exc_info=True)
             return
         
         eid = self.by_ident.get(queue_id, None)
         if eid is None:
             self.log.error("queue::unknown engine %r is sending a reply: "%queue_id)
             return
         
         parent = msg['parent_header']
         if not parent:
             return
         msg_id = parent['msg_id']
         if msg_id in self.pending:
             self.pending.remove(msg_id)
             self.all_completed.add(msg_id)
             self.queues[eid].remove(msg_id)
             self.completed[eid].append(msg_id)
         elif msg_id not in self.all_completed:
             # it could be a result from a dead engine that died before delivering the
             # result
             self.log.warn("queue:: unknown msg finished %r"%msg_id)
             return
         # update record anyway, because the unregistration could have been premature
         rheader = msg['header']
         completed = datetime.strptime(rheader['date'], util.ISO8601)
         started = rheader.get('started', None)
         if started is not None:
             started = datetime.strptime(started, util.ISO8601)
         result = {
             'result_header' : rheader,
             'result_content': msg['content'],
             'started' : started,
             'completed' : completed
         }
 
         result['result_buffers'] = msg['buffers']
         try:
             self.db.update_record(msg_id, result)
         except Exception:
             self.log.error("DB Error updating record %r"%msg_id, exc_info=True)
         
             
     #--------------------- Task Queue Traffic ------------------------------
     
     def save_task_request(self, idents, msg):
         """Save the submission of a task."""
         client_id = idents[0]
         
         try:
             msg = self.session.unpack_message(msg, content=False)
         except Exception:
             self.log.error("task::client %r sent invalid task message: %r"%(
                     client_id, msg), exc_info=True)
             return
         record = init_record(msg)
 
         record['client_uuid'] = client_id
         record['queue'] = 'task'
         header = msg['header']
         msg_id = header['msg_id']
         self.pending.add(msg_id)
         self.unassigned.add(msg_id)
         try:
             # it's posible iopub arrived first:
             existing = self.db.get_record(msg_id)
             if existing['resubmitted']:
                 for key in ('submitted', 'client_uuid', 'buffers'):
                     # don't clobber these keys on resubmit
                     # submitted and client_uuid should be different
                     # and buffers might be big, and shouldn't have changed
                     record.pop(key)
                     # still check content,header which should not change
                     # but are not expensive to compare as buffers
 
             for key,evalue in existing.iteritems():
                 if key.endswith('buffers'):
                     # don't compare buffers
                     continue
                 rvalue = record.get(key, None)
                 if evalue and rvalue and evalue != rvalue:
                     self.log.warn("conflicting initial state for record: %r:%r <%r> %r"%(msg_id, rvalue, key, evalue))
                 elif evalue and not rvalue:
                     record[key] = evalue
             self.db.update_record(msg_id, record)
         except KeyError:
             self.db.add_record(msg_id, record)
         except Exception:
             self.log.error("DB Error saving task request %r"%msg_id, exc_info=True)
     
     def save_task_result(self, idents, msg):
         """save the result of a completed task."""
         client_id = idents[0]
         try:
             msg = self.session.unpack_message(msg, content=False)
         except Exception:
             self.log.error("task::invalid task result message send to %r: %r"%(
                     client_id, msg), exc_info=True)
             return
         
         parent = msg['parent_header']
         if not parent:
             # print msg
             self.log.warn("Task %r had no parent!"%msg)
             return
         msg_id = parent['msg_id']
         if msg_id in self.unassigned:
             self.unassigned.remove(msg_id)
         
         header = msg['header']
         engine_uuid = header.get('engine', None)
         eid = self.by_ident.get(engine_uuid, None)
         
         if msg_id in self.pending:
             self.pending.remove(msg_id)
             self.all_completed.add(msg_id)
             if eid is not None:
                 self.completed[eid].append(msg_id)
                 if msg_id in self.tasks[eid]:
                     self.tasks[eid].remove(msg_id)
             completed = datetime.strptime(header['date'], util.ISO8601)
             started = header.get('started', None)
             if started is not None:
                 started = datetime.strptime(started, util.ISO8601)
             result = {
                 'result_header' : header,
                 'result_content': msg['content'],
                 'started' : started,
                 'completed' : completed,
                 'engine_uuid': engine_uuid
             }
 
             result['result_buffers'] = msg['buffers']
             try:
                 self.db.update_record(msg_id, result)
             except Exception:
                 self.log.error("DB Error saving task request %r"%msg_id, exc_info=True)
             
         else:
             self.log.debug("task::unknown task %r finished"%msg_id)
     
     def save_task_destination(self, idents, msg):
         try:
             msg = self.session.unpack_message(msg, content=True)
         except Exception:
             self.log.error("task::invalid task tracking message", exc_info=True)
             return
         content = msg['content']
         # print (content)
         msg_id = content['msg_id']
         engine_uuid = content['engine_id']
         eid = self.by_ident[engine_uuid]
         
         self.log.info("task::task %r arrived on %r"%(msg_id, eid))
         if msg_id in self.unassigned:
             self.unassigned.remove(msg_id)
         # else:
         #     self.log.debug("task::task %r not listed as MIA?!"%(msg_id))
         
         self.tasks[eid].append(msg_id)
         # self.pending[msg_id][1].update(received=datetime.now(),engine=(eid,engine_uuid))
         try:
             self.db.update_record(msg_id, dict(engine_uuid=engine_uuid))
         except Exception:
             self.log.error("DB Error saving task destination %r"%msg_id, exc_info=True)
             
     
     def mia_task_request(self, idents, msg):
         raise NotImplementedError
         client_id = idents[0]
         # content = dict(mia=self.mia,status='ok')
         # self.session.send('mia_reply', content=content, idents=client_id)
         
         
     #--------------------- IOPub Traffic ------------------------------
     
     def save_iopub_message(self, topics, msg):
         """save an iopub message into the db"""
         # print (topics)
         try:
             msg = self.session.unpack_message(msg, content=True)
         except Exception:
             self.log.error("iopub::invalid IOPub message", exc_info=True)
             return
         
         parent = msg['parent_header']
         if not parent:
             self.log.error("iopub::invalid IOPub message: %r"%msg)
             return
         msg_id = parent['msg_id']
         msg_type = msg['msg_type']
         content = msg['content']
         
         # ensure msg_id is in db
         try:
             rec = self.db.get_record(msg_id)
         except KeyError:
             rec = empty_record()
             rec['msg_id'] = msg_id
             self.db.add_record(msg_id, rec)
         # stream
         d = {}
         if msg_type == 'stream':
             name = content['name']
             s = rec[name] or ''
             d[name] = s + content['data']
             
         elif msg_type == 'pyerr':
             d['pyerr'] = content
         elif msg_type == 'pyin':
             d['pyin'] = content['code']
         else:
             d[msg_type] = content.get('data', '')
         
         try:
             self.db.update_record(msg_id, d)
         except Exception:
             self.log.error("DB Error saving iopub message %r"%msg_id, exc_info=True)
         
     
             
     #-------------------------------------------------------------------------
     # Registration requests
     #-------------------------------------------------------------------------
         
     def connection_request(self, client_id, msg):
         """Reply with connection addresses for clients."""
         self.log.info("client::client %r connected"%client_id)
         content = dict(status='ok')
         content.update(self.client_info)
         jsonable = {}
         for k,v in self.keytable.iteritems():
             if v not in self.dead_engines:
                 jsonable[str(k)] = v
         content['engines'] = jsonable
         self.session.send(self.query, 'connection_reply', content, parent=msg, ident=client_id)
     
     def register_engine(self, reg, msg):
         """Register a new engine."""
         content = msg['content']
         try:
             queue = content['queue']
         except KeyError:
             self.log.error("registration::queue not specified", exc_info=True)
             return
         heart = content.get('heartbeat', None)
         """register a new engine, and create the socket(s) necessary"""
         eid = self._next_id
         # print (eid, queue, reg, heart)
         
         self.log.debug("registration::register_engine(%i, %r, %r, %r)"%(eid, queue, reg, heart))
         
         content = dict(id=eid,status='ok')
         content.update(self.engine_info)
         # check if requesting available IDs:
         if queue in self.by_ident:
             try:
                 raise KeyError("queue_id %r in use"%queue)
             except:
                 content = error.wrap_exception()
                 self.log.error("queue_id %r in use"%queue, exc_info=True)
         elif heart in self.hearts: # need to check unique hearts?
             try:
                 raise KeyError("heart_id %r in use"%heart)
             except:
                 self.log.error("heart_id %r in use"%heart, exc_info=True)
                 content = error.wrap_exception()
         else:
             for h, pack in self.incoming_registrations.iteritems():
                 if heart == h:
                     try:
                         raise KeyError("heart_id %r in use"%heart)
                     except:
                         self.log.error("heart_id %r in use"%heart, exc_info=True)
                         content = error.wrap_exception()
                     break
                 elif queue == pack[1]:
                     try:
                         raise KeyError("queue_id %r in use"%queue)
                     except:
                         self.log.error("queue_id %r in use"%queue, exc_info=True)
                         content = error.wrap_exception()
                     break
         
         msg = self.session.send(self.query, "registration_reply", 
                 content=content, 
                 ident=reg)
         
         if content['status'] == 'ok':
             if heart in self.heartmonitor.hearts:
                 # already beating
                 self.incoming_registrations[heart] = (eid,queue,reg[0],None)
                 self.finish_registration(heart)
             else:
                 purge = lambda : self._purge_stalled_registration(heart)
                 dc = ioloop.DelayedCallback(purge, self.registration_timeout, self.loop)
                 dc.start()
                 self.incoming_registrations[heart] = (eid,queue,reg[0],dc)
         else:
             self.log.error("registration::registration %i failed: %r"%(eid, content['evalue']))
         return eid
     
     def unregister_engine(self, ident, msg):
         """Unregister an engine that explicitly requested to leave."""
         try:
             eid = msg['content']['id']
         except:
             self.log.error("registration::bad engine id for unregistration: %r"%ident, exc_info=True)
             return
         self.log.info("registration::unregister_engine(%r)"%eid)
         # print (eid)
         uuid = self.keytable[eid]
         content=dict(id=eid, queue=uuid)
         self.dead_engines.add(uuid)
         # self.ids.remove(eid)
         # uuid = self.keytable.pop(eid)
         # 
         # ec = self.engines.pop(eid)
         # self.hearts.pop(ec.heartbeat)
         # self.by_ident.pop(ec.queue)
         # self.completed.pop(eid)
         handleit = lambda : self._handle_stranded_msgs(eid, uuid)
         dc = ioloop.DelayedCallback(handleit, self.registration_timeout, self.loop)
         dc.start()
         ############## TODO: HANDLE IT ################
         
         if self.notifier:
             self.session.send(self.notifier, "unregistration_notification", content=content)
     
     def _handle_stranded_msgs(self, eid, uuid):
         """Handle messages known to be on an engine when the engine unregisters.
         
         It is possible that this will fire prematurely - that is, an engine will
         go down after completing a result, and the client will be notified
         that the result failed and later receive the actual result.
         """
         
         outstanding = self.queues[eid]
         
         for msg_id in outstanding:
             self.pending.remove(msg_id)
             self.all_completed.add(msg_id)
             try:
                 raise error.EngineError("Engine %r died while running task %r"%(eid, msg_id))
             except:
                 content = error.wrap_exception()
             # build a fake header:
             header = {}
             header['engine'] = uuid
             header['date'] = datetime.now()
             rec = dict(result_content=content, result_header=header, result_buffers=[])
             rec['completed'] = header['date']
             rec['engine_uuid'] = uuid
             try:
                 self.db.update_record(msg_id, rec)
             except Exception:
                 self.log.error("DB Error handling stranded msg %r"%msg_id, exc_info=True)
                 
     
     def finish_registration(self, heart):
         """Second half of engine registration, called after our HeartMonitor
         has received a beat from the Engine's Heart."""
         try: 
             (eid,queue,reg,purge) = self.incoming_registrations.pop(heart)
         except KeyError:
             self.log.error("registration::tried to finish nonexistant registration", exc_info=True)
             return
         self.log.info("registration::finished registering engine %i:%r"%(eid,queue))
         if purge is not None:
             purge.stop()
         control = queue
         self.ids.add(eid)
         self.keytable[eid] = queue
         self.engines[eid] = EngineConnector(id=eid, queue=queue, registration=reg, 
                                     control=control, heartbeat=heart)
         self.by_ident[queue] = eid
         self.queues[eid] = list()
         self.tasks[eid] = list()
         self.completed[eid] = list()
         self.hearts[heart] = eid
         content = dict(id=eid, queue=self.engines[eid].queue)
         if self.notifier:
             self.session.send(self.notifier, "registration_notification", content=content)
         self.log.info("engine::Engine Connected: %i"%eid)
     
     def _purge_stalled_registration(self, heart):
         if heart in self.incoming_registrations:
             eid = self.incoming_registrations.pop(heart)[0]
             self.log.info("registration::purging stalled registration: %i"%eid)
         else:
             pass
             
     #-------------------------------------------------------------------------
     # Client Requests
     #-------------------------------------------------------------------------
     
     def shutdown_request(self, client_id, msg):
         """handle shutdown request."""
         self.session.send(self.query, 'shutdown_reply', content={'status': 'ok'}, ident=client_id)
         # also notify other clients of shutdown
         self.session.send(self.notifier, 'shutdown_notice', content={'status': 'ok'})
         dc = ioloop.DelayedCallback(lambda : self._shutdown(), 1000, self.loop)
         dc.start()
     
     def _shutdown(self):
         self.log.info("hub::hub shutting down.")
         time.sleep(0.1)
         sys.exit(0)
         
     
     def check_load(self, client_id, msg):
         content = msg['content']
         try:
             targets = content['targets']
             targets = self._validate_targets(targets)
         except:
             content = error.wrap_exception()
             self.session.send(self.query, "hub_error", 
                     content=content, ident=client_id)
             return
         
         content = dict(status='ok')
         # loads = {}
         for t in targets:
             content[bytes(t)] = len(self.queues[t])+len(self.tasks[t])
         self.session.send(self.query, "load_reply", content=content, ident=client_id)
             
     
     def queue_status(self, client_id, msg):
         """Return the Queue status of one or more targets.
         if verbose: return the msg_ids
         else: return len of each type.
         keys: queue (pending MUX jobs)
             tasks (pending Task jobs)
             completed (finished jobs from both queues)"""
         content = msg['content']
         targets = content['targets']
         try:
             targets = self._validate_targets(targets)
         except:
             content = error.wrap_exception()
             self.session.send(self.query, "hub_error", 
                     content=content, ident=client_id)
             return
         verbose = content.get('verbose', False)
         content = dict(status='ok')
         for t in targets:
             queue = self.queues[t]
             completed = self.completed[t]
             tasks = self.tasks[t]
             if not verbose:
                 queue = len(queue)
                 completed = len(completed)
                 tasks = len(tasks)
             content[bytes(t)] = {'queue': queue, 'completed': completed , 'tasks': tasks}
         content['unassigned'] = list(self.unassigned) if verbose else len(self.unassigned)
         
         self.session.send(self.query, "queue_reply", content=content, ident=client_id)
     
     def purge_results(self, client_id, msg):
         """Purge results from memory. This method is more valuable before we move
         to a DB based message storage mechanism."""
         content = msg['content']
         msg_ids = content.get('msg_ids', [])
         reply = dict(status='ok')
         if msg_ids == 'all':
             try:
                 self.db.drop_matching_records(dict(completed={'$ne':None}))
             except Exception:
                 reply = error.wrap_exception()
         else:
             pending = filter(lambda m: m in self.pending, msg_ids)
             if pending:
                 try:
                     raise IndexError("msg pending: %r"%pending[0])
                 except:
                     reply = error.wrap_exception()
             else:
                 try:
                     self.db.drop_matching_records(dict(msg_id={'$in':msg_ids}))
                 except Exception:
                     reply = error.wrap_exception()
 
             if reply['status'] == 'ok':
                 eids = content.get('engine_ids', [])
                 for eid in eids:
                     if eid not in self.engines:
                         try:
                             raise IndexError("No such engine: %i"%eid)
                         except:
                             reply = error.wrap_exception()
                         break
                     msg_ids = self.completed.pop(eid)
                     uid = self.engines[eid].queue
                     try:
                         self.db.drop_matching_records(dict(engine_uuid=uid, completed={'$ne':None}))
                     except Exception:
                         reply = error.wrap_exception()
                         break
         
         self.session.send(self.query, 'purge_reply', content=reply, ident=client_id)
     
     def resubmit_task(self, client_id, msg):
         """Resubmit one or more tasks."""
         def finish(reply):
             self.session.send(self.query, 'resubmit_reply', content=reply, ident=client_id)
 
         content = msg['content']
         msg_ids = content['msg_ids']
         reply = dict(status='ok')
         try:
             records = self.db.find_records({'msg_id' : {'$in' : msg_ids}}, keys=[
                 'header', 'content', 'buffers'])
         except Exception:
             self.log.error('db::db error finding tasks to resubmit', exc_info=True)
             return finish(error.wrap_exception())
 
         # validate msg_ids
         found_ids = [ rec['msg_id'] for rec in records ]
         invalid_ids = filter(lambda m: m in self.pending, found_ids)
         if len(records) > len(msg_ids):
             try:
                 raise RuntimeError("DB appears to be in an inconsistent state."
                     "More matching records were found than should exist")
             except Exception:
                 return finish(error.wrap_exception())
         elif len(records) < len(msg_ids):
             missing = [ m for m in msg_ids if m not in found_ids ]
             try:
                 raise KeyError("No such msg(s): %r"%missing)
             except KeyError:
                 return finish(error.wrap_exception())
         elif invalid_ids:
             msg_id = invalid_ids[0]
             try:
                 raise ValueError("Task %r appears to be inflight"%(msg_id))
             except Exception:
                 return finish(error.wrap_exception())
 
         # clear the existing records
+        now = datetime.now()
         rec = empty_record()
         map(rec.pop, ['msg_id', 'header', 'content', 'buffers', 'submitted'])
-        rec['resubmitted'] = datetime.now()
+        rec['resubmitted'] = now
         rec['queue'] = 'task'
         rec['client_uuid'] = client_id[0]
         try:
             for msg_id in msg_ids:
                 self.all_completed.discard(msg_id)
                 self.db.update_record(msg_id, rec)
         except Exception:
             self.log.error('db::db error upating record', exc_info=True)
             reply = error.wrap_exception()
         else:
             # send the messages
+            now_s = now.strftime(util.ISO8601)
             for rec in records:
                 header = rec['header']
+                # include resubmitted in header to prevent digest collision
+                header['resubmitted'] = now_s
                 msg = self.session.msg(header['msg_type'])
                 msg['content'] = rec['content']
                 msg['header'] = header
                 msg['msg_id'] = rec['msg_id']
                 self.session.send(self.resubmit, msg, buffers=rec['buffers'])
 
         finish(dict(status='ok'))
 
     
     def _extract_record(self, rec):
         """decompose a TaskRecord dict into subsection of reply for get_result"""
         io_dict = {}
         for key in 'pyin pyout pyerr stdout stderr'.split():
                 io_dict[key] = rec[key]
         content = { 'result_content': rec['result_content'],
                             'header': rec['header'],
                             'result_header' : rec['result_header'],
                             'io' : io_dict,
                           }
         if rec['result_buffers']:
             buffers = map(str, rec['result_buffers'])
         else:
             buffers = []
         
         return content, buffers
     
     def get_results(self, client_id, msg):
         """Get the result of 1 or more messages."""
         content = msg['content']
         msg_ids = sorted(set(content['msg_ids']))
         statusonly = content.get('status_only', False)
         pending = []
         completed = []
         content = dict(status='ok')
         content['pending'] = pending
         content['completed'] = completed
         buffers = []
         if not statusonly:
             try:
                 matches = self.db.find_records(dict(msg_id={'$in':msg_ids}))
                 # turn match list into dict, for faster lookup
                 records = {}
                 for rec in matches:
                     records[rec['msg_id']] = rec
             except Exception:
                 content = error.wrap_exception()
                 self.session.send(self.query, "result_reply", content=content, 
                                                     parent=msg, ident=client_id)
                 return
         else:
             records = {}
         for msg_id in msg_ids:
             if msg_id in self.pending:
                 pending.append(msg_id)
             elif msg_id in self.all_completed:
                 completed.append(msg_id)
                 if not statusonly:
                     c,bufs = self._extract_record(records[msg_id])
                     content[msg_id] = c
                     buffers.extend(bufs)
             elif msg_id in records:
                 if rec['completed']:
                     completed.append(msg_id)
                     c,bufs = self._extract_record(records[msg_id])
                     content[msg_id] = c
                     buffers.extend(bufs)
                 else:
                     pending.append(msg_id)
             else:
                 try:
                     raise KeyError('No such message: '+msg_id)
                 except:
                     content = error.wrap_exception()
                 break
         self.session.send(self.query, "result_reply", content=content, 
                                             parent=msg, ident=client_id,
                                             buffers=buffers)
 
     def get_history(self, client_id, msg):
         """Get a list of all msg_ids in our DB records"""
         try:
             msg_ids = self.db.get_history()
         except Exception as e:
             content = error.wrap_exception()
         else:
             content = dict(status='ok', history=msg_ids)
         
         self.session.send(self.query, "history_reply", content=content, 
                                             parent=msg, ident=client_id)
 
     def db_query(self, client_id, msg):
         """Perform a raw query on the task record database."""
         content = msg['content']
         query = content.get('query', {})
         keys = content.get('keys', None)
         query = util.extract_dates(query)
         buffers = []
         empty = list()
         
         try:
             records = self.db.find_records(query, keys)
         except Exception as e:
             content = error.wrap_exception()
         else:
             # extract buffers from reply content:
             if keys is not None:
                 buffer_lens = [] if 'buffers' in keys else None
                 result_buffer_lens = [] if 'result_buffers' in keys else None
             else:
                 buffer_lens = []
                 result_buffer_lens = []
             
             for rec in records:
                 # buffers may be None, so double check
                 if buffer_lens is not None:
                     b = rec.pop('buffers', empty) or empty
                     buffer_lens.append(len(b))
                     buffers.extend(b)
                 if result_buffer_lens is not None:
                     rb = rec.pop('result_buffers', empty) or empty
                     result_buffer_lens.append(len(rb))
                     buffers.extend(rb)
             content = dict(status='ok', records=records, buffer_lens=buffer_lens,
                                     result_buffer_lens=result_buffer_lens)
         
         self.session.send(self.query, "db_reply", content=content, 
                                             parent=msg, ident=client_id,
                                             buffers=buffers)
 

IPython/parallel/controller/scheduler.py

@@ -1,679 +1,687 b''
 """The Python scheduler for rich scheduling.
 
 The Pure ZMQ scheduler does not allow routing schemes other than LRU,
 nor does it check msg_id DAG dependencies. For those, a slightly slower
 Python Scheduler exists.
 """
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010-2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #----------------------------------------------------------------------
 # Imports
 #----------------------------------------------------------------------
 
 from __future__ import print_function
 
 import logging
 import sys
 
 from datetime import datetime, timedelta
 from random import randint, random
 from types import FunctionType
 
 try:
     import numpy
 except ImportError:
     numpy = None
 
 import zmq
 from zmq.eventloop import ioloop, zmqstream
 
 # local imports
 from IPython.external.decorator import decorator
 from IPython.config.loader import Config
 from IPython.utils.traitlets import Instance, Dict, List, Set, Int, Str, Enum
 
 from IPython.parallel import error
 from IPython.parallel.factory import SessionFactory
 from IPython.parallel.util import connect_logger, local_logger
 
 from .dependency import Dependency
 
 @decorator
 def logged(f,self,*args,**kwargs):
     # print ("#--------------------")
     self.log.debug("scheduler::%s(*%s,**%s)"%(f.func_name, args, kwargs))
     # print ("#--")
     return f(self,*args, **kwargs)
 
 #----------------------------------------------------------------------
 # Chooser functions
 #----------------------------------------------------------------------
 
 def plainrandom(loads):
     """Plain random pick."""
     n = len(loads)
     return randint(0,n-1)
 
 def lru(loads):
     """Always pick the front of the line.
     
     The content of `loads` is ignored.
     
     Assumes LRU ordering of loads, with oldest first.
     """
     return 0
 
 def twobin(loads):
     """Pick two at random, use the LRU of the two.
     
     The content of loads is ignored.
     
     Assumes LRU ordering of loads, with oldest first.
     """
     n = len(loads)
     a = randint(0,n-1)
     b = randint(0,n-1)
     return min(a,b)
 
 def weighted(loads):
     """Pick two at random using inverse load as weight.
     
     Return the less loaded of the two.
     """
     # weight 0 a million times more than 1:
     weights = 1./(1e-6+numpy.array(loads))
     sums = weights.cumsum()
     t = sums[-1]
     x = random()*t
     y = random()*t
     idx = 0
     idy = 0
     while sums[idx] < x:
         idx += 1
     while sums[idy] < y:
         idy += 1
     if weights[idy] > weights[idx]:
         return idy
     else:
         return idx
 
 def leastload(loads):
     """Always choose the lowest load.
     
     If the lowest load occurs more than once, the first
     occurance will be used.  If loads has LRU ordering, this means
     the LRU of those with the lowest load is chosen.
     """
     return loads.index(min(loads))
 
 #---------------------------------------------------------------------
 # Classes
 #---------------------------------------------------------------------
 # store empty default dependency:
 MET = Dependency([])
 
 class TaskScheduler(SessionFactory):
     """Python TaskScheduler object.
     
     This is the simplest object that supports msg_id based
     DAG dependencies. *Only* task msg_ids are checked, not
     msg_ids of jobs submitted via the MUX queue.
     
     """
     
     hwm = Int(0, config=True, shortname='hwm',
         help="""specify the High Water Mark (HWM) for the downstream
         socket in the Task scheduler. This is the maximum number
         of allowed outstanding tasks on each engine."""
     )
     scheme_name = Enum(('leastload', 'pure', 'lru', 'plainrandom', 'weighted', 'twobin'),
         'leastload', config=True, shortname='scheme', allow_none=False,
         help="""select the task scheduler scheme  [default: Python LRU]
         Options are: 'pure', 'lru', 'plainrandom', 'weighted', 'twobin','leastload'"""
     )
     def _scheme_name_changed(self, old, new):
         self.log.debug("Using scheme %r"%new)
         self.scheme = globals()[new]
     
     # input arguments:
     scheme = Instance(FunctionType) # function for determining the destination
     def _scheme_default(self):
         return leastload
     client_stream = Instance(zmqstream.ZMQStream) # client-facing stream
     engine_stream = Instance(zmqstream.ZMQStream) # engine-facing stream
     notifier_stream = Instance(zmqstream.ZMQStream) # hub-facing sub stream
     mon_stream = Instance(zmqstream.ZMQStream) # hub-facing pub stream
     
     # internals:
     graph = Dict() # dict by msg_id of [ msg_ids that depend on key ]
     retries = Dict() # dict by msg_id of retries remaining (non-neg ints)
     # waiting = List() # list of msg_ids ready to run, but haven't due to HWM
     depending = Dict() # dict by msg_id of (msg_id, raw_msg, after, follow)
     pending = Dict() # dict by engine_uuid of submitted tasks
     completed = Dict() # dict by engine_uuid of completed tasks
     failed = Dict() # dict by engine_uuid of failed tasks
     destinations = Dict() # dict by msg_id of engine_uuids where jobs ran (reverse of completed+failed)
     clients = Dict() # dict by msg_id for who submitted the task
     targets = List() # list of target IDENTs
     loads = List() # list of engine loads
     # full = Set() # set of IDENTs that have HWM outstanding tasks
     all_completed = Set() # set of all completed tasks
     all_failed = Set() # set of all failed tasks
     all_done = Set() # set of all finished tasks=union(completed,failed)
     all_ids = Set() # set of all submitted task IDs
     blacklist = Dict() # dict by msg_id of locations where a job has encountered UnmetDependency
     auditor = Instance('zmq.eventloop.ioloop.PeriodicCallback')
     
     
     def start(self):
         self.engine_stream.on_recv(self.dispatch_result, copy=False)
         self._notification_handlers = dict(
             registration_notification = self._register_engine,
             unregistration_notification = self._unregister_engine
         )
         self.notifier_stream.on_recv(self.dispatch_notification)
         self.auditor = ioloop.PeriodicCallback(self.audit_timeouts, 2e3, self.loop) # 1 Hz
         self.auditor.start()
         self.log.info("Scheduler started...%r"%self)
     
     def resume_receiving(self):
         """Resume accepting jobs."""
         self.client_stream.on_recv(self.dispatch_submission, copy=False)
     
     def stop_receiving(self):
         """Stop accepting jobs while there are no engines.
         Leave them in the ZMQ queue."""
         self.client_stream.on_recv(None)
     
     #-----------------------------------------------------------------------
     # [Un]Registration Handling
     #-----------------------------------------------------------------------
     
     def dispatch_notification(self, msg):
         """dispatch register/unregister events."""
-        idents,msg = self.session.feed_identities(msg)
-        msg = self.session.unpack_message(msg)
+        try:
+            idents,msg = self.session.feed_identities(msg)
+        except ValueError:
+            self.log.warn("task::Invalid Message: %r"%msg)
+            return
+        try:
+            msg = self.session.unpack_message(msg)
+        except ValueError:
+            self.log.warn("task::Unauthorized message from: %r"%idents)
+            return
+        
         msg_type = msg['msg_type']
+        
         handler = self._notification_handlers.get(msg_type, None)
         if handler is None:
-            raise Exception("Unhandled message type: %s"%msg_type)
+            self.log.error("Unhandled message type: %r"%msg_type)
         else:
             try:
                 handler(str(msg['content']['queue']))
             except KeyError:
-                self.log.error("task::Invalid notification msg: %s"%msg)
+                self.log.error("task::Invalid notification msg: %r"%msg)
     
     @logged
     def _register_engine(self, uid):
         """New engine with ident `uid` became available."""
         # head of the line:
         self.targets.insert(0,uid)
         self.loads.insert(0,0)
         # initialize sets
         self.completed[uid] = set()
         self.failed[uid] = set()
         self.pending[uid] = {}
         if len(self.targets) == 1:
             self.resume_receiving()
         # rescan the graph:
         self.update_graph(None)
 
     def _unregister_engine(self, uid):
         """Existing engine with ident `uid` became unavailable."""
         if len(self.targets) == 1:
             # this was our only engine
             self.stop_receiving()
         
         # handle any potentially finished tasks:
         self.engine_stream.flush()
         
         # don't pop destinations, because they might be used later
         # map(self.destinations.pop, self.completed.pop(uid))
         # map(self.destinations.pop, self.failed.pop(uid))
 
         # prevent this engine from receiving work
         idx = self.targets.index(uid)
         self.targets.pop(idx)
         self.loads.pop(idx)
         
         # wait 5 seconds before cleaning up pending jobs, since the results might
         # still be incoming
         if self.pending[uid]:
             dc = ioloop.DelayedCallback(lambda : self.handle_stranded_tasks(uid), 5000, self.loop)
             dc.start()
         else:
             self.completed.pop(uid)
             self.failed.pop(uid)
 
     
     @logged
     def handle_stranded_tasks(self, engine):
         """Deal with jobs resident in an engine that died."""
         lost = self.pending[engine]
         for msg_id in lost.keys():
             if msg_id not in self.pending[engine]:
                 # prevent double-handling of messages
                 continue
 
             raw_msg = lost[msg_id][0]
             idents,msg = self.session.feed_identities(raw_msg, copy=False)
-            msg = self.session.unpack_message(msg, copy=False, content=False)
-            parent = msg['header']
+            parent = self.session.unpack(msg[1].bytes)
             idents = [engine, idents[0]]
 
             # build fake error reply
             try:
                 raise error.EngineError("Engine %r died while running task %r"%(engine, msg_id))
             except:
                 content = error.wrap_exception()
             msg = self.session.msg('apply_reply', content, parent=parent, subheader={'status':'error'})
             raw_reply = map(zmq.Message, self.session.serialize(msg, ident=idents))
             # and dispatch it
             self.dispatch_result(raw_reply)
 
         # finally scrub completed/failed lists
         self.completed.pop(engine)
         self.failed.pop(engine)
     
     
     #-----------------------------------------------------------------------
     # Job Submission
     #-----------------------------------------------------------------------
     @logged
     def dispatch_submission(self, raw_msg):
         """Dispatch job submission to appropriate handlers."""
         # ensure targets up to date:
         self.notifier_stream.flush()
         try:
             idents, msg = self.session.feed_identities(raw_msg, copy=False)
             msg = self.session.unpack_message(msg, content=False, copy=False)
         except Exception:
             self.log.error("task::Invaid task msg: %r"%raw_msg, exc_info=True)
             return
         
-
+        
         # send to monitor
         self.mon_stream.send_multipart(['intask']+raw_msg, copy=False)
         
         header = msg['header']
         msg_id = header['msg_id']
         self.all_ids.add(msg_id)
         
         # targets
         targets = set(header.get('targets', []))
         retries = header.get('retries', 0)
         self.retries[msg_id] = retries
         
         # time dependencies
         after = Dependency(header.get('after', []))
         if after.all:
             if after.success:
                 after.difference_update(self.all_completed)
             if after.failure:
                 after.difference_update(self.all_failed)
         if after.check(self.all_completed, self.all_failed):
             # recast as empty set, if `after` already met,
             # to prevent unnecessary set comparisons
             after = MET
         
         # location dependencies
         follow = Dependency(header.get('follow', []))
         
         # turn timeouts into datetime objects:
         timeout = header.get('timeout', None)
         if timeout:
             timeout = datetime.now() + timedelta(0,timeout,0)
         
         args = [raw_msg, targets, after, follow, timeout]
         
         # validate and reduce dependencies:
         for dep in after,follow:
             # check valid:
             if msg_id in dep or dep.difference(self.all_ids):
                 self.depending[msg_id] = args
                 return self.fail_unreachable(msg_id, error.InvalidDependency)
             # check if unreachable:
             if dep.unreachable(self.all_completed, self.all_failed):
                 self.depending[msg_id] = args
                 return self.fail_unreachable(msg_id)
         
         if after.check(self.all_completed, self.all_failed):
             # time deps already met, try to run
             if not self.maybe_run(msg_id, *args):
                 # can't run yet
                 if msg_id not in self.all_failed:
                     # could have failed as unreachable
                     self.save_unmet(msg_id, *args)
         else:
             self.save_unmet(msg_id, *args)
     
     # @logged
     def audit_timeouts(self):
         """Audit all waiting tasks for expired timeouts."""
         now = datetime.now()
         for msg_id in self.depending.keys():
             # must recheck, in case one failure cascaded to another:
             if msg_id in self.depending:
                 raw,after,targets,follow,timeout = self.depending[msg_id]
                 if timeout and timeout < now:
                     self.fail_unreachable(msg_id, error.TaskTimeout)
                 
     @logged
     def fail_unreachable(self, msg_id, why=error.ImpossibleDependency):
         """a task has become unreachable, send a reply with an ImpossibleDependency
         error."""
         if msg_id not in self.depending:
             self.log.error("msg %r already failed!"%msg_id)
             return
         raw_msg,targets,after,follow,timeout = self.depending.pop(msg_id)
         for mid in follow.union(after):
             if mid in self.graph:
                 self.graph[mid].remove(msg_id)
         
         # FIXME: unpacking a message I've already unpacked, but didn't save:
         idents,msg = self.session.feed_identities(raw_msg, copy=False)
-        msg = self.session.unpack_message(msg, copy=False, content=False)
-        header = msg['header']
+        header = self.session.unpack(msg[1].bytes)
         
         try:
             raise why()
         except:
             content = error.wrap_exception()
         
         self.all_done.add(msg_id)
         self.all_failed.add(msg_id)
         
         msg = self.session.send(self.client_stream, 'apply_reply', content, 
                                                 parent=header, ident=idents)
         self.session.send(self.mon_stream, msg, ident=['outtask']+idents)
         
         self.update_graph(msg_id, success=False)
     
     @logged
     def maybe_run(self, msg_id, raw_msg, targets, after, follow, timeout):
         """check location dependencies, and run if they are met."""
         blacklist = self.blacklist.setdefault(msg_id, set())
         if follow or targets or blacklist or self.hwm:
             # we need a can_run filter
             def can_run(idx):
                 # check hwm
                 if self.hwm and self.loads[idx] == self.hwm:
                     return False
                 target = self.targets[idx]
                 # check blacklist
                 if target in blacklist:
                     return False
                 # check targets
                 if targets and target not in targets:
                     return False
                 # check follow
                 return follow.check(self.completed[target], self.failed[target])
             
             indices = filter(can_run, range(len(self.targets)))
 
             if not indices:
                 # couldn't run
                 if follow.all:
                     # check follow for impossibility
                     dests = set()
                     relevant = set()
                     if follow.success:
                         relevant = self.all_completed
                     if follow.failure:
                         relevant = relevant.union(self.all_failed)
                     for m in follow.intersection(relevant):
                         dests.add(self.destinations[m])
                     if len(dests) > 1:
                         self.depending[msg_id] = (raw_msg, targets, after, follow, timeout)
                         self.fail_unreachable(msg_id)
                         return False
                 if targets:
                     # check blacklist+targets for impossibility
                     targets.difference_update(blacklist)
                     if not targets or not targets.intersection(self.targets):
                         self.depending[msg_id] = (raw_msg, targets, after, follow, timeout)
                         self.fail_unreachable(msg_id)
                         return False
                 return False
         else:
             indices = None
             
         self.submit_task(msg_id, raw_msg, targets, follow, timeout, indices)
         return True
             
     @logged
     def save_unmet(self, msg_id, raw_msg, targets, after, follow, timeout):
         """Save a message for later submission when its dependencies are met."""
         self.depending[msg_id] = [raw_msg,targets,after,follow,timeout]
         # track the ids in follow or after, but not those already finished
         for dep_id in after.union(follow).difference(self.all_done):
             if dep_id not in self.graph:
                 self.graph[dep_id] = set()
             self.graph[dep_id].add(msg_id)
     
     @logged
     def submit_task(self, msg_id, raw_msg, targets, follow, timeout, indices=None):
         """Submit a task to any of a subset of our targets."""
         if indices:
             loads = [self.loads[i] for i in indices]
         else:
             loads = self.loads
         idx = self.scheme(loads)
         if indices:
             idx = indices[idx]
         target = self.targets[idx]
         # print (target, map(str, msg[:3]))
         # send job to the engine
         self.engine_stream.send(target, flags=zmq.SNDMORE, copy=False)
         self.engine_stream.send_multipart(raw_msg, copy=False)
         # update load
         self.add_job(idx)
         self.pending[target][msg_id] = (raw_msg, targets, MET, follow, timeout)
         # notify Hub
         content = dict(msg_id=msg_id, engine_id=target)
         self.session.send(self.mon_stream, 'task_destination', content=content, 
                         ident=['tracktask',self.session.session])
         
     
     #-----------------------------------------------------------------------
     # Result Handling
     #-----------------------------------------------------------------------
     @logged
     def dispatch_result(self, raw_msg):
         """dispatch method for result replies"""
         try:
             idents,msg = self.session.feed_identities(raw_msg, copy=False)
             msg = self.session.unpack_message(msg, content=False, copy=False)
             engine = idents[0]
             try:
                 idx = self.targets.index(engine)
             except ValueError:
                 pass # skip load-update for dead engines
             else:
                 self.finish_job(idx)
         except Exception:
             self.log.error("task::Invaid result: %r"%raw_msg, exc_info=True)
             return
 
         header = msg['header']
         parent = msg['parent_header']
         if header.get('dependencies_met', True):
             success = (header['status'] == 'ok')
             msg_id = parent['msg_id']
             retries = self.retries[msg_id]
             if not success and retries > 0:
                 # failed
                 self.retries[msg_id] = retries - 1
                 self.handle_unmet_dependency(idents, parent)
             else:
                 del self.retries[msg_id]
                 # relay to client and update graph
                 self.handle_result(idents, parent, raw_msg, success)
                 # send to Hub monitor
                 self.mon_stream.send_multipart(['outtask']+raw_msg, copy=False)
         else:
             self.handle_unmet_dependency(idents, parent)
         
     @logged
     def handle_result(self, idents, parent, raw_msg, success=True):
         """handle a real task result, either success or failure"""
         # first, relay result to client
         engine = idents[0]
         client = idents[1]
         # swap_ids for XREP-XREP mirror
         raw_msg[:2] = [client,engine]
         # print (map(str, raw_msg[:4]))
         self.client_stream.send_multipart(raw_msg, copy=False)
         # now, update our data structures
         msg_id = parent['msg_id']
         self.blacklist.pop(msg_id, None)
         self.pending[engine].pop(msg_id)
         if success:
             self.completed[engine].add(msg_id)
             self.all_completed.add(msg_id)
         else:
             self.failed[engine].add(msg_id)
             self.all_failed.add(msg_id)
         self.all_done.add(msg_id)
         self.destinations[msg_id] = engine
         
         self.update_graph(msg_id, success)
         
     @logged
     def handle_unmet_dependency(self, idents, parent):
         """handle an unmet dependency"""
         engine = idents[0]
         msg_id = parent['msg_id']
         
         if msg_id not in self.blacklist:
             self.blacklist[msg_id] = set()
         self.blacklist[msg_id].add(engine)
         
         args = self.pending[engine].pop(msg_id)
         raw,targets,after,follow,timeout = args
         
         if self.blacklist[msg_id] == targets:
             self.depending[msg_id] = args
             self.fail_unreachable(msg_id)
         elif not self.maybe_run(msg_id, *args):
             # resubmit failed
             if msg_id not in self.all_failed:
                 # put it back in our dependency tree
                 self.save_unmet(msg_id, *args)
         
         if self.hwm:
             try:
                 idx = self.targets.index(engine)
             except ValueError:
                 pass # skip load-update for dead engines
             else:
                 if self.loads[idx] == self.hwm-1:
                     self.update_graph(None)
         
         
     
     @logged
     def update_graph(self, dep_id=None, success=True):
         """dep_id just finished. Update our dependency
         graph and submit any jobs that just became runable.
         
         Called with dep_id=None to update entire graph for hwm, but without finishing
         a task.
         """
         # print ("\n\n***********")
         # pprint (dep_id)
         # pprint (self.graph)
         # pprint (self.depending)
         # pprint (self.all_completed)
         # pprint (self.all_failed)
         # print ("\n\n***********\n\n")
         # update any jobs that depended on the dependency
         jobs = self.graph.pop(dep_id, [])
 
         # recheck *all* jobs if
         # a) we have HWM and an engine just become no longer full
         # or b) dep_id was given as None
         if dep_id is None or self.hwm and any( [ load==self.hwm-1 for load in self.loads ]):
             jobs = self.depending.keys()
         
         for msg_id in jobs:
             raw_msg, targets, after, follow, timeout = self.depending[msg_id]
             
             if after.unreachable(self.all_completed, self.all_failed) or follow.unreachable(self.all_completed, self.all_failed):
                 self.fail_unreachable(msg_id)
             
             elif after.check(self.all_completed, self.all_failed): # time deps met, maybe run
                 if self.maybe_run(msg_id, raw_msg, targets, MET, follow, timeout):
                     
                     self.depending.pop(msg_id)
                     for mid in follow.union(after):
                         if mid in self.graph:
                             self.graph[mid].remove(msg_id)
     
     #----------------------------------------------------------------------
     # methods to be overridden by subclasses
     #----------------------------------------------------------------------
     
     def add_job(self, idx):
         """Called after self.targets[idx] just got the job with header.
         Override with subclasses.  The default ordering is simple LRU.
         The default loads are the number of outstanding jobs."""
         self.loads[idx] += 1
         for lis in (self.targets, self.loads):
             lis.append(lis.pop(idx))
             
     
     def finish_job(self, idx):
         """Called after self.targets[idx] just finished a job.
         Override with subclasses."""
         self.loads[idx] -= 1
     
 
 
 def launch_scheduler(in_addr, out_addr, mon_addr, not_addr, config=None,logname='ZMQ', 
                             log_url=None, loglevel=logging.DEBUG,
                             identity=b'task'):
     from zmq.eventloop import ioloop
     from zmq.eventloop.zmqstream import ZMQStream
     
     if config:
         # unwrap dict back into Config
         config = Config(config)
 
     ctx = zmq.Context()
     loop = ioloop.IOLoop()
     ins = ZMQStream(ctx.socket(zmq.XREP),loop)
     ins.setsockopt(zmq.IDENTITY, identity)
     ins.bind(in_addr)
     
     outs = ZMQStream(ctx.socket(zmq.XREP),loop)
     outs.setsockopt(zmq.IDENTITY, identity)
     outs.bind(out_addr)
     mons = ZMQStream(ctx.socket(zmq.PUB),loop)
     mons.connect(mon_addr)
     nots = ZMQStream(ctx.socket(zmq.SUB),loop)
     nots.setsockopt(zmq.SUBSCRIBE, '')
     nots.connect(not_addr)
     
     # setup logging. Note that these will not work in-process, because they clobber
     # existing loggers.
     if log_url:
         connect_logger(logname, ctx, log_url, root="scheduler", loglevel=loglevel)
     else:
         local_logger(logname, loglevel)
     
     scheduler = TaskScheduler(client_stream=ins, engine_stream=outs,
                             mon_stream=mons, notifier_stream=nots,
                             loop=loop, logname=logname,
                             config=config)
     scheduler.start()
     try:
         loop.start()
     except KeyboardInterrupt:
         print ("interrupted, exiting...", file=sys.__stderr__)
 

IPython/parallel/streamsession.py

@@ -1,446 +1,483 b''
 #!/usr/bin/env python
 """edited session.py to work with streams, and move msg_type to the header
 """
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010-2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
+#-----------------------------------------------------------------------------
+# Imports
+#-----------------------------------------------------------------------------
 
+import hmac
 import os
 import pprint
 import uuid
 from datetime import datetime
 
 try:
     import cPickle
     pickle = cPickle
 except:
     cPickle = None
     import pickle
 
 import zmq
 from zmq.utils import jsonapi
 from zmq.eventloop.zmqstream import ZMQStream
 
 from IPython.config.configurable import Configurable
 from IPython.utils.importstring import import_item
-from IPython.utils.traitlets import CStr, Unicode, Bool, Any
+from IPython.utils.traitlets import CStr, Unicode, Bool, Any, Instance, Set
 
 from .util import ISO8601
 
+#-----------------------------------------------------------------------------
+# utility functions
+#-----------------------------------------------------------------------------
 
 def squash_unicode(obj):
     """coerce unicode back to bytestrings."""
     if isinstance(obj,dict):
         for key in obj.keys():
             obj[key] = squash_unicode(obj[key])
             if isinstance(key, unicode):
                 obj[squash_unicode(key)] = obj.pop(key)
     elif isinstance(obj, list):
         for i,v in enumerate(obj):
             obj[i] = squash_unicode(v)
     elif isinstance(obj, unicode):
         obj = obj.encode('utf8')
     return obj
 
 def _date_default(obj):
     if isinstance(obj, datetime):
         return obj.strftime(ISO8601)
     else:
         raise TypeError("%r is not JSON serializable"%obj)
 
+#-----------------------------------------------------------------------------
+# globals and defaults
+#-----------------------------------------------------------------------------
+
 _default_key = 'on_unknown' if jsonapi.jsonmod.__name__ == 'jsonlib' else 'default'
 json_packer = lambda obj: jsonapi.dumps(obj, **{_default_key:_date_default})
 json_unpacker = lambda s: squash_unicode(jsonapi.loads(s))
 
 pickle_packer = lambda o: pickle.dumps(o,-1)
 pickle_unpacker = pickle.loads
 
 default_packer = json_packer
 default_unpacker = json_unpacker
 
 
 DELIM="<IDS|MSG>"
 
+#-----------------------------------------------------------------------------
+# Classes
+#-----------------------------------------------------------------------------
+
 class Message(object):
     """A simple message object that maps dict keys to attributes.
 
     A Message can be created from a dict and a dict from a Message instance
     simply by calling dict(msg_obj)."""
     
     def __init__(self, msg_dict):
         dct = self.__dict__
         for k, v in dict(msg_dict).iteritems():
             if isinstance(v, dict):
                 v = Message(v)
             dct[k] = v
 
     # Having this iterator lets dict(msg_obj) work out of the box.
     def __iter__(self):
         return iter(self.__dict__.iteritems())
     
     def __repr__(self):
         return repr(self.__dict__)
 
     def __str__(self):
         return pprint.pformat(self.__dict__)
 
     def __contains__(self, k):
         return k in self.__dict__
 
     def __getitem__(self, k):
         return self.__dict__[k]
 
 
 def msg_header(msg_id, msg_type, username, session):
     date=datetime.now().strftime(ISO8601)
     return locals()
 
 def extract_header(msg_or_header):
     """Given a message or header, return the header."""
     if not msg_or_header:
         return {}
     try:
         # See if msg_or_header is the entire message.
         h = msg_or_header['header']
     except KeyError:
         try:
             # See if msg_or_header is just the header
             h = msg_or_header['msg_id']
         except KeyError:
             raise
         else:
             h = msg_or_header
     if not isinstance(h, dict):
         h = dict(h)
     return h
 
 class StreamSession(Configurable):
     """tweaked version of IPython.zmq.session.Session, for development in Parallel"""
     debug=Bool(False, config=True, help="""Debug output in the StreamSession""")
     packer = Unicode('json',config=True,
             help="""The name of the packer for serializing messages.
             Should be one of 'json', 'pickle', or an import name
             for a custom serializer.""")
     def _packer_changed(self, name, old, new):
         if new.lower() == 'json':
             self.pack = json_packer
             self.unpack = json_unpacker
         elif new.lower() == 'pickle':
             self.pack = pickle_packer
             self.unpack = pickle_unpacker
         else:
             self.pack = import_item(new)
 
     unpacker = Unicode('json',config=True,
         help="""The name of the unpacker for unserializing messages.
         Only used with custom functions for `packer`.""")
     def _unpacker_changed(self, name, old, new):
         if new.lower() == 'json':
             self.pack = json_packer
             self.unpack = json_unpacker
         elif new.lower() == 'pickle':
             self.pack = pickle_packer
             self.unpack = pickle_unpacker
         else:
             self.unpack = import_item(new)
         
     session = CStr('',config=True,
         help="""The UUID identifying this session.""")
     def _session_default(self):
         return bytes(uuid.uuid4())
-    username = Unicode(os.environ.get('USER','username'),config=True,
+    username = Unicode(os.environ.get('USER','username'), config=True,
         help="""Username for the Session. Default is your system username.""")
+    
+    # message signature related traits:
     key = CStr('', config=True,
         help="""execution key, for extra authentication.""")
-
+    def _key_changed(self, name, old, new):
+        if new:
+            self.auth = hmac.HMAC(new)
+        else:
+            self.auth = None
+    auth = Instance(hmac.HMAC)
+    counters = Instance('collections.defaultdict', (int,))
+    digest_history = Set()
+    
     keyfile = Unicode('', config=True,
         help="""path to file containing execution key.""")
     def _keyfile_changed(self, name, old, new):
         with open(new, 'rb') as f:
             self.key = f.read().strip()
 
     pack = Any(default_packer) # the actual packer function
     def _pack_changed(self, name, old, new):
         if not callable(new):
             raise TypeError("packer must be callable, not %s"%type(new))
         
     unpack = Any(default_unpacker) # the actual packer function
     def _unpack_changed(self, name, old, new):
         if not callable(new):
             raise TypeError("packer must be callable, not %s"%type(new))
 
     def __init__(self, **kwargs):
         super(StreamSession, self).__init__(**kwargs)
         self.none = self.pack({})
 
     @property
     def msg_id(self):
         """always return new uuid"""
         return str(uuid.uuid4())
 
     def msg_header(self, msg_type):
         return msg_header(self.msg_id, msg_type, self.username, self.session)
 
     def msg(self, msg_type, content=None, parent=None, subheader=None):
         msg = {}
         msg['header'] = self.msg_header(msg_type)
         msg['msg_id'] = msg['header']['msg_id']
         msg['parent_header'] = {} if parent is None else extract_header(parent)
         msg['msg_type'] = msg_type
         msg['content'] = {} if content is None else content
         sub = {} if subheader is None else subheader
         msg['header'].update(sub)
         return msg
 
     def check_key(self, msg_or_header):
         """Check that a message's header has the right key"""
         if not self.key:
             return True
         header = extract_header(msg_or_header)
         return header.get('key', '') == self.key
-            
+    
+    def sign(self, msg):
+        """Sign a message with HMAC digest. If no auth, return b''."""
+        if self.auth is None:
+            return b''
+        h = self.auth.copy()
+        for m in msg:
+            h.update(m)
+        return h.hexdigest()
 
     def serialize(self, msg, ident=None):
         content = msg.get('content', {})
         if content is None:
             content = self.none
         elif isinstance(content, dict):
             content = self.pack(content)
         elif isinstance(content, bytes):
             # content is already packed, as in a relayed message
             pass
         elif isinstance(content, unicode):
             # should be bytes, but JSON often spits out unicode
             content = content.encode('utf8')
         else:
             raise TypeError("Content incorrect type: %s"%type(content))
-
+        
+        real_message = [self.pack(msg['header']), 
+                        self.pack(msg['parent_header']), 
+                        content
+        ]
+        
         to_send = []
 
         if isinstance(ident, list):
             # accept list of idents
             to_send.extend(ident)
         elif ident is not None:
             to_send.append(ident)
         to_send.append(DELIM)
-        if self.key:
-            to_send.append(self.key)
-        to_send.append(self.pack(msg['header']))
-        to_send.append(self.pack(msg['parent_header']))
-        to_send.append(content)
+        
+        signature = self.sign(real_message)
+        to_send.append(signature)
+        
+        to_send.extend(real_message)
 
         return to_send
         
     def send(self, stream, msg_or_type, content=None, buffers=None, parent=None, subheader=None, ident=None, track=False):
         """Build and send a message via stream or socket.
         
         Parameters
         ----------
         
         stream : zmq.Socket or ZMQStream
             the socket-like object used to send the data
         msg_or_type : str or Message/dict
             Normally, msg_or_type will be a msg_type unless a message is being sent more
             than once.
         
         content : dict or None
             the content of the message (ignored if msg_or_type is a message)
         buffers : list or None
             the already-serialized buffers to be appended to the message
         parent : Message or dict or None
             the parent or parent header describing the parent of this message
         subheader : dict or None
             extra header keys for this message's header
         ident : bytes or list of bytes
             the zmq.IDENTITY routing path
         track : bool
             whether to track.  Only for use with Sockets, because ZMQStream objects cannot track messages.
         
         Returns
         -------
         msg : message dict
             the constructed message
         (msg,tracker) : (message dict, MessageTracker)
             if track=True, then a 2-tuple will be returned, the first element being the constructed
             message, and the second being the MessageTracker
             
         """
 
         if not isinstance(stream, (zmq.Socket, ZMQStream)):
             raise TypeError("stream must be Socket or ZMQStream, not %r"%type(stream))
         elif track and isinstance(stream, ZMQStream):
             raise TypeError("ZMQStream cannot track messages")
         
         if isinstance(msg_or_type, (Message, dict)):
             # we got a Message, not a msg_type
             # don't build a new Message
             msg = msg_or_type
         else:
             msg = self.msg(msg_or_type, content, parent, subheader)
         
         buffers = [] if buffers is None else buffers
         to_send = self.serialize(msg, ident)
         flag = 0
         if buffers:
             flag = zmq.SNDMORE
             _track = False
         else:
             _track=track
         if track:
             tracker = stream.send_multipart(to_send, flag, copy=False, track=_track)
         else:
             tracker = stream.send_multipart(to_send, flag, copy=False)
         for b in buffers[:-1]:
             stream.send(b, flag, copy=False)
         if buffers:
             if track:
                 tracker = stream.send(buffers[-1], copy=False, track=track)
             else:
                 tracker = stream.send(buffers[-1], copy=False)
                 
         # omsg = Message(msg)
         if self.debug:
             pprint.pprint(msg)
             pprint.pprint(to_send)
             pprint.pprint(buffers)
         
         msg['tracker'] = tracker
         
         return msg
     
     def send_raw(self, stream, msg, flags=0, copy=True, ident=None):
         """Send a raw message via ident path.
         
         Parameters
         ----------
         msg : list of sendable buffers"""
         to_send = []
         if isinstance(ident, bytes):
             ident = [ident]
         if ident is not None:
             to_send.extend(ident)
+            
         to_send.append(DELIM)
-        if self.key:
-            to_send.append(self.key)
+        to_send.append(self.sign(msg))
         to_send.extend(msg)
         stream.send_multipart(msg, flags, copy=copy)
     
     def recv(self, socket, mode=zmq.NOBLOCK, content=True, copy=True):
         """receives and unpacks a message
         returns [idents], msg"""
         if isinstance(socket, ZMQStream):
             socket = socket.socket
         try:
             msg = socket.recv_multipart(mode, copy=copy)
         except zmq.ZMQError as e:
             if e.errno == zmq.EAGAIN:
                 # We can convert EAGAIN to None as we know in this case
                 # recv_multipart won't return None.
                 return None
             else:
                 raise
         # return an actual Message object
         # determine the number of idents by trying to unpack them.
         # this is terrible:
         idents, msg = self.feed_identities(msg, copy)
         try:
             return idents, self.unpack_message(msg, content=content, copy=copy)
         except Exception as e:
             print (idents, msg)
             # TODO: handle it
             raise e
     
     def feed_identities(self, msg, copy=True):
         """feed until DELIM is reached, then return the prefix as idents and remainder as
         msg. This is easily broken by setting an IDENT to DELIM, but that would be silly.
         
         Parameters
         ----------
         msg : a list of Message or bytes objects
             the message to be split
         copy : bool
             flag determining whether the arguments are bytes or Messages
         
         Returns
         -------
         (idents,msg) : two lists
             idents will always be a list of bytes - the indentity prefix
             msg will be a list of bytes or Messages, unchanged from input
             msg should be unpackable via self.unpack_message at this point.
         """
-        ikey = int(self.key != '')
-        minlen = 3 + ikey
-        msg = list(msg)
-        idents = []
-        while len(msg) > minlen:
-            if copy:
-                s = msg[0]
-            else:
-                s = msg[0].bytes
-            if s == DELIM:
-                msg.pop(0)
-                break
-            else:
-                idents.append(s)
-                msg.pop(0)
-                
-        return idents, msg
+        if copy:
+            idx = msg.index(DELIM)
+            return msg[:idx], msg[idx+1:]
+        else:
+            failed = True
+            for idx,m in enumerate(msg):
+                if m.bytes == DELIM:
+                    failed = False
+                    break
+            if failed:
+                raise ValueError("DELIM not in msg")
+            idents, msg = msg[:idx], msg[idx+1:]
+            return [m.bytes for m in idents], msg
     
     def unpack_message(self, msg, content=True, copy=True):
         """Return a message object from the format
         sent by self.send.
         
         Parameters:
         -----------
         
         content : bool (True)
             whether to unpack the content dict (True), 
             or leave it serialized (False)
         
         copy : bool (True)
             whether to return the bytes (True), 
             or the non-copying Message object in each place (False)
         
         """
-        ikey = int(self.key != '')
-        minlen = 3 + ikey
+        minlen = 4
         message = {}
         if not copy:
             for i in range(minlen):
                 msg[i] = msg[i].bytes
-        if ikey:
-            if not self.key == msg[0]:
-                raise KeyError("Invalid Session Key: %s"%msg[0])
+        if self.auth is not None:
+            signature = msg[0]
+            if signature in self.digest_history:
+                raise ValueError("Duplicate Signature: %r"%signature)
+            self.digest_history.add(signature)
+            check = self.sign(msg[1:4])
+            if not signature == check:
+                raise ValueError("Invalid Signature: %r"%signature)
         if not len(msg) >= minlen:
             raise TypeError("malformed message, must have at least %i elements"%minlen)
-        message['header'] = self.unpack(msg[ikey+0])
+        message['header'] = self.unpack(msg[1])
         message['msg_type'] = message['header']['msg_type']
-        message['parent_header'] = self.unpack(msg[ikey+1])
+        message['parent_header'] = self.unpack(msg[2])
         if content:
-            message['content'] = self.unpack(msg[ikey+2])
+            message['content'] = self.unpack(msg[3])
         else:
-            message['content'] = msg[ikey+2]
+            message['content'] = msg[3]
         
-        message['buffers'] = msg[ikey+3:]# [ m.buffer for m in msg[3:] ]
+        message['buffers'] = msg[4:]
         return message
-        
 
 def test_msg2obj():
     am = dict(x=1)
     ao = Message(am)
     assert ao.x == am['x']
 
     am['y'] = dict(z=1)
     ao = Message(am)
     assert ao.y.z == am['y']['z']
     
     k1, k2 = 'y', 'z'
     assert ao[k1][k2] == am[k1][k2]
     
     am2 = dict(ao)
     assert am['x'] == am2['x']
     assert am['y']['z'] == am2['y']['z']

docs/source/parallel/parallel_security.txt

@@ -1,324 +1,324 b''
 .. _parallelsecurity:
 
 ===========================
 Security details of IPython
 ===========================
 
 .. note::
 
     This section is not thorough, and IPython.zmq needs a thorough security
     audit.
 
 IPython's :mod:`IPython.zmq` package exposes the full power of the
 Python interpreter over a TCP/IP network for the purposes of parallel
 computing. This feature brings up the important question of IPython's security
 model. This document gives details about this model and how it is implemented
 in IPython's architecture.
 
-Processs and network topology
-=============================
+Process and network topology
+============================
 
 To enable parallel computing, IPython has a number of different processes that
 run. These processes are discussed at length in the IPython documentation and
 are summarized here:
 
 * The IPython *engine*.  This process is a full blown Python
   interpreter in which user code is executed.  Multiple
   engines are started to make parallel computing possible.
 * The IPython *hub*.  This process monitors a set of 
   engines and schedulers, and keeps track of the state of the processes. It listens
   for registration connections from engines and clients, and monitor connections
   from schedulers.
 * The IPython *schedulers*. This is a set of processes that relay commands and results
   between clients and engines. They are typically on the same machine as the controller,
   and listen for connections from engines and clients, but connect to the Hub.
 * The IPython *client*.  This process is typically an
   interactive Python process that is used to coordinate the
   engines to get a parallel computation done.
 
-Collectively, these processes are called the IPython *kernel*, and the hub and schedulers
+Collectively, these processes are called the IPython *cluster*, and the hub and schedulers
 together are referred to as the *controller*.
 
-.. note::
-
-    Are these really still referred to as the Kernel?  It doesn't seem so to me.  'cluster'
-    seems more accurate.
-    
-    -MinRK
 
 These processes communicate over any transport supported by ZeroMQ (tcp,pgm,infiniband,ipc)
 with a well defined topology. The IPython hub and schedulers listen on sockets. Upon
 starting, an engine connects to a hub and registers itself, which then informs the engine
 of the connection information for the schedulers, and the engine then connects to the
 schedulers. These engine/hub and engine/scheduler connections persist for the
 lifetime of each engine.
 
 The IPython client also connects to the controller processes using a number of socket
 connections. As of writing, this is one socket per scheduler (4), and 3 connections to the
 hub for a total of 7. These connections persist for the lifetime of the client only.
 
 A given IPython controller and set of engines engines typically has a relatively
 short lifetime. Typically this lifetime corresponds to the duration of a single parallel
 simulation performed by a single user. Finally, the hub, schedulers, engines, and client
 processes typically execute with the permissions of that same user. More specifically, the
 controller and engines are *not* executed as root or with any other superuser permissions.
 
 Application logic
 =================
 
 When running the IPython kernel to perform a parallel computation, a user
 utilizes the IPython client to send Python commands and data through the
 IPython schedulers to the IPython engines, where those commands are executed
 and the data processed. The design of IPython ensures that the client is the
 only access point for the capabilities of the engines. That is, the only way
 of addressing the engines is through a client.
 
 A user can utilize the client to instruct the IPython engines to execute
 arbitrary Python commands. These Python commands can include calls to the
 system shell, access the filesystem, etc., as required by the user's
 application code. From this perspective, when a user runs an IPython engine on
 a host, that engine has the same capabilities and permissions as the user
 themselves (as if they were logged onto the engine's host with a terminal).
 
 Secure network connections
 ==========================
 
 Overview
 --------
 
 ZeroMQ provides exactly no security. For this reason, users of IPython must be very
 careful in managing connections, because an open TCP/IP socket presents access to
 arbitrary execution as the user on the engine machines. As a result, the default behavior
 of controller processes is to only listen for clients on the loopback interface, and the
 client must establish SSH tunnels to connect to the controller processes.
 
 .. warning:: 
 
     If the controller's loopback interface is untrusted, then IPython should be considered
     vulnerable, and this extends to the loopback of all connected clients, which have
     opened a loopback port that is redirected to the controller's loopback port.
 
 
 SSH
 ---
 
 Since ZeroMQ provides no security, SSH tunnels are the primary source of secure
 connections. A connector file, such as `ipcontroller-client.json`, will contain
 information for connecting to the controller, possibly including the address of an
 ssh-server through with the client is to tunnel. The Client object then creates tunnels
 using either [OpenSSH]_ or [Paramiko]_, depending on the platform. If users do not wish to
 use OpenSSH or Paramiko, or the tunneling utilities are insufficient, then they may
 construct the tunnels themselves, and simply connect clients and engines as if the
 controller were on loopback on the connecting machine.
 
 .. note::
 
     There is not currently tunneling available for engines.
 
 Authentication
 --------------
 
-To protect users of shared machines, an execution key is used to authenticate all messages.
+To protect users of shared machines, [HMAC]_ digests are used to sign messages, using a
+shared key.
 
 The Session object that handles the message protocol uses a unique key to verify valid
 messages. This can be any value specified by the user, but the default behavior is a
-pseudo-random 128-bit number, as generated by `uuid.uuid4()`. This key is checked on every
-message everywhere it is unpacked (Controller, Engine, and Client) to ensure that it came
-from an authentic user, and no messages that do not contain this key are acted upon in any
-way.
-
-There is exactly one key per cluster - it must be the same everywhere. Typically, the
-controller creates this key, and stores it in the private connection files
+pseudo-random 128-bit number, as generated by `uuid.uuid4()`. This key is used to
+initialize an HMAC object, which digests all messages, and includes that digest as a
+signature and part of the message. Every message that is unpacked (on Controller, Engine,
+and Client) will also be digested by the receiver, ensuring that the sender's key is the
+same as the receiver's. No messages that do not contain this key are acted upon in any
+way. The key itself is never sent over the network.
+
+There is exactly one shared key per cluster - it must be the same everywhere. Typically,
+the controller creates this key, and stores it in the private connection files
 `ipython-{engine|client}.json`. These files are typically stored in the
-`~/.ipython/cluster_<profile>/security` directory, and are maintained as readable only by
-the owner, just as is common practice with a user's keys in their `.ssh` directory.
+`~/.ipython/profile_<name>/security` directory, and are maintained as readable only by the
+owner, just as is common practice with a user's keys in their `.ssh` directory.
 
 .. warning::
 
     It is important to note that the key authentication, as emphasized by the use of
     a uuid rather than generating a key with a cryptographic library, provides a 
     defense against *accidental* messages more than it does against malicious attacks.
     If loopback is compromised, it would be trivial for an attacker to intercept messages
     and deduce the key, as there is no encryption.
 
 
 
 Specific security vulnerabilities
 =================================
 
 There are a number of potential security vulnerabilities present in IPython's
 architecture. In this section we discuss those vulnerabilities and detail how
 the security architecture described above prevents them from being exploited.
 
 Unauthorized clients
 --------------------
 
 The IPython client can instruct the IPython engines to execute arbitrary
 Python code with the permissions of the user who started the engines. If an
 attacker were able to connect their own hostile IPython client to the IPython
 controller, they could instruct the engines to execute code.
 
 
 On the first level, this attack is prevented by requiring access to the controller's
 ports, which are recommended to only be open on loopback if the controller is on an
 untrusted local network. If the attacker does have access to the Controller's ports, then
 the attack is prevented by the capabilities based client authentication of the execution
 key. The relevant authentication information is encoded into the JSON file that clients
 must present to gain access to the IPython controller. By limiting the distribution of
 those keys, a user can grant access to only authorized persons, just as with SSH keys.
 
 It is highly unlikely that an execution key could be guessed by an attacker
 in a brute force guessing attack. A given instance of the IPython controller
 only runs for a relatively short amount of time (on the order of hours). Thus
 an attacker would have only a limited amount of time to test a search space of
-size 2**128.  
+size 2**128.  For added security, users can have arbitrarily long keys.
 
 .. warning::
 
-    If the attacker has gained enough access to intercept loopback connections on
-    *either* the controller or client, then the key is easily deduced from network
-    traffic.
+    If the attacker has gained enough access to intercept loopback connections on *either* the
+    controller or client, then a duplicate message can be sent. To protect against this,
+    recipients only allow each signature once, and consider duplicates invalid.  However,
+    the duplicate message could be sent to *another* recipient using the same key,
+    and it would be considered valid.
 
 
 Unauthorized engines
 --------------------
 
 If an attacker were able to connect a hostile engine to a user's controller,
 the user might unknowingly send sensitive code or data to the hostile engine.
 This attacker's engine would then have full access to that code and data.
 
 This type of attack is prevented in the same way as the unauthorized client
 attack, through the usage of the capabilities based authentication scheme.
 
 Unauthorized controllers
 ------------------------
 
 It is also possible that an attacker could try to convince a user's IPython
 client or engine to connect to a hostile IPython controller. That controller
 would then have full access to the code and data sent between the IPython
 client and the IPython engines.
 
 Again, this attack is prevented through the capabilities in a connection file, which
 ensure that a client or engine connects to the correct controller. It is also important to
 note that the connection files also encode the IP address and port that the controller is
 listening on, so there is little chance of mistakenly connecting to a controller running
 on a different IP address and port.
 
 When starting an engine or client, a user must specify the key to use
 for that connection. Thus, in order to introduce a hostile controller, the
 attacker must convince the user to use the key associated with the
 hostile controller. As long as a user is diligent in only using keys from
 trusted sources, this attack is not possible.
 
 .. note::
 
     I may be wrong, the unauthorized controller may be easier to fake than this.
 
 Other security measures
 =======================
 
 A number of other measures are taken to further limit the security risks
 involved in running the IPython kernel.
 
 First, by default, the IPython controller listens on random port numbers.
 While this can be overridden by the user, in the default configuration, an
 attacker would have to do a port scan to even find a controller to attack.
 When coupled with the relatively short running time of a typical controller
 (on the order of hours), an attacker would have to work extremely hard and
 extremely *fast* to even find a running controller to attack.
 
 Second, much of the time, especially when run on supercomputers or clusters,
 the controller is running behind a firewall. Thus, for engines or client to
 connect to the controller:
 
 * The different processes have to all be behind the firewall.
 
 or:
 
 * The user has to use SSH port forwarding to tunnel the
   connections through the firewall.
    
 In either case, an attacker is presented with additional barriers that prevent
 attacking or even probing the system.
 
 Summary
 =======
 
 IPython's architecture has been carefully designed with security in mind. The
 capabilities based authentication model, in conjunction with SSH tunneled
 TCP/IP channels, address the core potential vulnerabilities in the system,
 while still enabling user's to use the system in open networks.
 
 Other questions
 ===============
 
 .. note::
 
     this does not apply to ZMQ, but I am sure there will be questions.
 
 About keys
 ----------
 
 Can you clarify the roles of the certificate and its keys versus the FURL,
 which is also called a key?
 
 The certificate created by IPython processes is a standard public key x509
 certificate, that is used by the SSL handshake protocol to setup encrypted
 channel between the controller and the IPython engine or client. This public
 and private key associated with this certificate are used only by the SSL
 handshake protocol in setting up this encrypted channel.
 
 The FURL serves a completely different and independent purpose from the
 key pair associated with the certificate. When we refer to a FURL as a
 key, we are using the word "key" in the capabilities based security model
 sense. This has nothing to do with "key" in the public/private key sense used
 in the SSL protocol.
 
 With that said the FURL is used as an cryptographic key, to grant
 IPython engines and clients access to particular capabilities that the
 controller offers.
 
 Self signed certificates
 ------------------------
 
 Is the controller creating a self-signed certificate? Is this created for per
 instance/session, one-time-setup or each-time the controller is started?
 
 The Foolscap network protocol, which handles the SSL protocol details, creates
 a self-signed x509 certificate using OpenSSL for each IPython process. The
 lifetime of the certificate is handled differently for the IPython controller
 and the engines/client.
 
 For the IPython engines and client, the certificate is only held in memory for
 the lifetime of its process. It is never written to disk.
 
 For the controller, the certificate can be created anew each time the
 controller starts or it can be created once and reused each time the
 controller starts. If at any point, the certificate is deleted, a new one is
 created the next time the controller starts.
 
 SSL private key
 ---------------
 
 How the private key (associated with the certificate) is distributed?
 
 In the usual implementation of the SSL protocol, the private key is never
 distributed. We follow this standard always.
 
 SSL versus Foolscap authentication
 ----------------------------------
 
 Many SSL connections only perform one sided authentication (the server to the
 client). How is the client authentication in IPython's system related to SSL
 authentication?
 
 We perform a two way SSL handshake in which both parties request and verify
 the certificate of their peer. This mutual authentication is handled by the
 SSL handshake and is separate and independent from the additional
 authentication steps that the CLIENT and SERVER perform after an encrypted
 channel is established.
 
 .. [RFC5246] <http://tools.ietf.org/html/rfc5246>
 
 .. [OpenSSH] <http://www.openssh.com/>
 .. [Paramiko] <http://www.lag.net/paramiko/>
+.. [HMAC] <http://tools.ietf.org/html/rfc2104.html>