From ba29da951be4f47af48979209b522f20bfb2fce6 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 18 Mar 2025 12:40:27 -0400 Subject: [PATCH] GH-9909: Fix `SftpSession` for shared client Fixes: #9909 Issue link: https://github.com/spring-projects/spring-integration/issues/9909 The `SftpSession.close()` closes `sftpClient` and its `clientSession` unconditionally. At the same time the `DefaultSftpSessionFactory.isSharedSession` is expected to expose only a single shared client. When this `DefaultSftpSessionFactory` is used concurrently, there is a chance that one thread would close that shared client and another won't be able to interact due to `clientSession` is closed. * Fix `SftpSession` accepting an `isSharedClient` flag and doing nothing in the `close()` if client is shared. * Propagate `isSharedSession` state down to the `SftpSession` from the `DefaultSftpSessionFactory` * Closed shared client and its `clientSession` in the `DefaultSftpSessionFactory.destroy()` (cherry picked from commit 2df9611dded3cbece7efc1451e452e3a8200e02c) --- .../session/DefaultSftpSessionFactory.java | 23 ++++++- .../integration/sftp/session/SftpSession.java | 18 +++++ .../sftp/session/SftpSessionFactoryTests.java | 67 +++++++++++++++++++ 3 files changed, 107 insertions(+), 1 deletion(-) diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java index 217e74e889..175833c114 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java @@ -18,6 +18,7 @@ package org.springframework.integration.sftp.session; import java.io.IOException; import java.io.InputStream; +import java.io.UncheckedIOException; import java.security.GeneralSecurityException; import java.security.KeyPair; import java.time.Duration; @@ -316,7 +317,7 @@ public class DefaultSftpSessionFactory sftpClient = createSftpClient(initClientSession(), this.sftpVersionSelector, SftpErrorDataHandler.EMPTY); freshSftpClient = true; } - sftpSession = new SftpSession(sftpClient); + sftpSession = new SftpSession(sftpClient, this.isSharedSession); sftpSession.connect(); if (this.isSharedSession && freshSftpClient) { this.sharedSftpClient = sftpClient; @@ -449,6 +450,26 @@ public class DefaultSftpSessionFactory if (this.isInnerClient && this.sshClient != null && this.sshClient.isStarted()) { this.sshClient.stop(); } + + SftpClient sharedSftpClientToClose = this.sharedSftpClient; + if (sharedSftpClientToClose != null) { + try { + sharedSftpClientToClose.close(); + } + catch (IOException ex) { + throw new UncheckedIOException("failed to close an SFTP client", ex); + } + + try { + ClientSession session = sharedSftpClientToClose.getSession(); + if (session != null && session.isOpen()) { + session.close(); + } + } + catch (IOException ex) { + throw new UncheckedIOException("failed to close an SFTP client (session)", ex); + } + } } /** diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java index 860cabea9e..9255fc4e25 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java @@ -55,9 +55,23 @@ public class SftpSession implements Session { private final SftpClient sftpClient; + private final boolean isSharedClient; + public SftpSession(SftpClient sftpClient) { + this(sftpClient, false); + } + + /** + * Construct an instance based on a {@link SftpClient} and its {@code shared} status. + * When {@code isSharedClient == true}, the {@link #close()} is void. + * @param sftpClient the {@link SftpClient} to use. + * @param isSharedClient whether the {@link SftpClient} is shared. + * @since 6.3.9 + */ + public SftpSession(SftpClient sftpClient, boolean isSharedClient) { Assert.notNull(sftpClient, "'sftpClient' must not be null"); this.sftpClient = sftpClient; + this.isSharedClient = isSharedClient; } @Override @@ -152,6 +166,10 @@ public class SftpSession implements Session { @Override public void close() { + if (this.isSharedClient) { + return; + } + try { this.sftpClient.close(); } diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpSessionFactoryTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpSessionFactoryTests.java index 05d245df5b..3c58b877e0 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpSessionFactoryTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpSessionFactoryTests.java @@ -24,6 +24,11 @@ import java.time.Duration; import java.util.ArrayList; import java.util.Collections; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.IntStream; import org.apache.sshd.client.SshClient; @@ -37,6 +42,8 @@ import org.apache.sshd.core.CoreModuleProperties; import org.apache.sshd.server.SshServer; import org.apache.sshd.server.keyprovider.SimpleGeneratorHostKeyProvider; import org.apache.sshd.sftp.client.SftpClient; +import org.apache.sshd.sftp.client.SftpErrorDataHandler; +import org.apache.sshd.sftp.client.SftpVersionSelector; import org.apache.sshd.sftp.client.impl.AbstractSftpClient; import org.apache.sshd.sftp.server.SftpSubsystemFactory; import org.junit.jupiter.api.Test; @@ -268,4 +275,64 @@ public class SftpSessionFactoryTests { } } + @Test + void sharedSessionConcurrentAccess() throws Exception { + try (SshServer server = SshServer.setUpDefaultServer()) { + server.setPasswordAuthenticator((arg0, arg1, arg2) -> true); + server.setPort(0); + server.setKeyPairProvider(new SimpleGeneratorHostKeyProvider(new File("hostkey.ser").toPath())); + server.setSubsystemFactories(Collections.singletonList(new SftpSubsystemFactory())); + server.start(); + + AtomicInteger clientInstances = new AtomicInteger(); + + DefaultSftpSessionFactory sftpSessionFactory = new DefaultSftpSessionFactory(true) { + + @Override + protected SftpClient createSftpClient(ClientSession clientSession, + SftpVersionSelector initialVersionSelector, SftpErrorDataHandler errorDataHandler) + throws IOException { + + clientInstances.incrementAndGet(); + return super.createSftpClient(clientSession, initialVersionSelector, errorDataHandler); + } + + }; + sftpSessionFactory.setHost("localhost"); + sftpSessionFactory.setPort(server.getPort()); + sftpSessionFactory.setUser("user"); + sftpSessionFactory.setPassword("pass"); + sftpSessionFactory.setAllowUnknownKeys(true); + + ExecutorService executorService = Executors.newFixedThreadPool(10); + + CountDownLatch executionLatch = new CountDownLatch(20); + List errors = Collections.synchronizedList(new ArrayList<>()); + + for (int i = 0; i < 20; i++) { + executorService.execute(() -> { + try (SftpSession session = sftpSessionFactory.getSession()) { + session.list("."); + } + catch (Exception e) { + errors.add(e); + } + executionLatch.countDown(); + }); + } + + assertThat(executionLatch.await(10, TimeUnit.SECONDS)).isTrue(); + synchronized (errors) { + assertThat(errors).isEmpty(); + } + + assertThat(clientInstances).hasValue(1); + + executorService.shutdown(); + assertThat(executorService.awaitTermination(10, TimeUnit.SECONDS)).isTrue(); + + sftpSessionFactory.destroy(); + } + } + }