WIP: Airbyte-jenny/connector ssh tunnels 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/base/SSHTunnel.java

@@ -0,0 +1,222 @@
+/*
+ * 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 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;
+
+  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 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
+   */
+  protected 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
+   */
+  public 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
+   */
+  public ClientSession openTunnel(SshClient client) throws IOException {
+    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;
+  }
+
+  public void closeTunnel(SshClient client, ClientSession session) throws IOException {
+    if (session != null) {
+      session.close();
+    }
+    if (client != null) {
+      client.stop();
+    }
+  }
+
+  @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

@@ -31,6 +31,7 @@
 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;
@@ -40,6 +41,8 @@
 import java.util.function.Consumer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.apache.sshd.client.SshClient;
+import org.apache.sshd.client.session.ClientSession;
 
 public abstract class AbstractJdbcDestination extends BaseConnector implements Destination {
 
@@ -71,8 +74,15 @@ public AbstractJdbcDestination(final String driverClass,
 
   @Override
   public AirbyteConnectionStatus check(JsonNode config) {
-
+    SSHTunnel tunnelConfig = null;
+    SshClient sshclient = null;
+    ClientSession tunnelSession = null;
     try (final JdbcDatabase database = getDatabase(config)) {
+      tunnelConfig = getSSHTunnelConfig(config);
+      if (tunnelConfig.shouldTunnel()) {
+        sshclient = tunnelConfig.createClient();
+        tunnelSession = tunnelConfig.openTunnel(sshclient);
+      }
       String outputSchema = namingResolver.getIdentifier(config.get("schema").asText());
       attemptSQLCreateAndDropTableOperations(outputSchema, database, namingResolver, sqlOperations);
       return new AirbyteConnectionStatus().withStatus(Status.SUCCEEDED);
@@ -81,6 +91,16 @@ public AirbyteConnectionStatus check(JsonNode config) {
       return new AirbyteConnectionStatus()
           .withStatus(Status.FAILED)
           .withMessage("Could not connect with provided configuration. \n" + e.getMessage());
+    } finally {
+      if (tunnelConfig.shouldTunnel()) {
+        try {
+          if (sshclient != null) {
+            tunnelConfig.closeTunnel(sshclient, tunnelSession);
+          }
+        } catch (Throwable t) {
+          t.printStackTrace();
+        }
+      }
     }
   }
 
@@ -110,11 +130,33 @@ protected JdbcDatabase getDatabase(JsonNode config) {
         driverClass);
   }
 
+
+
+  protected SSHTunnel getSSHTunnelConfig(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;
+  }
+
+  private String getConfigValueOrNull(JsonNode config, String key) {
+    return config != null && config.has(key) ? config.get(key).asText() : null;
+  }
+
   public abstract JsonNode toJdbcConfig(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), getSSHTunnelConfig(config), sqlOperations, namingResolver, config, catalog);
   }
 
 }

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

@@ -30,6 +30,7 @@
 import io.airbyte.db.jdbc.JdbcDatabase;
 import io.airbyte.integrations.base.AirbyteMessageConsumer;
 import io.airbyte.integrations.base.AirbyteStreamNameNamespacePair;
+import io.airbyte.integrations.base.SSHTunnel;
 import io.airbyte.integrations.destination.NamingConventionTransformer;
 import io.airbyte.integrations.destination.buffered_stream_consumer.BufferedStreamConsumer;
 import io.airbyte.integrations.destination.buffered_stream_consumer.OnCloseFunction;
@@ -67,6 +68,7 @@ public class JdbcBufferedConsumerFactory {
 
   public static AirbyteMessageConsumer create(Consumer<AirbyteMessage> outputRecordCollector,
                                               JdbcDatabase database,
+                                              SSHTunnel sshTunnel,
                                               SqlOperations sqlOperations,
                                               NamingConventionTransformer namingResolver,
                                               JsonNode config,
@@ -75,9 +77,9 @@ public static AirbyteMessageConsumer create(Consumer<AirbyteMessage> outputRecor
 
     return new BufferedStreamConsumer(
         outputRecordCollector,
-        onStartFunction(database, sqlOperations, writeConfigs),
+        onStartFunction(database, sshTunnel, sqlOperations, writeConfigs),
         recordWriterFunction(database, sqlOperations, writeConfigs, catalog),
-        onCloseFunction(database, sqlOperations, writeConfigs),
+        onCloseFunction(database, sshTunnel, sqlOperations, writeConfigs),
         catalog,
         sqlOperations::isValidData,
         MAX_BATCH_SIZE);
@@ -134,7 +136,7 @@ private static String getOutputSchema(AirbyteStream stream, String defaultDestSc
     return defaultDestSchema;
   }
 
-  private static OnStartFunction onStartFunction(JdbcDatabase database, SqlOperations sqlOperations, List<WriteConfig> writeConfigs) {
+  private static OnStartFunction onStartFunction(JdbcDatabase database, SSHTunnel sshTunnel, SqlOperations sqlOperations, List<WriteConfig> writeConfigs) {
     return () -> {
       LOGGER.info("Preparing tmp tables in destination started for {} streams", writeConfigs.size());
       for (final WriteConfig writeConfig : writeConfigs) {
@@ -168,7 +170,7 @@ private static RecordWriter recordWriterFunction(JdbcDatabase database,
     };
   }
 
-  private static OnCloseFunction onCloseFunction(JdbcDatabase database, SqlOperations sqlOperations, List<WriteConfig> writeConfigs) {
+  private static OnCloseFunction onCloseFunction(JdbcDatabase database, SSHTunnel sshTunnel, SqlOperations sqlOperations, List<WriteConfig> writeConfigs) {
     return (hasFailed) -> {
       // copy data
       if (!hasFailed) {