diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java index c909c3e67c..57c878a100 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/IpAdapterParserUtils.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2012 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -113,6 +113,12 @@ public abstract class IpAdapterParserUtils { public static final String SCHEDULER = "scheduler"; + public static final String SSL_CONTEXT_SUPPORT = "ssl-context-support"; + + public static final String SOCKET_SUPPORT = "socket-support"; + + public static final String SOCKET_FACTORY_SUPPORT = "socket-factory-support"; + private IpAdapterParserUtils() {} /** diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java index 9434e8b7aa..f56ed05bb7 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/config/TcpConnectionFactoryFactoryBean.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2012 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -31,12 +31,22 @@ import org.springframework.integration.ip.tcp.connection.TcpNetClientConnectionF import org.springframework.integration.ip.tcp.connection.TcpNetServerConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpNioClientConnectionFactory; import org.springframework.integration.ip.tcp.connection.TcpNioServerConnectionFactory; +import org.springframework.integration.ip.tcp.connection.support.DefaultTcpNetSSLSocketFactorySupport; +import org.springframework.integration.ip.tcp.connection.support.DefaultTcpNetSocketFactorySupport; +import org.springframework.integration.ip.tcp.connection.support.DefaultTcpNioConnectionSupport; +import org.springframework.integration.ip.tcp.connection.support.DefaultTcpNioSSLConnectionSupport; +import org.springframework.integration.ip.tcp.connection.support.DefaultTcpSocketSupport; +import org.springframework.integration.ip.tcp.connection.support.TcpNioConnectionSupport; +import org.springframework.integration.ip.tcp.connection.support.TcpSSLContextSupport; +import org.springframework.integration.ip.tcp.connection.support.TcpSocketFactorySupport; +import org.springframework.integration.ip.tcp.connection.support.TcpSocketSupport; import org.springframework.integration.ip.tcp.serializer.ByteArrayCrLfSerializer; +import org.springframework.util.Assert; /** * Instantiates a TcpN(et|io)(Server|Client)ConnectionFactory, depending * on type and using-nio attributes. - * + * * @author Gary Russell * @since 2.0.5 */ @@ -90,9 +100,17 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean getObjectType() { - return this.connectionFactory != null ? this.connectionFactory.getClass() + return this.connectionFactory != null ? this.connectionFactory.getClass() : AbstractConnectionFactory.class; } @@ -104,12 +122,14 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean deserializer) { + Assert.notNull(deserializer, "Deserializer may not be null"); this.deserializer = deserializer; } @@ -273,6 +325,7 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean serializer) { + Assert.notNull(serializer, "Serializer may not be null"); this.serializer = serializer; } @@ -281,7 +334,8 @@ public class TcpConnectionFactoryFactoryBean extends AbstractFactoryBean connections = new LinkedList(); + private volatile TcpSocketSupport tcpSocketSupport = new DefaultTcpSocketSupport(); + protected final Object lifecycleMonitor = new Object(); private volatile long nextCheckForClosedNioConnections; @@ -141,6 +145,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport socket.setTrafficClass(this.soTrafficClass); } socket.setKeepAlive(this.soKeepAlive); + this.tcpSocketSupport.postProcessSocket(socket); } /** @@ -678,4 +683,13 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport } } + protected TcpSocketSupport getTcpSocketSupport() { + return tcpSocketSupport; + } + + public void setTcpSocketSupport(TcpSocketSupport tcpSocketSupport) { + Assert.notNull(tcpSocketSupport, "TcpSocketSupport must not be null"); + this.tcpSocketSupport = tcpSocketSupport; + } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java index d981334c33..d551a7e8ee 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractServerConnectionFactory.java @@ -16,21 +16,22 @@ package org.springframework.integration.ip.tcp.connection; +import java.net.ServerSocket; import java.net.Socket; import java.net.SocketException; -/** +/** * Base class for all server connection factories. Server connection factories * listen on a port for incoming connections and create new TcpConnection objects * for each new connection. - * + * * @author Gary Russell * @since 2.0 */ public abstract class AbstractServerConnectionFactory extends AbstractConnectionFactory { private boolean listening; - + private String localAddress; @@ -60,18 +61,18 @@ public abstract class AbstractServerConnectionFactory extends AbstractConnection /** - * + * * @return true if the server is listening on the port. */ public boolean isListening() { return listening; } - + /** * Transfers attributes such as (de)serializer, singleUse etc to a new connection. * For single use sockets, enforces a socket timeout (default 10 seconds). * @param connection The new connection. - * @param socket The new socket. + * @param socket The new socket. */ protected void initializeConnection(TcpConnection connection, Socket socket) { TcpListener listener = this.getListener(); @@ -98,9 +99,13 @@ public abstract class AbstractServerConnectionFactory extends AbstractConnection } } - + + protected void postProcessServerSocket(ServerSocket serverSocket) { + this.getTcpSocketSupport().postProcessServerSocket(serverSocket); + } + /** - * + * * @return the localAddress */ public String getLocalAddress() { @@ -108,7 +113,7 @@ public abstract class AbstractServerConnectionFactory extends AbstractConnection } /** - * Used on multi-homed systems to enforce the server to listen + * Used on multi-homed systems to enforce the server to listen * on a specfic network address instead of all network adapters. * @param localAddress the ip address of the required adapter. */ diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java index 66ea8ec1ba..16db6a2c56 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetClientConnectionFactory.java @@ -22,8 +22,12 @@ import java.net.SocketException; import javax.net.SocketFactory; +import org.springframework.integration.ip.tcp.connection.support.DefaultTcpNetSocketFactorySupport; +import org.springframework.integration.ip.tcp.connection.support.TcpSocketFactorySupport; +import org.springframework.util.Assert; + /** - * A client connection factory that creates {@link TcpNetConnection}s. + * A client connection factory that creates {@link TcpNetConnection}s. * @author Gary Russell * @since 2.0 * @@ -31,6 +35,8 @@ import javax.net.SocketFactory; public class TcpNetClientConnectionFactory extends AbstractClientConnectionFactory { + private volatile TcpSocketFactorySupport tcpSocketFactorySupport = new DefaultTcpNetSocketFactorySupport(); + /** * Creates a TcpNetClientConnectionFactory for connections to the host and port. * @param host the host @@ -45,6 +51,7 @@ public class TcpNetClientConnectionFactory extends * @throws SocketException * @throws Exception */ + @Override protected TcpConnection getOrMakeConnection() throws Exception { TcpConnection theConnection = this.getTheConnection(); if (theConnection != null && theConnection.isOpen()) { @@ -73,13 +80,24 @@ public class TcpNetClientConnectionFactory extends * @throws IOException */ protected Socket createSocket(String host, int port) throws IOException { - return SocketFactory.getDefault().createSocket(host, port); + return this.tcpSocketFactorySupport.getSocketFactory().createSocket(host, port); } + @Override public void close() { } public void run() { } + protected TcpSocketFactorySupport getTcpSocketFactorySupport() { + return tcpSocketFactorySupport; + } + + public void setTcpSocketFactorySupport( + TcpSocketFactorySupport tcpSocketFactorySupport) { + Assert.notNull(tcpSocketFactorySupport, "TcpSocketFactorySupport may not be null"); + this.tcpSocketFactorySupport = tcpSocketFactorySupport; + } + } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java index e2a877e45c..09ffeec95a 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetConnection.java @@ -109,7 +109,11 @@ public class TcpNetConnection extends AbstractTcpConnection { logger.debug("Closing single use socket after timeout"); } else { if (this.noReadErrorOnClose) { - if (logger.isDebugEnabled()) { + if (logger.isTraceEnabled()) { + logger.trace("Read exception " + + this.getConnectionId(), e); + } + else if (logger.isDebugEnabled()) { logger.debug("Read exception " + this.getConnectionId() + " " + e.getClass().getSimpleName() + diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java index 108e02b539..b0df719676 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNetServerConnectionFactory.java @@ -21,9 +21,14 @@ import java.net.InetAddress; import java.net.ServerSocket; import java.net.Socket; import java.net.SocketException; +import java.net.SocketTimeoutException; import javax.net.ServerSocketFactory; +import org.springframework.integration.ip.tcp.connection.support.DefaultTcpNetSocketFactorySupport; +import org.springframework.integration.ip.tcp.connection.support.TcpSocketFactorySupport; +import org.springframework.util.Assert; + /** * Implements a server connection factory that produces {@link TcpNetConnection}s using * a {@link ServerSocket}. Must have a {@link TcpListener} registered. @@ -33,7 +38,9 @@ import javax.net.ServerSocketFactory; */ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFactory { - private ServerSocket serverSocket; + private volatile ServerSocket serverSocket; + + private volatile TcpSocketFactorySupport tcpSocketFactorySupport = new DefaultTcpNetSocketFactorySupport(); /** * Listens for incoming connections on the port. @@ -63,12 +70,27 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto InetAddress whichNic = InetAddress.getByName(this.getLocalAddress()); theServerSocket = createServerSocket(this.getPort(), this.getPoolSize(), whichNic); } + this.getTcpSocketSupport().postProcessServerSocket(theServerSocket); this.serverSocket = theServerSocket; this.setListening(true); logger.info("Listening on port " + this.getPort()); while (true) { - final Socket socket = serverSocket.accept(); - logger.debug("Accepted connection from " + socket.getInetAddress().getHostAddress()); + final Socket socket; + /* + * User hooks in the TcpSocketSupport may have set the server socket SO_TIMEOUT. + * Not fatal. + */ + try { + socket = serverSocket.accept(); + } catch (SocketTimeoutException ste) { + if (logger.isDebugEnabled()) { + logger.debug("Timed out on accept; continuing"); + } + continue; + } + if (logger.isDebugEnabled()) { + logger.debug("Accepted connection from " + socket.getInetAddress().getHostAddress()); + } setSocketAttributes(socket); TcpConnection connection = new TcpNetConnection(socket, true, this.isLookupHost()); connection = wrapConnection(connection); @@ -100,11 +122,12 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto * @throws IOException */ protected ServerSocket createServerSocket(int port, int backlog, InetAddress whichNic) throws IOException { + ServerSocketFactory serverSocketFactory = this.tcpSocketFactorySupport.getServerSocketFactory(); if (whichNic == null) { - return ServerSocketFactory.getDefault().createServerSocket(port, + return serverSocketFactory.createServerSocket(port, Math.abs(backlog)); } else { - return ServerSocketFactory.getDefault().createServerSocket(port, + return serverSocketFactory.createServerSocket(port, Math.abs(backlog), whichNic); } } @@ -125,5 +148,15 @@ public class TcpNetServerConnectionFactory extends AbstractServerConnectionFacto protected ServerSocket getServerSocket() { return serverSocket; } + + protected TcpSocketFactorySupport getTcpSocketFactorySupport() { + return tcpSocketFactorySupport; + } + + public void setTcpSocketFactorySupport( + TcpSocketFactorySupport tcpSocketFactorySupport) { + Assert.notNull(tcpSocketFactorySupport, "TcpSocketFactorySupport may not be null"); + this.tcpSocketFactorySupport = tcpSocketFactorySupport; + } } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java index ae0203e038..d196b73eac 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioClientConnectionFactory.java @@ -30,6 +30,10 @@ import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.LinkedBlockingQueue; +import org.springframework.integration.ip.tcp.connection.support.DefaultTcpNioConnectionSupport; +import org.springframework.integration.ip.tcp.connection.support.TcpNioConnectionSupport; +import org.springframework.util.Assert; + /** * A client connection factory that creates {@link TcpNioConnection}s. @@ -40,14 +44,15 @@ import java.util.concurrent.LinkedBlockingQueue; public class TcpNioClientConnectionFactory extends AbstractClientConnectionFactory { - private boolean usingDirectBuffers; + private volatile boolean usingDirectBuffers; - private Selector selector; + private volatile Selector selector; - private Map channelMap = new ConcurrentHashMap(); + private final Map channelMap = new ConcurrentHashMap(); - private BlockingQueue newChannels = new LinkedBlockingQueue(); + private final BlockingQueue newChannels = new LinkedBlockingQueue(); + private volatile TcpNioConnectionSupport tcpNioConnectionSupport = new DefaultTcpNioConnectionSupport(); /** * Creates a TcpNioClientConnectionFactory for connections to the host and port. @@ -84,7 +89,8 @@ public class TcpNioClientConnectionFactory extends } SocketChannel socketChannel = SocketChannel.open(new InetSocketAddress(this.getHost(), this.getPort())); setSocketAttributes(socketChannel.socket()); - TcpNioConnection connection = new TcpNioConnection(socketChannel, false, this.isLookupHost()); + TcpNioConnection connection = this.tcpNioConnectionSupport.createNewConnection( + socketChannel, false, this.isLookupHost()); connection.setUsingDirectBuffers(this.usingDirectBuffers); connection.setTaskExecutor(this.getTaskExecutor()); TcpConnection wrappedConnection = wrapConnection(connection); @@ -109,6 +115,11 @@ public class TcpNioClientConnectionFactory extends this.usingDirectBuffers = usingDirectBuffers; } + public void setTcpNioConnectionSupport(TcpNioConnectionSupport tcpNioSupport) { + Assert.notNull(tcpNioSupport, "TcpNioSupport must not be null"); + this.tcpNioConnectionSupport = tcpNioSupport; + } + public void close() { if (this.selector != null) { this.selector.wakeup(); diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java index fe1b94ac6e..afb76ab58f 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java @@ -33,10 +33,11 @@ import java.util.concurrent.atomic.AtomicInteger; import org.springframework.core.serializer.Serializer; import org.springframework.integration.Message; import org.springframework.integration.ip.tcp.serializer.SoftEndOfStreamException; +import org.springframework.util.Assert; /** * A TcpConnection that uses and underlying {@link SocketChannel}. - * + * * @author Gary Russell * @since 2.0 * @@ -45,7 +46,7 @@ public class TcpNioConnection extends AbstractTcpConnection { private final SocketChannel socketChannel; - private volatile OutputStream channelOutputStream; + private final ChannelOutputStream channelOutputStream; private volatile PipedOutputStream pipedOutputStream; @@ -79,6 +80,7 @@ public class TcpNioConnection extends AbstractTcpConnection { this.channelOutputStream = new ChannelOutputStream(); } + @Override public void close() { doClose(); } @@ -103,7 +105,7 @@ public class TcpNioConnection extends AbstractTcpConnection { public void send(Message message) throws Exception { synchronized(this.getMapper()) { Object object = this.getMapper().fromMessage(message); - ((Serializer) this.getSerializer()).serialize(object, this.channelOutputStream); + ((Serializer) this.getSerializer()).serialize(object, this.getChannelOutputStream()); this.afterSend(message); } } @@ -131,9 +133,9 @@ public class TcpNioConnection extends AbstractTcpConnection { } /** - * If there is no listener, and this connection is not for single use, + * If there is no listener, and this connection is not for single use, * this method exits. When there is a listener, this method assembles - * data into messages by invoking convertAndSend whenever there is + * data into messages by invoking convertAndSend whenever there is * data in the input Stream. Method exits when a message is complete * and there is no more data; thus freeing the thread to work on other * sockets. @@ -170,7 +172,7 @@ public class TcpNioConnection extends AbstractTcpConnection { } else { logger.error("Read exception " + this.getConnectionId() + " " + - e.getClass().getSimpleName() + + e.getClass().getSimpleName() + ":" + e.getCause() + ":" + e.getMessage()); } this.closeConnection(); @@ -244,7 +246,7 @@ public class TcpNioConnection extends AbstractTcpConnection { } /* * For single use sockets, we close after receipt if we are on the client - * side, and the data was not intercepted, + * side, and the data was not intercepted, * or the server side has no outbound adapter registered */ if (this.isSingleUse() && ((!this.isServer() && !intercepted) || (this.isServer() && this.getSender() == null))) { @@ -267,23 +269,37 @@ public class TcpNioConnection extends AbstractTcpConnection { } // If there is no assembler running, start one checkForAssembler(); - this.rawBuffer.clear(); + if (logger.isTraceEnabled()) { + logger.trace("Before read:" + this.rawBuffer.position() + "/" + this.rawBuffer.limit()); + } int len = this.socketChannel.read(this.rawBuffer); if (len < 0) { this.writingToPipe = false; this.closeConnection(); } + if (logger.isTraceEnabled()) { + logger.trace("After read:" + this.rawBuffer.position() + "/" + this.rawBuffer.limit()); + } this.rawBuffer.flip(); + if (logger.isTraceEnabled()) { + logger.trace("After flip:" + this.rawBuffer.position() + "/" + this.rawBuffer.limit()); + } if (logger.isDebugEnabled()) { logger.debug("Read " + rawBuffer.limit() + " into raw buffer"); } - this.pipedOutputStream.write(this.rawBuffer.array(), 0, this.rawBuffer.limit()); - this.pipedOutputStream.flush(); + this.sendToPipe(this.rawBuffer); } finally { this.writingToPipe = false; } } + protected void sendToPipe(ByteBuffer rawBuffer) throws IOException { + Assert.notNull(rawBuffer, "rawBuffer cannot be null"); + this.pipedOutputStream.write(rawBuffer.array(), 0, rawBuffer.limit()); + this.pipedOutputStream.flush(); + rawBuffer.clear(); + } + private void checkForAssembler() { synchronized(this.executionControl) { if (this.executionControl.incrementAndGet() <= 1) { @@ -311,9 +327,9 @@ public class TcpNioConnection extends AbstractTcpConnection { } this.closeConnection(); } catch (Exception e) { - logger.error("Exception on Read " + - this.getConnectionId() + " " + - e.getMessage()); + logger.error("Exception on Read " + + this.getConnectionId() + " " + + e.getMessage(), e); this.closeConnection(); } } @@ -326,7 +342,7 @@ public class TcpNioConnection extends AbstractTcpConnection { } /** - * + * * @param taskExecutor the taskExecutor to set */ public void setTaskExecutor(Executor taskExecutor) { @@ -342,8 +358,16 @@ public class TcpNioConnection extends AbstractTcpConnection { this.usingDirectBuffers = usingDirectBuffers; } + protected boolean isUsingDirectBuffers() { + return usingDirectBuffers; + } + + protected ChannelOutputStream getChannelOutputStream() { + return channelOutputStream; + } + /** - * + * * @return Time of last read. */ public long getLastRead() { @@ -351,7 +375,7 @@ public class TcpNioConnection extends AbstractTcpConnection { } /** - * + * * @param lastRead The time of the last read. */ public void setLastRead(long lastRead) { @@ -359,7 +383,7 @@ public class TcpNioConnection extends AbstractTcpConnection { } /** - * OutputStream to wrap a SocketChannel; implements timeout on write. + * OutputStream to wrap a SocketChannel; implements timeout on write. * */ class ChannelOutputStream extends OutputStream { @@ -397,7 +421,10 @@ public class TcpNioConnection extends AbstractTcpConnection { doWrite(buffer); } - private synchronized void doWrite(ByteBuffer buffer) throws IOException { + protected synchronized void doWrite(ByteBuffer buffer) throws IOException { + if (logger.isDebugEnabled()) { + logger.debug(getConnectionId() + " writing " + buffer.remaining()); + } socketChannel.write(buffer); int remaining = buffer.remaining(); if (remaining == 0) { diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioSSLConnection.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioSSLConnection.java new file mode 100644 index 0000000000..895246525b --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioSSLConnection.java @@ -0,0 +1,396 @@ +/* + * Copyright 2002-2012 the original author or authors. + * + * Licensed 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.springframework.integration.ip.tcp.connection; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.channels.SocketChannel; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; + +import javax.net.ssl.SSLEngine; +import javax.net.ssl.SSLEngineResult; +import javax.net.ssl.SSLEngineResult.HandshakeStatus; +import javax.net.ssl.SSLEngineResult.Status; +import javax.net.ssl.SSLException; + +import org.springframework.integration.MessagingException; +import org.springframework.util.Assert; + +/** + * Implementation of {@link TcpConnection} supporting SSL/TLS over NIO. + * Unlike TcpNetConnection, which uses Sockets, the JVM does not directly support SSL for + * SocketChannels, used by NIO. Instead, the SSLEngine is provided whereby the SSL + * encryption is performed by passing in a plain text buffer, and receiving an + * encrypted buffer to transmit over the network. Similarly, encrypted data read from + * the network is decrypted.

+ * However, before this can be done, certain handshaking operations are required, involving + * the creation of data buffers which must be exchanged by the peers. A number of such + * transfers are required; once the handshake is finished, it is relatively simple to + * encrypt/decrypt the data.

+ * Also, it may be deemed necessary to re-perform handshaking.

+ * This class supports the management of handshaking as necessary, both from the + * initiating and receiving peers. + * @author Gary Russell + * @since 2.2 + * + */ +public class TcpNioSSLConnection extends TcpNioConnection { + + private final SSLEngine sslEngine; + + private volatile ByteBuffer decoded; + + private volatile ByteBuffer encoded; + + private volatile SSLChannelOutputStream sslChannelOutputStream; + + private final Semaphore semaphore = new Semaphore(0); + + private final Object monitorLock = new Object(); + + private volatile boolean writerActive; + + private boolean needMoreNetworkData; + + public TcpNioSSLConnection(SocketChannel socketChannel, boolean server, + boolean lookupHost, SSLEngine sslEngine) throws Exception { + super(socketChannel, server, lookupHost); + this.sslEngine = sslEngine; + } + + /** + * Overrides super class method to perform decryption and/or participate + * in handshaking. Decrypted data is sent to the super class to be + * assembled into a Message. Data received from the network may + * constitute multiple SSL packets, and may end with a partial + * packet. In that case, the buffer is compacted, ready to receive + * the remainder of the packet. + */ + @Override + protected void sendToPipe(final ByteBuffer networkBuffer) throws IOException { + Assert.notNull(networkBuffer, "rawBuffer cannot be null"); + if (logger.isDebugEnabled()) { + logger.debug("sendToPipe " + sslEngine.getHandshakeStatus() + ", remaining:" + networkBuffer.remaining()); + } + SSLEngineResult result = null; + while (!this.needMoreNetworkData) { + result = decode(networkBuffer); + if (logger.isDebugEnabled()) { + logger.debug("result " + resultToString(result) + ", remaining:" + networkBuffer.remaining()); + } + } + this.needMoreNetworkData = false; + if (result.getStatus() == Status.BUFFER_UNDERFLOW) { + networkBuffer.compact(); + } + else { + networkBuffer.clear(); + } + if (logger.isDebugEnabled()) { + logger.debug("sendToPipe.x " + resultToString(result) + ", remaining:" + networkBuffer.remaining()); + } + } + + /** + * Performs the actual decryption of a received packet - which may be real + * data, or handshaking data. Appropriate action is taken with the data. + * If this side did not initiate the handshake, any handshaking data sent out + * is handled by the thread running in the {@link SSLChannelOutputStream#doWrite(ByteBuffer)} + * method, which is awoken here, as a result of reaching that stage in the handshaking. + */ + private SSLEngineResult decode(ByteBuffer networkBuffer) throws IOException { + SSLEngineResult result = new SSLEngineResult(Status.OK, this.sslEngine.getHandshakeStatus(), 0, 0); + HandshakeStatus handshakeStatus = this.sslEngine.getHandshakeStatus(); + switch (handshakeStatus) { + case NEED_TASK: + runTasks(); + break; + case NEED_UNWRAP: + case FINISHED: + case NOT_HANDSHAKING: + this.decoded.clear(); + result = this.sslEngine.unwrap(networkBuffer, this.decoded); + if (logger.isDebugEnabled()) { + logger.debug("After unwrap:" + resultToString(result)); + } + Status status = result.getStatus(); + if (status == Status.BUFFER_OVERFLOW) { + this.decoded = this.allocateEncryptionBuffer(this.sslEngine.getSession().getApplicationBufferSize()); + } + if (result.bytesProduced() > 0) { + this.decoded.flip(); + super.sendToPipe(this.decoded); + } + break; + case NEED_WRAP: + if (!resumeWriterIfNeeded()) { + this.encoded.clear(); + result = this.sslEngine.wrap(networkBuffer, this.encoded); + if (logger.isDebugEnabled()) { + logger.debug("After wrap:" + resultToString(result)); + } + if (result.getStatus() == Status.BUFFER_OVERFLOW) { + this.encoded = this.allocateEncryptionBuffer(this.sslEngine.getSession().getPacketBufferSize()); + } + else { + this.encoded.flip(); + getSSLChannelOutputStream().writeEncoded(this.encoded); + } + } + break; + default: + } + switch (result.getHandshakeStatus()) { + case FINISHED: + resumeWriterIfNeeded(); + // switch fall-through intended + case NOT_HANDSHAKING: + case NEED_UNWRAP: + this.needMoreNetworkData = result.getStatus() == Status.BUFFER_UNDERFLOW || networkBuffer.remaining() == 0; + break; + default: + } + return result; + } + + /** + * Handshake sends are handled by the initiator. + * @return false if we are the initiator. + */ + private boolean resumeWriterIfNeeded() { + if (this.writerActive) { + if (logger.isTraceEnabled()) { + logger.trace("Waking sender, permits:" + this.semaphore.availablePermits()); + } + this.semaphore.release(); + return true; + } + return false; + } + + /** + * Part of the SSLEngine handshaking protocol required at + * various stages. Tasks are run on the current thread. + */ + private void runTasks() { + Runnable task; + while ((task = this.sslEngine.getDelegatedTask()) != null) { + task.run(); + } + } + + /** + * Determines whether {@link #runTasks()} is needed and invokes if so. + */ + private HandshakeStatus runTasksIfNeeded(SSLEngineResult result) throws IOException { + if (result != null) { + if (logger.isDebugEnabled()) { + logger.debug("Running tasks if needed " + resultToString(result)); + } + if (result.getHandshakeStatus() == HandshakeStatus.NEED_TASK) { + runTasks(); + } + } + HandshakeStatus handshakeStatus = sslEngine.getHandshakeStatus(); + if (logger.isDebugEnabled()) { + logger.debug("New handshake status " + handshakeStatus); + } + return handshakeStatus; + } + + /** + * Initializes the SSLEngine and sets up the encryption/decryption buffers. + */ + public void init() throws IOException { + if (this.decoded == null) { + this.decoded = allocateEncryptionBuffer(2048); + this.encoded = allocateEncryptionBuffer(2048); + this.initilizeEngine(); + } + } + + private ByteBuffer allocateEncryptionBuffer(int size) { + if (this.isUsingDirectBuffers()) { + return ByteBuffer.allocateDirect(size); + } + else { + return ByteBuffer.allocate(size); + } + } + + private void initilizeEngine() throws IOException { + boolean client = !this.isServer(); + this.sslEngine.setUseClientMode(client); + } + + @Override + protected ChannelOutputStream getChannelOutputStream() { + synchronized (this.monitorLock) { + if (this.sslChannelOutputStream == null) { + this.sslChannelOutputStream = new SSLChannelOutputStream(super.getChannelOutputStream()); + } + return this.sslChannelOutputStream; + } + } + + protected SSLChannelOutputStream getSSLChannelOutputStream() { + if (this.sslChannelOutputStream == null) { + return (SSLChannelOutputStream) this.getChannelOutputStream(); + } + else { + return this.sslChannelOutputStream; + } + } + + private String resultToString(SSLEngineResult result) { + return result.toString().replace('\n', ' '); + } + + /** + * Subclass of {@link TcpNioConnection.ChannelOutputStream} to handle encryption + * of outbound data. Wraps an instance of the superclass, which is invoked to + * send to encrypted data to the SocketChannel. + * + */ + class SSLChannelOutputStream extends ChannelOutputStream { + + private final ChannelOutputStream channelOutputStream; + + public SSLChannelOutputStream(ChannelOutputStream channelOutputStream) { + this.channelOutputStream = channelOutputStream; + } + + /** + * Encrypts the plaintText buffer and writes it to the SocketChannel. + * Will participate in SSL handshaking as necessary. For very large + * data, the SSL packets will be limited by the engine's buffer sizes + * and multiple writes will be necessary. + */ + @Override + protected synchronized void doWrite(ByteBuffer plainText) + throws IOException { + try { + TcpNioSSLConnection.this.writerActive = true; + int remaining = plainText.remaining(); + while (remaining > 0) { + SSLEngineResult result = encode(plainText); + if (logger.isDebugEnabled()) { + logger.debug("doWrite: " + resultToString(result)); + } + if (result.getHandshakeStatus() == SSLEngineResult.HandshakeStatus.NOT_HANDSHAKING) { + writeEncodedIfAny(); + if (plainText.remaining() >= remaining) { + throw new MessagingException( + "Unexpected condition - SSL wrap did not consume any data; remaining = " + + remaining); + } + remaining = plainText.remaining(); + } else { + doClientSideHandshake(plainText, result); + writeEncodedIfAny(); + } + } + } + finally { + TcpNioSSLConnection.this.writerActive = false; + } + } + + /** + * Handles SSL handshaking; when network data is needed from the peer, suspends + * until that data is received. + */ + private void doClientSideHandshake(ByteBuffer plainText, + SSLEngineResult result) throws IOException, SSLException { + TcpNioSSLConnection.this.semaphore.drainPermits(); + HandshakeStatus status = TcpNioSSLConnection.this.sslEngine.getHandshakeStatus(); + while (status != HandshakeStatus.FINISHED) { + writeEncodedIfAny(); + status = runTasksIfNeeded(result); + if (status == HandshakeStatus.NEED_UNWRAP) { + status = waitForHandshakeData(result, status); + } + if (status == HandshakeStatus.NEED_WRAP || + status == HandshakeStatus.NOT_HANDSHAKING || + status == HandshakeStatus.FINISHED) { + result = encode(plainText); + status = result.getHandshakeStatus(); + if (status == HandshakeStatus.NOT_HANDSHAKING || + status == HandshakeStatus.FINISHED) { + break; + } + } + else { + logger.debug(status); + } + } + } + + private void writeEncodedIfAny() throws IOException { + TcpNioSSLConnection.this.encoded.flip(); + writeEncoded(TcpNioSSLConnection.this.encoded); + TcpNioSSLConnection.this.encoded.clear(); + } + + /** + * Suspend processing until data is received from the peer. + */ + private HandshakeStatus waitForHandshakeData(SSLEngineResult result, + HandshakeStatus status) throws IOException { + try { + if (logger.isTraceEnabled()) { + logger.trace("Writer waiting for handshake"); + } + if (!semaphore.tryAcquire(30, TimeUnit.SECONDS)) { + throw new MessagingException("SSL Handshaking taking too long"); + } + if (logger.isTraceEnabled()) { + logger.trace("Writer resuming handshake"); + } + status = runTasksIfNeeded(result); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new MessagingException("Interrupted during SSL Handshaking"); + } + return status; + } + + /** + * Encrypts plain text data. The result may indicate handshaking is needed. + */ + private SSLEngineResult encode(ByteBuffer plainText) + throws SSLException, IOException { + TcpNioSSLConnection.this.encoded.clear(); + SSLEngineResult result = TcpNioSSLConnection.this.sslEngine.wrap(plainText, TcpNioSSLConnection.this.encoded); + if (logger.isDebugEnabled()) { + logger.debug("After wrap:" + resultToString(result) + " Plaintext buffer @" + plainText.position() + "/" + plainText.limit()); + } + if (result.getStatus() == SSLEngineResult.Status.BUFFER_OVERFLOW) { + TcpNioSSLConnection.this.encoded = allocateEncryptionBuffer(sslEngine.getSession().getPacketBufferSize()); + result = TcpNioSSLConnection.this.sslEngine.wrap(plainText, TcpNioSSLConnection.this.encoded); + } + return result; + } + + /** + * Write data to the SocketChannel. + */ + void writeEncoded(ByteBuffer encoded) throws IOException { + this.channelOutputStream.doWrite(encoded); + } + } +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java index c5cd0ebfe8..74842dc0f7 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioServerConnectionFactory.java @@ -30,6 +30,10 @@ import java.nio.channels.SocketChannel; import java.util.HashMap; import java.util.Map; +import org.springframework.integration.ip.tcp.connection.support.DefaultTcpNioConnectionSupport; +import org.springframework.integration.ip.tcp.connection.support.TcpNioConnectionSupport; +import org.springframework.util.Assert; + /** /** * Implements a server connection factory that produces {@link TcpNioConnection}s using @@ -40,14 +44,16 @@ import java.util.Map; */ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFactory { - private ServerSocketChannel serverChannel; - - private boolean usingDirectBuffers; - - private Map channelMap = new HashMap(); + private volatile ServerSocketChannel serverChannel; + + private volatile boolean usingDirectBuffers; + + private final Map channelMap = new HashMap(); + + private volatile Selector selector; + + private volatile TcpNioConnectionSupport tcpNioConnectionSupport = new DefaultTcpNioConnectionSupport(); - private Selector selector; - /** * Listens for incoming connections on the port. * @param port The port. @@ -71,7 +77,10 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto try { this.serverChannel = ServerSocketChannel.open(); int port = this.getPort(); - logger.info("Listening on port " + port); + this.getTcpSocketSupport().postProcessServerSocket(this.serverChannel.socket()); + if (logger.isInfoEnabled()) { + logger.info("Listening on port " + port); + } this.serverChannel.configureBlocking(false); if (this.getLocalAddress() == null) { this.serverChannel.socket().bind(new InetSocketAddress(port), @@ -154,7 +163,9 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto private TcpNioConnection createTcpNioConnection(SocketChannel socketChannel) { try { - TcpNioConnection connection = new TcpNioConnection(socketChannel, true, this.isLookupHost()); + TcpNioConnection connection = this.tcpNioConnectionSupport + .createNewConnection(socketChannel, true, + this.isLookupHost()); connection.setUsingDirectBuffers(this.usingDirectBuffers); TcpConnection wrappedConnection = wrapConnection(connection); this.initializeConnection(wrappedConnection, socketChannel.socket()); @@ -182,6 +193,11 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto this.usingDirectBuffers = usingDirectBuffers; } + public void setTcpNioConnectionSupport(TcpNioConnectionSupport tcpNioSupport) { + Assert.notNull(tcpNioSupport, "TcpNioSupport must not be null"); + this.tcpNioConnectionSupport = tcpNioSupport; + } + /** * @return the serverChannel */ @@ -202,6 +218,5 @@ public class TcpNioServerConnectionFactory extends AbstractServerConnectionFacto protected Map getConnections() { return channelMap; } - } diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpNetSSLSocketFactorySupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpNetSSLSocketFactorySupport.java new file mode 100644 index 0000000000..7b7b8e678f --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpNetSSLSocketFactorySupport.java @@ -0,0 +1,58 @@ +/* + * Copyright 2002-2012 the original author or authors. + * + * Licensed 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.springframework.integration.ip.tcp.connection.support; + +import javax.net.ServerSocketFactory; +import javax.net.SocketFactory; +import javax.net.ssl.SSLContext; + +import org.springframework.beans.factory.InitializingBean; +import org.springframework.util.Assert; + +/** + * Implementation of TcpSocketFactorySupport + * for SSL sockets {@link javax.net.ssl.SSLServerSocket} and + * {@link javax.net.ssl.SSLSocket}. + * @author Gary Russell + * @since 2.2 + * + */ +public class DefaultTcpNetSSLSocketFactorySupport implements TcpSocketFactorySupport, + InitializingBean { + + private final TcpSSLContextSupport sslContextSupport; + + private volatile SSLContext sslContext; + + public DefaultTcpNetSSLSocketFactorySupport(TcpSSLContextSupport sslContextSupport) { + Assert.notNull(sslContextSupport, "TcpSSLContextSupport must not be null"); + this.sslContextSupport = sslContextSupport; + } + + public ServerSocketFactory getServerSocketFactory() { + return this.sslContext.getServerSocketFactory(); + } + + public SocketFactory getSocketFactory() { + return this.sslContext.getSocketFactory(); + } + + public void afterPropertiesSet() throws Exception { + this.sslContext = this.sslContextSupport.getSSLContext(); + Assert.notNull(this.sslContext, "SSLContex must not be null"); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpNetSocketFactorySupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpNetSocketFactorySupport.java new file mode 100644 index 0000000000..0131fb4da9 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpNetSocketFactorySupport.java @@ -0,0 +1,39 @@ +/* + * Copyright 2002-2012 the original author or authors. + * + * Licensed 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.springframework.integration.ip.tcp.connection.support; + +import javax.net.ServerSocketFactory; +import javax.net.SocketFactory; + +/** + * Implementation of TcpSocketFactorySupport + * for non-SSL sockets {@link java.net.ServerSocket} and + * {@link java.net.Socket}. + * @author Gary Russell + * @since 2.2 + * + */ +public class DefaultTcpNetSocketFactorySupport implements TcpSocketFactorySupport { + + public ServerSocketFactory getServerSocketFactory() { + return ServerSocketFactory.getDefault(); + } + + public SocketFactory getSocketFactory() { + return SocketFactory.getDefault(); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpNioConnectionSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpNioConnectionSupport.java new file mode 100644 index 0000000000..ff8b2313eb --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpNioConnectionSupport.java @@ -0,0 +1,36 @@ +/* + * Copyright 2002-2012 the original author or authors. + * + * Licensed 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.springframework.integration.ip.tcp.connection.support; + +import java.nio.channels.SocketChannel; + +import org.springframework.integration.ip.tcp.connection.TcpNioConnection; + +/** + * Implementation of {@link TcpNioConnectionSupport} for non-SSL + * NIO connections. + * @author Gary Russell + * @since 2.2 + * + */ +public class DefaultTcpNioConnectionSupport implements TcpNioConnectionSupport { + + public TcpNioConnection createNewConnection(SocketChannel socketChannel, + boolean server, boolean lookupHost) throws Exception { + return new TcpNioConnection(socketChannel, server, lookupHost); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpNioSSLConnectionSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpNioSSLConnectionSupport.java new file mode 100644 index 0000000000..60429efef9 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpNioSSLConnectionSupport.java @@ -0,0 +1,60 @@ +/* + * Copyright 2002-2012 the original author or authors. + * + * Licensed 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.springframework.integration.ip.tcp.connection.support; + +import java.nio.channels.SocketChannel; + +import javax.net.ssl.SSLContext; +import javax.net.ssl.SSLEngine; + +import org.springframework.beans.factory.InitializingBean; +import org.springframework.integration.ip.tcp.connection.TcpNioConnection; +import org.springframework.integration.ip.tcp.connection.TcpNioSSLConnection; +import org.springframework.util.Assert; + +/** + * Implementation of {@link TcpNioConnectionSupport} for SSL + * NIO connections. + * @author Gary Russell + * @since 2.2 + * + */ +public class DefaultTcpNioSSLConnectionSupport implements TcpNioConnectionSupport, InitializingBean { + + private volatile SSLContext sslContext; + + private final TcpSSLContextSupport sslContextSupport; + + public DefaultTcpNioSSLConnectionSupport(TcpSSLContextSupport sslContextSupport) { + Assert.notNull(sslContextSupport, "TcpSSLContextSupport must not be null"); + this.sslContextSupport = sslContextSupport; + } + + public TcpNioConnection createNewConnection(SocketChannel socketChannel, + boolean server, boolean lookupHost) throws Exception { + + SSLEngine sslEngine = this.sslContext.createSSLEngine(); + TcpNioSSLConnection tcpNioSSLConnection = new TcpNioSSLConnection(socketChannel, server, lookupHost, sslEngine); + tcpNioSSLConnection.init(); + return tcpNioSSLConnection; + } + + public void afterPropertiesSet() throws Exception { + this.sslContext = this.sslContextSupport.getSSLContext(); + Assert.notNull(this.sslContext, "SSLContex must not be null"); + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpSSLContextSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpSSLContextSupport.java new file mode 100644 index 0000000000..8451eeb655 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpSSLContextSupport.java @@ -0,0 +1,97 @@ +/* + * Copyright 2002-2012 the original author or authors. + * + * Licensed 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.springframework.integration.ip.tcp.connection.support; + +import java.io.FileInputStream; +import java.io.IOException; +import java.security.GeneralSecurityException; +import java.security.KeyStore; + +import javax.net.ssl.KeyManagerFactory; +import javax.net.ssl.SSLContext; +import javax.net.ssl.TrustManagerFactory; + +import org.springframework.core.io.ClassPathResource; +import org.springframework.core.io.Resource; +import org.springframework.util.Assert; + +/** + * Default implementation of {@link TcpSSLContextSupport}; uses a + * 'TLS' (by default) {@link SSLContext}, initialized with 'JKS' + * keystores, managed by 'SunX509' Key and Trust managers. + * @author Gary Russell + * @since 2.1 + * + */ +public class DefaultTcpSSLContextSupport implements TcpSSLContextSupport { + + private final String keyStore; + + private final String trustStore; + + private final char[] keyStorePassword; + + private final char[] trustStorePassword; + + private volatile String protocol = "TLS"; + + /** + * Prepares for the creation of an SSLContext using the supplied + * key/trust stores and passwords. + * @param keyStore A {@link Resource} pointing to the keyStore. + * @param trustStore A {@link Resource} pointing to the trustStore. + * @param keyStorePassword The passowrd for the keyStore. + * @param trustStorePassword The password for the trustStore. + */ + public DefaultTcpSSLContextSupport(String keyStore, String trustStore, + String keyStorePassword, String trustStorePassword) { + this.keyStore = keyStore; + this.trustStore = trustStore; + this.keyStorePassword = keyStorePassword.toCharArray(); + this.trustStorePassword = trustStorePassword.toCharArray(); + } + + public SSLContext getSSLContext() throws GeneralSecurityException, IOException { + KeyStore ks = KeyStore.getInstance("JKS"); + KeyStore ts = KeyStore.getInstance("JKS"); + + ks.load(new FileInputStream(new ClassPathResource(keyStore).getFile()), keyStorePassword); + ts.load(new FileInputStream(new ClassPathResource(trustStore).getFile()), trustStorePassword); + + KeyManagerFactory kmf = KeyManagerFactory.getInstance("SunX509"); + kmf.init(ks, keyStorePassword); + + TrustManagerFactory tmf = TrustManagerFactory.getInstance("SunX509"); + tmf.init(ts); + + SSLContext sslContext = SSLContext.getInstance(protocol); + + sslContext.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null); + + return sslContext; + + } + + /** + * The protocol used in {@link SSLContext#getInstance(String)}; default "TLS". + * @param protocol The protocol. + */ + public void setProtocol(String protocol) { + Assert.notNull(protocol, "protocol must not be null"); + this.protocol = protocol; + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpSocketSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpSocketSupport.java new file mode 100644 index 0000000000..4a031c1619 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/DefaultTcpSocketSupport.java @@ -0,0 +1,42 @@ +/* + * Copyright 2002-2012 the original author or authors. + * + * Licensed 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.springframework.integration.ip.tcp.connection.support; + +import java.net.ServerSocket; +import java.net.Socket; + +/** + * Default implementation of {@link TcpSocketSupport}; makes no + * changes to sockets. + * @author Gary Russell + * @since 2.2 + * + */ +public class DefaultTcpSocketSupport implements TcpSocketSupport { + + /** + * No-Op. + */ + public void postProcessServerSocket(ServerSocket serverSocket) { + } + + /** + * No-Op. + */ + public void postProcessSocket(Socket socket) { + } + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/TcpNioConnectionSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/TcpNioConnectionSupport.java new file mode 100644 index 0000000000..4561f17c9b --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/TcpNioConnectionSupport.java @@ -0,0 +1,34 @@ +/* + * Copyright 2002-2012 the original author or authors. + * + * Licensed 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.springframework.integration.ip.tcp.connection.support; + +import java.nio.channels.SocketChannel; + +import org.springframework.integration.ip.tcp.connection.TcpNioConnection; + +/** + * Used by NIO connection factories to instantiate a {@link TcpNioConnection} object. + * Implementations for SSL and non-SSL {@link TcpNioConnection}s are provided. + * @author Gary Russell + * @since 2.2 + * + */ +public interface TcpNioConnectionSupport { + + TcpNioConnection createNewConnection(SocketChannel socketChannel, + boolean server, boolean lookupHost) throws Exception; + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/TcpSSLContextSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/TcpSSLContextSupport.java new file mode 100644 index 0000000000..793c371dab --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/TcpSSLContextSupport.java @@ -0,0 +1,39 @@ +/* + * Copyright 2002-2012 the original author or authors. + * + * Licensed 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.springframework.integration.ip.tcp.connection.support; + +import java.io.IOException; +import java.security.GeneralSecurityException; + +import javax.net.ssl.SSLContext; + +/** + * Strategy interface for the creation of an {@link SSLContext} object + * for use with SSL/TLS sockets. + * @author Gary Russell + * @since 2.2 + * + */ +public interface TcpSSLContextSupport { + + /** + * Gets an SSLContext. + * @return the SSLContext. + * @throws Exception + */ + SSLContext getSSLContext() throws GeneralSecurityException, IOException; + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/TcpSocketFactorySupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/TcpSocketFactorySupport.java new file mode 100644 index 0000000000..619da0ae22 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/TcpSocketFactorySupport.java @@ -0,0 +1,46 @@ +/* + * Copyright 2002-2012 the original author or authors. + * + * Licensed 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.springframework.integration.ip.tcp.connection.support; + +import java.net.ServerSocket; +import java.net.Socket; + +import javax.net.ServerSocketFactory; +import javax.net.SocketFactory; + +/** + * Strategy interface for supplying Socket Factories. + * @author Gary Russell + * @since 2.2 + * + */ +public interface TcpSocketFactorySupport { + + /** + * Supplies the {@link ServerSocketFactory} to be used to + * create new {@link ServerSocket}s. + * @return the ServerSocketFacory + */ + ServerSocketFactory getServerSocketFactory(); + + /** + * Supplies the {@link SocketFactory} to be used to + * create new {@link Socket}s. + * @return the SocketFactory + */ + SocketFactory getSocketFactory(); + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/TcpSocketSupport.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/TcpSocketSupport.java new file mode 100644 index 0000000000..26609c0940 --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/TcpSocketSupport.java @@ -0,0 +1,47 @@ +/* + * Copyright 2002-2012 the original author or authors. + * + * Licensed 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.springframework.integration.ip.tcp.connection.support; + +import java.net.ServerSocket; +import java.net.Socket; + +/** + * Strategy interface for modifying sockets. + * @author Gary Russell + * @since 2.2 + * + */ +public interface TcpSocketSupport { + + /** + * Performs any further modifications to the server socket + * after the connection factory has created the socket and + * set any configured attributes, before invoking + * {@link ServerSocket#accept()}. + * @param serverSocket The ServerSocket + */ + void postProcessServerSocket(ServerSocket serverSocket); + + /** + * Performs any further modifications to the {@link Socket} after + * the socket has been created by a client, or accepted by + * a server, and after any configured atributes have been + * set. + * @param socket The Socket + */ + void postProcessSocket(Socket socket); + +} diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/package-info.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/package-info.java new file mode 100644 index 0000000000..82644b750a --- /dev/null +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/support/package-info.java @@ -0,0 +1,5 @@ +/** + * Provides classes supporting the creation/manipulation of sockets, + * SSLContexts etc. + */ +package org.springframework.integration.ip.tcp.connection.support; \ No newline at end of file diff --git a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayCrLfSerializer.java b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayCrLfSerializer.java index 596e2ef07d..9d99a53477 100644 --- a/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayCrLfSerializer.java +++ b/spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/serializer/ByteArrayCrLfSerializer.java @@ -22,19 +22,21 @@ import java.io.OutputStream; /** * Reads data in an InputStream to a byte[]; data must be terminated by \r\n - * (not included in resulting byte[]). + * (not included in resulting byte[]). * Writes a byte[] to an OutputStream and adds \r\n. - * + * * @author Gary Russell * @since 2.0 */ public class ByteArrayCrLfSerializer extends AbstractByteArraySerializer { + private static final byte[] CRLF = "\r\n".getBytes(); + /** * Reads the data in the inputstream to a byte[]. Data must be terminated * by CRLF (\r\n). Throws a {@link SoftEndOfStreamException} if the stream * is closed immediately after the \r\n (i.e. no data is in the process of - * being read). + * being read). */ public byte[] deserialize(InputStream inputStream) throws IOException { byte[] buffer = new byte[this.maxMessageSize]; @@ -69,8 +71,7 @@ public class ByteArrayCrLfSerializer extends AbstractByteArraySerializer { */ public void serialize(byte[] bytes, OutputStream outputStream) throws IOException { outputStream.write(bytes); - outputStream.write('\r'); - outputStream.write('\n'); + outputStream.write(CRLF); outputStream.flush(); } diff --git a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.2.xsd b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.2.xsd index 0fb50eb009..2b8ac02be4 100644 --- a/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.2.xsd +++ b/spring-integration-ip/src/main/resources/org/springframework/integration/ip/config/spring-integration-ip-2.2.xsd @@ -491,6 +491,52 @@ connections created by this factory. Facilitates resequencing if necessary. Defa + + + +A reference to a TcpSSLContextSupport strategy implementation. Providing this reference +enables SSL on connections created by this factory. A DefaultTcpSSLContextSupport implementation +is provided that takes keystore and trustore names and passwords. The SSLContext created by +this implementation is used to obtain socket factories (when using-nio="false") or SSLEngine +instances (when using-nio="true"). When this attribute is omitted, normal plain text +sockets are used. + + + + + + + + + + + +A reference to a TcpSocketSupport strategy implementation. Allows post-processing Socket +and ServerSocket instances after creation and after configured attributes are applied. + + + + + + + + + + + +A reference to a TcpSocketFactorySupport strategy implementation. Allows customization +of the factories used to create sockets. The default implementation returns default. +ServerSocketFactory and SocketFactory instances, unless an 'ssl-context-support' +attribute has been supplied, in which case the SSLContext obtained therefrom is +used to create SSLServerSocketFactory and SSLSocketFactory instances. + + + + + + + + diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml index 998c9eb771..0ba6b8ae23 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/config/ParserUnitTests-context.xml @@ -58,8 +58,33 @@ port="#{tcpIpUtils.findAvailableServerSocket(5200)}" lookup-host="false" apply-sequence="true" + ssl-context-support="sslContextSupport" /> + + + + + + + + + + + + + + + + + (){ + public Socket answer(InvocationOnMock invocation) throws Throwable { + latch1.countDown(); + latch2.await(10, TimeUnit.SECONDS); + return null; + }}); + TcpSocketSupport socketSupport = mock(TcpSocketSupport.class); + + TcpNetServerConnectionFactory connectionFactory = new TcpNetServerConnectionFactory(0); + connectionFactory.setTcpSocketFactorySupport(factorySupport); + connectionFactory.setTcpSocketSupport(socketSupport); + connectionFactory.registerListener(mock(TcpListener.class)); + connectionFactory.start(); + + assertTrue(latch1.await(10, TimeUnit.SECONDS)); + verify(socketSupport).postProcessServerSocket(serverSocket); + verify(socketSupport).postProcessSocket(socket); + latch2.countDown(); + connectionFactory.stop(); + } + + @Test + public void testNioClientAndServer() throws Exception { + int port = SocketTestUtils.findAvailableServerSocket(); + TcpNioClientConnectionFactory clientConnectionFactory = new TcpNioClientConnectionFactory("localhost", port); + final AtomicInteger ppSocketCountClient = new AtomicInteger(); + final AtomicInteger ppServerSocketCountClient = new AtomicInteger(); + TcpSocketSupport clientSocketSupport = new TcpSocketSupport() { + public void postProcessSocket(Socket socket) { + ppSocketCountClient.incrementAndGet(); + } + public void postProcessServerSocket(ServerSocket serverSocket) { + ppServerSocketCountClient.incrementAndGet(); + } + }; + clientConnectionFactory.setTcpSocketSupport(clientSocketSupport); + clientConnectionFactory.start(); + TcpNioServerConnectionFactory serverConnectionFactory = new TcpNioServerConnectionFactory(port); + serverConnectionFactory.registerListener(new TcpListener() { + public boolean onMessage(Message message) { + return false; + } + }); + final AtomicInteger ppSocketCountServer = new AtomicInteger(); + final AtomicInteger ppServerSocketCountServer = new AtomicInteger(); + TcpSocketSupport serverSocketSupport = new TcpSocketSupport() { + public void postProcessSocket(Socket socket) { + ppSocketCountServer.incrementAndGet(); + } + public void postProcessServerSocket(ServerSocket serverSocket) { + ppServerSocketCountServer.incrementAndGet(); + } + }; + serverConnectionFactory.setTcpSocketSupport(serverSocketSupport); + serverConnectionFactory.start(); + waitListening(serverConnectionFactory); + clientConnectionFactory.getConnection().send(new GenericMessage("Hello, world!")); + assertEquals(0, ppServerSocketCountClient.get()); + assertEquals(1, ppSocketCountClient.get()); + + assertEquals(1, ppServerSocketCountServer.get()); + assertEquals(1, ppSocketCountServer.get()); + } + + /* +$ keytool -genkeypair -alias sitestcertkey -keyalg RSA -validity 36500 -keystore src/test/resources/test.ks +Enter keystore password: secret +Re-enter new password: secret +What is your first and last name? + [Unknown]: Spring Integration +What is the name of your organizational unit? + [Unknown]: SpringSource +What is the name of your organization? + [Unknown]: VMware +What is the name of your City or Locality? + [Unknown]: Palo Alto +What is the name of your State or Province? + [Unknown]: CA +What is the two-letter country code for this unit? + [Unknown]: US +Is CN=Spring Integration, OU=SpringSource, O=VMware, L=Palo Alto, ST=CA, C=US correct? + [no]: yes + +Enter key password for + (RETURN if same as keystore password): + +$ keytool -list -v -keystore src/test/resources/test.ks +Enter keystore password: secret + +Keystore type: JKS +Keystore provider: SUN + +Your keystore contains 1 entry + +Alias name: sitestcertkey +Creation date: Feb 25, 2012 +Entry type: PrivateKeyEntry +Certificate chain length: 1 +Certificate[1]: +Owner: CN=Spring Integration, OU=SpringSource, O=VMware, L=Palo Alto, ST=CA, C=US +Issuer: CN=Spring Integration, OU=SpringSource, O=VMware, L=Palo Alto, ST=CA, C=US +Serial number: 4f491902 +Valid from: Sat Feb 25 12:23:14 EST 2012 until: Mon Feb 01 12:23:14 EST 2112 +Certificate fingerprints: + MD5: 4F:A9:76:0E:A9:C0:A8:B7:26:E7:7E:C7:E8:22:1F:8B + SHA1: 88:AC:9E:4D:29:0D:3A:59:3B:73:95:4A:E1:BB:D0:22:89:37:64:4C + Signature algorithm name: SHA1withRSA + Version: 3 + + +******************************************* +******************************************* + +$ keytool -export -alias sitestcertkey -keystore src/test/resources/test.ks -rfc -file src/test/resources/test.cer +Enter keystore password: +Certificate stored in file + +$ keytool -import -alias sitestcertkey -file src/test/resources/test.cer -keystore src/test/resources/test.truststore.ks +Enter keystore password: secret +Re-enter new password: secret +Owner: CN=Spring Integration, OU=SpringSource, O=VMware, L=Palo Alto, ST=CA, C=US +Issuer: CN=Spring Integration, OU=SpringSource, O=VMware, L=Palo Alto, ST=CA, C=US +Serial number: 4f491902 +Valid from: Sat Feb 25 12:23:14 EST 2012 until: Mon Feb 01 12:23:14 EST 2112 +Certificate fingerprints: + MD5: 4F:A9:76:0E:A9:C0:A8:B7:26:E7:7E:C7:E8:22:1F:8B + SHA1: 88:AC:9E:4D:29:0D:3A:59:3B:73:95:4A:E1:BB:D0:22:89:37:64:4C + Signature algorithm name: SHA1withRSA + Version: 3 +Trust this certificate? [no]: yes +Certificate was added to keystore + +$ keytool -list -v -keystore src/test/resources/test.truststore.ks +Enter keystore password: secret + +Keystore type: JKS +Keystore provider: SUN + +Your keystore contains 1 entry + +Alias name: sitestcertkey +Creation date: Feb 25, 2012 +Entry type: trustedCertEntry + +Owner: CN=Spring Integration, OU=SpringSource, O=VMware, L=Palo Alto, ST=CA, C=US +Issuer: CN=Spring Integration, OU=SpringSource, O=VMware, L=Palo Alto, ST=CA, C=US +Serial number: 4f491902 +Valid from: Sat Feb 25 12:23:14 EST 2012 until: Mon Feb 01 12:23:14 EST 2112 +Certificate fingerprints: + MD5: 4F:A9:76:0E:A9:C0:A8:B7:26:E7:7E:C7:E8:22:1F:8B + SHA1: 88:AC:9E:4D:29:0D:3A:59:3B:73:95:4A:E1:BB:D0:22:89:37:64:4C + Signature algorithm name: SHA1withRSA + Version: 3 + + +******************************************* +******************************************* + + */ + @Test + public void testNetClientAndServerSSL() throws Exception { + System.setProperty("javax.net.debug", "all"); // SSL activity in the console + int port = SocketTestUtils.findAvailableServerSocket(); + TcpNetServerConnectionFactory server = new TcpNetServerConnectionFactory(port); + TcpSSLContextSupport sslContextSupport = new DefaultTcpSSLContextSupport("test.ks", + "test.truststore.ks", "secret", "secret"); + DefaultTcpNetSSLSocketFactorySupport tcpSocketFactorySupport = new DefaultTcpNetSSLSocketFactorySupport(sslContextSupport); + tcpSocketFactorySupport.afterPropertiesSet(); + server.setTcpSocketFactorySupport(tcpSocketFactorySupport); + final List> messages = new ArrayList>(); + final CountDownLatch latch = new CountDownLatch(1); + server.registerListener(new TcpListener() { + public boolean onMessage(Message message) { + messages.add(message); + latch.countDown(); + return false; + } + }); + server.start(); + waitListening(server); + + TcpNetClientConnectionFactory client = new TcpNetClientConnectionFactory("localhost", port); + client.setTcpSocketFactorySupport(tcpSocketFactorySupport); + client.start(); + + TcpConnection connection = client.getConnection(); + connection.send(new GenericMessage("Hello, world!")); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertEquals("Hello, world!", new String((byte[]) messages.get(0).getPayload())); + } + + @Test + public void testNetClientAndServerSSLDifferentContexts() throws Exception { + System.setProperty("javax.net.debug", "all"); // SSL activity in the console + int port = SocketTestUtils.findAvailableServerSocket(); + TcpNetServerConnectionFactory server = new TcpNetServerConnectionFactory(port); + TcpSSLContextSupport serverSslContextSupport = new DefaultTcpSSLContextSupport("server.ks", + "server.truststore.ks", "secret", "secret"); + DefaultTcpNetSSLSocketFactorySupport serverTcpSocketFactorySupport = new DefaultTcpNetSSLSocketFactorySupport(serverSslContextSupport); + serverTcpSocketFactorySupport.afterPropertiesSet(); + server.setTcpSocketFactorySupport(serverTcpSocketFactorySupport); + final List> messages = new ArrayList>(); + final CountDownLatch latch = new CountDownLatch(1); + server.registerListener(new TcpListener() { + public boolean onMessage(Message message) { + messages.add(message); + latch.countDown(); + return false; + } + }); + server.start(); + waitListening(server); + + TcpNetClientConnectionFactory client = new TcpNetClientConnectionFactory("localhost", port); + TcpSSLContextSupport clientSslContextSupport = new DefaultTcpSSLContextSupport("client.ks", "client.truststore.ks", + "secret", "secret"); + DefaultTcpNetSSLSocketFactorySupport clientTcpSocketFactorySupport = new DefaultTcpNetSSLSocketFactorySupport(clientSslContextSupport); + clientTcpSocketFactorySupport.afterPropertiesSet(); + client.setTcpSocketFactorySupport(clientTcpSocketFactorySupport); + client.start(); + + TcpConnection connection = client.getConnection(); + connection.send(new GenericMessage("Hello, world!")); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertEquals("Hello, world!", new String((byte[]) messages.get(0).getPayload())); + } + + @Test + public void testNioClientAndServerSSL() throws Exception { + System.setProperty("javax.net.debug", "all"); // SSL activity in the console + int port = SocketTestUtils.findAvailableServerSocket(); + TcpNioServerConnectionFactory server = new TcpNioServerConnectionFactory(port); + DefaultTcpSSLContextSupport sslContextSupport = new DefaultTcpSSLContextSupport("test.ks", + "test.truststore.ks", "secret", "secret"); + sslContextSupport.setProtocol("SSL"); + DefaultTcpNioSSLConnectionSupport tcpNioConnectionSupport = new DefaultTcpNioSSLConnectionSupport(sslContextSupport); + tcpNioConnectionSupport.afterPropertiesSet(); + server.setTcpNioConnectionSupport(tcpNioConnectionSupport); + final List> messages = new ArrayList>(); + final CountDownLatch latch = new CountDownLatch(1); + server.registerListener(new TcpListener() { + public boolean onMessage(Message message) { + System.out.println("Server" + message); + messages.add(message); + latch.countDown(); + return false; + } + }); + server.start(); + waitListening(server); + + TcpNioClientConnectionFactory client = new TcpNioClientConnectionFactory("localhost", port); + client.setTcpNioConnectionSupport(tcpNioConnectionSupport); + client.registerListener(new TcpListener() { + public boolean onMessage(Message message) { + System.out.println("Client" + message); + return false; + } + }); + client.start(); + + TcpConnection connection = client.getConnection(); + connection.send(new GenericMessage("Hello, world!")); + assertTrue(latch.await(10, TimeUnit.SECONDS)); + assertEquals("Hello, world!", new String((byte[]) messages.get(0).getPayload())); + } + + @Test + public void testNioClientAndServerSSLDifferentContextsLargeDataWithReply() throws Exception { + System.setProperty("javax.net.debug", "all"); // SSL activity in the console + int port = SocketTestUtils.findAvailableServerSocket(); + TcpNioServerConnectionFactory server = new TcpNioServerConnectionFactory(port); + TcpSSLContextSupport serverSslContextSupport = new DefaultTcpSSLContextSupport("server.ks", + "server.truststore.ks", "secret", "secret"); + DefaultTcpNioSSLConnectionSupport serverTcpNioConnectionSupport = new DefaultTcpNioSSLConnectionSupport(serverSslContextSupport); + serverTcpNioConnectionSupport.afterPropertiesSet(); + server.setTcpNioConnectionSupport(serverTcpNioConnectionSupport); + final List> messages = new ArrayList>(); + final CountDownLatch latch = new CountDownLatch(2); + final Replier replier = new Replier(); + server.registerSender(replier); + server.registerListener(new TcpListener() { + public boolean onMessage(Message message) { + System.out.println("Server:" + message); + messages.add(message); + try { + replier.send(message); + } catch (Exception e) { + e.printStackTrace(); + } + latch.countDown(); + return false; + } + }); + ByteArrayCrLfSerializer deserializer = new ByteArrayCrLfSerializer(); + deserializer.setMaxMessageSize(120000); + server.setDeserializer(deserializer); + server.start(); + waitListening(server); + + TcpNioClientConnectionFactory client = new TcpNioClientConnectionFactory("localhost", port); + TcpSSLContextSupport clientSslContextSupport = new DefaultTcpSSLContextSupport("client.ks", + "client.truststore.ks", "secret", "secret"); + DefaultTcpNioSSLConnectionSupport clientTcpNioConnectionSupport = new DefaultTcpNioSSLConnectionSupport(clientSslContextSupport); + clientTcpNioConnectionSupport.afterPropertiesSet(); + client.setTcpNioConnectionSupport(clientTcpNioConnectionSupport); + client.registerListener(new TcpListener() { + public boolean onMessage(Message message) { + System.out.println("Client:" + message); + messages.add(message); + latch.countDown(); + return false; + } + }); + client.setDeserializer(deserializer); + client.start(); + + TcpConnection connection = client.getConnection(); + byte[] bytes = new byte[100000]; + connection.send(new GenericMessage("Hello, world!" + new String(bytes))); + assertTrue(latch.await(60, TimeUnit.SECONDS)); + byte[] payload = (byte[]) messages.get(0).getPayload(); + assertEquals(13 + bytes.length, payload.length); + assertEquals("Hello, world!", new String(payload).substring(0, 13)); + payload = (byte[]) messages.get(1).getPayload(); + assertEquals(13 + bytes.length, payload.length); + assertEquals("Hello, world!", new String(payload).substring(0, 13)); + } + + private void waitListening(AbstractServerConnectionFactory scf) throws Exception { + int n = 0; + while (!scf.isListening()) { + Thread.sleep(100); + if (++n > 100) { + fail("Server failed to start listening"); + } + } + } + + private class Replier implements TcpSender { + + private TcpConnection connection; + + public void addNewConnection(TcpConnection connection) { + this.connection = connection; + } + + public void removeDeadConnection(TcpConnection connection) { + } + + public void send(Message message) throws Exception { + // force a renegotiation from the server side + SSLEngine sslEngine = TestUtils.getPropertyValue(this.connection, "sslEngine", SSLEngine.class); + sslEngine.getSession().invalidate(); + sslEngine.beginHandshake(); + this.connection.send(message); + } + } +} diff --git a/spring-integration-ip/src/test/resources/client.cer b/spring-integration-ip/src/test/resources/client.cer new file mode 100644 index 0000000000..7fad781ccc --- /dev/null +++ b/spring-integration-ip/src/test/resources/client.cer @@ -0,0 +1,14 @@ +-----BEGIN CERTIFICATE----- +MIICdzCCAeCgAwIBAgIET0k29DANBgkqhkiG9w0BAQUFADB/MQswCQYDVQQGEwJVUzELMAkGA1UE +CBMCQ0ExEjAQBgNVBAcTCVBhbG8gQWx0bzEPMA0GA1UEChMGVk13YXJlMRUwEwYDVQQLEwxTcHJp +bmdTb3VyY2UxJzAlBgNVBAMTHlNwcmluZyBJbnRlZ3JhdGlvbiBUZXN0IENsaWVudDAgFw0xMjAy +MjUxOTMxMDBaGA8yMTEyMDIwMTE5MzEwMFowfzELMAkGA1UEBhMCVVMxCzAJBgNVBAgTAkNBMRIw +EAYDVQQHEwlQYWxvIEFsdG8xDzANBgNVBAoTBlZNd2FyZTEVMBMGA1UECxMMU3ByaW5nU291cmNl +MScwJQYDVQQDEx5TcHJpbmcgSW50ZWdyYXRpb24gVGVzdCBDbGllbnQwgZ8wDQYJKoZIhvcNAQEB +BQADgY0AMIGJAoGBAKL9XxG1vf6SXW7iNKR7bMbEqIgWzg6rxsr1e+35zsmTHhGwoRJJw/Fm/+Sc +bHxkO3YzM+WayQwe0NXttjMgjLnFoztEpTpWYsmzJt0r6fv7rGtixXFfLjmulONmakGFptwRZlWv +HhdDYkQiw3P8npC/ZWMQagaMHpxzlzyeD6BfAgMBAAEwDQYJKoZIhvcNAQEFBQADgYEAbsPnIYVl +1ClEkmPp/aSZooGtEd1JXUvW4JXWWt6oxTVe2SdU62GvxHDc71z73Fvqk1BFoKtniuOgrjZNeL2j +kGVbeB2wT+aG9soXifcYolNDXeNVF5xpYxHmDb1LyMiEpHl071mhD0q+a5Q1gFly5HbDb+RvrsNt +k6ooEx5Ycuo= +-----END CERTIFICATE----- diff --git a/spring-integration-ip/src/test/resources/client.ks b/spring-integration-ip/src/test/resources/client.ks new file mode 100644 index 0000000000..affaedd561 Binary files /dev/null and b/spring-integration-ip/src/test/resources/client.ks differ diff --git a/spring-integration-ip/src/test/resources/client.truststore.ks b/spring-integration-ip/src/test/resources/client.truststore.ks new file mode 100644 index 0000000000..1d06493923 Binary files /dev/null and b/spring-integration-ip/src/test/resources/client.truststore.ks differ diff --git a/spring-integration-ip/src/test/resources/server.cer b/spring-integration-ip/src/test/resources/server.cer new file mode 100644 index 0000000000..465e107a99 --- /dev/null +++ b/spring-integration-ip/src/test/resources/server.cer @@ -0,0 +1,14 @@ +-----BEGIN CERTIFICATE----- +MIICdzCCAeCgAwIBAgIET0k3GjANBgkqhkiG9w0BAQUFADB/MQswCQYDVQQGEwJVUzELMAkGA1UE +CBMCQ0ExEjAQBgNVBAcTCVBhbG8gQWx0bzEPMA0GA1UEChMGVk13YXJlMRUwEwYDVQQLEwxTcHJp +bmdTb3VyY2UxJzAlBgNVBAMTHlNwcmluZyBJbnRlZ3JhdGlvbiBUZXN0IFNlcnZlcjAgFw0xMjAy +MjUxOTMxMzhaGA8yMTEyMDIwMTE5MzEzOFowfzELMAkGA1UEBhMCVVMxCzAJBgNVBAgTAkNBMRIw +EAYDVQQHEwlQYWxvIEFsdG8xDzANBgNVBAoTBlZNd2FyZTEVMBMGA1UECxMMU3ByaW5nU291cmNl +MScwJQYDVQQDEx5TcHJpbmcgSW50ZWdyYXRpb24gVGVzdCBTZXJ2ZXIwgZ8wDQYJKoZIhvcNAQEB +BQADgY0AMIGJAoGBAM/YT0knf57/elITdhqQ5TgUKcbIKFncb3tmDrbZTm8kKWT21cUUvx5g0ZH4 +dHtCDOY1Zjd4Qyh0YQ1tXbpccpU014IKDHAc85BQGoUiFpUKAMrJ7jGff7iflbeXMgizFnQXtdb/ +fH7E8YGTP0XV2RizMPAClWcJoHBf7LjYzbknAgMBAAEwDQYJKoZIhvcNAQEFBQADgYEArGlIeCKf +Mj+IL9tJU4fxEGTTMCINBBOQB5sOmvsfZZddLcHGoYrFxxZxqUeVwv/VHRAGUe5kN6nTsGn5hFud +HeFswEgx/YIQtdRW1HYt3uuyZh+20cOq4uaNS7YFvPCIe7yBscK7PRwKrVspVZQFLFtcRIJUkF+0 +cOoQRaygXys= +-----END CERTIFICATE----- diff --git a/spring-integration-ip/src/test/resources/server.ks b/spring-integration-ip/src/test/resources/server.ks new file mode 100644 index 0000000000..f8f0cdc9c4 Binary files /dev/null and b/spring-integration-ip/src/test/resources/server.ks differ diff --git a/spring-integration-ip/src/test/resources/server.truststore.ks b/spring-integration-ip/src/test/resources/server.truststore.ks new file mode 100644 index 0000000000..4359d256d6 Binary files /dev/null and b/spring-integration-ip/src/test/resources/server.truststore.ks differ diff --git a/spring-integration-ip/src/test/resources/test.cer b/spring-integration-ip/src/test/resources/test.cer new file mode 100644 index 0000000000..e46e6ae864 --- /dev/null +++ b/spring-integration-ip/src/test/resources/test.cer @@ -0,0 +1,13 @@ +-----BEGIN CERTIFICATE----- +MIICXzCCAcigAwIBAgIET0kZAjANBgkqhkiG9w0BAQUFADBzMQswCQYDVQQGEwJVUzELMAkGA1UE +CBMCQ0ExEjAQBgNVBAcTCVBhbG8gQWx0bzEPMA0GA1UEChMGVk13YXJlMRUwEwYDVQQLEwxTcHJp +bmdTb3VyY2UxGzAZBgNVBAMTElNwcmluZyBJbnRlZ3JhdGlvbjAgFw0xMjAyMjUxNzIzMTRaGA8y +MTEyMDIwMTE3MjMxNFowczELMAkGA1UEBhMCVVMxCzAJBgNVBAgTAkNBMRIwEAYDVQQHEwlQYWxv +IEFsdG8xDzANBgNVBAoTBlZNd2FyZTEVMBMGA1UECxMMU3ByaW5nU291cmNlMRswGQYDVQQDExJT +cHJpbmcgSW50ZWdyYXRpb24wgZ8wDQYJKoZIhvcNAQEBBQADgY0AMIGJAoGBAM6hHqm4jCixwNgK +z5kBxsWbuGvSSLMiG8fMbg6RbVmbhh4ssVttzjcC3G2OxUxC2gQ9H/96PwgGJZp4VKZw8cPYVTZe +kX79NKvv1IBQ661LbFMF7yH0bMNtU8I/dT5P+hrvNbWT/oo5YYvI4LkDfrw4l4lqWNcW5Wyg40NO +7Yo7AgMBAAEwDQYJKoZIhvcNAQEFBQADgYEAbSkOrZKZ9caK4TJhJPD/6HC8PfJRcRc4hBdM54UX +4BxW9VRhrjZLS9luWrnVqfrqiZ49UuApTK+5K12GAcmkZGLJzDzaM6D55dW6JC7YlZEQQxHN0GvG +PqgOxu248fIqrasq4KXUGLvhL31ylRXZIcfEo15XpWwIhrKOWo2MBYw= +-----END CERTIFICATE----- diff --git a/spring-integration-ip/src/test/resources/test.ks b/spring-integration-ip/src/test/resources/test.ks new file mode 100644 index 0000000000..243b3d0244 Binary files /dev/null and b/spring-integration-ip/src/test/resources/test.ks differ diff --git a/spring-integration-ip/src/test/resources/test.truststore.ks b/spring-integration-ip/src/test/resources/test.truststore.ks new file mode 100644 index 0000000000..24ead4bc6d Binary files /dev/null and b/spring-integration-ip/src/test/resources/test.truststore.ks differ diff --git a/src/reference/docbook/ip.xml b/src/reference/docbook/ip.xml index 33ade1e89a..1f29dc42c1 100644 --- a/src/reference/docbook/ip.xml +++ b/src/reference/docbook/ip.xml @@ -351,6 +351,13 @@ + + + It is possible to modify the creation of and/or attributes of sockets - see + . As is noted there, such modifications are possible whether + or not SSL is being used. + +

TCP Connection Interceptors @@ -816,6 +823,157 @@ a new message is received.
+
+ SSL/TLS Support +
+ Overview + + Secure Sockets Layer/Transport Layer Security is supported. When using NIO, the JDK 5+ + SSLEngine feature is used to handle handshaking after the + connection is established. When not using NIO, standard + SSLSocketFactory and SSLServerSocketFactory objects are + used to create connections. A number of strategy interfaces are provided to allow + significant customization; default implementations of these interfaces provide for + the simplest way to get started with secure communications. + +
+
+ Getting Started + + Regardless of whether NIO is being used, you need to configure the + ssl-context-support attribute on the connection factory. + This attribute references a <bean/> definition that describes the location + and passwords for the required key stores. + + + SSL/TLS peers require two keystores each; a keystore containing private/public key + pairs identifying the peer; a truststore, containing the public keys for peers that + are trusted. See the documentation for the keytool utility + provided with the JDK. The essential steps are + + + + Create a new key pair and store in a keystore. + Export the public key. + Import the public key into the peer's truststore. + + + + Repeat for the other peer. + + + + It is common in test cases to use the same key stores on both peers, but this should + be avoided for production. + + + + After establishing the key stores, the next step is to indicate their locations to the + TcpSSLContextSupport bean, and provide a reference to that bean + to the connection factory. + + + + + + + + + + + + The DefaulTcpSSLContextSupport class also has an optional + 'protocol' property, which can be 'SSL' or 'TLS' (default). + + + 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. + +
+
+ Advanced Techniques + + In many cases, the configuration described above is all that is needed to enable secure + communication over TCP/IP. However, a number of strategy interfaces are provided to + allow customization and modification of socket factories and sockets. + + + + TcpSSLContextSupport + TcpSocketFactorySupport + TcpSocketSupport + + + + + + Implementations of this interface are responsible for creating an SSLContext. + The sole implementation provided by the framework is the + DefaultTcpSSLContextSupport described above. If you require + different behavior, implement this interface and provide the connection factory with + a reference to a bean of your class' implementation. + + + + + Implementations of this interface are responsible for obtaining references to + ServerSocketFactory and SocketFactory. + Two implementations are provided; the first is DefaultTcpNetSocketFactorySupport + for non-SSL sockets (when no 'ssl-context-support' attribute is defined); this simply + uses the JDK's default factories. The second implementation is + DefaultTcpNetSSLSocketFactorySupport; this is used, by default, + when an 'ssl-context-support' attribute is defined; it uses the SSLContext + created by that bean to create the socket factories. + + + + This interface only applies if using-nio is "false"; socket factories + are not used by NIO. + + + + + + Implementations of this interface can modify sockets after they are created, and after + all configured attributes have been applied, but before the sockets are used. This applies + whether or not NIO is being used. For example, + you could use an implementation of this interface to modify the supported cipher suites on + an SSL socket, or you could add a listener that gets notified after SSL handshaking is + complete. The sole implementation provided by the framework is the + DefaultTcpSocketSupport which does not modify the sockets in + any way + + + To supply your own implementation of TcpSocketFactorySupport or + TcpSocketSupport, provide the connection factory with references to + beans of your custom type using the socket-factory-support and + socket-support attributes, respectively. + +
+
IP Configuration Attributes @@ -1020,6 +1178,27 @@ See + + ssl-context-support + Y + Y + + See + + + socket-factory-support + Y + Y + + See + + + socket-support + Y + Y + + See +