qwebirc/ajaxengine.py

   1 from twisted.web import resource, server, static
   2 from twisted.names import client
   3 from twisted.internet import reactor
   4 from authgateengine import login_optional, getSessionData
   5 import simplejson, md5, sys, os, ircclient, time, config, weakref, traceback
   6
   7 Sessions = {}
   8
   9 def get_session_id():
  10   return md5.md5(os.urandom(16)).hexdigest()
  11
  12 class BufferOverflowException(Exception):
  13   pass
  14
  15 class AJAXException(Exception):
  16   pass
  17
  18 class IDGenerationException(Exception):
  19   pass
  20
  21 NOT_DONE_YET = None
  22
  23 def jsondump(fn):
  24   def decorator(*args, **kwargs):
  25     try:
  26       x = fn(*args, **kwargs)
  27       if x is None:
  28         return server.NOT_DONE_YET
  29       x = (True, x)
  30     except AJAXException, e:
  31       x = (False, e[0])
  32
  33     return simplejson.dumps(x)
  34   return decorator
  35
  36 def cleanupSession(id):
  37   try:
  38     del Sessions[id]
  39   except KeyError:
  40     pass
  41
  42 class IRCSession:
  43   def __init__(self, id):
  44     self.id = id
  45     self.subscriptions = []
  46     self.buffer = []
  47     self.throttle = 0
  48     self.schedule = None
  49     self.closed = False
  50     self.cleanupschedule = None
  51
  52   def subscribe(self, channel):
  53     if len(self.subscriptions) >= config.MAXSUBSCRIPTIONS:
  54       self.subscriptions.pop(0).close()
  55
  56     self.subscriptions.append(channel)
  57     self.flush()
  58
  59   def flush(self, scheduled=False):
  60     if scheduled:
  61       self.schedule = None
  62
  63     if not self.buffer or not self.subscriptions:
  64       return
  65
  66     t = time.time()
  67
  68     if t < self.throttle:
  69       if not self.schedule:
  70         self.schedule = reactor.callLater(self.throttle - t, self.flush, True)
  71       return
  72     else:
  73       # process the rest of the packet
  74       if not scheduled:
  75         if not self.schedule:
  76           self.schedule = reactor.callLater(0, self.flush, True)
  77         return
  78
  79     self.throttle = t + config.UPDATE_FREQ
  80
  81     encdata = simplejson.dumps(self.buffer)
  82     self.buffer = []
  83
  84     newsubs = []
  85     for x in self.subscriptions:
  86       if x.write(encdata):
  87         newsubs.append(x)
  88
  89     self.subscriptions = newsubs
  90     if self.closed and not self.subscriptions:
  91       cleanupSession(self.id)
  92
  93   def event(self, data):
  94     bufferlen = sum(map(len, self.buffer))
  95     if bufferlen + len(data) > config.MAXBUFLEN:
  96       self.buffer = []
  97       self.client.error("Buffer overflow")
  98       return
  99
 100     self.buffer.append(data)
 101     self.flush()
 102
 103   def push(self, data):
 104     if not self.closed:
 105       self.client.write(data)
 106
 107   def disconnect(self):
 108     # keep the session hanging around for a few seconds so the
 109     # client has a chance to see what the issue was
 110     self.closed = True
 111
 112     reactor.callLater(5, cleanupSession, self.id)
 113
 114 class Channel:
 115   def __init__(self, request):
 116     self.request = request
 117
 118 class SingleUseChannel(Channel):
 119   def write(self, data):
 120     self.request.write(data)
 121     self.request.finish()
 122     return False
 123
 124   def close(self):
 125     self.request.finish()
 126
 127 class MultipleUseChannel(Channel):
 128   def write(self, data):
 129     self.request.write(data)
 130     return True
 131
 132 class AJAXEngine(resource.Resource):
 133   isLeaf = True
 134
 135   def __init__(self, prefix):
 136     self.prefix = prefix
 137
 138   @jsondump
 139   def render_POST(self, request):
 140     path = request.path[len(self.prefix):]
 141     if path[0] == "/":
 142       handler = self.COMMANDS.get(path[1:])
 143       if handler is not None:
 144         return handler(self, request)
 145     raise AJAXException("404")
 146
 147 #  def render_GET(self, request):
 148 #    return self.render_POST(request)
 149
 150   def newConnection(self, request):
 151     ticket = login_optional(request)
 152
 153     _, ip, port = request.transport.getPeer()
 154
 155     nick, ident, realname = request.args.get("nick"), "webchat", config.REALNAME
 156
 157     if not nick:
 158       raise AJAXException("Nickname not supplied")
 159
 160     nick = nick[0]
 161
 162     for i in xrange(10):
 163       id = get_session_id()
 164       if not Sessions.get(id):
 165         break
 166     else:
 167       raise IDGenerationException()
 168
 169     session = IRCSession(id)
 170
 171     qticket = getSessionData(request).get("qticket")
 172     if qticket is None:
 173       perform = None
 174     else:
 175       perform = ["PRIVMSG %s :TICKETAUTH %s" % (config.QBOT, qticket)]
 176
 177     client = ircclient.createIRC(session, nick=nick, ident=ident, ip=ip, realname=realname, perform=perform)
 178     session.client = client
 179
 180     Sessions[id] = session
 181
 182     return id
 183
 184   def getSession(self, request):
 185     sessionid = request.args.get("s")
 186     if sessionid is None:
 187       raise AJAXException("Bad session ID")
 188
 189     session = Sessions.get(sessionid[0])
 190     if not session:
 191       raise AJAXException("Bad session ID")
 192     return session
 193
 194   def subscribe(self, request):
 195     self.getSession(request).subscribe(SingleUseChannel(request))
 196     return NOT_DONE_YET
 197
 198   def push(self, request):
 199     command = request.args.get("c")
 200     if command is None:
 201       raise AJAXException("No command specified")
 202
 203     command = command[0]
 204
 205     session = self.getSession(request)
 206
 207     try:
 208       decoded = command.decode("utf-8")
 209     except UnicodeDecodeError:
 210       decoded = command.decode("iso-8859-1", "ignore")
 211
 212     if len(decoded) > config.MAXLINELEN:
 213       session.disconnect()
 214       raise AJAXException("Line too long")
 215
 216     try:
 217       session.push(decoded)
 218     except AttributeError: # occurs when we haven't noticed an error
 219       session.disconnect()
 220       raise AJAXException("Connection closed by server.")
 221     except Exception, e: # catch all
 222       session.disconnect()
 223       traceback.print_exc(file=sys.stderr)
 224       raise AJAXException("Unknown error.")
 225
 226     return True
 227
 228   COMMANDS = dict(p=push, n=newConnection, s=subscribe)
 229