Add SASL configuration options for AMQP 1.0 driver.

Proton's SASL implementation provides several different configuration
options, such as the permitted mechanism set and the configuration
file.  This patch adds support for configuring these items.

Change-Id: Icf59643a6d557e3d91947664afedd420c9522fd8
Closes-Bug: #1495969

oslo_messaging/_drivers/protocols/amqp/controller.py

@@ -214,23 +214,26 @@ class Server(pyngus.ReceiverEventHandler):
 
 class Hosts(object):
     """An order list of TransportHost addresses. Connection failover
-    progresses from one host to the next.
+    progresses from one host to the next.  username and password come from the
+    configuration and are used only if no username/password was given in the
+    URL.
     """
-    def __init__(self, entries=None):
+    def __init__(self, entries=None, default_username=None,
-        self._entries = entries[:] if entries else []
+                 default_password=None):
+        if entries:
+            self._entries = entries[:]
+        else:
+            self._entries = [transport.TransportHost(hostname="localhost",
+                                                     port=5672)]
         for entry in self._entries:
             entry.port = entry.port or 5672
+            entry.username = entry.username or default_username
+            entry.password = entry.password or default_password
         self._current = 0
 
-    def add(self, transport_host):
-        self._entries.append(transport_host)
-
     @property
     def current(self):
-        if len(self._entries):
+        return self._entries[self._current]
-            return self._entries[self._current]
-        else:
-            return transport.TransportHost(hostname="localhost", port=5672)
 
     def next(self):
         if len(self._entries) > 1:
@@ -253,6 +256,7 @@ class Controller(pyngus.ConnectionEventHandler):
     """
     def __init__(self, hosts, default_exchange, config):
         self.processor = None
+        self._socket_connection = None
         # queue of Task() objects to execute on the eventloop once the
         # connection is ready:
         self._tasks = moves.queue.Queue(maxsize=500)
@@ -264,7 +268,6 @@ class Controller(pyngus.ConnectionEventHandler):
         self._senders = {}
         # Servers (set of receiving links), indexed by target:
         self._servers = {}
-        self.hosts = Hosts(hosts)
 
         opt_group = cfg.OptGroup(name='oslo_messaging_amqp',
                                  title='AMQP 1.0 driver options')
@@ -285,6 +288,11 @@ class Controller(pyngus.ConnectionEventHandler):
         self.ssl_key_password = config.oslo_messaging_amqp.ssl_key_password
         self.ssl_allow_insecure = \
             config.oslo_messaging_amqp.allow_insecure_clients
+        self.sasl_mechanisms = config.oslo_messaging_amqp.sasl_mechanisms
+        self.sasl_config_dir = config.oslo_messaging_amqp.sasl_config_dir
+        self.sasl_config_name = config.oslo_messaging_amqp.sasl_config_name
+        self.hosts = Hosts(hosts, config.oslo_messaging_amqp.username,
+                           config.oslo_messaging_amqp.password)
         self.separator = "."
         self.fanout_qualifier = "all"
         self.default_exchange = default_exchange
@@ -468,6 +476,14 @@ class Controller(pyngus.ConnectionEventHandler):
                                             self.ssl_key_file,
                                             self.ssl_key_password)
             conn_props["x-ssl-allow-cleartext"] = self.ssl_allow_insecure
+        # SASL configuration:
+        if self.sasl_mechanisms:
+            conn_props["x-sasl-mechs"] = self.sasl_mechanisms
+        if self.sasl_config_dir:
+            conn_props["x-sasl-config-dir"] = self.sasl_config_dir
+        if self.sasl_config_name:
+            conn_props["x-sasl-config-name"] = self.sasl_config_name
+
         self._socket_connection = self.processor.connect(host,
                                                          handler=self,
                                                          properties=conn_props)

oslo_messaging/_drivers/protocols/amqp/opts.py

@@ -69,5 +69,30 @@ amqp1_opts = [
     cfg.BoolOpt('allow_insecure_clients',
                 default=False,
                 deprecated_group='amqp1',
-                help='Accept clients using either SSL or plain TCP')
+                help='Accept clients using either SSL or plain TCP'),
+
+    cfg.StrOpt('sasl_mechanisms',
+               default='',
+               deprecated_group='amqp1',
+               help='Space separated list of acceptable SASL mechanisms'),
+
+    cfg.StrOpt('sasl_config_dir',
+               default='',
+               deprecated_group='amqp1',
+               help='Path to directory that contains the SASL configuration'),
+
+    cfg.StrOpt('sasl_config_name',
+               default='',
+               deprecated_group='amqp1',
+               help='Name of configuration file (without .conf suffix)'),
+
+    cfg.StrOpt('username',
+               default='',
+               deprecated_group='amqp1',
+               help='User name for message broker authentication'),
+
+    cfg.StrOpt('password',
+               default='',
+               deprecated_group='amqp1',
+               help='Password for message broker authentication')
 ]

oslo_messaging/tests/test_amqp_driver.py

@@ -379,6 +379,8 @@ class TestCyrusAuthentication(test_utils.BaseTestCase):
 
         # configure the SASL broker:
         conf = os.path.join(self._conf_dir, 'openstack.conf')
+        # Note: don't add ANONYMOUS or EXTERNAL without updating the
+        # test_authentication_bad_mechs test below
         mechs = "DIGEST-MD5 SCRAM-SHA-1 CRAM-MD5 PLAIN"
         t = Template("""sasldb_path: ${db}
 mech_list: ${mechs}
@@ -437,6 +439,50 @@ mech_list: ${mechs}
                           timeout=2.0)
         driver.cleanup()
 
+    def test_authentication_bad_mechs(self):
+        """Verify that the connection fails if the client's SASL mechanisms do
+        not match the broker's.
+        """
+        self.config(sasl_mechanisms="EXTERNAL ANONYMOUS",
+                    group="oslo_messaging_amqp")
+        addr = "amqp://joe:secret@%s:%d" % (self._broker.host,
+                                            self._broker.port)
+        url = oslo_messaging.TransportURL.parse(self.conf, addr)
+        driver = amqp_driver.ProtonDriver(self.conf, url)
+        target = oslo_messaging.Target(topic="test-topic")
+        _ListenerThread(driver.listen(target), 1)
+        self.assertRaises(oslo_messaging.MessagingTimeout,
+                          driver.send,
+                          target, {"context": True},
+                          {"method": "echo"},
+                          wait_for_reply=True,
+                          timeout=2.0)
+        driver.cleanup()
+        self.config(sasl_mechanisms=None,
+                    group="oslo_messaging_amqp")
+
+    def test_authentication_default_username(self):
+        """Verify that a configured username/password is used if none appears
+        in the URL.
+        """
+        addr = "amqp://%s:%d" % (self._broker.host, self._broker.port)
+        self.config(username="joe",
+                    password="secret",
+                    group="oslo_messaging_amqp")
+        url = oslo_messaging.TransportURL.parse(self.conf, addr)
+        driver = amqp_driver.ProtonDriver(self.conf, url)
+        target = oslo_messaging.Target(topic="test-topic")
+        listener = _ListenerThread(driver.listen(target), 1)
+        rc = driver.send(target, {"context": True},
+                         {"method": "echo"}, wait_for_reply=True)
+        self.assertIsNotNone(rc)
+        listener.join(timeout=30)
+        self.assertFalse(listener.isAlive())
+        driver.cleanup()
+        self.config(username=None,
+                    password=None,
+                    group="oslo_messaging_amqp")
+
 
 @testtools.skipUnless(pyngus, "proton modules not present")
 class TestFailover(test_utils.BaseTestCase):