External configured SshClient has it's configuration overridden by DefaultSftpSessionFactory

[ams-azaaiman]

In what version(s) of Spring Integration are you seeing this issue?
v6.0.3

Describe the bug
When providing an external configured SshClient to DefaultSftpSessionFactory, the configuration of the SshClient is partially overwritten by DefaultSftpSessionFactory, rendering it useless.

Our use case for providing an external configured SshClient is to be able to set additional Key Exchange Algorithms, due to old SFTP servers which are beyond our span of control.

Methods like #setKnownHostsResource, #setPassword throw an exception when being called on an DefaultSftpSessionFactory instance with an externally provided SshClient.
However in the private method #doInitClient their values are being used to configure the SshClient, ignoring if it is externally or internally provided.

To Reproduce

1. Create a SshClient with a simple ServerKeyVerifier:

   SshClient sshClient = SshClient.setUpDefaultClient();
   sshClient.setServerKeyVerifier(AcceptAllServerKeyVerifier.INSTANCE);
   

2. Use the built SshClient in a new DefaultSftpSessionFactory(sshClient, false), which in turn can be used for a SFTP polling or producing IntegrationFlow

3. Let the flow run, and see it will not connect due to an exception org.apache.sshd.common.SshException: Server key did not validate

Expected behavior
An externally configured SshClient will not have the settings overridden, when those cannot be configured using DefaultSftpSessionFactory.
The flow should have connected to any given SFTP server.

Suggested fix is to wrap line 338-366 of DefaultSftpSessionFactory in an if (this.isInnerClient) { ... }.

Sample
Below the full stacktrace:

ERROR o.s.i.handler.LoggingHandler - org.springframework.messaging.MessagingException: Failed to execute on session
        at org.springframework.integration.file.remote.RemoteFileTemplate.execute(RemoteFileTemplate.java:461)
        at org.springframework.integration.file.remote.RemoteFileTemplate.list(RemoteFileTemplate.java:422)
        at org.springframework.integration.file.remote.AbstractRemoteFileStreamingMessageSource.listFiles(AbstractRemoteFileStreamingMessageSource.java:267)
        at org.springframework.integration.file.remote.AbstractRemoteFileStreamingMessageSource.poll(AbstractRemoteFileStreamingMessageSource.java:246)
        at org.springframework.integration.file.remote.AbstractRemoteFileStreamingMessageSource.doReceive(AbstractRemoteFileStreamingMessageSource.java:198)
        at org.springframework.integration.endpoint.AbstractFetchLimitingMessageSource.doReceive(AbstractFetchLimitingMessageSource.java:47)
        at org.springframework.integration.endpoint.AbstractMessageSource.receive(AbstractMessageSource.java:142)
        at org.springframework.integration.endpoint.SourcePollingChannelAdapter.receiveMessage(SourcePollingChannelAdapter.java:212)
        at org.springframework.integration.endpoint.AbstractPollingEndpoint.doPoll(AbstractPollingEndpoint.java:443)
        at org.springframework.integration.endpoint.AbstractPollingEndpoint.pollForMessage(AbstractPollingEndpoint.java:412)
        at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$createPoller$4(AbstractPollingEndpoint.java:348)
        at org.springframework.integration.util.ErrorHandlingTaskExecutor.lambda$execute$0(ErrorHandlingTaskExecutor.java:57)
        at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50)
        at org.springframework.integration.util.ErrorHandlingTaskExecutor.execute(ErrorHandlingTaskExecutor.java:55)
        at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$createPoller$5(AbstractPollingEndpoint.java:341)
        at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
        at org.springframework.scheduling.concurrent.ReschedulingRunnable.run(ReschedulingRunnable.java:96)
        at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)
        at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
        at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
        at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
        at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
        at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: org.springframework.integration.util.PoolItemNotAvailableException: Failed to obtain pooled item
        at org.springframework.integration.util.SimplePool.getItem(SimplePool.java:196)
        at org.springframework.integration.file.remote.session.CachingSessionFactory.getSession(CachingSessionFactory.java:134)
        at org.springframework.integration.file.remote.RemoteFileTemplate.execute(RemoteFileTemplate.java:447)
        ... 22 more
Caused by: java.lang.IllegalStateException: failed to create SFTP Session
        at org.springframework.integration.sftp.session.DefaultSftpSessionFactory.getSession(DefaultSftpSessionFactory.java:291)
        at org.springframework.integration.sftp.session.DefaultSftpSessionFactory.getSession(DefaultSftpSessionFactory.java:67)
        at org.springframework.integration.file.remote.session.CachingSessionFactory$1.createForPool(CachingSessionFactory.java:85)
        at org.springframework.integration.file.remote.session.CachingSessionFactory$1.createForPool(CachingSessionFactory.java:82)
        at org.springframework.integration.util.SimplePool.doGetItem(SimplePool.java:206)
        at org.springframework.integration.util.SimplePool.getItem(SimplePool.java:187)
        ... 24 more
Caused by: org.apache.sshd.common.SshException: Server key did not validate
        at org.apache.sshd.common.future.AbstractSshFuture.verifyResult(AbstractSshFuture.java:127)
        at org.apache.sshd.client.future.DefaultAuthFuture.verify(DefaultAuthFuture.java:39)
        at org.apache.sshd.client.future.DefaultAuthFuture.verify(DefaultAuthFuture.java:32)
        at org.apache.sshd.common.future.VerifiableFuture.verify(VerifiableFuture.java:68)
        at org.springframework.integration.sftp.session.DefaultSftpSessionFactory.initClientSession(DefaultSftpSessionFactory.java:318)
        at org.springframework.integration.sftp.session.DefaultSftpSessionFactory.getSession(DefaultSftpSessionFactory.java:281)
        ... 29 more
Caused by: org.apache.sshd.common.SshException: Server key did not validate
        at org.apache.sshd.client.session.AbstractClientSession.checkKeys(AbstractClientSession.java:629)
        at org.apache.sshd.common.session.helpers.AbstractSession.handleKexMessage(AbstractSession.java:726)
        at org.apache.sshd.common.session.helpers.AbstractSession.doHandleMessage(AbstractSession.java:590)
        at org.apache.sshd.common.session.helpers.AbstractSession.lambda$handleMessage$0(AbstractSession.java:522)
        at org.apache.sshd.common.util.threads.ThreadUtils.runAsInternal(ThreadUtils.java:68)
        at org.apache.sshd.common.session.helpers.AbstractSession.handleMessage(AbstractSession.java:521)
        at org.apache.sshd.common.session.helpers.AbstractSession.decode(AbstractSession.java:1639)
        at org.apache.sshd.common.session.helpers.AbstractSession.messageReceived(AbstractSession.java:482)
        at org.apache.sshd.common.session.helpers.AbstractSessionIoHandler.messageReceived(AbstractSessionIoHandler.java:64)
        at org.apache.sshd.common.io.nio2.Nio2Session.handleReadCycleCompletion(Nio2Session.java:407)
        at org.apache.sshd.common.io.nio2.Nio2Session$1.onCompleted(Nio2Session.java:380)
        at org.apache.sshd.common.io.nio2.Nio2Session$1.onCompleted(Nio2Session.java:375)
        at org.apache.sshd.common.io.nio2.Nio2CompletionHandler.lambda$completed$0(Nio2CompletionHandler.java:38)
        at java.base/java.security.AccessController.doPrivileged(AccessController.java:318)
        at org.apache.sshd.common.io.nio2.Nio2CompletionHandler.completed(Nio2CompletionHandler.java:37)
        at java.base/sun.nio.ch.Invoker.invokeUnchecked(Invoker.java:129)
        at java.base/sun.nio.ch.Invoker$2.run(Invoker.java:221)
        at java.base/sun.nio.ch.AsynchronousChannelGroupImpl$1.run(AsynchronousChannelGroupImpl.java:113)
        ... 3 more

[artembilan]

Confirmed.
Feel free to contribute the fix: https://github.com/spring-projects/spring-integration/blob/main/CONTRIBUTING.adoc

Thanks