GH-3980: Fix concurrent initialization in DefaultSftpSessionFactory

spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 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.
@@ -22,7 +22,6 @@
 import java.security.KeyPair;
 import java.time.Duration;
 import java.util.Collection;
-import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
@@ -61,14 +60,15 @@
  * @author David Liu
  * @author Pat Turner
  * @author Artem Bilan
+ * @author Krzysztof Debski
  *
  * @since 2.0
  */
 public class DefaultSftpSessionFactory implements SessionFactory<SftpClient.DirEntry>, SharedSessionCapable {
 
 	private final SshClient sshClient;
 
-	private final AtomicBoolean initialized = new AtomicBoolean();
+	private volatile boolean initialized;
 
 	private final boolean isSharedSession;
 
@@ -321,41 +321,50 @@ private ClientSession initClientSession() throws IOException {
 	}
 
 	private void initClient() throws IOException {
-		if (this.initialized.compareAndSet(false, true)) {
-			if (this.port <= 0) {
-				this.port = SshConstants.DEFAULT_PORT;
-			}
-			ServerKeyVerifier serverKeyVerifier =
-					this.allowUnknownKeys ? AcceptAllServerKeyVerifier.INSTANCE : RejectAllServerKeyVerifier.INSTANCE;
-			if (this.knownHosts != null) {
-				serverKeyVerifier = new ResourceKnownHostsServerKeyVerifier(this.knownHosts);
+		if (!this.initialized) {
+			synchronized (this) {
+				if (!this.initialized) {
+					doInitClient();
+					this.initialized = true;
+				}
 			}
-			this.sshClient.setServerKeyVerifier(serverKeyVerifier);
-
-			this.sshClient.setPasswordIdentityProvider(PasswordIdentityProvider.wrapPasswords(this.password));
-			if (this.privateKey != null) {
-				IoResource<Resource> privateKeyResource =
-						new AbstractIoResource<>(Resource.class, this.privateKey) {
-
-							@Override
-							public InputStream openInputStream() throws IOException {
-								return getResourceValue().getInputStream();
-							}
-						};
-				try {
-					Collection<KeyPair> keys =
-							SecurityUtils.getKeyPairResourceParser()
+		}
+	}
+
+	private void doInitClient() throws IOException {
+		if (this.port <= 0) {
+			this.port = SshConstants.DEFAULT_PORT;
+		}
+		ServerKeyVerifier serverKeyVerifier =
+				this.allowUnknownKeys ? AcceptAllServerKeyVerifier.INSTANCE : RejectAllServerKeyVerifier.INSTANCE;
+		if (this.knownHosts != null) {
+			serverKeyVerifier = new ResourceKnownHostsServerKeyVerifier(this.knownHosts);
+		}
+		this.sshClient.setServerKeyVerifier(serverKeyVerifier);
+
+		this.sshClient.setPasswordIdentityProvider(PasswordIdentityProvider.wrapPasswords(this.password));
+		if (this.privateKey != null) {
+			IoResource<Resource> privateKeyResource =
+					new AbstractIoResource<>(Resource.class, this.privateKey) {
+
+						@Override
+						public InputStream openInputStream() throws IOException {
+							return getResourceValue().getInputStream();
+						}
+					};
+			try {
+				Collection<KeyPair> keys =
+						SecurityUtils.getKeyPairResourceParser()
 									.loadKeyPairs(null, privateKeyResource,
-											FilePasswordProvider.of(this.privateKeyPassphrase));
-					this.sshClient.setKeyIdentityProvider(KeyIdentityProvider.wrapKeyPairs(keys));
-				}
-				catch (GeneralSecurityException ex) {
-					throw new IOException("Cannot load private key: " + this.privateKey.getFilename(), ex);
-				}
+										FilePasswordProvider.of(this.privateKeyPassphrase));
+				this.sshClient.setKeyIdentityProvider(KeyIdentityProvider.wrapKeyPairs(keys));
+			}
+			catch (GeneralSecurityException ex) {
+				throw new IOException("Cannot load private key: " + this.privateKey.getFilename(), ex);
 			}
-			this.sshClient.setUserInteraction(this.userInteraction);
-			this.sshClient.start();
 		}
+		this.sshClient.setUserInteraction(this.userInteraction);
+		this.sshClient.start();
 	}
 
 	@Override

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/session/SftpSessionFactoryTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2022 the original author or authors.
+ * Copyright 2014-2023 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.
@@ -17,15 +17,24 @@
 package org.springframework.integration.sftp.session;
 
 import java.io.File;
+import java.io.IOException;
 import java.net.ConnectException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
 
 import org.apache.sshd.common.SshException;
 import org.apache.sshd.server.SshServer;
 import org.apache.sshd.server.keyprovider.SimpleGeneratorHostKeyProvider;
+import org.apache.sshd.sftp.server.SftpSubsystemFactory;
 import org.junit.jupiter.api.Test;
 
+import org.springframework.core.task.AsyncTaskExecutor;
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
+
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.fail;
+import static org.awaitility.Awaitility.await;
 
 /**
  * @author Gary Russell
@@ -84,4 +93,37 @@ public void testConnectFailSocketOpen() throws Exception {
 		}
 	}
 
+	@Test
+	public void concurrentGetSessionDoesntCauseFailure() throws IOException {
+		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();
+
+			DefaultSftpSessionFactory sftpSessionFactory = new DefaultSftpSessionFactory();
+			sftpSessionFactory.setHost("localhost");
+			sftpSessionFactory.setPort(server.getPort());
+			sftpSessionFactory.setUser("user");
+			sftpSessionFactory.setPassword("pass");
+			sftpSessionFactory.setAllowUnknownKeys(true);
+
+			List<SftpSession> concurrentSessions = new ArrayList<>();
+
+			AsyncTaskExecutor asyncTaskExecutor = new SimpleAsyncTaskExecutor();
+			for (int i = 0; i < 3; i++) {
+				asyncTaskExecutor.execute(() -> concurrentSessions.add(sftpSessionFactory.getSession()));
+			}
+
+			await().until(() -> concurrentSessions.size() == 3);
+
+			assertThat(concurrentSessions.get(0))
+					.isNotEqualTo(concurrentSessions.get(1))
+					.isNotEqualTo(concurrentSessions.get(2));
+
+			assertThat(concurrentSessions.get(1)).isNotEqualTo(concurrentSessions.get(2));
+		}
+	}
+
 }