qwebirc/engines/ajaxengine.py

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