🎉 New SFTP source connector (#13120)

* 11797 New SFTP source
* 11797 Bump sftp source version

* auto-bump connector version

Co-authored-by: Octavia Squidington III <octavia-squidington-iii@users.noreply.github.com>

airbyte-config/init/src/main/resources/seed/source_definitions.yaml

@@ -1051,3 +1051,10 @@
   documentationUrl: https://docs.airbyte.com/integrations/sources/zoho-crm
   sourceType: api
   releaseStage: alpha
+- name: SFTP
+  sourceDefinitionId: a827c52e-791c-4135-a245-e233c5255199
+  dockerRepository: airbyte/source-sftp
+  dockerImageTag: 0.1.1
+  documentationUrl: https://docs.airbyte.com/integrations/sources/sftp
+  sourceType: api
+  releaseStage: alpha

airbyte-config/init/src/main/resources/seed/source_specs.yaml

@@ -9986,3 +9986,107 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
+- dockerImage: "airbyte/source-sftp:0.1.1"
+  spec:
+    documentationUrl: "https://docs.airbyte.io/integrations/source/sftp"
+    connectionSpecification:
+      $schema: "http://json-schema.org/draft-07/schema#"
+      title: "SFTP Source Spec"
+      type: "object"
+      required:
+      - "user"
+      - "host"
+      - "port"
+      additionalProperties: true
+      properties:
+        user:
+          title: "User Name"
+          description: "The server user"
+          type: "string"
+          order: 0
+        host:
+          title: "Host Address"
+          description: "The server host address"
+          type: "string"
+          examples:
+          - "www.host.com"
+          - "192.0.2.1"
+          order: 1
+        port:
+          title: "Port"
+          description: "The server port"
+          type: "integer"
+          default: 22
+          examples:
+          - "22"
+          order: 2
+        credentials:
+          type: "object"
+          title: "Authentication *"
+          description: "The server authentication method"
+          order: 3
+          oneOf:
+          - title: "Password Authentication"
+            required:
+            - "auth_method"
+            - "auth_user_password"
+            properties:
+              auth_method:
+                description: "Connect through password authentication"
+                type: "string"
+                const: "SSH_PASSWORD_AUTH"
+                order: 0
+              auth_user_password:
+                title: "Password"
+                description: "OS-level password for logging into the jump server host"
+                type: "string"
+                airbyte_secret: true
+                order: 1
+          - title: "SSH Key Authentication"
+            required:
+            - "auth_method"
+            - "auth_ssh_key"
+            properties:
+              auth_method:
+                description: "Connect through ssh key"
+                type: "string"
+                const: "SSH_KEY_AUTH"
+                order: 0
+              auth_ssh_key:
+                title: "SSH Private Key"
+                description: "OS-level user account ssh key credentials in RSA PEM\
+                  \ format ( created with ssh-keygen -t rsa -m PEM -f myuser_rsa )"
+                type: "string"
+                airbyte_secret: true
+                multiline: true
+                order: 1
+        file_types:
+          title: "File types"
+          description: "Coma separated file types. Currently only 'csv' and 'json'\
+            \ types are supported."
+          type: "string"
+          default: "csv,json"
+          order: 4
+          examples:
+          - "csv,json"
+          - "csv"
+        folder_path:
+          title: "Folder Path (Optional)"
+          description: "The directory to search files for sync"
+          type: "string"
+          default: ""
+          examples:
+          - "/logs/2022"
+          order: 5
+        file_pattern:
+          title: "File Pattern (Optional)"
+          description: "The regular expression to specify files for sync in a chosen\
+            \ Folder Path"
+          type: "string"
+          default: ""
+          examples:
+          - "log-([0-9]{4})([0-9]{2})([0-9]{2}) - This will filter files which  `log-yearmmdd`"
+          order: 6
+    supportsNormalization: false
+    supportsDBT: false
+    supported_destination_sync_modes: []

airbyte-integrations/connectors/source-sftp/.dockerignore

@@ -0,0 +1,3 @@
+*
+!Dockerfile
+!build

airbyte-integrations/connectors/source-sftp/Dockerfile

@@ -0,0 +1,18 @@
+FROM airbyte/integration-base-java:dev AS build
+
+WORKDIR /airbyte
+ENV APPLICATION source-sftp
+
+COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar
+
+RUN tar xf ${APPLICATION}.tar --strip-components=1 && rm -rf ${APPLICATION}.tar
+
+FROM airbyte/integration-base-java:dev
+
+WORKDIR /airbyte
+ENV APPLICATION source-sftp
+
+COPY --from=build /airbyte /airbyte
+
+LABEL io.airbyte.version=0.1.1
+LABEL io.airbyte.name=airbyte/source-sftp

airbyte-integrations/connectors/source-sftp/README.md

@@ -0,0 +1,68 @@
+# Source Sftp
+
+This is the repository for the Sftp source connector in Java.
+For information about how to use this connector within Airbyte, see [the User Documentation](https://docs.airbyte.io/integrations/source/sftp).
+
+## Local development
+
+#### Building via Gradle
+From the Airbyte repository root, run:
+```
+./gradlew :airbyte-integrations:connectors:source-sftp:build
+```
+
+#### Create credentials
+**If you are a community contributor**, generate the necessary credentials and place them in `secrets/config.json` conforming to the spec file in `src/main/resources/spec.json`.
+Note that the `secrets` directory is git-ignored by default, so there is no danger of accidentally checking in sensitive information.
+
+**If you are an Airbyte core member**, follow the [instructions](https://docs.airbyte.io/connector-development#using-credentials-in-ci) to set up the credentials.
+
+### Locally running the connector docker image
+
+#### Build
+Build the connector image via Gradle:
+```
+./gradlew :airbyte-integrations:connectors:source-sftp:airbyteDocker
+```
+When building via Gradle, the docker image name and tag, respectively, are the values of the `io.airbyte.name` and `io.airbyte.version` `LABEL`s in
+the Dockerfile.
+
+#### Run
+Then run any of the connector commands as follows:
+```
+docker run --rm airbyte/source-sftp:dev spec
+docker run --rm -v $(pwd)/secrets:/secrets airbyte/source-sftp:dev check --config /secrets/config.json
+docker run --rm -v $(pwd)/secrets:/secrets airbyte/source-sftp:dev discover --config /secrets/config.json
+docker run --rm -v $(pwd)/secrets:/secrets -v $(pwd)/integration_tests:/integration_tests airbyte/source-sftp:dev read --config /secrets/config.json --catalog /integration_tests/configured_catalog.json
+```
+
+## Testing
+We use `JUnit` for Java tests.
+
+### Unit and Integration Tests
+Place unit tests under `src/test/io/airbyte/integrations/source/sftp`.
+
+#### Acceptance Tests
+Airbyte has a standard test suite that all source connectors must pass. Implement the `TODO`s in
+`src/test-integration/java/io/airbyte/integrations/source/sftpSourceAcceptanceTest.java`.
+
+### Using gradle to run tests
+All commands should be run from airbyte project root.
+To run unit tests:
+```
+./gradlew :airbyte-integrations:connectors:source-sftp:unitTest
+```
+To run acceptance and custom integration tests:
+```
+./gradlew :airbyte-integrations:connectors:source-sftp:integrationTest
+```
+
+## Dependency Management
+
+### Publishing a new version of the connector
+You've checked out the repo, implemented a million dollar feature, and you're ready to share your changes with the world. Now what?
+1. Make sure your changes are passing unit and integration tests.
+1. Bump the connector version in `Dockerfile` -- just increment the value of the `LABEL io.airbyte.version` appropriately (we use [SemVer](https://semver.org/)).
+1. Create a Pull Request.
+1. Pat yourself on the back for being an awesome contributor.
+1. Someone from Airbyte will take a look at your PR and iterate with you to merge it into master.

airbyte-integrations/connectors/source-sftp/build.gradle

@@ -0,0 +1,22 @@
+plugins {
+    id 'application'
+    id 'airbyte-docker'
+    id 'airbyte-integration-test-java'
+}
+
+application {
+    mainClass = 'io.airbyte.integrations.source.sftp.SftpSource'
+}
+
+dependencies {
+    implementation project(':airbyte-config:models')
+    implementation project(':airbyte-protocol:models')
+    implementation project(':airbyte-integrations:bases:base-java')
+    implementation files(project(':airbyte-integrations:bases:base-java').airbyteDocker.outputs)
+    implementation 'com.fasterxml.jackson.dataformat:jackson-dataformat-csv:2.13.2'
+    implementation 'com.jcraft:jsch:0.1.55'
+
+    integrationTestJavaImplementation project(':airbyte-integrations:bases:standard-source-test')
+    integrationTestJavaImplementation project(':airbyte-integrations:connectors:source-sftp')
+    integrationTestJavaImplementation "org.testcontainers:testcontainers:1.15.3"
+}

airbyte-integrations/connectors/source-sftp/src/main/java/io/airbyte/integrations/source/sftp/SftpClient.java

@@ -0,0 +1,125 @@
+package io.airbyte.integrations.source.sftp;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.jcraft.jsch.Channel;
+import com.jcraft.jsch.ChannelSftp;
+import com.jcraft.jsch.JSch;
+import com.jcraft.jsch.JSchException;
+import com.jcraft.jsch.Session;
+import com.jcraft.jsch.SftpException;
+import io.airbyte.integrations.source.sftp.enums.SftpAuthMethod;
+import io.airbyte.integrations.source.sftp.enums.SupportedFileExtension;
+import org.apache.commons.io.IOUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.Properties;
+import java.util.Vector;
+
+public class SftpClient {
+
+    protected static final Logger LOGGER = LoggerFactory.getLogger(SftpClient.class);
+    private static final String CHANNEL_SFTP = "sftp";
+    private static final String STRICT_HOST_KEY_CHECKING = "StrictHostKeyChecking";
+
+    private final String username;
+    private final String hostAddress;
+    private final int port;
+    private final SftpAuthMethod authMethod;
+    private final JsonNode config;
+    private final int connectionTimeoutMillis = 60000;
+    private final JSch jsch;
+    private Session session;
+    private ChannelSftp channelSftp;
+
+    public SftpClient(JsonNode config) {
+        this.config = config;
+        jsch = new JSch();
+        username = config.has("user") ? config.get("user").asText() : "";
+        hostAddress = config.has("host") ? config.get("host").asText() : "";
+        port = config.has("port") ? config.get("port").asInt() : 22;
+        authMethod = SftpAuthMethod.valueOf(config.get("credentials").get("auth_method").asText());
+    }
+
+    public void connect() {
+        try {
+            LOGGER.info("Connecting to the server");
+            configureSession();
+            configureAuthMethod();
+            LOGGER.debug("Connecting to host: {} at port: {}", hostAddress, port);
+            session.connect();
+            Channel channel = session.openChannel(CHANNEL_SFTP);
+            channel.connect();
+
+            channelSftp = (ChannelSftp) channel;
+            LOGGER.info("Connected successfully");
+        } catch (Exception e) {
+            LOGGER.error("Exception attempting to connect to the server:", e);
+            throw new RuntimeException(e);
+        }
+    }
+
+    private void configureSession() throws JSchException {
+        Properties properties = new Properties();
+        properties.put(STRICT_HOST_KEY_CHECKING, "no");
+        session = jsch.getSession(username, hostAddress, port);
+        session.setConfig(properties);
+        session.setTimeout(connectionTimeoutMillis);
+    }
+
+    public void disconnect() {
+        LOGGER.info("Disconnecting from the server");
+        if (channelSftp != null) {
+            channelSftp.disconnect();
+        }
+        if (session != null) {
+            session.disconnect();
+        }
+        LOGGER.info("Disconnected successfully");
+    }
+
+    public boolean isConnected() {
+        return channelSftp != null && channelSftp.isConnected();
+    }
+
+    public Vector lsFile(SupportedFileExtension fileExtension) {
+        try {
+            return channelSftp.ls("*." + fileExtension.typeName);
+        } catch (SftpException e) {
+            LOGGER.error("Exception occurred while trying to find files with type {} : ", fileExtension, e);
+            throw new RuntimeException(e);
+        }
+    }
+
+    public void changeWorkingDirectory(String path) throws SftpException {
+        channelSftp.cd(path);
+    }
+
+    public ByteArrayInputStream getFile(String fileName) {
+        try (InputStream inputStream = channelSftp.get(fileName)) {
+            return new ByteArrayInputStream(IOUtils.toByteArray(inputStream));
+        } catch (Exception e) {
+            LOGGER.error("Exception occurred while trying to download file {} : ", fileName, e);
+            throw new RuntimeException(e);
+        }
+    }
+
+    private void configureAuthMethod() throws Exception {
+        switch (authMethod) {
+            case SSH_PASSWORD_AUTH -> session.setPassword(config.get("credentials").get("auth_user_password").asText());
+            case SSH_KEY_AUTH -> {
+                File tempFile = File.createTempFile("private_key", "", null);
+                FileOutputStream fos = new FileOutputStream(tempFile);
+                fos.write(config.get("credentials").get("auth_ssh_key").asText().getBytes(StandardCharsets.UTF_8));
+                jsch.addIdentity(tempFile.getAbsolutePath());
+            }
+            default -> throw new RuntimeException("Unsupported SFTP Authentication type");
+        }
+    }
+
+}