Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-3980: Fix DefaultSftpSFactory for concurrency

Browse Source 
Fixes https://github.com/spring-projects/spring-integration/issues/3980

When not `isSharedSession`, the `initClient()` is called for every session
we request from the factory and in concurrent calls we end up with not initialized SSH client
in some threads.

* Add `synchronized` double check logic to the `initClient()` to block other threads while the first one
initialize the client
* Use `volatile boolean` instead of `AtomicBoolean` and change its state when `SshClient` is created and started
This commit is contained in:
kdebski85
2023-01-05 16:01:48 +01:00
committed by GitHub
parent f8fff81886
commit 854e555e4e
2 changed files with 87 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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.GeneralSecurityException;
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,6 +60,7 @@ import org.springframework.util.Assert;
* @author David Liu
* @author Pat Turner
* @author Artem Bilan
* @author Krzysztof Debski
*
* @since 2.0
*/
@@ -68,7 +68,7 @@ public class DefaultSftpSessionFactory implements SessionFactory<SftpClient.DirE
private final SshClient sshClient;
private final AtomicBoolean initialized = new AtomicBoolean();
private volatile boolean initialized;
private final boolean isSharedSession;
@@ -321,43 +321,52 @@ public class DefaultSftpSessionFactory implements SessionFactory<SftpClient.DirE
}
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);
}
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);
if (!this.initialized) {
synchronized (this) {
if (!this.initialized) {
doInitClient();
this.initialized = true;
}
}
this.sshClient.setUserInteraction(this.userInteraction);
this.sshClient.start();
}
}
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);
}
}
this.sshClient.setUserInteraction(this.userInteraction);
this.sshClient.start();
}
@Override
public final boolean isSharedSession() {
return this.isSharedSession;

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

@@ -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 class SftpSessionFactoryTests {
}
}
@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));
}
}
}