Merge "Add SASL configuration options for AMQP 1.0 driver."

3 files changed, 99 insertions, 12 deletions

diff --git a/oslo_messaging/_drivers/protocols/amqp/controller.py b/oslo_messaging/_drivers/protocols/amqp/controller.py
index 4f545e7..a997a3e 100644
--- a/oslo_messaging/_drivers/protocols/amqp/controller.py
+++ b/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):
-        self._entries = entries[:] if entries else []
+    def __init__(self, entries=None, default_username=None,
+                 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]
-        else:
-            return transport.TransportHost(hostname="localhost", port=5672)
+        return self._entries[self._current]
 
     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)
diff --git a/oslo_messaging/_drivers/protocols/amqp/opts.py b/oslo_messaging/_drivers/protocols/amqp/opts.py
index 01e9fa4..cba1fd3 100644
--- a/oslo_messaging/_drivers/protocols/amqp/opts.py
+++ b/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')
 ]
diff --git a/oslo_messaging/tests/test_amqp_driver.py b/oslo_messaging/tests/test_amqp_driver.py
index 18aa287..21cf7b2 100644
--- a/oslo_messaging/tests/test_amqp_driver.py
+++ b/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):