andreacavalli/netty5
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Port SSL echo test

Browse Source 
- Remove britspaces
This commit is contained in:
Trustin Lee 2012-06-06 23:18:37 +09:00
parent 843a94b989
commit a9cc75dd3e
2 changed files with 75 additions and 113 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

64
handler/src/main/java/io/netty/handler/ssl/SslHandler.java
View File

@ -164,26 +164,26 @@ public class SslHandler extends StreamToStreamCodec {
private final SSLEngine engine; private final SSLEngine engine;
private final SslBufferPool bufferPool; private final SslBufferPool bufferPool;
private final Executor delegatedTaskExecutor; private final Executor delegatedTaskExecutor;
private final boolean startTls; private final boolean startTls;
private boolean sentFirstMessage; private boolean sentFirstMessage;
private volatile boolean enableRenegotiation = true; private volatile boolean enableRenegotiation = true;
final Object handshakeLock = new Object(); final Object handshakeLock = new Object();
private boolean handshaking; private boolean handshaking;
private volatile boolean handshaken; private volatile boolean handshaken;
private ChannelFuture handshakeFuture; private ChannelFuture handshakeFuture;
private boolean sentCloseNotify; private boolean sentCloseNotify;
int ignoreClosedChannelException; int ignoreClosedChannelException;
final Object ignoreClosedChannelExceptionLock = new Object(); final Object ignoreClosedChannelExceptionLock = new Object();
private volatile boolean issueHandshake; private volatile boolean issueHandshake;
private final SSLEngineInboundCloseFuture sslEngineCloseFuture = new SSLEngineInboundCloseFuture(); private final SSLEngineInboundCloseFuture sslEngineCloseFuture = new SSLEngineInboundCloseFuture();
private int packetLength = -1; private int packetLength = -1;
/** /**
@ -337,36 +337,36 @@ public class SslHandler extends StreamToStreamCodec {
engine.beginHandshake(); engine.beginHandshake();
runDelegatedTasks(); runDelegatedTasks();
wrapNonAppData(ctx, channel); wrapNonAppData(ctx, channel);
} catch (Exception e) { } catch (Exception e) {
exception = e; exception = e;
} }
} else { } else {
ctx.executor().execute(new Runnable() { ctx.executor().execute(new Runnable() {
@Override @Override
public void run() { public void run() {
Throwable exception = null; Throwable exception = null;
synchronized (handshakeLock) { synchronized (handshakeLock) {
try { try {
engine.beginHandshake(); engine.beginHandshake();
runDelegatedTasks(); runDelegatedTasks();
wrapNonAppData(ctx, ctx.channel()); wrapNonAppData(ctx, ctx.channel());
} catch (Exception e) { } catch (Exception e) {
exception = e; exception = e;
} }
} }
if (exception != null) { // Failed to initiate handshake. if (exception != null) { // Failed to initiate handshake.
handshakeFuture.setFailure(exception); handshakeFuture.setFailure(exception);
ctx.fireExceptionCaught(exception); ctx.fireExceptionCaught(exception);
} }
} }
}); });
} }
} }
} }
@ -459,7 +459,7 @@ public class SslHandler extends StreamToStreamCodec {
out.writeBytes(in); out.writeBytes(in);
return; return;
} }
ByteBuffer outNetBuf = bufferPool.acquireBuffer(); ByteBuffer outNetBuf = bufferPool.acquireBuffer();
boolean success = true; boolean success = true;
boolean needsUnwrap = false; boolean needsUnwrap = false;
@ -467,7 +467,7 @@ public class SslHandler extends StreamToStreamCodec {
ByteBuffer outAppBuf = in.nioBuffer(); ByteBuffer outAppBuf = in.nioBuffer();
while(in.readable()) { while(in.readable()) {
int read; int read;
int remaining = outAppBuf.remaining(); int remaining = outAppBuf.remaining();
SSLEngineResult result = null; SSLEngineResult result = null;
@ -477,14 +477,14 @@ public class SslHandler extends StreamToStreamCodec {
} }
read = remaining - outAppBuf.remaining(); read = remaining - outAppBuf.remaining();
in.readerIndex(in.readerIndex() + read); in.readerIndex(in.readerIndex() + read);
if (result.bytesProduced() > 0) { if (result.bytesProduced() > 0) {
outNetBuf.flip(); outNetBuf.flip();
out.writeBytes(outNetBuf); out.writeBytes(outNetBuf);
outNetBuf.clear(); outNetBuf.clear();
} else if (result.getStatus() == Status.CLOSED) { } else if (result.getStatus() == Status.CLOSED) {
// SSLEngine has been closed already. // SSLEngine has been closed already.
@ -519,7 +519,7 @@ public class SslHandler extends StreamToStreamCodec {
throw new IllegalStateException("Unknown handshake status: " + handshakeStatus); throw new IllegalStateException("Unknown handshake status: " + handshakeStatus);
} }
} }
} }
} catch (SSLException e) { } catch (SSLException e) {
success = false; success = false;
@ -527,14 +527,14 @@ public class SslHandler extends StreamToStreamCodec {
throw e; throw e;
} finally { } finally {
bufferPool.releaseBuffer(outNetBuf); bufferPool.releaseBuffer(outNetBuf);
if (!success) { if (!success) {
// mark all bytes as read // mark all bytes as read
in.readerIndex(in.readerIndex() + in.readableBytes()); in.readerIndex(in.readerIndex() + in.readableBytes());
throw new IllegalStateException("SSLEngine already closed"); throw new IllegalStateException("SSLEngine already closed");
} }
} }
@ -608,7 +608,7 @@ public class SslHandler extends StreamToStreamCodec {
super.exceptionCaught(ctx, cause); super.exceptionCaught(ctx, cause);
} }
@Override @Override
public void decode(ChannelInboundHandlerContext<Byte> ctx, ChannelBuffer in, ChannelBuffer out) throws Exception { public void decode(ChannelInboundHandlerContext<Byte> ctx, ChannelBuffer in, ChannelBuffer out) throws Exception {
@ -647,7 +647,7 @@ public class SslHandler extends StreamToStreamCodec {
tls = false; tls = false;
} }
} }
if (!tls) { if (!tls) {
// SSLv2 or bad data - Check the version // SSLv2 or bad data - Check the version
boolean sslv2 = true; boolean sslv2 = true;
@ -677,11 +677,11 @@ public class SslHandler extends StreamToStreamCodec {
throw e; throw e;
} }
} }
assert packetLength > 0; assert packetLength > 0;
} }
if (in.readableBytes() < packetLength) { if (in.readableBytes() < packetLength) {
// not enough bytes left to read the packet // not enough bytes left to read the packet
// so return here for now // so return here for now
@ -1073,7 +1073,7 @@ public class SslHandler extends StreamToStreamCodec {
@Override @Override
public void beforeAdd(ChannelHandlerContext ctx) throws Exception { public void beforeAdd(ChannelHandlerContext ctx) throws Exception {
this.ctx = ctx; this.ctx = ctx;
this.handshakeFuture = ctx.newFuture(); handshakeFuture = ctx.newFuture();
} }
@ -1102,7 +1102,7 @@ public class SslHandler extends StreamToStreamCodec {
}); });
} else { } else {
super.channelActive(ctx); super.channelActive(ctx);
} }
} }
private final class SSLEngineInboundCloseFuture extends DefaultChannelFuture { private final class SSLEngineInboundCloseFuture extends DefaultChannelFuture {
@ -1134,6 +1134,6 @@ public class SslHandler extends StreamToStreamCodec {
return false; return false;
} }
} }
} }

124
testsuite/src/test/java/io/netty/testsuite/transport/socket/AbstractSocketSslEchoTest.java → testsuite/src/test/java/io/netty/testsuite/transport/socket/SocketSslEchoTest.java
View File

@ -16,22 +16,21 @@
package io.netty.testsuite.transport.socket; package io.netty.testsuite.transport.socket;
import static org.junit.Assert.*; import static org.junit.Assert.*;
import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap; import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ChannelBuffer; import io.netty.buffer.ChannelBuffer;
import io.netty.buffer.ChannelBuffers; import io.netty.buffer.ChannelBuffers;
import io.netty.channel.Channel; import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerContext;
import io.netty.channel.ChannelInboundStreamHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.SocketChannel;
import io.netty.handler.ssl.SslHandler; import io.netty.handler.ssl.SslHandler;
import io.netty.logging.InternalLogger;
import io.netty.logging.InternalLoggerFactory;
import io.netty.util.SocketAddresses;
import io.netty.util.internal.ExecutorUtil;
import java.io.ByteArrayInputStream; import java.io.ByteArrayInputStream;
import java.io.IOException; import java.io.IOException;
import java.io.InputStream; import java.io.InputStream;
import java.net.InetSocketAddress;
import java.security.InvalidAlgorithmParameterException; import java.security.InvalidAlgorithmParameterException;
import java.security.KeyStore; import java.security.KeyStore;
import java.security.KeyStoreException; import java.security.KeyStoreException;
@ -39,9 +38,6 @@ import java.security.Security;
import java.security.cert.CertificateException; import java.security.cert.CertificateException;
import java.security.cert.X509Certificate; import java.security.cert.X509Certificate;
import java.util.Random; import java.util.Random;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.atomic.AtomicReference;
import javax.net.ssl.KeyManagerFactory; import javax.net.ssl.KeyManagerFactory;
@ -52,83 +48,49 @@ import javax.net.ssl.TrustManager;
import javax.net.ssl.TrustManagerFactorySpi; import javax.net.ssl.TrustManagerFactorySpi;
import javax.net.ssl.X509TrustManager; import javax.net.ssl.X509TrustManager;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test; import org.junit.Test;
public abstract class AbstractSocketSslEchoTest { public class SocketSslEchoTest extends AbstractSocketTest {
static final InternalLogger logger =
InternalLoggerFactory.getInstance(AbstractSocketSslEchoTest.class);
private static final Random random = new Random(); private static final Random random = new Random();
static final byte[] data = new byte[1048576]; static final byte[] data = new byte[1048576];
private static ExecutorService executor;
private static ExecutorService eventExecutor;
static { static {
random.nextBytes(data); random.nextBytes(data);
} }
@BeforeClass
public static void init() {
executor = Executors.newCachedThreadPool();
eventExecutor = new OrderedMemoryAwareThreadPoolExecutor(16, 0, 0);
}
@AfterClass
public static void destroy() {
ExecutorUtil.terminate(executor, eventExecutor);
}
protected abstract ChannelFactory newServerSocketChannelFactory(Executor executor);
protected abstract ChannelFactory newClientSocketChannelFactory(Executor executor);
protected boolean isExecutorRequired() {
return false;
}
@Test @Test
public void testSslEcho() throws Throwable { public void testSslEcho() throws Throwable {
ServerBootstrap sb = new ServerBootstrap(newServerSocketChannelFactory(executor)); run();
ClientBootstrap cb = new ClientBootstrap(newClientSocketChannelFactory(executor)); }
EchoHandler sh = new EchoHandler(true); public void testSslEcho(ServerBootstrap sb, Bootstrap cb) throws Throwable {
EchoHandler ch = new EchoHandler(false); final EchoHandler sh = new EchoHandler(true);
final EchoHandler ch = new EchoHandler(false);
SSLEngine sse = BogusSslContextFactory.getServerContext().createSSLEngine(); final SSLEngine sse = BogusSslContextFactory.getServerContext().createSSLEngine();
SSLEngine cse = BogusSslContextFactory.getClientContext().createSSLEngine(); final SSLEngine cse = BogusSslContextFactory.getClientContext().createSSLEngine();
sse.setUseClientMode(false); sse.setUseClientMode(false);
cse.setUseClientMode(true); cse.setUseClientMode(true);
// Workaround for blocking I/O transport write-write dead lock. sb.childHandler(new ChannelInitializer<SocketChannel>() {
sb.setOption("receiveBufferSize", 1048576); @Override
sb.setOption("receiveBufferSize", 1048576); public void initChannel(SocketChannel sch) throws Exception {
sch.pipeline().addFirst("ssl", new SslHandler(sse));
sb.pipeline().addFirst("ssl", new SslHandler(sse)); sch.pipeline().addLast("handler", sh);
sb.pipeline().addLast("handler", sh);
cb.pipeline().addFirst("ssl", new SslHandler(cse));
cb.pipeline().addLast("handler", ch);
if (isExecutorRequired()) {
sb.pipeline().addFirst("executor", new ExecutionHandler(eventExecutor));
cb.pipeline().addFirst("executor", new ExecutionHandler(eventExecutor));
}
Channel sc = sb.bind(new InetSocketAddress(0));
int port = ((InetSocketAddress) sc.getLocalAddress()).getPort();
ChannelFuture ccf = cb.connect(new InetSocketAddress(SocketAddresses.LOCALHOST, port));
ccf.awaitUninterruptibly();
if (!ccf.isSuccess()) {
if(logger.isErrorEnabled()) {
logger.error("Connection attempt failed", ccf.cause());
} }
sc.close().awaitUninterruptibly(); });
}
assertTrue(ccf.isSuccess());
Channel cc = ccf.channel(); cb.handler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel sch) throws Exception {
sch.pipeline().addFirst("ssl", new SslHandler(cse));
sch.pipeline().addLast("handler", ch);
}
});
Channel sc = sb.bind().sync().channel();
Channel cc = cb.connect().sync().channel();
ChannelFuture hf = cc.pipeline().get(SslHandler.class).handshake(); ChannelFuture hf = cc.pipeline().get(SslHandler.class).handshake();
hf.awaitUninterruptibly(); hf.awaitUninterruptibly();
if (!hf.isSuccess()) { if (!hf.isSuccess()) {
@ -194,7 +156,7 @@ public abstract class AbstractSocketSslEchoTest {
} }
} }
private static class EchoHandler extends SimpleChannelUpstreamHandler { private class EchoHandler extends ChannelInboundStreamHandlerAdapter {
volatile Channel channel; volatile Channel channel;
final AtomicReference<Throwable> exception = new AtomicReference<Throwable>(); final AtomicReference<Throwable> exception = new AtomicReference<Throwable>();
volatile int counter; volatile int counter;
@ -205,41 +167,41 @@ public abstract class AbstractSocketSslEchoTest {
} }
@Override @Override
public void channelOpen(ChannelHandlerContext ctx, ChannelStateEvent e) public void channelActive(ChannelInboundHandlerContext<Byte> ctx)
throws Exception { throws Exception {
channel = e.channel(); channel = ctx.channel();
} }
@Override @Override
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) public void inboundBufferUpdated(
ChannelInboundHandlerContext<Byte> ctx, ChannelBuffer in)
throws Exception { throws Exception {
ChannelBuffer m = (ChannelBuffer) e.getMessage(); byte[] actual = new byte[in.readableBytes()];
byte[] actual = new byte[m.readableBytes()]; in.readBytes(actual);
m.getBytes(0, actual);
int lastIdx = counter; int lastIdx = counter;
for (int i = 0; i < actual.length; i ++) { for (int i = 0; i < actual.length; i ++) {
assertEquals(data[i + lastIdx], actual[i]); assertEquals(data[i + lastIdx], actual[i]);
} }
if (channel.getParent() != null) { if (channel.parent() != null) {
channel.write(m); channel.write(ChannelBuffers.wrappedBuffer(actual));
} }
counter += actual.length; counter += actual.length;
} }
@Override @Override
public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) public void exceptionCaught(ChannelInboundHandlerContext<Byte> ctx,
throws Exception { Throwable cause) throws Exception {
if (logger.isWarnEnabled()) { if (logger.isWarnEnabled()) {
logger.warn( logger.warn(
"Unexpected exception from the " + "Unexpected exception from the " +
(server? "server" : "client") + " side", e.cause()); (server? "server" : "client") + " side", cause);
} }
exception.compareAndSet(null, e.cause()); exception.compareAndSet(null, cause);
e.channel().close(); ctx.close();
} }
} }