Added possibility configure SFTP connection using private key #197.

docs/src/main/paradox/ftp.md

@@ -51,6 +51,8 @@ respectively.
 
 For non-anonymous connection, please provide an instance of @scaladoc[NonAnonFtpCredentials](akka.stream.alpakka.ftp.FtpCredentials$$NonAnonFtpCredentials) instead.
 
+For connection using a private key, please provide an instance of @scaladoc[SftpIdentity](akka.stream.alpakka.ftp.SftpIdentity) to @scaladoc[SftpSettings](akka.stream.alpakka.ftp.RemoteFileSettings$$SftpSettings).
+
 ### Traversing a remote FTP folder recursively
 
 In order to traverse a remote folder recursively, you need to use the `ls` method in the FTP API:

ftp/src/main/scala/akka/stream/alpakka/ftp/impl/FtpLike.scala

@@ -4,7 +4,6 @@
 package akka.stream.alpakka.ftp
 package impl
 
-import akka.stream.alpakka.ftp.RemoteFileSettings.SftpSettings
 import com.jcraft.jsch.JSch
 import org.apache.commons.net.ftp.FTPClient
 

ftp/src/main/scala/akka/stream/alpakka/ftp/impl/FtpSourceFactory.scala

@@ -4,7 +4,7 @@
 package akka.stream.alpakka.ftp.impl
 
 import akka.stream.alpakka.ftp.FtpCredentials.{AnonFtpCredentials, NonAnonFtpCredentials}
-import akka.stream.alpakka.ftp.{FtpFileSettings, RemoteFileSettings}
+import akka.stream.alpakka.ftp.{FtpFileSettings, RemoteFileSettings, SftpSettings}
 import akka.stream.alpakka.ftp.RemoteFileSettings._
 import com.jcraft.jsch.JSch
 import org.apache.commons.net.ftp.FTPClient

ftp/src/main/scala/akka/stream/alpakka/ftp/impl/SftpOperations.scala

@@ -4,21 +4,30 @@
 package akka.stream.alpakka.ftp
 package impl
 
-import akka.stream.alpakka.ftp.RemoteFileSettings.SftpSettings
 import com.jcraft.jsch.{ChannelSftp, JSch}
 
 import scala.collection.immutable
 import scala.util.Try
 import scala.collection.JavaConverters._
 import java.io.{InputStream, OutputStream}
 import java.nio.file.Paths
+import scala.collection.JavaConversions._
 import java.nio.file.attribute.PosixFilePermissions
 
 private[ftp] trait SftpOperations { _: FtpLike[JSch, SftpSettings] =>
 
   type Handler = ChannelSftp
 
+  private def configureIdentity(sftpIdentity: SftpIdentity)(implicit ftpClient: JSch) = sftpIdentity match {
+    case identity: RawKeySftpIdentity =>
+      ftpClient.addIdentity(identity.name, identity.privateKey, identity.publicKey.orNull, identity.password.orNull)
+    case identity: KeyFileSftpIdentity =>
+      ftpClient.addIdentity(identity.privateKey, identity.publicKey.orNull, identity.password.orNull)
+  }
+
   def connect(connectionSettings: SftpSettings)(implicit ftpClient: JSch): Try[Handler] = Try {
+    connectionSettings.sftpIdentity.foreach(configureIdentity)
+    connectionSettings.knownHosts.foreach(ftpClient.setKnownHosts)
     val session = ftpClient.getSession(
       connectionSettings.credentials.username,
       connectionSettings.host.getHostAddress,
@@ -27,6 +36,7 @@ private[ftp] trait SftpOperations { _: FtpLike[JSch, SftpSettings] =>
     session.setPassword(connectionSettings.credentials.password)
     val config = new java.util.Properties
     config.setProperty("StrictHostKeyChecking", if (connectionSettings.strictHostKeyChecking) "yes" else "no")
+    config.putAll(connectionSettings.options)
     session.setConfig(config)
     session.connect()
     val channel = session.openChannel("sftp").asInstanceOf[ChannelSftp]

ftp/src/main/scala/akka/stream/alpakka/ftp/model.scala

@@ -93,20 +93,52 @@ object RemoteFileSettings {
       passiveMode: Boolean = false
   ) extends FtpFileSettings
 
-  /**
-   * SFTP settings
-   *
-   * @param host host
-   * @param port port
-   * @param credentials credentials (username and password)
-   * @param strictHostKeyChecking sets whether to use strict host key checking.
-   */
-  final case class SftpSettings(
-      host: InetAddress,
-      port: Int = DefaultSftpPort,
-      credentials: FtpCredentials = AnonFtpCredentials,
-      strictHostKeyChecking: Boolean = true
-  ) extends RemoteFileSettings
+}
+
+/**
+ *
+ * @param host host
+ * @param port port
+ * @param credentials credentials (username and password)
+ * @param strictHostKeyChecking sets whether to use strict host key checking.
+ * @param knownHosts known hosts file to be used when connecting
+ * @param sftpIdentity private/public key config to use when connecting
+ * @param options additional options for ssh connection
+ */
+final case class SftpSettings(
+    host: InetAddress,
+    port: Int = RemoteFileSettings.DefaultSftpPort,
+    credentials: FtpCredentials = AnonFtpCredentials,
+    strictHostKeyChecking: Boolean = true,
+    knownHosts: Option[String] = None,
+    sftpIdentity: Option[SftpIdentity] = None,
+    options: Map[String, String] = Map.empty
+) extends RemoteFileSettings {
+  def withPort(port: Int) = copy(port = port)
+
+  def withCredentials(credentials: FtpCredentials) = copy(credentials = credentials)
+
+  def withStrictHostKeyChecking(strictHostKeyChecking: Boolean) = copy(strictHostKeyChecking = strictHostKeyChecking)
+
+  def withKnownHosts(knownHosts: String) = copy(knownHosts = Some(knownHosts))
+
+  def withSftpIdentity(sftpIdentity: SftpIdentity) = copy(sftpIdentity = Some(sftpIdentity))
+
+  def withOptions(option: (String, String), options: (String, String)*) =
+    copy(options = (option +: options).toMap)
+
+  @annotation.varargs
+  def withOptions(option: akka.japi.Pair[String, String], options: akka.japi.Pair[String, String]*) =
+    copy(options = (option +: options).map(_.toScala).toMap)
+
+}
+
+object SftpSettings {
+  def create(host: InetAddress) = SftpSettings(host)
+
+  def createEmptyIdentity(): Option[SftpIdentity] = None
+
+  def createEmptyKnownHosts(): Option[String] = None
 }
 
 /**
@@ -137,3 +169,60 @@ object FtpCredentials {
    */
   final case class NonAnonFtpCredentials(username: String, password: String) extends FtpCredentials
 }
+
+object SftpIdentity {
+
+  /** Java API */
+  def createRawSftpIdentity(name: String, privateKey: Array[Byte]): RawKeySftpIdentity =
+    RawKeySftpIdentity(name, privateKey)
+
+  def createFileSftpIdentity(privateKey: String): KeyFileSftpIdentity =
+    KeyFileSftpIdentity(privateKey)
+}
+
+sealed abstract class SftpIdentity {
+  type KeyType
+  val privateKey: KeyType
+  val publicKey: Option[KeyType]
+  val password: Option[Array[Byte]]
+}
+
+/**
+ * SFTP identity for authenticating using private/public key value
+ *
+ * @param name name of identity
+ * @param privateKey private key value to use when connecting
+ * @param password password to use to decrypt private key
+ * @param publicKey public key value to use when connecting
+ */
+final case class RawKeySftpIdentity(name: String,
+                                    privateKey: Array[Byte],
+                                    password: Option[Array[Byte]] = None,
+                                    publicKey: Option[Array[Byte]] = None)
+    extends SftpIdentity {
+
+  override type KeyType = Array[Byte]
+
+  def withPassword(password: Array[Byte]) = copy(password = Some(password))
+
+  def withPublicKey(publicKey: KeyType) = copy(publicKey = Some(publicKey))
+}
+
+/**
+ * SFTP identity for authenticating using private/public key file
+ *
+ * @param privateKey private key file to use when connecting
+ * @param password password to use to decrypt private key file
+ * @param publicKey public key file to use when connecting
+ */
+final case class KeyFileSftpIdentity(privateKey: String,
+                                     password: Option[Array[Byte]] = None,
+                                     publicKey: Option[String] = None)
+    extends SftpIdentity {
+
+  override type KeyType = String
+
+  def withPassword(password: Array[Byte]) = copy(password = Some(password))
+
+  def withPublicKey(publicKey: String) = copy(publicKey = Some(publicKey))
+}

ftp/src/test/java/akka/stream/alpakka/ftp/KeyFileSftpSourceTest.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright (C) 2016-2017 Lightbend Inc. <http://www.lightbend.com>
+ */
+package akka.stream.alpakka.ftp;
+
+import akka.NotUsed;
+import akka.stream.IOResult;
+import akka.stream.alpakka.ftp.javadsl.Sftp;
+import akka.stream.javadsl.Sink;
+import akka.stream.javadsl.Source;
+import akka.util.ByteString;
+import org.junit.Test;
+
+import java.net.InetAddress;
+import java.util.concurrent.CompletionStage;
+
+public class KeyFileSftpSourceTest extends SftpSupportImpl implements CommonFtpStageTest {
+
+  @Test
+  public void listFiles() throws Exception {
+    CommonFtpStageTest.super.listFiles();
+  }
+
+  @Test
+  public void fromPath() throws Exception {
+    CommonFtpStageTest.super.fromPath();
+  }
+
+  public Source<FtpFile, NotUsed> getBrowserSource(String basePath) throws Exception {
+    return Sftp.ls(basePath, settings());
+  }
+
+  public Source<ByteString, CompletionStage<IOResult>> getIOSource(String path) throws Exception {
+    return Sftp.fromPath(path, settings());
+  }
+
+  public Sink<ByteString, CompletionStage<IOResult>> getIOSink(String path) throws Exception {
+    return Sftp.toPath(path, settings());
+  }
+
+  private SftpSettings settings() throws Exception {
+    //#create-settings
+    final SftpSettings settings = SftpSettings.create(
+            InetAddress.getByName("localhost"))
+            .withPort(getPort())
+            .withCredentials(new FtpCredentials.NonAnonFtpCredentials("different user and password", "will fail password auth"))
+            .withStrictHostKeyChecking(false) // strictHostKeyChecking
+            .withSftpIdentity(SftpIdentity.createFileSftpIdentity("ftp/src/test/resources/client.pem")
+     );
+    //#create-settings
+    return settings;
+  }
+}

ftp/src/test/java/akka/stream/alpakka/ftp/RawKeySftpSourceTest.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright (C) 2016-2017 Lightbend Inc. <http://www.lightbend.com>
+ */
+package akka.stream.alpakka.ftp;
+
+import akka.NotUsed;
+import akka.stream.IOResult;
+import akka.stream.alpakka.ftp.javadsl.Sftp;
+import akka.stream.javadsl.Sink;
+import akka.stream.javadsl.Source;
+import akka.util.ByteString;
+import org.junit.Test;
+
+import java.net.InetAddress;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.concurrent.CompletionStage;
+
+public class RawKeySftpSourceTest extends SftpSupportImpl implements CommonFtpStageTest {
+
+  @Test
+  public void listFiles() throws Exception {
+    CommonFtpStageTest.super.listFiles();
+  }
+
+  @Test
+  public void fromPath() throws Exception {
+    CommonFtpStageTest.super.fromPath();
+  }
+
+  public Source<FtpFile, NotUsed> getBrowserSource(String basePath) throws Exception {
+    return Sftp.ls(basePath, settings());
+  }
+
+  public Source<ByteString, CompletionStage<IOResult>> getIOSource(String path) throws Exception {
+    return Sftp.fromPath(path, settings());
+  }
+
+  public Sink<ByteString, CompletionStage<IOResult>> getIOSink(String path) throws Exception {
+    return Sftp.toPath(path, settings());
+  }
+
+  private SftpSettings settings() throws Exception {
+    //#create-settings
+    final SftpSettings settings = SftpSettings.create(InetAddress.getByName("localhost"))
+            .withPort(getPort())
+            .withCredentials(new FtpCredentials.NonAnonFtpCredentials("different user and password", "will fail password auth"))
+            .withStrictHostKeyChecking(false) // strictHostKeyChecking
+            .withSftpIdentity(SftpIdentity.createRawSftpIdentity("id", Files.readAllBytes(Paths.get("ftp/src/test/resources/client.pem")))
+    );
+    //#create-settings
+    return settings;
+  }
+}

ftp/src/test/java/akka/stream/alpakka/ftp/SftpStageTest.java

@@ -5,7 +5,6 @@
 
 import akka.NotUsed;
 import akka.stream.IOResult;
-import akka.stream.alpakka.ftp.RemoteFileSettings.SftpSettings;
 import akka.stream.alpakka.ftp.javadsl.Sftp;
 import akka.stream.javadsl.Sink;
 import akka.stream.javadsl.Source;
@@ -46,12 +45,11 @@ public Sink<ByteString, CompletionStage<IOResult>> getIOSink(String path) throws
 
   private SftpSettings settings() throws Exception {
     //#create-settings
-    final SftpSettings settings = new SftpSettings(
-            InetAddress.getByName("localhost"),
-            getPort(),
-            FtpCredentials.createAnonCredentials(),
-            false // strictHostKeyChecking
-    );
+    final SftpSettings settings = SftpSettings.create(
+            InetAddress.getByName("localhost"))
+            .withPort(getPort())
+            .withCredentials(FtpCredentials.createAnonCredentials())
+            .withStrictHostKeyChecking(false);
     //#create-settings
     return settings;
   }

ftp/src/test/java/akka/stream/alpakka/ftp/SftpSupportImpl.java

@@ -18,6 +18,7 @@
 import org.apache.sshd.server.subsystem.sftp.SftpSubsystemFactory;
 import org.junit.After;
 import org.junit.Before;
+
 import java.io.File;
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -31,7 +32,7 @@ abstract class SftpSupportImpl extends FtpBaseSupport {
 
     SftpSupportImpl() {
         keyPairProviderFile =
-                new File(getClass().getClassLoader().getResource("hostkey.pem").getFile());
+                new File("ftp/src/test/resources/hostkey.pem");
     }
 
     @Before