INT-4183: Add SSL Handshake Timeout for TCP

JIRA: https://jira.spring.io/browse/INT-4183

Previously, this was hard-coded to 30 seconds.

* Fix typos according PR comments

spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java

@@ -124,6 +124,10 @@ public abstract class IpAdapterParserUtils {
 
 	public static final String MAPPER = "mapper";
 
+	public static final String READ_DELAY = "read-delay";
+
+	public static final String SSL_HANDSHAKE_TIMEOUT = "ssl-handshake-timeout";
+
 	private IpAdapterParserUtils() {
 	}
 

spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java

@@ -112,6 +112,8 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
 
 	private volatile TcpSSLContextSupport sslContextSupport;
 
+	private volatile Integer sslHandshakeTimeout;
+
 	private volatile TcpSocketSupport socketSupport = new DefaultTcpSocketSupport();
 
 	private volatile TcpNioConnectionSupport nioConnectionSupport;
@@ -166,6 +168,9 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
 				connectionFactory.setTcpNioConnectionSupport(this.obtainNioConnectionSupport());
 				this.connectionFactory = connectionFactory;
 			}
+			if (this.sslHandshakeTimeout != null) {
+				this.connectionFactory.setSslHandshakeTimeout(this.sslHandshakeTimeout);
+			}
 		}
 		else {
 			if (isServer()) {
@@ -497,6 +502,15 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean<Abstrac
 		this.applicationEventPublisher = applicationEventPublisher;
 	}
 
+	/**
+	 * Set the SSL handshake timeout (only used with SSL and NIO).
+	 * @param sslHandshakeTimeout the timeout.
+	 * @since 4.3.6
+	 */
+	public void setSslHandshakeTimeout(Integer sslHandshakeTimeout) {
+		this.sslHandshakeTimeout = sslHandshakeTimeout;
+	}
+
 	private boolean isClient() {
 		return "client".equals(this.type);
 	}

spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryParser.java

@@ -92,7 +92,10 @@ public class TcpConnectionFactoryParser extends AbstractBeanDefinitionParser {
 				IpAdapterParserUtils.SOCKET_SUPPORT);
 		IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,
 				IpAdapterParserUtils.MAPPER);
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "read-delay");
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
+				IpAdapterParserUtils.SSL_HANDSHAKE_TIMEOUT);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element,
+				IpAdapterParserUtils.READ_DELAY);
 
 		return builder.getBeanDefinition();
 	}

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java

@@ -127,6 +127,8 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
 
 	private volatile long readDelay = DEFAULT_READ_DELAY;
 
+	private volatile Integer sslHandshakeTimeout;
+
 	public AbstractConnectionFactory(int port) {
 		this.port = port;
 	}
@@ -456,6 +458,25 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
 		this.nioHarvestInterval = nioHarvestInterval;
 	}
 
+	/**
+	 * Set the handshake timeout used when waiting for SSL handshake data; only applies
+	 * to SSL connections, when using NIO.
+	 * @param sslHandshakeTimeout the timeout.
+	 * @since 4.3.6
+	 */
+	public void setSslHandshakeTimeout(int sslHandshakeTimeout) {
+		this.sslHandshakeTimeout = sslHandshakeTimeout;
+	}
+
+	/**
+	 * @return the handshake timeout.
+	 * @see #setSslHandshakeTimeout(int)
+	 * @since 4.3.6
+	 */
+	protected Integer getSslHandshakeTimeout() {
+		return this.sslHandshakeTimeout;
+	}
+
 	protected BlockingQueue<PendingIO> getDelayedReads() {
 		return this.delayedReads;
 	}

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java

@@ -88,6 +88,9 @@ public class TcpNioClientConnectionFactory extends
 				socketChannel, false, this.isLookupHost(), this.getApplicationEventPublisher(), this.getComponentName());
 		connection.setUsingDirectBuffers(this.usingDirectBuffers);
 		connection.setTaskExecutor(this.getTaskExecutor());
+		if (getSslHandshakeTimeout() != null && connection instanceof TcpNioSSLConnection) {
+			((TcpNioSSLConnection) connection).setHandshakeTimeout(getSslHandshakeTimeout());
+		}
 		TcpConnectionSupport wrappedConnection = wrapConnection(connection);
 		initializeConnection(wrappedConnection, socketChannel.socket());
 		socketChannel.configureBlocking(false);

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioSSLConnection.java

@@ -53,6 +53,8 @@ import org.springframework.util.Assert;
  */
 public class TcpNioSSLConnection extends TcpNioConnection {
 
+	private static final int DEFAULT_HANDSHAKE_TIMEOUT = 30;
+
 	private final SSLEngine sslEngine;
 
 	private volatile ByteBuffer decoded;
@@ -67,6 +69,8 @@ public class TcpNioSSLConnection extends TcpNioConnection {
 
 	private volatile boolean writerActive;
 
+	private volatile int handshakeTimeout = DEFAULT_HANDSHAKE_TIMEOUT;
+
 	private boolean needMoreNetworkData;
 
 	public TcpNioSSLConnection(SocketChannel socketChannel, boolean server, boolean lookupHost,
@@ -76,6 +80,15 @@ public class TcpNioSSLConnection extends TcpNioConnection {
 		this.sslEngine = sslEngine;
 	}
 
+	/**
+	 * Set the timeout while waiting for handshake data (in seconds). Default 30.
+	 * @param handshakeTimeout the timeout.
+	 * @since 4.3.6
+	 */
+	public void setHandshakeTimeout(int handshakeTimeout) {
+		this.handshakeTimeout = handshakeTimeout;
+	}
+
 	@Override
 	public SSLSession getSslSession() {
 		return this.sslEngine.getSession();
@@ -368,7 +381,8 @@ public class TcpNioSSLConnection extends TcpNioConnection {
 				if (logger.isTraceEnabled()) {
 					logger.trace("Writer waiting for handshake");
 				}
-				if (!TcpNioSSLConnection.this.semaphore.tryAcquire(30, TimeUnit.SECONDS)) {
+				if (!TcpNioSSLConnection.this.semaphore.tryAcquire(TcpNioSSLConnection.this.handshakeTimeout,
+						TimeUnit.SECONDS)) {
 					throw new MessagingException("SSL Handshaking taking too long");
 				}
 				if (logger.isTraceEnabled()) {

spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java

@@ -220,6 +220,9 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto
 				}
 				connection.setTaskExecutor(getTaskExecutor());
 				connection.setLastRead(now);
+				if (getSslHandshakeTimeout() != null && connection instanceof TcpNioSSLConnection) {
+					((TcpNioSSLConnection) connection).setHandshakeTimeout(getSslHandshakeTimeout());
+				}
 				this.channelMap.put(channel, connection);
 				channel.register(selector, SelectionKey.OP_READ, connection);
 				connection.publishConnectionOpenEvent();

spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-5.0.xsd

@@ -693,6 +693,14 @@
 					</xsd:appinfo>
 				</xsd:annotation>
 			</xsd:attribute>
+			<xsd:attribute name="ssl-handshake-timeout" type="xsd:string">
+				<xsd:annotation>
+					<xsd:documentation>
+						The timeout (in seconds) to use while performing handshakes on SSL sockets;
+						only applies when 'using-nio' is 'true'. Default: 30.
+					</xsd:documentation>
+				</xsd:annotation>
+			</xsd:attribute>
 			<xsd:attribute name="socket-support" type="xsd:string">
 				<xsd:annotation>
 					<xsd:documentation>

spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml

@@ -71,6 +71,7 @@
 		apply-sequence="true"
 		using-nio="true"
 		ssl-context-support="sslContextSupport"
+		ssl-handshake-timeout="43"
 		/>
 
 	<bean id="sslContextSupport" class="org.springframework.integration.ip.tcp.connection.DefaultTcpSSLContextSupport">
@@ -89,6 +90,17 @@
 		socket-support="socketSupport"
 		socket-factory-support="socketFactorySupport" />
 
+	<ip:tcp-connection-factory id="secureServerNio"
+		type="server"
+		port="#{tcpIpUtils.findAvailableServerSocket(5250)}"
+		lookup-host="false"
+		apply-sequence="true"
+		using-nio="true"
+		ssl-context-support="sslContextSupport"
+		ssl-handshake-timeout="34"
+		socket-support="socketSupport"
+		socket-factory-support="socketFactorySupport" />
+
 	<bean id="socketSupport" class="org.mockito.Mockito" factory-method="mock">
 		<constructor-arg value="org.springframework.integration.ip.tcp.connection.TcpSocketSupport" />
 	</bean>

spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests.java

@@ -256,6 +256,9 @@ public class ParserUnitTests {
 	@Autowired
 	TcpNetServerConnectionFactory secureServer;
 
+	@Autowired
+	TcpNioServerConnectionFactory secureServerNio;
+
 	@Autowired
 	TcpSocketFactorySupport socketFactorySupport;
 
@@ -343,6 +346,7 @@ public class ParserUnitTests {
 		Object connectionSupport = TestUtils.getPropertyValue(cfS1Nio, "tcpNioConnectionSupport");
 		assertTrue(connectionSupport instanceof DefaultTcpNioSSLConnectionSupport);
 		assertNotNull(TestUtils.getPropertyValue(connectionSupport, "sslContext"));
+		assertEquals(43, TestUtils.getPropertyValue(this.cfS1Nio, "sslHandshakeTimeout"));
 	}
 
 	@Test
@@ -656,6 +660,7 @@ public class ParserUnitTests {
 		DirectFieldAccessor dfa = new DirectFieldAccessor(secureServer);
 		assertSame(socketFactorySupport, dfa.getPropertyValue("tcpSocketFactorySupport"));
 		assertSame(socketSupport, dfa.getPropertyValue("tcpSocketSupport"));
+		assertEquals(34, TestUtils.getPropertyValue(this.secureServerNio, "sslHandshakeTimeout"));
 	}
 
 	public static class FooAdvice extends AbstractRequestHandlerAdvice {

spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/SocketSupportTests.java

@@ -34,6 +34,7 @@ import java.util.Map;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 
 import javax.net.ServerSocketFactory;
 import javax.net.SocketFactory;
@@ -295,6 +296,9 @@ Certificate fingerprints:
 		assertTrue(latch.await(10, TimeUnit.SECONDS));
 		assertEquals("Hello, world!", new String((byte[]) messages.get(0).getPayload()));
 		assertNotNull(messages.get(0).getHeaders().get("cipher"));
+
+		client.stop();
+		server.stop();
 	}
 
 	@Test
@@ -330,12 +334,16 @@ Certificate fingerprints:
 		connection.send(new GenericMessage<String>("Hello, world!"));
 		assertTrue(latch.await(10, TimeUnit.SECONDS));
 		assertEquals("Hello, world!", new String((byte[]) messages.get(0).getPayload()));
+
+		client.stop();
+		server.stop();
 	}
 
 	@Test
 	public void testNioClientAndServerSSL() throws Exception {
 		System.setProperty("javax.net.debug", "all"); // SSL activity in the console
 		TcpNioServerConnectionFactory server = new TcpNioServerConnectionFactory(0);
+		server.setSslHandshakeTimeout(43);
 		DefaultTcpSSLContextSupport sslContextSupport = new DefaultTcpSSLContextSupport("test.ks",
 				"test.truststore.ks", "secret", "secret");
 		sslContextSupport.setProtocol("SSL");
@@ -351,19 +359,36 @@ Certificate fingerprints:
 			return false;
 		});
 		server.setMapper(new SSLMapper());
+		final AtomicReference<String> serverConnectionId = new AtomicReference<>();
+		server.setApplicationEventPublisher(e -> {
+			if (e instanceof TcpConnectionOpenEvent) {
+				serverConnectionId.set(((TcpConnectionEvent) e).getConnectionId());
+			}
+		});
 		server.start();
 		TestingUtilities.waitListening(server, null);
 
 		TcpNioClientConnectionFactory client = new TcpNioClientConnectionFactory("localhost", server.getPort());
+		client.setSslHandshakeTimeout(34);
 		client.setTcpNioConnectionSupport(tcpNioConnectionSupport);
 		client.registerListener(message -> false);
+		client.setApplicationEventPublisher(e -> { });
 		client.start();
 
 		TcpConnection connection = client.getConnection();
+		assertEquals(34, TestUtils.getPropertyValue(connection, "handshakeTimeout"));
 		connection.send(new GenericMessage<String>("Hello, world!"));
 		assertTrue(latch.await(10, TimeUnit.SECONDS));
 		assertEquals("Hello, world!", new String((byte[]) messages.get(0).getPayload()));
 		assertNotNull(messages.get(0).getHeaders().get("cipher"));
+
+		Map<?, ?> connections = TestUtils.getPropertyValue(server, "connections", Map.class);
+		Object serverConnection = connections.get(serverConnectionId.get());
+		assertNotNull(serverConnection);
+		assertEquals(43, TestUtils.getPropertyValue(serverConnection, "handshakeTimeout"));
+
+		client.stop();
+		server.stop();
 	}
 
 	@Test
@@ -394,6 +419,12 @@ Certificate fingerprints:
 		ByteArrayCrLfSerializer deserializer = new ByteArrayCrLfSerializer();
 		deserializer.setMaxMessageSize(120000);
 		server.setDeserializer(deserializer);
+		final AtomicReference<String> serverConnectionId = new AtomicReference<>();
+		server.setApplicationEventPublisher(e -> {
+			if (e instanceof TcpConnectionOpenEvent) {
+				serverConnectionId.set(((TcpConnectionEvent) e).getConnectionId());
+			}
+		});
 		server.start();
 		TestingUtilities.waitListening(server, null);
 
@@ -410,9 +441,11 @@ Certificate fingerprints:
 			return false;
 		});
 		client.setDeserializer(deserializer);
+		client.setApplicationEventPublisher(e -> { });
 		client.start();
 
 		TcpConnection connection = client.getConnection();
+		assertEquals(30, TestUtils.getPropertyValue(connection, "handshakeTimeout"));
 		byte[] bytes = new byte[100000];
 		connection.send(new GenericMessage<String>("Hello, world!" + new String(bytes)));
 		assertTrue(latch.await(60, TimeUnit.SECONDS));
@@ -422,6 +455,14 @@ Certificate fingerprints:
 		payload = (byte[]) messages.get(1).getPayload();
 		assertEquals(13 + bytes.length, payload.length);
 		assertEquals("Hello, world!", new String(payload).substring(0, 13));
+
+		Map<?, ?> connections = TestUtils.getPropertyValue(server, "connections", Map.class);
+		Object serverConnection = connections.get(serverConnectionId.get());
+		assertNotNull(serverConnection);
+		assertEquals(30, TestUtils.getPropertyValue(serverConnection, "handshakeTimeout"));
+
+		client.stop();
+		server.stop();
 	}
 
 	private static class Replier implements TcpSender {

src/reference/asciidoc/ip.adoc

@@ -893,6 +893,9 @@ The `DefaulTcpSSLContextSupport` class also has an optional 'protocol' property,
 
 The keystore file names (first two constructor arguments) use the Spring `Resource` abstraction; by default the files will be located on the classpath, but this can be overridden by using the `file:` prefix, to find the files on the filesystem instead.
 
+Starting with _version 4.3.6_, when using NIO, you can specify an `ssl-handshake-timeout` (seconds) on the connection factory.
+This timeout (default 30) is used during SSL handshake when waiting for data; if the timeout is exceeded, the process is aborted and the socket closed.
+
 ==== Advanced Techniques
 
 In many cases, the configuration described above is all that is needed to enable secure communication over TCP/IP.