Add TLS transport config

include/pulsar/ClientConfiguration.h

@@ -156,6 +156,30 @@ class PULSAR_PUBLIC ClientConfiguration {
      */
     bool isUseTls() const;
 
+    /**
+     * Set the path to the TLS private key file.
+     *
+     * @param tlsPrivateKeyFilePath
+     */
+    ClientConfiguration& setTlsPrivateKeyFilePath(const std::string& tlsKeyFilePath);
+
+    /**
+     * @return the path to the TLS private key file
+     */
+    const std::string& getTlsPrivateKeyFilePath() const;
+
+    /**
+     * Set the path to the TLS certificate file.
+     *
+     * @param tlsCertificateFilePath
+     */
+    ClientConfiguration& setTlsCertificateFilePath(const std::string& tlsCertificateFilePath);
+
+    /**
+     * @return the path to the TLS certificate file
+     */
+    const std::string& getTlsCertificateFilePath() const;
+
     /**
      * Set the path to the trusted TLS certificate file.
      *

lib/ClientConfiguration.cc

@@ -82,6 +82,22 @@ ClientConfiguration& ClientConfiguration::setValidateHostName(bool validateHostN
 
 bool ClientConfiguration::isValidateHostName() const { return impl_->validateHostName; }
 
+ClientConfiguration& ClientConfiguration::setTlsPrivateKeyFilePath(const std::string& filePath) {
+    impl_->tlsPrivateKeyFilePath = filePath;
+    return *this;
+}
+
+const std::string& ClientConfiguration::getTlsPrivateKeyFilePath() const { return impl_->tlsPrivateKeyFilePath; }
+
+ClientConfiguration& ClientConfiguration::setTlsCertificateFilePath(const std::string& filePath) {
+    impl_->tlsCertificateFilePath = filePath;
+    return *this;
+}
+
+const std::string& ClientConfiguration::getTlsCertificateFilePath() const {
+    return impl_->tlsCertificateFilePath;
+}
+
 ClientConfiguration& ClientConfiguration::setTlsTrustCertsFilePath(const std::string& filePath) {
     impl_->tlsTrustCertsFilePath = filePath;
     return *this;

lib/ClientConfigurationImpl.h

@@ -32,6 +32,8 @@ struct ClientConfigurationImpl {
     int concurrentLookupRequest{50000};
     std::string logConfFilePath;
     bool useTls{false};
+    std::string tlsPrivateKeyFilePath;
+    std::string tlsCertificateFilePath;
     std::string tlsTrustCertsFilePath;
     bool tlsAllowInsecureConnection{false};
     unsigned int statsIntervalInSeconds{600};  // 10 minutes

lib/ClientConnection.cc

@@ -228,26 +228,30 @@ ClientConnection::ClientConnection(const std::string& logicalAddress, const std:
             return;
         }
 
+        std::string tlsCertificates = clientConfiguration.getTlsCertificateFilePath();
+        std::string tlsPrivateKey = clientConfiguration.getTlsPrivateKeyFilePath();
+
         AuthenticationDataPtr authData;
         if (authentication_->getAuthData(authData) == ResultOk && authData->hasDataForTls()) {
-            std::string tlsCertificates = authData->getTlsCertificates();
-            std::string tlsPrivateKey = authData->getTlsPrivateKey();
-
-            if (file_exists(tlsCertificates)) {
-                ctx.use_certificate_file(tlsCertificates, boost::asio::ssl::context::pem);
-            } else {
+            tlsCertificates = authData->getTlsCertificates();
+            tlsPrivateKey = authData->getTlsPrivateKey();
+            if (!file_exists(tlsCertificates)) {
                 LOG_ERROR(tlsCertificates << ": No such tlsCertificates");
                 close();
                 return;
             }
-
-            if (file_exists(tlsPrivateKey)) {
-                ctx.use_private_key_file(tlsPrivateKey, boost::asio::ssl::context::pem);
-            } else {
-                LOG_ERROR(tlsPrivateKey << ": No such tlsPrivateKey");
+            if (!file_exists(tlsCertificates)) {
+                LOG_ERROR(tlsCertificates << ": No such tlsCertificates");
                 close();
                 return;
             }
+            ctx.use_private_key_file(tlsPrivateKey, boost::asio::ssl::context::pem);
+            ctx.use_certificate_file(tlsCertificates, boost::asio::ssl::context::pem);
+        } else {
+            if (file_exists(tlsPrivateKey) && file_exists(tlsCertificates)) {
+                ctx.use_private_key_file(tlsPrivateKey, boost::asio::ssl::context::pem);
+                ctx.use_certificate_file(tlsCertificates, boost::asio::ssl::context::pem);
+            }
         }
 
         tlsSocket_ = ExecutorService::createTlsSocket(socket_, ctx);

lib/HTTPLookupService.cc

@@ -55,6 +55,8 @@ HTTPLookupService::HTTPLookupService(ServiceNameResolver &serviceNameResolver,
       serviceNameResolver_(serviceNameResolver),
       authenticationPtr_(authData),
       lookupTimeoutInSeconds_(clientConfiguration.getOperationTimeoutSeconds()),
+      tlsPrivateFilePath_(clientConfiguration.getTlsPrivateKeyFilePath()),
+      tlsCertificateFilePath_(clientConfiguration.getTlsCertificateFilePath()),
       tlsTrustCertsFilePath_(clientConfiguration.getTlsTrustCertsFilePath()),
       isUseTls_(clientConfiguration.isUseTls()),
       tlsAllowInsecure_(clientConfiguration.isTlsAllowInsecureConnection()),
@@ -231,6 +233,11 @@ Result HTTPLookupService::sendHTTPRequest(std::string completeUrl, std::string &
             if (authDataContent->hasDataForTls()) {
                 curl_easy_setopt(handle, CURLOPT_SSLCERT, authDataContent->getTlsCertificates().c_str());
                 curl_easy_setopt(handle, CURLOPT_SSLKEY, authDataContent->getTlsPrivateKey().c_str());
+            } else {
+                if (!tlsPrivateFilePath_.empty() && !tlsCertificateFilePath_.empty()) {
+                    curl_easy_setopt(handle, CURLOPT_SSLCERT, tlsCertificateFilePath_.c_str());
+                    curl_easy_setopt(handle, CURLOPT_SSLKEY, tlsPrivateFilePath_.c_str());
+                }
             }
         }
 

lib/HTTPLookupService.h

@@ -46,6 +46,8 @@ class HTTPLookupService : public LookupService, public std::enable_shared_from_t
     ServiceNameResolver& serviceNameResolver_;
     AuthenticationPtr authenticationPtr_;
     int lookupTimeoutInSeconds_;
+    std::string tlsPrivateFilePath_;
+    std::string tlsCertificateFilePath_;
     std::string tlsTrustCertsFilePath_;
     bool isUseTls_;
     bool tlsAllowInsecure_;

lib/c/c_ClientConfiguration.cc

@@ -119,6 +119,24 @@ int pulsar_client_configuration_is_validate_hostname(pulsar_client_configuration
     return conf->conf.isValidateHostName();
 }
 
+void pulsar_client_configuration_set_tls_private_key_file_path(pulsar_client_configuration_t *conf,
+                                                               const char *tlsPrivateKeyFilePath) {
+    conf->conf.setTlsPrivateKeyFilePath(tlsPrivateKeyFilePath);
+}
+
+const char *pulsar_client_configuration_get_tls_private_key_file_path(pulsar_client_configuration_t *conf) {
+    return conf->conf.getTlsPrivateKeyFilePath().c_str();
+}
+
+void pulsar_client_configuration_set_tls_certificate_file_path(pulsar_client_configuration_t *conf,
+                                                               const char *tlsCertificateFilePath) {
+    conf->conf.setTlsCertificateFilePath(tlsCertificateFilePath);
+}
+
+const char *pulsar_client_configuration_get_tls_certificate_file_path(pulsar_client_configuration_t *conf) {
+    return conf->conf.getTlsCertificateFilePath().c_str();
+}
+
 void pulsar_client_configuration_set_tls_trust_certs_file_path(pulsar_client_configuration_t *conf,
                                                                const char *tlsTrustCertsFilePath) {
     conf->conf.setTlsTrustCertsFilePath(tlsTrustCertsFilePath);

tests/AuthBasicTest.cc

@@ -28,6 +28,13 @@ using namespace pulsar;
 
 static const std::string serviceUrl = "pulsar://localhost:6650";
 static const std::string serviceUrlHttp = "http://localhost:8080";
+static const std::string serviceUrlTls = "pulsar+ssl://localhost:6651";
+static const std::string serviceUrlHttps = "https://localhost:8443";
+static const std::string caPath = "../../pulsar-broker/src/test/resources/authentication/tls/cacert.pem";
+static const std::string clientCertificatePath =
+    "../../pulsar-broker/src/test/resources/authentication/tls/client-cert.pem";
+static const std::string clientPrivateKeyPath =
+    "../../pulsar-broker/src/test/resources/authentication/tls/client-key.pem";
 
 TEST(AuthPluginBasic, testBasic) {
     ClientConfiguration config = ClientConfiguration();
@@ -138,3 +145,113 @@ TEST(AuthPluginBasic, testLoadAuth) {
     ASSERT_EQ(data->hasDataForTls(), false);
     ASSERT_EQ(data->hasDataForHttp(), true);
 }
+
+TEST(AuthPluginBasic, testAuthBasicWithServiceUrlTlsWithTlsTransport) {
+    ClientConfiguration config = ClientConfiguration();
+
+    config.setTlsPrivateKeyFilePath(clientPrivateKeyPath);
+    config.setTlsCertificateFilePath(clientCertificatePath);
+    config.setTlsTrustCertsFilePath(caPath);
+
+    AuthenticationPtr auth = pulsar::AuthBasic::create("admin", "123456");
+
+    ASSERT_TRUE(auth != NULL);
+    ASSERT_EQ(auth->getAuthMethodName(), "basic");
+
+    pulsar::AuthenticationDataPtr data;
+    ASSERT_EQ(auth->getAuthData(data), pulsar::ResultOk);
+    ASSERT_EQ(data->hasDataFromCommand(), true);
+    ASSERT_EQ(data->getCommandData(), "admin:123456");
+    ASSERT_EQ(data->hasDataForTls(), false);
+    ASSERT_EQ(data->hasDataForHttp(), true);
+
+    config.setAuth(auth);
+    Client client(serviceUrlTls, config);
+
+    std::string topicName = "persistent://private/auth/test-basic";
+
+    Producer producer;
+    Result result = client.createProducer(topicName, producer);
+    ASSERT_EQ(ResultOk, result);
+    producer.close();
+}
+
+TEST(AuthPluginBasic, testAuthBasicWithServiceUrlHttpsWithTlsTransport) {
+    ClientConfiguration config = ClientConfiguration();
+
+    config.setTlsPrivateKeyFilePath(clientPrivateKeyPath);
+    config.setTlsCertificateFilePath(clientCertificatePath);
+    config.setTlsTrustCertsFilePath(caPath);
+
+    AuthenticationPtr auth = pulsar::AuthBasic::create("admin", "123456");
+
+    ASSERT_TRUE(auth != NULL);
+    ASSERT_EQ(auth->getAuthMethodName(), "basic");
+
+    pulsar::AuthenticationDataPtr data;
+    ASSERT_EQ(auth->getAuthData(data), pulsar::ResultOk);
+    ASSERT_EQ(data->hasDataFromCommand(), true);
+    ASSERT_EQ(data->getCommandData(), "admin:123456");
+    ASSERT_EQ(data->hasDataForTls(), false);
+    ASSERT_EQ(data->hasDataForHttp(), true);
+
+    config.setAuth(auth);
+    Client client(serviceUrlHttps, config);
+
+    std::string topicName = "persistent://private/auth/test-basic";
+
+    Producer producer;
+    Result result = client.createProducer(topicName, producer);
+    ASSERT_EQ(ResultOk, result);
+    producer.close();
+}
+
+TEST(AuthPluginBasic, testAuthBasicWithServiceUrlTlsNoTlsTransport) {
+    ClientConfiguration config = ClientConfiguration();
+
+    AuthenticationPtr auth = pulsar::AuthBasic::create("admin", "123456");
+
+    ASSERT_TRUE(auth != NULL);
+    ASSERT_EQ(auth->getAuthMethodName(), "basic");
+
+    pulsar::AuthenticationDataPtr data;
+    ASSERT_EQ(auth->getAuthData(data), pulsar::ResultOk);
+    ASSERT_EQ(data->hasDataFromCommand(), true);
+    ASSERT_EQ(data->getCommandData(), "admin:123456");
+    ASSERT_EQ(data->hasDataForTls(), false);
+    ASSERT_EQ(data->hasDataForHttp(), true);
+
+    config.setAuth(auth);
+    Client client(serviceUrlTls, config);
+
+    std::string topicName = "persistent://private/auth/test-basic";
+
+    Producer producer;
+    Result result = client.createProducer(topicName, producer);
+    ASSERT_EQ(ResultConnectError, result);
+}
+
+TEST(AuthPluginBasic, testAuthBasicWithServiceUrlHttpsNoTlsTransport) {
+    ClientConfiguration config = ClientConfiguration();
+
+    AuthenticationPtr auth = pulsar::AuthBasic::create("admin", "123456");
+
+    ASSERT_TRUE(auth != NULL);
+    ASSERT_EQ(auth->getAuthMethodName(), "basic");
+
+    pulsar::AuthenticationDataPtr data;
+    ASSERT_EQ(auth->getAuthData(data), pulsar::ResultOk);
+    ASSERT_EQ(data->hasDataFromCommand(), true);
+    ASSERT_EQ(data->getCommandData(), "admin:123456");
+    ASSERT_EQ(data->hasDataForTls(), false);
+    ASSERT_EQ(data->hasDataForHttp(), true);
+
+    config.setAuth(auth);
+    Client client(serviceUrlHttps, config);
+
+    std::string topicName = "persistent://private/auth/test-basic";
+
+    Producer producer;
+    Result result = client.createProducer(topicName, producer);
+    ASSERT_EQ(ResultConnectError, result);
+}