| # |
| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you under the Apache License, Version 2.0 (the |
| # "License"); you may not use this file except in compliance |
| # with the License. You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, |
| # software distributed under the License is distributed on an |
| # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| # KIND, either express or implied. See the License for the |
| # specific language governing permissions and limitations |
| # under the License. |
| # |
| |
| from __future__ import absolute_import |
| |
| import os |
| |
| from proton import * |
| |
| from . import common |
| from . import engine |
| from .common import pump, Skipped |
| |
| def _sslCertpath(file): |
| """ Return the full path to the certificate, keyfile, etc. |
| """ |
| if os.name=="nt": |
| if file.find("private-key")!=-1: |
| # The private key is not in a separate store |
| return None |
| # Substitute pkcs#12 equivalent for the CA/key store |
| if file.endswith(".pem"): |
| file = file[:-4] + ".p12" |
| return os.path.join(os.path.dirname(__file__), |
| "ssl_db/%s" % file) |
| |
| def _testSaslMech(self, mech, clientUser='user@proton', authUser='user@proton', authzid=None, encrypted=False, authenticated=True): |
| self.s1.allowed_mechs(mech) |
| self.c1.open() |
| self.c2.open() |
| |
| pump(self.t1, self.t2, 1024) |
| |
| if encrypted is not None: |
| assert self.t2.encrypted == encrypted, encrypted |
| assert self.t1.encrypted == encrypted, encrypted |
| |
| if authzid is None: |
| authzid = authUser |
| |
| assert self.t2.authenticated == authenticated, authenticated |
| assert self.t1.authenticated == authenticated, authenticated |
| if authenticated: |
| # Server |
| assert self.t2.user == authUser |
| assert self.s2.user == authUser |
| assert self.s2.authorization == authzid, self.s2.authorization |
| assert self.s2.mech == mech.strip() |
| assert self.s2.outcome == SASL.OK, self.s2.outcome |
| assert self.c2.state & Endpoint.LOCAL_ACTIVE and self.c2.state & Endpoint.REMOTE_ACTIVE,\ |
| "local_active=%s, remote_active=%s" % (self.c1.state & Endpoint.LOCAL_ACTIVE, self.c1.state & Endpoint.REMOTE_ACTIVE) |
| # Client |
| assert self.t1.user == clientUser |
| assert self.s1.user == clientUser |
| assert self.s1.mech == mech.strip() |
| assert self.s1.outcome == SASL.OK, self.s1.outcome |
| assert self.c1.state & Endpoint.LOCAL_ACTIVE and self.c1.state & Endpoint.REMOTE_ACTIVE,\ |
| "local_active=%s, remote_active=%s" % (self.c1.state & Endpoint.LOCAL_ACTIVE, self.c1.state & Endpoint.REMOTE_ACTIVE) |
| else: |
| # Server |
| assert self.t2.user == None |
| assert self.s2.user == None |
| assert self.s2.authorization == None, self.s2.authorization |
| assert self.s2.outcome != SASL.OK, self.s2.outcome |
| # Client |
| assert self.t1.user == clientUser |
| assert self.s1.user == clientUser |
| assert self.s1.outcome != SASL.OK, self.s1.outcome |
| |
| class Test(common.Test): |
| pass |
| |
| def consumeAllOuput(t): |
| stops = 0 |
| while stops<1: |
| out = t.peek(1024) |
| l = len(out) if out else 0 |
| t.pop(l) |
| if l <= 0: |
| stops += 1 |
| |
| class SaslTest(Test): |
| |
| def setUp(self): |
| self.t1 = Transport() |
| self.s1 = SASL(self.t1) |
| self.t2 = Transport(Transport.SERVER) |
| self.t2.max_frame_size = 65536 |
| self.s2 = SASL(self.t2) |
| |
| def pump(self): |
| pump(self.t1, self.t2, 1024) |
| |
| # We have to generate the client frames manually because proton does not |
| # generate pipelined SASL and AMQP frames together |
| def testIllegalProtocolLayering(self): |
| # Server |
| self.s2.allowed_mechs('ANONYMOUS') |
| |
| c2 = Connection() |
| self.t2.bind(c2) |
| |
| assert self.s2.outcome is None |
| |
| # Push client bytes into server |
| self.t2.push( |
| # SASL |
| b'AMQP\x03\x01\x00\x00' |
| # @sasl-init(65) [mechanism=:ANONYMOUS, initial-response=b"anonymous@fuschia"] |
| b'\x00\x00\x002\x02\x01\x00\x00\x00SA\xd0\x00\x00\x00"\x00\x00\x00\x02\xa3\x09ANONYMOUS\xa0\x11anonymous@fuschia' |
| # SASL (again illegally) |
| b'AMQP\x03\x01\x00\x00' |
| # @sasl-init(65) [mechanism=:ANONYMOUS, initial-response=b"anonymous@fuschia"] |
| b'\x00\x00\x002\x02\x01\x00\x00\x00SA\xd0\x00\x00\x00"\x00\x00\x00\x02\xa3\x09ANONYMOUS\xa0\x11anonymous@fuschia' |
| # AMQP |
| b'AMQP\x00\x01\x00\x00' |
| # @open(16) [container-id="", channel-max=1234] |
| b'\x00\x00\x00!\x02\x00\x00\x00\x00S\x10\xd0\x00\x00\x00\x11\x00\x00\x00\x0a\xa1\x00@@`\x04\xd2@@@@@@' |
| ) |
| |
| consumeAllOuput(self.t2) |
| |
| assert self.t2.condition |
| assert self.t2.closed |
| assert not c2.state & Endpoint.REMOTE_ACTIVE |
| |
| def testPipelinedClient(self): |
| # Server |
| self.s2.allowed_mechs('ANONYMOUS') |
| |
| c2 = Connection() |
| self.t2.bind(c2) |
| |
| assert self.s2.outcome is None |
| |
| # Push client bytes into server |
| self.t2.push( |
| # SASL |
| b'AMQP\x03\x01\x00\x00' |
| # @sasl-init(65) [mechanism=:ANONYMOUS, initial-response=b"anonymous@fuschia"] |
| b'\x00\x00\x002\x02\x01\x00\x00\x00SA\xd0\x00\x00\x00"\x00\x00\x00\x02\xa3\x09ANONYMOUS\xa0\x11anonymous@fuschia' |
| # AMQP |
| b'AMQP\x00\x01\x00\x00' |
| # @open(16) [container-id="", channel-max=1234] |
| b'\x00\x00\x00!\x02\x00\x00\x00\x00S\x10\xd0\x00\x00\x00\x11\x00\x00\x00\x0a\xa1\x00@@`\x04\xd2@@@@@@' |
| ) |
| |
| consumeAllOuput(self.t2) |
| |
| assert not self.t2.condition |
| assert self.s2.outcome == SASL.OK |
| assert c2.state & Endpoint.REMOTE_ACTIVE |
| |
| def testPipelinedServer(self): |
| # Client |
| self.s1.allowed_mechs('ANONYMOUS') |
| |
| c1 = Connection() |
| self.t1.bind(c1) |
| |
| assert self.s1.outcome is None |
| |
| # Push server bytes into client |
| # Commented out lines in this test are where the client input processing doesn't |
| # run after output processing even though there is input waiting |
| self.t1.push( |
| # SASL |
| b'AMQP\x03\x01\x00\x00' |
| # @sasl-mechanisms(64) [sasl-server-mechanisms=@PN_SYMBOL[:ANONYMOUS]] |
| b'\x00\x00\x00\x1c\x02\x01\x00\x00\x00S@\xc0\x0f\x01\xe0\x0c\x01\xa3\tANONYMOUS' |
| # @sasl-outcome(68) [code=0] |
| b'\x00\x00\x00\x10\x02\x01\x00\x00\x00SD\xc0\x03\x01P\x00' |
| # AMQP |
| b'AMQP\x00\x01\x00\x00' |
| # @open(16) [container-id="", channel-max=1234] |
| b'\x00\x00\x00!\x02\x00\x00\x00\x00S\x10\xd0\x00\x00\x00\x11\x00\x00\x00\x0a\xa1\x00@@`\x04\xd2@@@@@@' |
| ) |
| |
| consumeAllOuput(self.t1) |
| |
| assert self.s1.outcome == SASL.OK |
| assert c1.state & Endpoint.REMOTE_ACTIVE |
| |
| def testPipelined2(self): |
| out1 = self.t1.peek(1024) |
| self.t1.pop(len(out1)) |
| self.t2.push(out1) |
| |
| self.s2.allowed_mechs('ANONYMOUS') |
| c2 = Connection() |
| c2.open() |
| self.t2.bind(c2) |
| |
| out2 = self.t2.peek(1024) |
| self.t2.pop(len(out2)) |
| self.t1.push(out2) |
| |
| out1 = self.t1.peek(1024) |
| assert len(out1) > 0 |
| |
| def testFracturedSASL(self): |
| """ PROTON-235 |
| """ |
| assert self.s1.outcome is None |
| |
| # self.t1.trace(Transport.TRACE_FRM) |
| |
| out = self.t1.peek(1024) |
| self.t1.pop(len(out)) |
| self.t1.push(b"AMQP\x03\x01\x00\x00") |
| out = self.t1.peek(1024) |
| self.t1.pop(len(out)) |
| self.t1.push(b"\x00\x00\x00") |
| out = self.t1.peek(1024) |
| self.t1.pop(len(out)) |
| |
| self.t1.push(b"6\x02\x01\x00\x00\x00S@\xc0\x29\x01\xe0\x26\x04\xa3\x05PLAIN\x0aDIGEST-MD5\x09ANONYMOUS\x08CRAM-MD5") |
| out = self.t1.peek(1024) |
| self.t1.pop(len(out)) |
| self.t1.push(b"\x00\x00\x00\x10\x02\x01\x00\x00\x00SD\xc0\x03\x01P\x00") |
| out = self.t1.peek(1024) |
| self.t1.pop(len(out)) |
| while out: |
| out = self.t1.peek(1024) |
| self.t1.pop(len(out)) |
| |
| assert self.s1.outcome == SASL.OK, self.s1.outcome |
| |
| def test_singleton(self): |
| """Verify that only a single instance of SASL can exist per Transport""" |
| transport = Transport() |
| attr = object() |
| sasl1 = SASL(transport) |
| sasl1.my_attribute = attr |
| sasl2 = transport.sasl() |
| sasl3 = SASL(transport) |
| assert sasl1 == sasl2 |
| assert sasl1 == sasl3 |
| assert sasl1.my_attribute == attr |
| assert sasl2.my_attribute == attr |
| assert sasl3.my_attribute == attr |
| transport = Transport() |
| sasl1 = transport.sasl() |
| sasl1.my_attribute = attr |
| sasl2 = SASL(transport) |
| assert sasl1 == sasl2 |
| assert sasl1.my_attribute == attr |
| assert sasl2.my_attribute == attr |
| |
| def testSaslSkipped(self): |
| """Verify that the server (with SASL) correctly handles a client without SASL""" |
| self.t1 = Transport() |
| self.t2.require_auth(False) |
| self.pump() |
| assert self.s2.outcome == None |
| assert self.t2.condition == None |
| assert self.t2.authenticated == False |
| assert self.s1.outcome == None |
| assert self.t1.condition == None |
| assert self.t1.authenticated == False |
| |
| def testSaslSkippedFail(self): |
| """Verify that the server (with SASL) correctly handles a client without SASL""" |
| self.t1 = Transport() |
| self.t2.require_auth(True) |
| self.pump() |
| assert self.s2.outcome == None |
| assert self.t2.condition != None |
| assert self.s1.outcome == None |
| assert self.t1.condition != None |
| |
| def testMechNotFound(self): |
| self.c1 = Connection() |
| self.c1.open() |
| self.t1.bind(self.c1) |
| self.s1.allowed_mechs('IMPOSSIBLE') |
| |
| self.pump() |
| |
| assert self.t2.authenticated == False |
| assert self.t1.authenticated == False |
| assert self.s1.outcome != SASL.OK |
| assert self.s2.outcome != SASL.OK |
| |
| class SASLMechTest(Test): |
| def setUp(self): |
| self.t1 = Transport() |
| self.s1 = SASL(self.t1) |
| self.t2 = Transport(Transport.SERVER) |
| self.s2 = SASL(self.t2) |
| |
| self.c1 = Connection() |
| self.c1.user = 'user@proton' |
| self.c1.password = 'password' |
| self.c1.hostname = 'localhost' |
| |
| self.c2 = Connection() |
| |
| def testANON(self): |
| self.t1.bind(self.c1) |
| self.t2.bind(self.c2) |
| _testSaslMech(self, 'ANONYMOUS', authUser='anonymous') |
| |
| def testCRAMMD5(self): |
| common.ensureCanTestExtendedSASL() |
| |
| self.t1.bind(self.c1) |
| self.t2.bind(self.c2) |
| _testSaslMech(self, 'CRAM-MD5') |
| |
| def testDIGESTMD5(self): |
| common.ensureCanTestExtendedSASL() |
| |
| self.t1.bind(self.c1) |
| self.t2.bind(self.c2) |
| _testSaslMech(self, 'DIGEST-MD5') |
| |
| # PLAIN shouldn't work without encryption without special setting |
| def testPLAINfail(self): |
| common.ensureCanTestExtendedSASL() |
| |
| self.t1.bind(self.c1) |
| self.t2.bind(self.c2) |
| _testSaslMech(self, 'PLAIN', authenticated=False) |
| |
| # Client won't accept PLAIN even if offered by server without special setting |
| def testPLAINClientFail(self): |
| common.ensureCanTestExtendedSASL() |
| |
| self.s2.allow_insecure_mechs = True |
| self.t1.bind(self.c1) |
| self.t2.bind(self.c2) |
| _testSaslMech(self, 'PLAIN', authenticated=False) |
| |
| # PLAIN will only work if both ends are specially set up |
| def testPLAIN(self): |
| common.ensureCanTestExtendedSASL() |
| |
| self.s1.allow_insecure_mechs = True |
| self.s2.allow_insecure_mechs = True |
| self.t1.bind(self.c1) |
| self.t2.bind(self.c2) |
| _testSaslMech(self, 'PLAIN') |
| |
| # SCRAM not supported before Cyrus SASL 2.1.26 |
| # so not universal and hence need a test for support |
| # to keep it in tests. |
| # def testSCRAMSHA1(self): |
| # common.ensureCanTestExtendedSASL() |
| # |
| # self.t1.bind(self.c1) |
| # self.t2.bind(self.c2) |
| # _testSaslMech(self, 'SCRAM-SHA-1') |
| |
| def _sslConnection(domain, transport, connection): |
| transport.bind(connection) |
| ssl = SSL(transport, domain, None ) |
| return connection |
| |
| class SSLSASLTest(Test): |
| def setUp(self): |
| if not common.isSSLPresent(): |
| raise Skipped("No SSL libraries found.") |
| |
| self.server_domain = SSLDomain(SSLDomain.MODE_SERVER) |
| self.client_domain = SSLDomain(SSLDomain.MODE_CLIENT) |
| |
| self.t1 = Transport() |
| self.s1 = SASL(self.t1) |
| self.t2 = Transport(Transport.SERVER) |
| self.s2 = SASL(self.t2) |
| |
| self.c1 = Connection() |
| self.c2 = Connection() |
| |
| def testSSLPlainSimple(self): |
| if not SASL.extended(): |
| raise Skipped("Simple SASL server does not support PLAIN") |
| common.ensureCanTestExtendedSASL() |
| |
| clientUser = 'user@proton' |
| mech = 'PLAIN' |
| |
| self.c1.user = clientUser |
| self.c1.password = 'password' |
| self.c1.hostname = 'localhost' |
| |
| ssl1 = _sslConnection(self.client_domain, self.t1, self.c1) |
| ssl2 = _sslConnection(self.server_domain, self.t2, self.c2) |
| |
| _testSaslMech(self, mech, encrypted=True) |
| |
| def testSSLPlainAuthzid(self): |
| if not SASL.extended(): |
| raise Skipped("Simple SASL server does not support PLAIN") |
| common.ensureCanTestExtendedSASL() |
| |
| clientUser = 'user@proton' |
| authzid = 'user2' |
| mech = 'PLAIN' |
| |
| self.c1.user = clientUser |
| self.c1.authorization = authzid |
| self.c1.password = 'password' |
| self.c1.hostname = 'localhost' |
| |
| ssl1 = _sslConnection(self.client_domain, self.t1, self.c1) |
| ssl2 = _sslConnection(self.server_domain, self.t2, self.c2) |
| |
| _testSaslMech(self, mech, authzid=authzid, encrypted=True) |
| |
| def testSSLPlainSimpleFail(self): |
| if not SASL.extended(): |
| raise Skipped("Simple SASL server does not support PLAIN") |
| common.ensureCanTestExtendedSASL() |
| |
| clientUser = 'usr@proton' |
| mech = 'PLAIN' |
| |
| self.c1.user = clientUser |
| self.c1.password = 'password' |
| self.c1.hostname = 'localhost' |
| |
| ssl1 = _sslConnection(self.client_domain, self.t1, self.c1) |
| ssl2 = _sslConnection(self.server_domain, self.t2, self.c2) |
| |
| _testSaslMech(self, mech, clientUser='usr@proton', encrypted=True, authenticated=False) |
| |
| def testSSLExternalSimple(self): |
| if os.name=="nt": |
| extUser = 'O=Client, CN=127.0.0.1' |
| else: |
| extUser = 'O=Client,CN=127.0.0.1' |
| mech = 'EXTERNAL' |
| |
| self.server_domain.set_credentials(_sslCertpath("server-certificate.pem"), |
| _sslCertpath("server-private-key.pem"), |
| "server-password") |
| self.server_domain.set_trusted_ca_db(_sslCertpath("ca-certificate.pem")) |
| self.server_domain.set_peer_authentication(SSLDomain.VERIFY_PEER, |
| _sslCertpath("ca-certificate.pem") ) |
| self.client_domain.set_credentials(_sslCertpath("client-certificate.pem"), |
| _sslCertpath("client-private-key.pem"), |
| "client-password") |
| self.client_domain.set_trusted_ca_db(_sslCertpath("ca-certificate.pem")) |
| self.client_domain.set_peer_authentication(SSLDomain.VERIFY_PEER) |
| |
| ssl1 = _sslConnection(self.client_domain, self.t1, self.c1) |
| ssl2 = _sslConnection(self.server_domain, self.t2, self.c2) |
| |
| _testSaslMech(self, mech, clientUser=None, authUser=extUser, encrypted=True) |
| |
| def testSSLExternalAuthzid(self): |
| if os.name=="nt": |
| extUser = 'O=Client, CN=127.0.0.1' |
| else: |
| extUser = 'O=Client,CN=127.0.0.1' |
| mech = 'EXTERNAL' |
| authzid = 'user_foo' |
| |
| self.c1.authorization = authzid |
| |
| self.server_domain.set_credentials(_sslCertpath("server-certificate.pem"), |
| _sslCertpath("server-private-key.pem"), |
| "server-password") |
| self.server_domain.set_trusted_ca_db(_sslCertpath("ca-certificate.pem")) |
| self.server_domain.set_peer_authentication(SSLDomain.VERIFY_PEER, |
| _sslCertpath("ca-certificate.pem") ) |
| self.client_domain.set_credentials(_sslCertpath("client-certificate.pem"), |
| _sslCertpath("client-private-key.pem"), |
| "client-password") |
| self.client_domain.set_trusted_ca_db(_sslCertpath("ca-certificate.pem")) |
| self.client_domain.set_peer_authentication(SSLDomain.VERIFY_PEER) |
| |
| ssl1 = _sslConnection(self.client_domain, self.t1, self.c1) |
| ssl2 = _sslConnection(self.server_domain, self.t2, self.c2) |
| |
| _testSaslMech(self, mech, clientUser=None, authUser=extUser, authzid=authzid, encrypted=True) |
| |
| def testSSLExternalSimpleFail(self): |
| mech = 'EXTERNAL' |
| |
| self.server_domain.set_credentials(_sslCertpath("server-certificate.pem"), |
| _sslCertpath("server-private-key.pem"), |
| "server-password") |
| self.server_domain.set_trusted_ca_db(_sslCertpath("ca-certificate.pem")) |
| self.server_domain.set_peer_authentication(SSLDomain.VERIFY_PEER, |
| _sslCertpath("ca-certificate.pem") ) |
| self.client_domain.set_trusted_ca_db(_sslCertpath("ca-certificate.pem")) |
| self.client_domain.set_peer_authentication(SSLDomain.VERIFY_PEER) |
| |
| ssl1 = _sslConnection(self.client_domain, self.t1, self.c1) |
| ssl2 = _sslConnection(self.server_domain, self.t2, self.c2) |
| |
| _testSaslMech(self, mech, clientUser=None, authUser=None, encrypted=None, authenticated=False) |
| |
| class SASLEventTest(engine.CollectorTest): |
| def setUp(self): |
| engine.CollectorTest.setUp(self) |
| self.t1 = Transport() |
| self.s1 = SASL(self.t1) |
| self.t2 = Transport(Transport.SERVER) |
| self.s2 = SASL(self.t2) |
| |
| self.c1 = Connection() |
| self.c1.user = 'user@proton' |
| self.c1.password = 'password' |
| self.c1.hostname = 'localhost' |
| |
| self.c2 = Connection() |
| |
| self.collector = Collector() |
| |
| def testNormalAuthenticationClient(self): |
| common.ensureCanTestExtendedSASL() |
| self.c1.collect(self.collector) |
| self.t1.bind(self.c1) |
| self.t2.bind(self.c2) |
| _testSaslMech(self, 'DIGEST-MD5') |
| self.expect(Event.CONNECTION_INIT, Event.CONNECTION_BOUND, |
| Event.CONNECTION_LOCAL_OPEN, Event.TRANSPORT, |
| Event.CONNECTION_REMOTE_OPEN) |
| |
| def testNormalAuthenticationServer(self): |
| common.ensureCanTestExtendedSASL() |
| self.c2.collect(self.collector) |
| self.t1.bind(self.c1) |
| self.t2.bind(self.c2) |
| _testSaslMech(self, 'DIGEST-MD5') |
| self.expect(Event.CONNECTION_INIT, Event.CONNECTION_BOUND, |
| Event.CONNECTION_LOCAL_OPEN, Event.TRANSPORT, |
| Event.CONNECTION_REMOTE_OPEN) |
| |
| def testFailedAuthenticationClient(self): |
| common.ensureCanTestExtendedSASL() |
| clientUser = "usr@proton" |
| self.c1.user = clientUser |
| self.c1.collect(self.collector) |
| self.t1.bind(self.c1) |
| self.t2.bind(self.c2) |
| _testSaslMech(self, 'DIGEST-MD5', clientUser=clientUser, authenticated=False) |
| self.expect(Event.CONNECTION_INIT, Event.CONNECTION_BOUND, |
| Event.CONNECTION_LOCAL_OPEN, Event.TRANSPORT, |
| Event.TRANSPORT_ERROR, |
| Event.TRANSPORT_TAIL_CLOSED, |
| Event.TRANSPORT_HEAD_CLOSED, Event.TRANSPORT_CLOSED) |
| |
| def testFailedAuthenticationServer(self): |
| common.ensureCanTestExtendedSASL() |
| clientUser = "usr@proton" |
| self.c1.user = clientUser |
| self.c2.collect(self.collector) |
| self.t1.bind(self.c1) |
| self.t2.bind(self.c2) |
| _testSaslMech(self, 'DIGEST-MD5', clientUser=clientUser, authenticated=False) |
| self.expect(Event.CONNECTION_INIT, Event.CONNECTION_BOUND, |
| Event.CONNECTION_LOCAL_OPEN, Event.TRANSPORT, |
| Event.TRANSPORT_ERROR, |
| Event.TRANSPORT_TAIL_CLOSED, |
| Event.TRANSPORT_HEAD_CLOSED, Event.TRANSPORT_CLOSED) |
| |
| def testNoMechClient(self): |
| common.ensureCanTestExtendedSASL() |
| self.c1.collect(self.collector) |
| self.s2.allowed_mechs('IMPOSSIBLE') |
| self.t1.bind(self.c1) |
| self.t2.bind(self.c2) |
| _testSaslMech(self, 'DIGEST-MD5', authenticated=False) |
| self.expect(Event.CONNECTION_INIT, Event.CONNECTION_BOUND, |
| Event.CONNECTION_LOCAL_OPEN, Event.TRANSPORT, |
| Event.TRANSPORT_ERROR, |
| Event.TRANSPORT_TAIL_CLOSED, Event.TRANSPORT_HEAD_CLOSED, Event.TRANSPORT_CLOSED) |
| |
| def testNoMechServer(self): |
| common.ensureCanTestExtendedSASL() |
| self.c2.collect(self.collector) |
| self.s2.allowed_mechs('IMPOSSIBLE') |
| self.t1.bind(self.c1) |
| self.t2.bind(self.c2) |
| _testSaslMech(self, 'DIGEST-MD5', authenticated=False) |
| self.expect(Event.CONNECTION_INIT, Event.CONNECTION_BOUND, |
| Event.CONNECTION_LOCAL_OPEN, Event.TRANSPORT, |
| Event.TRANSPORT_TAIL_CLOSED, |
| Event.TRANSPORT_ERROR, Event.TRANSPORT_HEAD_CLOSED, Event.TRANSPORT_CLOSED) |
| |
| def testDisallowedMechClient(self): |
| self.c1.collect(self.collector) |
| self.t1.bind(self.c1) |
| self.t2.bind(self.c2) |
| _testSaslMech(self, 'IMPOSSIBLE', authenticated=False) |
| self.expect(Event.CONNECTION_INIT, Event.CONNECTION_BOUND, |
| Event.CONNECTION_LOCAL_OPEN, Event.TRANSPORT, |
| Event.TRANSPORT_ERROR, |
| Event.TRANSPORT_TAIL_CLOSED, Event.TRANSPORT_HEAD_CLOSED, Event.TRANSPORT_CLOSED) |
| |
| def testDisallowedMechServer(self): |
| self.c2.collect(self.collector) |
| self.t1.bind(self.c1) |
| self.t2.bind(self.c2) |
| _testSaslMech(self, 'IMPOSSIBLE', authenticated=False) |
| self.expect(Event.CONNECTION_INIT, Event.CONNECTION_BOUND, |
| Event.CONNECTION_LOCAL_OPEN, Event.TRANSPORT, |
| Event.TRANSPORT_TAIL_CLOSED, |
| Event.TRANSPORT_ERROR, Event.TRANSPORT_HEAD_CLOSED, Event.TRANSPORT_CLOSED) |
| |
| def testDisallowedPlainClient(self): |
| self.c1.collect(self.collector) |
| self.t1.bind(self.c1) |
| self.t2.bind(self.c2) |
| _testSaslMech(self, 'PLAIN', authenticated=False) |
| self.expect(Event.CONNECTION_INIT, Event.CONNECTION_BOUND, |
| Event.CONNECTION_LOCAL_OPEN, Event.TRANSPORT, |
| Event.TRANSPORT_ERROR, |
| Event.TRANSPORT_TAIL_CLOSED, Event.TRANSPORT_HEAD_CLOSED, Event.TRANSPORT_CLOSED) |
| |
| def testDisallowedPlainServer(self): |
| self.c2.collect(self.collector) |
| self.t1.bind(self.c1) |
| self.t2.bind(self.c2) |
| _testSaslMech(self, 'PLAIN', authenticated=False) |
| self.expect(Event.CONNECTION_INIT, Event.CONNECTION_BOUND, |
| Event.CONNECTION_LOCAL_OPEN, Event.TRANSPORT, |
| Event.TRANSPORT_TAIL_CLOSED, |
| Event.TRANSPORT_ERROR, Event.TRANSPORT_HEAD_CLOSED, Event.TRANSPORT_CLOSED) |
