#!/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, 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="" #----------------------------------------------------------------------------- # 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, 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) 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) 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. """ 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) """ minlen = 4 message = {} if not copy: for i in range(minlen): msg[i] = msg[i].bytes 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[1]) message['msg_type'] = message['header']['msg_type'] message['parent_header'] = self.unpack(msg[2]) if content: message['content'] = self.unpack(msg[3]) else: message['content'] = 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']