src/python/osrf/http_translator.py

   1 import sys, os, time, md5, random
   2 from mod_python import apache, util
   3 import osrf.system, osrf.cache, osrf.json, osrf.conf, osrf.net, osrf.log
   4 from osrf.const import OSRF_MESSAGE_TYPE_DISCONNECT, OSRF_MESSAGE_TYPE_CONNECT, \
   5     OSRF_STATUS_CONTINUE, OSRF_STATUS_TIMEOUT, OSRF_MESSAGE_TYPE_STATUS
   6
   7
   8 '''
   9 Proof of concept OpenSRF-HTTP multipart streaming gateway.
  10
  11 Example Apache mod_python config:
  12
  13 <Location /osrf-http-translator>
  14    SetHandler mod_python
  15    PythonPath "['/path/to/osrf-python'] + sys.path"
  16    PythonHandler osrf.http_translator
  17    PythonOption OSRF_CONFIG /path/to/opensrf_core.xml
  18    PythonOption OSRF_CONFIG_CONTEXT config.gateway
  19    PythonOption OSRF_CACHE_SERVERS 127.0.0.1:11211
  20    # testing only
  21    PythonAutoReload On
  22 </Location>
  23 '''
  24
  25 OSRF_HTTP_HEADER_TO = 'X-OpenSRF-to'
  26 OSRF_HTTP_HEADER_XID = 'X-OpenSRF-thread'
  27 OSRF_HTTP_HEADER_FROM = 'X-OpenSRF-from'
  28 OSRF_HTTP_HEADER_THREAD = 'X-OpenSRF-thread'
  29 OSRF_HTTP_HEADER_TIMEOUT = 'X-OpenSRF-timeout'
  30 OSRF_HTTP_HEADER_SERVICE = 'X-OpenSRF-service'
  31 OSRF_HTTP_HEADER_MULTIPART = 'X-OpenSRF-multipart'
  32 MULTIPART_CONTENT_TYPE = 'multipart/x-mixed-replace;boundary="%s"'
  33 JSON_CONTENT_TYPE = 'text/plain'
  34 CACHE_TIME = 300
  35
  36 ROUTER_NAME = None
  37 OSRF_DOMAIN = None
  38
  39 # If DEBUG_WRITE = True, all data sent to the client is also written
  40 # to stderr (apache error log)
  41 DEBUG_WRITE = False
  42
  43 def _dbg(msg):
  44     ''' testing only '''
  45     sys.stderr.write("%s\n\n" % str(msg))
  46     sys.stderr.flush()
  47
  48
  49 INIT_COMPLETE = False
  50 def child_init(req):
  51     ''' At time of writing, mod_python doesn't support a child_init handler,
  52         so this function is called once per process to initialize
  53         the opensrf connection '''
  54
  55     global INIT_COMPLETE, ROUTER_NAME, OSRF_DOMAIN
  56     if INIT_COMPLETE:
  57         return
  58
  59     # Apache complains with: UnboundLocalError: local variable 'osrf' referenced before assignment
  60     # if the following import line is removed, even though its also at the top of the file...
  61     import osrf.system
  62
  63     ops = req.get_options()
  64     conf = ops['OSRF_CONFIG']
  65     ctxt = ops.get('OSRF_CONFIG_CONTEXT') or 'opensrf'
  66     osrf.system.System.net_connect(config_file=conf, config_context=ctxt)
  67
  68     ROUTER_NAME = osrf.conf.get('router_name')
  69     OSRF_DOMAIN = osrf.conf.get('domain')
  70     INIT_COMPLETE = True
  71
  72     servers = ops.get('OSRF_CACHE_SERVERS')
  73     if servers:
  74         servers = servers.split(',')
  75     else:
  76         # no cache servers configured, see if we can talk to the settings server
  77         import osrf.set
  78         servers = osrf.set.get('cache.global.servers.server')
  79         if not isinstance(servers, list):
  80             servers = [servers]
  81
  82     osrf.cache.CacheClient.connect(servers)
  83
  84
  85 def handler(req):
  86     ''' Create the translator and tell it to process the request. '''
  87     child_init(req)
  88     status = HTTPTranslator(req).process()
  89     osrf.log.log_debug("translator call resulted in status %d" % int(status))
  90     return status
  91
  92 class HTTPTranslator(object):
  93     def __init__(self, apreq):
  94
  95         self.apreq = apreq
  96         if apreq.header_only:
  97             return
  98
  99         try:
 100             post = util.parse_qsl(apreq.read(int(apreq.headers_in['Content-length'])))
 101             self.body = [d for d in post if d[0] == 'osrf-msg'][0][1]
 102         except:
 103             self.body = None
 104             return
 105
 106         self.messages = []
 107         self.complete = False
 108         self.handle = osrf.net.get_network_handle()
 109         self.handle.set_receive_callback(None)
 110
 111         self.recipient = apreq.headers_in.get(OSRF_HTTP_HEADER_TO)
 112         self.service = apreq.headers_in.get(OSRF_HTTP_HEADER_SERVICE)
 113         self.thread = apreq.headers_in.get(OSRF_HTTP_HEADER_THREAD) or \
 114             "%s%s" % (os.getpid(), time.time())
 115         self.timeout = apreq.headers_in.get(OSRF_HTTP_HEADER_TIMEOUT) or 1200
 116         self.multipart = str(
 117             apreq.headers_in.get(OSRF_HTTP_HEADER_MULTIPART)).lower() == 'true'
 118         self.connect_only = False
 119         self.disconnect_only = False
 120
 121         # generate a random multipart delimiter
 122         mpart = md5.new()
 123         mpart.update("%f%d%d" % (time.time(), os.getpid(), \
 124             random.randint(100, 10000000)))
 125         self.delim = mpart.hexdigest()
 126         self.remote_host = self.apreq.get_remote_host(apache.REMOTE_NOLOOKUP)
 127         self.cache = osrf.cache.CacheClient()
 128
 129
 130     def process(self):
 131
 132         if self.apreq.header_only:
 133             return apache.OK
 134         if not self.body:
 135             return apache.HTTP_BAD_REQUEST
 136         if not self.set_to_addr():
 137             return apache.HTTP_BAD_REQUEST
 138         if not self.parse_request():
 139             return apache.HTTP_BAD_REQUEST
 140
 141         while self.handle.recv(0):
 142             pass # drop stale messages
 143
 144
 145         net_msg = osrf.net.NetworkMessage(
 146             recipient=self.recipient, thread=self.thread, body=self.body)
 147         self.handle.send(net_msg)
 148
 149         if self.disconnect_only:
 150             osrf.log.log_debug("exiting early on DISCONNECT")
 151             return apache.OK
 152
 153         first_write = True
 154         while not self.complete:
 155
 156             net_msg = self.handle.recv(self.timeout)
 157             if not net_msg:
 158                 return apache.GATEWAY_TIME_OUT
 159
 160             if not self.check_status(net_msg):
 161                 continue
 162
 163             if first_write:
 164                 self.init_headers(net_msg)
 165                 first_write = False
 166
 167             if self.multipart:
 168                 self.respond_chunk(net_msg)
 169                 if self.connect_only:
 170                     break
 171             else:
 172                 self.messages.append(net_msg.body)
 173
 174                 # condense the sets of arrays into a single array of messages
 175                 if self.complete or self.connect_only:
 176                     json = self.messages.pop(0)
 177                     while len(self.messages) > 0:
 178                         msg = self.messages.pop(0)
 179                         json = "%s,%s" % (json[0:len(json)-1], msg[1:])
 180
 181                     self.write("%s" % json)
 182
 183
 184         return apache.OK
 185
 186     def parse_request(self):
 187         '''
 188         If this is solely a DISCONNECT message, we set self.disconnect_only
 189         to true
 190         @return True if the body parses correctly, False otherwise
 191         '''
 192         osrf_msgs = osrf.json.to_object(self.body)
 193         if not osrf_msgs:
 194             return False
 195
 196         if len(osrf_msgs) == 1:
 197             if osrf_msgs[0].type() == OSRF_MESSAGE_TYPE_CONNECT:
 198                 self.connect_only = True
 199             elif osrf_msgs[0].type() == OSRF_MESSAGE_TYPE_DISCONNECT:
 200                 self.disconnect_only = True
 201
 202         return True
 203
 204
 205     def set_to_addr(self):
 206         ''' Determines the TO address.  Returns false if
 207             the address is missing or ambiguous.
 208             Also returns false if an explicit TO is specified and the
 209             thread/IP/TO combination is not found in the session cache
 210             '''
 211         if self.service:
 212             if self.recipient:
 213                 osrf.log.log_warn("specifying both SERVICE and TO is not allowed")
 214                 return False
 215             self.recipient = "%s@%s/%s" % \
 216                 (ROUTER_NAME, OSRF_DOMAIN, self.service)
 217             osrf.log.log_debug("set service to %s" % self.recipient)
 218             return True
 219         else:
 220             if self.recipient:
 221                 # If the client specifies a specific TO address, verify it's
 222                 # the same address that was cached with the previous request.
 223                 obj = self.cache.get(self.thread)
 224                 if obj and obj['ip'] == self.remote_host and \
 225                     obj['jid'] == self.recipient:
 226                     return True
 227         osrf.log.log_warn("client [%s] attempted to send directly "
 228             "[%s] without a session" % (self.remote_host, self.recipient))
 229         return False
 230
 231
 232     def init_headers(self, net_msg):
 233         osrf.log.log_debug("initializing request headers")
 234         self.apreq.headers_out[OSRF_HTTP_HEADER_FROM] = net_msg.sender
 235         self.apreq.headers_out[OSRF_HTTP_HEADER_THREAD] = self.thread
 236         if self.multipart:
 237             self.apreq.content_type = MULTIPART_CONTENT_TYPE % self.delim
 238             self.write("--%s\n" % self.delim)
 239         else:
 240             self.apreq.content_type = JSON_CONTENT_TYPE
 241         self.cache.put(self.thread, \
 242             {'ip':self.remote_host, 'jid': net_msg.sender}, CACHE_TIME)
 243
 244         osrf.log.log_debug("caching session [%s] for host [%s] and server "
 245             " drone [%s]" % (self.thread, self.remote_host, net_msg.sender))
 246
 247
 248
 249     def check_status(self, net_msg):
 250         ''' Checks the status of the server response.
 251             If we received a timeout message, we drop it.
 252             if it's any other non-continue status, we mark this session as
 253             complete and carry on.
 254             @return False if there is no data to return to the caller
 255             (dropped message, eg. timeout), True otherwise '''
 256
 257         osrf.log.log_debug('checking status...')
 258         osrf_msgs = osrf.json.to_object(net_msg.body)
 259         last_msg = osrf_msgs.pop()
 260
 261         if last_msg.type() == OSRF_MESSAGE_TYPE_STATUS:
 262             code = int(last_msg.payload().statusCode())
 263             osrf.log.log_debug("got a status message with code %d" % code)
 264
 265             if code == OSRF_STATUS_TIMEOUT:
 266                 osrf.log.log_debug("removing cached session [%s] and "
 267                     "dropping TIMEOUT message" % net_msg.thread)
 268                 self.cache.delete(net_msg.thread)
 269                 return False
 270
 271             if code != OSRF_STATUS_CONTINUE:
 272                 self.complete = True
 273
 274         return True
 275
 276
 277     def respond_chunk(self, resp):
 278         ''' Writes a single multipart-delimited chunk of data '''
 279
 280         self.write("Content-type: %s\n\n" % JSON_CONTENT_TYPE)
 281         self.write("%s\n\n" % resp.body)
 282         if self.complete:
 283             self.write("--%s--\n" % self.delim)
 284         else:
 285             self.write("--%s\n" % self.delim)
 286         self.apreq.flush()
 287
 288     def write(self, msg):
 289         ''' Writes data to the client stream. '''
 290
 291         if DEBUG_WRITE:
 292             sys.stderr.write(msg)
 293             sys.stderr.flush()
 294         self.apreq.write(msg)
 295
 296