source: ralphm-patches/xmpp_client_service.patch @ 39:d6a0f8cbabf3

new file doc/examples/client_service.tac

@@ +1 @@
+from twisted.application import service, strports
+from twisted.internet import defer
+
+from wokkel import client, xmppim
+from wokkel.component import InternalComponent, Router
+from wokkel.generic import FallbackHandler
+from wokkel.ping import PingHandler
+from wokkel.xmppim import RosterItem
+
+from twisted.words.protocols.jabber.jid import internJID as JID
+
+import socket
+domain = socket.gethostname()
+
+RALPHM = JID('ralphm@'+domain)
+INTOSI = JID('intosi@'+domain)
+TERMIE = JID('termie@'+domain)
+
+roster = {
+    'ralphm': {
+        INTOSI: RosterItem(INTOSI,
+                           subscriptionTo=True,
+                           subscriptionFrom=True,
+                           name='Intosi',
+                           groups=set(['Friends'])),
+        TERMIE: RosterItem(TERMIE,
+                           subscriptionTo=True,
+                           subscriptionFrom=True,
+                           name='termie'),
+        },
+    'termie': {
+        RALPHM: RosterItem(RALPHM,
+                           subscriptionTo=True,
+                           subscriptionFrom=True,
+                           name='ralphm'),
+        }
+    }
+
+accounts = set(roster.keys())
+
+
+class StaticRoster(xmppim.RosterServerProtocol):
+
+    def __init__(self, roster):
+        xmppim.RosterServerProtocol.__init__(self)
+        self.roster = roster
+
+    def getRoster(self, entity):
+        return defer.succeed(self.roster[entity.user].values())
+
+
+
+application = service.Application("Jabber server")
+
+router = Router()
+component = InternalComponent(router, domain)
+component.setServiceParent(application)
+
+sessionManager = client.SessionManager(domain, accounts)
+sessionManager.setHandlerParent(component)
+
+xmppim.AccountIQHandler(sessionManager).setHandlerParent(component)
+xmppim.AccountMessageHandler(sessionManager).setHandlerParent(component)
+xmppim.PresenceServerHandler(sessionManager, domain, roster).setHandlerParent(component)
+FallbackHandler().setHandlerParent(component)
+StaticRoster(roster).setHandlerParent(component)
+PingHandler().setHandlerParent(component)
+
+c2sFactory = client.XMPPC2SServerFactory(sessionManager)
+c2sFactory.logTraffic = True
+c2sService = strports.service('5224', c2sFactory)
+c2sService.setServiceParent(application)
+
+sessionManager.connectionManager = c2sFactory

wokkel/client.py

@@ -10 +10 @@
 that should probably eventually move there.
 """
 
+import base64
+
 from twisted.application import service
-from twisted.internet import reactor
+from twisted.internet import defer, reactor
 from twisted.names.srvconnect import SRVConnector
-from twisted.words.protocols.jabber import client, sasl, xmlstream
+from twisted.python import log, randbytes
+from twisted.words.protocols.jabber import client, error, sasl, xmlstream
+from twisted.words.protocols.jabber.jid import JID, internJID
+from twisted.words.xish import domish
 
 from wokkel import generic
-from wokkel.subprotocols import StreamManager
+from wokkel.compat import XmlStreamServerFactory
+from wokkel.subprotocols import StreamManager, XMPPHandler
+
+NS_CLIENT = 'jabber:client'
+
+XPATH_ALL = "/*"
+XPATH_AUTH = "/auth[@xmlns='%s']" % sasl.NS_XMPP_SASL
+XPATH_BIND = "/iq[@type='set']/bind[@xmlns='%s']" % client.NS_XMPP_BIND
+XPATH_SESSION = "/iq[@type='set']/session[@xmlns='%s']" % \
+                client.NS_XMPP_SESSION
 
 class CheckAuthInitializer(object):
     """
@@ -51 +65 @@
     autentication.
     """
 
-    namespace = 'jabber:client'
+    namespace = NS_CLIENT
 
     def __init__(self, jid, password):
         xmlstream.ConnectAuthenticator.__init__(self, jid.host)
@@ -186 +200 @@
     c = XMPPClientConnector(reactor, domain, factory)
     c.connect()
     return factory.deferred
+
+
+
+class XMPPClientListenAuthenticator(xmlstream.ListenAuthenticator):
+    namespace = NS_CLIENT
+
+    def __init__(self, service):
+        self.service = service
+        self.failureGrace = 3
+        self.state = 'auth'
+
+
+    def associateWithStream(self, xs):
+        xmlstream.ListenAuthenticator.associateWithStream(self, xs)
+        self.xmlstream.addObserver(XPATH_ALL, self.onElementFallback, -1)
+
+
+    def onElementFallback(self, element):
+        if element.handled:
+            return
+
+        exc = error.StreamError('not-authorized')
+        self.xmlstream.sendStreamError(exc)
+
+
+    def streamStarted(self, rootElement):
+        xmlstream.ListenAuthenticator.streamStarted(self, rootElement)
+
+        # check namespace
+        #if self.xmlstream.namespace != self.namespace:
+        #    self.xmlstream.namespace = self.namespace
+        #    exc = error.StreamError('invalid-namespace')
+        #    self.xmlstream.sendStreamError(exc)
+        #    return
+
+        # TODO: check domain (self.service.domain)
+
+        self.xmlstream.sendHeader()
+
+        try:
+            stateHandlerName = 'streamStarted_' + self.state
+            stateHandler = getattr(self, stateHandlerName)
+        except AttributeError:
+            log.msg('streamStarted handler for', self.state, 'not found')
+        else:
+            stateHandler()
+
+
+    def toState(self, state):
+        self.state = state
+        if state == 'initialized':
+            self.xmlstream.removeObserver(XPATH_ALL, self.onElementFallback)
+            self.xmlstream.addOnetimeObserver(XPATH_SESSION, self.onSession, 1)
+            self.xmlstream.dispatch(self.xmlstream,
+                                    xmlstream.STREAM_AUTHD_EVENT)
+
+
+    def streamStarted_auth(self):
+        features = domish.Element((xmlstream.NS_STREAMS, 'features'))
+        features.addElement((sasl.NS_XMPP_SASL, 'mechanisms'))
+        features.mechanisms.addElement('mechanism', content='PLAIN')
+        self.xmlstream.send(features)
+        self.xmlstream.addOnetimeObserver(XPATH_AUTH, self.onAuth)
+
+
+    def onAuth(self, auth):
+        auth.handled = True
+
+        if auth.getAttribute('mechanism') != 'PLAIN':
+            failure = domish.Element((sasl.NS_XMPP_SASL, 'failure'))
+            failure.addElement('invalid-mechanism')
+            self.xmlstream.send(failure)
+
+            # Close stream on too many failing authentication attempts
+            self.failureGrace -= 1
+            if self.failureGrace == 0:
+                self.xmlstream.sendFooter()
+            else:
+                self.xmlstream.addOnetimeObserver(XPATH_AUTH, self.onAuth)
+
+            return
+
+        initialResponse = base64.b64decode(unicode(auth))
+        authzid, authcid, passwd = initialResponse.split('\x00')
+
+        # TODO: check passwd
+
+        # authenticated
+
+        self.username = authcid
+
+        success = domish.Element((sasl.NS_XMPP_SASL, 'success'))
+        self.xmlstream.send(success)
+        self.xmlstream.reset()
+
+        self.toState('bind')
+
+
+    def streamStarted_bind(self):
+        features = domish.Element((xmlstream.NS_STREAMS, 'features'))
+        features.addElement((client.NS_XMPP_BIND, 'bind'))
+        features.addElement((client.NS_XMPP_SESSION, 'session'))
+        self.xmlstream.send(features)
+        self.xmlstream.addOnetimeObserver(XPATH_BIND, self.onBind)
+
+
+    def onBind(self, iq):
+        def cb(boundJID):
+            self.xmlstream.otherEntity = boundJID
+            self.toState('initialized')
+
+            response = xmlstream.toResponse(iq, 'result')
+            response.addElement((client.NS_XMPP_BIND, 'bind'))
+            response.bind.addElement((client.NS_XMPP_BIND, 'jid'),
+                                  content=boundJID.full())
+
+            return response
+
+        def eb(failure):
+            if not isinstance(failure, error.StanzaError):
+                log.msg(failure)
+                exc = error.StanzaError('internal-server-error')
+            else:
+                exc = failure.value
+
+            return exc.toResponse(iq)
+
+        iq.handled = True
+        resource = unicode(iq.bind) or None
+        d = self.service.bindResource(self.username,
+                                      self.service.domain,
+                                      resource)
+        d.addCallback(cb)
+        d.addErrback(eb)
+        d.addCallback(self.xmlstream.send)
+
+
+    def onSession(self, iq):
+        iq.handled = True
+
+        reply = domish.Element((None, 'iq'))
+        reply['type'] = 'result'
+        if iq.getAttribute('id'):
+            reply['id'] = iq['id']
+        reply.addElement((client.NS_XMPP_SESSION, 'session'))
+        self.xmlstream.send(reply)
+
+
+
+class RecipientUnavailable(Exception):
+    """
+    The addressed entity is not, or no longer, available.
+    """
+
+
+
+class XMPPC2SServerFactory(XmlStreamServerFactory):
+
+    def __init__(self, service):
+        self.service = service
+
+        def authenticatorFactory():
+            return XMPPClientListenAuthenticator(service)
+
+        XmlStreamServerFactory.__init__(self, authenticatorFactory)
+        self.addBootstrap(xmlstream.STREAM_CONNECTED_EVENT,
+                          self.onConnectionMade)
+        self.addBootstrap(xmlstream.STREAM_AUTHD_EVENT,
+                          self.onAuthenticated)
+
+        self.serial = 0
+        self.streams = {}
+
+
+    def onConnectionMade(self, xs):
+        """
+        Called when a client-to-server connection was made.
+
+        This enables traffic debugging on incoming streams.
+        """
+        xs.serial = self.serial
+        self.serial += 1
+
+        log.msg("Client connection %d made" % xs.serial)
+
+        def logDataIn(buf):
+            log.msg("RECV (%d): %r" % (xs.serial, buf))
+
+        def logDataOut(buf):
+            log.msg("SEND (%d): %r" % (xs.serial, buf))
+
+        if self.logTraffic:
+            xs.rawDataInFn = logDataIn
+            xs.rawDataOutFn = logDataOut
+
+        xs.addObserver(xmlstream.STREAM_ERROR_EVENT, self.onError)
+
+
+    def onAuthenticated(self, xs):
+        log.msg("Client connection %d authenticated" % xs.serial)
+
+        xs.addObserver(xmlstream.STREAM_END_EVENT, self.onConnectionLost,
+                                                   0, xs)
+        xs.addObserver('/*', self.onElement, 0, xs)
+
+        # Record this stream as bound to the authenticated JID
+        self.streams[xs.otherEntity] = xs
+
+
+    def onConnectionLost(self, xs, reason):
+        log.msg("Client connection %d disconnected" % xs.serial)
+
+        entity = xs.otherEntity
+        self.service.unbindResource(entity.user,
+                                    entity.host,
+                                    entity.resource,
+                                    reason)
+
+        # If the lost connections had been bound, remove the reference
+        if xs.otherEntity in self.streams:
+            del self.streams[xs.otherEntity]
+
+
+    def onError(self, reason):
+        log.err(reason, "Stream Error")
+
+
+    def onElement(self, xs, stanza):
+        """
+        Called when an element was received from one of the connected streams.
+
+        """
+        if stanza.handled:
+            return
+        else:
+            self.service.onElement(stanza, xs.otherEntity)
+
+
+    def deliverStanza(self, element, recipient):
+        if recipient in self.streams:
+            self.streams[recipient].send(element)
+        else:
+            raise RecipientUnavailable(u"There is no connection for %s" %
+                                       recipient.full())
+
+
+class Session(object):
+    def __init__(self, entity):
+        self.entity = entity
+        self.connected = False
+        self.interested = False
+        self.presence = None
+
+
+
+class SessionManager(XMPPHandler):
+
+    """
+    Session Manager.
+
+    @ivar xmlstream: XML Stream to inject incoming stanzas from client
+        connections into. Stanzas where the C{'to'} attribute is not set
+        or is directed at the local domain are injected as if received on
+        the XML Stream (using C{dispatch}), other stanzas are injected as if
+        they were sent from the XML Stream (using C{send}).
+    """
+
+    def __init__(self, domain, accounts):
+        XMPPHandler.__init__(self)
+        self.domain = domain
+        self.accounts = accounts
+
+        self.connectionManager = None
+        self.sessions = {}
+
+
+    def bindResource(self, localpart, domain, resource):
+        if domain != self.domain:
+            raise Exception("I don't host this domain!")
+
+        try:
+            userSessions = self.sessions[localpart]
+        except KeyError:
+            userSessions = self.sessions[localpart] = {}
+
+        if resource is None:
+            resource = randbytes.secureRandom(8).encode('hex')
+        elif resource in self.userSessions:
+            resource = resource + ' ' + randbytes.secureRandom(8).encode('hex')
+
+        entity = JID(tuple=(localpart, domain, resource))
+        session = Session(entity)
+        session.connected = True
+        userSessions[resource] = session
+
+        return defer.succeed(entity)
+
+
+    def unbindResource(self, localpart, domain, resource, reason=None):
+        try:
+            session = self.sessions[localpart][resource]
+        except KeyError:
+            pass
+        else:
+            session.connected = False
+            del self.sessions[localpart][resource]
+            if not self.sessions[localpart]:
+                del self.sessions[localpart]
+
+        return defer.succeed(None)
+
+
+    def onElement(self, element, sender):
+        # Make sure each stanza has a sender address
+        if (element.name == 'presence' and
+            element.getAttribute('type') in ('subscribe', 'subscribed',
+                                             'unsubscribe', 'unsubscribed')):
+            element['from'] = sender.userhost()
+        else:
+            element['from'] = sender.full()
+
+        if (not element.hasAttribute('to') or
+                internJID(element['to']).host == self.domain):
+            # This stanza is for local delivery
+            self.xmlstream.dispatch(element)
+        else:
+            # This stanza is for remote delivery
+            self.xmlstream.send(element)
+
+
+    def deliverStanza(self, element, recipient):
+        if self.connectionManager:
+            self.connectionManager.deliverStanza(element, recipient)
+        else:
+            raise Exception("No connection manager set")

wokkel/component.py

@@ -313 +313 @@
         """
         destination = JID(stanza['to'])
 
-        log.msg("Routing to %s: %r" % (destination.full(), stanza.toXml()))
 
         if destination.host in self.routes:
+            msg = "Routing to %s (default route): %r"
+            log.msg("Routing to %s: %r" % (destination.full(),
+                                           stanza.toXml()))
             self.routes[destination.host].send(stanza)
+        elif None in self.routers:
+            log.msg("Routing to %s (default route): %r" % (destination.full(),
+                                                           stanza.toXml()))
+            self.routes[None].send(stanza)
         else:
-            self.routes[None].send(stanza)
+            log.msg("No route to %s: %r" % (destination.full(),
+                                            stanza.toXml()))
+            if stanza.getAttribute('type') not in ('result', 'error'):
+                # No route, send back error
+                exc = error.StanzaError('remote-server-timeout')
+                response = exc.toResponse(stanza)
+                self.route(response)
 
 
 

wokkel/xmppim.py

@@ -12 +12 @@
 All of it should eventually move to Twisted.
 """
 
+import copy
+
+from twisted.python import log
 from twisted.words.protocols.jabber import error, xmlstream
-from twisted.words.protocols.jabber.jid import JID
+from twisted.words.protocols.jabber.jid import JID, internJID
 from twisted.words.xish import domish
 
 from wokkel.compat import IQ
@@ -85 +88 @@
             handler(presence)
 
     def _onPresenceAvailable(self, presence):
-        entity = JID(presence["from"])
+        entity = internJID(presence["from"])
 
         show = unicode(presence.show or '')
         if show not in ['away', 'xa', 'chat', 'dnd']:
@@ -101 +104 @@
         self.availableReceived(entity, show, statuses, priority)
 
     def _onPresenceUnavailable(self, presence):
-        entity = JID(presence["from"])
+        entity = internJID(presence["from"])
 
         statuses = self._getStatuses(presence)
 
         self.unavailableReceived(entity, statuses)
 
     def _onPresenceSubscribed(self, presence):
-        self.subscribedReceived(JID(presence["from"]))
+        self.subscribedReceived(internJID(presence["from"]))
 
     def _onPresenceUnsubscribed(self, presence):
-        self.unsubscribedReceived(JID(presence["from"]))
+        self.unsubscribedReceived(internJID(presence["from"]))
 
     def _onPresenceSubscribe(self, presence):
-        self.subscribeReceived(JID(presence["from"]))
+        self.subscribeReceived(internJID(presence["from"]))
 
     def _onPresenceUnsubscribe(self, presence):
-        self.unsubscribeReceived(JID(presence["from"]))
+        self.unsubscribeReceived(internJID(presence["from"]))
 
 
     def availableReceived(self, entity, show=None, statuses=None, priority=0):
@@ -125 +128 @@
         Available presence was received.
 
         @param entity: entity from which the presence was received.
-        @type entity: {JID}
+        @type entity: L{JID}
         @param show: detailed presence information. One of C{'away'}, C{'xa'},
                      C{'chat'}, C{'dnd'} or C{None}.
         @type show: C{str} or C{NoneType}
@@ -143 +146 @@
         Unavailable presence was received.
 
         @param entity: entity from which the presence was received.
-        @type entity: {JID}
+        @type entity: L{JID}
         @param statuses: dictionary of natural language descriptions of the
                          availability status, keyed by the language
                          descriptor. A status without a language
@@ -156 +159 @@
         Subscription approval confirmation was received.
 
         @param entity: entity from which the confirmation was received.
-        @type entity: {JID}
+        @type entity: L{JID}
         """
 
     def unsubscribedReceived(self, entity):
@@ -164 +167 @@
         Unsubscription confirmation was received.
 
         @param entity: entity from which the confirmation was received.
-        @type entity: {JID}
+        @type entity: L{JID}
         """
 
     def subscribeReceived(self, entity):
@@ -172 +175 @@
         Subscription request was received.
 
         @param entity: entity from which the request was received.
-        @type entity: {JID}
+        @type entity: L{JID}
         """
 
     def unsubscribeReceived(self, entity):
@@ -180 +183 @@
         Unsubscription request was received.
 
         @param entity: entity from which the request was received.
-        @type entity: {JID}
+        @type entity: L{JID}
         """
 
     def available(self, entity=None, show=None, statuses=None, priority=0):
@@ -188 +191 @@
         Send available presence.
 
         @param entity: optional entity to which the presence should be sent.
-        @type entity: {JID}
+        @type entity: L{JID}
         @param show: optional detailed presence information. One of C{'away'},
                      C{'xa'}, C{'chat'}, C{'dnd'}.
         @type show: C{str}
@@ -207 +210 @@
         Send unavailable presence.
 
         @param entity: optional entity to which the presence should be sent.
-        @type entity: {JID}
+        @type entity: L{JID}
         @param statuses: dictionary of natural language descriptions of the
                          availability status, keyed by the language
                          descriptor. A status without a language
@@ -221 +224 @@
         Send subscription request
 
         @param entity: entity to subscribe to.
-        @type entity: {JID}
+        @type entity: L{JID}
         """
         self.send(Presence(to=entity, type='subscribe'))
 
@@ -230 +233 @@
         Send unsubscription request
 
         @param entity: entity to unsubscribe from.
-        @type entity: {JID}
+        @type entity: L{JID}
         """
         self.send(Presence(to=entity, type='unsubscribe'))
 
@@ -239 +242 @@
         Send subscription confirmation.
 
         @param entity: entity that subscribed.
-        @type entity: {JID}
+        @type entity: L{JID}
         """
         self.send(Presence(to=entity, type='subscribed'))
 
@@ -248 +251 @@
         Send unsubscription confirmation.
 
         @param entity: entity that unsubscribed.
-        @type entity: {JID}
+        @type entity: L{JID}
         """
         self.send(Presence(to=entity, type='unsubscribed'))
 
@@ -478 +481 @@
 
         @param recipient: Optional Recipient to which the presence should be
             sent.
-        @type recipient: {JID}
+        @type recipient: L{JID}
 
         @param show: Optional detailed presence information. One of C{'away'},
             C{'xa'}, C{'chat'}, C{'dnd'}.
@@ -503 +506 @@
         Send unavailable presence.
 
         @param recipient: Optional entity to which the presence should be sent.
-        @type recipient: {JID}
+        @type recipient: L{JID}
 
         @param statuses: dictionary of natural language descriptions of the
             availability status, keyed by the language descriptor. A status
@@ -520 +523 @@
         Send subscription request
 
         @param recipient: Entity to subscribe to.
-        @type recipient: {JID}
+        @type recipient: L{JID}
         """
         presence = SubscriptionPresence(recipient=recipient, sender=sender)
         presence.stanzaType = 'subscribe'
@@ -532 +535 @@
         Send unsubscription request
 
         @param recipient: Entity to unsubscribe from.
-        @type recipient: {JID}
+        @type recipient: L{JID}
         """
         presence = SubscriptionPresence(recipient=recipient, sender=sender)
         presence.stanzaType = 'unsubscribe'
@@ -544 +547 @@
         Send subscription confirmation.
 
         @param recipient: Entity that subscribed.
-        @type recipient: {JID}
+        @type recipient: L{JID}
         """
         presence = SubscriptionPresence(recipient=recipient, sender=sender)
         presence.stanzaType = 'subscribed'
@@ -556 +559 @@
         Send unsubscription confirmation.
 
         @param recipient: Entity that unsubscribed.
-        @type recipient: {JID}
+        @type recipient: L{JID}
         """
         presence = SubscriptionPresence(recipient=recipient, sender=sender)
         presence.stanzaType = 'unsubscribed'
@@ -568 +571 @@
         Send presence probe.
 
         @param recipient: Entity to be probed.
-        @type recipient: {JID}
+        @type recipient: L{JID}
         """
         presence = ProbePresence(recipient=recipient, sender=sender)
         self.send(presence.toElement())
@@ -652 +655 @@
 
 
     def _parseRosterItem(self, element):
-        jid = JID(element['jid'])
+        jid = internJID(element['jid'])
         item = RosterItem(jid)
         item.name = element.getAttribute('name')
         subscription = element.getAttribute('subscription')
@@ -715 +718 @@
         itemElement = iq.query.item
 
         if unicode(itemElement['subscription']) == 'remove':
-            self.onRosterRemove(JID(itemElement['jid']))
+            self.onRosterRemove(internJID(itemElement['jid']))
         else:
             item = self._parseRosterItem(iq.query.item)
             self.onRosterSet(item)
@@ -763 +766 @@
     def _onRosterGet(self, iq):
         iq.handled = True
 
-        d = self.getRoster(JID(iq["from"]))
+        d = self.getRoster(internJID(iq["from"]))
         d.addCallback(self._toRosterReply, iq)
         d.addErrback(lambda _: error.ErrorStanza('internal-error').toResponse(iq))
         d.addBoth(self.send)
@@ -808 +811 @@
         """
         Called when a message stanza was received.
         """
+
+
+
+class AccountIQHandler(XMPPHandler):
+
+    def __init__(self, sessionManager):
+        XMPPHandler.__init__(self)
+        self.sessionManager = sessionManager
+
+
+    def connectionMade(self):
+        self.xmlstream.addObserver('/iq', self.onIQ, 1)
+
+
+    def onIQ(self, iq):
+        """
+        Handler for iq stanzas to user accounts' connected resources.
+
+        If the recipient is a bare JID or there is no associated user, this
+        handler ignores the stanza, so that other handlers have a chance
+        to pick it up. If used, L{generic.FallbackHandler} will respond with a
+        C{'service-unavailable'} stanza error if no other handlers handle
+        the iq.
+        """
+
+        if iq.handled:
+            return
+
+        try:
+            recipient = internJID(iq['to'])
+        except KeyError:
+            return
+
+        if not recipient.user:
+            # This is not for an account, ignore it
+            return
+        elif recipient.user not in self.sessionManager.accounts:
+            # This is not a user, ignore it
+            return
+        elif not recipient.resource:
+            # Bare JID at local domain, ignore it
+            return
+        elif recipient.user in self.sessionManager.sessions:
+            # Full JID with connected resource, deliver the stanza
+            self.sessionManager.deliverStanza(iq, recipient)
+        else:
+            # Full JID without connected resource, return error
+            exc = error.StanzaError('service-unavailable')
+            if iq['type'] in ('result', 'error'):
+                log.err(exc, 'Could not deliver IQ response')
+            else:
+                self.send(exc.toResponse(iq))
+
+        iq.handled = True
+
+
+
+class AccountMessageHandler(XMPPHandler):
+
+    def __init__(self, sessionManager):
+        XMPPHandler.__init__(self)
+        self.sessionManager = sessionManager
+
+
+    def connectionMade(self):
+        self.xmlstream.addObserver('/message', self.onMessage, 1)
+
+
+    def onMessage(self, message):
+        """
+        Handler for message stanzas to user accounts.
+        """
+
+        if message.handled:
+            return
+
+        try:
+            recipient = internJID(message['to'])
+        except KeyError:
+            return
+
+        stanzaType = message.getAttribute('type', 'normal')
+
+        try:
+            if not recipient.user:
+                # This is not for an account, ignore it
+                return
+            elif recipient.user not in self.sessionManager.accounts:
+                # This is not a user, ignore it
+                return
+            elif recipient.resource:
+                userSessions = self.sessionManager.sessions.get(recipient.user,
+                                                                {})
+                if recipient.resource in userSessions:
+                    self.sessionManager.deliverStanza(message, recipient)
+                else:
+                    if stanzaType in ('normal', 'chat', 'headline'):
+                        self.onMessageBareJID(message, recipient.userhostJID())
+                    elif stanzaType == 'error':
+                        log.msg("Dropping message to unconnected resource %r" %
+                                recipient.full())
+                    elif stanzaType == 'groupchat':
+                        raise error.StanzaError('service-unavailable')
+            else:
+                self.onMessageBareJID(message, recipient)
+        except error.StanzaError, exc:
+            if stanzaType == 'error':
+                log.err(exc, "Undeliverable error")
+            else:
+                self.send(exc.toResponse(message))
+
+        message.handled = True
+
+
+    def onMessageBareJID(self, message, bareJID):
+        stanzaType = message.getAttribute('type', 'normal')
+
+        userSessions = self.sessionManager.sessions.get(bareJID.user, {})
+        print userSessions
+
+        recipients = set()
+
+        if stanzaType == 'headline':
+            for session in userSessions:
+                if session.presence.priority >= 0:
+                    recipients.add(session.entity)
+        elif stanzaType in ('chat', 'normal'):
+            priorities = {}
+            for session in userSessions.itervalues():
+                if not session.presence or not session.presence.available:
+                    continue
+                priority = session.presence.priority
+                if priority >= 0:
+                    priorities.setdefault(priority, set()).add(session.entity)
+                maxPriority = max(priorities.keys())
+                recipients.update(priorities[maxPriority])
+        elif stanzaType == 'groupchat':
+            raise error.StanzaError('service-unavailable')
+
+        if recipients:
+            for recipient in recipients:
+                self.sessionManager.deliverStanza(message, recipient)
+        elif stanzaType in ('chat', 'normal'):
+            raise error.StanzaError('service-unavailable')
+        else:
+            # silently discard
+            log.msg("Discarding message to %r" % message['to'])
+
+
+
+
+def clonePresence(presence):
+    """
+    Make a deep copy of a presence stanza.
+
+    The returned presence stanza is an orphaned deep copy of the given
+    original.
+
+    @note: Since the reference to the original parent, if any, is gone,
+    inherited attributes like C{xml:lang} are not preserved.
+    """
+    element = presence.element
+
+    parent = element.parent
+    element.parent = None
+    newElement = copy.deepcopy(element)
+    element.parent = parent
+    return newElement
+
+
+
+class PresenceServerHandler(PresenceProtocol):
+
+    def __init__(self, sessionManager, domain, roster):
+        PresenceProtocol.__init__(self)
+        self.sessionManager = sessionManager
+        self.domain = domain
+        self.roster = roster
+        self.presences = {} # user -> resource -> presence
+        self.offlinePresences = {} # user -> presence
+        self.remotePresences = {} # user -> remote entity -> presence
+
+
+    def _broadcastToOtherResources(self, presence):
+        """
+        Broadcast presence to other available resources.
+        """
+        fromJID = presence.sender
+        for otherResource in self.presences[fromJID.user]:
+            if otherResource == fromJID.resource:
+                continue
+
+            resourceJID = JID(tuple=(fromJID.user, fromJID.host, otherResource))
+            outPresence = clonePresence(presence)
+            outPresence['to'] = resourceJID.full()
+            self.sessionManager.deliverStanza(outPresence, resourceJID)
+
+
+    def _broadcastToContacts(self, presence):
+        """
+        Broadcast presence to subscribed entities.
+        """
+        fromJID = presence.sender
+        roster = self.roster[fromJID.user]
+
+        for item in roster.itervalues():
+            if not item.subscriptionFrom:
+                continue
+
+            outPresence = clonePresence(presence)
+            outPresence['to'] = item.jid.full()
+
+            if item.jid.host == self.domain:
+                # local contact
+                if item.jid.user in self.presences:
+                    # broadcast to contact's available resources
+                    for itemResource in self.presences[item.jid.user]:
+                        resourceJID = JID(tuple=(item.jid.user,
+                                                 item.jid.host,
+                                                 itemResource))
+                        self.sessionManager.deliverStanza(outPresence,
+                                                          resourceJID)
+            else:
+                # remote contact
+                self.send(outPresence)
+
+
+    def _on_availableBroadcast(self, presence):
+        fromJID = presence.sender
+        user, resource = fromJID.user, fromJID.resource
+        roster = self.roster[user]
+
+        if user not in self.presences:
+            # initial presence
+            self.presences[user] = {}
+            self.remotePresences[user] = {}
+
+            # send out probes
+            for item in roster.itervalues():
+                if item.subscriptionTo and item.jid.host != self.domain:
+                    self.probe(item.jid, fromJID)
+        else:
+            if resource not in self.presences[user]:
+                # initial presence with another available resource
+
+                # send last known presences from remote contacts
+                remotePresences = self.remotePresences[user]
+                for entity, remotePresence in remotePresences.iteritems():
+                    self.sessionManager.deliverStanza(remotePresence.element,
+                                                      fromJID)
+
+            # send presence to other resources
+            self._broadcastToOtherResources(presence)
+
+        # Send last known local presences
+        if user not in self.presences or resource not in self.presences[user]:
+            for item in roster.itervalues():
+                if item.subscriptionTo and \
+                   item.jid.host == self.domain and \
+                   item.jid.user in self.presences:
+                    for contactPresence in \
+                            self.presences[item.jid.user].itervalues():
+                        outPresence = clonePresence(contactPresence)
+                        outPresence['to'] = fromJID.userhost()
+                        self.sessionManager.deliverStanza(outPresence, fromJID)
+
+        # broadcast presence
+        self._broadcastToContacts(presence)
+
+        # save presence
+        self.presences[user][resource] = presence
+        self.sessionManager.sessions[user][resource].presence = presence
+
+
+    #def _on_availableDirected(self, presence):
+    #    pass
+
+
+    def _on_availableInbound(self, presence):
+        fromJID = presence.sender
+        toJID = presence.recipient
+        if (toJID.user in self.roster and
+            toJID.user in self.presences):
+            for resource in self.presences[toJID.user]:
+                resourceJID = JID(tuple=(toJID.user, toJID.host, resource))
+                self.sessionManager.deliverStanza(presence.element, resourceJID)
+            self.remotePresences[toJID.user][fromJID] = presence
+        else:
+            # no such user or no available resource, ignore this stanza
+            pass
+
+
+    def _on_unavailableBroadcast(self, presence):
+        fromJID = presence.sender
+        user, resource = fromJID.user, fromJID.resource
+
+        # broadcast presence
+        self._broadcastToContacts(presence)
+
+        if user in self.presences:
+            # send presence to other resources
+            self._broadcastToOtherResources(presence)
+
+            # update stored presences
+            if resource in self.presences[user]:
+                del self.presences[user][resource]
+
+            if not self.presences[user]:
+                # last resource to become unavailable
+                del self.presences[user]
+
+                # TODO: save last unavailable presence
+
+
+#    def _on_unavailableDirected(self, presence):
+#        pass
+
+
+#    def _on_unavailableInbound(self, presence):
+#        pass
+
+
+    def getDirection(self, presence):
+        if not presence.recipient:
+            if presence.sender.host == self.domain:
+                # broadcast presence from local domain
+                return 'Broadcast'
+            else:
+                raise Exception("Unexpected missing to address")
+        else:
+            if presence.sender.host == self.domain:
+                # directed presence from local domain
+                return 'Directed'
+            elif presence.recipient.host == self.domain:
+                # incoming remote presence
+                return 'Inbound'
+            else:
+                raise Exception("Badly routed presence")
+
+    def availableReceived(self, presence):
+        direction = self.getDirection(presence)
+        handler = getattr(self, "_on_available%s" % direction)
+        if handler:
+            handler(presence)
+            presence.handled = True
+        else:
+            print "Unhandled: %r" % presence.element.toXml()
+
+
+    def unavailableReceived(self, presence):
+        direction = self.getDirection(presence)
+        handler = getattr(self, "_on_unavailable%s" % direction)
+        if handler:
+            handler(presence)
+            presence.handled = True
+        else:
+            print "Unhandled: %r" % presence.element.toXml()
+
+
+
+    def probeReceived(self, presence):
+        fromJID = presence.sender
+        toJID = presence.recipient
+
+        if toJID.user not in self.roster or \
+           fromJID.userhost() not in self.roster[toJID.user] or \
+           not self.roster[toJID.user][fromJID.userhost()].subscriptionFrom:
+            # send unsubscribed
+            pass
+        elif toJID.user not in self.presences:
+            # send last unavailable or nothing
+            pass
+        else:
+            for resourcePresence in self.presences[toJID.user].itervalues():
+                outPresence = clonePresence(resourcePresence)
+                outPresence['to'] = fromJID.userhost()
+                self.send(outPresence)