~launchpad-pqm/launchpad/devel

# Copyright 2009-2010 Canonical Ltd.  This software is licensed under the

# GNU Affero General Public License version 3 (see the file LICENSE).

import os

from testtools.deferredruntest import (

    assert_fails_with,

    flush_logged_errors,

    AsynchronousDeferredRunTest,

    )

from twisted.conch.checkers import SSHPublicKeyDatabase

from twisted.conch.error import ConchError

from twisted.conch.ssh import userauth

from twisted.conch.ssh.common import (

    getNS,

    NS,

    )

from twisted.conch.ssh.keys import (

    BadKeyError,

    Key,

    )

from twisted.conch.ssh.transport import (

    SSHCiphers,

    SSHServerTransport,

    )

from twisted.cred.error import UnauthorizedLogin

from twisted.cred.portal import (

    IRealm,

    Portal,

    )

from twisted.internet import defer

from twisted.python import failure

from twisted.python.util import sibpath

from zope.interface import implements

from canonical.launchpad.xmlrpc import faults

from lp.services.sshserver import auth

from lp.services.twistedsupport import suppress_stderr

from lp.testing import TestCase

class MockRealm:

    """A mock realm for testing userauth.SSHUserAuthServer.

    This realm is not actually used in the course of testing, so calls to

    requestAvatar will raise an exception.

    """

    implements(IRealm)

    def requestAvatar(self, avatar_id, mind, *interfaces):

        user_dict = {

            'id': avatar_id, 'name': avatar_id, 'teams': [],

            'initialBranches': []}

        return (

            interfaces[0], auth.LaunchpadAvatar(user_dict), lambda: None)

class MockSSHTransport(SSHServerTransport):

    """A mock SSH transport for testing userauth.SSHUserAuthServer.

    SSHUserAuthServer expects an SSH transport which has a factory attribute

    which in turn has a portal attribute. Because the portal is important for

    testing authentication, we need to be able to provide an interesting portal

    object to the SSHUserAuthServer.

    In addition, we want to be able to capture any packets sent over the

    transport.

    """

    class Factory:

        def getService(self, transport, nextService):

            return lambda: None

    def __init__(self, portal):

        # In Twisted 8.0.1, Conch's transport starts referring to

        # currentEncryptions where it didn't before. Provide a dummy value for

        # it.

        self.currentEncryptions = SSHCiphers('none', 'none', 'none', 'none')

        self.packets = []

        self.factory = self.Factory()

        self.factory.portal = portal

    def sendPacket(self, messageType, payload):

        self.packets.append((messageType, payload))

    def setService(self, service):

        pass

class UserAuthServerMixin(object):

    def setUp(self):

        self.portal = Portal(MockRealm())

        self.transport = MockSSHTransport(self.portal)

        self.user_auth = auth.SSHUserAuthServer(self.transport)

    def _getMessageName(self, message_type):

        """Get the name of the message for the given message type constant."""

        return userauth.SSHUserAuthServer.protocolMessages[message_type]

    def assertMessageOrder(self, message_types):

        """Assert that SSH messages were sent in the given order."""

        messages = userauth.SSHUserAuthServer.protocolMessages

        self.assertEqual(

            [messages[msg_type] for msg_type in message_types],

            [messages[packet_type]

             for packet_type, contents in self.transport.packets])

    def assertBannerSent(self, banner_message, expected_language='en'):

        """Assert that 'banner_message' was sent as an SSH banner."""

        # Check that we received a BANNER, then a FAILURE.

        for packet_type, packet_content in self.transport.packets:

            if packet_type == userauth.MSG_USERAUTH_BANNER:

                bytes, language, empty = getNS(packet_content, 2)

                self.assertEqual(banner_message, bytes.decode('UTF8'))

                self.assertEqual(expected_language, language)

                self.assertEqual('', empty)

                break

        else:

            self.fail("No banner logged.")

class TestUserAuthServer(TestCase, UserAuthServerMixin):

    def setUp(self):

        TestCase.setUp(self)

        UserAuthServerMixin.setUp(self)

    def test_sendBanner(self):

        # sendBanner should send an SSH 'packet' with type MSG_USERAUTH_BANNER

        # and two fields. The first field is the message itself, and the

        # second is the language tag.

        #

        # sendBanner automatically adds a trailing newline, because openssh

        # and Twisted don't add one when displaying the banner.

        #

        # See RFC 4252, Section 5.4.

        message = u"test message"

        self.user_auth.sendBanner(message, language='en-US')

        self.assertBannerSent(message + '\r\n', 'en-US')

        self.assertEqual(

            1, len(self.transport.packets),

            "More than just banner was sent: %r" % self.transport.packets)

    def test_sendBannerUsesCRLF(self):

        # sendBanner should make sure that any line breaks in the message are

        # sent as CR LF pairs.

        #

        # See RFC 4252, Section 5.4.

        self.user_auth.sendBanner(u"test\nmessage")

        [(messageType, payload)] = self.transport.packets

        bytes, language, empty = getNS(payload, 2)

        self.assertEqual(bytes.decode('UTF8'), u"test\r\nmessage\r\n")

    def test_requestRaisesConchError(self):

        # ssh_USERAUTH_REQUEST should raise a ConchError if tryAuth returns

        # None. Added to catch a bug noticed by pyflakes.

        # Whitebox test.

        def mock_try_auth(kind, user, data):

            return None

        def mock_eb_bad_auth(reason):

            reason.trap(ConchError)

        tryAuth = self.user_auth.tryAuth

        self.user_auth.tryAuth = mock_try_auth

        _ebBadAuth, self.user_auth._ebBadAuth = (self.user_auth._ebBadAuth,

                                                 mock_eb_bad_auth)

        self.user_auth.serviceStarted()

        try:

            packet = NS('jml') + NS('foo') + NS('public_key') + NS('data')

            self.user_auth.ssh_USERAUTH_REQUEST(packet)

        finally:

            self.user_auth.serviceStopped()

            self.user_auth.tryAuth = tryAuth

            self.user_auth._ebBadAuth = _ebBadAuth

class MockChecker(SSHPublicKeyDatabase):

    """A very simple public key checker which rejects all offered credentials.

    Used by TestAuthenticationBannerDisplay to test that errors raised by

    checkers are sent to SSH clients.

    """

    error_message = u'error message'

    def requestAvatarId(self, credentials):

        if credentials.username == 'success':

            return credentials.username

        else:

            return failure.Failure(

                auth.UserDisplayedUnauthorizedLogin(self.error_message))

class TestAuthenticationBannerDisplay(UserAuthServerMixin, TestCase):

    """Check that auth error information is passed through to the client.

    Normally, SSH servers provide minimal information on failed authentication.

    With Launchpad, much more user information is public, so it is helpful and

    not insecure to tell users why they failed to authenticate.

    SSH doesn't provide a standard way of doing this, but the

    MSG_USERAUTH_BANNER message is allowed and seems appropriate. See RFC 4252,

    Section 5.4 for more information.

    """

    run_tests_with = AsynchronousDeferredRunTest

    def setUp(self):

        UserAuthServerMixin.setUp(self)

        TestCase.setUp(self)

        self.portal.registerChecker(MockChecker())

        self.user_auth.serviceStarted()

        self.key_data = self._makeKey()

    def tearDown(self):

        self.user_auth.serviceStopped()

        TestCase.tearDown(self)

    def _makeKey(self):

        keydir = sibpath(__file__, 'keys')

        public_key = Key.fromString(

            open(os.path.join(keydir, 'ssh_host_key_rsa.pub'), 'rb').read())

        if isinstance(public_key, str):

            return chr(0) + NS('rsa') + NS(public_key)

        else:

            return chr(0) + NS('rsa') + NS(public_key.blob())

    def requestFailedAuthentication(self):

        return self.user_auth.ssh_USERAUTH_REQUEST(

            NS('failure') + NS('') + NS('publickey') + self.key_data)

    def requestSuccessfulAuthentication(self):

        return self.user_auth.ssh_USERAUTH_REQUEST(

            NS('success') + NS('') + NS('publickey') + self.key_data)

    def requestUnsupportedAuthentication(self):

        # Note that it doesn't matter how the checker responds -- the server

        # doesn't get that far.

        return self.user_auth.ssh_USERAUTH_REQUEST(

            NS('success') + NS('') + NS('none') + NS(''))

    def test_bannerNotSentOnSuccess(self):

        # No banner is printed when the user authenticates successfully.

        self.user_auth._banner = None

        d = self.requestSuccessfulAuthentication()

        def check(ignored):

            # Check that no banner was sent to the user.

            self.assertMessageOrder([userauth.MSG_USERAUTH_SUCCESS])

        return d.addCallback(check)

    def test_defaultBannerSentOnSuccess(self):

        # If a banner was passed to the user auth agent then we send it to the

        # user when they log in.

        self.user_auth._banner = "Boogedy boo"

        d = self.requestSuccessfulAuthentication()

        def check(ignored):

            self.assertMessageOrder(

                [userauth.MSG_USERAUTH_BANNER, userauth.MSG_USERAUTH_SUCCESS])

            self.assertBannerSent(self.user_auth._banner + '\r\n')

        return d.addCallback(check)

    def test_defaultBannerSentOnlyOnce(self):

        # We don't send the banner on each authentication attempt, just on the

        # first one. It is usual for there to be many authentication attempts

        # per SSH session.

        self.user_auth._banner = "Boogedy boo"

        d = self.requestUnsupportedAuthentication()

        d.addCallback(lambda ignored: self.requestSuccessfulAuthentication())

        def check(ignored):

            # Check that no banner was sent to the user.

            self.assertMessageOrder(

                [userauth.MSG_USERAUTH_FAILURE, userauth.MSG_USERAUTH_BANNER,

                 userauth.MSG_USERAUTH_SUCCESS])

            self.assertBannerSent(self.user_auth._banner + '\r\n')

        return d.addCallback(check)

    def test_defaultBannerNotSentOnFailure(self):

        # Failed authentication attempts do not get the default banner

        # sent.

        self.user_auth._banner = "You come away two hundred quid down"

        d = self.requestFailedAuthentication()

        def check(ignored):

            self.assertMessageOrder(

                [userauth.MSG_USERAUTH_BANNER, userauth.MSG_USERAUTH_FAILURE])

            self.assertBannerSent(MockChecker.error_message + '\r\n')

        return d.addCallback(check)

    def test_loggedToBanner(self):

        # When there's an authentication failure, we display an informative

        # error message through the SSH authentication protocol 'banner'.

        d = self.requestFailedAuthentication()

        def check(ignored):

            # Check that we received a BANNER, then a FAILURE.

            self.assertMessageOrder(

                [userauth.MSG_USERAUTH_BANNER, userauth.MSG_USERAUTH_FAILURE])

            self.assertBannerSent(MockChecker.error_message + '\r\n')

        return d.addCallback(check)

    def test_unsupportedAuthMethodNotLogged(self):

        # Trying various authentication methods is a part of the normal

        # operation of the SSH authentication protocol. We should not spam the

        # client with warnings about this, as whenever it becomes a problem,

        # we can rely on the SSH client itself to report it to the user.

        d = self.requestUnsupportedAuthentication()

        def check(ignored):

            # Check that we received only a FAILRE.

            self.assertMessageOrder([userauth.MSG_USERAUTH_FAILURE])

        return d.addCallback(check)

class TestPublicKeyFromLaunchpadChecker(TestCase):

    """Tests for the SSH server authentication mechanism.

    PublicKeyFromLaunchpadChecker accepts the SSH authentication information

    and contacts the authserver to determine if the given details are valid.

    Any authentication errors are displayed back to the user via an SSH

    MSG_USERAUTH_BANNER message.

    """

    run_tests_with = AsynchronousDeferredRunTest

    class FakeAuthenticationEndpoint:

        """A fake client for enough of `IAuthServer` for this test.

        """

        valid_user = 'valid_user'

        no_key_user = 'no_key_user'

        valid_key = 'valid_key'

        def __init__(self):

            self.calls = []

        def callRemote(self, function_name, *args, **kwargs):

            return getattr(

                self, 'xmlrpc_%s' % function_name)(*args, **kwargs)

        def xmlrpc_getUserAndSSHKeys(self, username):

            self.calls.append(username)

            if username == self.valid_user:

                return defer.succeed({

                    'name': username,

                    'keys': [('DSA', self.valid_key.encode('base64'))],

                    })

            elif username == self.no_key_user:

                return defer.succeed({

                    'name': username,

                    'keys': [],

                    })

            else:

                try:

                    raise faults.NoSuchPersonWithName(username)

                except faults.NoSuchPersonWithName:

                    return defer.fail()

    def makeCredentials(self, username, public_key, mind=None):

        if mind is None:

            mind = auth.UserDetailsMind()

        return auth.SSHPrivateKeyWithMind(

            username, 'ssh-dss', public_key, '', None, mind)

    def makeChecker(self, do_signature_checking=False):

        """Construct a PublicKeyFromLaunchpadChecker.

        :param do_signature_checking: if False, as is the default, monkeypatch

            the returned instance to not verify the signatures of the keys.

        """

        checker = auth.PublicKeyFromLaunchpadChecker(self.authserver)

        if not do_signature_checking:

            checker._cbRequestAvatarId = self._cbRequestAvatarId

        return checker

    def _cbRequestAvatarId(self, is_key_valid, credentials):

        if is_key_valid:

            return credentials.username

        return failure.Failure(UnauthorizedLogin())

    def setUp(self):

        TestCase.setUp(self)

        self.authserver = self.FakeAuthenticationEndpoint()

    def test_successful(self):

        # Attempting to log in with a username and key known to the

        # authentication end-point succeeds.

        creds = self.makeCredentials(

            self.authserver.valid_user, self.authserver.valid_key)

        checker = self.makeChecker()

        d = checker.requestAvatarId(creds)

        return d.addCallback(self.assertEqual, self.authserver.valid_user)

    @suppress_stderr

    def test_invalid_signature(self):

        # The checker requests attempts to authenticate if the requests have

        # an invalid signature.

        creds = self.makeCredentials(

            self.authserver.valid_user, self.authserver.valid_key)

        creds.signature = 'a'

        checker = self.makeChecker(True)

        d = checker.requestAvatarId(creds)

        def flush_errback(f):

            flush_logged_errors(BadKeyError)

            return f

        d.addErrback(flush_errback)

        return assert_fails_with(d, UnauthorizedLogin)

    def assertLoginError(self, checker, creds, error_message):

        """Logging in with 'creds' against 'checker' fails with 'message'.

        In particular, this tests that the login attempt fails in a way that

        is sent to the client.

        :param checker: The `ICredentialsChecker` used.

        :param creds: SSHPrivateKey credentials.

        :param error_message: String excepted to match the exception's message.

        :return: Deferred. You must return this from your test.

        """

        d = assert_fails_with(

            checker.requestAvatarId(creds),

            auth.UserDisplayedUnauthorizedLogin)

        d.addCallback(

            lambda exception: self.assertEqual(str(exception), error_message))

        return d

    def test_noSuchUser(self):

        # When someone signs in with a non-existent user, they should be told

        # that. The usual security issues don't apply here because the list of

        # Launchpad user names is public.

        checker = self.makeChecker()

        creds = self.makeCredentials(

            'no-such-user', self.authserver.valid_key)

        return self.assertLoginError(

            checker, creds, 'No such Launchpad account: no-such-user')

    def test_noKeys(self):

        # When you sign into an existing account with no SSH keys, the SSH

        # server informs you that the account has no keys.

        checker = self.makeChecker()

        creds = self.makeCredentials(

            self.authserver.no_key_user, self.authserver.valid_key)

        return self.assertLoginError(

            checker, creds,

            "Launchpad user %r doesn't have a registered SSH key"

            % self.authserver.no_key_user)

    def test_wrongKey(self):

        # When you sign into an existing account using the wrong key, you

        # are *not* informed of the wrong key. This is because SSH often

        # tries several keys as part of normal operation.

        checker = self.makeChecker()

        creds = self.makeCredentials(

            self.authserver.valid_user, 'invalid key')

        # We cannot use assertLoginError because we are checking that we fail

        # with UnauthorizedLogin and not its subclass

        # UserDisplayedUnauthorizedLogin.

        d = assert_fails_with(

            checker.requestAvatarId(creds),

            UnauthorizedLogin)

        d.addCallback(

            lambda exception:

            self.failIf(isinstance(exception,

                                   auth.UserDisplayedUnauthorizedLogin),

                        "Should not be a UserDisplayedUnauthorizedLogin"))

        return d

    def test_successful_with_second_key_calls_authserver_once(self):

        # It is normal in SSH authentication to be presented with a number of

        # keys.  When the valid key is presented after some invalid ones (a)

        # the login succeeds and (b) only one call is made to the authserver

        # to retrieve the user's details.

        checker = self.makeChecker()

        mind = auth.UserDetailsMind()

        wrong_key_creds = self.makeCredentials(

            self.authserver.valid_user, 'invalid key', mind)

        right_key_creds = self.makeCredentials(

            self.authserver.valid_user, self.authserver.valid_key, mind)

        d = checker.requestAvatarId(wrong_key_creds)

        def try_second_key(failure):

            failure.trap(UnauthorizedLogin)

            return checker.requestAvatarId(right_key_creds)

        d.addErrback(try_second_key)

        d.addCallback(self.assertEqual, self.authserver.valid_user)

        def check_one_call(r):

            self.assertEqual(

                [self.authserver.valid_user], self.authserver.calls)

            return r

        d.addCallback(check_one_call)

        return d

    def test_noSuchUser_with_two_keys_calls_authserver_once(self):

        # When more than one key is presented for a username that does not

        # exist, only one call is made to the authserver.

        checker = self.makeChecker()

        mind = auth.UserDetailsMind()

        creds_1 = self.makeCredentials(

            'invalid-user', 'invalid key 1', mind)

        creds_2 = self.makeCredentials(

            'invalid-user', 'invalid key 2', mind)

        d = checker.requestAvatarId(creds_1)

        def try_second_key(failure):

            return assert_fails_with(

                checker.requestAvatarId(creds_2),

                UnauthorizedLogin)

        d.addErrback(try_second_key)

        def check_one_call(r):

            self.assertEqual(

                ['invalid-user'], self.authserver.calls)

            return r

        d.addCallback(check_one_call)

        return d