source: ralphm-patches/listening-authenticator-stream-features.patch @ 72:727b4d29c48e

wokkel/generic.py

@@ -10 +10 @@
 from zope.interface import implements
 
 from twisted.internet import defer, protocol
-from twisted.python import reflect
+from twisted.python import log, reflect
 from twisted.python.deprecate import deprecated
 from twisted.python.versions import Version
 from twisted.words.protocols.jabber import error, jid, xmlstream
@@ -18 +18 @@
 from twisted.words.xish import domish, utility
 from twisted.words.xish.xmlstream import BootstrapMixin
 
-from wokkel.iwokkel import IDisco
+from wokkel.iwokkel import IDisco, IReceivingInitializer
 from wokkel.subprotocols import XMPPHandler
 
 IQ_GET = '/iq[@type="get"]'
@@ -27 +27 @@
 NS_VERSION = 'jabber:iq:version'
 VERSION = IQ_GET + '/query[@xmlns="' + NS_VERSION + '"]'
 
+XPATH_ALL = "/*"
+
 def parseXml(string):
     """
     Parse serialized XML into a DOM structure.
@@ -332 +334 @@
 
 
 
+class TestableXmlStream(xmlstream.XmlStream):
+    """
+    XML Stream that buffers outgoing data and catches special events.
+
+    This implementation overrides relevant methods to prevent any data
+    to be sent out on a transport. Instead it buffers all outgoing stanzas,
+    sets flags instead of sending the stream header and footer and logs stream
+    errors so it can be caught by a logging observer.
+
+    @ivar output: Sequence of objects sent out using L{send}. Usually these are
+        L{domish.Element} instances.
+    @type output: C{list}
+
+    @ivar headerSent: Flag set when a stream header would have been sent. When
+        a stream restart occurs through L{reset}, this flag is reset as well.
+    @type headerSent: C{bool}
+
+    @ivar footerSent: Flag set when a stream footer would have been sent
+        explicitly. Note that it is not set when a stream error is sent
+        using L{sendStreamError}.
+    @type footerSent: C{bool}
+    """
+
+
+    def __init__(self, authenticator):
+        xmlstream.XmlStream.__init__(self, authenticator)
+        self.headerSent = False
+        self.footerSent = False
+        self.output = []
+
+
+    def reset(self):
+        xmlstream.XmlStream.reset(self)
+        self.headerSent = False
+
+
+    def sendHeader(self):
+        self.headerSent = True
+
+
+    def sendFooter(self):
+        self.footerSent = True
+
+
+    def sendStreamError(self, streamError):
+        """
+        Log a stream error.
+
+        If this is called from a Twisted Trial test case, the stream error
+        will be observed by the Trial logging observer. If it is not explicitly
+        tested for (i.e. flushed), this will cause the test case to
+        automatically fail. See L{assertStreamError} for a convenience method
+        to test for stream errors.
+
+        @type streamError: L{error.StreamError}
+        """
+        log.err(streamError)
+
+
+    @staticmethod
+    def assertStreamError(testcase, condition=None, exc=None):
+        """
+        Check if a stream error was sent out.
+
+        To check for stream errors sent out by L{sendStreamError}, this method
+        will flush logged stream errors and inspect the last one. If
+        C{condition} was passed, the logged error is asserted to match that
+        condition. If C{exc} was passed, the logged error is asserted to be
+        identical to it.
+
+        Note that this is takes the calling test case as the first argument, to
+        be able to hook into its methods for flushing errors and making
+        assertions.
+
+        @param testcase: The test case instance that is calling this method.
+        @type testcase: {twisted.trial.unittest.TestCase}
+
+        @param condition: The optional stream error condition to match against.
+        @type condition: C{unicode}.
+
+        @param exc: The optional stream error to check identity against.
+        @type exc: L{error.StreamError}
+        """
+
+        loggedErrors = testcase.flushLoggedErrors(error.StreamError)
+        testcase.assertTrue(loggedErrors, "No stream error was sent")
+        streamError = loggedErrors[-1].value
+        if condition:
+            testcase.assertEqual(condition, streamError.condition)
+        elif exc:
+            testcase.assertIdentical(exc, streamError)
+
+
+    def send(self, obj):
+        """
+        Buffer all outgoing stanzas.
+
+        @type obj: L{domish.Element}
+        """
+        self.output.append(obj)
+
+
+
+class BaseReceivingInitializer(object):
+    """
+    Base stream initializer for receiving entities.
+    """
+    implements(IReceivingInitializer)
+
+    required = False
+
+    def __init__(self, name, xs):
+        self.name = name
+        self.xmlstream = xs
+        self.deferred = defer.Deferred()
+
+
+    def getFeatures(self):
+        raise NotImplementedError()
+
+
+    def initialize(self):
+        return self.deferred
+
+
+
+class FeatureListenAuthenticator(xmlstream.ListenAuthenticator):
+    """
+    Authenticator for receiving entities with support for initializers.
+    """
+
+    def __init__(self):
+        self.completedInitializers = []
+
+
+    def _onElementFallback(self, element):
+        """
+        Fallback observer that rejects XML Stanzas.
+
+        This observer is active while stream feature negotiation has not yet
+        completed.
+        """
+        # ignore elements that are not XML Stanzas
+        if (element.uri not in (self.namespace) or
+            element.name not in ('iq', 'message', 'presence')):
+            return
+
+        # ignore elements that have already been handled
+        if element.handled:
+            return
+
+        exc = error.StreamError('not-authorized')
+        self.xmlstream.sendStreamError(exc)
+
+
+    def connectionMade(self):
+        """
+        Called when the connection has been made.
+
+        Adds an observer to reject XML Stanzas until stream feature negotiation
+        has completed.
+        """
+        xmlstream.ListenAuthenticator.connectionMade(self)
+        self.xmlstream.addObserver(XPATH_ALL, self._onElementFallback, -1)
+
+
+    def _cbInit(self, result):
+        """
+        Mark the initializer as completed and continue to the next.
+        """
+        result, index = result
+        self.completedInitializers.append(self._initializers[index].name)
+        del self._initializers[index]
+
+        if result is xmlstream.Reset:
+            # The initializer initiated a stream restart, bail.
+            return
+        else:
+            self._initializeStream()
+
+
+    def _ebInit(self, failure):
+        """
+        Called when an initializer raises an exception.
+
+        If the exception is a L{error.StreamError} it is sent out, otherwise
+        the error is logged and a stream error with condition
+        C{'internal-server-error'} is sent out instead.
+        """
+        firstError = failure.value
+        subFailure = firstError.subFailure
+        if subFailure.check(error.StreamError):
+            exc = subFailure.value
+        else:
+            log.err(subFailure, index=firstError.index)
+            exc = error.StreamError('internal-server-error')
+
+        self.xmlstream.sendStreamError(exc)
+
+
+    def _initializeStream(self):
+        """
+        Initialize the stream.
+
+        This walks all initializers to retrieve their features and determine
+        if there is at least one required initializer. If not, the stream is
+        ready for the exchange of stanzas. The features are sent out and each
+        initializer will have its C{initialize} method called.
+
+        If negation has completed, L{xmlstream.STREAM_AUTHD_EVENT} is
+        dispatched and the observer rejecting incoming stanzas is removed.
+        """
+        features = domish.Element((xmlstream.NS_STREAMS, 'features'))
+        ds = []
+        required = False
+        for initializer in self._initializers:
+            required = required or initializer.required
+            for feature in initializer.getFeatures():
+                features.addChild(feature)
+            d = initializer.initialize()
+            ds.append(d)
+
+        self.xmlstream.send(features)
+
+        if not required:
+            # There are no required initializers anymore. This stream is
+            # now ready for the exchange of stanzas.
+            self.xmlstream.removeObserver(XPATH_ALL, self._onElementFallback)
+            self.xmlstream.dispatch(self.xmlstream, xmlstream.STREAM_AUTHD_EVENT)
+
+        if ds:
+            d = defer.DeferredList(ds, fireOnOneCallback=True,
+                                       fireOnOneErrback=True,
+                                       consumeErrors=True)
+            d.addCallbacks(self._cbInit, self._ebInit)
+
+
+    def getInitializers(self):
+        """
+        Get the initializers for the current stage of stream negotiation.
+
+        This will be called at the start of each stream start to retrieve
+        the initializers for which the features are advertised and initiated.
+
+        @rtype: C{list} of C{IReceivingInitializer} instances.
+        """
+        raise NotImplementedError()
+
+
+    def checkStream(self):
+        """
+        Check the stream before sending out a stream header and initialization.
+
+        This is the place to inspect the stream properties and raise a relevant
+        L{error.StreamError} if needed.
+        """
+        # Check stream namespace
+        if self.xmlstream.namespace != self.namespace:
+            self.xmlstream.namespace = self.namespace
+            raise error.StreamError('invalid-namespace')
+
+
+    def streamStarted(self, rootElement):
+        """
+        Called when the stream header has been received.
+
+        Check the stream properties through L{checkStream}, send out
+        a stream header, and retrieve and initialize the stream initializers.
+        """
+        xmlstream.ListenAuthenticator.streamStarted(self, rootElement)
+
+        try:
+            self.checkStream()
+        except error.StreamError, exc:
+            self.xmlstream.sendStreamError(exc)
+            return
+
+        self.xmlstream.sendHeader()
+
+        self._initializers = self.getInitializers()
+        self._initializeStream()
+
+
+
 @deprecated(Version("Wokkel", 0, 8, 0), "unicode.encode('idna')")
 def prepareIDNName(name):
     """

wokkel/iwokkel.py

@@ -11 +11 @@
            'IPubSubClient', 'IPubSubService', 'IPubSubResource',
            'IMUCClient', 'IMUCStatuses']
 
-from zope.interface import Interface
+from zope.interface import Attribute, Interface
 from twisted.python.deprecate import deprecatedModuleAttribute
 from twisted.python.versions import Version
 from twisted.words.protocols.jabber.ijabber import IXMPPHandler
@@ -982 +982 @@
         """
         Return the number of status conditions.
         """
+
+
+
+class IReceivingInitializer(Interface):
+    """
+    Interface for XMPP stream initializers for receiving entities.
+    """
+
+    required = Attribute(
+        """
+        This initializer is required to complete feature negotiation.
+        """)
+    name = Attribute(
+        """
+        Identifier for this initializer.
+
+        This identifier is included in
+        L{wokkel.generic.FeatureListenAuthenticator} when an initializer has
+        completed.
+        """)
+    xmlstream = Attribute(
+        """
+        The XML Stream.
+        """)
+    deferred = Attribute(
+        """
+        The deferred returned from initialize.
+        """)
+
+
+    def getFeatures():
+        """
+        Get stream features for this initializer.
+
+        @rtype: C{list} of L{twisted.words.xish.domish.Element}
+        """
+
+
+    def initialize():
+        """
+        Initialize the initializer.
+
+        This is where observers for feature negotiation are set up. When
+        the returned deferred fires, it is assumed to have completed.
+
+        @rtype: L{twisted.internet.defer.Deferred}
+        """

wokkel/test/test_generic.py

@@ -7 +7 @@
 
 import re
 
+from zope.interface import verify
+
+from twisted.internet import defer
 from twisted.python import deprecate
 from twisted.python.versions import Version
+from twisted.test import proto_helpers
 from twisted.trial import unittest
 from twisted.trial.util import suppress as SUPPRESS
 from twisted.words.xish import domish
 from twisted.words.protocols.jabber.jid import JID
+from twisted.words.protocols.jabber import error, xmlstream
 
-from wokkel import generic
+from wokkel import generic, iwokkel
 from wokkel.test.helpers import XmlStreamStub
 
 NS_VERSION = 'jabber:iq:version'
@@ -276 +281 @@
 
 
 
+class BaseReceivingInitializerTest(unittest.TestCase):
+    """
+    Tests for L{generic.BaseReceivingInitializer}.
+    """
+
+    def setUp(self):
+        self.init = generic.BaseReceivingInitializer('init', None)
+
+
+    def test_interface(self):
+        verify.verifyObject(iwokkel.IReceivingInitializer, self.init)
+
+
+    def test_getFeatures(self):
+        self.assertRaises(NotImplementedError, self.init.getFeatures)
+
+
+    def test_initialize(self):
+        d = self.init.initialize()
+        self.init.deferred.callback(None)
+        return d
+
+
+
+class TestableReceivingInitializer(generic.BaseReceivingInitializer):
+    """
+    Testable initializer for receiving entities.
+
+    This initializer advertises support for a stream feature denoted by
+    C{uri} and C{name}. Its C{deferred} should be fired to complete
+    initialization for this initializer.
+
+    @ivar uri: Namespace of the stream feature.
+    @ivar localname: Element localname for the stream feature.
+    """
+    required = True
+
+    def __init__(self, name, xs, uri, localname):
+        generic.BaseReceivingInitializer.__init__(self, name, xs)
+        self.uri = uri
+        self.localname = localname
+        self.deferred = defer.Deferred()
+
+
+    def getFeatures(self):
+        return [domish.Element((self.uri, self.localname))]
+
+
+
+class FeatureListenAuthenticatorTest(unittest.TestCase):
+    """
+    Tests for L{generic.FeatureListenAuthenticator}.
+    """
+
+    def setUp(self):
+        self.gotAuthenticated = False
+        self.initFailure = None
+        self.authenticator = generic.FeatureListenAuthenticator()
+        self.authenticator.namespace = 'jabber:server'
+        self.xmlstream = generic.TestableXmlStream(self.authenticator)
+        self.xmlstream.addObserver('//event/stream/authd',
+                                   self.onAuthenticated)
+        self.xmlstream.addObserver('//event/xmpp/initfailed',
+                                   self.onInitFailed)
+
+        self.init = TestableReceivingInitializer('init', self.xmlstream,
+                                                 'testns', 'test')
+
+        def getInitializers():
+            return [self.init]
+
+        self.authenticator.getInitializers = getInitializers
+
+
+    def onAuthenticated(self, obj):
+        self.gotAuthenticated = True
+
+
+    def onInitFailed(self, failure):
+        self.initFailure = failure
+
+
+    def test_getInitializers(self):
+        """
+        Unoverridden getInitializers raises NotImplementedError.
+        """
+        authenticator = generic.FeatureListenAuthenticator()
+        self.assertRaises(
+            NotImplementedError,
+            authenticator.getInitializers)
+
+
+    def test_streamStarted(self):
+        """
+        Upon stream start, stream initializers are set up.
+
+        The method getInitializers will determine the available stream
+        initializers given the current state of stream initialization.
+        hen, each of the returned initializers will be called to set
+        themselves up.
+        """
+        xs = self.xmlstream
+        xs.makeConnection(proto_helpers.StringTransport())
+        xs.dataReceived("<stream:stream xmlns='jabber:server' "
+                         "xmlns:stream='http://etherx.jabber.org/streams' "
+                         "from='example.com' to='example.org' id='12345' "
+                         "version='1.0'>")
+
+        self.assertTrue(xs.headerSent)
+
+        # Check if features were sent
+        features = xs.output[-1]
+        self.assertEquals(xmlstream.NS_STREAMS, features.uri)
+        self.assertEquals('features', features.name)
+        feature = features.elements().next()
+        self.assertEqual('testns', feature.uri)
+        self.assertEqual('test', feature.name)
+
+        self.assertFalse(self.gotAuthenticated)
+
+        self.init.deferred.callback(None)
+        self.assertTrue(self.gotAuthenticated)
+
+
+    def test_streamStartedStreamError(self):
+        """
+        A stream error raised by the initializer is sent out.
+        """
+        xs = self.xmlstream
+        xs.makeConnection(proto_helpers.StringTransport())
+        xs.dataReceived("<stream:stream xmlns='jabber:server' "
+                         "xmlns:stream='http://etherx.jabber.org/streams' "
+                         "from='example.com' to='example.org' id='12345' "
+                         "version='1.0'>")
+
+        self.assertTrue(xs.headerSent)
+
+        xs.output = []
+        exc = error.StreamError('policy-violation')
+        self.init.deferred.errback(exc)
+
+        self.xmlstream.assertStreamError(self, exc=exc)
+        self.assertFalse(xs.output)
+        self.assertFalse(self.gotAuthenticated)
+
+
+    def test_streamStartedOtherError(self):
+        """
+        Initializer exceptions are logged and yield a internal-server-error.
+        """
+        xs = self.xmlstream
+        xs.makeConnection(proto_helpers.StringTransport())
+        xs.dataReceived("<stream:stream xmlns='jabber:server' "
+                         "xmlns:stream='http://etherx.jabber.org/streams' "
+                         "from='example.com' to='example.org' id='12345' "
+                         "version='1.0'>")
+
+        self.assertTrue(xs.headerSent)
+
+        xs.output = []
+        class Error(Exception):
+            pass
+        self.init.deferred.errback(Error())
+
+        self.xmlstream.assertStreamError(self, condition='internal-server-error')
+        self.assertFalse(xs.output)
+        self.assertFalse(self.gotAuthenticated)
+        self.assertEqual(1, len(self.flushLoggedErrors(Error)))
+
+
+    def test_streamStartedInitializerCompleted(self):
+        """
+        Succesfully finished initializers are recorded.
+        """
+        xs = self.xmlstream
+        xs.makeConnection(proto_helpers.StringTransport())
+        xs.dataReceived("<stream:stream xmlns='jabber:server' "
+                         "xmlns:stream='http://etherx.jabber.org/streams' "
+                         "from='example.com' to='example.org' id='12345' "
+                         "version='1.0'>")
+
+        xs.output = []
+        self.init.deferred.callback(None)
+        self.assertEqual(['init'], self.authenticator.completedInitializers)
+
+
+    def test_streamStartedInitializerCompletedFeatures(self):
+        """
+        After completing an initializer, stream features are sent again.
+
+        In this case, with only one initializer, there are no more features.
+        """
+        xs = self.xmlstream
+        xs.makeConnection(proto_helpers.StringTransport())
+        xs.dataReceived("<stream:stream xmlns='jabber:server' "
+                         "xmlns:stream='http://etherx.jabber.org/streams' "
+                         "from='example.com' to='example.org' id='12345' "
+                         "version='1.0'>")
+
+        xs.output = []
+        self.init.deferred.callback(None)
+
+        self.assertEqual(1, len(xs.output))
+        features = xs.output[-1]
+        self.assertEqual('features', features.name)
+        self.assertEqual(xmlstream.NS_STREAMS, features.uri)
+        self.assertFalse(features.children)
+
+
+    def test_streamStartedInitializerCompletedReset(self):
+        """
+        If an initializer completes with Reset, no features are sent.
+        """
+        xs = self.xmlstream
+        xs.makeConnection(proto_helpers.StringTransport())
+        xs.dataReceived("<stream:stream xmlns='jabber:server' "
+                         "xmlns:stream='http://etherx.jabber.org/streams' "
+                         "from='example.com' to='example.org' id='12345' "
+                         "version='1.0'>")
+
+        xs.output = []
+        self.init.deferred.callback(xmlstream.Reset)
+
+        self.assertEqual(0, len(xs.output))
+
+
+    def test_streamStartedXmlStanzasRejected(self):
+        """
+        XML Stanzas may not be sent before feature negotiation has completed.
+        """
+        xs = self.xmlstream
+        xs.makeConnection(proto_helpers.StringTransport())
+        xs.dataReceived("<stream:stream xmlns='jabber:server' "
+                         "xmlns:stream='http://etherx.jabber.org/streams' "
+                         "from='example.com' to='example.org' id='12345' "
+                         "version='1.0'>")
+
+        xs.dataReceived("<iq to='example.org' from='example.com' type='set'>"
+                        "  <query xmlns='jabber:iq:version'/>"
+                        "</iq>")
+
+        self.xmlstream.assertStreamError(self, condition='not-authorized')
+
+
+    def test_streamStartedCompleteXmlStanzasAllowed(self):
+        """
+        XML Stanzas may sent after feature negotiation has completed.
+        """
+        xs = self.xmlstream
+        xs.makeConnection(proto_helpers.StringTransport())
+        xs.dataReceived("<stream:stream xmlns='jabber:server' "
+                         "xmlns:stream='http://etherx.jabber.org/streams' "
+                         "from='example.com' to='example.org' id='12345' "
+                         "version='1.0'>")
+
+        self.init.deferred.callback(None)
+
+        xs.dataReceived("<iq to='example.org' from='example.com' type='set'>"
+                        "  <query xmlns='jabber:iq:version'/>"
+                        "</iq>")
+
+
+    def test_streamStartedXmlStanzasHandledIgnored(self):
+        """
+        XML Stanzas that have already been handled are ignored.
+        """
+        xs = self.xmlstream
+        xs.makeConnection(proto_helpers.StringTransport())
+        xs.dataReceived("<stream:stream xmlns='jabber:server' "
+                         "xmlns:stream='http://etherx.jabber.org/streams' "
+                         "from='example.com' to='example.org' id='12345' "
+                         "version='1.0'>")
+
+        iq = generic.parseXml("<iq to='example.org' from='example.com' type='set'>"
+                              "  <query xmlns='jabber:iq:version'/>"
+                              "</iq>")
+        iq.handled = True
+        xs.dispatch(iq)
+
+
+    def test_streamStartedNonXmlStanzasIgnored(self):
+        """
+        Elements that are not XML Stranzas are not rejected.
+        """
+        xs = self.xmlstream
+        xs.makeConnection(proto_helpers.StringTransport())
+        xs.dataReceived("<stream:stream xmlns='jabber:server' "
+                         "xmlns:stream='http://etherx.jabber.org/streams' "
+                         "from='example.com' to='example.org' id='12345' "
+                         "version='1.0'>")
+
+        xs.dataReceived("<test xmlns='myns'/>")
+
+
+    def test_streamStartedCheckStream(self):
+        """
+        Stream errors raised by checkStream are sent out.
+        """
+        def checkStream():
+            raise error.StreamError('undefined-condition')
+
+        self.authenticator.checkStream = checkStream
+        xs = self.xmlstream
+        xs.makeConnection(proto_helpers.StringTransport())
+        xs.dataReceived("<stream:stream xmlns='jabber:server' "
+                         "xmlns:stream='http://etherx.jabber.org/streams' "
+                         "from='example.com' to='example.org' id='12345' "
+                         "version='1.0'>")
+
+        self.xmlstream.assertStreamError(self, condition='undefined-condition')
+        self.assertFalse(xs.headerSent)
+
+
+    def test_checkStreamNamespace(self):
+        """
+        The stream namespace must match the pre-defined stream namespace.
+        """
+        xs = self.xmlstream
+        xs.makeConnection(proto_helpers.StringTransport())
+        xs.dataReceived("<stream:stream xmlns='jabber:client' "
+                         "xmlns:stream='http://etherx.jabber.org/streams' "
+                         "from='example.com' to='example.org' id='12345' "
+                         "version='1.0'>")
+
+        self.xmlstream.assertStreamError(self, condition='invalid-namespace')
+
+
+
 class PrepareIDNNameTests(unittest.TestCase):
     """
     Tests for L{wokkel.generic.prepareIDNName}.