Add possibility to use particular certificate for client during 2 way ssl connection

Author: Simca

docs/HTTP-batchsource.md

@@ -403,6 +403,20 @@ error. Do not disable this in production environment on a network you do not ent
 
 **Keystore Key Algorithm:** An algorithm used for keystore.
 
+**Keystore Cert Alias**
+
+ Alias of the key in the keystore to be used for communication. This options is supported only by X.509 keys or keystores.
+ 
+Below is an example how the store need to be prepared:
+ ```
+ cat  client.crt client.key > client-bundle.pem
+
+ openssl pkcs12 -export -in client-bundle.pem -out full-chain.keycert.p12 -name ${CERT_ALIAS}
+
+ keytool -importkeystore -srckeystore full-chain.keycert.p12 -srcstoretype pkcs12 -srcalias ${CERT_ALIAS} \
+                         -destkeystore identity.jks -deststoretype jks -destalias ${CERT_ALIAS}
+ ```
+
 **TrustStore File:** A path to a file which contains truststore.
 
 **TrustStore Type:** Format of a truststore.

docs/HTTP-streamingsource.md

@@ -410,6 +410,20 @@ error. Do not disable this in production environment on a network you do not ent
 
 **Keystore Key Algorithm:** An algorithm used for keystore.
 
+**Keystore Cert Alias**
+
+Alias of the key in the keystore to be used for communication. This options is supported only by X.509 keys or keystores.
+
+Below is an example how the store need to be prepared:
+ ```
+ cat  client.crt client.key > client-bundle.pem
+
+ openssl pkcs12 -export -in client-bundle.pem -out full-chain.keycert.p12 -name ${CERT_ALIAS}
+
+ keytool -importkeystore -srckeystore full-chain.keycert.p12 -srcstoretype pkcs12 -srcalias ${CERT_ALIAS} \
+                         -destkeystore identity.jks -deststoretype jks -destalias ${CERT_ALIAS}
+ ```
+
 **TrustStore File:** A path to a file which contains truststore.
 
 **TrustStore Type:** Format of a truststore.

pom.xml

@@ -21,7 +21,7 @@
   <name>HTTP Plugins</name>
   <groupId>io.cdap</groupId>
   <artifactId>http-plugins</artifactId>
-  <version>1.4.0-SNAPSHOT</version>
+  <version>1.4.1-SNAPSHOT</version>
 
   <licenses>
     <license>
@@ -79,10 +79,12 @@
     <cdap.version>6.1.1</cdap.version>
     <commons.version>3.9</commons.version>
     <common.codec.version>1.12</common.codec.version>
+    <common.logging.version>1.2</common.logging.version>
+    <log4j.version>1.2.17</log4j.version>
     <gson.version>2.8.5</gson.version>
     <hadoop.version>2.3.0</hadoop.version>
     <httpcomponents.version>4.5.9</httpcomponents.version>
-    <hydrator.version>2.4.0-SNAPSHOT</hydrator.version>
+    <hydrator.version>2.4.0</hydrator.version>
     <jackson.version>2.9.9</jackson.version>
     <junit.version>4.11</junit.version>
     <jython.version>2.7.1</jython.version>
@@ -93,6 +95,20 @@
   </properties>
 
   <dependencies>
+    <!-- Required to work properly on external hadoop clusters-->
+    <dependency>
+      <groupId>commons-logging</groupId>
+      <artifactId>commons-logging</artifactId>
+      <version>${common.logging.version}</version>
+      <scope>compile</scope>
+    </dependency>
+    <dependency>
+     <groupId>log4j</groupId>
+     <artifactId>log4j</artifactId>
+     <version>${log4j.version}</version>
+     <scope>compile</scope>
+    </dependency>
+    <!-- / end required -->
     <dependency>
       <groupId>io.cdap.cdap</groupId>
       <artifactId>cdap-api</artifactId>

src/main/java/io/cdap/plugin/http/source/common/BaseHttpSourceConfig.java

@@ -100,6 +100,8 @@ public abstract class BaseHttpSourceConfig extends ReferencePluginConfig {
   public static final String PROPERTY_CIPHER_SUITES = "cipherSuites";
   public static final String PROPERTY_SCHEMA = "schema";
 
+  public static final String PROPERTY_KEYSTORE_CERT_ALIAS = "keystoreCertAlias";
+
   public static final String PAGINATION_INDEX_PLACEHOLDER_REGEX = "\\{pagination.index\\}";
   public static final String PAGINATION_INDEX_PLACEHOLDER = "{pagination.index}";
 
@@ -390,6 +392,12 @@ public abstract class BaseHttpSourceConfig extends ReferencePluginConfig {
   @Description("Output schema. Is required to be set.")
   protected String schema;
 
+  @Name(PROPERTY_KEYSTORE_CERT_ALIAS)
+  @Macro
+  @Nullable
+  @Description("Alias of the key in the keystore to be used for communication")
+  protected String keystoreCertAliasName;
+
   protected BaseHttpSourceConfig(String referenceName) {
     super(referenceName);
   }
@@ -627,6 +635,11 @@ public Schema getSchema() {
     }
   }
 
+  @Nullable
+  public String getKeystoreCertAliasName() {
+    return keystoreCertAliasName;
+  }
+
   @Nullable
   public Map<String, String> getHeadersMap() {
     return getMapFromKeyValueString(headers);

src/main/java/io/cdap/plugin/http/source/common/http/SSLConnectionSocketFactoryCreator.java

@@ -59,6 +59,7 @@ public SSLConnectionSocketFactory create() {
       SSLContext sslContext = SSLContext.getInstance("TLS"); // "TLS" means rely system properties
       sslContext.init(getKeyManagers(), getTrustManagers(), null);
 
+
       return new SSLConnectionSocketFactory(sslContext, config.getTransportProtocolsList().toArray(new String[0]),
                                             cipherSuites, SSLConnectionSocketFactory.getDefaultHostnameVerifier());
     } catch (KeyManagementException | CertificateException | NoSuchAlgorithmException | KeyStoreException
@@ -70,27 +71,30 @@ public SSLConnectionSocketFactory create() {
   private KeyManager[] getKeyManagers() throws CertificateException, NoSuchAlgorithmException,
     KeyStoreException, IOException, UnrecoverableKeyException {
 
-    KeyStore keystore = loadKeystore(config.getKeystoreFile(), config.getKeystoreType().name(),
-                                     config.getKeystorePassword());
-
     String keyStorePassword = config.getKeystorePassword();
+    KeyStore keystore = loadKeystore(config.getKeystoreFile(), config.getKeystoreType().name(), keyStorePassword);
 
     // we have to manually fall back to default keystore. SSLContext won't provide such a functionality.
     if (keystore == null) {
       String keyStore = System.getProperty("javax.net.ssl.keyStore");
       String keyStoreType = System.getProperty("javax.net.ssl.keyStoreType", KeyStore.getDefaultType());
       keyStorePassword = System.getProperty("javax.net.ssl.keyStorePassword", "");
-
       keystore = loadKeystore(keyStore, keyStoreType, keyStorePassword);
     }
 
-    String keystoreAlgorithm =
-      (Strings.isNullOrEmpty(config.getKeystoreKeyAlgorithm())) ? KeyManagerFactory.getDefaultAlgorithm()
+    String keystoreAlgorithm = (Strings.isNullOrEmpty(config.getKeystoreKeyAlgorithm()))
+        ? KeyManagerFactory.getDefaultAlgorithm()
         : config.getKeystoreKeyAlgorithm();
+
     KeyManagerFactory keyManagerFactory = KeyManagerFactory.getInstance(keystoreAlgorithm);
-    char[] passwordArr = (keyStorePassword == null) ? null : keyStorePassword.toCharArray();
-    keyManagerFactory.init(keystore, passwordArr);
-    return keyManagerFactory.getKeyManagers();
+    keyManagerFactory.init(
+      keystore,
+      (keyStorePassword == null) ? null : keyStorePassword.toCharArray()
+    );
+
+    return (Strings.isNullOrEmpty(config.getKeystoreCertAliasName()))
+      ? keyManagerFactory.getKeyManagers()
+      : X509KeyManagerAliasWrapper.getKeyManagers(keyManagerFactory, config.getKeystoreCertAliasName());
   }
 
   private TrustManager[] getTrustManagers()
@@ -100,13 +104,17 @@ private TrustManager[] getTrustManagers()
       return new TrustManager[] { new TrustAllTrustManager() };
     }
 
-    KeyStore trustStore = loadKeystore(config.getTrustStoreFile(), config.getTrustStoreType().name(),
-                                       config.getTrustStorePassword());
+    KeyStore trustStore = loadKeystore(
+      config.getTrustStoreFile(),
+      config.getTrustStoreType().name(),
+      config.getTrustStorePassword()
+    );
+
     TrustManager[] trustManagers = null;
     if (trustStore != null) {
-      String trustStoreAlgorithm =
-        (Strings.isNullOrEmpty(config.getTrustStoreKeyAlgorithm())) ? TrustManagerFactory.getDefaultAlgorithm()
-          : config.getTrustStoreKeyAlgorithm();
+      String trustStoreAlgorithm = (Strings.isNullOrEmpty(config.getTrustStoreKeyAlgorithm()))
+        ? TrustManagerFactory.getDefaultAlgorithm()
+        : config.getTrustStoreKeyAlgorithm();
       TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance(trustStoreAlgorithm);
       trustManagerFactory.init(trustStore);
       trustManagers = trustManagerFactory.getTrustManagers();
@@ -117,13 +125,15 @@ private TrustManager[] getTrustManagers()
   private static KeyStore loadKeystore(String keystoreFile, String type, String password)
     throws IOException, CertificateException, NoSuchAlgorithmException, KeyStoreException {
 
-    KeyStore keystore = null;
-    if (keystoreFile != null) {
-      keystore = KeyStore.getInstance(type);
-      char[] passwordArr = (password == null) ? null : password.toCharArray();
-      try (InputStream is = Files.newInputStream(Paths.get(keystoreFile))) {
-        keystore.load(is, passwordArr);
-      }
+    if (keystoreFile == null) {
+      return null;
+    }
+
+    KeyStore keystore = KeyStore.getInstance(type);
+    char[] passwordArr = (password == null) ? null : password.toCharArray();
+
+    try (InputStream is = Files.newInputStream(Paths.get(keystoreFile))) {
+      keystore.load(is, passwordArr);
     }
     return keystore;
   }

src/main/java/io/cdap/plugin/http/source/common/http/X509KeyManagerAliasWrapper.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright © 2021 Cask Data, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package io.cdap.plugin.http.source.common.http;
+
+import java.net.Socket;
+import java.security.Principal;
+import java.security.PrivateKey;
+import java.security.cert.X509Certificate;
+
+import javax.net.ssl.KeyManager;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.X509KeyManager;
+
+
+/**
+ * This is just wrapper over SunX509KeyManagerImpl with possibility to provide specific alias
+ *
+ * Usage example:
+ *    X509KeyManagerAliasWrapper.getKeyManagers(keyManagerFactory, CERT_ALIAS);
+ */
+public class X509KeyManagerAliasWrapper implements X509KeyManager  {
+
+  private final X509KeyManager originalKeyManager;
+  private final String certAlias;
+
+  public X509KeyManagerAliasWrapper(X509KeyManager originalKeyManager, String certAlias) {
+    this.originalKeyManager = originalKeyManager;
+    this.certAlias = certAlias;
+  }
+
+  public static KeyManager[] getKeyManagers(KeyManagerFactory keyManagerFactory, String certAlias) {
+    KeyManager[] keyManagers = keyManagerFactory.getKeyManagers();
+
+    // Current implementation only support X509 Certificates
+    if (keyManagers.length != 1) {
+      return keyManagers;
+    }
+    if (!(keyManagers[0] instanceof X509KeyManager)) {
+      return keyManagers;
+    }
+
+   return new KeyManager[]{ new X509KeyManagerAliasWrapper((X509KeyManager) keyManagers[0], certAlias) };
+ };
+
+  @Override
+  public String chooseClientAlias(String[] strings, Principal[] principals, Socket socket) {
+    return certAlias;
+  }
+
+  @Override
+  public String[] getClientAliases(String s, Principal[] principals) {
+    return originalKeyManager.getClientAliases(s, principals);
+  }
+
+  @Override
+  public String[] getServerAliases(String s, Principal[] principals) {
+    return originalKeyManager.getServerAliases(s, principals);
+  }
+
+  @Override
+  public String chooseServerAlias(String s, Principal[] principals, Socket socket) {
+    return originalKeyManager.chooseServerAlias(s, principals, socket);
+  }
+
+  @Override
+  public X509Certificate[] getCertificateChain(String s) {
+    return originalKeyManager.getCertificateChain(s);
+  }
+
+  @Override
+  public PrivateKey getPrivateKey(String s) {
+    return originalKeyManager.getPrivateKey(s);
+  }
+}

widgets/HTTP-batchsource.json

@@ -426,6 +426,11 @@
             "default": "SunX509"
           }
         },
+        {
+          "widget-type": "textbox",
+          "label": "Keystore Cert Alias",
+          "name": "keystoreCertAlias"
+        },
         {
           "widget-type": "textbox",
           "label": "TrustStore File",

widgets/HTTP-streamingsource.json

@@ -425,6 +425,11 @@
             "default": "SunX509"
           }
         },
+        {
+          "widget-type": "textbox",
+          "label": "Keystore Cert Alias",
+          "name": "keystoreCertAlias"
+        },
         {
           "widget-type": "textbox",
           "label": "TrustStore File",