Enforce connection timeout in NIO TLS handshake

The handshake is in blocking mode and reading from a channel
in this mode does not honor so_timeout but using temporary
channels based on the socket input/output streams offers
a decent workaround for this stage.

Fixes #719

(cherry picked from commit 2751d46)

Conflicts:
	pom.xml

Author: acogoluegnes

pom.xml

@@ -63,7 +63,8 @@
     <mockito.version>4.0.0</mockito.version>
     <assertj.version>3.21.0</assertj.version>
     <jetty.version>9.4.44.v20210927</jetty.version>
-    <bouncycastle.version>1.69</bouncycastle.version>
+    <bouncycastle.version>1.70</bouncycastle.version>
+    <netcrusher.version>0.10</netcrusher.version>
 
     <maven.javadoc.plugin.version>3.2.0</maven.javadoc.plugin.version>
     <maven.release.plugin.version>2.5.3</maven.release.plugin.version>
@@ -759,6 +760,13 @@
       <version>${bouncycastle.version}</version>
       <scope>test</scope>
     </dependency>
+    <dependency>
+      <groupId>com.github.netcrusherorg</groupId>
+      <artifactId>netcrusher-core</artifactId>
+      <version>${netcrusher.version}</version>
+      <scope>test</scope>
+    </dependency>
+
   </dependencies>
 
   <build>

src/main/java/com/rabbitmq/client/impl/nio/SocketChannelFrameHandlerFactory.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2007-2022 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Java client library, is triple-licensed under the
 // Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2
@@ -21,6 +21,9 @@
 import com.rabbitmq.client.impl.AbstractFrameHandlerFactory;
 import com.rabbitmq.client.impl.FrameHandler;
 import com.rabbitmq.client.impl.TlsUtils;
+import java.nio.channels.Channels;
+import java.nio.channels.ReadableByteChannel;
+import java.nio.channels.WritableByteChannel;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -95,14 +98,23 @@ public FrameHandler create(Address addr, String connectionName) throws IOExcepti
 
             channel.connect(address);
 
+
             if (ssl) {
+                int initialSoTimeout = channel.socket().getSoTimeout();
+                channel.socket().setSoTimeout(this.connectionTimeout);
                 sslEngine.beginHandshake();
                 try {
-                    boolean handshake = SslEngineHelper.doHandshake(channel, sslEngine);
+                    ReadableByteChannel wrappedReadChannel = Channels.newChannel(
+                        channel.socket().getInputStream());
+                    WritableByteChannel wrappedWriteChannel = Channels.newChannel(
+                        channel.socket().getOutputStream());
+                    boolean handshake = SslEngineHelper.doHandshake(
+                        wrappedWriteChannel, wrappedReadChannel, sslEngine);
                     if (!handshake) {
                         LOGGER.error("TLS connection failed");
                         throw new SSLException("TLS handshake failed");
                     }
+                    channel.socket().setSoTimeout(initialSoTimeout);
                 } catch (SSLHandshakeException e) {
                     LOGGER.error("TLS connection failed: {}", e.getMessage());
                     throw e;

src/main/java/com/rabbitmq/client/impl/nio/SslEngineHelper.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2007-2021 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2007-2022 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Java client library, is triple-licensed under the
 // Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2
@@ -21,7 +21,6 @@
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.nio.channels.ReadableByteChannel;
-import java.nio.channels.SocketChannel;
 import java.nio.channels.WritableByteChannel;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -38,7 +37,7 @@ public class SslEngineHelper {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(SslEngineHelper.class);
 
-    public static boolean doHandshake(SocketChannel socketChannel, SSLEngine engine) throws IOException {
+    public static boolean doHandshake(WritableByteChannel writeChannel, ReadableByteChannel readChannel, SSLEngine engine) throws IOException {
 
         ByteBuffer plainOut = ByteBuffer.allocate(engine.getSession().getApplicationBufferSize());
         ByteBuffer plainIn = ByteBuffer.allocate(engine.getSession().getApplicationBufferSize());
@@ -58,11 +57,11 @@ public static boolean doHandshake(SocketChannel socketChannel, SSLEngine engine)
                 break;
             case NEED_UNWRAP:
                 LOGGER.debug("Unwrapping...");
-                handshakeStatus = unwrap(cipherIn, plainIn, socketChannel, engine);
+                handshakeStatus = unwrap(cipherIn, plainIn, readChannel, engine);
                 break;
             case NEED_WRAP:
                 LOGGER.debug("Wrapping...");
-                handshakeStatus = wrap(plainOut, cipherOut, socketChannel, engine);
+                handshakeStatus = wrap(plainOut, cipherOut, writeChannel, engine);
                 break;
             case FINISHED:
                 break;

src/test/java/com/rabbitmq/client/test/ssl/NioTlsUnverifiedConnection.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2007-2021 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2007-2022 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Java client library, is triple-licensed under the
 // Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2
@@ -15,31 +15,39 @@
 
 package com.rabbitmq.client.test.ssl;
 
-import com.rabbitmq.client.*;
+import static com.rabbitmq.client.test.TestUtils.basicGetBasicConsume;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.TrustEverythingTrustManager;
 import com.rabbitmq.client.impl.nio.NioParams;
 import com.rabbitmq.client.test.BrokerTestCase;
 import com.rabbitmq.client.test.TestUtils;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.SocketTimeoutException;
 import java.util.Collection;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
 import javax.net.ssl.TrustManager;
 import org.junit.Test;
+import org.netcrusher.core.reactor.NioReactor;
+import org.netcrusher.tcp.TcpCrusher;
+import org.netcrusher.tcp.TcpCrusherBuilder;
 import org.slf4j.LoggerFactory;
 
-import javax.net.ssl.SSLEngine;
-import java.io.IOException;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import static com.rabbitmq.client.test.TestUtils.basicGetBasicConsume;
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-
 /**
  *
  */
@@ -149,6 +157,59 @@ public void connectionGetConsumeProtocols() throws Exception {
         }
     }
 
+  @Test
+  public void connectionShouldEnforceConnectionTimeout() throws Exception {
+    int amqpPort = 5671; // assumes RabbitMQ server running on localhost;
+    int amqpProxyPort = TestUtils.randomNetworkPort();
+
+    int connectionTimeout = 3_000;
+    int handshakeTimeout = 1_000;
+
+    try (NioReactor reactor = new NioReactor();
+        TcpCrusher tcpProxy =
+            TcpCrusherBuilder.builder()
+                .withReactor(reactor)
+                .withBindAddress(new InetSocketAddress(amqpProxyPort))
+                .withConnectAddress("localhost", amqpPort)
+                .build()) {
+
+      tcpProxy.open();
+      tcpProxy.freeze();
+
+      ConnectionFactory factory = new ConnectionFactory();
+      factory.setHost("localhost");
+      factory.setPort(amqpProxyPort);
+
+      factory.useSslProtocol();
+      factory.useNio();
+
+      factory.setConnectionTimeout(connectionTimeout);
+      factory.setHandshakeTimeout(handshakeTimeout);
+
+      ExecutorService executorService = Executors.newSingleThreadExecutor();
+      try {
+        CountDownLatch latch = new CountDownLatch(1);
+        executorService.submit(
+            () -> {
+              try {
+                factory.newConnection();
+                latch.countDown();
+              } catch (SocketTimeoutException e) {
+                latch.countDown();
+              } catch (Exception e) {
+                // not supposed to happen
+              }
+            });
+
+        boolean connectionCreatedTimedOut = latch.await(10, TimeUnit.SECONDS);
+        assertThat(connectionCreatedTimedOut).isTrue();
+
+      } finally {
+        executorService.shutdownNow();
+      }
+    }
+  }
+
     private void sendAndVerifyMessage(int size) throws Exception {
         CountDownLatch latch = new CountDownLatch(1);
         boolean messageReceived = basicGetBasicConsume(connection, QUEUE, latch, size);