From 58f6c79338ab73305dfefb55ca2a151ac19b0b87 Mon Sep 17 00:00:00 2001
From: Guillaume Nodet <gnodet@gmail.com>
Date: Thu, 25 Jul 2024 12:34:58 +0200
Subject: [PATCH] Implement no-flow-control extension (fixes #539)

---
 CHANGES.md                                    |   1 +
 .../sshd/cli/client/ScpCommandMain.java       |  25 +-
 .../sshd/cli/client/SftpCommandMain.java      |  11 +-
 .../sshd/cli/client/SshClientCliSupport.java  |  13 +-
 .../kex/extension/parser/NoFlowControl.java   |   3 +
 .../sshd/common/channel/LocalWindow.java      |   5 +
 .../sshd/common/channel/RemoteWindow.java     |   5 +
 .../apache/sshd/common/channel/Window.java    |   5 +
 .../DefaultClientKexExtensionHandler.java     |  22 +-
 .../DefaultServerKexExtensionHandler.java     |  32 +-
 .../apache/sshd/common/session/Session.java   |   6 +
 .../helpers/AbstractConnectionService.java    |   7 +
 .../session/helpers/AbstractSession.java      |  18 +
 .../helpers/KeyExchangeMessageHandler.java    |   3 +-
 .../sshd/core/CoreModuleProperties.java       |   8 +
 .../org/apache/sshd/NoFlowControlTest.java    | 352 ++++++++++++++++++
 16 files changed, 503 insertions(+), 13 deletions(-)
 create mode 100644 sshd-core/src/test/java/org/apache/sshd/NoFlowControlTest.java

diff --git a/CHANGES.md b/CHANGES.md
index 09d2d4c15..6b7239fdc 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -43,6 +43,7 @@
 ## New Features
 
 * New utility methods `SftpClient.put(Path localFile, String remoteFileName)` and `SftpClient.put(InputStream in, String remoteFileName)` facilitate SFTP file uploading.
+* [GH-539](https://github.com/apache/mina-sshd/issues/539) Implement no-flow-control extension 
 
 ## Potential compatibility issues
 
diff --git a/sshd-cli/src/main/java/org/apache/sshd/cli/client/ScpCommandMain.java b/sshd-cli/src/main/java/org/apache/sshd/cli/client/ScpCommandMain.java
index 72102c8cf..977b7ce8f 100644
--- a/sshd-cli/src/main/java/org/apache/sshd/cli/client/ScpCommandMain.java
+++ b/sshd-cli/src/main/java/org/apache/sshd/cli/client/ScpCommandMain.java
@@ -32,6 +32,7 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.EnumSet;
+import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Set;
@@ -42,12 +43,14 @@
 import org.apache.sshd.client.auth.AuthenticationIdentitiesProvider;
 import org.apache.sshd.client.config.hosts.HostConfigEntry;
 import org.apache.sshd.client.session.ClientSession;
+import org.apache.sshd.common.PropertyResolver;
 import org.apache.sshd.common.SshConstants;
 import org.apache.sshd.common.session.Session;
 import org.apache.sshd.common.util.GenericUtils;
 import org.apache.sshd.common.util.ReflectionUtils;
 import org.apache.sshd.common.util.io.input.NoCloseInputStream;
 import org.apache.sshd.common.util.threads.ThreadUtils;
+import org.apache.sshd.core.CoreModuleProperties;
 import org.apache.sshd.scp.client.ScpClient;
 import org.apache.sshd.scp.client.ScpClient.Option;
 import org.apache.sshd.scp.client.ScpClientCreator;
@@ -61,6 +64,8 @@
 import org.apache.sshd.scp.common.helpers.ScpTimestampCommandDetails;
 import org.slf4j.Logger;
 
+import static org.apache.sshd.common.PropertyResolverUtils.toPropertyResolver;
+
 /**
  * @see    <A HREF="https://man7.org/linux/man-pages/man1/scp.1.html">SCP(1) - manual page</A>
  * @author <a href="mailto:dev@mina.apache.org">Apache MINA SSHD Project</a>
@@ -151,7 +156,7 @@ public static String[] normalizeCommandArguments(PrintStream stdout, PrintStream
             return null;
         }
 
-        return effective.toArray(new String[effective.size()]);
+        return effective.toArray(new String[0]);
     }
 
     /* -------------------------------------------------------------------------------- */
@@ -248,11 +253,11 @@ public static void showUsageMessage(PrintStream stderr) {
     public static void xferLocalToRemote(
             BufferedReader stdin, PrintStream stdout, PrintStream stderr, String[] args,
             ScpLocation source, ScpLocation target, Collection<Option> options,
-            OutputStream logStream, Level level, boolean quiet)
+            OutputStream logStream, Level level, boolean quiet, PropertyResolver defaultOptions)
             throws Exception {
         ScpClientCreator creator = resolveScpClientCreator(stderr, args);
         ClientSession session = ((logStream == null) || (creator == null) || GenericUtils.isEmpty(args))
-                ? null : setupClientSession(SCP_PORT_OPTION, stdin, level, stdout, stderr, args);
+                ? null : setupClientSession(SCP_PORT_OPTION, stdin, level, stdout, stderr, args, defaultOptions);
         if (session == null) {
             showUsageMessage(stderr);
             System.exit(-1);
@@ -330,10 +335,10 @@ private void logEvent(
     public static void xferRemoteToRemote(
             BufferedReader stdin, PrintStream stdout, PrintStream stderr, String[] args,
             ScpLocation source, ScpLocation target, Collection<Option> options,
-            OutputStream logStream, Level level, boolean quiet)
+            OutputStream logStream, Level level, boolean quiet, PropertyResolver defaultOptions)
             throws Exception {
         ClientSession srcSession = ((logStream == null) || GenericUtils.isEmpty(args))
-                ? null : setupClientSession(SCP_PORT_OPTION, stdin, level, stdout, stderr, args);
+                ? null : setupClientSession(SCP_PORT_OPTION, stdin, level, stdout, stderr, args, defaultOptions);
         if (srcSession == null) {
             showUsageMessage(stderr);
             System.exit(-1);
@@ -444,6 +449,10 @@ public static void main(String[] args) throws Exception {
                 new InputStreamReader(new NoCloseInputStream(System.in), Charset.defaultCharset()))) {
             args = normalizeCommandArguments(stdout, stderr, args);
 
+            PropertyResolver defaultOptions = toPropertyResolver(new HashMap<>());
+            CoreModuleProperties.NO_FLOW_CONTROL.set(defaultOptions, Boolean.TRUE);
+            CoreModuleProperties.WINDOW_SIZE.set(defaultOptions, 1024L * 1024L * 1024L);
+
             Level level = Level.SEVERE;
             int numArgs = GenericUtils.length(args);
             // see the way normalizeCommandArguments works...
@@ -472,9 +481,11 @@ public static void main(String[] args) throws Exception {
             }
 
             if (threeWay) {
-                xferRemoteToRemote(stdin, stdout, stderr, args, source, target, options, logStream, level, quiet);
+                xferRemoteToRemote(stdin, stdout, stderr, args, source, target, options, logStream, level, quiet,
+                        defaultOptions);
             } else {
-                xferLocalToRemote(stdin, stdout, stderr, args, source, target, options, logStream, level, quiet);
+                xferLocalToRemote(stdin, stdout, stderr, args, source, target, options, logStream, level, quiet,
+                        defaultOptions);
             }
         } finally {
             if ((logStream != stdout) && (logStream != stderr)) {
diff --git a/sshd-cli/src/main/java/org/apache/sshd/cli/client/SftpCommandMain.java b/sshd-cli/src/main/java/org/apache/sshd/cli/client/SftpCommandMain.java
index fde0e8868..6220a14bf 100644
--- a/sshd-cli/src/main/java/org/apache/sshd/cli/client/SftpCommandMain.java
+++ b/sshd-cli/src/main/java/org/apache/sshd/cli/client/SftpCommandMain.java
@@ -34,6 +34,7 @@
 import java.nio.file.Paths;
 import java.util.Arrays;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
 import java.util.ServiceLoader;
@@ -47,6 +48,7 @@
 import org.apache.sshd.client.ClientFactoryManager;
 import org.apache.sshd.client.session.ClientSession;
 import org.apache.sshd.common.NamedResource;
+import org.apache.sshd.common.PropertyResolver;
 import org.apache.sshd.common.ServiceFactory;
 import org.apache.sshd.common.channel.ChannelFactory;
 import org.apache.sshd.common.cipher.CipherFactory;
@@ -71,6 +73,7 @@
 import org.apache.sshd.common.util.io.output.LineLevelAppenderStream;
 import org.apache.sshd.common.util.io.output.NullOutputStream;
 import org.apache.sshd.common.util.threads.ThreadUtils;
+import org.apache.sshd.core.CoreModuleProperties;
 import org.apache.sshd.server.config.SshServerConfigFileReader;
 import org.apache.sshd.sftp.client.SftpClient;
 import org.apache.sshd.sftp.client.SftpClient.Attributes;
@@ -91,6 +94,8 @@
 import org.apache.sshd.sftp.common.extensions.openssh.StatVfsExtensionParser;
 import org.slf4j.Logger;
 
+import static org.apache.sshd.common.PropertyResolverUtils.toPropertyResolver;
+
 /**
  * TODO Add javadoc
  *
@@ -366,9 +371,13 @@ public static void main(String[] args) throws Exception {
                 setupLogging(level, stdout, stderr, logStream);
             }
 
+            PropertyResolver defaultOptions = toPropertyResolver(new HashMap<>());
+            CoreModuleProperties.NO_FLOW_CONTROL.set(defaultOptions, Boolean.TRUE);
+            CoreModuleProperties.WINDOW_SIZE.set(defaultOptions, 1024L * 1024L * 1024L);
+
             ClientSession session = (logStream == null)
                     ? null
-                    : setupClientSession(SFTP_PORT_OPTION, stdin, level, stdout, stderr, args);
+                    : setupClientSession(SFTP_PORT_OPTION, stdin, level, stdout, stderr, args, defaultOptions);
             if (session == null) {
                 System.err.println("usage: sftp [-v[v][v]] [-E logoutput] [-i identity] [-io nio2|mina|netty]"
                                    + " [-J proxyJump] [-l login] [" + SFTP_PORT_OPTION + " port] [-o option=value]"
diff --git a/sshd-cli/src/main/java/org/apache/sshd/cli/client/SshClientCliSupport.java b/sshd-cli/src/main/java/org/apache/sshd/cli/client/SshClientCliSupport.java
index 204e70743..9b9b7eabb 100644
--- a/sshd-cli/src/main/java/org/apache/sshd/cli/client/SshClientCliSupport.java
+++ b/sshd-cli/src/main/java/org/apache/sshd/cli/client/SshClientCliSupport.java
@@ -119,7 +119,15 @@ public static boolean isArgumentedOption(String portOption, String argName) {
     // CHECKSTYLE:OFF
     public static ClientSession setupClientSession(
             String portOption, BufferedReader stdin, Level level,
-            PrintStream stdout, PrintStream stderr, String... args)
+            PrintStream stdout, PrintStream stderr, String[] args)
+            throws Exception {
+        return setupClientSession(portOption, stdin, level, stdout, stderr, args, null);
+    }
+
+    public static ClientSession setupClientSession(
+            String portOption, BufferedReader stdin, Level level,
+            PrintStream stdout, PrintStream stderr, String[] args,
+            PropertyResolver defaultOptions)
             throws Exception {
         int port = -1;
         String host = null;
@@ -240,7 +248,8 @@ public static ClientSession setupClientSession(
             return null;
         }
 
-        PropertyResolver resolver = PropertyResolverUtils.toPropertyResolver(options);
+        PropertyResolver resolver = PropertyResolverUtils.toPropertyResolver(options,
+                defaultOptions);
         SshClient client = setupClient(
                 resolver, ciphers, macs, compressions, identities,
                 stdin, stdout, stderr, level, args);
diff --git a/sshd-common/src/main/java/org/apache/sshd/common/kex/extension/parser/NoFlowControl.java b/sshd-common/src/main/java/org/apache/sshd/common/kex/extension/parser/NoFlowControl.java
index 2c6990342..71774ac6b 100644
--- a/sshd-common/src/main/java/org/apache/sshd/common/kex/extension/parser/NoFlowControl.java
+++ b/sshd-common/src/main/java/org/apache/sshd/common/kex/extension/parser/NoFlowControl.java
@@ -31,6 +31,9 @@
 public class NoFlowControl extends AbstractKexExtensionParser<String> {
     public static final String NAME = "no-flow-control";
 
+    public static final String SUPPORTED = "s";
+    public static final String PREFERRED = "p";
+
     public static final NoFlowControl INSTANCE = new NoFlowControl();
 
     public NoFlowControl() {
diff --git a/sshd-core/src/main/java/org/apache/sshd/common/channel/LocalWindow.java b/sshd-core/src/main/java/org/apache/sshd/common/channel/LocalWindow.java
index ff46a2a03..afdd98fe3 100644
--- a/sshd-core/src/main/java/org/apache/sshd/common/channel/LocalWindow.java
+++ b/sshd-core/src/main/java/org/apache/sshd/common/channel/LocalWindow.java
@@ -70,6 +70,11 @@ public void consume(long len) throws IOException {
         BufferUtils.validateUint32Value(len, "Invalid consumption length: %d");
         checkInitialized("consume");
 
+        if (noFlowControl) {
+            // flow control is disabled, so just bail out
+            return;
+        }
+
         long remainLen;
         synchronized (lock) {
             remainLen = getSize() - len;
diff --git a/sshd-core/src/main/java/org/apache/sshd/common/channel/RemoteWindow.java b/sshd-core/src/main/java/org/apache/sshd/common/channel/RemoteWindow.java
index e92513eeb..d83b38435 100644
--- a/sshd-core/src/main/java/org/apache/sshd/common/channel/RemoteWindow.java
+++ b/sshd-core/src/main/java/org/apache/sshd/common/channel/RemoteWindow.java
@@ -67,6 +67,11 @@ public void consume(long len) {
         BufferUtils.validateUint32Value(len, "Invalid consumption length: %d");
         checkInitialized("consume");
 
+        if (noFlowControl) {
+            // flow control is disabled, so just bail out
+            return;
+        }
+
         long remainLen;
         synchronized (lock) {
             remainLen = getSize() - len;
diff --git a/sshd-core/src/main/java/org/apache/sshd/common/channel/Window.java b/sshd-core/src/main/java/org/apache/sshd/common/channel/Window.java
index 716f43abf..01fec44ed 100644
--- a/sshd-core/src/main/java/org/apache/sshd/common/channel/Window.java
+++ b/sshd-core/src/main/java/org/apache/sshd/common/channel/Window.java
@@ -25,6 +25,7 @@
 import java.util.function.Predicate;
 
 import org.apache.sshd.common.PropertyResolver;
+import org.apache.sshd.common.session.Session;
 import org.apache.sshd.common.util.ValidateUtils;
 import org.apache.sshd.common.util.buffer.BufferUtils;
 import org.apache.sshd.common.util.logging.AbstractLoggingBean;
@@ -46,6 +47,8 @@ public abstract class Window extends AbstractLoggingBean implements ChannelHolde
 
     protected final Object lock = new Object();
 
+    protected boolean noFlowControl;
+
     private final AtomicBoolean closed = new AtomicBoolean(false);
     private final AtomicBoolean initialized = new AtomicBoolean(false);
     private final Channel channelInstance;
@@ -94,6 +97,8 @@ protected void init(long size, long packetSize, PropertyResolver resolver) {
         }
 
         synchronized (lock) {
+            Session session = channelInstance.getSession(); // this should only be null during tests
+            this.noFlowControl = session != null && session.isNoFlowControl();
             this.maxSize = size;
             this.packetSize = packetSize;
             updateSize(size);
diff --git a/sshd-core/src/main/java/org/apache/sshd/common/kex/extension/DefaultClientKexExtensionHandler.java b/sshd-core/src/main/java/org/apache/sshd/common/kex/extension/DefaultClientKexExtensionHandler.java
index a0924c559..b089139a0 100644
--- a/sshd-core/src/main/java/org/apache/sshd/common/kex/extension/DefaultClientKexExtensionHandler.java
+++ b/sshd-core/src/main/java/org/apache/sshd/common/kex/extension/DefaultClientKexExtensionHandler.java
@@ -26,6 +26,7 @@
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.TreeSet;
 import java.util.function.BiConsumer;
@@ -33,11 +34,15 @@
 import org.apache.sshd.common.AttributeRepository.AttributeKey;
 import org.apache.sshd.common.NamedFactory;
 import org.apache.sshd.common.kex.extension.parser.HostBoundPubkeyAuthentication;
+import org.apache.sshd.common.kex.extension.parser.NoFlowControl;
 import org.apache.sshd.common.kex.extension.parser.ServerSignatureAlgorithms;
 import org.apache.sshd.common.session.Session;
+import org.apache.sshd.common.session.helpers.AbstractSession;
 import org.apache.sshd.common.signature.Signature;
+import org.apache.sshd.common.util.ValidateUtils;
 import org.apache.sshd.common.util.buffer.Buffer;
 import org.apache.sshd.common.util.logging.AbstractLoggingBean;
+import org.apache.sshd.core.CoreModuleProperties;
 
 /**
  * Detects if the server sends a
@@ -92,6 +97,15 @@ public boolean handleKexExtensionRequest(
             } else {
                 session.setAttribute(HOSTBOUND_AUTHENTICATION, version);
             }
+        } else if (NoFlowControl.NAME.equals(name)) {
+            String o = NoFlowControl.INSTANCE.parseExtension(data);
+            Optional<Boolean> nfc = CoreModuleProperties.NO_FLOW_CONTROL.get(session);
+            if (NoFlowControl.PREFERRED.equals(o) && nfc.orElse(Boolean.TRUE)
+                    || NoFlowControl.SUPPORTED.equals(o) && nfc.orElse(Boolean.FALSE)) {
+                AbstractSession abstractSession
+                        = ValidateUtils.checkInstanceOf(session, AbstractSession.class, "Not a supported session: %s", session);
+                abstractSession.activateNoFlowControl();
+            }
         }
         return true;
     }
@@ -157,7 +171,7 @@ public void sendKexExtensions(Session session, KexPhase phase) throws Exception
      * Collects extension info records, handing them off to the given {@code marshaller} for writing into an
      * {@link KexExtensions#SSH_MSG_EXT_INFO} message.
      * <p>
-     * This default implementation does not marshal any extension.
+     * This default implementation marshals a {@link NoFlowControl} extension}.
      * </p>
      *
      * @param session    {@link Session} to send the KEX extension information for
@@ -165,5 +179,11 @@ public void sendKexExtensions(Session session, KexPhase phase) throws Exception
      * @param marshaller {@link BiConsumer} writing the extensions into an SSH message
      */
     public void collectExtensions(Session session, KexPhase phase, BiConsumer<String, Object> marshaller) {
+        // no-flow-control
+        Boolean nfc = CoreModuleProperties.NO_FLOW_CONTROL.get(session).orElse(null);
+        if (nfc == null || nfc) {
+            marshaller.accept(NoFlowControl.NAME,
+                    nfc != null ? NoFlowControl.PREFERRED : NoFlowControl.SUPPORTED);
+        }
     }
 }
diff --git a/sshd-core/src/main/java/org/apache/sshd/common/kex/extension/DefaultServerKexExtensionHandler.java b/sshd-core/src/main/java/org/apache/sshd/common/kex/extension/DefaultServerKexExtensionHandler.java
index 7d5924622..47a2db70e 100644
--- a/sshd-core/src/main/java/org/apache/sshd/common/kex/extension/DefaultServerKexExtensionHandler.java
+++ b/sshd-core/src/main/java/org/apache/sshd/common/kex/extension/DefaultServerKexExtensionHandler.java
@@ -19,23 +19,29 @@
 
 package org.apache.sshd.common.kex.extension;
 
+import java.io.IOException;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.LinkedHashMap;
 import java.util.Map;
+import java.util.Optional;
 import java.util.function.BiConsumer;
 
 import org.apache.sshd.common.AttributeRepository.AttributeKey;
 import org.apache.sshd.common.kex.KexProposalOption;
+import org.apache.sshd.common.kex.extension.parser.NoFlowControl;
 import org.apache.sshd.common.kex.extension.parser.ServerSignatureAlgorithms;
 import org.apache.sshd.common.session.Session;
+import org.apache.sshd.common.session.helpers.AbstractSession;
 import org.apache.sshd.common.util.GenericUtils;
+import org.apache.sshd.common.util.ValidateUtils;
 import org.apache.sshd.common.util.buffer.Buffer;
 import org.apache.sshd.common.util.logging.AbstractLoggingBean;
+import org.apache.sshd.core.CoreModuleProperties;
 
 /**
  * A basic default implementation of a server-side {@link KexExtensionHandler} handling the
- * {@link ServerSignatureAlgorithms} KEX extension.
+ * {@link ServerSignatureAlgorithms} KEX extension along with the {@link NoFlowControl} extension.
  *
  * @see <a href="https://tools.ietf.org/html/rfc8308">RFC 8308</a>
  */
@@ -130,6 +136,23 @@ public void sendKexExtensions(Session session, KexPhase phase) throws Exception
         }
     }
 
+    @Override
+    public boolean handleKexExtensionRequest(
+            Session session, int index, int count, String name, byte[] data)
+            throws IOException {
+        if (NoFlowControl.NAME.equals(name)) {
+            String o = NoFlowControl.INSTANCE.parseExtension(data);
+            Optional<Boolean> nfc = CoreModuleProperties.NO_FLOW_CONTROL.get(session);
+            if (NoFlowControl.PREFERRED.equals(o) && nfc.orElse(Boolean.TRUE)
+                    || NoFlowControl.SUPPORTED.equals(o) && nfc.orElse(Boolean.FALSE)) {
+                AbstractSession abstractSession
+                        = ValidateUtils.checkInstanceOf(session, AbstractSession.class, "Not a supported session: %s", session);
+                abstractSession.activateNoFlowControl();
+            }
+        }
+        return true;
+    }
+
     /**
      * Collects extension info records, handing them off to the given {@code marshaller} for writing into an
      * {@link KexExtensions#SSH_MSG_EXT_INFO} message.
@@ -144,6 +167,7 @@ public void sendKexExtensions(Session session, KexPhase phase) throws Exception
      */
     @SuppressWarnings("javadoc")
     public void collectExtensions(Session session, KexPhase phase, BiConsumer<String, Object> marshaller) {
+        // server-sig-algs
         if (phase == KexPhase.NEWKEYS) {
             Collection<String> algorithms = session.getSignatureFactoriesNames();
             if (!GenericUtils.isEmpty(algorithms)) {
@@ -157,5 +181,11 @@ public void collectExtensions(Session session, KexPhase phase, BiConsumer<String
                         ServerSignatureAlgorithms.NAME);
             }
         }
+        // no-flow-control
+        Boolean nfc = CoreModuleProperties.NO_FLOW_CONTROL.get(session).orElse(null);
+        if (nfc != Boolean.FALSE) {
+            marshaller.accept(NoFlowControl.NAME,
+                    nfc != null ? NoFlowControl.PREFERRED : NoFlowControl.SUPPORTED);
+        }
     }
 }
diff --git a/sshd-core/src/main/java/org/apache/sshd/common/session/Session.java b/sshd-core/src/main/java/org/apache/sshd/common/session/Session.java
index 07eb80fd7..48ea9df86 100644
--- a/sshd-core/src/main/java/org/apache/sshd/common/session/Session.java
+++ b/sshd-core/src/main/java/org/apache/sshd/common/session/Session.java
@@ -382,4 +382,10 @@ static <T> T resolveAttribute(Session session, AttributeRepository.AttributeKey<
         T value = session.getAttribute(key);
         return (value != null) ? value : FactoryManager.resolveAttribute(session.getFactoryManager(), key);
     }
+
+    /**
+     * Check if the no-flow-control KEX extension has been activated.
+     */
+    boolean isNoFlowControl();
+
 }
diff --git a/sshd-core/src/main/java/org/apache/sshd/common/session/helpers/AbstractConnectionService.java b/sshd-core/src/main/java/org/apache/sshd/common/session/helpers/AbstractConnectionService.java
index 9251d651d..e3f42f662 100644
--- a/sshd-core/src/main/java/org/apache/sshd/common/session/helpers/AbstractConnectionService.java
+++ b/sshd-core/src/main/java/org/apache/sshd/common/session/helpers/AbstractConnectionService.java
@@ -757,6 +757,13 @@ protected void channelOpen(Buffer buffer) throws Exception {
             return;
         }
 
+        if (getSession().isNoFlowControl() && !channels.isEmpty()) {
+            // TODO add language tag configurable control
+            sendChannelOpenFailure(buffer, sender, SshConstants.SSH_OPEN_CONNECT_FAILED,
+                    "Only a single channel can be opened when using no-flow-control", "");
+            return;
+        }
+
         Session session = getSession();
         FactoryManager manager = Objects.requireNonNull(session.getFactoryManager(), "No factory manager");
         Channel channel = ChannelFactory.createChannel(session, manager.getChannelFactories(), type);
diff --git a/sshd-core/src/main/java/org/apache/sshd/common/session/helpers/AbstractSession.java b/sshd-core/src/main/java/org/apache/sshd/common/session/helpers/AbstractSession.java
index 31589a4bf..12a325994 100644
--- a/sshd-core/src/main/java/org/apache/sshd/common/session/helpers/AbstractSession.java
+++ b/sshd-core/src/main/java/org/apache/sshd/common/session/helpers/AbstractSession.java
@@ -304,6 +304,8 @@ public abstract class AbstractSession extends SessionHelper {
     private byte[] clientKexData; // the payload of the client's SSH_MSG_KEXINIT
     private byte[] serverKexData; // the payload of the server's SSH_MSG_KEXINIT
 
+    private boolean noFlowControl;
+
     /**
      * Create a new session.
      *
@@ -2769,6 +2771,22 @@ protected abstract void receiveKexInit(
             Map<KexProposalOption, String> proposal, byte[] seed)
             throws IOException;
 
+    /**
+     * Activate the no-flow-control KEX extension specified by https://tools.ietf.org/html/rfc8308#section-3.3
+     */
+    public void activateNoFlowControl() {
+        this.noFlowControl = true;
+        log.info("activateNoFlowControl({}): deactivating flow control", this);
+    }
+
+    /**
+     * Check if the no-flow-control KEX extension has been activated.
+     */
+    @Override
+    public boolean isNoFlowControl() {
+        return noFlowControl;
+    }
+
     /**
      * Retrieve the SSH session from the I/O session. If the session has not been attached, an exception will be thrown
      *
diff --git a/sshd-core/src/main/java/org/apache/sshd/common/session/helpers/KeyExchangeMessageHandler.java b/sshd-core/src/main/java/org/apache/sshd/common/session/helpers/KeyExchangeMessageHandler.java
index a5853e06e..03c400963 100644
--- a/sshd-core/src/main/java/org/apache/sshd/common/session/helpers/KeyExchangeMessageHandler.java
+++ b/sshd-core/src/main/java/org/apache/sshd/common/session/helpers/KeyExchangeMessageHandler.java
@@ -40,6 +40,7 @@
 import org.apache.sshd.common.io.AbstractIoWriteFuture;
 import org.apache.sshd.common.io.IoWriteFuture;
 import org.apache.sshd.common.kex.KexState;
+import org.apache.sshd.common.kex.extension.KexExtensions;
 import org.apache.sshd.common.util.ExceptionUtils;
 import org.apache.sshd.common.util.ValidateUtils;
 import org.apache.sshd.common.util.buffer.Buffer;
@@ -236,7 +237,7 @@ public IoWriteFuture writePacket(Buffer buffer, long timeout, TimeUnit unit) thr
         int cmd = bufData[buffer.rpos()] & 0xFF;
         boolean enqueued = false;
         boolean isLowLevelMessage = cmd <= SshConstants.SSH_MSG_KEX_LAST && cmd != SshConstants.SSH_MSG_SERVICE_REQUEST
-                && cmd != SshConstants.SSH_MSG_SERVICE_ACCEPT;
+                && cmd != SshConstants.SSH_MSG_SERVICE_ACCEPT && cmd != KexExtensions.SSH_MSG_EXT_INFO;
         IoWriteFuture future = null;
         try {
             if (isLowLevelMessage) {
diff --git a/sshd-core/src/main/java/org/apache/sshd/core/CoreModuleProperties.java b/sshd-core/src/main/java/org/apache/sshd/core/CoreModuleProperties.java
index d8c8d35c1..3b6f4e4e5 100644
--- a/sshd-core/src/main/java/org/apache/sshd/core/CoreModuleProperties.java
+++ b/sshd-core/src/main/java/org/apache/sshd/core/CoreModuleProperties.java
@@ -26,6 +26,7 @@
 import org.apache.sshd.common.Property;
 import org.apache.sshd.common.SshConstants;
 import org.apache.sshd.common.channel.Channel;
+import org.apache.sshd.common.kex.extension.parser.NoFlowControl;
 import org.apache.sshd.common.session.Session;
 import org.apache.sshd.common.util.OsUtils;
 import org.apache.sshd.common.util.ValidateUtils;
@@ -800,6 +801,13 @@ public final class CoreModuleProperties {
         }
     });
 
+    /**
+     * Configuration value to enable {@code no-flow-control}. When set to {@code true}, the option will be used if the
+     * connected peer also supports it. A value of {@code false} disables the {@code no-flow-control} completely, while
+     * the default {@code null} value, will support it, but not request it.
+     */
+    public static final Property<Boolean> NO_FLOW_CONTROL = Property.bool(NoFlowControl.NAME);
+
     private CoreModuleProperties() {
         throw new UnsupportedOperationException("No instance");
     }
diff --git a/sshd-core/src/test/java/org/apache/sshd/NoFlowControlTest.java b/sshd-core/src/test/java/org/apache/sshd/NoFlowControlTest.java
new file mode 100644
index 000000000..b19f3808b
--- /dev/null
+++ b/sshd-core/src/test/java/org/apache/sshd/NoFlowControlTest.java
@@ -0,0 +1,352 @@
+/*
+ * 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.sshd;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Collection;
+import java.util.Deque;
+import java.util.EnumSet;
+import java.util.LinkedList;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.sshd.client.SshClient;
+import org.apache.sshd.client.channel.ClientChannel;
+import org.apache.sshd.client.channel.ClientChannelEvent;
+import org.apache.sshd.client.session.ClientSession;
+import org.apache.sshd.common.io.IoInputStream;
+import org.apache.sshd.common.io.IoOutputStream;
+import org.apache.sshd.common.io.WritePendingException;
+import org.apache.sshd.common.util.buffer.Buffer;
+import org.apache.sshd.common.util.buffer.ByteArrayBuffer;
+import org.apache.sshd.common.util.io.IoUtils;
+import org.apache.sshd.common.util.io.output.NoCloseOutputStream;
+import org.apache.sshd.common.util.logging.AbstractLoggingBean;
+import org.apache.sshd.common.util.threads.ThreadUtils;
+import org.apache.sshd.core.CoreModuleProperties;
+import org.apache.sshd.server.Environment;
+import org.apache.sshd.server.ExitCallback;
+import org.apache.sshd.server.SshServer;
+import org.apache.sshd.server.channel.ChannelSession;
+import org.apache.sshd.server.command.AsyncCommand;
+import org.apache.sshd.server.keyprovider.SimpleGeneratorHostKeyProvider;
+import org.apache.sshd.util.test.BaseTestSupport;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.FixMethodOrder;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.Timeout;
+import org.junit.runners.MethodSorters;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Test the no-flow-control extension.
+ */
+@FixMethodOrder(MethodSorters.NAME_ASCENDING)
+public class NoFlowControlTest extends BaseTestSupport {
+
+    public static final byte END_FILE = '#';
+    public static final int BIG_MSG_SEND_COUNT = 10000;
+
+    @Rule
+    public Timeout timeout = Timeout.seconds(TimeUnit.MINUTES.toSeconds(6));
+
+    private SshServer sshServer;
+    private int port;
+
+    public NoFlowControlTest() {
+        super();
+    }
+
+    @Before
+    public void setUp() throws Exception {
+        sshServer = setupTestServer();
+
+        byte[] msg = IoUtils.toByteArray(getClass().getResourceAsStream("/big-msg.txt"));
+        sshServer.setShellFactory(
+                channel -> new FloodingAsyncCommand(msg, BIG_MSG_SEND_COUNT, END_FILE));
+
+        sshServer.setKeyPairProvider(new SimpleGeneratorHostKeyProvider());
+        sshServer.start();
+        port = sshServer.getPort();
+    }
+
+    @After
+    public void tearDown() throws Exception {
+        if (sshServer != null) {
+            sshServer.stop();
+            sshServer.close(true);
+        }
+    }
+
+    @Test
+    public void testNoFlowControl() throws Exception {
+        try (SshClient client = setupTestClient()) {
+            CoreModuleProperties.NO_FLOW_CONTROL.set(client, true);
+            client.start();
+
+            try (ClientSession session = client.connect(getCurrentTestName(), TEST_LOCALHOST, port)
+                    .verify(CONNECT_TIMEOUT).getSession()) {
+                session.addPasswordIdentity(getCurrentTestName());
+                session.auth().verify(AUTH_TIMEOUT);
+
+                try (ClientChannel channel = session.createShellChannel()) {
+                    channel.setOut(new VerifyingOutputStream(channel, END_FILE));
+                    channel.setErr(new NoCloseOutputStream(System.err));
+                    channel.open().verify(OPEN_TIMEOUT);
+
+                    assertTrue(session.isNoFlowControl());
+                    assertTrue(sshServer.getActiveSessions().get(0).isNoFlowControl());
+
+                    Collection<ClientChannelEvent> result
+                            = channel.waitFor(EnumSet.of(ClientChannelEvent.CLOSED), TimeUnit.MINUTES.toMillis(2L));
+                    assertFalse("Timeout while waiting for channel closure", result.contains(ClientChannelEvent.TIMEOUT));
+                }
+            } finally {
+                client.stop();
+            }
+        }
+    }
+
+    /**
+     * Read all incoming data and if END_FILE symbol is detected, kill client session to end test
+     */
+    private static class VerifyingOutputStream extends OutputStream {
+        private final Logger log;
+        private final ClientChannel channel;
+        private final byte eofSignal;
+
+        VerifyingOutputStream(ClientChannel channel, final byte eofSignal) {
+            this.log = LoggerFactory.getLogger(getClass());
+            this.channel = channel;
+            this.eofSignal = eofSignal;
+        }
+
+        @Override
+        public void write(int b) throws IOException {
+            if (channel.isClosed() || channel.isClosing()) {
+                throw new IOException("Channel (" + channel + ") is closing / closed on write single byte");
+            }
+
+            if (b == (eofSignal & 0xff)) {
+                log.info("Closing channel (" + channel + ") due to single byte EOF");
+                channel.close(true);
+            }
+        }
+
+        @Override
+        public void write(byte[] b, int off, int len) throws IOException {
+            if (channel.isClosed() || channel.isClosing()) {
+                throw new IOException("Channel (" + channel + ") is closing / closed on write " + len + " bytes");
+            }
+
+            if (len <= 0) {
+                return;
+            }
+
+            int lastPos = off + len - 1;
+            if ((b[lastPos] & 0xFF) == (eofSignal & 0xFF)) {
+                log.info("Closing channel (" + channel + ") due to last byte EOF");
+                channel.close(true);
+            }
+        }
+    }
+
+    public static final class FloodingAsyncCommand extends AbstractLoggingBean implements AsyncCommand {
+        private static final AtomicInteger POOL_COUNT = new AtomicInteger(0);
+
+        private final AtomicReference<ExecutorService> executorHolder = new AtomicReference<>();
+        private final AtomicReference<Future<?>> futureHolder = new AtomicReference<>();
+
+        private AsyncInPendingWrapper pendingWrapper;
+        private byte[] msg;
+        private int sendCount;
+        private byte eofSignal;
+
+        public FloodingAsyncCommand(final byte[] msg, final int sendCount, final byte eofSignal) {
+            this.msg = msg;
+            this.sendCount = sendCount;
+            this.eofSignal = eofSignal;
+        }
+
+        @Override
+        public void setIoInputStream(IoInputStream in) {
+            // ignored
+        }
+
+        @Override
+        public void setIoOutputStream(IoOutputStream out) {
+            pendingWrapper = new AsyncInPendingWrapper(out);
+        }
+
+        @Override
+        public void setIoErrorStream(IoOutputStream err) {
+            // ignored
+        }
+
+        @Override
+        public void setInputStream(InputStream in) {
+            // ignored
+        }
+
+        @Override
+        public void setOutputStream(OutputStream out) {
+            // ignored
+        }
+
+        @Override
+        public void setErrorStream(OutputStream err) {
+            // ignored
+        }
+
+        @Override
+        public void setExitCallback(ExitCallback callback) {
+            // ignored
+        }
+
+        @Override
+        public void start(ChannelSession channel, Environment env) throws IOException {
+            log.info("Starting");
+
+            ExecutorService service
+                    = ThreadUtils.newSingleThreadExecutor(getClass().getSimpleName() + "-" + POOL_COUNT.incrementAndGet());
+            executorHolder.set(service);
+
+            futureHolder.set(service.submit((Runnable) () -> {
+                log.info("Start heavy load sending " + sendCount + " messages of " + msg.length + " bytes");
+                for (int i = 0; i < sendCount; i++) {
+                    try {
+                        pendingWrapper.write(new ByteArrayBuffer(msg));
+                    } catch (IOException e) {
+                        log.error("Failed ({}) to send message #{}/{}: {}",
+                                e.getClass().getSimpleName(), i + 1, sendCount, e.getMessage());
+                        throw new RuntimeException(e);
+                    }
+                }
+                log.info("Sending EOF signal");
+
+                try {
+                    pendingWrapper.write(new ByteArrayBuffer(new byte[] { eofSignal }));
+                } catch (IOException e) {
+                    log.error("Failed ({}) to send EOF message after {} messages: {}",
+                            e.getClass().getSimpleName(), sendCount, e.getMessage());
+                    throw new RuntimeException(e);
+                }
+            }));
+            log.info("Started");
+        }
+
+        @Override
+        public void destroy(ChannelSession channel) {
+            log.info("Destroying");
+
+            Future<?> future = futureHolder.getAndSet(null);
+            if ((future != null) && (!future.isDone())) {
+                log.info("Cancelling");
+                future.cancel(true);
+            }
+
+            ExecutorService service = executorHolder.getAndSet(null);
+            if ((service != null) && (!service.isShutdown())) {
+                log.info("Shutdown");
+                service.shutdownNow();
+            }
+        }
+    }
+
+    /**
+     * Wrapper for asyncIn stream that catches Pending exception and queues the pending messages for later retry (send
+     * after previous messages were fully transfered)
+     */
+    private static class AsyncInPendingWrapper extends AbstractLoggingBean {
+        private IoOutputStream asyncIn;
+
+        // Order has to be preserved for queued writes
+        private final Deque<Buffer> pending = new LinkedList<>();
+
+        AsyncInPendingWrapper(IoOutputStream out) {
+            this.asyncIn = out;
+        }
+
+        public synchronized void write(Object msg) throws IOException {
+            if ((asyncIn != null) && (!asyncIn.isClosed()) && (!asyncIn.isClosing())) {
+                Buffer byteBufferMsg = (Buffer) msg;
+                if (!pending.isEmpty()) {
+                    queueRequest(byteBufferMsg);
+                    return;
+                }
+
+                writeWithPendingDetection(byteBufferMsg, false);
+            }
+        }
+
+        private void writeWithPendingDetection(Buffer msg, boolean wasPending) throws IOException {
+            try {
+                asyncIn.writeBuffer(msg).addListener(future -> {
+                    if (future.isWritten()) {
+                        try {
+                            if (wasPending) {
+                                synchronized (AsyncInPendingWrapper.this) {
+                                    pending.remove();
+                                }
+                            }
+                            writePendingIfAny();
+                        } catch (Throwable e) {
+                            log.error("Failed ({}) to re-write pending: {}", e.getClass().getSimpleName(), e.getMessage());
+                        }
+                    } else {
+                        Throwable t = future.getException();
+                        log.warn("Failed to write message", t);
+                    }
+                });
+            } catch (WritePendingException e) {
+                if (!wasPending) {
+                    queueRequest(msg);
+                }
+            } catch (Throwable t) {
+                log.error("Failed to write or queue", t);
+                if (t instanceof IOException) {
+                    throw (IOException) t;
+                }
+                throw new IOException("Failed to write or queue", t);
+            }
+        }
+
+        private synchronized void writePendingIfAny() throws IOException {
+            if (pending.peek() == null) {
+                return;
+            }
+
+            Buffer msg = pending.peek();
+            writeWithPendingDetection(msg, true);
+        }
+
+        private void queueRequest(Buffer msg) {
+            msg.rpos(0);
+            pending.add(msg);
+        }
+    }
+}