Add Tls with keystore type config support (#6853)

Fixes #6640

### Motivation

Add Tls with keystore type config.

### Modifications

Add Tls with keystore type config.

### Verifying this change

- Unit test passed

conf/broker.conf

@@ -406,6 +406,62 @@ tlsCiphers=
 # authentication.
 tlsRequireTrustedClientCertOnConnect=false
 
+### --- KeyStore TLS config variables --- ###
+# Enable TLS with KeyStore type configuration in broker.
+tlsEnabledWithKeyStore=false
+
+# TLS Provider for KeyStore type
+tlsProvider=
+
+# TLS KeyStore type configuration in broker: JKS, PKCS12
+tlsKeyStoreType=JKS
+
+# TLS KeyStore path in broker
+tlsKeyStore=
+
+# TLS KeyStore password for broker
+tlsKeyStorePassword=
+
+# TLS TrustStore type configuration in broker: JKS, PKCS12
+tlsTrustStoreType=JKS
+
+# TLS TrustStore path in broker
+tlsTrustStore=
+
+# TLS TrustStore password in broker
+tlsTrustStorePassword=
+
+# Whether internal client use KeyStore type to authenticate with Pulsar brokers
+brokerClientTlsEnabledWithKeyStore=false
+
+# The TLS Provider used by internal client to authenticate with other Pulsar brokers
+brokerClientSslProvider=
+
+# TLS TrustStore type configuration for internal client: JKS, PKCS12
+# used by the internal client to authenticate with Pulsar brokers
+brokerClientTlsTrustStoreType=JKS
+
+# TLS TrustStore path for internal client
+# used by the internal client to authenticate with Pulsar brokers
+brokerClientTlsTrustStore=
+
+# TLS TrustStore password for internal client,
+# used by the internal client to authenticate with Pulsar brokers
+brokerClientTlsTrustStorePassword=
+
+# Specify the tls cipher the internal client will use to negotiate during TLS Handshake
+# (a comma-separated list of ciphers)
+# e.g.  [TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256].
+# used by the internal client to authenticate with Pulsar brokers
+brokerClientTlsCiphers=
+
+# Specify the tls protocols the broker will use to negotiate during TLS handshake
+# (a comma-separated list of protocol names).
+# e.g.  [TLSv1.2, TLSv1.1, TLSv1]
+# used by the internal client to authenticate with Pulsar brokers
+brokerClientTlsProtocols=
+
+
 ### --- Authentication --- ###
 
 # Enable authentication

conf/client.conf

@@ -56,3 +56,14 @@ tlsEnableHostnameVerification=false
 # fails, then the cert is untrusted and the connection is dropped.
 tlsTrustCertsFilePath=
 
+# Enable TLS with KeyStore type configuration in broker.
+useKeyStoreTls=false;
+
+# TLS KeyStore type configuration: JKS, PKCS12
+tlsTrustStoreType=JKS
+
+# TLS TrustStore path
+tlsTrustStorePath=
+
+# TLS TrustStore password
+tlsTrustStorePassword=

conf/standalone.conf

@@ -225,6 +225,102 @@ maxConsumersPerSubscription=0
 # Use 0 or negative number to disable the check
 maxNumPartitionsPerPartitionedTopic=0
 
+### --- TLS --- ###
+# Deprecated - Use webServicePortTls and brokerServicePortTls instead
+tlsEnabled=false
+
+# Tls cert refresh duration in seconds (set 0 to check on every new connection)
+tlsCertRefreshCheckDurationSec=300
+
+# Path for the TLS certificate file
+tlsCertificateFilePath=
+
+# Path for the TLS private key file
+tlsKeyFilePath=
+
+# Path for the trusted TLS certificate file.
+# This cert is used to verify that any certs presented by connecting clients
+# are signed by a certificate authority. If this verification
+# fails, then the certs are untrusted and the connections are dropped.
+tlsTrustCertsFilePath=
+
+# Accept untrusted TLS certificate from client.
+# If true, a client with a cert which cannot be verified with the
+# 'tlsTrustCertsFilePath' cert will allowed to connect to the server,
+# though the cert will not be used for client authentication.
+tlsAllowInsecureConnection=false
+
+# Specify the tls protocols the broker will use to negotiate during TLS handshake
+# (a comma-separated list of protocol names).
+# Examples:- [TLSv1.2, TLSv1.1, TLSv1]
+tlsProtocols=
+
+# Specify the tls cipher the broker will use to negotiate during TLS Handshake
+# (a comma-separated list of ciphers).
+# Examples:- [TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256]
+tlsCiphers=
+
+# Trusted client certificates are required for to connect TLS
+# Reject the Connection if the Client Certificate is not trusted.
+# In effect, this requires that all connecting clients perform TLS client
+# authentication.
+tlsRequireTrustedClientCertOnConnect=false
+
+### --- KeyStore TLS config variables --- ###
+# Enable TLS with KeyStore type configuration in broker.
+tlsEnabledWithKeyStore=false
+
+# TLS Provider for KeyStore type
+tlsProvider=
+
+# TLS KeyStore type configuration in broker: JKS, PKCS12
+tlsKeyStoreType=JKS
+
+# TLS KeyStore path in broker
+tlsKeyStore=
+
+# TLS KeyStore password for broker
+tlsKeyStorePassword=
+
+# TLS TrustStore type configuration in broker: JKS, PKCS12
+tlsTrustStoreType=JKS
+
+# TLS TrustStore path in broker
+tlsTrustStore=
+
+# TLS TrustStore password for broker
+tlsTrustStorePassword=
+
+# Whether internal client use KeyStore type to authenticate with Pulsar brokers
+brokerClientTlsEnabledWithKeyStore=false
+
+# The TLS Provider used by internal client to authenticate with other Pulsar brokers
+brokerClientSslProvider=
+
+# TLS TrustStore type configuration for internal client: JKS, PKCS12
+# used by the internal client to authenticate with Pulsar brokers
+brokerClientTlsTrustStoreType=JKS
+
+# TLS TrustStore path for internal client
+# used by the internal client to authenticate with Pulsar brokers
+brokerClientTlsTrustStore=
+
+# TLS TrustStore password for internal client,
+# used by the internal client to authenticate with Pulsar brokers
+brokerClientTlsTrustStorePassword=
+
+# Specify the tls cipher the internal client will use to negotiate during TLS Handshake
+# (a comma-separated list of ciphers)
+# e.g.  [TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256].
+# used by the internal client to authenticate with Pulsar brokers
+brokerClientTlsCiphers=
+
+# Specify the tls protocols the broker will use to negotiate during TLS handshake
+# (a comma-separated list of protocol names).
+# e.g.  [TLSv1.2, TLSv1.1, TLSv1]
+# used by the internal client to authenticate with Pulsar brokers
+brokerClientTlsProtocols=
+
 ### --- Authentication --- ###
 # Role names that are treated as "proxy roles". If the broker sees a request with
 #role as proxyRoles - it will demand to see a valid original principal.

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -77,6 +77,8 @@ public class ServiceConfiguration implements PulsarConfiguration {
     @Category
     private static final String CATEGORY_TLS = "TLS";
     @Category
+    private static final String CATEGORY_KEYSTORE_TLS = "KeyStoreTLS";
+    @Category
     private static final String CATEGORY_AUTHENTICATION = "Authentication";
     @Category
     private static final String CATEGORY_AUTHORIZATION = "Authorization";
@@ -1581,6 +1583,102 @@ public class ServiceConfiguration implements PulsarConfiguration {
     private String transactionMetadataStoreProviderClassName =
             "org.apache.pulsar.transaction.coordinator.impl.InMemTransactionMetadataStoreProvider";
 
+    /**** --- KeyStore TLS config variables --- ****/
+    @FieldContext(
+            category = CATEGORY_KEYSTORE_TLS,
+            doc = "Enable TLS with KeyStore type configuration in broker"
+    )
+    private boolean tlsEnabledWithKeyStore = false;
+
+    @FieldContext(
+            category = CATEGORY_KEYSTORE_TLS,
+            doc = "TLS Provider for KeyStore type"
+    )
+    private String tlsProvider = null;
+
+    @FieldContext(
+            category = CATEGORY_KEYSTORE_TLS,
+            doc = "TLS KeyStore type configuration in broker: JKS, PKCS12"
+    )
+    private String tlsKeyStoreType = "JKS";
+
+    @FieldContext(
+            category = CATEGORY_KEYSTORE_TLS,
+            doc = "TLS KeyStore path in broker"
+    )
+    private String tlsKeyStore = null;
+
+    @FieldContext(
+            category = CATEGORY_KEYSTORE_TLS,
+            doc = "TLS KeyStore password for broker"
+    )
+    private String tlsKeyStorePassword = null;
+
+    @FieldContext(
+            category = CATEGORY_KEYSTORE_TLS,
+            doc = "TLS TrustStore type configuration in broker: JKS, PKCS12"
+    )
+    private String tlsTrustStoreType = "JKS";
+
+    @FieldContext(
+            category = CATEGORY_KEYSTORE_TLS,
+            doc = "TLS TrustStore path in broker"
+    )
+    private String tlsTrustStore = null;
+
+    @FieldContext(
+            category = CATEGORY_KEYSTORE_TLS,
+            doc = "TLS TrustStore password for broker"
+    )
+    private String tlsTrustStorePassword = null;
+
+    /**** --- KeyStore TLS config variables used for internal client/admin to auth with other broker--- ****/
+    @FieldContext(
+            category = CATEGORY_KEYSTORE_TLS,
+            doc = "Whether internal client use KeyStore type to authenticate with other Pulsar brokers"
+    )
+    private boolean brokerClientTlsEnabledWithKeyStore = false;
+    @FieldContext(
+            category = CATEGORY_KEYSTORE_TLS,
+            doc = "The TLS Provider used by internal client to authenticate with other Pulsar brokers"
+    )
+    private String brokerClientSslProvider = null;
+    // needed when client auth is required
+    @FieldContext(
+            category = CATEGORY_KEYSTORE_TLS,
+            doc = "TLS TrustStore type configuration for internal client: JKS, PKCS12 "
+                  + " used by the internal client to authenticate with Pulsar brokers"
+    )
+    private String brokerClientTlsTrustStoreType = "JKS";
+    @FieldContext(
+            category = CATEGORY_KEYSTORE_TLS,
+            doc = "TLS TrustStore path for internal client, "
+                  + " used by the internal client to authenticate with Pulsar brokers"
+    )
+    private String brokerClientTlsTrustStore = null;
+    @FieldContext(
+            category = CATEGORY_KEYSTORE_TLS,
+            doc = "TLS TrustStore password for internal client, "
+                  + " used by the internal client to authenticate with Pulsar brokers"
+    )
+    private String brokerClientTlsTrustStorePassword = null;
+    @FieldContext(
+            category = CATEGORY_KEYSTORE_TLS,
+            doc = "Specify the tls cipher the internal client will use to negotiate during TLS Handshake"
+                  + " (a comma-separated list of ciphers).\n\n"
+                  + "Examples:- [TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256].\n"
+                  + " used by the internal client to authenticate with Pulsar brokers"
+    )
+    private Set<String> brokerClientTlsCiphers = Sets.newTreeSet();
+    @FieldContext(
+            category = CATEGORY_KEYSTORE_TLS,
+            doc = "Specify the tls protocols the broker will use to negotiate during TLS handshake"
+                  + " (a comma-separated list of protocol names).\n\n"
+                  + "Examples:- [TLSv1.2, TLSv1.1, TLSv1] \n"
+                  + " used by the internal client to authenticate with Pulsar brokers"
+    )
+    private Set<String> brokerClientTlsProtocols = Sets.newTreeSet();
+
     /**
      * @deprecated See {@link #getConfigurationStoreServers}
      */

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/AuthenticationDataHttps.java

@@ -34,7 +34,6 @@ public AuthenticationDataHttps(HttpServletRequest request) {
     /*
      * TLS
      */
-
     @Override
     public boolean hasDataFromTls() {
         return (certificates != null);

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authentication/OneStageAuthenticationState.java

@@ -42,7 +42,7 @@ public OneStageAuthenticationState(AuthData authData,
                                        SSLSession sslSession,
                                        AuthenticationProvider provider) throws AuthenticationException {
         this.authenticationDataSource = new AuthenticationDataCommand(
-            new String(authData.getBytes(), UTF_8), remoteAddress, sslSession);;
+            new String(authData.getBytes(), UTF_8), remoteAddress, sslSession);
         this.authRole = provider.authenticate(authenticationDataSource);
     }
 

pulsar-broker/src/main/java/org/apache/pulsar/PulsarStandalone.java

@@ -33,6 +33,7 @@
 import org.apache.pulsar.broker.ServiceConfiguration;
 import org.apache.pulsar.broker.ServiceConfigurationUtils;
 import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminBuilder;
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.common.naming.TopicName;
 import org.apache.pulsar.common.policies.data.ClusterData;
@@ -332,11 +333,29 @@ public void start() throws Exception {
             createSampleNameSpace(clusterData, cluster);
         } else {
             URL webServiceUrlTls = new URL(
-                    String.format("http://%s:%d", config.getAdvertisedAddress(), config.getWebServicePortTls().get()));
+                    String.format("https://%s:%d", config.getAdvertisedAddress(), config.getWebServicePortTls().get()));
             String brokerServiceUrlTls = String.format("pulsar+ssl://%s:%d", config.getAdvertisedAddress(),
                     config.getBrokerServicePortTls().get());
-            admin = PulsarAdmin.builder().serviceHttpUrl(webServiceUrlTls.toString()).authentication(
-                    config.getBrokerClientAuthenticationPlugin(), config.getBrokerClientAuthenticationParameters()).build();
+            PulsarAdminBuilder builder = PulsarAdmin.builder()
+                    .serviceHttpUrl(webServiceUrlTls.toString())
+                    .authentication(
+                            config.getBrokerClientAuthenticationPlugin(),
+                            config.getBrokerClientAuthenticationParameters());
+
+            // set trust store if needed.
+            if (config.isBrokerClientTlsEnabled()) {
+                if (config.isBrokerClientTlsEnabledWithKeyStore()) {
+                    builder.useKeyStoreTls(true)
+                            .tlsTrustStoreType(config.getBrokerClientTlsTrustStoreType())
+                            .tlsTrustStorePath(config.getBrokerClientTlsTrustStore())
+                            .tlsTrustStorePassword(config.getBrokerClientTlsTrustStorePassword());
+                } else {
+                    builder.tlsTrustCertsFilePath(config.getBrokerClientTrustCertsFilePath());
+                }
+                builder.allowTlsInsecureConnection(config.isTlsAllowInsecureConnection());
+            }
+
+            admin = builder.build();
             ClusterData clusterData = new ClusterData(null, webServiceUrlTls.toString(), null, brokerServiceUrlTls);
             createSampleNameSpace(clusterData, cluster);
         }