[pulsar-broker] add broker-bookie mTLS support (apache#5042)

* [pulsar-broker] add broker-bookie mTLS support

* add default values

conf/broker.conf

@@ -457,6 +457,34 @@ bookkeeperClientMinAvailableBookiesInIsolationGroups=
 # Disable Sticy Read until {@link https://github.com/apache/bookkeeper/issues/1970} is fixed
 bookkeeperEnableStickyReads=false
 
+# Set the client security provider factory class name. 
+# Default: org.apache.bookkeeper.tls.TLSContextFactory
+bookkeeperTLSProviderFactoryClass=org.apache.bookkeeper.tls.TLSContextFactory
+
+# Enable tls authentication with bookie
+bookkeeperTLSClientAuthentication=false
+
+# Supported type: PEM, JKS, PKCS12. Default value: PEM
+bookkeeperTLSKeyFileType=PEM
+
+#Supported type: PEM, JKS, PKCS12. Default value: PEM
+bookkeeperTLSTrustCertTypes=PEM
+
+# Path to file containing keystore password, if the client keystore is password protected.
+bookkeeperTLSKeyStorePasswordPath=
+
+# Path to file containing truststore password, if the client truststore is password protected.
+bookkeeperTLSTrustStorePasswordPath=
+
+# Path for the TLS private key file
+bookkeeperTLSKeyFilePath=
+
+# Path for the TLS certificate file
+bookkeeperTLSCertificateFilePath=
+
+# Path for the trusted TLS certificate file
+bookkeeperTLSTrustCertsFilePath=
+
 ### --- Managed Ledger --- ###
 
 # Number of bookies to use when creating a ledger

conf/standalone.conf

@@ -280,6 +280,35 @@ bookkeeperClientSecondaryIsolationGroups=
 # else broker will include bookkeeperClientSecondaryIsolationGroups bookies in isolated list.
 bookkeeperClientMinAvailableBookiesInIsolationGroups=
 
+# Set the client security provider factory class name. 
+# Default: org.apache.bookkeeper.tls.TLSContextFactory
+bookkeeperTLSProviderFactoryClass=org.apache.bookkeeper.tls.TLSContextFactory
+
+# Enable tls authentication with bookie
+bookkeeperTLSClientAuthentication=false
+
+# Supported type: PEM, JKS, PKCS12. Default value: PEM
+bookkeeperTLSKeyFileType=PEM
+
+#Supported type: PEM, JKS, PKCS12. Default value: PEM
+bookkeeperTLSTrustCertTypes=PEM
+
+# Path to file containing keystore password, if the client keystore is password protected.
+bookkeeperTLSKeyStorePasswordPath=
+
+# Path to file containing truststore password, if the client truststore is password protected.
+bookkeeperTLSTrustStorePasswordPath=
+
+# Path for the TLS private key file
+bookkeeperTLSKeyFilePath=
+
+# Path for the TLS certificate file
+bookkeeperTLSCertificateFilePath=
+
+# Path for the trusted TLS certificate file
+bookkeeperTLSTrustCertsFilePath=
+
+
 ### --- Managed Ledger --- ###
 
 # Number of bookies to use when creating a ledger

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

@@ -771,6 +771,36 @@ public class ServiceConfiguration implements PulsarConfiguration {
             "preference) to read all entries for a ledger.")
     private boolean bookkeeperEnableStickyReads = true;
 
+    @FieldContext(category = CATEGORY_STORAGE_BK, doc = "Set the client security provider factory class name. "
+            + "Default: org.apache.bookkeeper.tls.TLSContextFactory")
+    private String bookkeeperTLSProviderFactoryClass = "org.apache.bookkeeper.tls.TLSContextFactory";
+
+    @FieldContext(category = CATEGORY_STORAGE_BK, doc = "Enable tls authentication with bookie")
+    private boolean bookkeeperTLSClientAuthentication = false;
+
+    @FieldContext(category = CATEGORY_STORAGE_BK, doc = "Supported type: PEM, JKS, PKCS12. Default value: PEM")
+    private String bookkeeperTLSKeyFileType = "PEM";
+
+    @FieldContext(category = CATEGORY_STORAGE_BK, doc = "Supported type: PEM, JKS, PKCS12. Default value: PEM")
+    private String bookkeeperTLSTrustCertTypes = "PEM";
+
+    @FieldContext(category = CATEGORY_STORAGE_BK, doc = "Path to file containing keystore password, "
+            + "if the client keystore is password protected.")
+    private String bookkeeperTLSKeyStorePasswordPath;
+
+    @FieldContext(category = CATEGORY_STORAGE_BK, doc = "Path to file containing truststore password, "
+            + "if the client truststore is password protected.")
+    private String bookkeeperTLSTrustStorePasswordPath;
+
+    @FieldContext(category = CATEGORY_STORAGE_BK, doc = "Path for the TLS private key file")
+    private String bookkeeperTLSKeyFilePath;
+
+    @FieldContext(category = CATEGORY_STORAGE_BK, doc = "Path for the TLS certificate file")
+    private String bookkeeperTLSCertificateFilePath;
+
+    @FieldContext(category = CATEGORY_STORAGE_BK, doc = "Path for the trusted TLS certificate file")
+    private String bookkeeperTLSTrustCertsFilePath;
+
     /**** --- Managed Ledger --- ****/
     @FieldContext(
         minValue = 1,

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

@@ -74,6 +74,18 @@ private ClientConfiguration createBkClientConfiguration(ServiceConfiguration con
             bkConf.setProperty(conf.getBookkeeperClientAuthenticationParametersName(),
                     conf.getBookkeeperClientAuthenticationParameters());
         }
+
+        if (conf.isBookkeeperTLSClientAuthentication()) {
+            bkConf.setTLSClientAuthentication(true);
+            bkConf.setTLSCertificatePath(conf.getBookkeeperTLSCertificateFilePath());
+            bkConf.setTLSKeyStore(conf.getBookkeeperTLSKeyFilePath());
+            bkConf.setTLSKeyStoreType(conf.getBookkeeperTLSKeyFileType());
+            bkConf.setTLSKeyStorePasswordPath(conf.getBookkeeperTLSKeyStorePasswordPath());
+            bkConf.setTLSProviderFactoryClass(conf.getBookkeeperTLSProviderFactoryClass());
+            bkConf.setTLSTrustStore(conf.getBookkeeperTLSTrustCertsFilePath());
+            bkConf.setTLSTrustStoreType(conf.getBookkeeperTLSTrustCertTypes());
+            bkConf.setTLSTrustStorePasswordPath(conf.getBookkeeperTLSTrustStorePasswordPath());
+        }
 
         bkConf.setThrottleValue(0);
         bkConf.setAddEntryTimeout((int) conf.getBookkeeperClientTimeoutInSeconds());