Add embedded sftp server

.gitlab-ci.yml

@@ -8,10 +8,8 @@ stages:
 test:
   stage: test
   image: openjdk:8
-  services:
-    - swissbib/sftp-test-env:latest
   script:
-    - ./gradlew --stacktrace --debug --no-daemon test --fail-fast --tests "org.memobase.Tests"
+    - ./gradlew --no-daemon test --fail-fast --tests "org.memobase.Tests"
 
 .build-image:
   stage: publish

build.gradle

@@ -69,6 +69,13 @@ dependencies {
     // MOCK KAFKA STREAMS
     // https://mvnrepository.com/artifact/org.apache.kafka/kafka-streams-test-utils
     //testCompile group: 'org.apache.kafka', name: 'kafka-streams-test-utils', version: kafkaV
+
+    // https://mvnrepository.com/artifact/com.github.marschall/memoryfilesystem
+    testCompile group: 'com.github.marschall', name: 'memoryfilesystem', version: '2.1.0'
+    // https://mvnrepository.com/artifact/org.apache.sshd/sshd-core
+    testCompile group: 'org.apache.sshd', name: 'sshd-core', version: '2.4.0'
+    // https://mvnrepository.com/artifact/org.apache.sshd/sshd-sftp
+    testCompile group: 'org.apache.sshd', name: 'sshd-sftp', version: '2.4.0'
 }
 
 compileKotlin {

docker-compose.yml

-sftp:
-  image: swissbib/sftp-test-env:latest
-  ports:
-    - "22:22"
\ No newline at end of file

env/Dockerfile

-FROM atmoz/sftp:alpine
-
-COPY create_test_env.sh /etc/sftp.d/create_test_env.sh
-COPY ssh_host_rsa_key /etc/ssh/ssh_host_rsa_key
-COPY ssh_host_rsa_key.pub /etc/ssh/ssh_host_rsa_key.pub:ro
-COPY test.csv /home/user/sftp/test_institution_1/test_record_set_1/test.csv
-COPY users.conf /etc/sftp/users.conf
\ No newline at end of file

env/create_test_env.sh

-chown -R 744:744 /home/user/sftp/test_institution_1
\ No newline at end of file

env/ssh_host_rsa_key

------BEGIN RSA PRIVATE KEY-----
-MIIJKAIBAAKCAgEAp7ZwT1Z4rfN8Z1kKps5htZikaLP6g+f+w13rJbE+tH4HkTwh
-oAMylRsWCMWmyDAk1SDK6EUHPkCLxD7FYNLqZvjWrB7ljikbuBM3VAg0k+5bKGwv
-RAEUWRcP1ux3hRdWENvkhgogUEAhojMSwFOZXPhOPlt4bcFnBiCccBjwiZiYrGUX
-/TnSKFJ7R9BM9xPA82kB17R0/BEoT7OM7YOYp8kK0eJBJso97Scb8PeWinj3rNF2
-OZNZFALDqiL7FiiRodKEmcf/AaUXcjSvPXlM2fX0C1Qf9UyyEOGN1jVXFfiKJbtd
-Sxh+5XneTNklfq+jr2HIFw9LgBIBUPekYIRnnWGBdjW3xbIM86VQg2ll86148qqV
-Ebvdb5th7RWh2zEXeSZ/85rx0H78FkeC6yLyMrVyoWJ2ROcdCLzxJJexyZebojFH
-QcNzBSoODl1wEwEQNwxunpKAj4nDWHMLtOE7TGXCAMOJJFgaAsExr19qywkv7ADE
-TESoC+LM77dT8QLSPdDCUSmzoxKr1LSSaE6WE109hO5rEaPgsfB2jWEN8dUenyBg
-jJGCAAw1tZndgEpkI+Ur4mKaknbibkxD5/5tlaEMK5HPMSyGk9y5hN0H7A9zNSRb
-6/JdoFk/ouYotb3YlEjMvA9fiOKWaiotwrqgQSmJysm3y8j+ik/b+Qcjmd0CAwEA
-AQKCAgAdw6wU+IS8Yn0OnmfciL4gi3GKP37TUCYPqChmbRly0Pco2GIjUs30VnVH
-o0RhPOIcjRBVvoJ1kuD7OCSxdV0yXzFCJM2auEL6HTbxi+9A3cmB2AlfaU2iLgya
-mSbxEN4uacgZ3cw0Ud1mOug2B+As9sh0Gqm52Nwe40TARDzUPqfB2MM0JsHcdB76
-9MXkB7ZzvIweKyGj5quf0X7OXE+IX46zBxOG/deVsh4sWtR1yNgz7Iyt1S+29HKj
-TEgBe7u023a7EBp6wUDy/NSo9VElUZg5Nnnhf/YONumEPb0x4gUqgyEghb1nnoRM
-YTFdVXCYXKM1LHq6xPdE1l5u3rDAnOoKHNeVhCXNuYlpCwZXjlPZqR24QjSlT6t8
-+mfNFaRcFwnnT3CQrRFnstouRNy647kGYRVUviKY4ruRKuOOTthvZeEwat+NE01u
-howUq3MTNIPPMjk2cM1XimQ+vOlL/LOD7LEXvh50A47Osc3Z9dbxw61Gm4h/fFkX
-Bk2obFokM5pDIjFyJmJQ77B94EqM3axFlpQuuziGKFwI0ewNmY4CKc8zIF3DbPBs
-n89ZLs8K6+y5GxTRIZaqVdiOSLahJgzPcvPpa5nEtaVGVafqOq4ZRbfT5MMqjB+e
-N2MO9XdhFTZfnfCpkFopwn0NXX7M/fAfUjhSyhSwCqQ8bDCGxQKCAQEA3YoSZ/1t
-ghfpWnVRUxWrSa7hO96kqE6CF7bqbIj9C+BfKpJdeGbrhAcfuQJnzAe38w4DfFdk
-MzLWU98s5VzlAxaCnTPzw72tYjt9rLxlWKCmYb/eIMHdYzqXYoMYCW/NX4BIR1mg
-Lh7RWy125YiVphDAfoj7vKA8IqNswYRaJARMtCHqTWPTq2+FWTuFhmqhUIWr773T
-3StjbX2M7pECYQb8ReFLB8GmeEJxG18aWZY8cr+il9wnCZ2XJGYyQ9ktYqajy6cY
-GWc92VOBieDZ4/itZmV4jpNw5bdLZpNrz/1U0LDvxipj7/EPQEzn0WWNtuaFekg9
-WAgBe5nXi2fVkwKCAQEAwczug6BdV/bvliwGshIoSMYg6AYuvsyiAzihpGROyJY2
-lqjXUkkJnP5pHSFh7Vr08yAY0rxuITWw9iUvQx7U5mzoJmLMfWMR8hIiYDgstfpH
-2Bx6ihRBuO2vZXQYqyhWU+W/nKDQn2EH7l+OoeLUe7svw9fSjKwd6XwKmZ4M7ro1
-ZJbr1ftqbKlt4n6JI62yteg4XjpRy+tEaVk/h30OUgHaMOajmlPSjV9edf+ZERaW
-aAUWQ1xyoyHYkzvjcB8NWVDftbUFTF3cJ++IZlFmJb+v6iKGCpp15j7uM6RhUHn5
-Fve/xv1LIpcLy3bmyNOjhKlcoi4CjtH+tU5SsEt4zwKCAQEAhpwXlQIi4PJAgwtX
-z8ER4+KTzrn4mJ+jYl9tT2dpQiciDA8FJlx67C0b9GFmyk1CUzgHnCzJoGZaXnnz
-oKXyLQ8na/eePShqSo7VTPjoJ5LtpeVcRdEmAN4gD+aR22IIiue6g0gNERj+ooUc
-glmcfFwfaoM3WqSOBYoBUhBmaQ4HwUf+QunOOpO3lcGZ31O5EuE12KUiL3fmoSex
-U1/e7y+8Z4V9/oeG6/mLGlOOAjNMJXkVhWpqeeg8ZwyFrD4w2olgayTrerwFk6Cu
-zCVIn8GBMv+i4hbqeVoHQZt/3dATEf8Aogst0CRL3QkdrlkjY6fsIKH2TCAJLp4K
-nxUHawKCAQAi+g2CDAtMuPB8te1vbf9/QuLlfVgqb1w+IJZryP6/DP1FK6vQ2gW2
-I+RssX2vDN9wkbZpMkDeFYaepg9lmcbq33T2mJY3ew3eFo/Ftd276jPVOS6UhRtn
-eN5S/SUGnv0Vnz150zxTx3ta9jwT05Bt1FbGjckeQmITpaN0HiZPX6QLR4HA2ONY
-QSvn2NZ/bfX3BrZFq1jf6NIsAUOJ/HP9MQBkGvwj+kTh5vhxa3QAtYbntyNRfPnj
-n1QrHn/p2HDcUdBORyFxqu709jIz5TT+Ux44r4ppl3730xvCjkRR9fGSx5wBGe3Z
-jFFAo3D7hdbZNofVbWBgzl2d80jRMI3ZAoIBAFSumxeIzXEL4U37lvj6x5o2sFMr
-rHpN2JbOZPrIhyB1kbsNY97avutgXE+rNvwTF+jNeLBWqKOmQ8iFJoCuHFAzk1M6
-+q1ui6p8+o3oBspG4iNI7+YvApLpOemZYIjgVo8/+5E8LgCxwvrVAAB/Jochs665
-SFv3Hsk5Z8X6PcbVEZKOanFWQ7aVw97kth6WkC38hebfogOEuMGQm7Of3O7bjLHF
-lySVcv5SxUsbQm+fcnEE+AVndLB07SpWoTDFxiX9Sn+Mj4k9OXou1tpzgpSWWwH3
-yee/MV5ONjadGO5GWqLQGTvXbC7LAsJ/3eyhPpEno3PYBf/YOnUaRjjVKUE=
------END RSA PRIVATE KEY-----

env/ssh_host_rsa_key.pub

-ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAACAQCntnBPVnit83xnWQqmzmG1mKRos/qD5/7DXeslsT60fgeRPCGgAzKVGxYIxabIMCTVIMroRQc+QIvEPsVg0upm+NasHuWOKRu4EzdUCDST7lsobC9EARRZFw/W7HeFF1YQ2+SGCiBQQCGiMxLAU5lc+E4+W3htwWcGIJxwGPCJmJisZRf9OdIoUntH0Ez3E8DzaQHXtHT8EShPs4ztg5inyQrR4kEmyj3tJxvw95aKePes0XY5k1kUAsOqIvsWKJGh0oSZx/8BpRdyNK89eUzZ9fQLVB/1TLIQ4Y3WNVcV+Iolu11LGH7led5M2SV+r6OvYcgXD0uAEgFQ96RghGedYYF2NbfFsgzzpVCDaWXzrXjyqpURu91vm2HtFaHbMRd5Jn/zmvHQfvwWR4LrIvIytXKhYnZE5x0IvPEkl7HJl5uiMUdBw3MFKg4OXXATARA3DG6ekoCPicNYcwu04TtMZcIAw4kkWBoCwTGvX2rLCS/sAMRMRKgL4szvt1PxAtI90MJRKbOjEqvUtJJoTpYTXT2E7msRo+Cx8HaNYQ3x1R6fIGCMkYIADDW1md2ASmQj5SviYpqSduJuTEPn/m2VoQwrkc8xLIaT3LmE3QfsD3M1JFvr8l2gWT+i5ii1vdiUSMy8D1+I4pZqKi3CuqBBKYnKybfLyP6KT9v5ByOZ3Q== jonas@jonas

env/test.csv

-hello,stuff,,,,,
-,,,,,more,
\ No newline at end of file

env/users.conf

-user:password:744:744:sftp
\ No newline at end of file

src/test/kotlin/EmbeddedSftpServer.kt

+/*
+ * record-parser
+ * Copyright (C) 2020 Memoriav
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package org.memobase
+
+import com.github.marschall.memoryfilesystem.MemoryFileSystemBuilder.newLinux
+import java.io.Closeable
+import java.io.IOException
+import java.io.InputStream
+import java.nio.charset.Charset
+import java.nio.file.FileStore
+import java.nio.file.FileSystem
+import java.nio.file.Files
+import java.nio.file.Path
+import java.nio.file.PathMatcher
+import java.nio.file.WatchService
+import java.nio.file.attribute.UserPrincipalLookupService
+import java.nio.file.spi.FileSystemProvider
+import org.apache.sshd.server.SshServer
+import org.apache.sshd.server.keyprovider.SimpleGeneratorHostKeyProvider
+import org.apache.sshd.server.subsystem.sftp.SftpSubsystemFactory
+
+class EmbeddedSftpServer(port: Int, user: String, password: String) : Closeable {
+    private val credentials = mapOf(Pair(user, password))
+    private val fileSystem: FileSystem = newLinux().build("EmbeddedSftpServerFileSystem@" + hashCode())
+    private val server: SshServer = SshServer.setUpDefaultServer()
+
+    init {
+        server.port = port
+        server.keyPairProvider = SimpleGeneratorHostKeyProvider()
+        server.setPasswordAuthenticator { authUser, authPassword, _ -> authenticate(authUser, authPassword) }
+        server.subsystemFactories = listOf(SftpSubsystemFactory())
+        /* When a channel is closed SshServer calls close() on the file system.
+         * In order to use the file system for multiple channels/sessions we
+         * have to use a file system wrapper whose close() does nothing.
+         */
+        server.setFileSystemFactory { DoNotClose(fileSystem) }
+        server.start()
+    }
+
+    /**
+     * Put a text file on the SFTP folder. The file is available by the
+     * specified path.
+     * @param path the path to the file.
+     * @param content the files content.
+     * @param encoding the encoding of the file.
+     * @throws IOException if the file cannot be written.
+     */
+    @Throws(IOException::class)
+    fun putFile(
+        path: String,
+        content: String,
+        encoding: Charset = Charset.defaultCharset()
+    ) {
+        val contentAsBytes = content.toByteArray(encoding)
+        putFile(path, contentAsBytes)
+    }
+
+    /**
+     * Put a file on the SFTP folder. The file is available by the specified
+     * path.
+     * @param path the path to the file.
+     * @param content the files content.
+     * @throws IOException if the file cannot be written.
+     */
+    @Throws(IOException::class)
+    fun putFile(
+        path: String,
+        content: ByteArray
+    ) {
+        val pathAsObject = fileSystem.getPath(path)
+        ensureDirectoryOfPathExists(pathAsObject)
+        Files.write(pathAsObject, content)
+    }
+
+    /**
+     * Put a file on the SFTP folder. The file is available by the specified
+     * path. The file content is read from an `InputStream`.
+     * @param path the path to the file.
+     * @param inputStream an `InputStream` that provides the file's content.
+     * @throws IOException if the file cannot be written or the input stream
+     * cannot be read.
+     */
+    @Throws(IOException::class)
+    fun putFile(
+        path: String,
+        inputStream: InputStream
+    ) {
+        val pathAsObject = fileSystem.getPath(path)
+        ensureDirectoryOfPathExists(pathAsObject)
+        Files.copy(inputStream, pathAsObject)
+    }
+
+    /**
+     * Create a directory on the SFTP server.
+     * @param path the directory's path.
+     * @throws IOException if the directory cannot be created.
+     */
+    @Throws(IOException::class)
+    fun createDirectory(
+        path: String
+    ) {
+        val pathAsObject = fileSystem.getPath(path)
+        Files.createDirectories(pathAsObject)
+    }
+
+    /**
+     * Create multiple directories on the SFTP server.
+     * @param paths the directories' paths.
+     * @throws IOException if at least one directory cannot be created.
+     */
+    @Throws(IOException::class)
+    fun createDirectories(
+        vararg paths: String
+    ) {
+        for (path in paths) createDirectory(path)
+    }
+
+    /**
+     * Get a text file from the SFTP server.
+     * @param path the path to the file.
+     * @return the content of the text file.
+     * @throws IOException if the file cannot be read.
+     * @throws IllegalStateException if not called from within a test.
+     */
+    @Throws(IOException::class)
+    fun getFileContent(
+        path: String,
+        encoding: Charset = Charset.defaultCharset()
+    ): String {
+        return getFileContent(path).toString(encoding)
+    }
+
+    /**
+     * Get a file from the SFTP server.
+     * @param path the path to the file.
+     * @return the content of the file.
+     * @throws IOException if the file cannot be read.
+     * @throws IllegalStateException if not called from within a test.
+     */
+    @Throws(IOException::class)
+    fun getFileContent(
+        path: String
+    ): ByteArray {
+        val pathAsObject = fileSystem.getPath(path)
+        return Files.readAllBytes(pathAsObject)
+    }
+
+    /**
+     * Checks the existence of a file. returns `true` iff the file exists
+     * and it is not a directory.
+     * @param path the path to the file.
+     * @return `true` iff the file exists and it is not a directory.
+     * @throws IllegalStateException if not called from within a test.
+     */
+    fun existsFile(
+        path: String
+    ): Boolean {
+        val pathAsObject = fileSystem.getPath(path)
+        return Files.exists(pathAsObject) && !Files.isDirectory(pathAsObject)
+    }
+
+    private fun authenticate(
+        username: String,
+        password: String
+    ): Boolean {
+        return (credentials.isEmpty() ||
+                credentials[username] == password)
+    }
+
+    @Throws(IOException::class)
+    private fun ensureDirectoryOfPathExists(
+        path: Path
+    ) {
+        val directory = path.parent
+        if (directory != null && directory != path.root) Files.createDirectories(directory)
+    }
+
+    private class DoNotClose internal constructor(
+        val fileSystem: FileSystem
+    ) : FileSystem() {
+        override fun provider(): FileSystemProvider {
+            return fileSystem.provider()
+        }
+
+        @Throws(IOException::class)
+        override fun close() {
+            // will not be closed
+        }
+
+        override fun isOpen(): Boolean {
+            return fileSystem.isOpen
+        }
+
+        override fun isReadOnly(): Boolean {
+            return fileSystem.isReadOnly
+        }
+
+        override fun getSeparator(): String {
+            return fileSystem.separator
+        }
+
+        override fun getRootDirectories(): Iterable<Path> {
+            return fileSystem.rootDirectories
+        }
+
+        override fun getFileStores(): Iterable<FileStore> {
+            return fileSystem.fileStores
+        }
+
+        override fun supportedFileAttributeViews(): Set<String> {
+            return fileSystem.supportedFileAttributeViews()
+        }
+
+        override fun getPath(
+            first: String,
+            vararg more: String
+        ): Path {
+            return fileSystem.getPath(first, *more)
+        }
+
+        override fun getPathMatcher(
+            syntaxAndPattern: String
+        ): PathMatcher {
+            return fileSystem.getPathMatcher(syntaxAndPattern)
+        }
+
+        override fun getUserPrincipalLookupService(): UserPrincipalLookupService {
+            return fileSystem.userPrincipalLookupService
+        }
+
+        @Throws(IOException::class)
+        override fun newWatchService(): WatchService {
+            return fileSystem.newWatchService()
+        }
+    }
+
+    override fun close() {
+        this.fileSystem.close()
+        this.server.close()
+    }
+}

src/test/kotlin/EmbeddedSftpServerExtension.kt

+/*
+ * record-parser
+ * Copyright (C) 2020 Memoriav
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+package org.memobase
+
+import java.io.File
+import java.io.FileInputStream
+import org.junit.jupiter.api.extension.BeforeAllCallback
+import org.junit.jupiter.api.extension.ExtensionContext
+
+class EmbeddedSftpServerExtension : BeforeAllCallback {
+
+    lateinit var sftpServer: EmbeddedSftpServer
+
+    private val port: Int = 22000
+    private val user: String = "user"
+    private val password: String = "password"
+
+    override fun beforeAll(context: ExtensionContext?) {
+        sftpServer = EmbeddedSftpServer(port, user, password)
+        sftpServer.createDirectories("/memobase/test_institution_1/test_record_set_1")
+        sftpServer.putFile("/memobase/test_institution_1/test_record_set_1/test.csv", FileInputStream(File("src/test/resources/data/test.csv")))
+        val wrapper = SftpWrapper(sftpServer)
+        if (context != null) {
+            context.getStore(ExtensionContext.Namespace.create(EmbeddedSftpServerExtension::class.java))
+                .put("sftp", wrapper)
+            Runtime.getRuntime().addShutdownHook(Thread(Runnable {
+                wrapper.close()
+            }))
+        }
+    }
+
+    class SftpWrapper(private val embedded: EmbeddedSftpServer) : ExtensionContext.Store.CloseableResource {
+        override fun close() {
+            embedded.close()
+        }
+    }
+}

src/test/kotlin/Tests.kt

@@ -21,7 +21,7 @@ import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.TestInstance
 import org.junit.jupiter.api.extension.ExtendWith
 
-@ExtendWith(EmbeddedKafkaExtension::class)
+@ExtendWith(EmbeddedKafkaExtension::class, EmbeddedSftpServerExtension::class)
 @TestInstance(TestInstance.Lifecycle.PER_CLASS)
 class Tests {
 

src/test/resources/app.yml

 sftp:
   host: localhost
-  port: 22
+  port: 22000
   user: user
   password: password
-  fingerprint: MD5:34:1d:ec:5b:8e:63:db:54:a1:dc:06:26:88:f3:2b:e1
 app:
-  directory: sftp/test_institution_1/test_record_set_1/
+  directory: /memobase/test_institution_1/test_record_set_1
 kafka:
   producer:
     bootstrap.servers: localhost:12345