Add separated TLS transport configs (#252)

Author: BewareMyPower

pulsar/__init__.py

@@ -489,7 +489,9 @@ def __init__(self, service_url,
                  tls_validate_hostname=False,
                  logger=None,
                  connection_timeout_ms=10000,
-                 listener_name=None
+                 listener_name=None,
+                 tls_private_key_file_path: Optional[str] = None,
+                 tls_certificate_file_path: Optional[str] = None,
                  ):
         """
         Create a new Pulsar client instance.
@@ -555,6 +557,10 @@ def __init__(self, service_url,
             Listener name for lookup. Clients can use listenerName to choose one of the listeners as
             the service URL to create a connection to the broker as long as the network is accessible.
             ``advertisedListeners`` must be enabled in broker side.
+        tls_private_key_file_path: str, optional
+            The path to the TLS private key file
+        tls_certificate_file_path: str, optional
+            The path to the TLS certificate file.
         """
         _check_type(str, service_url, 'service_url')
         _check_type_or_none(Authentication, authentication, 'authentication')
@@ -570,6 +576,8 @@ def __init__(self, service_url,
         _check_type(bool, tls_allow_insecure_connection, 'tls_allow_insecure_connection')
         _check_type(bool, tls_validate_hostname, 'tls_validate_hostname')
         _check_type_or_none(str, listener_name, 'listener_name')
+        _check_type_or_none(str, tls_private_key_file_path, 'tls_private_key_file_path')
+        _check_type_or_none(str, tls_certificate_file_path, 'tls_certificate_file_path')
 
         conf = _pulsar.ClientConfiguration()
         if authentication:
@@ -601,6 +609,10 @@ def __init__(self, service_url,
             conf.tls_trust_certs_file_path(certifi.where())
         conf.tls_allow_insecure_connection(tls_allow_insecure_connection)
         conf.tls_validate_hostname(tls_validate_hostname)
+        if tls_private_key_file_path is not None:
+            conf.tls_private_key_file_path(tls_private_key_file_path)
+        if tls_certificate_file_path is not None:
+            conf.tls_certificate_file_path(tls_certificate_file_path)
         self._client = _pulsar.Client(service_url, conf)
         self._consumers = []
 

src/config.cc

@@ -166,6 +166,14 @@ void export_config(py::module_& m) {
              return_value_policy::copy)
         .def("tls_trust_certs_file_path", &ClientConfiguration::setTlsTrustCertsFilePath,
              return_value_policy::reference)
+        .def("tls_private_key_file_path", &ClientConfiguration::getTlsPrivateKeyFilePath,
+             return_value_policy::copy)
+        .def("tls_private_key_file_path", &ClientConfiguration::setTlsPrivateKeyFilePath,
+             return_value_policy::reference)
+        .def("tls_certificate_file_path", &ClientConfiguration::getTlsCertificateFilePath,
+             return_value_policy::copy)
+        .def("tls_certificate_file_path", &ClientConfiguration::setTlsCertificateFilePath,
+             return_value_policy::reference)
         .def("tls_allow_insecure_connection", &ClientConfiguration::isTlsAllowInsecureConnection)
         .def("tls_allow_insecure_connection", &ClientConfiguration::setTlsAllowInsecureConnection,
              return_value_policy::reference)

tests/pulsar_test.py

@@ -22,6 +22,7 @@
 import random
 import threading
 import logging
+from typing import Optional
 from unittest import TestCase, main
 import time
 import os
@@ -1511,10 +1512,22 @@ def _check_type_error(self, fun):
         with self.assertRaises(TypeError):
             fun()
 
-    def _test_basic_auth(self, id, auth):
-        client = Client(self.adminUrl, authentication=auth)
-
-        topic = "persistent://private/auth/my-python-topic-basic-auth-" + str(id)
+    def _test_basic_auth(self, topic_id: int, auth,
+                         use_tls: bool = False,
+                         tls_private_key_file_path: Optional[str] = None,
+                         tls_certificate_file_path: Optional[str] = None) -> None:
+        if use_tls:
+            service_url = self.serviceUrlTls
+            tls_trust_certs_file_path = CERTS_DIR + 'cacert.pem'
+        else:
+            service_url = self.adminUrl
+            tls_trust_certs_file_path = None
+        client = Client(service_url, authentication=auth,
+                        tls_trust_certs_file_path=tls_trust_certs_file_path,
+                        tls_private_key_file_path=tls_private_key_file_path,
+                        tls_certificate_file_path=tls_certificate_file_path)
+
+        topic = "persistent://private/auth/my-python-topic-basic-auth-" + str(topic_id)
         consumer = client.subscribe(topic, "my-sub", consumer_type=ConsumerType.Shared)
         producer = client.create_producer(topic)
         producer.send(b"hello")
@@ -1546,6 +1559,17 @@ def test_basic_auth_method(self):
                 auth_params_string='{{"username": "{}","password": "{}", "method": "unknown"}}'.format(username, password)
             ))
 
+    def test_tls_encryption_with_other_auth(self):
+        self._test_basic_auth(6, AuthenticationBasic('admin', '123456'),
+                              use_tls=True,
+                              tls_private_key_file_path=CERTS_DIR + 'client-key.pem',
+                              tls_certificate_file_path=CERTS_DIR + 'client-cert.pem')
+        with self.assertRaises(pulsar.ConnectError):
+            self._test_basic_auth(7, AuthenticationBasic('admin', '123456'),
+                                  use_tls=True,
+                                  tls_private_key_file_path=CERTS_DIR + 'client-cert.pem',
+                                  tls_certificate_file_path=CERTS_DIR + 'client-key.pem')
+
     def test_invalid_basic_auth(self):
         username = "invalid"
         password = "123456"