Add HTTP Client keystore for mutual TLS

Author: hackedd

kafka-connect-http/src/main/java/com/github/castorm/kafka/connect/http/client/okhttp/OkHttpClient.java

@@ -35,9 +35,23 @@
 import okhttp3.Response;
 import okhttp3.logging.HttpLoggingInterceptor;
 
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.TrustManagerFactory;
+import javax.net.ssl.X509TrustManager;
+import java.io.FileInputStream;
 import java.io.IOException;
+import java.io.InputStream;
 import java.net.InetSocketAddress;
 import java.net.Proxy;
+import java.security.KeyManagementException;
+import java.security.KeyStore;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.security.UnrecoverableKeyException;
+import java.security.cert.CertificateException;
+import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -68,7 +82,8 @@ public void configure(Map<String, ?> configs) {
         OkHttpClientConfig config = new OkHttpClientConfig(configs);
 
         authenticator = config.getAuthenticator();
-        client = new okhttp3.OkHttpClient.Builder()
+
+        okhttp3.OkHttpClient.Builder builder = new okhttp3.OkHttpClient.Builder()
                 .connectionPool(new ConnectionPool(config.getMaxIdleConnections(), config.getKeepAliveDuration(), MILLISECONDS))
                 .connectTimeout(config.getConnectionTimeoutMillis(), MILLISECONDS)
                 .readTimeout(config.getReadTimeoutMillis(), MILLISECONDS)
@@ -77,8 +92,11 @@ public void configure(Map<String, ?> configs) {
                 .addInterceptor(chain -> chain.proceed(authorize(chain.request())))
                 .authenticator((route, response) -> authorize(response.request()))
                 .proxy(resolveProxy(config.getProxyHost(), config.getProxyPort()))
-                .proxyAuthenticator(resolveProxyAuthenticator(config.getProxyUsername(), config.getProxyPassword()))
-                .build();
+                .proxyAuthenticator(resolveProxyAuthenticator(config.getProxyUsername(), config.getProxyPassword()));
+
+        resolveSslSocketFactory(builder, config.getKeyStore(), config.getKeyStorePassword().value());
+
+        client = builder.build();
     }
 
     private Request authorize(Request request) {
@@ -108,6 +126,57 @@ private static Authenticator resolveProxyAuthenticator(String username, String p
                         .build();
     }
 
+    private static void resolveSslSocketFactory(okhttp3.OkHttpClient.Builder builder, String keyStorePath, String keyStorePassword) {
+        if (keyStorePath.isEmpty()) {
+            return;
+        }
+
+        KeyStore keyStore;
+        try {
+            keyStore = KeyStore.getInstance("PKCS12");
+        } catch (KeyStoreException e) {
+            throw new IllegalStateException("Unable to create keystore", e);
+        }
+
+        try (InputStream is = new FileInputStream(keyStorePath)) {
+            keyStore.load(is, keyStorePassword.toCharArray());
+        }
+        catch (CertificateException | IOException | NoSuchAlgorithmException e) {
+            throw new IllegalStateException(String.format("Unable to load keystore '%s'", keyStorePath), e);
+        }
+
+        KeyManagerFactory kmf;
+        try {
+            kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+            kmf.init(keyStore, keyStorePassword.toCharArray());
+        } catch (UnrecoverableKeyException | NoSuchAlgorithmException | KeyStoreException e) {
+            throw new IllegalStateException("Unable to initialize key manager", e);
+        }
+
+        SSLContext sslContext;
+        try {
+            sslContext = SSLContext.getInstance("TLS");
+            sslContext.init(kmf.getKeyManagers(), null, null);
+        } catch (NoSuchAlgorithmException | KeyManagementException e) {
+            throw new IllegalStateException("Unable to initialize SSL context", e);
+        }
+
+        TrustManagerFactory trustManagerFactory;
+        try {
+            trustManagerFactory = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+            trustManagerFactory.init((KeyStore) null);
+        } catch (NoSuchAlgorithmException | KeyStoreException e) {
+            throw new IllegalStateException("Unable to initialize trust manager", e);
+        }
+
+        TrustManager[] trustManagers = trustManagerFactory.getTrustManagers();
+        if (trustManagers.length != 1 || !(trustManagers[0] instanceof X509TrustManager)) {
+            throw new IllegalStateException("Unexpected default trust managers:" + Arrays.toString(trustManagers));
+        }
+
+        builder.sslSocketFactory(sslContext.getSocketFactory(), (X509TrustManager) trustManagers[0]);
+    }
+
     @Override
     @SneakyThrows(IOException.class)
     public HttpResponse execute(HttpRequest httpRequest) {

kafka-connect-http/src/main/java/com/github/castorm/kafka/connect/http/client/okhttp/OkHttpClientConfig.java

@@ -25,6 +25,7 @@
 import lombok.Getter;
 import org.apache.kafka.common.config.AbstractConfig;
 import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.types.Password;
 
 import java.util.Map;
 
@@ -34,6 +35,7 @@
 import static org.apache.kafka.common.config.ConfigDef.Type.INT;
 import static org.apache.kafka.common.config.ConfigDef.Type.LONG;
 import static org.apache.kafka.common.config.ConfigDef.Type.STRING;
+import static org.apache.kafka.common.config.ConfigDef.Type.PASSWORD;
 
 @Getter
 public class OkHttpClientConfig extends AbstractConfig {
@@ -47,6 +49,8 @@ public class OkHttpClientConfig extends AbstractConfig {
     private static final String PROXY_PORT = "http.client.proxy.port";
     private static final String PROXY_USERNAME = "http.client.proxy.username";
     private static final String PROXY_PASSWORD = "http.client.proxy.password";
+    private static final String KEYSTORE = "http.client.keystore";
+    private static final String KEYSTORE_PASSWORD = "http.client.keystore.password";
 
     private final Long connectionTimeoutMillis;
     private final Long readTimeoutMillis;
@@ -57,6 +61,8 @@ public class OkHttpClientConfig extends AbstractConfig {
     private final Integer proxyPort;
     private final String proxyUsername;
     private final String proxyPassword;
+    private final String keyStore;
+    private final Password keyStorePassword;
 
     OkHttpClientConfig(Map<String, ?> originals) {
         super(config(), originals);
@@ -69,6 +75,8 @@ public class OkHttpClientConfig extends AbstractConfig {
         proxyPort = getInt(PROXY_PORT);
         proxyUsername = getString(PROXY_USERNAME);
         proxyPassword = getString(PROXY_PASSWORD);
+        keyStore = getString(KEYSTORE);
+        keyStorePassword = getPassword(KEYSTORE_PASSWORD);
     }
 
     public static ConfigDef config() {
@@ -82,6 +90,8 @@ public static ConfigDef config() {
                 .define(PROXY_PORT, INT, 3128, MEDIUM, "Proxy port")
                 .define(PROXY_USERNAME, STRING, "", MEDIUM, "Proxy username")
                 .define(PROXY_PASSWORD, STRING, "", MEDIUM, "Proxy password")
+                .define(KEYSTORE, STRING, "", MEDIUM, "Keystore")
+                .define(KEYSTORE_PASSWORD, PASSWORD, "", MEDIUM, "Keystore password")
                 ;
     }
 }