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, qwebirc.config_options as config_options, 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     password = request.args.get("password")
 194     if password is not None:
 195       password = ircclient.irc_decode(password[0])
 196
 197     for i in xrange(10):
 198       id = get_session_id()
 199       if not Sessions.get(id):
 200         break
 201     else:
 202       raise IDGenerationException()
 203
 204     session = IRCSession(id)
 205
 206     qticket = getSessionData(request).get("qticket")
 207     if qticket is None:
 208       perform = None
 209     else:
 210       service_mask = config.AUTH_SERVICE
 211       msg_mask = service_mask.split("!")[0] + "@" + service_mask.split("@", 1)[1]
 212       perform = ["PRIVMSG %s :TICKETAUTH %s" % (msg_mask, qticket)]
 213
 214     ident, realname = config.IDENT, config.REALNAME
 215     if ident is config_options.IDENT_HEX or ident is None: # latter is legacy
 216       ident = socket.inet_aton(ip).encode("hex")
 217     elif ident is config_options.IDENT_NICKNAME:
 218       ident = nick
 219
 220     self.__connect_hit()
 221
 222     def proceed(hostname):
 223       kwargs = dict(nick=nick, ident=ident, ip=ip, realname=realname, perform=perform, hostname=hostname)
 224       if password is not None:
 225         kwargs["password"] = password
 226
 227       client = ircclient.createIRC(session, **kwargs)
 228       session.client = client
 229
 230     if not hasattr(config, "WEBIRC_MODE") or config.WEBIRC_MODE == "hmac":
 231       proceed(None)
 232     elif config.WEBIRC_MODE != "hmac":
 233       notice = lambda x: session.event(connect_notice(x))
 234       notice("Looking up your hostname...")
 235       def callback(hostname):
 236         notice("Found your hostname.")
 237         proceed(hostname)
 238       def errback(failure):
 239         notice("Couldn't look up your hostname!")
 240         proceed(ip)
 241       qdns.lookupAndVerifyPTR(ip, timeout=[config.DNS_TIMEOUT]).addCallbacks(callback, errback)
 242
 243     Sessions[id] = session
 244
 245     return id
 246
 247   def getSession(self, request):
 248     bad_session_message = "Invalid session, this most likely means the server has restarted; close this dialog and then try refreshing the page."
 249
 250     sessionid = request.args.get("s")
 251     if sessionid is None:
 252       raise AJAXException, bad_session_message
 253
 254     session = Sessions.get(sessionid[0])
 255     if not session:
 256       raise AJAXException, bad_session_message
 257     return session
 258
 259   def subscribe(self, request):
 260     request.channel.cancelTimeout()
 261     self.getSession(request).subscribe(SingleUseChannel(request), request.notifyFinish())
 262     return NOT_DONE_YET
 263
 264   def push(self, request):
 265     command = request.args.get("c")
 266     if command is None:
 267       raise AJAXException, "No command specified."
 268     self.__total_hit()
 269
 270     decoded = ircclient.irc_decode(command[0])
 271
 272     session = self.getSession(request)
 273
 274     if len(decoded) > config.MAXLINELEN:
 275       session.disconnect()
 276       raise AJAXException, "Line too long."
 277
 278     try:
 279       session.push(decoded)
 280     except AttributeError: # occurs when we haven't noticed an error
 281       session.disconnect()
 282       raise AJAXException, "Connection closed by server; try reconnecting by reloading the page."
 283     except Exception, e: # catch all
 284       session.disconnect()
 285       traceback.print_exc(file=sys.stderr)
 286       raise AJAXException, "Unknown error."
 287
 288     return True
 289
 290   def closeById(self, k):
 291     s = Sessions.get(k)
 292     if s is None:
 293       return
 294     s.client.client.error("Closed by admin interface")
 295
 296   @property
 297   def adminEngine(self):
 298     return {
 299       "Sessions": [(str(v.client.client), AdminEngineAction("close", self.closeById, k)) for k, v in Sessions.iteritems() if not v.closed],
 300       "Connections": [(self.__connect_hit,)],
 301       "Total hits": [(self.__total_hit,)],
 302     }
 303
 304   COMMANDS = dict(p=push, n=newConnection, s=subscribe)
 305