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