source: ralphm-patches/xmpp_client_service.patch @ 52:a6ed3b9703cb

new file doc/examples/client_service.tac

@@ +1 @@
+from twisted.application import service, strports
+from twisted.internet import defer
+
+from wokkel import client, im
+from wokkel.component import InternalComponent, Router
+from wokkel.generic import FallbackHandler
+from wokkel.ping import PingHandler
+from wokkel.im 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(im.RosterServerProtocol):
+
+    def __init__(self, roster):
+        im.RosterServerProtocol.__init__(self)
+        self.roster = roster
+
+    def getRoster(self, request):
+        user = request.sender.user
+        return defer.succeed(self.roster[user].values())
+
+
+
+application = service.Application("Jabber server")
+
+router = Router()
+component = InternalComponent(router, domain)
+component.setServiceParent(application)
+
+sessionManager = client.SessionManager(domain, accounts)
+sessionManager.setHandlerParent(component)
+
+im.AccountIQHandler(sessionManager).setHandlerParent(component)
+im.AccountMessageHandler(sessionManager).setHandlerParent(component)
+im.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, utility
 
 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 = {}
+        self.clientStream = utility.EventDispatcher()
+        self.clientStream.addObserver('/*', self.routeOrDeliver, -1)
+
+
+    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()
+
+        self.clientStream.dispatch(element)
+
+
+    def routeOrDeliver(self, element):
+        if element.handled:
+            return
+
+        if (not element.hasAttribute('to') or
+            internJID(element['to']).host == self.domain):
+            # This stanza is for local delivery
+            log.msg("Delivering locally: %r" % element.toXml())
+            self.xmlstream.dispatch(element)
+        else:
+            # This stanza is for remote routing
+            log.msg("Routing remotely: %r" % element.toXml())
+            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.routes:
+            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', type='wait')
+                exc.code = '504'
+                response = exc.toResponse(stanza)
+                self.route(response)
 
 
 

wokkel/im.py

@@ -10 +10 @@
 U{RFC 6121<http://www.xmpp.org/rfcs/rfc6121.html>} (XMPP IM).
 """
 
+import copy
+
 from twisted.internet import defer
+from twisted.python import log
 from twisted.words.protocols.jabber import jid
 from twisted.words.protocols.jabber import error
 from twisted.words.xish import domish
@@ -168 +171 @@
         self.xmlstream.addObserver("/presence", self._onPresence)
 
 
-    def _onPresence(self, element):
+    def parsePresence(self, element):
+        """
+        Parse presence.
+        """
         stanza = Stanza.fromElement(element)
 
         presenceType = stanza.stanzaType or 'available'
@@ -178 +184 @@
         except KeyError:
             return
 
-        presence = parser.fromElement(element)
+        return parser.fromElement(element)
+
+
+    def _onPresence(self, element):
+        presence = self.parsePresence(element)
+        presenceType = presence.stanzaType or 'available'
 
         try:
             handler = getattr(self, '%sReceived' % presenceType)
         except AttributeError:
             return
         else:
-            handler(presence)
+            element.handled = handler(presence)
 
 
     def errorReceived(self, presence):
@@ -553 +564 @@
 
 
 
+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 = jid.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
+
+        userSessions = self.sessionManager.sessions.get(recipient.user,
+                                                        {})
+        if recipient.resource in userSessions:
+            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 = jid.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, {})
+
+        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
+
+        self.sessionManager.clientStream.addObserver('/presence',
+                                                     self._onPresenceOutbound)
+
+
+    def _onPresenceOutbound(self, element):
+        log.msg("Got outbound presence: %r" % element.toXml())
+        presence = self.parsePresence(element)
+
+        presenceType = presence.stanzaType or 'available'
+        method = '%sReceivedOutbound' % presenceType
+        print method
+
+        try:
+            handler = getattr(self, method)
+        except AttributeError:
+            return
+        else:
+            element.handled = handler(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.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.entity.full()
+
+            if item.entity.host == self.domain:
+                # local contact
+                if item.entity.user in self.presences:
+                    # broadcast to contact's available resources
+                    for itemResource in self.presences[item.entity.user]:
+                        resourceJID = jid.JID(tuple=(item.entity.user,
+                                                     item.entity.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.entity.host != self.domain:
+                    self.probe(item.entity, 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.entity.host == self.domain and \
+                   item.entity.user in self.presences:
+                    for contactPresence in \
+                            self.presences[item.entity.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
+
+        return True
+
+
+    def _on_availableDirected(self, presence):
+        self.send(presence.element)
+        return True
+
+
+    def availableReceivedOutbound(self, presence):
+        if presence.recipient:
+            return self._on_availableDirected(presence)
+        else:
+            return self._on_availableBroadcast(presence)
+
+
+    def availableReceived(self, presence):
+        fromJID = presence.sender
+        toJID = presence.recipient
+
+        if toJID.user not in self.roster:
+            return False
+
+        if toJID.user in self.presences:
+            for resource in self.presences[toJID.user]:
+                resourceJID = jid.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
+
+        return True
+
+
+    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
+
+        return True
+
+
+    def _on_unavailableDirected(self, presence):
+        self.send(presence.element)
+        return True
+
+
+    def unavailableReceivedOutbound(self, presence):
+        if presence.recipient:
+            return self._on_unavailableDirected(presence)
+        else:
+            return self._on_unavailableBroadcast(presence)
+
+#    def unavailableReceived(self, presence):
+
+
+    def subscribedReceivedOutbound(self, presence):
+        log.msg("%r subscribed %s to its presence" % (presence.sender,
+                                                      presence.recipient))
+        self.send(presence.element)
+        return True
+
+
+    def subscribedReceived(self, presence):
+        log.msg("%r subscribed %s to its presence" % (presence.sender,
+                                                      presence.recipient))
+
+
+    def unsubscribedReceivedOutbound(self, presence):
+        log.msg("%r unsubscribed %s from its presence" % (presence.sender,
+                                                          presence.recipient))
+        self.send(presence.element)
+        return True
+
+
+    def unsubscribedReceived(self, presence):
+        log.msg("%r unsubscribed %s from its presence" % (presence.sender,
+                                                          presence.recipient))
+
+
+    def subscribeReceivedOutbound(self, presence):
+        log.msg("%r requests subscription to %s" % (presence.sender,
+                                                    presence.recipient))
+        self.send(presence.element)
+        return True
+
+
+    def subscribeReceived(self, presence):
+        log.msg("%r requests subscription to %s" % (presence.sender,
+                                                    presence.recipient))
+
+
+    def unsubscribeReceivedOutbound(self, presence):
+        log.msg("%r requests unsubscription from %s" % (presence.sender,
+                                                        presence.recipient))
+        self.send(presence.element)
+        return True
+
+
+    def unsubscribeReceived(self, presence):
+        log.msg("%r requests unsubscription from %s" % (presence.sender,
+                                                        presence.recipient))
+
+
+    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)
+
+
+
 class RosterServerProtocol(XMPPHandler, IQHandlerMixin):
     """
     XMPP subprotocol handler for the roster, server side.

wokkel/test/test_im.py

@@ -13 +13 @@
 from twisted.words.xish import domish, utility
 
 from wokkel import im
-from wokkel.generic import ErrorStanza, parseXml
+from wokkel.generic import ErrorStanza, Stanza, parseXml
 from wokkel.test.helpers import TestableRequestHandlerMixin, XmlStreamStub
 
 NS_XML = 'http://www.w3.org/XML/1998/namespace'
@@ -846 +846 @@
 
 
 
+class AccountIQHandlerTest(unittest.TestCase):
+    """
+    Tests for L{im.AccountIQHandler}.
+    """
+
+    def setUp(self):
+        self.stub = XmlStreamStub()
+        self.protocol = im.AccountIQHandler(None)
+        self.protocol.makeConnection(self.stub.xmlstream)
+        self.protocol.connectionInitialized()
+
+
+    def test_onIQNotUser(self):
+        """
+        IQs to JIDs without local part are ignored.
+        """
+        xml = """
+          <iq to='example.org'>
+            <query xmlns='jabber:iq:version'/>
+          </iq>
+        """
+
+        iq = parseXml(xml)
+        self.stub.send(iq)
+
+        self.assertFalse(getattr(iq, 'handled'))
+
+
+
+class AccountMessageHandlerTest(unittest.TestCase):
+    """
+    Tests for L{im.AccountMessageHandler}.
+    """
+
+    def setUp(self):
+        self.stub = XmlStreamStub()
+        self.protocol = im.AccountMessageHandler(None)
+        self.protocol.makeConnection(self.stub.xmlstream)
+        self.protocol.connectionInitialized()
+
+
+    def test_onMessageNotUser(self):
+        """
+        Messages to JIDs without local part are ignored.
+        """
+        xml = """
+          <message to='example.org'>
+            <body>Hello</body>
+          </message>
+        """
+
+        message = parseXml(xml)
+        self.stub.send(message)
+
+        self.assertFalse(getattr(message, 'handled'))
+
+
+
+class ClonePresenceTest(unittest.TestCase):
+    """
+    Tests for L{im.clonePresence}.
+    """
+
+    def test_rootElement(self):
+        """
+        The copied presence stanza is not identical, but renders identically.
+        """
+        originalElement = domish.Element((None, 'presence'))
+        stanza = Stanza.fromElement(originalElement)
+        copyElement = im.clonePresence(stanza)
+
+        self.assertNotIdentical(copyElement, originalElement)
+        self.assertEquals(copyElement.toXml(), originalElement.toXml())
+
+
+
 class RosterServerProtocolTest(unittest.TestCase, TestableRequestHandlerMixin):
     """
     Tests for L{im.RosterServerProtocol}.