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 utils
  99 import xdb
 100 import avatar
 101 import session
 102 import jabw
 103 import disco
 104 import register
 105 import misciq
 106 import ft
 107 import lang
 108 import legacy
 109 import housekeep
 110
 111
 112
 113 class PyTransport(component.Service):
 114         def __init__(self):
 115                 LogEvent(INFO)
 116                 LogEvent(INFO, msg="Reactor: " + str(reactor))
 117
 118                 # Discovery, as well as some builtin features
 119                 self.discovery = disco.ServerDiscovery(self)
 120                 self.discovery.addIdentity("gateway", legacy.id, config.discoName, config.jid)
 121                 self.discovery.addIdentity("conference", "text", config.discoName + " Chatrooms", config.jid)
 122                 self.discovery.addFeature(disco.XCONFERENCE, None, config.jid) # So that clients know you can create groupchat rooms on the server
 123                 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
 124                 self.discovery.addIdentity("client", "pc", "MSN Messenger", "USER")
 125                 self.discovery.addIdentity("conference", "text", "MSN Groupchat", "ROOM")
 126
 127                 self.xdb = xdb.XDB(config.jid, legacy.mangle)
 128                 self.avatarCache = avatar.AvatarCache()
 129                 self.registermanager = register.RegisterManager(self)
 130                 self.gatewayTranslator = misciq.GatewayTranslator(self)
 131                 self.versionTeller = misciq.VersionTeller(self)
 132                 self.pingService = misciq.PingService(self)
 133                 self.adHocCommands = misciq.AdHocCommands(self)
 134                 self.vCardFactory = misciq.VCardFactory(self)
 135                 self.iqAvatarFactor = misciq.IqAvatarFactory(self)
 136                 self.connectUsers = misciq.ConnectUsers(self)
 137                 if config.ftJabberPort:
 138                         self.ftSOCKS5Receive = ft.Proxy65(int(config.ftJabberPort))
 139                         self.ftSOCKS5Send = misciq.Socks5FileTransfer(self)
 140                 if config.ftOOBPort:
 141                         self.ftOOBReceive = ft.FileTransferOOBReceive(int(config.ftOOBPort))
 142                         self.ftOOBSend = misciq.FileTransferOOBSend(self)
 143                 self.statistics = misciq.Statistics(self)
 144                 self.startTime = int(time.time())
 145
 146                 self.xmlstream = None
 147                 self.sessions = {}
 148
 149                 # Groupchat ID handling
 150                 self.lastID = 0
 151                 self.reservedIDs = []
 152
 153                 # Message IDs
 154                 self.messageID = 0
 155
 156                 self.loopTask = task.LoopingCall(self.loopFunc)
 157                 self.loopTask.start(60.0)
 158
 159         def removeMe(self):
 160                 LogEvent(INFO)
 161                 for session in self.sessions.copy():
 162                         self.sessions[session].removeMe()
 163
 164         def makeMessageID(self):
 165                 self.messageID += 1
 166                 return str(self.messageID)
 167
 168         def makeID(self):
 169                 newID = "r" + str(self.lastID)
 170                 self.lastID += 1
 171                 if self.reservedIDs.count(newID) > 0:
 172                         # Ack, it's already used.. Try again
 173                         return self.makeID()
 174                 else:
 175                         return newID
 176
 177         def reserveID(self, ID):
 178                 self.reservedIDs.append(ID)
 179
 180         def loopFunc(self):
 181                 numsessions = len(self.sessions)
 182
 183                 #if config.debugOn and numsessions > 0:
 184                 #       print "Sessions:"
 185                 #       for key in self.sessions:
 186                 #               print "\t" + self.sessions[key].jabberID
 187
 188                 self.statistics.stats["Uptime"] = int(time.time()) - self.startTime
 189                 self.statistics.stats["OnlineUsers"] = numsessions
 190                 legacy.updateStats(self.statistics)
 191                 if numsessions > 0:
 192                         oldDict = self.sessions.copy()
 193                         self.sessions = {}
 194                         for key in oldDict:
 195                                 s = oldDict[key]
 196                                 if not s.alive:
 197                                         LogEvent(WARN, "", "Ghost session found.")
 198                                         # Don't add it to the new dictionary. Effectively removing it
 199                                 else:
 200                                         self.sessions[key] = s
 201
 202         def componentConnected(self, xmlstream):
 203                 LogEvent(INFO)
 204                 self.xmlstream = xmlstream
 205                 self.xmlstream.addObserver("/iq", self.discovery.onIq)
 206                 self.xmlstream.addObserver("/presence", self.onPresence)
 207                 self.xmlstream.addObserver("/message", self.onMessage)
 208                 self.xmlstream.addObserver("/route", self.onRouteMessage)
 209                 if config.useXCP:
 210                         pres = Element((None, "presence"))
 211                         pres.attributes["to"] = "presence@-internal"
 212                         pres.attributes["from"] = config.compjid
 213                         x = pres.addElement("x")
 214                         x.attributes["xmlns"] = "http://www.jabber.com/schemas/component-presence.xsd"
 215                         x.attributes["xmlns:config"] = "http://www.jabber.com/config"
 216                         x.attributes["config:version"] = "1"
 217                         x.attributes["protocol-version"] = "1.0"
 218                         x.attributes["config-ns"] = legacy.url + "/component"
 219                         self.send(pres)
 220
 221         def componentDisconnected(self):
 222                 LogEvent(INFO)
 223                 self.xmlstream = None
 224
 225         def onRouteMessage(self, el):
 226                 for child in el.elements():
 227                         if child.name == "message":
 228                                 self.onMessage(child)
 229                         elif child.name == "presence":
 230                                 # Ignore any presence broadcasts about other XCP components
 231                                 if child.getAttribute("to") and child.getAttribute("to").find("@-internal") > 0: return
 232                                 self.onPresence(child)
 233                         elif child.name == "iq":
 234                                 self.discovery.onIq(child)
 235
 236         def onMessage(self, el):
 237                 fro = el.getAttribute("from")
 238                 try:
 239                         froj = internJID(fro)
 240                 except Exception, e:
 241                         LogEvent(WARN, "", "Failed stringprep.")
 242                         return
 243                 mtype = el.getAttribute("type")
 244                 s = self.sessions.get(froj.userhost(), None)
 245                 if mtype == "error" and s:
 246                         LogEvent(INFO, s.jabberID, "Removing session because of message type=error")
 247                         s.removeMe()
 248                 elif s:
 249                         s.onMessage(el)
 250                 elif mtype != "error":
 251                         to = el.getAttribute("to")
 252                         ulang = utils.getLang(el)
 253                         body = None
 254                         for child in el.elements():
 255                                 if child.name == "body":
 256                                         body = child.__str__()
 257                         LogEvent(INFO, "", "Sending error response to a message outside of session.")
 258                         jabw.sendErrorMessage(self, fro, to, "auth", "not-authorized", lang.get(ulang).notLoggedIn, body)
 259                         jabw.sendPresence(self, fro, to, ptype="unavailable")
 260
 261         def onPresence(self, el):
 262                 fro = el.getAttribute("from")
 263                 to = el.getAttribute("to")
 264                 try:
 265                         froj = internJID(fro)
 266                         toj = internJID(to)
 267                 except Exception, e:
 268                         LogEvent(WARN, "", "Failed stringprep.")
 269                         return
 270
 271                 ptype = el.getAttribute("type")
 272                 s = self.sessions.get(froj.userhost())
 273                 if ptype == "error" and s:
 274                         LogEvent(INFO, s.jabberID, "Removing session because of message type=error")
 275                         s.removeMe()
 276                 elif s:
 277                         s.onPresence(el)
 278                 else:
 279                         ulang = utils.getLang(el)
 280                         ptype = el.getAttribute("type")
 281                         if to.find('@') < 0:
 282                                 # If the presence packet is to the transport (not a user) and there isn't already a session
 283                                 if not el.getAttribute("type"): # Don't create a session unless they're sending available presence
 284                                         LogEvent(INFO, "", "Attempting to create a new session.")
 285                                         s = session.makeSession(self, froj.userhost(), ulang)
 286                                         if s:
 287                                                 self.statistics.stats["TotalUsers"] += 1
 288                                                 self.sessions[froj.userhost()] = s
 289                                                 LogEvent(INFO, "", "New session created.")
 290                                                 # Send the first presence
 291                                                 s.onPresence(el)
 292                                         else:
 293                                                 LogEvent(INFO, "", "Failed to create session")
 294                                                 jabw.sendMessage(self, to=froj.userhost(), fro=config.jid, body=lang.get(ulang).notRegistered)
 295
 296                                 elif el.getAttribute("type") != "error":
 297                                         LogEvent(INFO, "", "Sending unavailable presence to non-logged in user.")
 298                                         jabw.sendPresence(self, fro, to, ptype="unavailable")
 299                                         return
 300
 301                         elif ptype and (ptype.startswith("subscribe") or ptype.startswith("unsubscribe")):
 302                                 # They haven't logged in, and are trying to change subscription to a user
 303                                 # No, lets not log them in. Lets send an error :)
 304                                 jabw.sendPresence(self, fro, to, ptype="error")
 305
 306                                 # Lets log them in and then do it
 307                                 #LogEvent(INFO, "", "Attempting to create a session to do subscription stuff.")
 308                                 #s = session.makeSession(self, froj.userhost(), ulang)
 309                                 #if s:
 310                                 #       self.sessions[froj.userhost()] = s
 311                                 #       LogEvent(INFO, "", "New session created.")
 312                                 #       # Tell the session there's a new resource
 313                                 #       s.handleResourcePresence(froj.userhost(), froj.resource, toj.userhost(), toj.resource, 0, None, None, None)
 314                                 #       # Send this subscription
 315                                 #       s.onPresence(el)
 316
 317
 318 class App:
 319         def __init__(self):
 320                 # Check for any other instances
 321                 if config.pid and os.name != "posix":
 322                         config.pid = ""
 323                 if config.pid:
 324                         twistd.checkPID(config.pid)
 325
 326                 # Do any auto-update stuff
 327                 housekeep.init()
 328
 329                 # Daemonise the process and write the PID file
 330                 if config.background and os.name == "posix":
 331                         twistd.daemonize()
 332                 if config.pid:
 333                         self.writePID()
 334
 335                 # Initialise debugging, and do the other half of SIGHUPstuff
 336                 debug.reloadConfig()
 337                 legacy.reloadConfig()
 338
 339                 # Start the service
 340                 jid = config.jid
 341                 if config.useXCP and config.compjid:
 342                         jid = config.compjid
 343                 self.c = component.buildServiceManager(jid, config.secret, "tcp:%s:%s" % (config.mainServer, config.port))
 344                 self.transportSvc = PyTransport()
 345                 self.transportSvc.setServiceParent(self.c)
 346                 self.c.startService()
 347                 reactor.addSystemEventTrigger('before', 'shutdown', self.shuttingDown)
 348
 349         def writePID(self):
 350                 # Create a PID file
 351                 pid = str(os.getpid())
 352                 pf = open(config.pid, "w")
 353                 pf.write("%s\n" % pid)
 354                 pf.close()
 355
 356         def shuttingDown(self):
 357                 self.transportSvc.removeMe()
 358                 # Keep the transport running for another 3 seconds
 359                 def cb(ignored=None):
 360                         if config.pid:
 361                                 twistd.removePID(config.pid)
 362                 d = Deferred()
 363                 d.addCallback(cb)
 364                 reactor.callLater(3.0, d.callback, None)
 365                 return d
 366
 367
 368
 369 def SIGHUPstuff(*args):
 370         global configFile, configOptions
 371         xmlconfig.reloadConfig(configFile, configOptions)
 372         if config.pid and os.name != "posix":
 373                 config.pid = ""
 374         debug.reloadConfig()
 375         legacy.reloadConfig()
 376
 377 if os.name == "posix":
 378         import signal
 379         # Set SIGHUP to reload the config file & close & open debug file
 380         signal.signal(signal.SIGHUP, SIGHUPstuff)
 381         # Load some scripts for PID and daemonising
 382         from twisted.scripts import _twistd_unix as twistd
 383
 384
 385 def main():
 386         # Create the application
 387         app = App()
 388         reactor.run()
 389
 390 if __name__ == "__main__":
 391         main()
 392