#
# 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.
#

import unittest, os, time
from subprocess import PIPE, Popen
from system_test import TestCase, Qdrouterd, main_module, DIR, TIMEOUT

from qpid_dispatch.management.client import Node

class RouterTestPlainSaslCommon(TestCase):

    @classmethod
    def router(cls, name, connection):

        config = [
            ('fixedAddress', {'prefix': '/closest/', 'fanout': 'single', 'bias': 'closest'}),
            ('fixedAddress', {'prefix': '/spread/', 'fanout': 'single', 'bias': 'spread'}),
            ('fixedAddress', {'prefix': '/multicast/', 'fanout': 'multiple'}),
            ('fixedAddress', {'prefix': '/', 'fanout': 'multiple'}),

        ] + connection

        config = Qdrouterd.Config(config)

        cls.routers.append(cls.tester.qdrouterd(name, config, wait=False))

    @classmethod
    def createSaslFiles(cls):
        # Create a sasl database.
        p = Popen(['saslpasswd2', '-c', '-p', '-f', 'qdrouterd.sasldb', '-u', 'domain.com', 'test'],
                  stdin=PIPE, stdout=PIPE, stderr=PIPE)
        result = p.communicate('password')
        assert p.returncode == 0, \
            "saslpasswd2 exit status %s, output:\n%s" % (p.returncode, result)

        # Create a SASL configuration file.
        with open('tests-mech-PLAIN.conf', 'w') as sasl_conf:
            sasl_conf.write("""
pwcheck_method: auxprop
auxprop_plugin: sasldb
sasldb_path: qdrouterd.sasldb
mech_list: ANONYMOUS DIGEST-MD5 EXTERNAL PLAIN
# The following line stops spurious 'sql_select option missing' errors when cyrus-sql-sasl plugin is installed
sql_select: dummy select
""")

class RouterTestPlainSasl(RouterTestPlainSaslCommon):

    @classmethod
    def setUpClass(cls):
        """
        Tests the sasl_username, sasl_password property of the dispatch router.

        Creates two routers (QDR.X and QDR.Y) and sets up PLAIN authentication on QDR.X.
        QDR.Y connects to QDR.X by providing a sasl_username and a sasl_password.

        """
        super(RouterTestPlainSasl, cls).setUpClass()

        super(RouterTestPlainSasl, cls).createSaslFiles()

        cls.routers = []

        x_listener_port = cls.tester.get_port()
        y_listener_port = cls.tester.get_port()

        super(RouterTestPlainSasl, cls).router('X', [
                     ('listener', {'host': '0.0.0.0', 'role': 'inter-router', 'port': x_listener_port,
                                   'saslMechanisms':'PLAIN', 'authenticatePeer': 'yes'}),
                     # This unauthenticated listener is for qdstat to connect to it.
                     ('listener', {'host': '0.0.0.0', 'role': 'normal', 'port': cls.tester.get_port(),
                                   'authenticatePeer': 'no'}),
                     ('listener', {'host': '0.0.0.0', 'role': 'normal', 'port': cls.tester.get_port(),
                                   'saslMechanisms':'PLAIN', 'authenticatePeer': 'yes'}),
                     ('router', {'workerThreads': 1,
                                 'id': 'QDR.X',
                                 'mode': 'interior',
                                 'saslConfigName': 'tests-mech-PLAIN',
                                 'saslConfigPath': os.getcwd()}),
        ])

        super(RouterTestPlainSasl, cls).router('Y', [
                     ('connector', {'host': '0.0.0.0', 'role': 'inter-router', 'port': x_listener_port,
                                    # Provide a sasl user name and password to connect to QDR.X
                                   'saslMechanisms': 'PLAIN',
                                    'saslUsername': 'test@domain.com',
                                    'saslPassword': 'password'}),
                     ('router', {'workerThreads': 1,
                                 'mode': 'interior',
                                 'id': 'QDR.Y'}),
                     ('listener', {'host': '0.0.0.0', 'role': 'normal', 'port': y_listener_port}),
        ])

        cls.routers[1].wait_router_connected('QDR.X')

    def test_inter_router_plain_exists(self):
        """The setUpClass sets up two routers with SASL PLAIN enabled.

        This test makes executes a qdstat -c via an unauthenticated listener to
        QDR.X and makes sure that the output has an "inter-router" connection to
        QDR.Y whose authentication is PLAIN. This ensures that QDR.Y did not
        somehow use SASL ANONYMOUS to connect to QDR.X

        """
        p = self.popen(
            ['qdstat', '-b', str(self.routers[0].addresses[1]), '-c'],
            name='qdstat-'+self.id(), stdout=PIPE, expect=None)
        out = p.communicate()[0]
        assert p.returncode == 0, \
            "qdstat exit status %s, output:\n%s" % (p.returncode, out)

        self.assertIn("inter-router", out)
        self.assertIn("test@domain.com(PLAIN)", out)

    def test_qdstat_connect_sasl(self):
        """
        Make qdstat use sasl plain authentication.
        """
        p = self.popen(
            ['qdstat', '-b', str(self.routers[0].addresses[2]), '-c', '--sasl-mechanisms=PLAIN',
             '--sasl-username=test@domain.com', '--sasl-password=password'],
            name='qdstat-'+self.id(), stdout=PIPE, expect=None)

        out = p.communicate()[0]
        assert p.returncode == 0, \
            "qdstat exit status %s, output:\n%s" % (p.returncode, out)

        split_list = out.split()

        # There will be 2 connections that have authenticated using SASL PLAIN. One inter-router connection
        # and the other connection that this qdstat client is making
        self.assertEqual(2, split_list.count("test@domain.com(PLAIN)"))
        self.assertEqual(1, split_list.count("inter-router"))
        self.assertEqual(1, split_list.count("normal"))

    def test_qdstat_connect_sasl_password_file(self):
        """
        Make qdstat use sasl plain authentication with client password specified in a file.
        """
        password_file = os.getcwd() + '/sasl-client-password-file.txt'
        # Create a SASL configuration file.
        with open(password_file, 'w') as sasl_client_password_file:
            sasl_client_password_file.write("password")

        sasl_client_password_file.close()

        p = self.popen(
            ['qdstat', '-b', str(self.routers[0].addresses[2]), '-c', '--sasl-mechanisms=PLAIN',
             '--sasl-username=test@domain.com', '--sasl-password-file=' + password_file],
            name='qdstat-'+self.id(), stdout=PIPE, expect=None)

        out = p.communicate()[0]
        assert p.returncode == 0, \
            "qdstat exit status %s, output:\n%s" % (p.returncode, out)

        split_list = out.split()

        # There will be 2 connections that have authenticated using SASL PLAIN. One inter-router connection
        # and the other connection that this qdstat client is making
        self.assertEqual(2, split_list.count("test@domain.com(PLAIN)"))
        self.assertEqual(1, split_list.count("inter-router"))
        self.assertEqual(1, split_list.count("normal"))


class RouterTestPlainSaslOverSsl(RouterTestPlainSaslCommon):

    @staticmethod
    def ssl_file(name):
        return os.path.join(DIR, 'ssl_certs', name)

    @classmethod
    def setUpClass(cls):
        """
        Tests the sasl_username, sasl_password property of the dispatch router.

        Creates two routers (QDR.X and QDR.Y) and sets up PLAIN authentication on QDR.X.
        QDR.Y connects to QDR.X by providing a sasl_username and a sasl_password.
        This PLAIN authentication is done over an TLS/SSLv3 connection.

        """
        super(RouterTestPlainSaslOverSsl, cls).setUpClass()

        super(RouterTestPlainSaslOverSsl, cls).createSaslFiles()

        cls.routers = []

        x_listener_port = cls.tester.get_port()
        y_listener_port = cls.tester.get_port()

        super(RouterTestPlainSaslOverSsl, cls).router('X', [
                     ('listener', {'host': '0.0.0.0', 'role': 'inter-router', 'port': x_listener_port,
                                   'sslProfile':'server-ssl-profile',
                                   'saslMechanisms':'PLAIN', 'authenticatePeer': 'yes'}),
                     ('listener', {'host': '0.0.0.0', 'role': 'normal', 'port': cls.tester.get_port(),
                                   'authenticatePeer': 'no'}),
                     ('listener', {'host': '0.0.0.0', 'role': 'normal', 'port': cls.tester.get_port(),
                                   'sslProfile':'server-ssl-profile',
                                   'saslMechanisms':'PLAIN', 'authenticatePeer': 'yes'}),
                     ('sslProfile', {'name': 'server-ssl-profile',
                                     'certDb': cls.ssl_file('ca-certificate.pem'),
                                     'certFile': cls.ssl_file('server-certificate.pem'),
                                     'keyFile': cls.ssl_file('server-private-key.pem'),
                                     'password': 'server-password'}),
                     ('router', {'workerThreads': 1,
                                 'id': 'QDR.X',
                                 'mode': 'interior',
                                 'saslConfigName': 'tests-mech-PLAIN',
                                 'saslConfigPath': os.getcwd()}),
        ])

        super(RouterTestPlainSaslOverSsl, cls).router('Y', [
                     # This router will act like a client. First an SSL connection will be established and then
                     # we will have SASL plain authentication over SSL.
                     ('connector', {'host': '0.0.0.0', 'role': 'inter-router', 'port': x_listener_port,
                                    'sslProfile': 'client-ssl-profile',
                                    'verifyHostName': 'no',
                                    # Provide a sasl user name and password to connect to QDR.X
                                    'saslMechanisms': 'PLAIN',
                                    'saslUsername': 'test@domain.com',
                                    'saslPassword': 'password'}),
                     ('router', {'workerThreads': 1,
                                 'mode': 'interior',
                                 'id': 'QDR.Y'}),
                     ('listener', {'host': '0.0.0.0', 'role': 'normal', 'port': y_listener_port}),
                     ('sslProfile', {'name': 'client-ssl-profile',
                                     'certDb': cls.ssl_file('ca-certificate.pem'),
                                     'certFile': cls.ssl_file('client-certificate.pem'),
                                     'keyFile': cls.ssl_file('client-private-key.pem'),
                                     'password': 'client-password'}),
        ])

        cls.routers[1].wait_router_connected('QDR.X')

    def test_aaa_qdstat_connect_sasl_over_ssl(self):
        """
        Make qdstat use sasl plain authentication over ssl.
        """
        p = self.popen(
            ['qdstat', '-b', str(self.routers[0].addresses[2]), '-c',
             # The following are SASL args
             '--sasl-mechanisms=PLAIN',
             '--sasl-username=test@domain.com',
             '--sasl-password=password',
             # The following are SSL args
             '--ssl-disable-peer-name-verify',
             '--ssl-trustfile=' + self.ssl_file('ca-certificate.pem'),
             '--ssl-certificate=' + self.ssl_file('client-certificate.pem'),
             '--ssl-key=' + self.ssl_file('client-private-key.pem'),
             '--ssl-password=client-password'],
            name='qdstat-'+self.id(), stdout=PIPE, expect=None)

        out = p.communicate()[0]
        assert p.returncode == 0, \
            "qdstat exit status %s, output:\n%s" % (p.returncode, out)

        split_list = out.split()

        # There will be 2 connections that have authenticated using SASL PLAIN. One inter-router connection
        # and the other connection that this qdstat client is making
        self.assertEqual(2, split_list.count("test@domain.com(PLAIN)"))
        self.assertEqual(1, split_list.count("inter-router"))
        self.assertEqual(1, split_list.count("normal"))

    def test_inter_router_plain_over_ssl_exists(self):
        """The setUpClass sets up two routers with SASL PLAIN enabled over TLS/SSLv3.

        This test makes executes a query for type='org.apache.qpid.dispatch.connection' over
        an unauthenticated listener to
        QDR.X and makes sure that the output has an "inter-router" connection to
        QDR.Y whose authentication is PLAIN. This ensures that QDR.Y did not
        somehow use SASL ANONYMOUS to connect to QDR.X
        Also makes sure that TLSv1/SSLv3 was used as sslProto

        """
        local_node = Node.connect(self.routers[0].addresses[1], timeout=TIMEOUT)

        # sslProto should be TLSv1/SSLv3
        self.assertEqual(u'TLSv1/SSLv3', local_node.query(type='org.apache.qpid.dispatch.connection').results[0][4])

        # role should be inter-router
        self.assertEqual(u'inter-router', local_node.query(type='org.apache.qpid.dispatch.connection').results[0][9])

        # sasl must be plain
        self.assertEqual(u'PLAIN', local_node.query(type='org.apache.qpid.dispatch.connection').results[0][12])

        # user must be test@domain.com
        self.assertEqual(u'test@domain.com', local_node.query(type='org.apache.qpid.dispatch.connection').results[0][16])


class RouterTestVerifyHostNameYes(RouterTestPlainSaslCommon):
    @staticmethod
    def ssl_file(name):
        return os.path.join(DIR, 'ssl_certs', name)

    @classmethod
    def setUpClass(cls):
        """
        Tests the verifyHostName property of the connector. The hostname on the server certificate we use is
        A1.Good.Server.domain.com and the host is 0.0.0.0 on the client router initiating the SSL connection.
        Since the host names do not match and the verifyHostName is set to true, the client router
        will NOT be able make a successful SSL connection the server router.
        """
        super(RouterTestVerifyHostNameYes, cls).setUpClass()

        super(RouterTestVerifyHostNameYes, cls).createSaslFiles()

        cls.routers = []

        x_listener_port = cls.tester.get_port()
        y_listener_port = cls.tester.get_port()

        super(RouterTestVerifyHostNameYes, cls).router('X', [
                     ('listener', {'addr': '0.0.0.0', 'role': 'inter-router', 'port': x_listener_port,
                                   'sslProfile':'server-ssl-profile',
                                   'saslMechanisms':'PLAIN', 'authenticatePeer': 'yes'}),
                     # This unauthenticated listener is for qdstat to connect to it.
                     ('listener', {'addr': '0.0.0.0', 'role': 'normal', 'port': cls.tester.get_port(),
                                   'authenticatePeer': 'no'}),
                     ('sslProfile', {'name': 'server-ssl-profile',
                                     'certDb': cls.ssl_file('ca-certificate.pem'),
                                     'certFile': cls.ssl_file('server-certificate.pem'),
                                     'keyFile': cls.ssl_file('server-private-key.pem'),
                                     'password': 'server-password'}),
                     ('router', {'workerThreads': 1,
                                 'routerId': 'QDR.X',
                                 'mode': 'interior',
                                 'saslConfigName': 'tests-mech-PLAIN',
                                 'saslConfigPath': os.getcwd()}),
        ])

        super(RouterTestVerifyHostNameYes, cls).router('Y', [
                     ('connector', {'addr': '127.0.0.1', 'role': 'inter-router', 'port': x_listener_port,
                                    'sslProfile': 'client-ssl-profile',
                                    'verifyHostName': 'yes',
                                    'saslMechanisms': 'PLAIN',
                                    'saslUsername': 'test@domain.com',
                                    'saslPassword': 'password'}),
                     ('router', {'workerThreads': 1,
                                 'mode': 'interior',
                                 'routerId': 'QDR.Y'}),
                     ('listener', {'addr': '0.0.0.0', 'role': 'normal', 'port': y_listener_port}),
                     ('sslProfile', {'name': 'client-ssl-profile',
                                     'certDb': cls.ssl_file('ca-certificate.pem'),
                                     'certFile': cls.ssl_file('client-certificate.pem'),
                                     'keyFile': cls.ssl_file('client-private-key.pem'),
                                     'password': 'client-password'}),
        ])

        cls.routers[0].wait_ports()
        cls.routers[1].wait_ports()
        try:
            # This will time out because there is no inter-router connection
            cls.routers[1].wait_connectors(timeout=3)
        except:
            pass

    def test_no_inter_router_connection(self):
        """
        Tests to make sure that there are no 'inter-router' connections.
        The connection to the other router will not happen because the connection failed
        due to setting 'verifyHostName': 'yes'
        """
        local_node = Node.connect(self.routers[1].addresses[0], timeout=TIMEOUT)

        # There should be only two connections.
        # There will be no inter-router connection
        self.assertEqual(2, len(local_node.query(type='org.apache.qpid.dispatch.connection').results))
        self.assertEqual('in', local_node.query(type='org.apache.qpid.dispatch.connection').results[0][15])
        self.assertEqual('normal', local_node.query(type='org.apache.qpid.dispatch.connection').results[0][9])
        self.assertEqual('anonymous', local_node.query(type='org.apache.qpid.dispatch.connection').results[0][16])

        self.assertEqual('normal', local_node.query(type='org.apache.qpid.dispatch.connection').results[1][9])
        self.assertEqual('anonymous', local_node.query(type='org.apache.qpid.dispatch.connection').results[1][16])

class RouterTestVerifyHostNameNo(RouterTestPlainSaslCommon):

    @staticmethod
    def ssl_file(name):
        return os.path.join(DIR, 'ssl_certs', name)

    @classmethod
    def setUpClass(cls):
        """
        Tests the verifyHostName property of the connector. The hostname on the server certificate we use is
        A1.Good.Server.domain.com and the host is 0.0.0.0 on the client router initiating the SSL connection.
        Since the host names do not match but verifyHostName is set to false, the client router
        will be successfully able to make an SSL connection the server router.
        """
        super(RouterTestVerifyHostNameNo, cls).setUpClass()

        super(RouterTestVerifyHostNameNo, cls).createSaslFiles()

        cls.routers = []

        x_listener_port = cls.tester.get_port()
        y_listener_port = cls.tester.get_port()

        super(RouterTestVerifyHostNameNo, cls).router('X', [
                     ('listener', {'addr': '0.0.0.0', 'role': 'inter-router', 'port': x_listener_port,
                                   'sslProfile':'server-ssl-profile',
                                   'saslMechanisms':'PLAIN', 'authenticatePeer': 'yes'}),
                     # This unauthenticated listener is for qdstat to connect to it.
                     ('listener', {'addr': '0.0.0.0', 'role': 'normal', 'port': cls.tester.get_port(),
                                   'authenticatePeer': 'no'}),
                     ('sslProfile', {'name': 'server-ssl-profile',
                                     'certDb': cls.ssl_file('ca-certificate.pem'),
                                     'certFile': cls.ssl_file('server-certificate.pem'),
                                     'keyFile': cls.ssl_file('server-private-key.pem'),
                                     'password': 'server-password'}),
                     ('router', {'workerThreads': 1,
                                 'routerId': 'QDR.X',
                                 'mode': 'interior',
                                 'saslConfigName': 'tests-mech-PLAIN',
                                 'saslConfigPath': os.getcwd()}),
        ])

        super(RouterTestVerifyHostNameNo, cls).router('Y', [
                     # This router will act like a client. First an SSL connection will be established and then
                     # we will have SASL plain authentication over SSL.
                     ('connector', {'addr': '127.0.0.1', 'role': 'inter-router', 'port': x_listener_port,
                                    'sslProfile': 'client-ssl-profile',
                                    # Provide a sasl user name and password to connect to QDR.X
                                    'saslMechanisms': 'PLAIN',
                                    'verifyHostName': 'no',
                                    'saslUsername': 'test@domain.com', 'saslPassword': 'password'}),
                     ('router', {'workerThreads': 1,
                                 'mode': 'interior',
                                 'routerId': 'QDR.Y'}),
                     ('listener', {'addr': '0.0.0.0', 'role': 'normal', 'port': y_listener_port}),
                     ('sslProfile', {'name': 'client-ssl-profile',
                                     'certDb': cls.ssl_file('ca-certificate.pem'),
                                     'certFile': cls.ssl_file('client-certificate.pem'),
                                     'keyFile': cls.ssl_file('client-private-key.pem'),
                                     'password': 'client-password'}),
        ])

        cls.routers[0].wait_ports()
        cls.routers[1].wait_ports()
        cls.routers[1].wait_router_connected('QDR.X')

    def test_inter_router_plain_over_ssl_exists(self):
        """
        Tests to make sure that an inter-router connection exists between the routers since verifyHostName is 'no'.
        """
        local_node = Node.connect(self.routers[1].addresses[0], timeout=TIMEOUT)

        results = local_node.query(type='org.apache.qpid.dispatch.connection').results

        self.assertEqual(4, len(results))

        search = "QDR.X"
        found = False

        for N in range(0,3):
            if results[N][0] == search:
                found = True
                break

        self.assertTrue(found, "Connection to %s not found" % search)

        # sslProto should be TLSv1/SSLv3
        self.assertEqual(u'TLSv1/SSLv3', results[N][4])

        # role should be inter-router
        self.assertEqual(u'inter-router', results[N][9])

        # sasl must be plain
        self.assertEqual(u'PLAIN', results[N][12])

        # user must be test@domain.com
        self.assertEqual(u'test@domain.com', results[N][16])

if __name__ == '__main__':
    unittest.main(main_module())