Changeset 35:eb020b49a77d

Timestamp:

Oct 10, 2008, 5:24:28 PM (14 years ago)

Author:

Ralph Meijer <ralphm@…>

Branch:

default

Convert:

svn:b33ecbfc-034c-dc11-8662-000475d9059e/trunk@90

Message:

Add XMPP router and server side component authenticator.

Author: ralphm.
Fixed #30.

Location:

wokkel

Files:

• component.py (modified) (3 diffs)
• generic.py (modified) (2 diffs)
• test/test_component.py (added)
• test/test_generic.py (modified) (1 diff)

wokkel/component.py

@@ -1 @@
-# Copyright (c) 2003-2007 Ralph Meijer
+# -*- test-case-name: wokkel.test.test_component -*-
+#
+# Copyright (c) 2003-2008 Ralph Meijer
 # See LICENSE for details.
 
@@ -8 +10 @@
 from twisted.application import service
 from twisted.internet import reactor
-from twisted.words.protocols.jabber import component
+from twisted.python import log
+from twisted.words.protocols.jabber.jid import internJID as JID
+from twisted.words.protocols.jabber import component, error, xmlstream
 from twisted.words.xish import domish
 
 from wokkel.subprotocols import StreamManager
+
+NS_COMPONENT_ACCEPT = 'jabber:component:accept'
 
 class Component(StreamManager, service.Service):
@@ -57 +63 @@
     def _getConnection(self):
         return reactor.connectTCP(self.host, self.port, self.factory)
+
+
+
+class ListenComponentAuthenticator(xmlstream.ListenAuthenticator):
+    """
+    Authenticator for accepting components.
+    """
+    namespace = NS_COMPONENT_ACCEPT
+
+    def __init__(self, secret):
+        self.secret = secret
+        xmlstream.ListenAuthenticator.__init__(self)
+
+
+    def associateWithStream(self, xs):
+        xs.version = (0, 0)
+        xmlstream.ListenAuthenticator.associateWithStream(self, xs)
+
+
+    def streamStarted(self, rootElement):
+        xmlstream.ListenAuthenticator.streamStarted(self, rootElement)
+
+        if rootElement.defaultUri != self.namespace:
+            exc = error.StreamError('invalid-namespace')
+            self.xmlstream.sendStreamError(exc)
+            return
+
+        # self.xmlstream.thisEntity is set to the address the component
+        # wants to assume. This should probably be checked.
+        if not self.xmlstream.thisEntity:
+            exc = error.StreamError('improper-addressing')
+            self.xmlstream.sendStreamError(exc)
+            return
+
+        self.xmlstream.sid = 'random' # FIXME
+
+        self.xmlstream.sendHeader()
+        self.xmlstream.addOnetimeObserver('/*', self.onElement)
+
+
+    def onElement(self, element):
+        if (element.uri, element.name) == (self.namespace, 'handshake'):
+            self.onHandshake(unicode(element))
+        else:
+            exc = error.streamError('not-authorized')
+            self.xmlstream.sendStreamError(exc)
+
+
+    def onHandshake(self, handshake):
+        calculatedHash = xmlstream.hashPassword(self.xmlstream.sid, self.secret)
+        if handshake != calculatedHash:
+            exc = error.StreamError('not-authorized', text='Invalid hash')
+            self.xmlstream.sendStreamError(exc)
+        else:
+            self.xmlstream.send('<handshake/>')
+            self.xmlstream.dispatch(self.xmlstream,
+                                    xmlstream.STREAM_AUTHD_EVENT)
+
+
+
+class RouterService(service.Service):
+    """
+    XMPP Server's Router Service.
+
+    This service connects the different components of the XMPP service and
+    routes messages between them based on the given routing table.
+
+    Connected components are trusted to have correct addressing in the
+    stanzas they offer for routing.
+
+    A route destination of C{None} adds a default route. Traffic for which no
+    specific route exists, will be routed to this default route.
+
+    @ivar routes: Routes based on the host part of JIDs. Maps host names to the
+                  L{EventDispatcher<utility.EventDispatcher>}s that should
+                  receive the traffic. A key of C{None} means the default
+                  route.
+    @type routes: C{dict}
+    """
+
+    def __init__(self):
+        self.routes = {}
+
+
+    def addRoute(self, destination, xs):
+        """
+        Add a new route.
+
+        The passed XML Stream C{xs} will have an observer for all stanzas
+        added to route its outgoing traffic. In turn, traffic for
+        C{destination} will be passed to this stream.
+
+        @param destination: Destination of the route to be added as a host name
+                            or C{None} for the default route.
+        @type destination: C{str} or C{NoneType}.
+        @param xs: XML Stream to register the route for.
+        @type xs: L{EventDispatcher<utility.EventDispatcher>}.
+        """
+        self.routes[destination] = xs
+        xs.addObserver('/*', self.route)
+
+
+    def removeRoute(self, destination, xs):
+        """
+        Remove a route.
+
+        @param destination: Destination of the route that should be removed.
+        @type destination: C{str}.
+        @param xs: XML Stream to remove the route for.
+        @type xs: L{EventDispatcher<utility.EventDispatcher>}.
+        """
+        xs.removeObserver('/*', self.route)
+        if (xs == self.routes[destination]):
+            del self.routes[destination]
+
+
+    def route(self, stanza):
+        """
+        Route a stanza.
+
+        @param stanza: The stanza to be routed.
+        @type stanza: L{domish.Element}.
+        """
+        if not list(stanza.elements()):
+            return
+
+        destination = JID(stanza['to'])
+
+        log.msg("Routing to %s: %r" % (destination.full(), stanza.toXml()))
+
+        if destination.host in self.routes:
+            self.routes[destination.host].send(stanza)
+        else:
+            self.routes[None].send(stanza)
+
+
+
+class ComponentServer(service.Service):
+    """
+    XMPP Component Server service.
+
+    This service accepts XMPP external component connections and makes
+    the router service route traffic for a component's bound domain
+    to that component.
+    """
+
+    logTraffic = False
+
+    def __init__(self, router, port=5347, secret='secret'):
+        self.router = router
+        self.port = port
+        self.secret = secret
+
+        def authenticatorFactory():
+            return ListenComponentAuthenticator(self.secret)
+
+        self.factory = xmlstream.XmlStreamServerFactory(authenticatorFactory)
+        self.factory.addBootstrap(xmlstream.STREAM_CONNECTED_EVENT,
+                                  self.makeConnection)
+        self.factory.addBootstrap(xmlstream.STREAM_AUTHD_EVENT,
+                                  self.connectionInitialized)
+
+        self.serial = 0
+
+
+    def startService(self):
+        service.Service.startService(self)
+        reactor.listenTCP(self.port, self.factory)
+
+
+    def makeConnection(self, xs):
+        """
+        Called when a component connection was made.
+
+        This enables traffic debugging on incoming streams.
+        """
+        xs.serial = self.serial
+        self.serial += 1
+
+        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 connectionInitialized(self, xs):
+        """
+        Called when a component has succesfully authenticated.
+
+        Add the component to the routing table and establish a handler
+        for a closed connection.
+        """
+        destination = xs.thisEntity.host
+
+        self.router.addRoute(destination, xs)
+        xs.addObserver(xmlstream.STREAM_END_EVENT, self.connectionLost, 0,
+                                                   destination, xs)
+
+
+    def onError(self, reason):
+        log.err(reason, "Stream Error")
+
+
+    def connectionLost(self, destination, xs, reason):
+        self.router.removeRoute(destination, xs)

wokkel/generic.py

@@ -13 +13 @@
 from twisted.words.protocols.jabber import error
 from twisted.words.protocols.jabber.xmlstream import toResponse
-from twisted.words.xish import domish
+from twisted.words.xish import domish, utility
 
 from wokkel import disco
@@ -121 +121 @@
     def getDiscoItems(self, requestor, target, node):
         return defer.succeed([])
+
+
+
+class XmlPipe(object):
+    """
+    XML stream pipe.
+
+    Connects two objects that communicate stanzas through an XML stream like
+    interface. Each of the ends of the pipe (sink and source) can be used to
+    send XML stanzas to the other side, or add observers to process XML stanzas
+    that were sent from the other side.
+
+    XML pipes are usually used in place of regular XML streams that are
+    transported over TCP. This is the reason for the use of the names source
+    and sink for both ends of the pipe. The source side corresponds with the
+    entity that initiated the TCP connection, whereas the sink corresponds with
+    the entity that accepts that connection. In this object, though, the source
+    and sink are treated equally.
+
+    Unlike Jabber
+    L{XmlStream<twisted.words.protocols.jabber.xmlstream.XmlStream>}s, the sink
+    and source objects are assumed to represent an eternal connected and
+    initialized XML stream. As such, events corresponding to connection,
+    disconnection, initialization and stream errors are not dispatched or
+    processed.
+
+    @ivar source: Source XML stream.
+    @ivar sink: Sink XML stream.
+    """
+
+    def __init__(self):
+        self.source = utility.EventDispatcher()
+        self.sink = utility.EventDispatcher()
+        self.source.send = lambda obj: self.sink.dispatch(obj)
+        self.sink.send = lambda obj: self.source.dispatch(obj)

wokkel/test/test_generic.py

@@ -49 +49 @@
         self.assertEquals(1, len(elements))
         self.assertEquals('0.1.0', unicode(elements[0]))
+
+
+
+class XmlPipeTest(unittest.TestCase):
+    """
+    Tests for L{wokkel.generic.XmlPipe}.
+    """
+
+    def setUp(self):
+        self.pipe = generic.XmlPipe()
+
+
+    def test_sendFromSource(self):
+        """
+        Send an element from the source and observe it from the sink.
+        """
+        def cb(obj):
+            called.append(obj)
+
+        called = []
+        self.pipe.sink.addObserver('/test[@xmlns="testns"]', cb)
+        element = domish.Element(('testns', 'test'))
+        self.pipe.source.send(element)
+        self.assertEquals([element], called)
+
+
+    def test_sendFromSink(self):
+        """
+        Send an element from the sink and observe it from the source.
+        """
+        def cb(obj):
+            called.append(obj)
+
+        called = []
+        self.pipe.source.addObserver('/test[@xmlns="testns"]', cb)
+        element = domish.Element(('testns', 'test'))
+        self.pipe.sink.send(element)
+        self.assertEquals([element], called)