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, socket
   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     for i in xrange(10):
 190       id = get_session_id()
 191       if not Sessions.get(id):
 192         break
 193     else:
 194       raise IDGenerationException()
 195
 196     session = IRCSession(id)
 197
 198     qticket = getSessionData(request).get("qticket")
 199     if qticket is None:
 200       perform = None
 201     else:
 202       service_mask = config.AUTH_SERVICE
 203       msg_mask = service_mask.split("!")[0] + "@" + service_mask.split("@", 1)[1]
 204       perform = ["PRIVMSG %s :TICKETAUTH %s" % (msg_mask, qticket)]
 205
 206     ident, realname = config.IDENT, config.REALNAME
 207     if ident is None:
 208       ident = socket.inet_aton(ip).encode("hex")
 209
 210     self.__connect_hit()
 211     client = ircclient.createIRC(session, nick=nick, ident=ident, ip=ip, realname=realname, perform=perform, hostname=ip)
 212     session.client = client
 213
 214     Sessions[id] = session
 215
 216     return id
 217
 218   def getSession(self, request):
 219     bad_session_message = "Invalid session, this most likely means the server has restarted; close this dialog and then try refreshing the page."
 220
 221     sessionid = request.args.get("s")
 222     if sessionid is None:
 223       raise AJAXException, bad_session_message
 224
 225     session = Sessions.get(sessionid[0])
 226     if not session:
 227       raise AJAXException, bad_session_message
 228     return session
 229
 230   def subscribe(self, request):
 231     request.channel.cancelTimeout()
 232     self.getSession(request).subscribe(SingleUseChannel(request), request.notifyFinish())
 233     return NOT_DONE_YET
 234
 235   def push(self, request):
 236     command = request.args.get("c")
 237     if command is None:
 238       raise AJAXException, "No command specified."
 239     self.__total_hit()
 240
 241     decoded = ircclient.irc_decode(command[0])
 242
 243     session = self.getSession(request)
 244
 245     if len(decoded) > config.MAXLINELEN:
 246       session.disconnect()
 247       raise AJAXException, "Line too long."
 248
 249     try:
 250       session.push(decoded)
 251     except AttributeError: # occurs when we haven't noticed an error
 252       session.disconnect()
 253       raise AJAXException, "Connection closed by server; try reconnecting by reloading the page."
 254     except Exception, e: # catch all
 255       session.disconnect()
 256       traceback.print_exc(file=sys.stderr)
 257       raise AJAXException, "Unknown error."
 258
 259     return True
 260
 261   def closeById(self, k):
 262     s = Sessions.get(k)
 263     if s is None:
 264       return
 265     s.client.client.error("Closed by admin interface")
 266
 267   @property
 268   def adminEngine(self):
 269     return {
 270       "Sessions": [(str(v.client.client), AdminEngineAction("close", self.closeById, k)) for k, v in Sessions.iteritems() if not v.closed],
 271       "Connections": [(self.__connect_hit,)],
 272       "Total hits": [(self.__total_hit,)],
 273     }
 274
 275   COMMANDS = dict(p=push, n=newConnection, s=subscribe)
 276