[apacheGH-445] Added StrictKexTest

CHANGES.md

@@ -56,7 +56,18 @@ Provide (read-only) public access to internal session state values related to KE
 * *getSessionKexDetails*
 * *getSessionCountersDetails*
 
+### Added `SessionListener#sessionPeerIdentificationSent` callback
+
+Invoked after successful or failed attempt to send client/server identification to peer. The callback provides the failure error if such occurred.
+
 ## Potential compatibility issues
 
+### Added finite wait time for default implementation of `ClientSession#executeRemoteCommand`
+
+* `CoreModuleProperties#EXEC_CHANNEL_OPEN_TIMEOUT` - default = 30 seconds.
+* `CoreModuleProperties#EXEC_CHANNEL_CMD_TIMEOUT` - default = 30 seconds.
+
+This may cause failures for code that was running long execution commands using the default method implementations.
+
 ## Major Code Re-factoring
 

sshd-common/src/test/java/org/apache/sshd/util/test/JUnitTestSupport.java

@@ -703,7 +703,7 @@ public static void outputDebugMessage(String format, Object o) {
 
     public static void outputDebugMessage(String format, Object... args) {
         if (OUTPUT_DEBUG_MESSAGES) {
-            outputDebugMessage(String.format(format, args));
+            outputDebugMessage(GenericUtils.isEmpty(args) ? format : String.format(format, args));
         }
     }
 
@@ -713,6 +713,24 @@ public static void outputDebugMessage(Object message) {
         }
     }
 
+    public static void failWithWrittenErrorMessage(String format, Object... args) {
+        failWithWrittenErrorMessage(GenericUtils.isEmpty(args) ? format : String.format(format, args));
+    }
+
+    public static void failWithWrittenErrorMessage(Object message) {
+        writeErrorMessage(message);
+        fail(Objects.toString(message));
+    }
+
+    public static void writeErrorMessage(String format, Object... args) {
+        writeErrorMessage(GenericUtils.isEmpty(args) ? format : String.format(format, args));
+    }
+
+    public static void writeErrorMessage(Object message) {
+        System.err.append("===[ERROR]=== ").println(message);
+        System.err.flush();
+    }
+
     /* ---------------------------------------------------------------------------- */
 
     public static void replaceJULLoggers() {

sshd-core/src/main/java/org/apache/sshd/client/session/ClientSession.java

@@ -59,6 +59,7 @@
 import org.apache.sshd.common.util.io.output.NoCloseOutputStream;
 import org.apache.sshd.common.util.io.output.NullOutputStream;
 import org.apache.sshd.common.util.net.SshdSocketAddress;
+import org.apache.sshd.core.CoreModuleProperties;
 
 /**
  * <P>
@@ -304,10 +305,12 @@ default void executeRemoteCommand(
              ClientChannel channel = createExecChannel(command)) {
             channel.setOut(channelOut);
             channel.setErr(channelErr);
-            channel.open().await(); // TODO use verify and a configurable timeout
 
-            // TODO use a configurable timeout
-            Collection<ClientChannelEvent> waitMask = channel.waitFor(REMOTE_COMMAND_WAIT_EVENTS, 0L);
+            Duration openTimeout = CoreModuleProperties.EXEC_CHANNEL_OPEN_TIMEOUT.getRequired(channel);
+            channel.open().verify(openTimeout);
+
+            Duration execTimeout = CoreModuleProperties.EXEC_CHANNEL_CMD_TIMEOUT.getRequired(channel);
+            Collection<ClientChannelEvent> waitMask = channel.waitFor(REMOTE_COMMAND_WAIT_EVENTS, execTimeout);
             if (waitMask.contains(ClientChannelEvent.TIMEOUT)) {
                 throw new SocketTimeoutException("Failed to retrieve command result in time: " + command);
             }

sshd-core/src/main/java/org/apache/sshd/client/session/ClientSessionImpl.java

@@ -74,6 +74,7 @@ public class ClientSessionImpl extends AbstractClientSession {
 
     public ClientSessionImpl(ClientFactoryManager client, IoSession ioSession) throws Exception {
         super(client, ioSession);
+
         if (log.isDebugEnabled()) {
             log.debug("Client session created: {}", ioSession);
         }

sshd-core/src/main/java/org/apache/sshd/common/session/SessionListener.java

@@ -102,6 +102,21 @@ default void sessionPeerIdentificationLine(
             Session session, String line, List<String> extraLines) {
         // ignored
     }
+    /**
+     * Identification sent to peer
+     *
+     * @param session    The {@link Session} instance
+     * @param version    The resolved identification version
+     * @param extraLines Extra data preceding the identification to be sent. <B>Note:</B> the list is modifiable only if
+     *                   this is a server session. The user may modify it based on the peer.
+     * @param error {@code null} if sending was successful
+     * @see              <A HREF="https://tools.ietf.org/html/rfc4253#section-4.2">RFC 4253 - section 4.2 - Protocol
+     *                   Version Exchange</A>
+     */
+    default void sessionPeerIdentificationSent(
+            Session session, String version, List<String> extraLines, Throwable error) {
+        // ignored
+    }
 
     /**
      * The peer's identification version was received

sshd-core/src/main/java/org/apache/sshd/common/session/helpers/AbstractSession.java

@@ -601,7 +601,8 @@ protected void doHandleMessage(Buffer buffer) throws Exception {
                 && CoreModuleProperties.USE_STRICT_KEX.getRequired(this)
                 && (cmd != SshConstants.SSH_MSG_KEXINIT)) {
             log.error("doHandleMessage({}) invalid 1st message: {}", this, SshConstants.getCommandMessageName(cmd));
-            throw new SshException(SshConstants.SSH2_DISCONNECT_PROTOCOL_ERROR, "Strict KEX Error");
+            disconnect(SshConstants.SSH2_DISCONNECT_PROTOCOL_ERROR, "Strict KEX Error");
+            return;
         }
 
         if (log.isDebugEnabled()) {

sshd-core/src/main/java/org/apache/sshd/common/session/helpers/SessionHelper.java

@@ -49,6 +49,7 @@
 import org.apache.sshd.common.channel.throttle.ChannelStreamWriterResolverManager;
 import org.apache.sshd.common.digest.Digest;
 import org.apache.sshd.common.forward.Forwarder;
+import org.apache.sshd.common.future.SshFutureListener;
 import org.apache.sshd.common.io.IoSession;
 import org.apache.sshd.common.io.IoWriteFuture;
 import org.apache.sshd.common.kex.AbstractKexFactoryManager;
@@ -77,6 +78,7 @@
 /**
  * Contains split code in order to make {@link AbstractSession} class smaller
  */
+@SuppressWarnings("checkstyle:MethodCount")
 public abstract class SessionHelper extends AbstractKexFactoryManager implements Session {
 
     // Session timeout measurements
@@ -628,6 +630,31 @@ protected void signalSendIdentification(SessionListener listener, String version
         listener.sessionPeerIdentificationSend(this, version, extraLines);
     }
 
+    protected void signalIdentificationSent(String version, List<String> extraLines, Throwable error) throws Exception {
+        try {
+            invokeSessionSignaller(l -> {
+                signalIdentificationSent(l, version, extraLines, error);
+                return null;
+            });
+        } catch (Throwable err) {
+            Throwable e = ExceptionUtils.peelException(err);
+            if (e instanceof Exception) {
+                throw (Exception) e;
+            } else {
+                throw new RuntimeSshException(e);
+            }
+        }
+    }
+
+    protected void signalIdentificationSent(
+            SessionListener listener, String version, List<String> extraLines, Throwable err) throws Exception {
+        if (listener == null) {
+            return;
+        }
+
+        listener.sessionPeerIdentificationSent(this, version, extraLines, err);
+    }
+
     protected void signalReadPeerIdentificationLine(String line, List<String> extraLines) throws Exception {
         try {
             invokeSessionSignaller(l -> {
@@ -833,28 +860,45 @@ protected IoWriteFuture sendIdentification(String version, List<String> extraLin
         ReservedSessionMessagesHandler handler = getReservedSessionMessagesHandler();
         IoWriteFuture future = (handler == null) ? null : handler.sendIdentification(this, version, extraLines);
         boolean debugEnabled = log.isDebugEnabled();
-        if (future != null) {
+        if (future == null) {
+            String ident = version;
+            if (GenericUtils.size(extraLines) > 0) {
+                ident = GenericUtils.join(extraLines, "\r\n") + "\r\n" + version;
+            }
+
+            if (debugEnabled) {
+                log.debug("sendIdentification({}): {}",
+                        this, ident.replace('\r', '|').replace('\n', '|'));
+            }
+
+            IoSession networkSession = getIoSession();
+            byte[] data = (ident + "\r\n").getBytes(StandardCharsets.UTF_8);
+            future = networkSession.writeBuffer(new ByteArrayBuffer(data));
+        } else {
             if (debugEnabled) {
                 log.debug("sendIdentification({})[{}] sent {} lines via reserved handler",
                         this, version, GenericUtils.size(extraLines));
             }
-
-            return future;
         }
 
-        String ident = version;
-        if (GenericUtils.size(extraLines) > 0) {
-            ident = GenericUtils.join(extraLines, "\r\n") + "\r\n" + version;
-        }
+        future.addListener(new SshFutureListener<IoWriteFuture>() {
+            @Override
+            public void operationComplete(IoWriteFuture future) {
+                try {
+                    signalIdentificationSent(version, extraLines, future.getException());
+                } catch(Throwable err) {
+                    Throwable e = ExceptionUtils.peelException(err);
+                    if (e instanceof RuntimeException) {
+                        throw (RuntimeException) e;
+                    } else {
+                        throw new RuntimeSshException(e);
+                    }
 
-        if (debugEnabled) {
-            log.debug("sendIdentification({}): {}",
-                    this, ident.replace('\r', '|').replace('\n', '|'));
-        }
+                }
+            }
+        });
 
-        IoSession networkSession = getIoSession();
-        byte[] data = (ident + "\r\n").getBytes(StandardCharsets.UTF_8);
-        return networkSession.writeBuffer(new ByteArrayBuffer(data));
+        return future;
     }
 
     /**

sshd-core/src/main/java/org/apache/sshd/core/CoreModuleProperties.java

@@ -61,6 +61,21 @@ public final class CoreModuleProperties {
     public static final Property<Duration> CHANNEL_OPEN_TIMEOUT
             = Property.duration("ssh-agent-server-channel-open-timeout", Duration.ofSeconds(30));
 
+    /**
+     * Value that can be set on the {@link org.apache.sshd.common.FactoryManager} the session or the channel to
+     * configure the channel open timeout value (millis) for executing a remote command using default implementation.
+     */
+    public static final Property<Duration> EXEC_CHANNEL_OPEN_TIMEOUT
+            = Property.duration("ssh-exec-channel-open-timeout", Duration.ofSeconds(30));
+
+    /**
+     * Value that can be set on the {@link org.apache.sshd.common.FactoryManager} the session or the channel to
+     * configure the channel command execution timeout value (millis) for executing a remote command using default
+     * implementation.
+     */
+    public static final Property<Duration> EXEC_CHANNEL_CMD_TIMEOUT
+            = Property.duration("ssh-exec-channel-cmd-timeout", Duration.ofSeconds(30));
+
     /**
      * Value used to configure the type of proxy forwarding channel to be used. See also
      * https://tools.ietf.org/html/draft-ietf-secsh-agent-02

sshd-core/src/test/java/org/apache/sshd/client/ClientTest.java

@@ -1510,7 +1510,6 @@ public void testKeyboardInteractiveInSessionUserInteractiveFailure() throws Exce
         CoreModuleProperties.PASSWORD_PROMPTS.set(client, maxPrompts);
         AtomicInteger numberOfRequests = new AtomicInteger();
         UserAuthKeyboardInteractiveFactory auth = new UserAuthKeyboardInteractiveFactory() {
-
             @Override
             public UserAuthKeyboardInteractive createUserAuth(ClientSession session) throws IOException {
                 return new UserAuthKeyboardInteractive() {