Merge "Add "ssl" option for amqp driver"

oslo_messaging/_drivers/amqp1_driver/controller.py

@@ -829,6 +829,7 @@ class Controller(pyngus.ConnectionEventHandler):
         self._container_name = config.oslo_messaging_amqp.container_name
         self.idle_timeout = config.oslo_messaging_amqp.idle_timeout
         self.trace_protocol = config.oslo_messaging_amqp.trace
+        self.ssl = config.oslo_messaging_amqp.ssl
         self.ssl_ca_file = config.oslo_messaging_amqp.ssl_ca_file
         self.ssl_cert_file = config.oslo_messaging_amqp.ssl_cert_file
         self.ssl_key_file = config.oslo_messaging_amqp.ssl_key_file
@@ -975,6 +976,8 @@ class Controller(pyngus.ConnectionEventHandler):
             conn_props["idle-time-out"] = float(self.idle_timeout)
         if self.trace_protocol:
             conn_props["x-trace-protocol"] = self.trace_protocol
+        if self.ssl:
+            conn_props["x-ssl"] = self.ssl
         if self.ssl_ca_file:
             conn_props["x-ssl-ca-file"] = self.ssl_ca_file
         if self.ssl_cert_file:

oslo_messaging/_drivers/amqp1_driver/opts.py

@@ -31,6 +31,12 @@ amqp1_opts = [
                 deprecated_group='amqp1',
                 help='Debug: dump AMQP frames to stdout'),
 
+    cfg.BoolOpt('ssl',
+                default=False,
+                help=("Attempt to connect via SSL. If no other ssl-related "
+                      "parameters are given, it will use the system's "
+                      "CA-bundle to verify the server's certificate.")),
+
     cfg.StrOpt('ssl_ca_file',
                default='',
                deprecated_group='amqp1',

oslo_messaging/tests/drivers/test_amqp_driver.py

@@ -17,6 +17,7 @@ import os
 import select
 import shlex
 import shutil
+from six.moves import mock
 import socket
 import subprocess
 import sys
@@ -1532,6 +1533,32 @@ class TestSSL(test_utils.BaseTestCase):
         self.assertFalse(listener.isAlive())
         driver.cleanup()
 
+    @mock.patch('ssl.get_default_verify_paths')
+    def test_server_ok_with_ssl_set_in_transport_url(self, mock_verify_paths):
+        # test client authenticates server
+        self._broker = FakeBroker(self.conf.oslo_messaging_amqp,
+                                  sock_addr=self._ssl_config['s_name'],
+                                  ssl_config=self._ssl_config)
+        url = oslo_messaging.TransportURL.parse(
+            self.conf, "amqp://%s:%d?ssl=1" % (self._broker.host,
+                                               self._broker.port))
+        self._broker.start()
+        mock_verify_paths.return_value = mock.Mock(
+            cafile=self._ssl_config['ca_cert'])
+        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)
+
+        driver.send(target,
+                    {"context": "whatever"},
+                    {"method": "echo", "a": "b"},
+                    wait_for_reply=True,
+                    timeout=30)
+        listener.join(timeout=30)
+        self.assertFalse(listener.isAlive())
+        driver.cleanup()
+
     def test_bad_server_fail(self):
         # test client does not connect to invalid server
         self._ssl_config['s_cert'] = self._ssl_config['bad_cert']