qwebirc/engines/ajaxengine.py

   1 from twisted.web import resource, server, static, error as http_error
   2 from twisted.names import client
   3 from twisted.internet import reactor, error
   4 from authgateengine import login_optional, getSessionData
   5 import simplejson, md5, sys, os, time, config, weakref, traceback
   6 import qwebirc.ircclient as ircclient
   7 from adminengine import AdminEngineAction
   8 from qwebirc.util import HitCounter
   9
  10 Sessions = {}
  11
  12 def get_session_id():
  13   return md5.md5(os.urandom(16)).hexdigest()
  14
  15 class BufferOverflowException(Exception):
  16   pass
  17
  18 class AJAXException(Exception):
  19   pass
  20
  21 class IDGenerationException(Exception):
  22   pass
  23
  24 class PassthruException(Exception):
  25   pass
  26
  27 NOT_DONE_YET = None
  28
  29 def jsondump(fn):
  30   def decorator(*args, **kwargs):
  31     try:
  32       x = fn(*args, **kwargs)
  33       if x is None:
  34         return server.NOT_DONE_YET
  35       x = (True, x)
  36     except AJAXException, e:
  37       x = (False, e[0])
  38     except PassthruException, e:
  39       return str(e)
  40
  41     return simplejson.dumps(x)
  42   return decorator
  43
  44 def cleanupSession(id):
  45   try:
  46     del Sessions[id]
  47   except KeyError:
  48     pass
  49
  50 class IRCSession:
  51   def __init__(self, id):
  52     self.id = id
  53     self.subscriptions = []
  54     self.buffer = []
  55     self.throttle = 0
  56     self.schedule = None
  57     self.closed = False
  58     self.cleanupschedule = None
  59
  60   def subscribe(self, channel, notifier):
  61     timeout_entry = reactor.callLater(config.HTTP_AJAX_REQUEST_TIMEOUT, self.timeout, channel)
  62     def cancel_timeout(result):
  63       if channel in self.subscriptions:
  64         self.subscriptions.remove(channel)
  65       try:
  66         timeout_entry.cancel()
  67       except error.AlreadyCalled:
  68         pass
  69     notifier.addCallbacks(cancel_timeout, cancel_timeout)
  70
  71     if len(self.subscriptions) >= config.MAXSUBSCRIPTIONS:
  72       self.subscriptions.pop(0).close()
  73
  74     self.subscriptions.append(channel)
  75     self.flush()
  76
  77   def timeout(self, channel):
  78     if self.schedule:
  79       return
  80
  81     channel.write(simplejson.dumps([]))
  82     if channel in self.subscriptions:
  83       self.subscriptions.remove(channel)
  84
  85   def flush(self, scheduled=False):
  86     if scheduled:
  87       self.schedule = None
  88
  89     if not self.buffer or not self.subscriptions:
  90       return
  91
  92     t = time.time()
  93
  94     if t < self.throttle:
  95       if not self.schedule:
  96         self.schedule = reactor.callLater(self.throttle - t, self.flush, True)
  97       return
  98     else:
  99       # process the rest of the packet
 100       if not scheduled:
 101         if not self.schedule:
 102           self.schedule = reactor.callLater(0, self.flush, True)
 103         return
 104
 105     self.throttle = t + config.UPDATE_FREQ
 106
 107     encdata = simplejson.dumps(self.buffer)
 108     self.buffer = []
 109
 110     newsubs = []
 111     for x in self.subscriptions:
 112       if x.write(encdata):
 113         newsubs.append(x)
 114
 115     self.subscriptions = newsubs
 116     if self.closed and not self.subscriptions:
 117       cleanupSession(self.id)
 118
 119   def event(self, data):
 120     bufferlen = sum(map(len, self.buffer))
 121     if bufferlen + len(data) > config.MAXBUFLEN:
 122       self.buffer = []
 123       self.client.error("Buffer overflow.")
 124       return
 125
 126     self.buffer.append(data)
 127     self.flush()
 128
 129   def push(self, data):
 130     if not self.closed:
 131       self.client.write(data)
 132
 133   def disconnect(self):
 134     # keep the session hanging around for a few seconds so the
 135     # client has a chance to see what the issue was
 136     self.closed = True
 137
 138     reactor.callLater(5, cleanupSession, self.id)
 139
 140 class Channel:
 141   def __init__(self, request):
 142     self.request = request
 143
 144 class SingleUseChannel(Channel):
 145   def write(self, data):
 146     self.request.write(data)
 147     self.request.finish()
 148     return False
 149
 150   def close(self):
 151     self.request.finish()
 152
 153 class MultipleUseChannel(Channel):
 154   def write(self, data):
 155     self.request.write(data)
 156     return True
 157
 158 class AJAXEngine(resource.Resource):
 159   isLeaf = True
 160
 161   def __init__(self, prefix):
 162     self.prefix = prefix
 163     self.__connect_hit = HitCounter()
 164     self.__total_hit = HitCounter()
 165
 166   @jsondump
 167   def render_POST(self, request):
 168     path = request.path[len(self.prefix):]
 169     if path[0] == "/":
 170       handler = self.COMMANDS.get(path[1:])
 171       if handler is not None:
 172         return handler(self, request)
 173
 174     raise PassthruException, http_error.NoResource().render(request)
 175
 176   #def render_GET(self, request):
 177     #return self.render_POST(request)
 178
 179   def newConnection(self, request):
 180     ticket = login_optional(request)
 181
 182     _, ip, port = request.transport.getPeer()
 183
 184     nick = request.args.get("nick")
 185     if not nick:
 186       raise AJAXException, "Nickname not supplied."
 187     nick = ircclient.irc_decode(nick[0])
 188
 189     ident, realname = "webchat", config.REALNAME
 190
 191     for i in xrange(10):
 192       id = get_session_id()
 193       if not Sessions.get(id):
 194         break
 195     else:
 196       raise IDGenerationException()
 197
 198     session = IRCSession(id)
 199
 200     qticket = getSessionData(request).get("qticket")
 201     if qticket is None:
 202       perform = None
 203     else:
 204       perform = ["PRIVMSG %s :TICKETAUTH %s" % (config.QBOT, qticket)]
 205
 206     self.__connect_hit()
 207     client = ircclient.createIRC(session, nick=nick, ident=ident, ip=ip, realname=realname, perform=perform)
 208     session.client = client
 209
 210     Sessions[id] = session
 211
 212     return id
 213
 214   def getSession(self, request):
 215     bad_session_message = "Invalid session, this most likely means the server has restarted; close this dialog and then try refreshing the page."
 216
 217     sessionid = request.args.get("s")
 218     if sessionid is None:
 219       raise AJAXException, bad_session_message
 220
 221     session = Sessions.get(sessionid[0])
 222     if not session:
 223       raise AJAXException, bad_session_message
 224     return session
 225
 226   def subscribe(self, request):
 227     request.channel.cancelTimeout()
 228     self.getSession(request).subscribe(SingleUseChannel(request), request.notifyFinish())
 229     return NOT_DONE_YET
 230
 231   def push(self, request):
 232     command = request.args.get("c")
 233     if command is None:
 234       raise AJAXException, "No command specified."
 235     self.__total_hit()
 236
 237     decoded = ircclient.irc_decode(command[0])
 238
 239     session = self.getSession(request)
 240
 241     if len(decoded) > config.MAXLINELEN:
 242       session.disconnect()
 243       raise AJAXException, "Line too long."
 244
 245     try:
 246       session.push(decoded)
 247     except AttributeError: # occurs when we haven't noticed an error
 248       session.disconnect()
 249       raise AJAXException, "Connection closed by server; try reconnecting by reloading the page."
 250     except Exception, e: # catch all
 251       session.disconnect()
 252       traceback.print_exc(file=sys.stderr)
 253       raise AJAXException, "Unknown error."
 254
 255     return True
 256
 257   def closeById(self, k):
 258     s = Sessions.get(k)
 259     if s is None:
 260       return
 261     s.client.client.error("Closed by admin interface")
 262
 263   @property
 264   def adminEngine(self):
 265     return {
 266       "Sessions": [(str(v.client.client), AdminEngineAction("close", self.closeById, k)) for k, v in Sessions.iteritems() if not v.closed],
 267       "Connections": [(self.__connect_hit,)],
 268       "Total hits": [(self.__total_hit,)],
 269     }
 270
 271   COMMANDS = dict(p=push, n=newConnection, s=subscribe)
 272