This change adds 2 optional configuration properties that can be set to true when the keystore location string is actually a base64 encoded keystore string used in the PCF environments.

Author: rahulnirgude

clients/src/main/java/org/apache/kafka/common/config/SslConfigs.java

@@ -94,6 +94,12 @@ public class SslConfigs {
         + "This is optional for client and only needed if 'ssl.keystore.location' is configured. "
         + "Key store password is not supported for PEM format.";
 
+    public static final String SSL_KEYSTORE_AS_STRING = "ssl.keystore.as.string";
+    public static final String SSL_KEYSTORE_AS_STRING_DOC = "True when using a base64 encoded keystore string";
+
+    public static final String SSL_TRUSTSTORE_AS_STRING = "ssl.truststore.as.string";
+    public static final String SSL_TRUSTSTORE_AS_STRING_DOC = "True when using a base64 encoded truststore string";
+
     public static final String SSL_KEY_PASSWORD_CONFIG = "ssl.key.password";
     public static final String SSL_KEY_PASSWORD_DOC = "The password of the private key in the key store file or "
         + "the PEM key specified in 'ssl.keystore.key'.";
@@ -154,7 +160,9 @@ public static void addClientSslSupport(ConfigDef config) {
                 .define(SslConfigs.SSL_TRUSTMANAGER_ALGORITHM_CONFIG, ConfigDef.Type.STRING, SslConfigs.DEFAULT_SSL_TRUSTMANAGER_ALGORITHM, ConfigDef.Importance.LOW, SslConfigs.SSL_TRUSTMANAGER_ALGORITHM_DOC)
                 .define(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, ConfigDef.Type.STRING, SslConfigs.DEFAULT_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM, ConfigDef.Importance.LOW, SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_DOC)
                 .define(SslConfigs.SSL_SECURE_RANDOM_IMPLEMENTATION_CONFIG, ConfigDef.Type.STRING, null, ConfigDef.Importance.LOW, SslConfigs.SSL_SECURE_RANDOM_IMPLEMENTATION_DOC)
-                .define(SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG, ConfigDef.Type.CLASS, null, ConfigDef.Importance.LOW, SslConfigs.SSL_ENGINE_FACTORY_CLASS_DOC);
+                .define(SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG, ConfigDef.Type.CLASS, null, ConfigDef.Importance.LOW, SslConfigs.SSL_ENGINE_FACTORY_CLASS_DOC)
+                .define(SslConfigs.SSL_KEYSTORE_AS_STRING, ConfigDef.Type.STRING, null, ConfigDef.Importance.LOW,SslConfigs.SSL_KEYSTORE_AS_STRING_DOC)
+                .define(SslConfigs.SSL_TRUSTSTORE_AS_STRING, ConfigDef.Type.STRING, null, ConfigDef.Importance.LOW,SslConfigs.SSL_TRUSTSTORE_AS_STRING_DOC);
     }
 
     public static final Set<String> RECONFIGURABLE_CONFIGS = Utils.mkSet(

clients/src/main/java/org/apache/kafka/common/security/ssl/PcfTruststoreUtility.java

@@ -0,0 +1,36 @@
+package org.apache.kafka.common.security.ssl;
+
+import java.io.BufferedInputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.security.GeneralSecurityException;
+import java.security.KeyStore;
+import java.security.cert.Certificate;
+import java.security.cert.CertificateFactory;
+
+public class PcfTruststoreUtility {
+
+    public static final String CRT = "CRT";
+
+    public static KeyStore createTrustStore(String locationOfCerts, String trustStorePass) throws GeneralSecurityException, IOException {
+        if(!new File(locationOfCerts).exists()){
+            locationOfCerts = System.getenv(locationOfCerts);
+        }
+        KeyStore ks = KeyStore.getInstance("JKS");
+        ks.load(null, trustStorePass.toCharArray());
+        try (FileInputStream fis = new FileInputStream(locationOfCerts)) {
+            try (BufferedInputStream bis = new BufferedInputStream(fis)) {
+                CertificateFactory cf = CertificateFactory.getInstance("X.509");
+                Certificate cert = null;
+
+                while (bis.available() > 0) {
+                    cert = cf.generateCertificate(bis);
+                    ks.setCertificateEntry(String.valueOf(bis.available()), cert);
+                }
+                ks.setCertificateEntry(String.valueOf(bis.available()), cert);
+                return ks;
+            }
+        }
+    }
+}