qwebirc/ajaxengine.py

   1 from twisted.web import resource, server, static
   2 from twisted.names import client
   3 from twisted.internet import reactor
   4 from authgateengine import login_optional
   5 import simplejson, md5, sys, os, ircclient, time, config, weakref, traceback
   6
   7 Sessions = {}
   8
   9 def get_session_id():
  10   return md5.md5(os.urandom(16)).hexdigest()
  11
  12 class BufferOverflowException(Exception):
  13   pass
  14
  15 class AJAXException(Exception):
  16   pass
  17
  18 class IDGenerationException(Exception):
  19   pass
  20
  21 NOT_DONE_YET = None
  22
  23 def jsondump(fn):
  24   def decorator(*args, **kwargs):
  25     try:
  26       x = fn(*args, **kwargs)
  27       if x is None:
  28         return server.NOT_DONE_YET
  29       x = (True, x)
  30     except AJAXException, e:
  31       x = (False, e[0])
  32
  33     return simplejson.dumps(x)
  34   return decorator
  35
  36 def cleanupSession(id):
  37   try:
  38     del Sessions[id]
  39   except KeyError:
  40     pass
  41
  42 class IRCSession:
  43   def __init__(self, id):
  44     self.id = id
  45     self.subscriptions = []
  46     self.buffer = []
  47     self.throttle = 0
  48     self.schedule = None
  49     self.closed = False
  50     self.cleanupschedule = None
  51
  52   def subscribe(self, channel):
  53     if len(self.subscriptions) >= config.MAXSUBSCRIPTIONS:
  54       self.subscriptions.pop(0).close()
  55
  56     self.subscriptions.append(channel)
  57     self.flush()
  58
  59   def flush(self, scheduled=False):
  60     if scheduled:
  61       self.schedule = None
  62
  63     if not self.buffer or not self.subscriptions:
  64       return
  65
  66     t = time.time()
  67
  68     if t < self.throttle:
  69       if not self.schedule:
  70         self.schedule = reactor.callLater(self.throttle - t, self.flush, True)
  71       return
  72     else:
  73       # process the rest of the packet
  74       if not scheduled:
  75         if not self.schedule:
  76           self.schedule = reactor.callLater(0, self.flush, True)
  77         return
  78
  79     self.throttle = t + config.UPDATE_FREQ
  80
  81     encdata = simplejson.dumps(self.buffer)
  82     self.buffer = []
  83
  84     newsubs = []
  85     for x in self.subscriptions:
  86       if x.write(encdata):
  87         newsubs.append(x)
  88
  89     self.subscriptions = newsubs
  90     if self.closed and not self.subscriptions:
  91       cleanupSession(self.id)
  92
  93   def event(self, data):
  94     bufferlen = sum(map(len, self.buffer))
  95     if bufferlen + len(data) > config.MAXBUFLEN:
  96       self.buffer = []
  97       self.client.error("Buffer overflow")
  98       return
  99
 100     self.buffer.append(data)
 101     self.flush()
 102
 103   def push(self, data):
 104     if not self.closed:
 105       self.client.write(data)
 106
 107   def disconnect(self):
 108     # keep the session hanging around for a few seconds so the
 109     # client has a chance to see what the issue was
 110     self.closed = True
 111
 112     reactor.callLater(5, cleanupSession, self.id)
 113
 114 class Channel:
 115   def __init__(self, request):
 116     self.request = request
 117
 118 class SingleUseChannel(Channel):
 119   def write(self, data):
 120     self.request.write(data)
 121     self.request.finish()
 122     return False
 123
 124   def close(self):
 125     self.request.finish()
 126
 127 class MultipleUseChannel(Channel):
 128   def write(self, data):
 129     self.request.write(data)
 130     return True
 131
 132 class AJAXEngine(resource.Resource):
 133   isLeaf = True
 134
 135   def __init__(self, prefix):
 136     self.prefix = prefix
 137
 138   @jsondump
 139   def render_POST(self, request):
 140     path = request.path[len(self.prefix):]
 141     if path[0] == "/":
 142       handler = self.COMMANDS.get(path[1:])
 143       if handler is not None:
 144         return handler(self, request)
 145     raise AJAXException("404")
 146
 147 #  def render_GET(self, request):
 148 #    return self.render_POST(request)
 149
 150   def newConnection(self, request):
 151     ticket = login_optional(request)
 152
 153     _, ip, port = request.transport.getPeer()
 154
 155     nick, ident, realname = request.args.get("nick"), "webchat", config.REALNAME
 156
 157     if not ticket is None:
 158       realname = "%s (%s:%d:%s)" % (realname, ticket.username, ticket.id, ticket.authflags)
 159
 160     if not nick:
 161       raise AJAXException("Nickname not supplied")
 162
 163     nick = nick[0]
 164
 165     for i in xrange(10):
 166       id = get_session_id()
 167       if not Sessions.get(id):
 168         break
 169     else:
 170       raise IDGenerationException()
 171
 172     session = IRCSession(id)
 173
 174     client = ircclient.createIRC(session, nick=nick, ident=ident, ip=ip, realname=realname)
 175     session.client = client
 176
 177     Sessions[id] = session
 178
 179     return id
 180
 181   def getSession(self, request):
 182     sessionid = request.args.get("s")
 183     if sessionid is None:
 184       raise AJAXException("Bad session ID")
 185
 186     session = Sessions.get(sessionid[0])
 187     if not session:
 188       raise AJAXException("Bad session ID")
 189     return session
 190
 191   def subscribe(self, request):
 192     self.getSession(request).subscribe(SingleUseChannel(request))
 193     return NOT_DONE_YET
 194
 195   def push(self, request):
 196     command = request.args.get("c")
 197     if command is None:
 198       raise AJAXException("No command specified")
 199
 200     command = command[0]
 201
 202     session = self.getSession(request)
 203
 204     try:
 205       decoded = command.decode("utf-8")
 206     except UnicodeDecodeError:
 207       decoded = command.decode("iso-8859-1", "ignore")
 208
 209     if len(decoded) > config.MAXLINELEN:
 210       session.disconnect()
 211       raise AJAXException("Line too long")
 212
 213     try:
 214       session.push(decoded)
 215     except AttributeError: # occurs when we haven't noticed an error
 216       session.disconnect()
 217       raise AJAXException("Connection closed by server.")
 218     except Exception, e: # catch all
 219       session.disconnect()
 220       traceback.print_exc(file=sys.stderr)
 221       raise AJAXException("Unknown error.")
 222
 223     return True
 224
 225   COMMANDS = dict(p=push, n=newConnection, s=subscribe)
 226