Add ssh tunnel support for jdbc connectors, issue #312

airbyte-integrations/bases/base-java/build.gradle

@@ -5,6 +5,12 @@ plugins {
 
 dependencies {
     implementation 'commons-cli:commons-cli:1.4'
+    implementation group: 'org.apache.sshd', name: 'sshd-mina', version: '2.7.0'
+    // bouncycastle is pinned to version-match the transitive dependency from kubernetes client-java
+    // because a version conflict causes "parameter object not a ECParameterSpec" on ssh tunnel initiation
+    implementation group: 'org.bouncycastle', name: 'bcprov-jdk15on', version: '1.66'
+    implementation group: 'org.bouncycastle', name: 'bcpkix-jdk15on', version: '1.66'
+    implementation group: 'org.bouncycastle', name: 'bctls-jdk15on', version: '1.66'
 
     implementation project(':airbyte-protocol:models')
     implementation project(":airbyte-json-validation")

airbyte-integrations/bases/base-java/src/main/java/io/airbyte/integrations/BaseConnector.java

@@ -24,10 +24,12 @@
 
 package io.airbyte.integrations;
 
+import com.fasterxml.jackson.databind.JsonNode;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.commons.resources.MoreResources;
 import io.airbyte.integrations.base.Integration;
 import io.airbyte.protocol.models.ConnectorSpecification;
+import java.io.IOException;
 
 public abstract class BaseConnector implements Integration {
 
@@ -42,7 +44,18 @@ public abstract class BaseConnector implements Integration {
   public ConnectorSpecification spec() throws Exception {
     // return a JsonSchema representation of the spec for the integration.
     final String resourceString = MoreResources.readResource("spec.json");
-    return Jsons.deserialize(resourceString, ConnectorSpecification.class);
+    return Jsons.object(addToSpec(Jsons.deserialize(resourceString)), ConnectorSpecification.class);
+  }
+
+  /**
+   * Extension point for child classes to add things to the spec json tree before it's deserialized
+   * into a ConnectorSpecification object.
+   *
+   * @param root
+   * @return root
+   */
+  public JsonNode addToSpec(JsonNode root) throws IOException {
+    return root;
   }
 
 }

airbyte-integrations/bases/base-java/src/main/java/io/airbyte/integrations/base/SSHTunnel.java

@@ -0,0 +1,276 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2020 Airbyte
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.airbyte.integrations.base;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import java.io.IOException;
+import java.io.StringReader;
+import java.net.URISyntaxException;
+import java.security.KeyPair;
+import java.security.NoSuchAlgorithmException;
+import java.security.interfaces.RSAPrivateKey;
+import java.security.interfaces.RSAPublicKey;
+import java.security.spec.InvalidKeySpecException;
+import org.apache.sshd.client.SshClient;
+import org.apache.sshd.client.keyverifier.AcceptAllServerKeyVerifier;
+import org.apache.sshd.client.session.ClientSession;
+import org.apache.sshd.common.util.net.SshdSocketAddress;
+import org.apache.sshd.server.forward.AcceptAllForwardingFilter;
+import org.bouncycastle.asn1.x509.SubjectPublicKeyInfo;
+import org.bouncycastle.openssl.PEMKeyPair;
+import org.bouncycastle.openssl.PEMParser;
+import org.bouncycastle.openssl.jcajce.JcaPEMKeyConverter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Encapsulates the connection configuration for an ssh tunnel port forward through a proxy/bastion
+ * host plus the remote host and remote port to forward to a specified local port.
+ */
+public class SSHTunnel {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(SSHTunnel.class);
+
+  public static final int TIMEOUT_MILLIS = 15000; // 15 seconds
+  private final String method;
+  private final String host;
+  private final String tunnelSshPort;
+  private final String user;
+  private final String sshkey;
+  private final String password;
+  private final String remoteDatabaseHost;
+  private final String remoteDatabasePort;
+  private final String tunnelDatabasePort;
+
+  private SSHTunnel tunnelConfig = null;
+  private SshClient sshclient = null;
+  private ClientSession tunnelSession = null;
+
+  public SSHTunnel(String method,
+                   String host,
+                   String tunnelSshPort,
+                   String user,
+                   String sshkey,
+                   String password,
+                   String remoteDatabaseHost,
+                   String remoteDatabasePort,
+                   String tunnelDatabasePort) {
+    if (method == null) {
+      this.method = "NO_TUNNEL";
+    } else {
+      this.method = method;
+    }
+    this.host = host;
+    this.tunnelSshPort = tunnelSshPort;
+    this.user = user;
+    this.sshkey = sshkey;
+    this.password = password;
+    this.remoteDatabaseHost = remoteDatabaseHost;
+    this.remoteDatabasePort = remoteDatabasePort;
+    this.tunnelDatabasePort = tunnelDatabasePort;
+  }
+
+  public static SSHTunnel getInstance(JsonNode config) {
+    JsonNode ourConfig = config.get("tunnel_method");
+    SSHTunnel sshconfig = new SSHTunnel(
+        getConfigValueOrNull(ourConfig, "tunnel_method"),
+        getConfigValueOrNull(ourConfig, "tunnel_host"),
+        getConfigValueOrNull(ourConfig, "tunnel_ssh_port"),
+        getConfigValueOrNull(ourConfig, "tunnel_username"),
+        getConfigValueOrNull(ourConfig, "tunnel_usersshkey"),
+        getConfigValueOrNull(ourConfig, "tunnel_userpass"),
+        getConfigValueOrNull(ourConfig, "tunnel_db_remote_host"),
+        getConfigValueOrNull(ourConfig, "tunnel_db_remote_port"),
+        getConfigValueOrNull(ourConfig, "tunnel_localport"));
+    return sshconfig;
+  }
+
+  static String getConfigValueOrNull(JsonNode config, String key) {
+    return config != null && config.has(key) ? config.get(key).asText() : null;
+  }
+
+  /**
+   * Starts an ssh session; wrap this in a try-finally and use closeTunnel() to close it.
+   *
+   * @throws IOException
+   */
+  public void openTunnelIfRequested() {
+    if (shouldTunnel()) {
+      if (tunnelSession != null || sshclient != null) {
+        throw new RuntimeException("SSH Tunnel was requested to be opened while it was already open.  This is a coding error.");
+      }
+      sshclient = createClient();
+      tunnelSession = openTunnel(sshclient);
+    }
+  }
+
+  /**
+   * Closes a tunnel if one was open, and otherwise doesn't do anything (safe to run).
+   */
+  public void closeTunnel() {
+    try {
+      if (shouldTunnel()) {
+        if (tunnelSession != null) {
+          tunnelSession.close();
+        }
+        if (sshclient != null) {
+          sshclient.stop();
+        }
+        tunnelSession = null;
+        sshclient = null;
+      }
+    } catch (Throwable t) {
+      throw new RuntimeException(t);
+    }
+  }
+
+  public boolean shouldTunnel() {
+    return method != null && !"NO_TUNNEL".equals(method);
+  }
+
+  public String getMethod() {
+    return method;
+  }
+
+  public String getHost() {
+    return host;
+  }
+
+  public String getTunnelSshPort() {
+    return tunnelSshPort;
+  }
+
+  public String getUser() {
+    return user;
+  }
+
+  private String getSSHKey() {
+    return sshkey;
+  }
+
+  private String getPassword() {
+    return password;
+  }
+
+  public String getRemoteDatabaseHost() {
+    return remoteDatabaseHost;
+  }
+
+  public String getRemoteDatabasePort() {
+    return remoteDatabasePort;
+  }
+
+  public String getTunnelDatabasePort() {
+    return tunnelDatabasePort;
+  }
+
+  /**
+   * From the RSA format private key string, use bouncycastle to deserialize the key pair, reconstruct
+   * the keys from the key info, and return the key pair for use in authentication.
+   *
+   * @return
+   * @throws IOException
+   */
+  private KeyPair getPrivateKeyPair() throws IOException {
+    PEMParser pemParser = new PEMParser(new StringReader(getSSHKey()));
+    PEMKeyPair keypair = (PEMKeyPair) pemParser.readObject();
+    JcaPEMKeyConverter converter = new JcaPEMKeyConverter();
+    return new KeyPair(
+        (RSAPublicKey) converter.getPublicKey(SubjectPublicKeyInfo.getInstance(keypair.getPublicKeyInfo())),
+        (RSAPrivateKey) converter.getPrivateKey(keypair.getPrivateKeyInfo()));
+  }
+
+  /**
+   * Generates a new ssh client and returns it, with forwarding set to accept all types; use this
+   * before opening a tunnel.
+   *
+   * @return
+   */
+  private SshClient createClient() {
+    java.security.Security.addProvider(
+        new org.bouncycastle.jce.provider.BouncyCastleProvider());
+    SshClient client = SshClient.setUpDefaultClient();
+    client.setForwardingFilter(AcceptAllForwardingFilter.INSTANCE);
+    client.setServerKeyVerifier(AcceptAllServerKeyVerifier.INSTANCE);
+    return client;
+  }
+
+  private void validate() {
+    if (getHost() == null) {
+      throw new RuntimeException("SSH Tunnel host is null - verify configuration before starting tunnel!");
+    }
+  }
+
+  /**
+   * Starts an ssh session; wrap this in a try-finally and use closeTunnel() to close it.
+   *
+   * @return
+   * @throws IOException
+   * @throws InvalidKeySpecException
+   * @throws NoSuchAlgorithmException
+   * @throws URISyntaxException
+   */
+  private ClientSession openTunnel(SshClient client) {
+    try {
+      validate();
+      client.start();
+      ClientSession session = client.connect(
+          getUser().trim(),
+          getHost().trim(),
+          Integer.parseInt(
+              getTunnelSshPort().trim()))
+          .verify(TIMEOUT_MILLIS)
+          .getSession();
+      if (getMethod().equals("SSH_KEY_AUTH")) {
+        session.addPublicKeyIdentity(getPrivateKeyPair());
+      }
+      if (getMethod().equals("SSH_PASSWORD_AUTH")) {
+        session.addPasswordIdentity(getPassword());
+      }
+      session.auth().verify(TIMEOUT_MILLIS);
+      SshdSocketAddress address = session.startLocalPortForwarding(
+          new SshdSocketAddress(SshdSocketAddress.LOCALHOST_ADDRESS.getHostName(), Integer.parseInt(getTunnelDatabasePort().trim())),
+          new SshdSocketAddress(getRemoteDatabaseHost().trim(), Integer.parseInt(getRemoteDatabasePort().trim())));
+      LOGGER.info("Established tunneling session.  Port forwarding started on " + address.toInetSocketAddress());
+      return session;
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  @Override
+  public String toString() {
+    return "SSHTunnel{" +
+        "method='" + method + '\'' +
+        ", host='" + host + '\'' +
+        ", tunnelSshPort='" + tunnelSshPort + '\'' +
+        ", user='" + user + '\'' +
+        ", remoteDatabaseHost='" + remoteDatabaseHost + '\'' +
+        ", remoteDatabasePort='" + remoteDatabasePort + '\'' +
+        ", tunnelDatabasePort='" + tunnelDatabasePort + '\'' +
+        '}';
+  }
+
+}

airbyte-integrations/connectors/destination-jdbc/build.gradle

@@ -11,6 +11,7 @@ application {
 dependencies {
     implementation 'com.google.cloud:google-cloud-storage:1.113.16'
     implementation 'com.google.auth:google-auth-library-oauth2-http:0.25.5'
+    implementation group: 'org.apache.sshd', name: 'sshd-mina', version: '2.7.0'
 
     implementation project(':airbyte-db')
     implementation project(':airbyte-integrations:bases:base-java')

airbyte-integrations/connectors/destination-jdbc/src/main/java/io/airbyte/integrations/destination/jdbc/AbstractJdbcDestination.java

@@ -25,17 +25,22 @@
 package io.airbyte.integrations.destination.jdbc;
 
 import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.airbyte.commons.json.Jsons;
+import io.airbyte.commons.resources.MoreResources;
 import io.airbyte.db.Databases;
 import io.airbyte.db.jdbc.JdbcDatabase;
 import io.airbyte.db.jdbc.JdbcUtils;
 import io.airbyte.integrations.BaseConnector;
 import io.airbyte.integrations.base.AirbyteMessageConsumer;
 import io.airbyte.integrations.base.Destination;
+import io.airbyte.integrations.base.SSHTunnel;
 import io.airbyte.integrations.destination.NamingConventionTransformer;
 import io.airbyte.protocol.models.AirbyteConnectionStatus;
 import io.airbyte.protocol.models.AirbyteConnectionStatus.Status;
 import io.airbyte.protocol.models.AirbyteMessage;
 import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
+import java.io.IOException;
 import java.util.UUID;
 import java.util.function.Consumer;
 import org.slf4j.Logger;
@@ -57,6 +62,13 @@ protected NamingConventionTransformer getNamingResolver() {
     return namingResolver;
   }
 
+  @Override
+  public JsonNode addToSpec(JsonNode spec) throws IOException {
+    ObjectNode propNode = (ObjectNode) spec.get("connectionSpecification").get("properties");
+    propNode.set("tunnel_method", Jsons.deserialize(MoreResources.readResource("ssh-tunnel-spec.json")));
+    return spec;
+  }
+
   protected SqlOperations getSqlOperations() {
     return sqlOperations;
   }
@@ -71,8 +83,9 @@ public AbstractJdbcDestination(final String driverClass,
 
   @Override
   public AirbyteConnectionStatus check(JsonNode config) {
-
+    SSHTunnel sshTunnel = SSHTunnel.getInstance(config);
     try (final JdbcDatabase database = getDatabase(config)) {
+      sshTunnel.openTunnelIfRequested();
       String outputSchema = namingResolver.getIdentifier(config.get("schema").asText());
       attemptSQLCreateAndDropTableOperations(outputSchema, database, namingResolver, sqlOperations);
       return new AirbyteConnectionStatus().withStatus(Status.SUCCEEDED);
@@ -81,6 +94,8 @@ public AirbyteConnectionStatus check(JsonNode config) {
       return new AirbyteConnectionStatus()
           .withStatus(Status.FAILED)
           .withMessage("Could not connect with provided configuration. \n" + e.getMessage());
+    } finally {
+      sshTunnel.closeTunnel();
     }
   }
 
@@ -114,7 +129,8 @@ protected JdbcDatabase getDatabase(JsonNode config) {
 
   @Override
   public AirbyteMessageConsumer getConsumer(JsonNode config, ConfiguredAirbyteCatalog catalog, Consumer<AirbyteMessage> outputRecordCollector) {
-    return JdbcBufferedConsumerFactory.create(outputRecordCollector, getDatabase(config), sqlOperations, namingResolver, config, catalog);
+    return JdbcBufferedConsumerFactory
+        .create(outputRecordCollector, getDatabase(config), SSHTunnel.getInstance(config), sqlOperations, namingResolver, config, catalog);
   }
 
 }