This is an automated email from the ASF dual-hosted git repository.
bamaer pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new 0c4bcc5bf9 ssh, attach inputstream on connection, fixes #9159 (#9160)
0c4bcc5bf9 is described below
commit 0c4bcc5bf903aecc0dd6a601ae6975a4e0027381
Author: Hans Van Akelyen <[email protected]>
AuthorDate: Fri Oct 30 07:41:54 2026 +0100
ssh, attach inputstream on connection, fixes #9159 (#9160)
---
.../hop/vfs/sftp/provider/SftpStreamProxy.java | 12 ++--
.../hop/vfs/sftp/provider/SftpStreamProxyTest.java | 71 ++++++++++++++++++++++
.../hop/pipeline/transforms/ssh/SessionResult.java | 16 +++--
.../pipeline/transforms/ssh/SessionResultTest.java | 22 +++++++
4 files changed, 111 insertions(+), 10 deletions(-)
diff --git
a/plugins/tech/sftp/src/main/java/org/apache/hop/vfs/sftp/provider/SftpStreamProxy.java
b/plugins/tech/sftp/src/main/java/org/apache/hop/vfs/sftp/provider/SftpStreamProxy.java
index 5d648df6dc..5d8b074155 100644
---
a/plugins/tech/sftp/src/main/java/org/apache/hop/vfs/sftp/provider/SftpStreamProxy.java
+++
b/plugins/tech/sftp/src/main/java/org/apache/hop/vfs/sftp/provider/SftpStreamProxy.java
@@ -44,6 +44,9 @@ public class SftpStreamProxy implements Proxy {
private ChannelExec channel;
+ /** The output of the proxy command, attached before the channel connects. */
+ private InputStream inputStream;
+
/**
* Command pattern to execute on the proxy host.
*
@@ -135,16 +138,15 @@ public class SftpStreamProxy implements Proxy {
proxyOptions);
channel = (ChannelExec) session.openChannel("exec");
channel.setCommand(String.format(commandFormat, targetHost, targetPort));
+ // JSch drops output that arrives before a stream is attached, such as the
target server's SSH
+ // greeting: attach it before connecting
+ inputStream = channel.getInputStream();
channel.connect(timeout);
}
@Override
public InputStream getInputStream() {
- try {
- return channel.getInputStream();
- } catch (final IOException e) {
- throw new IllegalStateException("IOException getting the SSH proxy input
stream", e);
- }
+ return inputStream;
}
@Override
diff --git
a/plugins/tech/sftp/src/test/java/org/apache/hop/vfs/sftp/provider/SftpStreamProxyTest.java
b/plugins/tech/sftp/src/test/java/org/apache/hop/vfs/sftp/provider/SftpStreamProxyTest.java
new file mode 100644
index 0000000000..ff1c6bc498
--- /dev/null
+++
b/plugins/tech/sftp/src/test/java/org/apache/hop/vfs/sftp/provider/SftpStreamProxyTest.java
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.hop.vfs.sftp.provider;
+
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.jcraft.jsch.ChannelExec;
+import com.jcraft.jsch.Session;
+import java.io.ByteArrayInputStream;
+import java.io.InputStream;
+import org.apache.commons.vfs2.FileSystemOptions;
+import org.junit.jupiter.api.Test;
+import org.mockito.InOrder;
+import org.mockito.MockedStatic;
+
+class SftpStreamProxyTest {
+
+ @Test
+ void connectAttachesTheInputStreamBeforeConnectingTheChannel() throws
Exception {
+ // JSch drops output that arrives before a stream is attached: the target
server's SSH greeting
+ // could be lost before the session read it
+ Session session = mock(Session.class);
+ ChannelExec channel = mock(ChannelExec.class);
+ InputStream input = new ByteArrayInputStream(new byte[0]);
+ when(session.openChannel("exec")).thenReturn(channel);
+ when(channel.getInputStream()).thenReturn(input);
+
+ SftpStreamProxy proxy =
+ new SftpStreamProxy(
+ SftpStreamProxy.NETCAT_COMMAND, "user", "proxy", 22, "secret", new
FileSystemOptions());
+ try (MockedStatic<SftpClientFactory> factory =
mockStatic(SftpClientFactory.class)) {
+ factory
+ .when(
+ () ->
+ SftpClientFactory.createConnection(
+ anyString(), anyInt(), any(char[].class),
any(char[].class), any()))
+ .thenReturn(session);
+
+ proxy.connect(null, "target", 2222, 1000);
+ }
+
+ verify(channel).setCommand("nc -q 0 target 2222");
+ InOrder order = inOrder(channel);
+ order.verify(channel).getInputStream();
+ order.verify(channel).connect(1000);
+ assertSame(input, proxy.getInputStream());
+ }
+}
diff --git
a/plugins/transforms/ssh/src/main/java/org/apache/hop/pipeline/transforms/ssh/SessionResult.java
b/plugins/transforms/ssh/src/main/java/org/apache/hop/pipeline/transforms/ssh/SessionResult.java
index 666bcfbdf5..e3e77f40da 100644
---
a/plugins/transforms/ssh/src/main/java/org/apache/hop/pipeline/transforms/ssh/SessionResult.java
+++
b/plugins/transforms/ssh/src/main/java/org/apache/hop/pipeline/transforms/ssh/SessionResult.java
@@ -75,11 +75,14 @@ public class SessionResult {
try {
channel = (ChannelExec) session.openChannel("exec");
channel.setCommand(command);
+ // JSch drops output that arrives before a stream is attached: attach
them before connecting
+ InputStream isOut = channel.getInputStream();
+ InputStream isErr = channel.getErrStream();
channel.connect();
SessionResult result = new SessionResult();
- result.readFromChannel(channel, commandTimeoutMs, stopped, clock);
+ result.readFromChannel(channel, isOut, isErr, commandTimeoutMs, stopped,
clock);
return result;
- } catch (JSchException e) {
+ } catch (JSchException | IOException e) {
throw new HopException(e);
} finally {
if (channel != null) {
@@ -100,11 +103,14 @@ public class SessionResult {
}
private void readFromChannel(
- ChannelExec channel, long commandTimeoutMs, BooleanSupplier stopped,
LongSupplier clock)
+ ChannelExec channel,
+ InputStream isOut,
+ InputStream isErr,
+ long commandTimeoutMs,
+ BooleanSupplier stopped,
+ LongSupplier clock)
throws HopException {
try {
- InputStream isOut = channel.getInputStream();
- InputStream isErr = channel.getErrStream();
byte[] buffer = new byte[8192];
StringBuilder stdout = new StringBuilder();
StringBuilder stderr = new StringBuilder();
diff --git
a/plugins/transforms/ssh/src/test/java/org/apache/hop/pipeline/transforms/ssh/SessionResultTest.java
b/plugins/transforms/ssh/src/test/java/org/apache/hop/pipeline/transforms/ssh/SessionResultTest.java
index 5fc5625c9b..b2ed25274e 100644
---
a/plugins/transforms/ssh/src/test/java/org/apache/hop/pipeline/transforms/ssh/SessionResultTest.java
+++
b/plugins/transforms/ssh/src/test/java/org/apache/hop/pipeline/transforms/ssh/SessionResultTest.java
@@ -23,6 +23,7 @@ import static
org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -37,6 +38,7 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.function.LongSupplier;
import org.apache.hop.core.exception.HopException;
import org.junit.jupiter.api.Test;
+import org.mockito.InOrder;
class SessionResultTest {
@@ -68,6 +70,26 @@ class SessionResultTest {
verify(channel).disconnect();
}
+ @Test
+ void executeCommandAttachesTheStreamsBeforeConnecting() throws Exception {
+ // JSch drops output that arrives before a stream is attached: a fast
command lost its output
+ Session session = mock(Session.class);
+ ChannelExec channel = mock(ChannelExec.class);
+ when(session.openChannel("exec")).thenReturn(channel);
+ when(channel.getInputStream()).thenReturn(stream("out\n"));
+ when(channel.getErrStream()).thenReturn(stream(""));
+ when(channel.isClosed()).thenReturn(true);
+
+ SessionResult.executeCommand(session, "echo out");
+
+ InOrder order = inOrder(channel);
+ order.verify(channel).getInputStream();
+ order.verify(channel).connect();
+ order = inOrder(channel);
+ order.verify(channel).getErrStream();
+ order.verify(channel).connect();
+ }
+
@Test
void executeCommandWrapsJSchException() throws Exception {
Session session = mock(Session.class);
