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/security/ssl/DefaultSslEngineFactory.java

@@ -30,6 +30,31 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.Reconfigurable;
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.config.SslConfigs;
+import org.apache.kafka.common.config.internals.BrokerSecurityConfigs;
+import org.apache.kafka.common.config.types.Password;
+import org.apache.kafka.common.network.Mode;
+import org.apache.kafka.common.utils.Base64;
+import org.apache.kafka.common.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.net.ssl.KeyManager;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.SSLEngineResult;
+import javax.net.ssl.SSLException;
+import javax.net.ssl.SSLParameters;
+import javax.net.ssl.TrustManagerFactory;
+import javax.net.ssl.X509ExtendedKeyManager;
+import javax.net.ssl.X509KeyManager;
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
@@ -164,12 +189,14 @@ public void configure(Map<String, ?> configs) {
                 (Password) configs.get(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG),
                 (Password) configs.get(SslConfigs.SSL_KEY_PASSWORD_CONFIG),
                 (Password) configs.get(SslConfigs.SSL_KEYSTORE_KEY_CONFIG),
-                (Password) configs.get(SslConfigs.SSL_KEYSTORE_CERTIFICATE_CHAIN_CONFIG));
+                (Password) configs.get(SslConfigs.SSL_KEYSTORE_CERTIFICATE_CHAIN_CONFIG)),
+                Boolean.parseBoolean((String) configs.get(SslConfigs.SSL_KEYSTORE_AS_STRING)));
 
         this.truststore = createTruststore((String) configs.get(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG),
                 (String) configs.get(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG),
                 (Password) configs.get(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG),
-                (Password) configs.get(SslConfigs.SSL_TRUSTSTORE_CERTIFICATES_CONFIG));
+                (Password) configs.get(SslConfigs.SSL_TRUSTSTORE_CERTIFICATES_CONFIG)),
+                Boolean.parseBoolean((String) configs.get(SslConfigs.SSL_TRUSTSTORE_AS_STRING)));
 
         this.sslContext = createSSLContext(keystore, truststore);
     }
@@ -275,7 +302,7 @@ protected TrustManager[] getTrustManagers(SecurityStore truststore, String tmfAl
     }
 
     // Visibility to override for testing
-    protected SecurityStore createKeystore(String type, String path, Password password, Password keyPassword, Password privateKey, Password certificateChain) {
+    protected SecurityStore createKeystore(String type, String path, Password password, Password keyPassword, Password privateKey, Password certificateChain, boolean pathAsBase64EncodedString) {
         if (privateKey != null) {
             if (!PEM_TYPE.equals(type))
                 throw new InvalidConfigurationException("SSL private key can be specified only for PEM, but key store type is " + type + ".");
@@ -299,12 +326,12 @@ else if (password != null)
         } else if (path != null && password == null) {
             throw new InvalidConfigurationException("SSL key store is specified, but key store password is not specified.");
         } else if (path != null && password != null) {
-            return new FileBasedStore(type, path, password, keyPassword, true);
+            return new FileBasedStore(type, path, password, keyPassword, true, pathAsBase64EncodedString);
         } else
             return null; // path == null, clients may use this path with brokers that don't require client auth
     }
 
-    private static SecurityStore createTruststore(String type, String path, Password password, Password trustStoreCerts) {
+    private static SecurityStore createTruststore(String type, String path, Password password, Password trustStoreCerts, boolean pathAsBase64EncodedString) {
         if (trustStoreCerts != null) {
             if (!PEM_TYPE.equals(type))
                 throw new InvalidConfigurationException("SSL trust store certs can be specified only for PEM, but trust store type is " + type + ".");
@@ -322,7 +349,7 @@ else if (password != null)
         } else if (path == null && password != null) {
             throw new InvalidConfigurationException("SSL trust store is not specified, but trust store password is specified.");
         } else if (path != null) {
-            return new FileBasedStore(type, path, password, null, false);
+            return new FileBasedStore(type, path, password, null, false, pathAsBase64EncodedString);
         } else
             return null;
     }
@@ -341,15 +368,17 @@ static class FileBasedStore implements SecurityStore {
         protected final Password keyPassword;
         private final Long fileLastModifiedMs;
         private final KeyStore keyStore;
+        private final boolean pathAsBase64EncodedString;
 
-        FileBasedStore(String type, String path, Password password, Password keyPassword, boolean isKeyStore) {
+        FileBasedStore(String type, String path, Password password, Password keyPassword, boolean isKeyStore, boolean pathAsBase64EncodedString) {
             Objects.requireNonNull(type, "type must not be null");
             this.type = type;
             this.path = path;
             this.password = password;
             this.keyPassword = keyPassword;
             fileLastModifiedMs = lastModifiedMs(path);
             this.keyStore = load(isKeyStore);
+            this.pathAsBase64EncodedString = pathAsBase64EncodedString;
         }
 
         @Override
@@ -370,11 +399,24 @@ public char[] keyPassword() {
          *   using the specified configs (e.g. if the password or keystore type is invalid)
          */
         protected KeyStore load(boolean isKeyStore) {
-            try (InputStream in = Files.newInputStream(Paths.get(path))) {
+            if (path == null) {
+                throw new KafkaException("Failed to load SSL keystore: path was null");
+            }
+            InputStream in;
+            try {
+                if (pathAsBase64EncodedString) {
+                    String encodedKeyStore = System.getenv(path);
+                    in = new ByteArrayInputStream(Base64.decoder().decode(encodedKeyStore));
+                } else if (type.equalsIgnoreCase(TruststoreUtility.CRT)) {
+                    return TruststoreUtility.createTrustStore(path, password.value());
+                } else {
+                    in = new FileInputStream(path);
+                }
                 KeyStore ks = KeyStore.getInstance(type);
                 // If a password is not set access to the truststore is still available, but integrity checking is disabled.
                 char[] passwordChars = password != null ? password.value().toCharArray() : null;
                 ks.load(in, passwordChars);
+                in.close();
                 return ks;
             } catch (GeneralSecurityException | IOException e) {
                 throw new KafkaException("Failed to load SSL keystore " + path + " of type " + type, e);

…n/security/ssl/PcfTruststoreUtility.java …mmon/security/ssl/TruststoreUtility.javaclients/src/main/java/org/apache/kafka/common/security/ssl/PcfTruststoreUtility.java renamed to clients/src/main/java/org/apache/kafka/common/security/ssl/TruststoreUtility.java clients/src/main/java/org/apache/kafka/common/security/ssl/PcfTruststoreUtility.java renamed to clients/src/main/java/org/apache/kafka/common/security/ssl/TruststoreUtility.java

@@ -9,7 +9,7 @@
 import java.security.cert.Certificate;
 import java.security.cert.CertificateFactory;
 
-public class PcfTruststoreUtility {
+public class TruststoreUtility {
 
     public static final String CRT = "CRT";
 

clients/src/test/java/org/apache/kafka/common/security/ssl/SslFactoryTest.java

@@ -591,7 +591,8 @@ private KeyStore sslKeyStore(Map<String, Object> sslConfig) {
                     (String) sslConfig.get(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG),
                     (Password) sslConfig.get(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG),
                     (Password) sslConfig.get(SslConfigs.SSL_KEY_PASSWORD_CONFIG),
-                    true
+                    true,
+                    Boolean.parseBoolean((String) sslConfig.get(SslConfigs.SSL_KEYSTORE_AS_STRING))
             );
         } else {
             store = new PemStore(