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 import qwebirc.dns as qdns
  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.buflen = 0
  56     self.throttle = 0
  57     self.schedule = None
  58     self.closed = False
  59     self.cleanupschedule = None
  60
  61   def subscribe(self, channel, notifier):
  62     timeout_entry = reactor.callLater(config.HTTP_AJAX_REQUEST_TIMEOUT, self.timeout, channel)
  63     def cancel_timeout(result):
  64       if channel in self.subscriptions:
  65         self.subscriptions.remove(channel)
  66       try:
  67         timeout_entry.cancel()
  68       except error.AlreadyCalled:
  69         pass
  70     notifier.addCallbacks(cancel_timeout, cancel_timeout)
  71
  72     if len(self.subscriptions) >= config.MAXSUBSCRIPTIONS:
  73       self.subscriptions.pop(0).close()
  74
  75     self.subscriptions.append(channel)
  76     self.flush()
  77
  78   def timeout(self, channel):
  79     if self.schedule:
  80       return
  81
  82     channel.write(simplejson.dumps([]))
  83     if channel in self.subscriptions:
  84       self.subscriptions.remove(channel)
  85
  86   def flush(self, scheduled=False):
  87     if scheduled:
  88       self.schedule = None
  89
  90     if not self.buffer or not self.subscriptions:
  91       return
  92
  93     t = time.time()
  94
  95     if t < self.throttle:
  96       if not self.schedule:
  97         self.schedule = reactor.callLater(self.throttle - t, self.flush, True)
  98       return
  99     else:
 100       # process the rest of the packet
 101       if not scheduled:
 102         if not self.schedule:
 103           self.schedule = reactor.callLater(0, self.flush, True)
 104         return
 105
 106     self.throttle = t + config.UPDATE_FREQ
 107
 108     encdata = simplejson.dumps(self.buffer)
 109     self.buffer = []
 110     self.buflen = 0
 111
 112     newsubs = []
 113     for x in self.subscriptions:
 114       if x.write(encdata):
 115         newsubs.append(x)
 116
 117     self.subscriptions = newsubs
 118     if self.closed and not self.subscriptions:
 119       cleanupSession(self.id)
 120
 121   def event(self, data):
 122     newbuflen = self.buflen + len(data)
 123     if newbuflen > config.MAXBUFLEN:
 124       self.buffer = []
 125       self.client.error("Buffer overflow.")
 126       return
 127
 128     self.buffer.append(data)
 129     self.buflen = newbuflen
 130     self.flush()
 131
 132   def push(self, data):
 133     if not self.closed:
 134       self.client.write(data)
 135
 136   def disconnect(self):
 137     # keep the session hanging around for a few seconds so the
 138     # client has a chance to see what the issue was
 139     self.closed = True
 140
 141     reactor.callLater(5, cleanupSession, self.id)
 142
 143 # DANGER! Breach of encapsulation!
 144 def connect_notice(line):
 145   return "c", "NOTICE", "", ("AUTH", "*** (qwebirc) %s" % line)
 146
 147 class Channel:
 148   def __init__(self, request):
 149     self.request = request
 150
 151 class SingleUseChannel(Channel):
 152   def write(self, data):
 153     self.request.write(data)
 154     self.request.finish()
 155     return False
 156
 157   def close(self):
 158     self.request.finish()
 159
 160 class MultipleUseChannel(Channel):
 161   def write(self, data):
 162     self.request.write(data)
 163     return True
 164
 165 class AJAXEngine(resource.Resource):
 166   isLeaf = True
 167
 168   def __init__(self, prefix):
 169     self.prefix = prefix
 170     self.__connect_hit = HitCounter()
 171     self.__total_hit = HitCounter()
 172
 173   @jsondump
 174   def render_POST(self, request):
 175     path = request.path[len(self.prefix):]
 176     if path[0] == "/":
 177       handler = self.COMMANDS.get(path[1:])
 178       if handler is not None:
 179         return handler(self, request)
 180
 181     raise PassthruException, http_error.NoResource().render(request)
 182
 183   def newConnection(self, request):
 184     ticket = login_optional(request)
 185
 186     _, ip, port = request.transport.getPeer()
 187
 188     nick = request.args.get("nick")
 189     if not nick:
 190       raise AJAXException, "Nickname not supplied."
 191     nick = ircclient.irc_decode(nick[0])
 192
 193     for i in xrange(10):
 194       id = get_session_id()
 195       if not Sessions.get(id):
 196         break
 197     else:
 198       raise IDGenerationException()
 199
 200     session = IRCSession(id)
 201
 202     qticket = getSessionData(request).get("qticket")
 203     if qticket is None:
 204       perform = None
 205     else:
 206       service_mask = config.AUTH_SERVICE
 207       msg_mask = service_mask.split("!")[0] + "@" + service_mask.split("@", 1)[1]
 208       perform = ["PRIVMSG %s :TICKETAUTH %s" % (msg_mask, qticket)]
 209
 210     ident, realname = config.IDENT, config.REALNAME
 211     if ident is None:
 212       ident = socket.inet_aton(ip).encode("hex")
 213
 214     self.__connect_hit()
 215
 216     def proceed(hostname):
 217       client = ircclient.createIRC(session, nick=nick, ident=ident, ip=ip, realname=realname, perform=perform, hostname=hostname)
 218       session.client = client
 219
 220     if config.WEBIRC_MODE != "hmac":
 221       notice = lambda x: session.event(connect_notice(x))
 222       notice("Looking up your hostname...")
 223       def callback(hostname):
 224         notice("Found your hostname.")
 225         proceed(hostname)
 226       def errback(failure):
 227         notice("Couldn't look up your hostname!")
 228         proceed(ip)
 229       qdns.lookupAndVerifyPTR(ip, timeout=[config.DNS_TIMEOUT]).addCallbacks(callback, errback)
 230     else:
 231       proceed(None) # hmac doesn't care
 232
 233     Sessions[id] = session
 234
 235     return id
 236
 237   def getSession(self, request):
 238     bad_session_message = "Invalid session, this most likely means the server has restarted; close this dialog and then try refreshing the page."
 239
 240     sessionid = request.args.get("s")
 241     if sessionid is None:
 242       raise AJAXException, bad_session_message
 243
 244     session = Sessions.get(sessionid[0])
 245     if not session:
 246       raise AJAXException, bad_session_message
 247     return session
 248
 249   def subscribe(self, request):
 250     request.channel.cancelTimeout()
 251     self.getSession(request).subscribe(SingleUseChannel(request), request.notifyFinish())
 252     return NOT_DONE_YET
 253
 254   def push(self, request):
 255     command = request.args.get("c")
 256     if command is None:
 257       raise AJAXException, "No command specified."
 258     self.__total_hit()
 259
 260     decoded = ircclient.irc_decode(command[0])
 261
 262     session = self.getSession(request)
 263
 264     if len(decoded) > config.MAXLINELEN:
 265       session.disconnect()
 266       raise AJAXException, "Line too long."
 267
 268     try:
 269       session.push(decoded)
 270     except AttributeError: # occurs when we haven't noticed an error
 271       session.disconnect()
 272       raise AJAXException, "Connection closed by server; try reconnecting by reloading the page."
 273     except Exception, e: # catch all
 274       session.disconnect()
 275       traceback.print_exc(file=sys.stderr)
 276       raise AJAXException, "Unknown error."
 277
 278     return True
 279
 280   def closeById(self, k):
 281     s = Sessions.get(k)
 282     if s is None:
 283       return
 284     s.client.client.error("Closed by admin interface")
 285
 286   @property
 287   def adminEngine(self):
 288     return {
 289       "Sessions": [(str(v.client.client), AdminEngineAction("close", self.closeById, k)) for k, v in Sessions.iteritems() if not v.closed],
 290       "Connections": [(self.__connect_hit,)],
 291       "Total hits": [(self.__total_hit,)],
 292     }
 293
 294   COMMANDS = dict(p=push, n=newConnection, s=subscribe)
 295