support alias for keystore

Author: rahulnirgude

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

@@ -89,6 +89,11 @@ public class SslConfigs {
     public static final String SSL_KEYSTORE_LOCATION_DOC = "The location of the key store file. "
         + "This is optional for client and can be used for two-way authentication for client.";
 
+    public static final String SSL_KEYSTORE_ALIAS_CONFIG = "ssl.keystore.alias";
+    public static final String SSL_KEYSTORE_ALIAS_DOC = "The Alias of key in the key store file. "
+            + "This is optional for client and can be used for two-way authentication for client.";
+
+
     public static final String SSL_KEYSTORE_PASSWORD_CONFIG = "ssl.keystore.password";
     public static final String SSL_KEYSTORE_PASSWORD_DOC = "The store password for the key store file. "
         + "This is optional for client and only needed if 'ssl.keystore.location' is configured. "
@@ -148,6 +153,7 @@ public static void addClientSslSupport(ConfigDef config) {
                 .define(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, ConfigDef.Type.LIST, SslConfigs.DEFAULT_SSL_ENABLED_PROTOCOLS, ConfigDef.Importance.MEDIUM, SslConfigs.SSL_ENABLED_PROTOCOLS_DOC)
                 .define(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, ConfigDef.Type.STRING, SslConfigs.DEFAULT_SSL_KEYSTORE_TYPE, ConfigDef.Importance.MEDIUM, SslConfigs.SSL_KEYSTORE_TYPE_DOC)
                 .define(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, ConfigDef.Type.STRING, null,  ConfigDef.Importance.HIGH, SslConfigs.SSL_KEYSTORE_LOCATION_DOC)
+                .define(SslConfigs.SSL_KEYSTORE_ALIAS_CONFIG, ConfigDef.Type.STRING, null,  ConfigDef.Importance.HIGH, SslConfigs.SSL_KEYSTORE_ALIAS_DOC)
                 .define(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, ConfigDef.Type.PASSWORD, null, ConfigDef.Importance.HIGH, SslConfigs.SSL_KEYSTORE_PASSWORD_DOC)
                 .define(SslConfigs.SSL_KEY_PASSWORD_CONFIG, ConfigDef.Type.PASSWORD, null, ConfigDef.Importance.HIGH, SslConfigs.SSL_KEY_PASSWORD_DOC)
                 .define(SslConfigs.SSL_KEYSTORE_KEY_CONFIG, ConfigDef.Type.PASSWORD, null,  ConfigDef.Importance.HIGH, SslConfigs.SSL_KEYSTORE_KEY_DOC)

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

@@ -26,7 +26,7 @@
 import org.apache.kafka.common.security.auth.SslEngineFactory;
 import org.apache.kafka.common.utils.SecurityUtils;
 import org.apache.kafka.common.utils.Utils;
-
+import java.net.Socket;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -41,7 +41,7 @@
 import org.apache.kafka.common.utils.Utils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-
+import java.security.PrivateKey;
 import javax.net.ssl.KeyManager;
 import javax.net.ssl.KeyManagerFactory;
 import javax.net.ssl.SSLContext;
@@ -54,7 +54,8 @@
 import javax.net.ssl.X509KeyManager;
 import java.io.ByteArrayInputStream;
 import java.io.InputStream;
-
+import javax.net.ssl.X509ExtendedKeyManager;
+import javax.net.ssl.X509KeyManager;
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.io.InputStream;
@@ -198,7 +199,7 @@ public void configure(Map<String, ?> configs) {
                 (Password) configs.get(SslConfigs.SSL_TRUSTSTORE_CERTIFICATES_CONFIG)),
                 Boolean.parseBoolean((String) configs.get(SslConfigs.SSL_TRUSTSTORE_AS_STRING)));
 
-        this.sslContext = createSSLContext(keystore, truststore);
+        this.sslContext = createSSLContext(keystore, truststore, configs);
     }
 
     @Override
@@ -261,7 +262,7 @@ private static SecureRandom createSecureRandom(String key) {
         }
     }
 
-    private SSLContext createSSLContext(SecurityStore keystore, SecurityStore truststore) {
+    private SSLContext createSSLContext(SecurityStore keystore, SecurityStore truststore, Map<String, ?> configs) {
         try {
             SSLContext sslContext;
             if (provider != null)
@@ -285,7 +286,7 @@ private SSLContext createSSLContext(SecurityStore keystore, SecurityStore trusts
             String tmfAlgorithm = this.tmfAlgorithm != null ? this.tmfAlgorithm : TrustManagerFactory.getDefaultAlgorithm();
             TrustManager[] trustManagers = getTrustManagers(truststore, tmfAlgorithm);
 
-            sslContext.init(keyManagers, trustManagers, this.secureRandomImplementation);
+            sslContext.init(applyAliasToKM(keyManagers, (String)configs.get("ssl.keystore.alias")), trustManagers, this.secureRandomImplementation);
             log.debug("Created SSL context with keystore {}, truststore {}, provider {}.",
                     keystore, truststore, sslContext.getProvider().getName());
             return sslContext;
@@ -294,6 +295,63 @@ private SSLContext createSSLContext(SecurityStore keystore, SecurityStore trusts
         }
     }
 
+    private KeyManager[] applyAliasToKM(KeyManager[] kms, final String alias) {
+        if(alias == null || alias.isEmpty()){
+            return kms;
+        }
+
+        log.debug("Applying the custom KeyManagers for alias: {}", alias);
+
+        KeyManager[] updatedKMs = new KeyManager[kms.length];
+
+        int i=0;
+        for(KeyManager km : kms){
+            final X509KeyManager origKM = (X509KeyManager)km;
+            X509ExtendedKeyManager exKM = new X509ExtendedKeyManager() {
+                /* (non-Javadoc)
+                 * @see javax.net.ssl.X509ExtendedKeyManager#chooseEngineClientAlias(java.lang.String[], java.security.Principal[], javax.net.ssl.SSLEngine)
+                 */
+                @Override
+                public String chooseEngineClientAlias(String[] arg0,
+                                                      Principal[] arg1, SSLEngine arg2) {
+                    return alias;
+                }
+
+                @Override
+                public String[] getServerAliases(String arg0, Principal[] arg1) {
+                    return origKM.getServerAliases(arg0, arg1);
+                }
+
+                @Override
+                public PrivateKey getPrivateKey(String arg0) {
+                    return origKM.getPrivateKey(arg0);
+                }
+
+                @Override
+                public String[] getClientAliases(String arg0, Principal[] arg1) {
+                    return origKM.getClientAliases(arg0, arg1);
+                }
+
+                @Override
+                public X509Certificate[] getCertificateChain(String arg0) {
+                    return origKM.getCertificateChain(arg0);
+                }
+
+                @Override
+                public String chooseServerAlias(String arg0, Principal[] arg1, Socket arg2) {
+                    return origKM.chooseServerAlias(arg0, arg1, arg2);
+                }
+
+                @Override
+                public String chooseClientAlias(String[] arg0, Principal[] arg1, Socket arg2) {
+                    return alias;
+                }
+            };
+            updatedKMs[i++] = exKM;
+        }
+        return updatedKMs;
+    }
+
     protected TrustManager[] getTrustManagers(SecurityStore truststore, String tmfAlgorithm) throws NoSuchAlgorithmException, KeyStoreException {
         TrustManagerFactory tmf = TrustManagerFactory.getInstance(tmfAlgorithm);
         KeyStore ts = truststore == null ? null : truststore.get();