🎉 New SFTP source connector

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.0
+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");
+        }
+    }
+
+}

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

@@ -0,0 +1,133 @@
+package io.airbyte.integrations.source.sftp;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.jcraft.jsch.ChannelSftp;
+import com.jcraft.jsch.ChannelSftp.LsEntry;
+import com.jcraft.jsch.SftpException;
+import io.airbyte.integrations.source.sftp.enums.SupportedFileExtension;
+import io.airbyte.integrations.source.sftp.parsers.SftpFileParser;
+import io.airbyte.integrations.source.sftp.parsers.SftpFileParserFactory;
+import io.airbyte.integrations.source.sftp.util.JsonSchemaGenerator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.Vector;
+import org.apache.commons.io.FilenameUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayInputStream;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+
+public class SftpCommand {
+
+    protected static final Logger LOGGER = LoggerFactory.getLogger(SftpClient.class);
+
+    private static final String FILE_TYPE_SEPARATOR = ",";
+    private final SftpClient client;
+    private final Set<SupportedFileExtension> selectedFileExtensions;
+    private final Pattern filePattern;
+    private final SftpFileParserFactory sftpFileParserFactory;
+
+    public SftpCommand(SftpClient client, JsonNode config) {
+        this.client = client;
+        sftpFileParserFactory = new SftpFileParserFactory();
+        String commaSeparatedFileExtension = config.has("file_type") ? config.get("file_type").asText() : "";
+        Set<String> selectedFileExtension = Set.of(commaSeparatedFileExtension.split(FILE_TYPE_SEPARATOR));
+        selectedFileExtensions = selectedFileExtension.stream()
+                .map(this::transformFileExtension)
+                .collect(Collectors.toSet());
+        filePattern = Pattern.compile(config.has("file_pattern") ? config.get("file_pattern").asText() : "");
+    }
+
+    private SupportedFileExtension transformFileExtension(String fileExtension) {
+        try {
+            return SupportedFileExtension.valueOf(fileExtension.toUpperCase());
+        } catch (Exception e) {
+            LOGGER.error("Unsupported file extension : {}", fileExtension);
+            throw new RuntimeException();
+        }
+    }
+
+    public void tryChangeWorkingDirectory(String path) {
+        try {
+            checkIfConnected();
+            client.changeWorkingDirectory(FilenameUtils.normalize(path));
+            LOGGER.info("Working directory set to : {}", path);
+        } catch (SftpException e) {
+            LOGGER.error("Could not find path : {} : ", path, e);
+            throw new RuntimeException(e);
+        }
+    }
+
+    public Map<String, JsonNode> getFilesSchemas() {
+        checkIfConnected();
+        Map<String, JsonNode> fileSchemas = new HashMap<>();
+        Set<String> fileNames = getFileNames();
+        LOGGER.info("Found file for sync : {}", fileNames);
+        fileNames.forEach(fileName -> {
+            ByteArrayInputStream file = client.getFile(fileName);
+            JsonNode parsedFile = tryGetFirstNode(file, fileName);
+            JsonNode schema = tryGetSchema(parsedFile, fileName);
+            fileSchemas.put(fileName, schema);
+        });
+        return fileSchemas;
+    }
+
+    private Set<String> getFileNames() {
+        checkIfConnected();
+        Vector<LsEntry> entries = new Vector<>();
+        for (SupportedFileExtension fileExtension : selectedFileExtensions) {
+            entries.addAll(client.lsFile(fileExtension));
+        }
+        return entries.stream()
+                .map(ChannelSftp.LsEntry::getFilename)
+                .filter(filePattern.asPredicate())
+                .collect(Collectors.toSet());
+    }
+
+    private JsonNode tryGetSchema(JsonNode jsonNode, String fileName) {
+        try {
+            return JsonSchemaGenerator.getJsonSchema(jsonNode);
+        } catch (Exception e) {
+            LOGGER.error("Exception occurred while trying to parse schema of file {} : ", fileName, e);
+            throw new RuntimeException(e);
+        }
+    }
+
+    public JsonNode tryGetFirstNode(ByteArrayInputStream file, String fileName) {
+        try {
+            String extension = FilenameUtils.getExtension(fileName);
+            SftpFileParser parser = sftpFileParserFactory.create(transformFileExtension(extension));
+            return parser.parseFileFirstEntity(file);
+        } catch (Exception e) {
+            LOGGER.error("Exception occurred while trying to parse file {} : ", fileName, e);
+            throw new RuntimeException();
+        }
+    }
+
+    public List<JsonNode> getFileData(String fileName) {
+        ByteArrayInputStream file = client.getFile(fileName);
+        return tryParseFile(file, fileName);
+    }
+
+    private List<JsonNode> tryParseFile(ByteArrayInputStream file, String fileName) {
+        try {
+            String extension = FilenameUtils.getExtension(fileName);
+            SftpFileParser parser = sftpFileParserFactory.create(transformFileExtension(extension));
+            return parser.parseFile(file);
+        } catch (Exception e) {
+            LOGGER.error("Exception occurred while trying to parse file {} : ", fileName, e);
+            throw new RuntimeException();
+        }
+    }
+
+    private void checkIfConnected() {
+        if (!client.isConnected()) {
+            LOGGER.info("SFTP client is not connected.");
+            client.connect();
+        }
+    }
+}