Make ChannelInputStream respect connection timeout (#1034)

* Make ChannelInputStream respect connection timeout

* If wait ended with a timeout throw exception

* Do not throw timeout exception in eof state

* Change synchronization from synchronized/wait/notifyAll to ReentrantLock and Condition for safely identify timeouts

* Source channel read timeout from Config instead of connection timeout

Introduce Config#getChannelReadTimeoutMs and getChannelErrorReadTimeoutMs
(both defaulting to 0) so stream read timeouts are configured independently
of the connection-level timeout.

Additional fixes on top of the original PR:
- Restore the 3-arg ChannelInputStream constructor for binary compatibility
- Fix notifyError idempotency (restores if (!eof) guard)
- Move eof check in receive() inside the lock for consistent locking discipline
- Throw SocketTimeoutException on timeout instead of IOException

* Add tests

---------

Co-authored-by: Jeroen van Erp <jeroen@hierynomus.com>

Author: jpstotz

src/main/java/net/schmizz/sshj/Config.java

@@ -204,4 +204,12 @@ public interface Config {
     int getMaxCircularBufferSize();
 
     void setMaxCircularBufferSize(int maxCircularBufferSize);
+
+    int getChannelReadTimeoutMs();
+
+    void setChannelReadTimeoutMs(int channelReadTimeoutMs);
+
+    int getChannelErrorReadTimeoutMs();
+
+    void setChannelErrorReadTimeoutMs(int channelErrorReadTimeoutMs);
 }

src/main/java/net/schmizz/sshj/ConfigImpl.java

@@ -51,6 +51,8 @@ public class ConfigImpl
     private boolean verifyHostKeyCertificates = true;
     // HF-982: default to 16MB buffers.
     private int maxCircularBufferSize = 16 * 1024 * 1024;
+    private int channelReadTimeoutMs = 0;
+    private int channelErrorReadTimeoutMs = 0;
 
     @Override
     public List<Factory.Named<Cipher>> getCipherFactories() {
@@ -187,6 +189,26 @@ public void setMaxCircularBufferSize(int maxCircularBufferSize) {
         this.maxCircularBufferSize = maxCircularBufferSize;
     }
 
+    @Override
+    public int getChannelReadTimeoutMs() {
+        return channelReadTimeoutMs;
+    }
+
+    @Override
+    public void setChannelReadTimeoutMs(int channelReadTimeoutMs) {
+        this.channelReadTimeoutMs = channelReadTimeoutMs;
+    }
+
+    @Override
+    public int getChannelErrorReadTimeoutMs() {
+        return channelErrorReadTimeoutMs;
+    }
+
+    @Override
+    public void setChannelErrorReadTimeoutMs(int channelErrorReadTimeoutMs) {
+        this.channelErrorReadTimeoutMs = channelErrorReadTimeoutMs;
+    }
+
     @Override
     public void setLoggerFactory(LoggerFactory loggerFactory) {
         this.loggerFactory = loggerFactory;

src/main/java/net/schmizz/sshj/connection/channel/AbstractChannel.java

@@ -95,7 +95,7 @@ protected AbstractChannel(Connection conn, String type, Charset remoteCharset) {
         id = conn.nextID();
 
         lwin = new Window.Local(conn.getWindowSize(), conn.getMaxPacketSize(), loggerFactory);
-        in = new ChannelInputStream(this, trans, lwin);
+        in = new ChannelInputStream(this, trans, lwin, trans.getConfig().getChannelReadTimeoutMs());
 
         openEvent = new Event<ConnectionException>("chan#" + id + " / " + "open", ConnectionException.chainer, openCloseLock, loggerFactory);
         closeEvent = new Event<ConnectionException>("chan#" + id + " / " + "close", ConnectionException.chainer, openCloseLock, loggerFactory);

src/main/java/net/schmizz/sshj/connection/channel/ChannelInputStream.java

@@ -24,6 +24,10 @@
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.InterruptedIOException;
+import java.net.SocketTimeoutException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
 
 /**
  * {@link InputStream} for channels. Can {@link #receive(byte[], int, int) receive} data into its buffer for serving to
@@ -38,25 +42,37 @@ public final class ChannelInputStream
     private final Channel chan;
     private final Transport trans;
     private final Window.Local win;
+    private final int timeoutMs;
     private final CircularBuffer.PlainCircularBuffer buf;
+    private final ReentrantLock lock = new ReentrantLock();
+    private final Condition dataArrived = lock.newCondition();
     private final byte[] b = new byte[1];
 
     private boolean eof;
     private SSHException error;
 
+    /** Creates a non-timeout channel input stream. Reads will block indefinitely until data arrives or EOF. */
     public ChannelInputStream(Channel chan, Transport trans, Window.Local win) {
+        this(chan, trans, win, 0);
+    }
+
+    public ChannelInputStream(Channel chan, Transport trans, Window.Local win, int timeoutMs) {
         this.chan = chan;
         this.log = chan.getLoggerFactory().getLogger(getClass());
         this.trans = trans;
         this.win = win;
+        this.timeoutMs = timeoutMs;
         this.buf = new CircularBuffer.PlainCircularBuffer(
             chan.getLocalMaxPacketSize(), trans.getConfig().getMaxCircularBufferSize());
     }
 
     @Override
     public int available() {
-        synchronized (buf) {
+        lock.lock();
+        try {
             return buf.available();
+        } finally {
+            lock.unlock();
         }
     }
 
@@ -66,18 +82,29 @@ public void close() {
     }
 
     public void eof() {
-        synchronized (buf) {
+        lock.lock();
+        try {
             if (!eof) {
                 eof = true;
-                buf.notifyAll();
+                dataArrived.signalAll();
             }
+        } finally {
+            lock.unlock();
         }
     }
 
     @Override
-    public synchronized void notifyError(SSHException error) {
-        this.error = error;
-        eof();
+    public void notifyError(SSHException error) {
+        lock.lock();
+        try {
+            if (!eof) {
+                this.error = error;
+                eof = true;
+                dataArrived.signalAll();
+            }
+        } finally {
+            lock.unlock();
+        }
     }
 
     @Override
@@ -91,11 +118,9 @@ public int read()
     @Override
     public int read(byte[] b, int off, int len)
             throws IOException {
-        synchronized (buf) {
-            for (; ; ) {
-                if (buf.available() > 0) {
-                    break;
-                }
+        lock.lock();
+        try {
+            while (buf.available() == 0) {
                 if (eof) {
                     if (error != null) {
                         throw error;
@@ -104,39 +129,51 @@ public int read(byte[] b, int off, int len)
                     }
                 }
                 try {
-                    buf.wait();
+                    if (timeoutMs > 0) {
+                        if (!dataArrived.await(timeoutMs, TimeUnit.MILLISECONDS)) {
+                            throw new SocketTimeoutException("Timeout of " + timeoutMs + "ms while waiting for data");
+                        }
+                    } else {
+                        dataArrived.await();
+                    }
                 } catch (InterruptedException e) {
                     Thread.currentThread().interrupt();
                     throw (IOException) new InterruptedIOException().initCause(e);
                 }
             }
-            if (len > buf.available()) {
-                len = buf.available();
+            int available = buf.available();
+            if (len > available) {
+                len = available;
             }
             buf.readRawBytes(b, off, len);
 
             if (!chan.getAutoExpand()) {
                 checkWindow();
             }
+        } finally {
+            lock.unlock();
         }
 
         return len;
     }
 
     public void receive(byte[] data, int offset, int len) throws SSHException {
-        if (eof) {
-            throw new ConnectionException("Getting data on EOF'ed stream");
-        }
-        synchronized (buf) {
+        lock.lock();
+        try {
+            if (eof) {
+                throw new ConnectionException("Getting data on EOF'ed stream");
+            }
             buf.putRawBytes(data, offset, len);
-            buf.notifyAll();
+            dataArrived.signalAll();
             // Potential fix for #203 (window consumed below 0).
             // This seems to be a race condition if we receive more data, while we're already sending a SSH_MSG_CHANNEL_WINDOW_ADJUST
             // And the window has not expanded yet.
             win.consume(len);
             if (chan.getAutoExpand()) {
                 checkWindow();
             }
+        } finally {
+            lock.unlock();
         }
     }
 

src/main/java/net/schmizz/sshj/connection/channel/direct/SessionChannel.java

@@ -32,7 +32,7 @@ public class SessionChannel
         extends AbstractDirectChannel
         implements Session, Session.Command, Session.Shell, Session.Subsystem {
 
-    private final ChannelInputStream err = new ChannelInputStream(this, trans, lwin);
+    private final ChannelInputStream err = new ChannelInputStream(this, trans, lwin, trans.getConfig().getChannelErrorReadTimeoutMs());
 
     private volatile Integer exitStatus;
 

src/test/java/net/schmizz/sshj/connection/channel/ChannelInputStreamTest.java

@@ -0,0 +1,126 @@
+/*
+ * Copyright (C)2009 - SSHJ Contributors
+ *
+ * Licensed 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 net.schmizz.sshj.connection.channel;
+
+import net.schmizz.sshj.Config;
+import net.schmizz.sshj.common.LoggerFactory;
+import net.schmizz.sshj.common.SSHException;
+import net.schmizz.sshj.transport.Transport;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.net.SocketTimeoutException;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+public class ChannelInputStreamTest {
+
+    private Channel chan;
+    private Window.Local win;
+    private ScheduledExecutorService scheduler;
+
+    @BeforeEach
+    void setUp() {
+        Config config = mock(Config.class);
+        when(config.getMaxCircularBufferSize()).thenReturn(16 * 1024 * 1024);
+
+        Transport trans = mock(Transport.class);
+        when(trans.getConfig()).thenReturn(config);
+
+        chan = mock(Channel.class);
+        when(chan.getLoggerFactory()).thenReturn(LoggerFactory.DEFAULT);
+        when(chan.getLocalMaxPacketSize()).thenReturn(32768);
+        when(chan.getAutoExpand()).thenReturn(false);
+
+        win = new Window.Local(2097152, 32768, LoggerFactory.DEFAULT);
+        scheduler = Executors.newSingleThreadScheduledExecutor();
+    }
+
+    @AfterEach
+    void tearDown() {
+        scheduler.shutdownNow();
+    }
+
+    @Test
+    void timeoutFiresWhenNoDataArrives() {
+        ChannelInputStream stream = newStream(100);
+        assertThrows(SocketTimeoutException.class, () -> stream.read(new byte[8], 0, 8));
+    }
+
+    @Test
+    void eofBeforeTimeoutReturnsMinusOne() throws Exception {
+        ChannelInputStream stream = newStream(500);
+        scheduleAfter(50, stream::eof);
+        assertEquals(-1, stream.read(new byte[8], 0, 8));
+    }
+
+    @Test
+    void dataArrivesBeforeTimeout() throws Exception {
+        ChannelInputStream stream = newStream(500);
+        byte[] data = "hello".getBytes();
+        scheduleAfter(50, () -> stream.receive(data, 0, data.length));
+
+        byte[] buf = new byte[data.length];
+        int read = stream.read(buf, 0, buf.length);
+        assertEquals(data.length, read);
+        assertArrayEquals(data, buf);
+    }
+
+    @Test
+    void notifyErrorThrowsBeforeTimeout() throws Exception {
+        ChannelInputStream stream = newStream(500);
+        SSHException error = new SSHException("remote error");
+        scheduleAfter(50, () -> stream.notifyError(error));
+
+        SSHException thrown = assertThrows(SSHException.class, () -> stream.read(new byte[8], 0, 8));
+        assertSame(error, thrown);
+    }
+
+    @Test
+    void zeroTimeoutBlocksUntilEof() throws Exception {
+        ChannelInputStream stream = newStream(0);
+        scheduleAfter(100, stream::eof);
+        assertEquals(-1, stream.read(new byte[8], 0, 8));
+    }
+
+    private ChannelInputStream newStream(int timeoutMs) {
+        Config config = mock(Config.class);
+        when(config.getMaxCircularBufferSize()).thenReturn(16 * 1024 * 1024);
+        Transport trans = mock(Transport.class);
+        when(trans.getConfig()).thenReturn(config);
+        return new ChannelInputStream(chan, trans, win, timeoutMs);
+    }
+
+    @FunctionalInterface
+    private interface ThrowingRunnable {
+        void run() throws Exception;
+    }
+
+    private void scheduleAfter(long delayMs, ThrowingRunnable action) {
+        scheduler.schedule(() -> {
+            try {
+                action.run();
+            } catch (Exception e) {
+                throw new RuntimeException(e);
+            }
+        }, delayMs, TimeUnit.MILLISECONDS);
+    }
+}