1 # Copyright (c) Twisted Matrix Laboratories.
2 # See LICENSE for details.
5 Tests for L{twisted.words.protocols.jabber.client}
8 from twisted.internet import defer
9 from twisted.python.hashlib import sha1
10 from twisted.trial import unittest
11 from twisted.words.protocols.jabber import client, error, jid, xmlstream
12 from twisted.words.protocols.jabber.sasl import SASLInitiatingInitializer
13 from twisted.words.xish import utility
15 IQ_AUTH_GET = '/iq[@type="get"]/query[@xmlns="jabber:iq:auth"]'
16 IQ_AUTH_SET = '/iq[@type="set"]/query[@xmlns="jabber:iq:auth"]'
17 NS_BIND = 'urn:ietf:params:xml:ns:xmpp-bind'
18 IQ_BIND_SET = '/iq[@type="set"]/bind[@xmlns="%s"]' % NS_BIND
19 NS_SESSION = 'urn:ietf:params:xml:ns:xmpp-session'
20 IQ_SESSION_SET = '/iq[@type="set"]/session[@xmlns="%s"]' % NS_SESSION
22 class CheckVersionInitializerTest(unittest.TestCase):
24 a = xmlstream.Authenticator()
25 xs = xmlstream.XmlStream(a)
26 self.init = client.CheckVersionInitializer(xs)
29 def testSupported(self):
31 Test supported version number 1.0
33 self.init.xmlstream.version = (1, 0)
34 self.init.initialize()
37 def testNotSupported(self):
39 Test unsupported version number 0.0, and check exception.
41 self.init.xmlstream.version = (0, 0)
42 exc = self.assertRaises(error.StreamError, self.init.initialize)
43 self.assertEqual('unsupported-version', exc.condition)
47 class InitiatingInitializerHarness(object):
49 Testing harness for interacting with XML stream initializers.
51 This sets up an L{utility.XmlPipe} to create a communication channel between
52 the initializer and the stubbed receiving entity. It features a sink and
53 source side that both act similarly to a real L{xmlstream.XmlStream}. The
54 sink is augmented with an authenticator to which initializers can be added.
56 The harness also provides some utility methods to work with event observers
62 self.pipe = utility.XmlPipe()
63 self.xmlstream = self.pipe.sink
64 self.authenticator = xmlstream.ConnectAuthenticator('example.org')
65 self.xmlstream.authenticator = self.authenticator
68 def waitFor(self, event, handler):
70 Observe an output event, returning a deferred.
72 The returned deferred will be fired when the given event has been
73 observed on the source end of the L{XmlPipe} tied to the protocol
74 under test. The handler is added as the first callback.
76 @param event: The event to be observed. See
77 L{utility.EventDispatcher.addOnetimeObserver}.
78 @param handler: The handler to be called with the observed event object.
79 @rtype: L{defer.Deferred}.
82 d.addCallback(handler)
83 self.pipe.source.addOnetimeObserver(event, d.callback)
88 class IQAuthInitializerTest(InitiatingInitializerHarness, unittest.TestCase):
90 Tests for L{client.IQAuthInitializer}.
94 super(IQAuthInitializerTest, self).setUp()
95 self.init = client.IQAuthInitializer(self.xmlstream)
96 self.authenticator.jid = jid.JID('user@example.com/resource')
97 self.authenticator.password = 'secret'
100 def testPlainText(self):
102 Test plain-text authentication.
104 Act as a server supporting plain-text authentication and expect the
105 C{password} field to be filled with the password. Then act as if
106 authentication succeeds.
111 Called when the initializer sent a query for authentication methods.
113 The response informs the client that plain-text authentication
117 # Create server response
118 response = xmlstream.toResponse(iq, 'result')
119 response.addElement(('jabber:iq:auth', 'query'))
120 response.query.addElement('username')
121 response.query.addElement('password')
122 response.query.addElement('resource')
124 # Set up an observer for the next request we expect.
125 d = self.waitFor(IQ_AUTH_SET, onAuthSet)
127 # Send server response
128 self.pipe.source.send(response)
134 Called when the initializer sent the authentication request.
136 The server checks the credentials and responds with an empty result
139 self.assertEqual('user', unicode(iq.query.username))
140 self.assertEqual('secret', unicode(iq.query.password))
141 self.assertEqual('resource', unicode(iq.query.resource))
143 # Send server response
144 response = xmlstream.toResponse(iq, 'result')
145 self.pipe.source.send(response)
147 # Set up an observer for the request for authentication fields
148 d1 = self.waitFor(IQ_AUTH_GET, onAuthGet)
150 # Start the initializer
151 d2 = self.init.initialize()
152 return defer.gatherResults([d1, d2])
155 def testDigest(self):
157 Test digest authentication.
159 Act as a server supporting digest authentication and expect the
160 C{digest} field to be filled with a sha1 digest of the concatenated
161 stream session identifier and password. Then act as if authentication
167 Called when the initializer sent a query for authentication methods.
169 The response informs the client that digest authentication is
173 # Create server response
174 response = xmlstream.toResponse(iq, 'result')
175 response.addElement(('jabber:iq:auth', 'query'))
176 response.query.addElement('username')
177 response.query.addElement('digest')
178 response.query.addElement('resource')
180 # Set up an observer for the next request we expect.
181 d = self.waitFor(IQ_AUTH_SET, onAuthSet)
183 # Send server response
184 self.pipe.source.send(response)
190 Called when the initializer sent the authentication request.
192 The server checks the credentials and responds with an empty result
195 self.assertEqual('user', unicode(iq.query.username))
196 self.assertEqual(sha1('12345secret').hexdigest(),
197 unicode(iq.query.digest).encode('utf-8'))
198 self.assertEqual('resource', unicode(iq.query.resource))
200 # Send server response
201 response = xmlstream.toResponse(iq, 'result')
202 self.pipe.source.send(response)
204 # Digest authentication relies on the stream session identifier. Set it.
205 self.xmlstream.sid = u'12345'
207 # Set up an observer for the request for authentication fields
208 d1 = self.waitFor(IQ_AUTH_GET, onAuthGet)
210 # Start the initializer
211 d2 = self.init.initialize()
213 return defer.gatherResults([d1, d2])
216 def testFailRequestFields(self):
218 Test initializer failure of request for fields for authentication.
222 Called when the initializer sent a query for authentication methods.
224 The server responds that the client is not authorized to authenticate.
226 response = error.StanzaError('not-authorized').toResponse(iq)
227 self.pipe.source.send(response)
229 # Set up an observer for the request for authentication fields
230 d1 = self.waitFor(IQ_AUTH_GET, onAuthGet)
232 # Start the initializer
233 d2 = self.init.initialize()
235 # The initialized should fail with a stanza error.
236 self.assertFailure(d2, error.StanzaError)
238 return defer.gatherResults([d1, d2])
241 def testFailAuth(self):
243 Test initializer failure to authenticate.
248 Called when the initializer sent a query for authentication methods.
250 The response informs the client that plain-text authentication
254 # Send server response
255 response = xmlstream.toResponse(iq, 'result')
256 response.addElement(('jabber:iq:auth', 'query'))
257 response.query.addElement('username')
258 response.query.addElement('password')
259 response.query.addElement('resource')
261 # Set up an observer for the next request we expect.
262 d = self.waitFor(IQ_AUTH_SET, onAuthSet)
264 # Send server response
265 self.pipe.source.send(response)
271 Called when the initializer sent the authentication request.
273 The server checks the credentials and responds with a not-authorized
276 response = error.StanzaError('not-authorized').toResponse(iq)
277 self.pipe.source.send(response)
279 # Set up an observer for the request for authentication fields
280 d1 = self.waitFor(IQ_AUTH_GET, onAuthGet)
282 # Start the initializer
283 d2 = self.init.initialize()
285 # The initializer should fail with a stanza error.
286 self.assertFailure(d2, error.StanzaError)
288 return defer.gatherResults([d1, d2])
292 class BindInitializerTest(InitiatingInitializerHarness, unittest.TestCase):
294 Tests for L{client.BindInitializer}.
298 super(BindInitializerTest, self).setUp()
299 self.init = client.BindInitializer(self.xmlstream)
300 self.authenticator.jid = jid.JID('user@example.com/resource')
305 Set up a stream, and act as if resource binding succeeds.
308 response = xmlstream.toResponse(iq, 'result')
309 response.addElement((NS_BIND, 'bind'))
310 response.bind.addElement('jid',
311 content='user@example.com/other resource')
312 self.pipe.source.send(response)
315 self.assertEqual(jid.JID('user@example.com/other resource'),
316 self.authenticator.jid)
318 d1 = self.waitFor(IQ_BIND_SET, onBind)
319 d2 = self.init.start()
321 return defer.gatherResults([d1, d2])
324 def testFailure(self):
326 Set up a stream, and act as if resource binding fails.
329 response = error.StanzaError('conflict').toResponse(iq)
330 self.pipe.source.send(response)
332 d1 = self.waitFor(IQ_BIND_SET, onBind)
333 d2 = self.init.start()
334 self.assertFailure(d2, error.StanzaError)
335 return defer.gatherResults([d1, d2])
339 class SessionInitializerTest(InitiatingInitializerHarness, unittest.TestCase):
341 Tests for L{client.SessionInitializer}.
345 super(SessionInitializerTest, self).setUp()
346 self.init = client.SessionInitializer(self.xmlstream)
349 def testSuccess(self):
351 Set up a stream, and act as if session establishment succeeds.
355 response = xmlstream.toResponse(iq, 'result')
356 self.pipe.source.send(response)
358 d1 = self.waitFor(IQ_SESSION_SET, onSession)
359 d2 = self.init.start()
360 return defer.gatherResults([d1, d2])
363 def testFailure(self):
365 Set up a stream, and act as if session establishment fails.
368 response = error.StanzaError('forbidden').toResponse(iq)
369 self.pipe.source.send(response)
371 d1 = self.waitFor(IQ_SESSION_SET, onSession)
372 d2 = self.init.start()
373 self.assertFailure(d2, error.StanzaError)
374 return defer.gatherResults([d1, d2])
378 class XMPPAuthenticatorTest(unittest.TestCase):
380 Test for both XMPPAuthenticator and XMPPClientFactory.
384 Test basic operations.
386 Setup an XMPPClientFactory, which sets up an XMPPAuthenticator, and let
387 it produce a protocol instance. Then inspect the instance variables of
388 the authenticator and XML stream objects.
390 self.client_jid = jid.JID('user@example.com/resource')
392 # Get an XmlStream instance. Note that it gets initialized with the
393 # XMPPAuthenticator (that has its associateWithXmlStream called) that
394 # is in turn initialized with the arguments to the factory.
395 xs = client.XMPPClientFactory(self.client_jid,
396 'secret').buildProtocol(None)
398 # test authenticator's instance variables
399 self.assertEqual('example.com', xs.authenticator.otherHost)
400 self.assertEqual(self.client_jid, xs.authenticator.jid)
401 self.assertEqual('secret', xs.authenticator.password)
403 # test list of initializers
404 version, tls, sasl, bind, session = xs.initializers
406 self.assert_(isinstance(tls, xmlstream.TLSInitiatingInitializer))
407 self.assert_(isinstance(sasl, SASLInitiatingInitializer))
408 self.assert_(isinstance(bind, client.BindInitializer))
409 self.assert_(isinstance(session, client.SessionInitializer))
411 self.assertFalse(tls.required)
412 self.assertTrue(sasl.required)
413 self.assertFalse(bind.required)
414 self.assertFalse(session.required)