<commit_before>#
# 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 unicode_literals
from __future__ import division
from __future__ import absolute_import
from __future__ import print_function

import os, json
from subprocess import PIPE, STDOUT, Popen
from system_test import TestCase, Qdrouterd, main_module, DIR, TIMEOUT, SkipIfNeeded, Process
from system_test import unittest, QdManager
from qpid_dispatch.management.client import Node
from proton import SASL

class RouterTestPlainSaslCommon(TestCase):
    @classmethod
    def router(cls, name, connection):

        config = Qdrouterd.Config(connection)

        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, universal_newlines=True)
        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 RouterTestPlainSaslFailure(RouterTestPlainSaslCommon):
    @staticmethod
    def sasl_file(name):
        return os.path.join(DIR, 'sasl_files', 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 bad sasl_password
        as a non-existent file.

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

        if not SASL.extended():
            return

        super(RouterTestPlainSaslFailure, cls).createSaslFiles()

        cls.routers = []

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

        super(RouterTestPlainSaslFailure, 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',
                                 # Leave as saslConfigPath for testing backward compatibility
                                 'saslConfigPath': os.getcwd()}),
        ])

        super(RouterTestPlainSaslFailure, 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': '<EMAIL>',
                                    # Provide a non-existen file.
                                    'saslPassword': 'file:' + cls.sasl_file('non-existent-password-file.txt')}),
                     ('router', {'workerThreads': 1,
                                 'mode': 'interior',
                                 'id': 'QDR.Y'}),
                     ('listener', {'host': '0.0.0.0', 'role': 'normal', 'port': y_listener_port}),
        ])

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

    @SkipIfNeeded(not SASL.extended(), "Cyrus library not available. skipping test")
    def test_inter_router_sasl_fail(self):
        passed = False
        long_type = 'org.apache.qpid.dispatch.connection'
        qd_manager = QdManager(self, address=self.routers[1].addresses[0])
        connections = qd_manager.query(long_type)
        for connection in connections:
            if connection['role'] == 'inter-router':
                passed = True
                break

        # There was no inter-router connection established.
        self.assertFalse(passed)

        qd_manager = QdManager(self, address=self.routers[1].addresses[0])
        logs = qd_manager.get_log()

        sasl_failed = False
        file_open_failed = False
        for log in logs:
            if log[0] == 'SERVER' and log[1] == "info" and "amqp:unauthorized-access Authentication failed [mech=PLAIN]" in log[2]:
                sasl_failed = True
            if log[0] == "CONN_MGR" and log[1] == "error" and "Unable to open password file" in log[2] and "error: No such file or directory" in log[2]:
                file_open_failed = True

        self.assertTrue(sasl_failed)
        self.assertTrue(file_open_failed)


class RouterTestPlainSaslFailureUsingLiteral(RouterTestPlainSaslCommon):
    @staticmethod
    def sasl_file(name):
        return os.path.join(DIR, 'sasl_files', 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 bad sasl_password
        using the literal: prefix.

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

        if not SASL.extended():
            return
<commit_msg>NO-JIRA: Add test print statement in system_tests_sasl_plain to detect failure<commit_after>#
# 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 unicode_literals
from __future__ import division
from __future__ import absolute_import
from __future__ import print_function

import os, json
from subprocess import PIPE, STDOUT, Popen
from system_test import TestCase, Qdrouterd, main_module, DIR, TIMEOUT, SkipIfNeeded, Process
from system_test import unittest, QdManager
from qpid_dispatch.management.client import Node
from proton import SASL

class RouterTestPlainSaslCommon(TestCase):
    @classmethod
    def router(cls, name, connection):

        config = Qdrouterd.Config(connection)

        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, universal_newlines=True)
        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 RouterTestPlainSaslFailure(RouterTestPlainSaslCommon):
    @staticmethod
    def sasl_file(name):
        return os.path.join(DIR, 'sasl_files', 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 bad sasl_password
        as a non-existent file.

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

        if not SASL.extended():
            return

        super(RouterTestPlainSaslFailure, cls).createSaslFiles()

        cls.routers = []

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

        super(RouterTestPlainSaslFailure, 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',
                                 # Leave as saslConfigPath for testing backward compatibility
                                 'saslConfigPath': os.getcwd()}),
        ])

        super(RouterTestPlainSaslFailure, 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': '<EMAIL>',
                                    # Provide a non-existen file.
                                    'saslPassword': 'file:' + cls.sasl_file('non-existent-password-file.txt')}),
                     ('router', {'workerThreads': 1,
                                 'mode': 'interior',
                                 'id': 'QDR.Y'}),
                     ('listener', {'host': '0.0.0.0', 'role': 'normal', 'port': y_listener_port}),
        ])

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

    @SkipIfNeeded(not SASL.extended(), "Cyrus library not available. skipping test")
    def test_inter_router_sasl_fail(self):
        passed = False
        long_type = 'org.apache.qpid.dispatch.connection'
        qd_manager = QdManager(self, address=self.routers[1].addresses[0])
        connections = qd_manager.query(long_type)
        for connection in connections:
            if connection['role'] == 'inter-router':
                passed = True
                break

        # There was no inter-router connection established.
        self.assertFalse(passed)

        qd_manager = QdManager(self, address=self.routers[1].addresses[0])
        logs = qd_manager.get_log()

        sasl_failed = False
        file_open_failed = False
        for log in logs:
            print (log)
            if log[0] == 'SERVER' and log[1] == "info" and "amqp:unauthorized-access Authentication failed [mech=PLAIN]" in log[2]:
                sasl_failed = True
            if log[0] == "CONN_MGR" and log[1] == "error" and "Unable to open password file" in log[2] and "error: No such file or directory" in log[2]:
                file_open_failed = True

        self.assertTrue(sasl_failed)
        self.assertTrue(file_open_failed)


class RouterTestPlainSaslFailureUsingLiteral(RouterTestPlainSaslCommon):
    @staticmethod
    def sasl_file(name):
        return os.path.join(DIR, 'sasl_files', 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 bad sasl_password
        using the literal: prefix.

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

        if not SASL.extended():
            return

{'fraction_non_alphanumeric': 0.10371398819854218, 'fraction_numerical': 0.006386671294689344, 'mean_word_length': 2.3003436426116837, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 10, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 30, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4501365', 'n_tokens_mistral': 4325, 'n_tokens_neox': 4176, 'n_words': 1165}