code.delx.au - pymsnt/blob - src/main.py

   1 # Copyright 2004-2006 James Bunton <james@delx.cjb.net>
   2 # Licensed for distribution under the GPL version 2, check COPYING for details
   3
   4 import os, os.path, time, sys, codecs, getopt
   5 reload(sys)
   6 sys.setdefaultencoding("utf-8")
   7 sys.stdout = codecs.lookup('utf-8')[-1](sys.stdout)
   8
   9
  10 # Find the best reactor
  11 selectWarning = "Unable to install any good reactors (kqueue, epoll, poll).\nWe fell back to using select. You may have scalability problems.\nThis reactor will not support more than 1024 connections at a time."
  12 reactors = [("epollreactor", True), ("pollreactor", True), ("selectreactor", False), ("default", False)]
  13 for tryReactor, good in reactors:
  14         try:
  15                 bestReactor = __import__("twisted.internet." + tryReactor)
  16                 if not good:
  17                         print >> sys.stderr, selectWarning
  18                 break
  19         except ImportError:
  20                 pass
  21 else:
  22         print >> sys.stderr, "Unable to find a reactor. Please make sure you have Twisted properly installed.\nExiting..."
  23         sys.exit(1)
  24
  25
  26 import twistfix
  27 twistfix.main()
  28
  29
  30 # Must load config before everything else
  31 import config
  32 import xmlconfig
  33 configFile = "config.xml"
  34 configOptions = {}
  35 opts, args = getopt.getopt(sys.argv[1:], "bc:o:dDgtlp:h", ["background", "config=", "option=", "debug", "Debug", "garbage", "traceback", "log=", "pid=", "help"])
  36 for o, v in opts:
  37         if o in ("-c", "--config"):
  38                 configFile = v
  39         elif o in ("-b", "--background"):
  40                 config.background = True
  41         elif o in ("-d", "--debug"):
  42                 config.debugLevel = "2"
  43         elif o in ("-D", "--Debug"):
  44                 config.debugLevel = "3"
  45         elif o in ("-g", "--garbage"):
  46                 import gc
  47                 gc.set_debug(gc.DEBUG_LEAK|gc.DEBUG_STATS)
  48         elif o in ("-t", "--traceback"):
  49                 config.debugLevel = "1"
  50         elif o in ("-l", "--log"):
  51                 config.debugFile = v
  52         elif o in ("-p", "--pid"):
  53                 config.pid = v
  54         elif o in ("-o", "--option"):
  55                 var, setting = v.split("=", 2)
  56                 configOptions[var] = setting
  57         elif o in ("-h", "--help"):
  58                 print "%s [options]" % sys.argv[0]
  59                 print "   -h                  print this help"
  60                 print "   -b                  daemonize/background transport"
  61                 print "   -c <file>           read configuration from this file"
  62                 print "   -d                  print debugging output"
  63                 print "   -D                  print extended debugging output"
  64                 print "   -g                  print garbage collection output"
  65                 print "   -t                  print debugging only on traceback"
  66                 print "   -l <file>           write debugging output to file"
  67                 print "   -p <file>           write process ID to file"
  68                 print "   -o <var>=<setting>  set config var to setting"
  69                 sys.exit(0)
  70
  71 xmlconfig.reloadConfig(configFile, configOptions)
  72
  73 if config.reactor:
  74         # They picked their own reactor. Lets install it.
  75         del sys.modules["twisted.internet.reactor"]
  76         if config.reactor == "epoll":
  77                 from twisted.internet import epollreactor
  78                 epollreactor.install()
  79         elif config.reactor == "poll":
  80                 from twisted.internet import pollreactor
  81                 pollreactor.install()
  82         elif config.reactor == "kqueue":
  83                 from twisted.internet import kqreactor
  84                 kqreactor.install()
  85         elif len(config.reactor) > 0:
  86                 print >> sys.stderr, "Unknown reactor: ", config.reactor, ". Using best available reactor."
  87
  88
  89 from twisted.internet import reactor, task
  90 from twisted.internet.defer import Deferred
  91 from twisted.words.xish.domish import Element
  92 from twisted.words.protocols.jabber import component
  93 from twisted.words.protocols.jabber.jid import internJID
  94
  95 from debug import LogEvent, INFO, WARN, ERROR
  96
  97 import debug
  98 import svninfo
  99 import utils
 100 import xdb
 101 import avatar
 102 import session
 103 import jabw
 104 import disco
 105 import register
 106 import misciq
 107 import ft
 108 import lang
 109 import legacy
 110 import housekeep
 111
 112
 113
 114 class PyTransport(component.Service):
 115         def __init__(self):
 116                 LogEvent(INFO)
 117                 try:
 118                         LogEvent(INFO, msg="SVN r" + str(svninfo.getSVNVersion()))
 119                 except:
 120                         pass
 121                 LogEvent(INFO, msg="Reactor: " + str(reactor))
 122
 123                 # Discovery, as well as some builtin features
 124                 self.discovery = disco.ServerDiscovery(self)
 125                 self.discovery.addIdentity("gateway", legacy.id, config.discoName, config.jid)
 126                 self.discovery.addIdentity("conference", "text", config.discoName + " Chatrooms", config.jid)
 127                 self.discovery.addFeature(disco.XCONFERENCE, None, config.jid) # So that clients know you can create groupchat rooms on the server
 128                 self.discovery.addFeature("jabber:iq:conference", None, config.jid) # We don't actually support this, but Psi has a bug where it looks for this instead of the above
 129                 self.discovery.addIdentity("client", "pc", "MSN Messenger", "USER")
 130                 self.discovery.addIdentity("conference", "text", "MSN Groupchat", "ROOM")
 131
 132                 self.xdb = xdb.XDB(config.jid, legacy.mangle)
 133                 self.avatarCache = avatar.AvatarCache()
 134                 self.registermanager = register.RegisterManager(self)
 135                 self.gatewayTranslator = misciq.GatewayTranslator(self)
 136                 self.versionTeller = misciq.VersionTeller(self)
 137                 self.pingService = misciq.PingService(self)
 138                 self.adHocCommands = misciq.AdHocCommands(self)
 139                 self.vCardFactory = misciq.VCardFactory(self)
 140                 self.iqAvatarFactor = misciq.IqAvatarFactory(self)
 141                 self.connectUsers = misciq.ConnectUsers(self)
 142                 if config.ftJabberPort:
 143                         self.ftSOCKS5Receive = ft.Proxy65(int(config.ftJabberPort))
 144                         self.ftSOCKS5Send = misciq.Socks5FileTransfer(self)
 145                 if config.ftOOBPort:
 146                         self.ftOOBReceive = ft.FileTransferOOBReceive(int(config.ftOOBPort))
 147                         self.ftOOBSend = misciq.FileTransferOOBSend(self)
 148                 self.statistics = misciq.Statistics(self)
 149                 self.startTime = int(time.time())
 150
 151                 self.xmlstream = None
 152                 self.sessions = {}
 153
 154                 # Groupchat ID handling
 155                 self.lastID = 0
 156                 self.reservedIDs = []
 157
 158                 # Message IDs
 159                 self.messageID = 0
 160
 161                 self.loopTask = task.LoopingCall(self.loopFunc)
 162                 self.loopTask.start(60.0)
 163
 164         def removeMe(self):
 165                 LogEvent(INFO)
 166                 for session in self.sessions.copy():
 167                         self.sessions[session].removeMe()
 168
 169         def makeMessageID(self):
 170                 self.messageID += 1
 171                 return str(self.messageID)
 172
 173         def makeID(self):
 174                 newID = "r" + str(self.lastID)
 175                 self.lastID += 1
 176                 if self.reservedIDs.count(newID) > 0:
 177                         # Ack, it's already used.. Try again
 178                         return self.makeID()
 179                 else:
 180                         return newID
 181
 182         def reserveID(self, ID):
 183                 self.reservedIDs.append(ID)
 184
 185         def loopFunc(self):
 186                 numsessions = len(self.sessions)
 187
 188                 #if config.debugOn and numsessions > 0:
 189                 #       print "Sessions:"
 190                 #       for key in self.sessions:
 191                 #               print "\t" + self.sessions[key].jabberID
 192
 193                 self.statistics.stats["Uptime"] = int(time.time()) - self.startTime
 194                 self.statistics.stats["OnlineUsers"] = numsessions
 195                 legacy.updateStats(self.statistics)
 196                 if numsessions > 0:
 197                         oldDict = self.sessions.copy()
 198                         self.sessions = {}
 199                         for key in oldDict:
 200                                 s = oldDict[key]
 201                                 if not s.alive:
 202                                         LogEvent(WARN, "", "Ghost session found.")
 203                                         # Don't add it to the new dictionary. Effectively removing it
 204                                 else:
 205                                         self.sessions[key] = s
 206
 207         def componentConnected(self, xmlstream):
 208                 LogEvent(INFO)
 209                 self.xmlstream = xmlstream
 210                 self.xmlstream.addObserver("/iq", self.discovery.onIq)
 211                 self.xmlstream.addObserver("/presence", self.onPresence)
 212                 self.xmlstream.addObserver("/message", self.onMessage)
 213                 self.xmlstream.addObserver("/route", self.onRouteMessage)
 214                 if config.useXCP:
 215                         pres = Element((None, "presence"))
 216                         pres.attributes["to"] = "presence@-internal"
 217                         pres.attributes["from"] = config.compjid
 218                         x = pres.addElement("x")
 219                         x.attributes["xmlns"] = "http://www.jabber.com/schemas/component-presence.xsd"
 220                         x.attributes["xmlns:config"] = "http://www.jabber.com/config"
 221                         x.attributes["config:version"] = "1"
 222                         x.attributes["protocol-version"] = "1.0"
 223                         x.attributes["config-ns"] = legacy.url + "/component"
 224                         self.send(pres)
 225
 226         def componentDisconnected(self):
 227                 LogEvent(INFO)
 228                 self.xmlstream = None
 229
 230         def onRouteMessage(self, el):
 231                 for child in el.elements():
 232                         if child.name == "message":
 233                                 self.onMessage(child)
 234                         elif child.name == "presence":
 235                                 # Ignore any presence broadcasts about other XCP components
 236                                 if child.getAttribute("to") and child.getAttribute("to").find("@-internal") > 0: return
 237                                 self.onPresence(child)
 238                         elif child.name == "iq":
 239                                 self.discovery.onIq(child)
 240
 241         def onMessage(self, el):
 242                 fro = el.getAttribute("from")
 243                 try:
 244                         froj = internJID(fro)
 245                 except Exception, e:
 246                         LogEvent(WARN, "", "Failed stringprep.")
 247                         return
 248                 mtype = el.getAttribute("type")
 249                 s = self.sessions.get(froj.userhost(), None)
 250                 if mtype == "error" and s:
 251                         LogEvent(INFO, s.jabberID, "Removing session because of message type=error")
 252                         s.removeMe()
 253                 elif s:
 254                         s.onMessage(el)
 255                 elif mtype != "error":
 256                         to = el.getAttribute("to")
 257                         ulang = utils.getLang(el)
 258                         body = None
 259                         for child in el.elements():
 260                                 if child.name == "body":
 261                                         body = child.__str__()
 262                         LogEvent(INFO, "", "Sending error response to a message outside of session.")
 263                         jabw.sendErrorMessage(self, fro, to, "auth", "not-authorized", lang.get(ulang).notLoggedIn, body)
 264                         jabw.sendPresence(self, fro, to, ptype="unavailable")
 265
 266         def onPresence(self, el):
 267                 fro = el.getAttribute("from")
 268                 to = el.getAttribute("to")
 269                 try:
 270                         froj = internJID(fro)
 271                         toj = internJID(to)
 272                 except Exception, e:
 273                         LogEvent(WARN, "", "Failed stringprep.")
 274                         return
 275
 276                 ptype = el.getAttribute("type")
 277                 s = self.sessions.get(froj.userhost())
 278                 if ptype == "error" and s:
 279                         LogEvent(INFO, s.jabberID, "Removing session because of message type=error")
 280                         s.removeMe()
 281                 elif s:
 282                         s.onPresence(el)
 283                 else:
 284                         ulang = utils.getLang(el)
 285                         ptype = el.getAttribute("type")
 286                         if to.find('@') < 0:
 287                                 # If the presence packet is to the transport (not a user) and there isn't already a session
 288                                 if not el.getAttribute("type"): # Don't create a session unless they're sending available presence
 289                                         LogEvent(INFO, "", "Attempting to create a new session.")
 290                                         s = session.makeSession(self, froj.userhost(), ulang)
 291                                         if s:
 292                                                 self.statistics.stats["TotalUsers"] += 1
 293                                                 self.sessions[froj.userhost()] = s
 294                                                 LogEvent(INFO, "", "New session created.")
 295                                                 # Send the first presence
 296                                                 s.onPresence(el)
 297                                         else:
 298                                                 LogEvent(INFO, "", "Failed to create session")
 299                                                 jabw.sendMessage(self, to=froj.userhost(), fro=config.jid, body=lang.get(ulang).notRegistered)
 300
 301                                 elif el.getAttribute("type") != "error":
 302                                         LogEvent(INFO, "", "Sending unavailable presence to non-logged in user.")
 303                                         jabw.sendPresence(self, fro, to, ptype="unavailable")
 304                                         return
 305
 306                         elif ptype and (ptype.startswith("subscribe") or ptype.startswith("unsubscribe")):
 307                                 # They haven't logged in, and are trying to change subscription to a user
 308                                 # No, lets not log them in. Lets send an error :)
 309                                 jabw.sendPresence(self, fro, to, ptype="error")
 310
 311                                 # Lets log them in and then do it
 312                                 #LogEvent(INFO, "", "Attempting to create a session to do subscription stuff.")
 313                                 #s = session.makeSession(self, froj.userhost(), ulang)
 314                                 #if s:
 315                                 #       self.sessions[froj.userhost()] = s
 316                                 #       LogEvent(INFO, "", "New session created.")
 317                                 #       # Tell the session there's a new resource
 318                                 #       s.handleResourcePresence(froj.userhost(), froj.resource, toj.userhost(), toj.resource, 0, None, None, None)
 319                                 #       # Send this subscription
 320                                 #       s.onPresence(el)
 321
 322
 323 class App:
 324         def __init__(self):
 325                 # Check for any other instances
 326                 if config.pid and os.name != "posix":
 327                         config.pid = ""
 328                 if config.pid:
 329                         twistd.checkPID(config.pid)
 330
 331                 # Do any auto-update stuff
 332                 housekeep.init()
 333
 334                 # Daemonise the process and write the PID file
 335                 if config.background and os.name == "posix":
 336                         twistd.daemonize()
 337                 if config.pid:
 338                         self.writePID()
 339
 340                 # Initialise debugging, and do the other half of SIGHUPstuff
 341                 debug.reloadConfig()
 342                 legacy.reloadConfig()
 343
 344                 # Start the service
 345                 jid = config.jid
 346                 if config.useXCP and config.compjid:
 347                         jid = config.compjid
 348                 self.c = component.buildServiceManager(jid, config.secret, "tcp:%s:%s" % (config.mainServer, config.port))
 349                 self.transportSvc = PyTransport()
 350                 self.transportSvc.setServiceParent(self.c)
 351                 self.c.startService()
 352                 reactor.addSystemEventTrigger('before', 'shutdown', self.shuttingDown)
 353
 354         def writePID(self):
 355                 # Create a PID file
 356                 pid = str(os.getpid())
 357                 pf = open(config.pid, "w")
 358                 pf.write("%s\n" % pid)
 359                 pf.close()
 360
 361         def shuttingDown(self):
 362                 self.transportSvc.removeMe()
 363                 # Keep the transport running for another 3 seconds
 364                 def cb(ignored=None):
 365                         if config.pid:
 366                                 twistd.removePID(config.pid)
 367                 d = Deferred()
 368                 d.addCallback(cb)
 369                 reactor.callLater(3.0, d.callback, None)
 370                 return d
 371
 372
 373
 374 def SIGHUPstuff(*args):
 375         global configFile, configOptions
 376         xmlconfig.reloadConfig(configFile, configOptions)
 377         if config.pid and os.name != "posix":
 378                 config.pid = ""
 379         debug.reloadConfig()
 380         legacy.reloadConfig()
 381
 382 if os.name == "posix":
 383         import signal
 384         # Set SIGHUP to reload the config file & close & open debug file
 385         signal.signal(signal.SIGHUP, SIGHUPstuff)
 386         # Load some scripts for PID and daemonising
 387         from twisted.scripts import _twistd_unix as twistd
 388
 389
 390 def main():
 391         # Create the application
 392         app = App()
 393         reactor.run()
 394
 395 if __name__ == "__main__":
 396         main()
 397