1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16 package io.netty.testsuite.transport.socket;
17
18 import io.netty.bootstrap.Bootstrap;
19 import io.netty.bootstrap.ServerBootstrap;
20 import io.netty.buffer.ByteBuf;
21 import io.netty.buffer.Unpooled;
22 import io.netty.channel.Channel;
23 import io.netty.channel.ChannelFuture;
24 import io.netty.channel.ChannelHandler.Sharable;
25 import io.netty.channel.ChannelHandlerContext;
26 import io.netty.channel.ChannelInboundHandlerAdapter;
27 import io.netty.channel.ChannelInitializer;
28 import io.netty.channel.ChannelOption;
29 import io.netty.channel.SimpleChannelInboundHandler;
30 import io.netty.handler.ssl.OpenSsl;
31 import io.netty.handler.ssl.OpenSslContext;
32 import io.netty.handler.ssl.SslContext;
33 import io.netty.handler.ssl.SslContextBuilder;
34 import io.netty.handler.ssl.SslHandler;
35 import io.netty.handler.ssl.SslHandshakeCompletionEvent;
36 import io.netty.handler.ssl.SslProvider;
37 import io.netty.handler.stream.ChunkedWriteHandler;
38 import io.netty.pkitesting.CertificateBuilder;
39 import io.netty.pkitesting.X509Bundle;
40 import io.netty.testsuite.util.TestUtils;
41 import io.netty.util.concurrent.Future;
42 import io.netty.util.internal.PlatformDependent;
43 import io.netty.util.internal.logging.InternalLogger;
44 import io.netty.util.internal.logging.InternalLoggerFactory;
45 import org.junit.jupiter.api.AfterAll;
46 import org.junit.jupiter.api.TestInfo;
47 import org.junit.jupiter.api.Timeout;
48 import org.junit.jupiter.params.ParameterizedTest;
49 import org.junit.jupiter.params.provider.MethodSource;
50
51 import javax.net.ssl.SSLEngine;
52 import java.io.File;
53 import java.io.IOException;
54 import java.util.ArrayList;
55 import java.util.Collection;
56 import java.util.List;
57 import java.util.Random;
58 import java.util.SplittableRandom;
59 import java.util.concurrent.CountDownLatch;
60 import java.util.concurrent.ExecutorService;
61 import java.util.concurrent.Executors;
62 import java.util.concurrent.TimeUnit;
63 import java.util.concurrent.atomic.AtomicInteger;
64 import java.util.concurrent.atomic.AtomicReference;
65
66 import static io.netty.testsuite.transport.TestsuitePermutation.randomBufferType;
67 import static org.assertj.core.api.Assertions.assertThat;
68 import static org.junit.jupiter.api.Assertions.assertEquals;
69 import static org.junit.jupiter.api.Assertions.assertFalse;
70 import static org.junit.jupiter.api.Assertions.assertNotSame;
71 import static org.junit.jupiter.api.Assertions.assertSame;
72 import static org.junit.jupiter.api.Assertions.assertTrue;
73
74 public class SocketSslEchoTest extends AbstractSocketTest {
75
76 private static final InternalLogger logger = InternalLoggerFactory.getInstance(SocketSslEchoTest.class);
77
78 private static final int FIRST_MESSAGE_SIZE = 16384;
79 private static final Random random = new Random();
80 private static final File CERT_FILE;
81 private static final File KEY_FILE;
82 static final byte[] data = new byte[1048576];
83
84 static {
85 PlatformDependent.splittableRandomNextBytes(new SplittableRandom(random.nextLong()), data);
86
87 try {
88 X509Bundle cert = new CertificateBuilder()
89 .rsa2048()
90 .subject("cn=localhost")
91 .setIsCertificateAuthority(true)
92 .buildSelfSigned();
93 CERT_FILE = cert.toTempCertChainPem();
94 KEY_FILE = cert.toTempPrivateKeyPem();
95 } catch (Exception e) {
96 throw new ExceptionInInitializerError(e);
97 }
98 }
99
100 protected enum RenegotiationType {
101 NONE,
102 CLIENT_INITIATED,
103 SERVER_INITIATED,
104 }
105
106 protected static class Renegotiation {
107 static final Renegotiation NONE = new Renegotiation(RenegotiationType.NONE, null);
108
109 final RenegotiationType type;
110 final String cipherSuite;
111
112 Renegotiation(RenegotiationType type, String cipherSuite) {
113 this.type = type;
114 this.cipherSuite = cipherSuite;
115 }
116
117 @Override
118 public String toString() {
119 if (type == RenegotiationType.NONE) {
120 return "NONE";
121 }
122
123 return type + "(" + cipherSuite + ')';
124 }
125 }
126
127 public static Collection<Object[]> data() throws Exception {
128 List<SslContext> serverContexts = new ArrayList<SslContext>();
129 serverContexts.add(SslContextBuilder.forServer(CERT_FILE, KEY_FILE)
130 .sslProvider(SslProvider.JDK)
131
132 .protocols("TLSv1.2")
133 .build());
134
135 List<SslContext> clientContexts = new ArrayList<SslContext>();
136 clientContexts.add(SslContextBuilder.forClient()
137 .sslProvider(SslProvider.JDK)
138 .trustManager(CERT_FILE)
139
140 .protocols("TLSv1.2")
141 .endpointIdentificationAlgorithm(null)
142 .build());
143
144 boolean hasOpenSsl = OpenSsl.isAvailable();
145 if (hasOpenSsl) {
146 serverContexts.add(SslContextBuilder.forServer(CERT_FILE, KEY_FILE)
147 .sslProvider(SslProvider.OPENSSL)
148
149 .protocols("TLSv1.2")
150 .build());
151 clientContexts.add(SslContextBuilder.forClient()
152 .sslProvider(SslProvider.OPENSSL)
153 .trustManager(CERT_FILE)
154
155 .protocols("TLSv1.2")
156 .endpointIdentificationAlgorithm(null)
157 .build());
158 } else {
159 logger.warn("OpenSSL is unavailable and thus will not be tested.", OpenSsl.unavailabilityCause());
160 }
161
162 List<Object[]> params = new ArrayList<Object[]>();
163 for (SslContext sc: serverContexts) {
164 for (SslContext cc: clientContexts) {
165 for (RenegotiationType rt: RenegotiationType.values()) {
166 if (rt != RenegotiationType.NONE &&
167 (sc instanceof OpenSslContext || cc instanceof OpenSslContext)) {
168
169 continue;
170 }
171
172 final Renegotiation r;
173 switch (rt) {
174 case NONE:
175 r = Renegotiation.NONE;
176 break;
177 case SERVER_INITIATED:
178 r = new Renegotiation(rt, sc.cipherSuites().get(sc.cipherSuites().size() - 1));
179 break;
180 case CLIENT_INITIATED:
181 r = new Renegotiation(rt, cc.cipherSuites().get(cc.cipherSuites().size() - 1));
182 break;
183 default:
184 throw new Error("Unexpected renegotiation type: " + rt);
185 }
186
187 for (int i = 0; i < 32; i++) {
188 params.add(new Object[] {
189 sc, cc, r,
190 (i & 16) != 0, (i & 8) != 0, (i & 4) != 0, (i & 2) != 0, (i & 1) != 0 });
191 }
192 }
193 }
194 }
195
196 return params;
197 }
198
199 private final AtomicReference<Throwable> clientException = new AtomicReference<Throwable>();
200 private final AtomicReference<Throwable> serverException = new AtomicReference<Throwable>();
201 private final AtomicInteger clientSendCounter = new AtomicInteger();
202 private final AtomicInteger clientRecvCounter = new AtomicInteger();
203 private final AtomicInteger serverRecvCounter = new AtomicInteger();
204
205 private final AtomicInteger clientNegoCounter = new AtomicInteger();
206 private final AtomicInteger serverNegoCounter = new AtomicInteger();
207
208 private volatile Channel clientChannel;
209 private volatile Channel serverChannel;
210
211 private volatile SslHandler clientSslHandler;
212 private volatile SslHandler serverSslHandler;
213
214 private final EchoClientHandler clientHandler =
215 new EchoClientHandler(clientRecvCounter, clientNegoCounter, clientException);
216
217 private final EchoServerHandler serverHandler =
218 new EchoServerHandler(serverRecvCounter, serverNegoCounter, serverException);
219
220 private SslContext serverCtx;
221 private SslContext clientCtx;
222 private Renegotiation renegotiation;
223 private boolean serverUsesDelegatedTaskExecutor;
224 private boolean clientUsesDelegatedTaskExecutor;
225 private boolean autoRead;
226 private boolean useChunkedWriteHandler;
227 private boolean useCompositeByteBuf;
228
229 @AfterAll
230 public static void compressHeapDumps() throws Exception {
231 TestUtils.compressHeapDumps();
232 }
233
234 @ParameterizedTest(name =
235 "{index}: serverEngine = {0}, clientEngine = {1}, renegotiation = {2}, " +
236 "serverUsesDelegatedTaskExecutor = {3}, clientUsesDelegatedTaskExecutor = {4}, " +
237 "autoRead = {5}, useChunkedWriteHandler = {6}, useCompositeByteBuf = {7}")
238 @MethodSource("data")
239 @Timeout(value = 30000, unit = TimeUnit.MILLISECONDS)
240 public void testSslEcho(
241 SslContext serverCtx, SslContext clientCtx, Renegotiation renegotiation,
242 boolean serverUsesDelegatedTaskExecutor, boolean clientUsesDelegatedTaskExecutor,
243 boolean autoRead, boolean useChunkedWriteHandler, boolean useCompositeByteBuf,
244 TestInfo testInfo) throws Throwable {
245 this.serverCtx = serverCtx;
246 this.clientCtx = clientCtx;
247 this.serverUsesDelegatedTaskExecutor = serverUsesDelegatedTaskExecutor;
248 this.clientUsesDelegatedTaskExecutor = clientUsesDelegatedTaskExecutor;
249 this.renegotiation = renegotiation;
250 this.autoRead = autoRead;
251 this.useChunkedWriteHandler = useChunkedWriteHandler;
252 this.useCompositeByteBuf = useCompositeByteBuf;
253 run(testInfo, new Runner<ServerBootstrap, Bootstrap>() {
254 @Override
255 public void run(ServerBootstrap serverBootstrap, Bootstrap bootstrap) throws Throwable {
256 testSslEcho(serverBootstrap, bootstrap);
257 }
258 });
259 }
260
261 public void testSslEcho(ServerBootstrap sb, Bootstrap cb) throws Throwable {
262 final ExecutorService delegatedTaskExecutor = Executors.newCachedThreadPool();
263 reset();
264
265 sb.childOption(ChannelOption.AUTO_READ, autoRead);
266 cb.option(ChannelOption.AUTO_READ, autoRead);
267
268 sb.childHandler(new ChannelInitializer<Channel>() {
269 @Override
270 public void initChannel(Channel sch) {
271 serverChannel = sch;
272
273 if (serverUsesDelegatedTaskExecutor) {
274 SSLEngine sse = serverCtx.newEngine(sch.alloc());
275 serverSslHandler = new SslHandler(sse, delegatedTaskExecutor);
276 } else {
277 serverSslHandler = serverCtx.newHandler(sch.alloc());
278 }
279 serverSslHandler.setHandshakeTimeoutMillis(0);
280
281 sch.pipeline().addLast("ssl", serverSslHandler);
282 if (useChunkedWriteHandler) {
283 sch.pipeline().addLast(new ChunkedWriteHandler());
284 }
285 sch.pipeline().addLast("serverHandler", serverHandler);
286 }
287 });
288
289 final CountDownLatch clientHandshakeEventLatch = new CountDownLatch(1);
290 cb.handler(new ChannelInitializer<Channel>() {
291 @Override
292 public void initChannel(Channel sch) {
293 clientChannel = sch;
294
295 if (clientUsesDelegatedTaskExecutor) {
296 SSLEngine cse = clientCtx.newEngine(sch.alloc());
297 clientSslHandler = new SslHandler(cse, delegatedTaskExecutor);
298 } else {
299 clientSslHandler = clientCtx.newHandler(sch.alloc());
300 }
301 clientSslHandler.setHandshakeTimeoutMillis(0);
302
303 sch.pipeline().addLast("ssl", clientSslHandler);
304 if (useChunkedWriteHandler) {
305 sch.pipeline().addLast(new ChunkedWriteHandler());
306 }
307 sch.pipeline().addLast("clientHandler", clientHandler);
308 sch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
309 @Override
310 public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
311 if (evt instanceof SslHandshakeCompletionEvent) {
312 clientHandshakeEventLatch.countDown();
313 }
314 ctx.fireUserEventTriggered(evt);
315 }
316 });
317 }
318 });
319
320 final Channel sc = sb.bind().sync().channel();
321 cb.connect(sc.localAddress()).sync();
322
323 final Future<Channel> clientHandshakeFuture = clientSslHandler.handshakeFuture();
324
325
326 clientHandshakeFuture.sync();
327 clientHandshakeEventLatch.await();
328
329 clientChannel.writeAndFlush(randomBufferType(clientChannel.alloc(), data, 0, FIRST_MESSAGE_SIZE));
330 clientSendCounter.set(FIRST_MESSAGE_SIZE);
331
332 boolean needsRenegotiation = renegotiation.type == RenegotiationType.CLIENT_INITIATED;
333 Future<Channel> renegoFuture = null;
334 SplittableRandom rng = new SplittableRandom(random.nextLong());
335 while (clientSendCounter.get() < data.length) {
336 int clientSendCounterVal = clientSendCounter.get();
337 int length = Math.min(rng.nextInt(1024 * 64), data.length - clientSendCounterVal);
338 ByteBuf buf = randomBufferType(clientChannel.alloc(), data, clientSendCounterVal, length);
339 if (useCompositeByteBuf) {
340 buf = Unpooled.compositeBuffer().addComponent(true, buf);
341 }
342
343 ChannelFuture future = clientChannel.writeAndFlush(buf);
344 clientSendCounter.set(clientSendCounterVal += length);
345 future.sync();
346
347 if (needsRenegotiation && clientSendCounterVal >= data.length / 2) {
348 needsRenegotiation = false;
349 clientSslHandler.engine().setEnabledCipherSuites(new String[] { renegotiation.cipherSuite });
350 renegoFuture = clientSslHandler.renegotiate();
351 logStats("CLIENT RENEGOTIATES");
352 assertNotSame(renegoFuture, clientHandshakeFuture);
353 }
354 }
355
356
357 while (clientRecvCounter.get() < data.length) {
358 if (serverException.get() != null) {
359 break;
360 }
361 if (clientException.get() != null) {
362 break;
363 }
364
365 Thread.sleep(50);
366 }
367
368 while (serverRecvCounter.get() < data.length) {
369 if (serverException.get() != null) {
370 break;
371 }
372 if (clientException.get() != null) {
373 break;
374 }
375
376 Thread.sleep(50);
377 }
378
379
380 if (renegoFuture != null) {
381 renegoFuture.sync();
382 }
383 if (serverHandler.renegoFuture != null) {
384 serverHandler.renegoFuture.sync();
385 }
386
387 serverChannel.close().awaitUninterruptibly();
388 clientChannel.close().awaitUninterruptibly();
389 sc.close().awaitUninterruptibly();
390 delegatedTaskExecutor.shutdown();
391 assertTrue(delegatedTaskExecutor.awaitTermination(5, TimeUnit.SECONDS));
392
393 if (serverException.get() != null && !(serverException.get() instanceof IOException)) {
394 throw serverException.get();
395 }
396 if (clientException.get() != null && !(clientException.get() instanceof IOException)) {
397 throw clientException.get();
398 }
399 if (serverException.get() != null) {
400 throw serverException.get();
401 }
402 if (clientException.get() != null) {
403 throw clientException.get();
404 }
405
406
407 try {
408 switch (renegotiation.type) {
409 case SERVER_INITIATED:
410 assertEquals(renegotiation.cipherSuite, serverSslHandler.engine().getSession().getCipherSuite());
411 assertEquals(2, serverNegoCounter.get());
412 assertThat(clientNegoCounter.get()).isIn(1, 2);
413 break;
414 case CLIENT_INITIATED:
415 assertThat(serverNegoCounter.get()).isIn(1, 2);
416 assertEquals(renegotiation.cipherSuite, clientSslHandler.engine().getSession().getCipherSuite());
417 assertEquals(2, clientNegoCounter.get());
418 break;
419 case NONE:
420 assertEquals(1, serverNegoCounter.get());
421 assertEquals(1, clientNegoCounter.get());
422 }
423 } finally {
424 logStats("STATS");
425 }
426 }
427
428 private void reset() {
429 clientException.set(null);
430 serverException.set(null);
431
432 clientSendCounter.set(0);
433 clientRecvCounter.set(0);
434 serverRecvCounter.set(0);
435
436 clientNegoCounter.set(0);
437 serverNegoCounter.set(0);
438
439 clientChannel = null;
440 serverChannel = null;
441
442 clientSslHandler = null;
443 serverSslHandler = null;
444 }
445
446 void logStats(String message) {
447 logger.debug(
448 "{}:\n" +
449 "\tclient { sent: {}, rcvd: {}, nego: {}, cipher: {} },\n" +
450 "\tserver { rcvd: {}, nego: {}, cipher: {} }",
451 message,
452 clientSendCounter, clientRecvCounter, clientNegoCounter,
453 clientSslHandler.engine().getSession().getCipherSuite(),
454 serverRecvCounter, serverNegoCounter,
455 serverSslHandler.engine().getSession().getCipherSuite());
456 }
457
458 @Sharable
459 private abstract class EchoHandler extends SimpleChannelInboundHandler<ByteBuf> {
460
461 protected final AtomicInteger recvCounter;
462 protected final AtomicInteger negoCounter;
463 protected final AtomicReference<Throwable> exception;
464
465 EchoHandler(
466 AtomicInteger recvCounter, AtomicInteger negoCounter,
467 AtomicReference<Throwable> exception) {
468
469 this.recvCounter = recvCounter;
470 this.negoCounter = negoCounter;
471 this.exception = exception;
472 }
473
474 @Override
475 public final void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
476
477
478 if (!autoRead) {
479 ctx.read();
480 }
481 ctx.fireChannelReadComplete();
482 }
483
484 @Override
485 public final void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
486 if (evt instanceof SslHandshakeCompletionEvent) {
487 SslHandshakeCompletionEvent handshakeEvt = (SslHandshakeCompletionEvent) evt;
488 if (handshakeEvt.cause() != null) {
489 logger.warn("Handshake failed:", handshakeEvt.cause());
490 }
491 assertSame(SslHandshakeCompletionEvent.SUCCESS, evt);
492 negoCounter.incrementAndGet();
493 logStats("HANDSHAKEN");
494 }
495 ctx.fireUserEventTriggered(evt);
496 }
497
498 @Override
499 public final void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
500 if (logger.isWarnEnabled()) {
501 logger.warn("Unexpected exception from the client side:", cause);
502 }
503
504 exception.compareAndSet(null, cause);
505 ctx.close();
506 }
507 }
508
509 private class EchoClientHandler extends EchoHandler {
510
511 EchoClientHandler(
512 AtomicInteger recvCounter, AtomicInteger negoCounter,
513 AtomicReference<Throwable> exception) {
514
515 super(recvCounter, negoCounter, exception);
516 }
517
518 @Override
519 public void handlerAdded(final ChannelHandlerContext ctx) {
520 if (!autoRead) {
521 ctx.pipeline().get(SslHandler.class).handshakeFuture().addListener(future -> ctx.read());
522 }
523 }
524
525 @Override
526 public void channelRead0(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
527 byte[] actual = new byte[in.readableBytes()];
528 in.readBytes(actual);
529
530 int lastIdx = recvCounter.get();
531 for (int i = 0; i < actual.length; i ++) {
532 assertEquals(data[i + lastIdx], actual[i]);
533 }
534
535 recvCounter.addAndGet(actual.length);
536 }
537 }
538
539 private class EchoServerHandler extends EchoHandler {
540 volatile Future<Channel> renegoFuture;
541
542 EchoServerHandler(
543 AtomicInteger recvCounter, AtomicInteger negoCounter,
544 AtomicReference<Throwable> exception) {
545
546 super(recvCounter, negoCounter, exception);
547 }
548
549 @Override
550 public final void channelRegistered(ChannelHandlerContext ctx) {
551 renegoFuture = null;
552 }
553
554 @Override
555 public void channelActive(final ChannelHandlerContext ctx) throws Exception {
556 if (!autoRead) {
557 ctx.read();
558 }
559 ctx.fireChannelActive();
560 }
561
562 @Override
563 public void channelRead0(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
564 byte[] actual = new byte[in.readableBytes()];
565 in.readBytes(actual);
566
567 int lastIdx = recvCounter.get();
568 for (int i = 0; i < actual.length; i ++) {
569 assertEquals(data[i + lastIdx], actual[i]);
570 }
571
572 ByteBuf buf = randomBufferType(ctx.alloc(), actual, 0, actual.length);
573 if (useCompositeByteBuf) {
574 buf = Unpooled.compositeBuffer().addComponent(true, buf);
575 }
576 ctx.writeAndFlush(buf);
577
578 recvCounter.addAndGet(actual.length);
579
580
581 if (renegotiation.type == RenegotiationType.SERVER_INITIATED &&
582 recvCounter.get() > data.length / 2 && renegoFuture == null) {
583
584 SslHandler sslHandler = ctx.pipeline().get(SslHandler.class);
585
586 Future<Channel> hf = sslHandler.handshakeFuture();
587 assertTrue(hf.isDone());
588
589 sslHandler.engine().setEnabledCipherSuites(new String[] { renegotiation.cipherSuite });
590 logStats("SERVER RENEGOTIATES");
591 renegoFuture = sslHandler.renegotiate();
592 assertNotSame(renegoFuture, hf);
593 assertSame(renegoFuture, sslHandler.handshakeFuture());
594 assertFalse(renegoFuture.isDone());
595 }
596 }
597 }
598 }