diff --git a/applications/sink/pom.xml b/applications/sink/pom.xml index 5cb0aec1..df07b03e 100644 --- a/applications/sink/pom.xml +++ b/applications/sink/pom.xml @@ -20,5 +20,6 @@ redis-sink rabbit-sink router-sink + sftp-sink diff --git a/applications/sink/sftp-sink/README.adoc b/applications/sink/sftp-sink/README.adoc new file mode 100644 index 00000000..1dd2762e --- /dev/null +++ b/applications/sink/sftp-sink/README.adoc @@ -0,0 +1,58 @@ +//tag::ref-doc[] += SFTP Sink + +SFTP sink is a simple option to push files to an SFTP server from incoming messages. + +It uses an `sftp-outbound-adapter`, therefore incoming messages can be either a `java.io.File` object, a `String` (content of the file) +or an array of `bytes` (file content as well). + +To use this sink, you need a username and a password to login. + +NOTE: By default Spring Integration will use `o.s.i.file.DefaultFileNameGenerator` if none is specified. `DefaultFileNameGenerator` will determine the file name +based on the value of the `file_name` header (if it exists) in the `MessageHeaders`, or if the payload of the `Message` is already a `java.io.File`, then it will +use the original name of that file. + +When configuring the `sftp.factory.known-hosts-expression` option, the root object of the evaluation is the application context, an example might be `sftp.factory.known-hosts-expression = @systemProperties['user.home'] + '/.ssh/known_hosts'`. + +== Input + +=== Headers + +* `file_name` (See note above) + +=== Payload + +* `java.io.File` +* `java.io.InputStream` +* `byte[]` +* `String` + +== Output + +N/A (writes to the SFTP server). + +== Options + +The **$$sftp$$** $$sink$$ has the following options: + +//tag::configuration-properties[] +$$sftp.consumer.auto-create-dir$$:: $$Whether or not to create the remote directory.$$ *($$Boolean$$, default: `$$true$$`)* +$$sftp.consumer.factory.allow-unknown-keys$$:: $$True to allow an unknown or changed key.$$ *($$Boolean$$, default: `$$false$$`)* +$$sftp.consumer.factory.cache-sessions$$:: $$Cache sessions.$$ *($$Boolean$$, default: `$$$$`)* +$$sftp.consumer.factory.host$$:: $$The host name of the server.$$ *($$String$$, default: `$$localhost$$`)* +$$sftp.consumer.factory.known-hosts-expression$$:: $$A SpEL expression resolving to the location of the known hosts file.$$ *($$Expression$$, default: `$$$$`)* +$$sftp.consumer.factory.pass-phrase$$:: $$Passphrase for user's private key.$$ *($$String$$, default: `$$$$`)* +$$sftp.consumer.factory.password$$:: $$The password to use to connect to the server.$$ *($$String$$, default: `$$$$`)* +$$sftp.consumer.factory.port$$:: $$The port of the server.$$ *($$Integer$$, default: `$$22$$`)* +$$sftp.consumer.factory.private-key$$:: $$Resource location of user's private key.$$ *($$Resource$$, default: `$$$$`)* +$$sftp.consumer.factory.username$$:: $$The username to use to connect to the server.$$ *($$String$$, default: `$$$$`)* +$$sftp.consumer.filename-expression$$:: $$A SpEL expression to generate the remote file name.$$ *($$String$$, default: `$$$$`)* +$$sftp.consumer.mode$$:: $$Action to take if the remote file already exists.$$ *($$FileExistsMode$$, default: `$$$$`, possible values: `APPEND`,`APPEND_NO_FLUSH`,`FAIL`,`IGNORE`,`REPLACE`,`REPLACE_IF_MODIFIED`)* +$$sftp.consumer.remote-dir$$:: $$The remote FTP directory.$$ *($$String$$, default: `$$/$$`)* +$$sftp.consumer.remote-file-separator$$:: $$The remote file separator.$$ *($$String$$, default: `$$/$$`)* +$$sftp.consumer.temporary-remote-dir$$:: $$A temporary directory where the file will be written if 'isUseTemporaryFilename()' is true.$$ *($$String$$, default: `$$/$$`)* +$$sftp.consumer.tmp-file-suffix$$:: $$The suffix to use while the transfer is in progress.$$ *($$String$$, default: `$$.tmp$$`)* +$$sftp.consumer.use-temporary-filename$$:: $$Whether or not to write to a temporary file and rename.$$ *($$Boolean$$, default: `$$true$$`)* +//end::configuration-properties[] + +//end::ref-doc[] diff --git a/applications/sink/sftp-sink/pom.xml b/applications/sink/sftp-sink/pom.xml new file mode 100644 index 00000000..87dacddd --- /dev/null +++ b/applications/sink/sftp-sink/pom.xml @@ -0,0 +1,78 @@ + + + 4.0.0 + sftp-sink + 3.0.0-SNAPSHOT + sftp-sink + sftp sink apps + jar + + + org.springframework.cloud.stream.app + stream-applications-core + 3.0.0-SNAPSHOT + + + + + + org.springframework.boot + spring-boot-starter-test + test + + + org.springframework.cloud.fn + sftp-consumer + ${java-functions.version} + + + + + + + org.springframework.cloud + spring-cloud-app-starter-doc-maven-plugin + + + org.springframework.cloud.stream.app.plugin + spring-cloud-stream-app-maven-plugin + + + sftp + sink + ${project.version} + org.springframework.cloud.fn.consumer.sftp.SftpConsumerConfiguration.class + + + + org.springframework.cloud.fn + sftp-consumer + ${java-functions.version} + + + + + + + + + + + true + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + + + false + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + + + diff --git a/applications/sink/sftp-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties b/applications/sink/sftp-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties new file mode 100644 index 00000000..b944bfbb --- /dev/null +++ b/applications/sink/sftp-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties @@ -0,0 +1,2 @@ +configuration-properties.classes=org.springframework.cloud.fn.consumer.sftp.SftpConsumerProperties, \ + org.springframework.cloud.fn.consumer.sftp.SftpConsumerProperties$Factory diff --git a/functions/common/function-test-support/pom.xml b/functions/common/function-test-support/pom.xml index 2766569f..e0262c11 100644 --- a/functions/common/function-test-support/pom.xml +++ b/functions/common/function-test-support/pom.xml @@ -13,6 +13,10 @@ ../../spring-functions-parent + + 1.6.0 + + org.apache.ftpserver @@ -35,6 +39,11 @@ spring-boot-starter-test compile + + org.apache.sshd + sshd-core + ${sshd-core.version} + diff --git a/functions/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/sftp/SftpTestSupport.java b/functions/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/sftp/SftpTestSupport.java new file mode 100644 index 00000000..ad82a7f7 --- /dev/null +++ b/functions/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/sftp/SftpTestSupport.java @@ -0,0 +1,120 @@ +/* + * Copyright 2015-2020 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.fn.test.support.sftp; + +import java.io.File; +import java.io.InputStream; +import java.math.BigInteger; +import java.nio.ByteBuffer; +import java.security.KeyFactory; +import java.security.PublicKey; +import java.security.spec.RSAPublicKeySpec; +import java.util.Arrays; +import java.util.Collections; + +import org.apache.sshd.common.file.virtualfs.VirtualFileSystemFactory; +import org.apache.sshd.server.SshServer; +import org.apache.sshd.server.keyprovider.SimpleGeneratorHostKeyProvider; +import org.apache.sshd.server.subsystem.sftp.SftpSubsystemFactory; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; + +import org.springframework.cloud.fn.test.support.file.remote.RemoteFileTestSupport; +import org.springframework.core.io.ClassPathResource; +import org.springframework.util.Base64Utils; +import org.springframework.util.FileCopyUtils; +import org.springframework.util.StringUtils; + +/** + * Provides an embedded SFTP Server for test cases. + * + * @author David Turanski + * @author Gary Russell + * @author Artem Bilan + */ +public class SftpTestSupport extends RemoteFileTestSupport { + + private static SshServer server; + + @Override + public String prefix() { + return "sftp"; + } + + @BeforeAll + public static void createServer() throws Exception { + server = SshServer.setUpDefaultServer(); + server.setPasswordAuthenticator((username, password, session) -> + StringUtils.hasText(password) && !"badPassword".equals(password)); // fail if pub key validation failed + server.setPublickeyAuthenticator((username, key, session) -> key.equals(decodePublicKey("id_rsa_pp.pub.rename2"))); + server.setPort(0); + server.setKeyPairProvider(new SimpleGeneratorHostKeyProvider(new File("hostkey.ser"))); + server.setSubsystemFactories(Collections.singletonList(new SftpSubsystemFactory())); + server.setFileSystemFactory(new VirtualFileSystemFactory(remoteTemporaryFolder)); + server.start(); + System.setProperty("sftp.consumer.factory.port", String.valueOf(server.getPort())); + System.setProperty("sftp.consumer.localDir", + localTemporaryFolder + File.separator + "localTarget"); + } + + @AfterAll + public static void stopServer() throws Exception { + server.stop(); + File hostkey = new File("hostkey.ser"); + if (hostkey.exists()) { + hostkey.delete(); + } + System.clearProperty("sftp.consumer.factory.port"); + System.clearProperty("sftp.consumer.localDir"); + } + + private static PublicKey decodePublicKey(String key) { + try { + InputStream stream = new ClassPathResource(key).getInputStream(); + byte[] keyBytes = FileCopyUtils.copyToByteArray(stream); + // strip any newline chars + while (keyBytes[keyBytes.length - 1] == 0x0a || keyBytes[keyBytes.length - 1] == 0x0d) { + keyBytes = Arrays.copyOf(keyBytes, keyBytes.length - 1); + } + byte[] decodeBuffer = Base64Utils.decode(keyBytes); + ByteBuffer bb = ByteBuffer.wrap(decodeBuffer); + int len = bb.getInt(); + byte[] type = new byte[len]; + bb.get(type); + if ("ssh-rsa".equals(new String(type))) { + BigInteger e = decodeBigInt(bb); + BigInteger m = decodeBigInt(bb); + RSAPublicKeySpec spec = new RSAPublicKeySpec(m, e); + return KeyFactory.getInstance("RSA").generatePublic(spec); + + } + else { + throw new IllegalArgumentException("Only supports RSA"); + } + } + catch (Exception e) { + throw new IllegalStateException("Failed to determine the test public key", e); + } + } + + private static BigInteger decodeBigInt(ByteBuffer bb) { + int len = bb.getInt(); + byte[] bytes = new byte[len]; + bb.get(bytes); + return new BigInteger(bytes); + } +} diff --git a/functions/consumer/sftp-consumer/pom.xml b/functions/consumer/sftp-consumer/pom.xml new file mode 100644 index 00000000..160804cf --- /dev/null +++ b/functions/consumer/sftp-consumer/pom.xml @@ -0,0 +1,46 @@ + + + 4.0.0 + sftp-consumer + 1.0.0-SNAPSHOT + sftp-consumer + file consumer + + + org.springframework.cloud.fn + spring-functions-parent + 1.0.0-SNAPSHOT + ../../spring-functions-parent + + + + + org.springframework.integration + spring-integration-sftp + + + org.springframework.boot + spring-boot-starter-integration + + + org.springframework.boot + spring-boot-starter-validation + + + org.springframework.boot + spring-boot-configuration-processor + provided + + + org.springframework.boot + spring-boot-starter-test + test + + + org.springframework.cloud.fn + function-test-support + ${project.version} + + + + diff --git a/functions/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerConfiguration.java b/functions/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerConfiguration.java new file mode 100644 index 00000000..5273dc45 --- /dev/null +++ b/functions/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerConfiguration.java @@ -0,0 +1,69 @@ +/* + * Copyright 2015-2020 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.fn.consumer.sftp; + +import java.util.function.Consumer; + +import com.jcraft.jsch.ChannelSftp; + +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.expression.ExpressionParser; +import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlowBuilder; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.sftp.dsl.Sftp; +import org.springframework.integration.sftp.dsl.SftpMessageHandlerSpec; +import org.springframework.integration.sftp.session.SftpRemoteFileTemplate; +import org.springframework.messaging.Message; + +@Configuration +@EnableConfigurationProperties(SftpConsumerProperties.class) +@Import(SftpConsumerSessionFactoryConfiguration.class) +public class SftpConsumerConfiguration { + + private static final ExpressionParser EXPRESSION_PARSER = new SpelExpressionParser(); + + @Bean + public IntegrationFlow ftpInboundFlow(SftpConsumerProperties properties, SessionFactory ftpSessionFactory) { + IntegrationFlowBuilder integrationFlowBuilder = + IntegrationFlows.from(MessageConsumer.class, (gateway) -> gateway.beanName("sftpConsumer")); + + SftpMessageHandlerSpec handlerSpec = + Sftp.outboundAdapter(new SftpRemoteFileTemplate(ftpSessionFactory), properties.getMode()) + .remoteDirectory(properties.getRemoteDir()) + .remoteFileSeparator(properties.getRemoteFileSeparator()) + .autoCreateDirectory(properties.isAutoCreateDir()) + .temporaryFileSuffix(properties.getTmpFileSuffix()); + if (properties.getFilenameExpression() != null) { + handlerSpec.fileNameExpression(EXPRESSION_PARSER.parseExpression(properties.getFilenameExpression()) + .getExpressionString()); + } + return integrationFlowBuilder + .handle(handlerSpec) + .get(); + } + + private interface MessageConsumer extends Consumer> { + + } + +} diff --git a/functions/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerProperties.java b/functions/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerProperties.java new file mode 100644 index 00000000..b15038b8 --- /dev/null +++ b/functions/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerProperties.java @@ -0,0 +1,278 @@ +/* + * Copyright 2015-2020 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.fn.consumer.sftp; + +import javax.validation.constraints.NotBlank; +import javax.validation.constraints.NotNull; + +import org.hibernate.validator.constraints.Range; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.core.io.Resource; +import org.springframework.expression.Expression; +import org.springframework.integration.file.support.FileExistsMode; +import org.springframework.validation.annotation.Validated; + +/** + * @author Gary Russell + * @author Artem Bilan + */ +@ConfigurationProperties("sftp.consumer") +@Validated +public class SftpConsumerProperties { + + private final Factory factory = new Factory(); + + /** + * A temporary directory where the file will be written if 'isUseTemporaryFilename()' is true. + */ + private String temporaryRemoteDir = "/"; + + /** + * Whether or not to create the remote directory. + */ + private boolean autoCreateDir = true; + + /** + * Action to take if the remote file already exists. + */ + private FileExistsMode mode = FileExistsMode.REPLACE; + + /** + * Whether or not to write to a temporary file and rename. + */ + private boolean useTemporaryFilename = true; + + /** + * A SpEL expression to generate the remote file name. + */ + private String filenameExpression; + + /** + * The remote FTP directory. + */ + private String remoteDir = "/"; + + /** + * The suffix to use while the transfer is in progress. + */ + private String tmpFileSuffix = ".tmp"; + + /** + * The remote file separator. + */ + private String remoteFileSeparator = "/"; + + @NotBlank + public String getTemporaryRemoteDir() { + return this.temporaryRemoteDir; + } + + public void setTemporaryRemoteDir(String temporaryRemoteDir) { + this.temporaryRemoteDir = temporaryRemoteDir; + } + + public boolean isAutoCreateDir() { + return this.autoCreateDir; + } + + public void setAutoCreateDir(boolean autoCreateDir) { + this.autoCreateDir = autoCreateDir; + } + + @NotNull + public FileExistsMode getMode() { + return this.mode; + } + + public void setMode(FileExistsMode mode) { + this.mode = mode; + } + + public boolean isUseTemporaryFilename() { + return this.useTemporaryFilename; + } + + public void setUseTemporaryFilename(boolean useTemporaryFilename) { + this.useTemporaryFilename = useTemporaryFilename; + } + + public String getFilenameExpression() { + return this.filenameExpression; + } + + public void setFilenameExpression(String filenameExpression) { + this.filenameExpression = filenameExpression; + } + + @NotBlank + public String getRemoteDir() { + return remoteDir; + } + + public final void setRemoteDir(String remoteDir) { + this.remoteDir = remoteDir; + } + + @NotBlank + public String getTmpFileSuffix() { + return tmpFileSuffix; + } + + public void setTmpFileSuffix(String tmpFileSuffix) { + this.tmpFileSuffix = tmpFileSuffix; + } + + @NotBlank + public String getRemoteFileSeparator() { + return remoteFileSeparator; + } + + public void setRemoteFileSeparator(String remoteFileSeparator) { + this.remoteFileSeparator = remoteFileSeparator; + } + + public Factory getFactory() { + return this.factory; + } + + public static class Factory { + + /** + * The host name of the server. + */ + private String host = "localhost"; + + /** + * The username to use to connect to the server. + */ + + private String username; + + /** + * The password to use to connect to the server. + */ + private String password; + + /** + * Cache sessions. + */ + private Boolean cacheSessions; + + /** + * The port of the server. + */ + private int port = 22; + + /** + * Resource location of user's private key. + */ + private Resource privateKey; + + /** + * Passphrase for user's private key. + */ + private String passPhrase = ""; + + /** + * True to allow an unknown or changed key. + */ + private boolean allowUnknownKeys = false; + + /** + * A SpEL expression resolving to the location of the known hosts file. + */ + private Expression knownHostsExpression = null; + + + @NotBlank + public String getHost() { + return this.host; + } + + public void setHost(String host) { + this.host = host; + } + + @NotBlank + public String getUsername() { + return this.username; + } + + public void setUsername(String username) { + this.username = username; + } + + public String getPassword() { + return this.password; + } + + public void setPassword(String password) { + this.password = password; + } + + public Boolean getCacheSessions() { + return this.cacheSessions; + } + + public void setCacheSessions(Boolean cacheSessions) { + this.cacheSessions = cacheSessions; + } + + @Range(min = 0, max = 65535) + public int getPort() { + return this.port; + } + + public void setPort(int port) { + this.port = port; + } + + public Resource getPrivateKey() { + return this.privateKey; + } + + public void setPrivateKey(Resource privateKey) { + this.privateKey = privateKey; + } + + public String getPassPhrase() { + return this.passPhrase; + } + + public void setPassPhrase(String passPhrase) { + this.passPhrase = passPhrase; + } + + public boolean isAllowUnknownKeys() { + return this.allowUnknownKeys; + } + + public void setAllowUnknownKeys(boolean allowUnknownKeys) { + this.allowUnknownKeys = allowUnknownKeys; + } + + public Expression getKnownHostsExpression() { + return this.knownHostsExpression; + } + + public void setKnownHostsExpression(Expression knownHosts) { + this.knownHostsExpression = knownHosts; + } + + } +} diff --git a/functions/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerSessionFactoryConfiguration.java b/functions/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerSessionFactoryConfiguration.java new file mode 100644 index 00000000..550f61c2 --- /dev/null +++ b/functions/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerSessionFactoryConfiguration.java @@ -0,0 +1,61 @@ +/* + * Copyright 2015-2020 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.fn.consumer.sftp; + +import com.jcraft.jsch.ChannelSftp; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.context.annotation.Bean; +import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.integration.file.remote.session.CachingSessionFactory; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.sftp.session.DefaultSftpSessionFactory; + +/** + * Session factory configuration. + * + * @author Gary Russell + * + */ +public class SftpConsumerSessionFactoryConfiguration { + + @Bean + @ConditionalOnMissingBean + public SessionFactory sftpSessionFactory(SftpConsumerProperties properties, BeanFactory beanFactory) { + DefaultSftpSessionFactory sftpSessionFactory = new DefaultSftpSessionFactory(); + SftpConsumerProperties.Factory factory = properties.getFactory(); + sftpSessionFactory.setHost(factory.getHost()); + sftpSessionFactory.setPort(factory.getPort()); + sftpSessionFactory.setUser(factory.getUsername()); + sftpSessionFactory.setPassword(factory.getPassword()); + sftpSessionFactory.setPrivateKey(factory.getPrivateKey()); + sftpSessionFactory.setPrivateKeyPassphrase(factory.getPassPhrase()); + sftpSessionFactory.setAllowUnknownKeys(factory.isAllowUnknownKeys()); + if (factory.getKnownHostsExpression() != null) { + sftpSessionFactory.setKnownHosts(factory.getKnownHostsExpression() + .getValue(IntegrationContextUtils.getEvaluationContext(beanFactory), String.class)); + } + if (factory.getCacheSessions() != null) { + CachingSessionFactory csf = new CachingSessionFactory<>(sftpSessionFactory); + return csf; + } + else { + return sftpSessionFactory; + } + } +} diff --git a/functions/consumer/sftp-consumer/src/test/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerPropertiesTests.java b/functions/consumer/sftp-consumer/src/test/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerPropertiesTests.java new file mode 100644 index 00000000..a9e88d72 --- /dev/null +++ b/functions/consumer/sftp-consumer/src/test/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerPropertiesTests.java @@ -0,0 +1,203 @@ +/* + * Copyright 2015-2020 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.fn.consumer.sftp; + +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.boot.context.properties.ConfigurationPropertiesBinding; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.boot.test.util.TestPropertyValues; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.context.annotation.Lazy; +import org.springframework.core.convert.converter.Converter; +import org.springframework.expression.EvaluationContext; +import org.springframework.expression.Expression; +import org.springframework.expression.ParseException; +import org.springframework.expression.spel.standard.SpelExpression; +import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.config.IntegrationConverter; +import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.file.support.FileExistsMode; +import org.springframework.integration.test.util.TestUtils; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author David Turanski + * @author Gary Russell + * @author Artem Bilan + * @author Chris Schaefer + */ +public class SftpConsumerPropertiesTests { + + @Test + public void remoteDirCanBeCustomized() { + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); + testPropertyValues(context, "sftp.consumer.remoteDir:/remote"); + context.register(Conf.class); + context.refresh(); + SftpConsumerProperties properties = context.getBean(SftpConsumerProperties.class); + assertThat(properties.getRemoteDir()).isEqualTo("/remote"); + context.close(); + } + + @Test + public void autoCreateDirCanBeDisabled() { + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); + testPropertyValues(context, "sftp.consumer.autoCreateDir:false"); + context.register(Conf.class); + context.refresh(); + SftpConsumerProperties properties = context.getBean(SftpConsumerProperties.class); + assertThat(!properties.isAutoCreateDir()).isTrue(); + context.close(); + } + + @Test + public void tmpFileSuffixCanBeCustomized() { + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); + testPropertyValues(context, "sftp.consumer.tmpFileSuffix:.foo"); + context.register(Conf.class); + context.refresh(); + SftpConsumerProperties properties = context.getBean(SftpConsumerProperties.class); + assertThat(properties.getTmpFileSuffix()).isEqualTo(".foo"); + context.close(); + } + + @Test + public void tmpFileRemoteDirCanBeCustomized() { + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); + testPropertyValues(context, "sftp.consumer.temporaryRemoteDir:/foo"); + context.register(Conf.class); + context.refresh(); + SftpConsumerProperties properties = context.getBean(SftpConsumerProperties.class); + assertThat(properties.getTemporaryRemoteDir()).isEqualTo("/foo"); + context.close(); + } + + @Test + public void remoteFileSeparatorCanBeCustomized() { + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); + testPropertyValues(context, "sftp.consumer.remoteFileSeparator:\\"); + context.register(Conf.class); + context.refresh(); + SftpConsumerProperties properties = context.getBean(SftpConsumerProperties.class); + assertThat(properties.getRemoteFileSeparator()).isEqualTo("\\"); + context.close(); + } + + @Test + public void useTemporaryFileNameCanBeCustomized() { + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); + testPropertyValues(context, "sftp.consumer.useTemporaryFilename:false"); + context.register(Conf.class); + context.refresh(); + SftpConsumerProperties properties = context.getBean(SftpConsumerProperties.class); + assertThat(properties.isUseTemporaryFilename()).isFalse(); + context.close(); + } + + @Test + public void fileExistsModeCanBeCustomized() { + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); + testPropertyValues(context, "sftp.consumer.mode:FAIL"); + context.register(Conf.class); + context.refresh(); + SftpConsumerProperties properties = context.getBean(SftpConsumerProperties.class); + assertThat(properties.getMode()).isEqualTo(FileExistsMode.FAIL); + context.close(); + } + + @Test + public void knownHostsExpression() { + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); + testPropertyValues(context, + "sftp.consumer.factory.known-hosts-expression = @systemProperties[\"user.home\"] + \"/.ssh/known_hosts\"", + "sftp.consumer.factory.cache-sessions = true"); + context.register(Factory.class); + context.refresh(); + SessionFactory sessionFactory = context.getBean(SessionFactory.class); + assertThat(TestUtils.getPropertyValue(sessionFactory, "sessionFactory.knownHosts").toString().endsWith( + "/.ssh/known_hosts]")).isTrue(); + context.close(); + } + + private void testPropertyValues(ConfigurableApplicationContext context, String... props) { + TestPropertyValues.of("sftp.consumer.factory.username=foo").and(props).applyTo(context); + } + + @Configuration + @EnableConfigurationProperties(SftpConsumerProperties.class) + static class Conf { + + } + + @Configuration + @EnableConfigurationProperties(SftpConsumerProperties.class) + @EnableIntegration + @Import(SftpConsumerSessionFactoryConfiguration.class) + static class Factory { + + @Bean + @ConfigurationPropertiesBinding + @IntegrationConverter + public Converter spelConverter() { + return new SpelConverter(); + } + + /** + * TODO: This needs to be refactored into a generic place for any functions to use. + * + * A simple converter from String to Expression. + * + * @author Eric Bottard + */ + public static class SpelConverter implements Converter { + + private SpelExpressionParser parser = new SpelExpressionParser(); + + @Autowired + @Qualifier(IntegrationContextUtils.INTEGRATION_EVALUATION_CONTEXT_BEAN_NAME) + @Lazy + private EvaluationContext evaluationContext; + + @Override + public Expression convert(String source) { + try { + Expression expression = this.parser.parseExpression(source); + if (expression instanceof SpelExpression) { + ((SpelExpression) expression) + .setEvaluationContext(this.evaluationContext); + } + return expression; + } + catch (ParseException e) { + throw new IllegalArgumentException(String.format( + "Could not convert '%s' into a SpEL expression", source), e); + } + } + + } + } +} diff --git a/functions/consumer/sftp-consumer/src/test/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerTests.java b/functions/consumer/sftp-consumer/src/test/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerTests.java new file mode 100644 index 00000000..53dd5a01 --- /dev/null +++ b/functions/consumer/sftp-consumer/src/test/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerTests.java @@ -0,0 +1,78 @@ +/* + * Copyright 2015-2020 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.fn.consumer.sftp; + +import java.io.File; +import java.util.function.Consumer; + +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.cloud.fn.test.support.sftp.SftpTestSupport; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.annotation.DirtiesContext; + +import static org.assertj.core.api.Assertions.assertThat; + +@DirtiesContext +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, + properties = { + "sftp.consumer.remoteDir = sftpTarget", + "sftp.consumer.factory.username = foo", + "sftp.consumer.factory.password = foo", + "sftp.consumer.mode = FAIL", + "sftp.consumer.filenameExpression = payload.name.toUpperCase()", + "sftp.consumer.factory.allowUnknownKeys = true" + }) +public class SftpConsumerTests extends SftpTestSupport { + + @Autowired + Consumer> sftpConsumer; + + @Test + public void sendFiles() { + for (int i = 1; i <= 2; i++) { + String pathname = "/localSource" + i + ".txt"; + String upperPathname = pathname.toUpperCase(); + new File(getTargetRemoteDirectory() + upperPathname).delete(); + assertThat(new File(getTargetRemoteDirectory() + upperPathname).exists()).isFalse(); + sftpConsumer.accept(new GenericMessage<>(new File(getSourceLocalDirectory() + pathname))); + File expected = new File(getTargetRemoteDirectory() + upperPathname); + assertThat(expected.exists()).isTrue(); + // verify the uppercase on a case-insensitive file system + File[] files = getTargetRemoteDirectory().listFiles(); + for (File file : files) { + assertThat(file.getName().startsWith("LOCALSOURCE")).isTrue(); + } + } + } + + @Test + public void serverRefreshed() { // noop test to test the dirs are refreshed properly + String pathname = "/LOCALSOURCE1.TXT"; + assertThat(getTargetRemoteDirectory().exists()).isTrue(); + assertThat(new File(getTargetRemoteDirectory() + pathname).exists()).isFalse(); + } + + @SpringBootApplication + static class TestApplication { + } +} + diff --git a/functions/pom.xml b/functions/pom.xml index fd22ecbe..c8f0dbd2 100644 --- a/functions/pom.xml +++ b/functions/pom.xml @@ -52,6 +52,7 @@ consumer/mongodb-consumer consumer/rabbit-consumer consumer/redis-consumer + consumer/sftp-consumer function/filter-function function/header-enricher-function