[AMQP 1.0] Add default SASL realm setting

Change-Id: I2ac680bb9b594f0493dfb6d52c3f40bf5282dec7
Closes-Bug: 1681927

oslo_messaging/_drivers/amqp1_driver/controller.py

@@ -763,13 +763,14 @@ class NotificationServer(Server):
 
 
 class Hosts(object):
-    """An order list of TransportHost addresses. Connection failover
+    """An order list of TransportHost addresses. Connection failover progresses
-    progresses from one host to the next.  username and password come from the
+    from one host to the next.  username, password , and realm come from the
-    configuration and are used only if no username/password was given in the
+    configuration and are used only if no username/password/realm is present in
-    URL.
+    the URL.
     """
     def __init__(self, entries=None, default_username=None,
-                 default_password=None):
+                 default_password=None,
+                 default_realm=None):
         if entries:
             self._entries = entries[:]
         else:
@@ -779,6 +780,8 @@ class Hosts(object):
             entry.port = entry.port or 5672
             entry.username = entry.username or default_username
             entry.password = entry.password or default_password
+            if default_realm and entry.username and '@' not in entry.username:
+                entry.username = entry.username + '@' + default_realm
         self._current = random.randint(0, len(self._entries) - 1)
 
     @property
@@ -840,7 +843,8 @@ class Controller(pyngus.ConnectionEventHandler):
         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)
+                           config.oslo_messaging_amqp.password,
+                           config.oslo_messaging_amqp.sasl_default_realm)
         self.conn_retry_interval = \
             config.oslo_messaging_amqp.connection_retry_interval
         self.conn_retry_backoff = \

oslo_messaging/_drivers/amqp1_driver/opts.py

@@ -83,6 +83,10 @@ amqp1_opts = [
                deprecated_group='amqp1',
                help='Name of configuration file (without .conf suffix)'),
 
+    cfg.StrOpt('sasl_default_realm',
+               default='',
+               help='SASL realm to use if no realm present in username'),
+
     cfg.StrOpt('username',
                default='',
                deprecated_group='amqp1',

oslo_messaging/tests/drivers/test_amqp_driver.py

@@ -711,7 +711,7 @@ class TestCyrusAuthentication(test_utils.BaseTestCase):
         # the temp dir after the first test is run
         os.makedirs(cls._conf_dir)
         db = os.path.join(cls._conf_dir, 'openstack.sasldb')
-        _t = "echo secret | saslpasswd2 -c -p -f ${db} joe"
+        _t = "echo secret | saslpasswd2 -c -p -f ${db} -u myrealm joe"
         cmd = Template(_t).substitute(db=db)
         try:
             subprocess.check_call(args=cmd, shell=True)
@@ -744,7 +744,7 @@ mech_list: ${mechs}
         _dir = TestCyrusAuthentication._conf_dir
         self._broker = FakeBroker(self.conf.oslo_messaging_amqp,
                                   sasl_mechanisms=_mechs,
-                                  user_credentials=["\0joe\0secret"],
+                                  user_credentials=["\0joe@myrealm\0secret"],
                                   sasl_config_dir=_dir,
                                   sasl_config_name="openstack")
         self._broker.start()
@@ -757,45 +757,44 @@ mech_list: ${mechs}
             self._broker = None
         super(TestCyrusAuthentication, self).tearDown()
 
-    def test_authentication_ok(self):
+    def _authentication_test(self, addr, retry=None):
-        """Verify that username and password given in TransportHost are
-        accepted by the broker.
-        """
-        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")
         listener = _ListenerThread(
             driver.listen(target, None, None)._poll_style_listener, 1)
-        rc = driver.send(target, {"context": True},
+        try:
-                         {"method": "echo"}, wait_for_reply=True)
+            rc = driver.send(target, {"context": True},
-        self.assertIsNotNone(rc)
+                             {"method": "echo"}, wait_for_reply=True,
-        listener.join(timeout=30)
+                             retry=retry)
-        self.assertFalse(listener.isAlive())
+            self.assertIsNotNone(rc)
-        driver.cleanup()
+            listener.join(timeout=30)
+            self.assertFalse(listener.isAlive())
+        finally:
+            driver.cleanup()
+
+    def test_authentication_ok(self):
+        """Verify that username and password given in TransportHost are
+        accepted by the broker.
+        """
+        addr = "amqp://joe@myrealm:secret@%s:%d" % (self._broker.host,
+                                                    self._broker.port)
+        self._authentication_test(addr)
 
     def test_authentication_failure(self):
         """Verify that a bad password given in TransportHost is
         rejected by the broker.
         """
-        addr = "amqp://joe:badpass@%s:%d" % (self._broker.host,
+        addr = "amqp://joe@myrealm:badpass@%s:%d" % (self._broker.host,
-                                             self._broker.port)
+                                                     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, None, None)._poll_style_listener, 1)
         try:
-            driver.send(target, {"context": True}, {"method": "echo"},
+            self._authentication_test(addr, retry=2)
-                        wait_for_reply=True, retry=2)
         except oslo_messaging.MessageDeliveryFailure as e:
             # verify the exception indicates the failure was an authentication
             # error
             self.assertTrue('amqp:unauthorized-access' in str(e))
         else:
             self.assertIsNone("Expected authentication failure")
-        driver.cleanup()
 
     def test_authentication_bad_mechs(self):
         """Verify that the connection fails if the client's SASL mechanisms do
@@ -803,40 +802,41 @@ mech_list: ${mechs}
         """
         self.config(sasl_mechanisms="EXTERNAL ANONYMOUS",
                     group="oslo_messaging_amqp")
-        addr = "amqp://joe:secret@%s:%d" % (self._broker.host,
+        addr = "amqp://joe@myrealm:secret@%s:%d" % (self._broker.host,
-                                            self._broker.port)
+                                                    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, None, None)._poll_style_listener, 1)
         self.assertRaises(oslo_messaging.MessageDeliveryFailure,
-                          driver.send,
+                          self._authentication_test,
-                          target, {"context": True},
+                          addr,
-                          {"method": "echo"},
-                          wait_for_reply=True,
                           retry=0)
-        driver.cleanup()
 
     def test_authentication_default_username(self):
         """Verify that a configured username/password is used if none appears
         in the URL.
+        Deprecated: username password deprecated in favor of transport_url
         """
         addr = "amqp://%s:%d" % (self._broker.host, self._broker.port)
-        self.config(username="joe",
+        self.config(username="joe@myrealm",
                     password="secret",
                     group="oslo_messaging_amqp")
-        url = oslo_messaging.TransportURL.parse(self.conf, addr)
+        self._authentication_test(addr)
-        driver = amqp_driver.ProtonDriver(self.conf, url)
+
-        target = oslo_messaging.Target(topic="test-topic")
+    def test_authentication_default_realm(self):
-        listener = _ListenerThread(
+        """Verify that default realm is used if none present in username"""
-            driver.listen(target, None, None)._poll_style_listener, 1)
+        addr = "amqp://joe:secret@%s:%d" % (self._broker.host,
-        rc = driver.send(target, {"context": True},
+                                            self._broker.port)
-                         {"method": "echo"}, wait_for_reply=True)
+        self.config(sasl_default_realm="myrealm",
-        self.assertIsNotNone(rc)
+                    group="oslo_messaging_amqp")
-        listener.join(timeout=30)
+        self._authentication_test(addr)
-        self.assertFalse(listener.isAlive())
+
-        driver.cleanup()
+    def test_authentication_ignore_default_realm(self):
+        """Verify that default realm is not used if realm present in
+        username
+        """
+        addr = "amqp://joe@myrealm:secret@%s:%d" % (self._broker.host,
+                                                    self._broker.port)
+        self.config(sasl_default_realm="bad-realm",
+                    group="oslo_messaging_amqp")
+        self._authentication_test(addr)
 
 
 @testtools.skipUnless(pyngus, "proton modules not present")