Fix hierynomus#805: Prevent CHANNEL_CLOSE to be sent between Channel.isOpen and a Transport.write call

Otherwise, a disconnect with a "packet referred to nonexistent channel" message can occur.

This particularly happens when the transport.Reader thread passes an eof from the server to the ChannelInputStream, the reading library-user thread returns, and closes the channel at the same time as the transport.Reader thread receives the subsequent CHANNEL_CLOSE from the server.

Author: kegelh

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

@@ -304,6 +304,25 @@ public boolean isOpen() {
         }
     }
 
+    // Prevent CHANNEL_CLOSE to be sent between isOpen and a Transport.write call in the runnable, otherwise
+    // a disconnect with a "packet referred to nonexistent channel" message can occur.
+    //
+    // This particularly happens when the transport.Reader thread passes an eof from the server to the
+    // ChannelInputStream, the reading library-user thread returns, and closes the channel at the same time as the
+    // transport.Reader thread receives the subsequent CHANNEL_CLOSE from the server.
+    boolean whileOpen(TransportRunnable runnable) throws TransportException, ConnectionException {
+        openCloseLock.lock();
+        try {
+            if (isOpen()) {
+                runnable.run();
+                return true;
+            }
+        } finally {
+            openCloseLock.unlock();
+        }
+        return false;
+    }
+
     private void gotChannelRequest(SSHPacket buf)
             throws ConnectionException, TransportException {
         final String reqType;
@@ -427,5 +446,8 @@ public String toString() {
                 + rwin + " >";
     }
 
+    public interface TransportRunnable {
+        void run() throws TransportException, ConnectionException;
+    }
 
 }

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

@@ -30,7 +30,7 @@
  */
 public final class ChannelOutputStream extends OutputStream implements ErrorNotifiable {
 
-    private final Channel chan;
+    private final AbstractChannel chan;
     private final Transport trans;
     private final Window.Remote win;
 
@@ -47,6 +47,12 @@ private final class DataBuffer {
 
         private final SSHPacket packet = new SSHPacket(Message.CHANNEL_DATA);
         private final Buffer.PlainBuffer leftOvers = new Buffer.PlainBuffer();
+        private final AbstractChannel.TransportRunnable packetWriteRunnable = new AbstractChannel.TransportRunnable() {
+            @Override
+            public void run() throws TransportException {
+                trans.write(packet);
+            }
+        };
 
         DataBuffer() {
             headerOffset = packet.rpos();
@@ -99,8 +105,9 @@ boolean flush(int bufferSize, boolean canAwaitExpansion) throws TransportExcepti
                 if (leftOverBytes > 0) {
                     leftOvers.putRawBytes(packet.array(), packet.wpos(), leftOverBytes);
                 }
-
-                trans.write(packet);
+                if (!chan.whileOpen(packetWriteRunnable)) {
+                    throwStreamClosed();
+                }
                 win.consume(writeNow);
 
                 packet.rpos(headerOffset);
@@ -119,7 +126,7 @@ boolean flush(int bufferSize, boolean canAwaitExpansion) throws TransportExcepti
 
     }
 
-    public ChannelOutputStream(Channel chan, Transport trans, Window.Remote win) {
+    public ChannelOutputStream(AbstractChannel chan, Transport trans, Window.Remote win) {
         this.chan = chan;
         this.trans = trans;
         this.win = win;
@@ -157,17 +164,22 @@ private void checkClose() throws SSHException {
             if (error != null) {
                 throw error;
             } else {
-                throw new ConnectionException("Stream closed");
+                throwStreamClosed();
             }
         }
     }
 
     @Override
     public synchronized void close() throws IOException {
         // Not closed yet, and underlying channel is open to flush the data to.
-        if (!closed.getAndSet(true) && chan.isOpen()) {
-            buffer.flush(false);
-            trans.write(new SSHPacket(Message.CHANNEL_EOF).putUInt32(chan.getRecipient()));
+        if (!closed.getAndSet(true)) {
+            chan.whileOpen(new AbstractChannel.TransportRunnable() {
+                @Override
+                public void run() throws TransportException, ConnectionException {
+                    buffer.flush(false);
+                    trans.write(new SSHPacket(Message.CHANNEL_EOF).putUInt32(chan.getRecipient()));
+                }
+            });
         }
     }
 
@@ -188,4 +200,7 @@ public String toString() {
         return "< ChannelOutputStream for Channel #" + chan.getID() + " >";
     }
 
+    private static void throwStreamClosed() throws ConnectionException {
+        throw new ConnectionException("Stream closed");
+    }
 }