Changes to support selection of appropriate alias from a keystore with multiple alias entries

Author: rahulnirgude

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

@@ -99,12 +99,6 @@ 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'.";
@@ -166,9 +160,7 @@ 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_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);
+                .define(SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG, ConfigDef.Type.CLASS, null, ConfigDef.Importance.LOW, SslConfigs.SSL_ENGINE_FACTORY_CLASS_DOC);
     }
 
     public static final Set<String> RECONFIGURABLE_CONFIGS = Utils.mkSet(

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

@@ -190,14 +190,12 @@ 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)),
-                Boolean.parseBoolean((String) configs.get(SslConfigs.SSL_KEYSTORE_AS_STRING)));
+                (Password) configs.get(SslConfigs.SSL_KEYSTORE_CERTIFICATE_CHAIN_CONFIG));
 
         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)),
-                Boolean.parseBoolean((String) configs.get(SslConfigs.SSL_TRUSTSTORE_AS_STRING)));
+                (Password) configs.get(SslConfigs.SSL_TRUSTSTORE_CERTIFICATES_CONFIG));
 
         this.sslContext = createSSLContext(keystore, truststore, configs);
     }
@@ -360,7 +358,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, boolean pathAsBase64EncodedString) {
+    protected SecurityStore createKeystore(String type, String path, Password password, Password keyPassword, Password privateKey, Password certificateChain) {
         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 + ".");
@@ -384,12 +382,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, pathAsBase64EncodedString);
+            return new FileBasedStore(type, path, password, keyPassword, true);
         } 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, boolean pathAsBase64EncodedString) {
+    private static SecurityStore createTruststore(String type, String path, Password password, Password trustStoreCerts) {
         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 + ".");
@@ -407,7 +405,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, pathAsBase64EncodedString);
+            return new FileBasedStore(type, path, password, null, false);
         } else
             return null;
     }
@@ -428,15 +426,14 @@ static class FileBasedStore implements SecurityStore {
         private final KeyStore keyStore;
         private final boolean pathAsBase64EncodedString;
 
-        FileBasedStore(String type, String path, Password password, Password keyPassword, boolean isKeyStore, boolean pathAsBase64EncodedString) {
+        FileBasedStore(String type, String path, Password password, Password keyPassword, boolean isKeyStore) {
             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
@@ -457,28 +454,15 @@ public char[] keyPassword() {
          *   using the specified configs (e.g. if the password or keystore type is invalid)
          */
         protected KeyStore load(boolean isKeyStore) {
-            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);
+            try (InputStream in = Files.newInputStream(Paths.get(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);
+                    return ks;
+                } catch (GeneralSecurityException | IOException e) {
+                    throw new KafkaException("Failed to load SSL keystore " + path + " of type " + type, e);
                 }
-                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);
-            }
         }
 
         private Long lastModifiedMs(String path) {

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

@@ -591,8 +591,7 @@ 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,
-                    Boolean.parseBoolean((String) sslConfig.get(SslConfigs.SSL_KEYSTORE_AS_STRING))
+                    true
             );
         } else {
             store = new PemStore(