From 5a43d4548a8cab82222d8d4d0fddc246a1f1fa32 Mon Sep 17 00:00:00 2001
From: Michal Arbet <michal.arbet@ultimum.io>
Date: Fri, 15 Nov 2019 11:30:50 +0100
Subject: [PATCH] Add support for kafka SSL autentication

Change-Id: Idef066a2e3b4923789a6b081d5442e931aba4507
---
 oslo_messaging/_drivers/impl_kafka.py              | 11 ++++++++++-
 .../_drivers/kafka_driver/kafka_options.py         | 14 +++++++++++++-
 oslo_messaging/tests/drivers/test_impl_kafka.py    |  8 +++++++-
 releasenotes/notes/add-ssl-support-for-kafka.yaml  |  9 +++++++++
 4 files changed, 39 insertions(+), 3 deletions(-)
 create mode 100644 releasenotes/notes/add-ssl-support-for-kafka.yaml

diff --git a/oslo_messaging/_drivers/impl_kafka.py b/oslo_messaging/_drivers/impl_kafka.py
index 88fdb7eb0..6729f8701 100644
--- a/oslo_messaging/_drivers/impl_kafka.py
+++ b/oslo_messaging/_drivers/impl_kafka.py
@@ -101,6 +101,9 @@ class Connection(object):
         self.security_protocol = self.driver_conf.security_protocol
         self.sasl_mechanism = self.driver_conf.sasl_mechanism
         self.ssl_cafile = self.driver_conf.ssl_cafile
+        self.ssl_client_cert_file = self.driver_conf.ssl_client_cert_file
+        self.ssl_client_key_file = self.driver_conf.ssl_client_key_file
+        self.ssl_client_key_password = self.driver_conf.ssl_client_key_password
         self.url = url
         self.virtual_host = url.virtual_host
         self._parse_url()
@@ -238,6 +241,9 @@ class ConsumerConnection(Connection):
             'sasl.username': self.username,
             'sasl.password': self.password,
             'ssl.ca.location': self.ssl_cafile,
+            'ssl.certificate.location': self.ssl_client_cert_file,
+            'ssl.key.location': self.ssl_client_key_file,
+            'ssl.key.password': self.ssl_client_key_password,
             'enable.partition.eof': False,
             'default.topic.config': {'auto.offset.reset': 'latest'}
         }
@@ -323,7 +329,10 @@ class ProducerConnection(Connection):
                 'sasl.mechanism': self.sasl_mechanism,
                 'sasl.username': self.username,
                 'sasl.password': self.password,
-                'ssl.ca.location': self.ssl_cafile
+                'ssl.ca.location': self.ssl_cafile,
+                'ssl.certificate.location': self.ssl_client_cert_file,
+                'ssl.key.location': self.ssl_client_key_file,
+                'ssl.key.password': self.ssl_client_key_password
             }
             self.producer = confluent_kafka.Producer(conf)
 
diff --git a/oslo_messaging/_drivers/kafka_driver/kafka_options.py b/oslo_messaging/_drivers/kafka_driver/kafka_options.py
index c1b8bef71..754711e4f 100644
--- a/oslo_messaging/_drivers/kafka_driver/kafka_options.py
+++ b/oslo_messaging/_drivers/kafka_driver/kafka_options.py
@@ -73,7 +73,19 @@ KAFKA_OPTS = [
     cfg.StrOpt('ssl_cafile',
                default='',
                help='CA certificate PEM file used to verify the server'
-               ' certificate')
+               ' certificate'),
+
+    cfg.StrOpt('ssl_client_cert_file',
+               default='',
+               help='Client certificate PEM file used for authentication.'),
+
+    cfg.StrOpt('ssl_client_key_file',
+               default='',
+               help='Client key PEM file used for authentication.'),
+
+    cfg.StrOpt('ssl_client_key_password',
+               default='',
+               help='Client key password file used for authentication.')
 ]
 
 
diff --git a/oslo_messaging/tests/drivers/test_impl_kafka.py b/oslo_messaging/tests/drivers/test_impl_kafka.py
index 0af8c053e..72a86831f 100644
--- a/oslo_messaging/tests/drivers/test_impl_kafka.py
+++ b/oslo_messaging/tests/drivers/test_impl_kafka.py
@@ -113,7 +113,10 @@ class TestKafkaDriver(test_utils.BaseTestCase):
                 'sasl.mechanism': 'PLAIN',
                 'sasl.username': mock.ANY,
                 'sasl.password': mock.ANY,
-                'ssl.ca.location': ''
+                'ssl.ca.location': '',
+                'ssl.certificate.location': '',
+                'ssl.key.location': '',
+                'ssl.key.password': '',
             })
 
     def test_listen(self):
@@ -139,6 +142,9 @@ class TestKafkaDriver(test_utils.BaseTestCase):
                 'sasl.username': mock.ANY,
                 'sasl.password': mock.ANY,
                 'ssl.ca.location': '',
+                'ssl.certificate.location': '',
+                'ssl.key.location': '',
+                'ssl.key.password': '',
                 'default.topic.config': {'auto.offset.reset': 'latest'}
             })
 
diff --git a/releasenotes/notes/add-ssl-support-for-kafka.yaml b/releasenotes/notes/add-ssl-support-for-kafka.yaml
new file mode 100644
index 000000000..170c17e8e
--- /dev/null
+++ b/releasenotes/notes/add-ssl-support-for-kafka.yaml
@@ -0,0 +1,9 @@
+---
+features:
+  - |
+    | SSL support for oslo_messaging's kafka driver
+    | Next configuration params was added
+
+    * *ssl_client_cert_file* (default='')
+    * *ssl_client_key_file* (default='')
+    * *ssl_client_key_password* (default='')