FTP: Add for FTPS the ability to set KeyManager and TrustManager (#2993)

ftp/src/main/mima-filters/6.0.3.backwards.excludes

@@ -0,0 +1,2 @@
+# Allow change to FtpsSettings
+ProblemFilters.exclude[DirectMissingMethodProblem]("akka.stream.alpakka.ftp.FtpsSettings.this")

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

@@ -21,6 +21,9 @@ private[ftp] trait FtpsOperations extends CommonFtpOperations {
     Try {
       connectionSettings.proxy.foreach(ftpClient.setProxy)
 
+      connectionSettings.keyManager.foreach(ftpClient.setKeyManager)
+      connectionSettings.trustManager.foreach(ftpClient.setTrustManager)
+
       ftpClient.connect(connectionSettings.host, connectionSettings.port)
 
       connectionSettings.configureConnection(ftpClient)

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

@@ -6,6 +6,8 @@ package akka.stream.alpakka.ftp
 
 import java.net.InetAddress
 import java.net.Proxy
+import javax.net.ssl.KeyManager
+import javax.net.ssl.TrustManager
 import java.nio.file.attribute.PosixFilePermission
 
 import akka.annotation.{DoNotInherit, InternalApi}
@@ -171,7 +173,9 @@ final class FtpsSettings private (
     val binary: Boolean,
     val passiveMode: Boolean,
     val configureConnection: FTPSClient => Unit,
-    val proxy: Option[Proxy]
+    val proxy: Option[Proxy],
+    val keyManager: Option[KeyManager],
+    val trustManager: Option[TrustManager]
 ) extends FtpFileSettings {
 
   def withHost(value: java.net.InetAddress): FtpsSettings = copy(host = value)
@@ -181,6 +185,8 @@ final class FtpsSettings private (
   def withPassiveMode(value: Boolean): FtpsSettings =
     if (passiveMode == value) this else copy(passiveMode = value)
   def withProxy(value: Proxy): FtpsSettings = copy(proxy = Some(value))
+  def withKeyManager(value: KeyManager): FtpsSettings = copy(keyManager = Some(value))
+  def withTrustManager(value: TrustManager): FtpsSettings = copy(trustManager = Some(value))
 
   /**
    * Scala API:
@@ -205,15 +211,19 @@ final class FtpsSettings private (
       binary: Boolean = binary,
       passiveMode: Boolean = passiveMode,
       configureConnection: FTPSClient => Unit = configureConnection,
-      proxy: Option[Proxy] = proxy
+      proxy: Option[Proxy] = proxy,
+      keyManager: Option[KeyManager] = keyManager,
+      trustManager: Option[TrustManager] = trustManager
   ): FtpsSettings = new FtpsSettings(
     host = host,
     port = port,
     credentials = credentials,
     binary = binary,
     passiveMode = passiveMode,
     configureConnection = configureConnection,
-    proxy = proxy
+    proxy = proxy,
+    keyManager = keyManager,
+    trustManager = trustManager
   )
 
   override def toString =
@@ -224,7 +234,9 @@ final class FtpsSettings private (
     s"binary=$binary," +
     s"passiveMode=$passiveMode," +
     s"configureConnection=$configureConnection," +
-    s"proxy=$proxy)"
+    s"proxy=$proxy," +
+    s"keyManager=$keyManager," +
+    s"trustManager=$trustManager)"
 }
 
 /**
@@ -243,7 +255,9 @@ object FtpsSettings {
     binary = false,
     passiveMode = false,
     configureConnection = _ => (),
-    proxy = None
+    proxy = None,
+    keyManager = None,
+    trustManager = None
   )
 
   /** Java API */

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

@@ -0,0 +1,84 @@
+/*
+ * Copyright (C) 2016-2020 Lightbend Inc. <https://www.lightbend.com>
+ */
+
+package akka.stream.alpakka.ftp;
+
+import nl.altindag.ssl.util.PemUtils;
+import akka.NotUsed;
+import akka.stream.IOResult;
+import akka.stream.alpakka.ftp.javadsl.Ftps;
+import akka.stream.alpakka.testkit.javadsl.LogCapturingJunit4;
+import akka.stream.javadsl.Sink;
+import akka.stream.javadsl.Source;
+import akka.util.ByteString;
+import org.junit.Rule;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import javax.net.ssl.KeyManager;
+import javax.net.ssl.TrustManager;
+import java.util.concurrent.CompletionStage;
+import java.util.function.Function;
+
+public class FtpsWithTrustAndKeyManagersStageTest extends BaseFtpSupport
+    implements CommonFtpStageTest {
+  private static final String PEM_PATH = "ftpd/pure-ftpd.pem";
+
+  @Rule public final LogCapturingJunit4 logCapturing = new LogCapturingJunit4();
+
+  @Test
+  public void listFiles() throws Exception {
+    CommonFtpStageTest.super.listFiles();
+  }
+
+  public Source<FtpFile, NotUsed> getBrowserSource(String basePath) throws Exception {
+    return Ftps.ls(basePath, settings());
+  }
+
+  public Source<ByteString, CompletionStage<IOResult>> getIOSource(String path) throws Exception {
+    return Ftps.fromPath(path, settings());
+  }
+
+  public Sink<ByteString, CompletionStage<IOResult>> getIOSink(String path) throws Exception {
+    return Ftps.toPath(path, settings());
+  }
+
+  public Sink<FtpFile, CompletionStage<IOResult>> getRemoveSink() throws Exception {
+    return Ftps.remove(settings());
+  }
+
+  public Sink<FtpFile, CompletionStage<IOResult>> getMoveSink(
+      Function<FtpFile, String> destinationPath) throws Exception {
+    return Ftps.move(destinationPath, settings());
+  }
+
+  private FtpsSettings settings() throws Exception {
+    return FtpsSettings.create(InetAddress.getByName(HOSTNAME))
+        .withPort(PORT)
+        .withCredentials(CREDENTIALS)
+        .withBinary(false)
+        .withPassiveMode(true)
+        .withTrustManager(trustManager())
+        .withKeyManager(keyManager());
+  }
+
+  private KeyManager keyManager() throws IOException {
+    try (InputStream stream = classLoader().getResourceAsStream(PEM_PATH)) {
+      return PemUtils.loadIdentityMaterial(stream);
+    }
+  }
+
+  private TrustManager trustManager() throws IOException {
+    try (InputStream stream = classLoader().getResourceAsStream(PEM_PATH)) {
+      return PemUtils.loadTrustMaterial(stream);
+    }
+  }
+
+  private ClassLoader classLoader() {
+    return FtpsWithTrustAndKeyManagersStageTest.class.getClassLoader();
+  }
+}

project/Dependencies.scala

@@ -181,7 +181,8 @@ object Dependencies {
   val Ftp = Seq(
     libraryDependencies ++= Seq(
         "commons-net" % "commons-net" % "3.8.0", // ApacheV2
-        "com.hierynomus" % "sshj" % "0.33.0" // ApacheV2
+        "com.hierynomus" % "sshj" % "0.33.0", // ApacheV2
+        "io.github.hakky54" % "sslcontext-kickstart-for-pem" % "6.8.0" % Test
       )
   )