1 /*
2 * Copyright 2012 The Netty Project
3 *
4 * The Netty Project licenses this file to you under the Apache License,
5 * version 2.0 (the "License"); you may not use this file except in compliance
6 * with the License. You may obtain a copy of the License at:
7 *
8 * https://www.apache.org/licenses/LICENSE-2.0
9 *
10 * Unless required by applicable law or agreed to in writing, software
11 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13 * License for the specific language governing permissions and limitations
14 * under the License.
15 */
16 package io.netty.handler.ssl;
17
18 import io.netty.buffer.ByteBuf;
19 import io.netty.buffer.ByteBufAllocator;
20 import io.netty.buffer.ByteBufUtil;
21 import io.netty.buffer.CompositeByteBuf;
22 import io.netty.buffer.Unpooled;
23 import io.netty.channel.Channel;
24 import io.netty.channel.ChannelConfig;
25 import io.netty.channel.ChannelException;
26 import io.netty.channel.ChannelFuture;
27 import io.netty.channel.ChannelHandlerContext;
28 import io.netty.channel.ChannelInboundHandler;
29 import io.netty.channel.ChannelOption;
30 import io.netty.channel.ChannelOutboundBuffer;
31 import io.netty.channel.ChannelOutboundHandler;
32 import io.netty.channel.ChannelPipeline;
33 import io.netty.channel.ChannelPromise;
34 import io.netty.channel.StacklessClosedChannelException;
35 import io.netty.channel.unix.UnixChannel;
36 import io.netty.handler.codec.ByteToMessageDecoder;
37 import io.netty.handler.codec.DecoderException;
38 import io.netty.handler.codec.UnsupportedMessageTypeException;
39 import io.netty.util.ReferenceCountUtil;
40 import io.netty.util.concurrent.DefaultPromise;
41 import io.netty.util.concurrent.EventExecutor;
42 import io.netty.util.concurrent.Future;
43 import io.netty.util.concurrent.ImmediateExecutor;
44 import io.netty.util.concurrent.Promise;
45 import io.netty.util.concurrent.PromiseNotifier;
46 import io.netty.util.internal.ObjectUtil;
47 import io.netty.util.internal.PlatformDependent;
48 import io.netty.util.internal.ThrowableUtil;
49 import io.netty.util.internal.UnstableApi;
50 import io.netty.util.internal.logging.InternalLogger;
51 import io.netty.util.internal.logging.InternalLoggerFactory;
52
53 import java.io.IOException;
54 import java.net.SocketAddress;
55 import java.nio.ByteBuffer;
56 import java.nio.channels.ClosedChannelException;
57 import java.nio.channels.DatagramChannel;
58 import java.nio.channels.SocketChannel;
59 import java.security.cert.CertificateException;
60 import java.util.List;
61 import java.util.concurrent.Executor;
62 import java.util.concurrent.RejectedExecutionException;
63 import java.util.concurrent.TimeUnit;
64 import java.util.regex.Pattern;
65
66 import javax.net.ssl.SSLEngine;
67 import javax.net.ssl.SSLEngineResult;
68 import javax.net.ssl.SSLEngineResult.HandshakeStatus;
69 import javax.net.ssl.SSLEngineResult.Status;
70 import javax.net.ssl.SSLException;
71 import javax.net.ssl.SSLHandshakeException;
72 import javax.net.ssl.SSLSession;
73
74 import static io.netty.handler.ssl.SslUtils.NOT_ENOUGH_DATA;
75 import static io.netty.handler.ssl.SslUtils.getEncryptedPacketLength;
76 import static io.netty.util.internal.ObjectUtil.checkNotNull;
77 import static io.netty.util.internal.ObjectUtil.checkPositiveOrZero;
78
79 /**
80 * Adds <a href="https://en.wikipedia.org/wiki/Transport_Layer_Security">SSL
81 * · TLS</a> and StartTLS support to a {@link Channel}. Please refer
82 * to the <strong>"SecureChat"</strong> example in the distribution or the web
83 * site for the detailed usage.
84 *
85 * <h3>Beginning the handshake</h3>
86 * <p>
87 * Beside using the handshake {@link ChannelFuture} to get notified about the completion of the handshake it's
88 * also possible to detect it by implement the
89 * {@link ChannelInboundHandler#userEventTriggered(ChannelHandlerContext, Object)}
90 * method and check for a {@link SslHandshakeCompletionEvent}.
91 *
92 * <h3>Handshake</h3>
93 * <p>
94 * The handshake will be automatically issued for you once the {@link Channel} is active and
95 * {@link SSLEngine#getUseClientMode()} returns {@code true}.
96 * So no need to bother with it by your self.
97 *
98 * <h3>Closing the session</h3>
99 * <p>
100 * To close the SSL session, the {@link #closeOutbound()} method should be
101 * called to send the {@code close_notify} message to the remote peer. One
102 * exception is when you close the {@link Channel} - {@link SslHandler}
103 * intercepts the close request and send the {@code close_notify} message
104 * before the channel closure automatically. Once the SSL session is closed,
105 * it is not reusable, and consequently you should create a new
106 * {@link SslHandler} with a new {@link SSLEngine} as explained in the
107 * following section.
108 *
109 * <h3>Restarting the session</h3>
110 * <p>
111 * To restart the SSL session, you must remove the existing closed
112 * {@link SslHandler} from the {@link ChannelPipeline}, insert a new
113 * {@link SslHandler} with a new {@link SSLEngine} into the pipeline,
114 * and start the handshake process as described in the first section.
115 *
116 * <h3>Implementing StartTLS</h3>
117 * <p>
118 * <a href="https://en.wikipedia.org/wiki/STARTTLS">StartTLS</a> is the
119 * communication pattern that secures the wire in the middle of the plaintext
120 * connection. Please note that it is different from SSL · TLS, that
121 * secures the wire from the beginning of the connection. Typically, StartTLS
122 * is composed of three steps:
123 * <ol>
124 * <li>Client sends a StartTLS request to server.</li>
125 * <li>Server sends a StartTLS response to client.</li>
126 * <li>Client begins SSL handshake.</li>
127 * </ol>
128 * If you implement a server, you need to:
129 * <ol>
130 * <li>create a new {@link SslHandler} instance with {@code startTls} flag set
131 * to {@code true},</li>
132 * <li>insert the {@link SslHandler} to the {@link ChannelPipeline}, and</li>
133 * <li>write a StartTLS response.</li>
134 * </ol>
135 * Please note that you must insert {@link SslHandler} <em>before</em> sending
136 * the StartTLS response. Otherwise the client can send begin SSL handshake
137 * before {@link SslHandler} is inserted to the {@link ChannelPipeline}, causing
138 * data corruption.
139 * <p>
140 * The client-side implementation is much simpler.
141 * <ol>
142 * <li>Write a StartTLS request,</li>
143 * <li>wait for the StartTLS response,</li>
144 * <li>create a new {@link SslHandler} instance with {@code startTls} flag set
145 * to {@code false},</li>
146 * <li>insert the {@link SslHandler} to the {@link ChannelPipeline}, and</li>
147 * <li>Initiate SSL handshake.</li>
148 * </ol>
149 *
150 * <h3>Known issues</h3>
151 * <p>
152 * Because of a known issue with the current implementation of the SslEngine that comes
153 * with Java it may be possible that you see blocked IO-Threads while a full GC is done.
154 * <p>
155 * So if you are affected you can workaround this problem by adjust the cache settings
156 * like shown below:
157 *
158 * <pre>
159 * SslContext context = ...;
160 * context.getServerSessionContext().setSessionCacheSize(someSaneSize);
161 * context.getServerSessionContext().setSessionTime(someSameTimeout);
162 * </pre>
163 * <p>
164 * What values to use here depends on the nature of your application and should be set
165 * based on monitoring and debugging of it.
166 * For more details see
167 * <a href="https://github.com/netty/netty/issues/832">#832</a> in our issue tracker.
168 */
169 public class SslHandler extends ByteToMessageDecoder implements ChannelOutboundHandler {
170 private static final InternalLogger logger =
171 InternalLoggerFactory.getInstance(SslHandler.class);
172 private static final Pattern IGNORABLE_ERROR_MESSAGE = Pattern.compile(
173 "^.*(?:connection.*(?:reset|closed|abort|broken)|broken.*pipe).*$", Pattern.CASE_INSENSITIVE);
174 private static final int STATE_SENT_FIRST_MESSAGE = 1;
175 private static final int STATE_FLUSHED_BEFORE_HANDSHAKE = 1 << 1;
176 private static final int STATE_READ_DURING_HANDSHAKE = 1 << 2;
177 private static final int STATE_HANDSHAKE_STARTED = 1 << 3;
178 /**
179 * Set by wrap*() methods when something is produced.
180 * {@link #channelReadComplete(ChannelHandlerContext)} will check this flag, clear it, and call ctx.flush().
181 */
182 private static final int STATE_NEEDS_FLUSH = 1 << 4;
183 private static final int STATE_OUTBOUND_CLOSED = 1 << 5;
184 private static final int STATE_CLOSE_NOTIFY = 1 << 6;
185 private static final int STATE_PROCESS_TASK = 1 << 7;
186 /**
187 * This flag is used to determine if we need to call {@link ChannelHandlerContext#read()} to consume more data
188 * when {@link ChannelConfig#isAutoRead()} is {@code false}.
189 */
190 private static final int STATE_FIRE_CHANNEL_READ = 1 << 8;
191 private static final int STATE_UNWRAP_REENTRY = 1 << 9;
192
193 /**
194 * <a href="https://tools.ietf.org/html/rfc5246#section-6.2">2^14</a> which is the maximum sized plaintext chunk
195 * allowed by the TLS RFC.
196 */
197 private static final int MAX_PLAINTEXT_LENGTH = 16 * 1024;
198
199 /**
200 * The JDK {@link SSLEngine} wants at least {@code packetLength - 5 - 16} bytes to unwrap a record, as it subtracts
201 * the 5 byte header and at least 16 bytes of tag or MAC for every cipher suite.
202 * <p>
203 * The {@code BUFFER_OVERFLOW} check calls {@code estimateFragmentSize(packetLen)}:
204 * https://github.com/openjdk/jdk/blob/6917797/src/java.base/share/classes/sun/security/ssl/SSLEngineImpl.java#L672
205 * which for AEAD ciphers subtracts the header and tag size from the packet length:
206 * https://github.com/openjdk/jdk/blob/6917797/src/java.base/share/classes/sun/security/ssl/SSLCipher.java#L1969
207 */
208 private static final int JDK_MIN_UNWRAP_OVERHEAD = 5 + 16;
209
210 private enum SslEngineType {
211 TCNATIVE(true, COMPOSITE_CUMULATOR) {
212 @Override
213 SSLEngineResult unwrap(SslHandler handler, ByteBuf in, int len, ByteBuf out) throws SSLException {
214 int nioBufferCount = in.nioBufferCount();
215 int writerIndex = out.writerIndex();
216 final SSLEngineResult result;
217 if (nioBufferCount > 1) {
218 /*
219 * If {@link OpenSslEngine} is in use,
220 * we can use a special {@link OpenSslEngine#unwrap(ByteBuffer[], ByteBuffer[])} method
221 * that accepts multiple {@link ByteBuffer}s without additional memory copies.
222 */
223 ReferenceCountedOpenSslEngine opensslEngine = (ReferenceCountedOpenSslEngine) handler.engine;
224 try {
225 handler.singleBuffer[0] = toByteBuffer(out, writerIndex, out.writableBytes());
226 result = opensslEngine.unwrap(in.nioBuffers(in.readerIndex(), len), handler.singleBuffer);
227 } finally {
228 handler.singleBuffer[0] = null;
229 }
230 } else {
231 result = handler.engine.unwrap(toByteBuffer(in, in.readerIndex(), len),
232 toByteBuffer(out, writerIndex, out.writableBytes()));
233 }
234 out.writerIndex(writerIndex + result.bytesProduced());
235 return result;
236 }
237
238 @Override
239 ByteBuf allocateWrapBuffer(SslHandler handler, ByteBufAllocator allocator,
240 int pendingBytes, int numComponents) {
241 return allocator.directBuffer(((ReferenceCountedOpenSslEngine) handler.engine)
242 .calculateOutNetBufSize(pendingBytes, numComponents));
243 }
244
245 @Override
246 int calculateRequiredOutBufSpace(SslHandler handler, int pendingBytes, int numComponents) {
247 return ((ReferenceCountedOpenSslEngine) handler.engine)
248 .calculateMaxLengthForWrap(pendingBytes, numComponents);
249 }
250
251 @Override
252 int calculatePendingData(SslHandler handler, int guess) {
253 int sslPending = ((ReferenceCountedOpenSslEngine) handler.engine).sslPending();
254 return sslPending > 0 ? sslPending : guess;
255 }
256
257 @Override
258 int calculateInitialUnwrapBufSize(SslHandler handler, int length) {
259 // The plaintext of a record is never longer than MAX_PLAINTEXT_LENGTH.
260 return Math.min(length, MAX_PLAINTEXT_LENGTH);
261 }
262
263 @Override
264 boolean jdkCompatibilityMode(SSLEngine engine) {
265 return ((ReferenceCountedOpenSslEngine) engine).jdkCompatibilityMode;
266 }
267 },
268 CONSCRYPT(true, COMPOSITE_CUMULATOR) {
269 @Override
270 SSLEngineResult unwrap(SslHandler handler, ByteBuf in, int len, ByteBuf out) throws SSLException {
271 int nioBufferCount = in.nioBufferCount();
272 int writerIndex = out.writerIndex();
273 final SSLEngineResult result;
274 if (nioBufferCount > 1) {
275 /*
276 * Use a special unwrap method without additional memory copies.
277 */
278 try {
279 handler.singleBuffer[0] = toByteBuffer(out, writerIndex, out.writableBytes());
280 result = ((ConscryptAlpnSslEngine) handler.engine).unwrap(
281 in.nioBuffers(in.readerIndex(), len),
282 handler.singleBuffer);
283 } finally {
284 handler.singleBuffer[0] = null;
285 }
286 } else {
287 result = handler.engine.unwrap(toByteBuffer(in, in.readerIndex(), len),
288 toByteBuffer(out, writerIndex, out.writableBytes()));
289 }
290 out.writerIndex(writerIndex + result.bytesProduced());
291 return result;
292 }
293
294 @Override
295 ByteBuf allocateWrapBuffer(SslHandler handler, ByteBufAllocator allocator,
296 int pendingBytes, int numComponents) {
297 return allocator.directBuffer(
298 ((ConscryptAlpnSslEngine) handler.engine).calculateOutNetBufSize(pendingBytes, numComponents));
299 }
300
301 @Override
302 int calculateRequiredOutBufSpace(SslHandler handler, int pendingBytes, int numComponents) {
303 return ((ConscryptAlpnSslEngine) handler.engine)
304 .calculateRequiredOutBufSpace(pendingBytes, numComponents);
305 }
306
307 @Override
308 int calculatePendingData(SslHandler handler, int guess) {
309 return guess;
310 }
311
312 @Override
313 int calculateInitialUnwrapBufSize(SslHandler handler, int length) {
314 // Conscrypt does not need more space than the plaintext, which is at most MAX_PLAINTEXT_LENGTH.
315 return Math.min(length, MAX_PLAINTEXT_LENGTH);
316 }
317
318 @Override
319 boolean jdkCompatibilityMode(SSLEngine engine) {
320 return true;
321 }
322 },
323 JDK(false, MERGE_CUMULATOR) {
324 @Override
325 SSLEngineResult unwrap(SslHandler handler, ByteBuf in, int len, ByteBuf out) throws SSLException {
326 int writerIndex = out.writerIndex();
327 ByteBuffer inNioBuffer = getUnwrapInputBuffer(handler, toByteBuffer(in, in.readerIndex(), len));
328 int position = inNioBuffer.position();
329 final SSLEngineResult result = handler.engine.unwrap(inNioBuffer,
330 toByteBuffer(out, writerIndex, out.writableBytes()));
331 out.writerIndex(writerIndex + result.bytesProduced());
332
333 // This is a workaround for a bug in Android 5.0. Android 5.0 does not correctly update the
334 // SSLEngineResult.bytesConsumed() in some cases and just return 0.
335 //
336 // See:
337 // - https://android-review.googlesource.com/c/platform/external/conscrypt/+/122080
338 // - https://github.com/netty/netty/issues/7758
339 if (result.bytesConsumed() == 0) {
340 int consumed = inNioBuffer.position() - position;
341 if (consumed != result.bytesConsumed()) {
342 // Create a new SSLEngineResult with the correct bytesConsumed().
343 return new SSLEngineResult(
344 result.getStatus(), result.getHandshakeStatus(), consumed, result.bytesProduced());
345 }
346 }
347 return result;
348 }
349
350 private ByteBuffer getUnwrapInputBuffer(SslHandler handler, ByteBuffer inNioBuffer) {
351 int javaVersion = PlatformDependent.javaVersion();
352 if (javaVersion >= 22 && javaVersion < 25 && inNioBuffer.isDirect()) {
353 // Work-around for https://bugs.openjdk.org/browse/JDK-8357268
354 int remaining = inNioBuffer.remaining();
355 ByteBuffer copy = handler.unwrapInputCopy;
356 if (copy == null || copy.capacity() < remaining) {
357 handler.unwrapInputCopy = copy = ByteBuffer.allocate(remaining);
358 } else {
359 copy.clear();
360 }
361 copy.put(inNioBuffer).flip();
362 return copy;
363 }
364 return inNioBuffer;
365 }
366
367 @Override
368 ByteBuf allocateWrapBuffer(SslHandler handler, ByteBufAllocator allocator,
369 int pendingBytes, int numComponents) {
370 // For JDK we don't have a good source for the max wrap overhead. We need at least one packet buffer
371 // size, but may be able to fit more in based on the total requested.
372 return allocator.heapBuffer(Math.max(pendingBytes, handler.engine.getSession().getPacketBufferSize()));
373 }
374
375 @Override
376 int calculateRequiredOutBufSpace(SslHandler handler, int pendingBytes, int numComponents) {
377 // As for the JDK SSLEngine we always need to operate on buffer space required by the SSLEngine
378 // (normally ~16KB). This is required even if the amount of data to encrypt is very small. Use heap
379 // buffers to reduce the native memory usage.
380 //
381 // Beside this the JDK SSLEngine also (as of today) will do an extra heap to direct buffer copy
382 // if a direct buffer is used as its internals operate on byte[].
383 return handler.engine.getSession().getPacketBufferSize();
384 }
385
386 @Override
387 int calculatePendingData(SslHandler handler, int guess) {
388 return guess;
389 }
390
391 @Override
392 int calculateInitialUnwrapBufSize(SslHandler handler, int length) {
393 // length is a single record, which may need a bit more space than MAX_PLAINTEXT_LENGTH.
394 return length <= MAX_PLAINTEXT_LENGTH ? length : Math.max(MAX_PLAINTEXT_LENGTH,
395 length - JDK_MIN_UNWRAP_OVERHEAD);
396 }
397
398 @Override
399 boolean jdkCompatibilityMode(SSLEngine engine) {
400 return true;
401 }
402 };
403
404 static SslEngineType forEngine(SSLEngine engine) {
405 return engine instanceof ReferenceCountedOpenSslEngine ? TCNATIVE :
406 engine instanceof ConscryptAlpnSslEngine ? CONSCRYPT : JDK;
407 }
408
409 SslEngineType(boolean wantsDirectBuffer, Cumulator cumulator) {
410 this.wantsDirectBuffer = wantsDirectBuffer;
411 this.cumulator = cumulator;
412 }
413
414 abstract SSLEngineResult unwrap(SslHandler handler, ByteBuf in, int len, ByteBuf out) throws SSLException;
415
416 abstract int calculatePendingData(SslHandler handler, int guess);
417
418 abstract int calculateInitialUnwrapBufSize(SslHandler handler, int length);
419
420 abstract boolean jdkCompatibilityMode(SSLEngine engine);
421
422 abstract ByteBuf allocateWrapBuffer(SslHandler handler, ByteBufAllocator allocator,
423 int pendingBytes, int numComponents);
424
425 abstract int calculateRequiredOutBufSpace(SslHandler handler, int pendingBytes, int numComponents);
426
427 // BEGIN Platform-dependent flags
428
429 /**
430 * {@code true} if and only if {@link SSLEngine} expects a direct buffer and so if a heap buffer
431 * is given will make an extra memory copy.
432 */
433 final boolean wantsDirectBuffer;
434
435 // END Platform-dependent flags
436
437 /**
438 * When using JDK {@link SSLEngine}, we use {@link #MERGE_CUMULATOR} because it works only with
439 * one {@link ByteBuffer}.
440 *
441 * When using {@link OpenSslEngine}, we can use {@link #COMPOSITE_CUMULATOR} because it has
442 * {@link OpenSslEngine#unwrap(ByteBuffer[], ByteBuffer[])} which works with multiple {@link ByteBuffer}s
443 * and which does not need to do extra memory copies.
444 */
445 final Cumulator cumulator;
446 }
447
448 private volatile ChannelHandlerContext ctx;
449 private final SSLEngine engine;
450 private final SslEngineType engineType;
451 private final Executor delegatedTaskExecutor;
452 private final boolean jdkCompatibilityMode;
453
454 /**
455 * Used if {@link SSLEngine#wrap(ByteBuffer[], ByteBuffer)} and {@link SSLEngine#unwrap(ByteBuffer, ByteBuffer[])}
456 * should be called with a {@link ByteBuf} that is only backed by one {@link ByteBuffer} to reduce the object
457 * creation.
458 */
459 private final ByteBuffer[] singleBuffer = new ByteBuffer[1];
460 private ByteBuffer unwrapInputCopy;
461
462 private final boolean startTls;
463 private final ResumptionController resumptionController;
464
465 private final SslTasksRunner sslTaskRunnerForUnwrap = new SslTasksRunner(true);
466 private final SslTasksRunner sslTaskRunner = new SslTasksRunner(false);
467
468 private SslHandlerCoalescingBufferQueue pendingUnencryptedWrites;
469 private Promise<Channel> handshakePromise = new LazyChannelPromise();
470 private final LazyChannelPromise sslClosePromise = new LazyChannelPromise();
471
472 private int packetLength;
473 private short state;
474
475 private volatile long handshakeTimeoutMillis = 10000;
476 private volatile long closeNotifyFlushTimeoutMillis = 3000;
477 private volatile long closeNotifyReadTimeoutMillis;
478 volatile int wrapDataSize = MAX_PLAINTEXT_LENGTH;
479
480 /**
481 * Creates a new instance which runs all delegated tasks directly on the {@link EventExecutor}.
482 *
483 * @param engine the {@link SSLEngine} this handler will use
484 */
485 public SslHandler(SSLEngine engine) {
486 this(engine, false);
487 }
488
489 /**
490 * Creates a new instance which runs all delegated tasks directly on the {@link EventExecutor}.
491 *
492 * @param engine the {@link SSLEngine} this handler will use
493 * @param startTls {@code true} if the first write request shouldn't be
494 * encrypted by the {@link SSLEngine}
495 */
496 public SslHandler(SSLEngine engine, boolean startTls) {
497 this(engine, startTls, ImmediateExecutor.INSTANCE);
498 }
499
500 /**
501 * Creates a new instance.
502 *
503 * @param engine the {@link SSLEngine} this handler will use
504 * @param delegatedTaskExecutor the {@link Executor} that will be used to execute tasks that are returned by
505 * {@link SSLEngine#getDelegatedTask()}.
506 */
507 public SslHandler(SSLEngine engine, Executor delegatedTaskExecutor) {
508 this(engine, false, delegatedTaskExecutor);
509 }
510
511 /**
512 * Creates a new instance.
513 *
514 * @param engine the {@link SSLEngine} this handler will use
515 * @param startTls {@code true} if the first write request shouldn't be
516 * encrypted by the {@link SSLEngine}
517 * @param delegatedTaskExecutor the {@link Executor} that will be used to execute tasks that are returned by
518 * {@link SSLEngine#getDelegatedTask()}.
519 */
520 public SslHandler(SSLEngine engine, boolean startTls, Executor delegatedTaskExecutor) {
521 this(engine, startTls, delegatedTaskExecutor, null);
522 }
523
524 SslHandler(SSLEngine engine, boolean startTls, Executor delegatedTaskExecutor,
525 ResumptionController resumptionController) {
526 this.engine = ObjectUtil.checkNotNull(engine, "engine");
527 this.delegatedTaskExecutor = ObjectUtil.checkNotNull(delegatedTaskExecutor, "delegatedTaskExecutor");
528 engineType = SslEngineType.forEngine(engine);
529 this.startTls = startTls;
530 this.jdkCompatibilityMode = engineType.jdkCompatibilityMode(engine);
531 setCumulator(engineType.cumulator);
532 this.resumptionController = resumptionController;
533 }
534
535 public long getHandshakeTimeoutMillis() {
536 return handshakeTimeoutMillis;
537 }
538
539 public void setHandshakeTimeout(long handshakeTimeout, TimeUnit unit) {
540 checkNotNull(unit, "unit");
541 setHandshakeTimeoutMillis(unit.toMillis(handshakeTimeout));
542 }
543
544 public void setHandshakeTimeoutMillis(long handshakeTimeoutMillis) {
545 this.handshakeTimeoutMillis = checkPositiveOrZero(handshakeTimeoutMillis, "handshakeTimeoutMillis");
546 }
547
548 /**
549 * Sets the number of bytes to pass to each {@link SSLEngine#wrap(ByteBuffer[], int, int, ByteBuffer)} call.
550 * <p>
551 * This value will partition data which is passed to write
552 * {@link #write(ChannelHandlerContext, Object, ChannelPromise)}. The partitioning will work as follows:
553 * <ul>
554 * <li>If {@code wrapDataSize <= 0} then we will write each data chunk as is.</li>
555 * <li>If {@code wrapDataSize > data size} then we will attempt to aggregate multiple data chunks together.</li>
556 * <li>If {@code wrapDataSize > data size} Else if {@code wrapDataSize <= data size} then we will divide the data
557 * into chunks of {@code wrapDataSize} when writing.</li>
558 * </ul>
559 * <p>
560 * If the {@link SSLEngine} doesn't support a gather wrap operation (e.g. {@link SslProvider#OPENSSL}) then
561 * aggregating data before wrapping can help reduce the ratio between TLS overhead vs data payload which will lead
562 * to better goodput. Writing fixed chunks of data can also help target the underlying transport's (e.g. TCP)
563 * frame size. Under lossy/congested network conditions this may help the peer get full TLS packets earlier and
564 * be able to do work sooner, as opposed to waiting for the all the pieces of the TLS packet to arrive.
565 * @param wrapDataSize the number of bytes which will be passed to each
566 * {@link SSLEngine#wrap(ByteBuffer[], int, int, ByteBuffer)} call.
567 */
568 @UnstableApi
569 public final void setWrapDataSize(int wrapDataSize) {
570 this.wrapDataSize = wrapDataSize;
571 }
572
573 /**
574 * @deprecated use {@link #getCloseNotifyFlushTimeoutMillis()}
575 */
576 @Deprecated
577 public long getCloseNotifyTimeoutMillis() {
578 return getCloseNotifyFlushTimeoutMillis();
579 }
580
581 /**
582 * @deprecated use {@link #setCloseNotifyFlushTimeout(long, TimeUnit)}
583 */
584 @Deprecated
585 public void setCloseNotifyTimeout(long closeNotifyTimeout, TimeUnit unit) {
586 setCloseNotifyFlushTimeout(closeNotifyTimeout, unit);
587 }
588
589 /**
590 * @deprecated use {@link #setCloseNotifyFlushTimeoutMillis(long)}
591 */
592 @Deprecated
593 public void setCloseNotifyTimeoutMillis(long closeNotifyFlushTimeoutMillis) {
594 setCloseNotifyFlushTimeoutMillis(closeNotifyFlushTimeoutMillis);
595 }
596
597 /**
598 * Gets the timeout for flushing the close_notify that was triggered by closing the
599 * {@link Channel}. If the close_notify was not flushed in the given timeout the {@link Channel} will be closed
600 * forcibly.
601 */
602 public final long getCloseNotifyFlushTimeoutMillis() {
603 return closeNotifyFlushTimeoutMillis;
604 }
605
606 /**
607 * Sets the timeout for flushing the close_notify that was triggered by closing the
608 * {@link Channel}. If the close_notify was not flushed in the given timeout the {@link Channel} will be closed
609 * forcibly.
610 */
611 public final void setCloseNotifyFlushTimeout(long closeNotifyFlushTimeout, TimeUnit unit) {
612 setCloseNotifyFlushTimeoutMillis(unit.toMillis(closeNotifyFlushTimeout));
613 }
614
615 /**
616 * See {@link #setCloseNotifyFlushTimeout(long, TimeUnit)}.
617 */
618 public final void setCloseNotifyFlushTimeoutMillis(long closeNotifyFlushTimeoutMillis) {
619 this.closeNotifyFlushTimeoutMillis = checkPositiveOrZero(closeNotifyFlushTimeoutMillis,
620 "closeNotifyFlushTimeoutMillis");
621 }
622
623 /**
624 * Gets the timeout (in ms) for receiving the response for the close_notify that was triggered by closing the
625 * {@link Channel}. This timeout starts after the close_notify message was successfully written to the
626 * remote peer. Use {@code 0} to directly close the {@link Channel} and not wait for the response.
627 */
628 public final long getCloseNotifyReadTimeoutMillis() {
629 return closeNotifyReadTimeoutMillis;
630 }
631
632 /**
633 * Sets the timeout for receiving the response for the close_notify that was triggered by closing the
634 * {@link Channel}. This timeout starts after the close_notify message was successfully written to the
635 * remote peer. Use {@code 0} to directly close the {@link Channel} and not wait for the response.
636 */
637 public final void setCloseNotifyReadTimeout(long closeNotifyReadTimeout, TimeUnit unit) {
638 setCloseNotifyReadTimeoutMillis(unit.toMillis(closeNotifyReadTimeout));
639 }
640
641 /**
642 * See {@link #setCloseNotifyReadTimeout(long, TimeUnit)}.
643 */
644 public final void setCloseNotifyReadTimeoutMillis(long closeNotifyReadTimeoutMillis) {
645 this.closeNotifyReadTimeoutMillis = checkPositiveOrZero(closeNotifyReadTimeoutMillis,
646 "closeNotifyReadTimeoutMillis");
647 }
648
649 /**
650 * Returns the {@link SSLEngine} which is used by this handler.
651 */
652 public SSLEngine engine() {
653 return engine;
654 }
655
656 /**
657 * Returns the name of the current application-level protocol.
658 *
659 * @return the protocol name or {@code null} if application-level protocol has not been negotiated
660 */
661 public String applicationProtocol() {
662 SSLEngine engine = engine();
663 if (!(engine instanceof ApplicationProtocolAccessor)) {
664 return null;
665 }
666
667 return ((ApplicationProtocolAccessor) engine).getNegotiatedApplicationProtocol();
668 }
669
670 /**
671 * Returns a {@link Future} that will get notified once the current TLS handshake completes.
672 *
673 * @return the {@link Future} for the initial TLS handshake if {@link #renegotiate()} was not invoked.
674 * The {@link Future} for the most recent {@linkplain #renegotiate() TLS renegotiation} otherwise.
675 */
676 public Future<Channel> handshakeFuture() {
677 return handshakePromise;
678 }
679
680 /**
681 * Use {@link #closeOutbound()}
682 */
683 @Deprecated
684 public ChannelFuture close() {
685 return closeOutbound();
686 }
687
688 /**
689 * Use {@link #closeOutbound(ChannelPromise)}
690 */
691 @Deprecated
692 public ChannelFuture close(ChannelPromise promise) {
693 return closeOutbound(promise);
694 }
695
696 /**
697 * Sends an SSL {@code close_notify} message to the specified channel and
698 * destroys the underlying {@link SSLEngine}. This will <strong>not</strong> close the underlying
699 * {@link Channel}. If you want to also close the {@link Channel} use {@link Channel#close()} or
700 * {@link ChannelHandlerContext#close()}
701 */
702 public ChannelFuture closeOutbound() {
703 return closeOutbound(ctx.newPromise());
704 }
705
706 /**
707 * Sends an SSL {@code close_notify} message to the specified channel and
708 * destroys the underlying {@link SSLEngine}. This will <strong>not</strong> close the underlying
709 * {@link Channel}. If you want to also close the {@link Channel} use {@link Channel#close()} or
710 * {@link ChannelHandlerContext#close()}
711 */
712 public ChannelFuture closeOutbound(final ChannelPromise promise) {
713 final ChannelHandlerContext ctx = this.ctx;
714 if (ctx.executor().inEventLoop()) {
715 closeOutbound0(promise);
716 } else {
717 ctx.executor().execute(() -> closeOutbound0(promise));
718 }
719 return promise;
720 }
721
722 private void closeOutbound0(ChannelPromise promise) {
723 setState(STATE_OUTBOUND_CLOSED);
724 engine.closeOutbound();
725 try {
726 flush(ctx, promise);
727 } catch (Exception e) {
728 if (!promise.tryFailure(e)) {
729 logger.warn("{} flush() raised a masked exception.", ctx.channel(), e);
730 }
731 }
732 }
733
734 /**
735 * Return the {@link Future} that will get notified if the inbound of the {@link SSLEngine} is closed.
736 *
737 * This method will return the same {@link Future} all the time.
738 *
739 * @see SSLEngine
740 */
741 public Future<Channel> sslCloseFuture() {
742 return sslClosePromise;
743 }
744
745 @Override
746 public void handlerRemoved0(ChannelHandlerContext ctx) throws Exception {
747 try {
748 if (pendingUnencryptedWrites != null && !pendingUnencryptedWrites.isEmpty()) {
749 // Check if queue is not empty first because create a new ChannelException is expensive
750 pendingUnencryptedWrites.releaseAndFailAll(ctx,
751 new ChannelException("Pending write on removal of SslHandler"));
752 }
753 pendingUnencryptedWrites = null;
754
755 SSLException cause = null;
756
757 // If the handshake or SSLEngine closure is not done yet we should fail corresponding promise and
758 // notify the rest of the
759 // pipeline.
760 if (!handshakePromise.isDone()) {
761 cause = new SSLHandshakeException("SslHandler removed before handshake completed");
762 if (handshakePromise.tryFailure(cause)) {
763 ctx.fireUserEventTriggered(new SslHandshakeCompletionEvent(cause));
764 }
765 }
766 if (!sslClosePromise.isDone()) {
767 if (cause == null) {
768 cause = new SSLException("SslHandler removed before SSLEngine was closed");
769 }
770 notifyClosePromise(cause);
771 }
772 } finally {
773 ReferenceCountUtil.release(engine);
774 }
775 }
776
777 @Override
778 public void bind(ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) throws Exception {
779 ctx.bind(localAddress, promise);
780 }
781
782 @Override
783 public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, SocketAddress localAddress,
784 ChannelPromise promise) throws Exception {
785 ctx.connect(remoteAddress, localAddress, promise);
786 }
787
788 @Override
789 public void deregister(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
790 ctx.deregister(promise);
791 }
792
793 @Override
794 public void disconnect(final ChannelHandlerContext ctx,
795 final ChannelPromise promise) throws Exception {
796 closeOutboundAndChannel(ctx, promise, true);
797 }
798
799 @Override
800 public void close(final ChannelHandlerContext ctx,
801 final ChannelPromise promise) throws Exception {
802 closeOutboundAndChannel(ctx, promise, false);
803 }
804
805 @Override
806 public void read(ChannelHandlerContext ctx) throws Exception {
807 if (!handshakePromise.isDone()) {
808 setState(STATE_READ_DURING_HANDSHAKE);
809 }
810
811 ctx.read();
812 }
813
814 private static IllegalStateException newPendingWritesNullException() {
815 return new IllegalStateException("pendingUnencryptedWrites is null, handlerRemoved0 called?");
816 }
817
818 @Override
819 public void write(final ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
820 if (!(msg instanceof ByteBuf)) {
821 UnsupportedMessageTypeException exception = new UnsupportedMessageTypeException(msg, ByteBuf.class);
822 ReferenceCountUtil.safeRelease(msg);
823 promise.setFailure(exception);
824 } else if (pendingUnencryptedWrites == null) {
825 ReferenceCountUtil.safeRelease(msg);
826 promise.setFailure(newPendingWritesNullException());
827 } else {
828 pendingUnencryptedWrites.add((ByteBuf) msg, promise);
829 }
830 }
831
832 @Override
833 public void flush(ChannelHandlerContext ctx) throws Exception {
834 // Do not encrypt the first write request if this handler is
835 // created with startTLS flag turned on.
836 if (startTls && !isStateSet(STATE_SENT_FIRST_MESSAGE)) {
837 setState(STATE_SENT_FIRST_MESSAGE);
838 pendingUnencryptedWrites.writeAndRemoveAll(ctx);
839 forceFlush(ctx);
840 // Explicit start handshake processing once we send the first message. This will also ensure
841 // we will schedule the timeout if needed.
842 startHandshakeProcessing(true);
843 return;
844 }
845
846 if (isStateSet(STATE_PROCESS_TASK)) {
847 return;
848 }
849
850 try {
851 wrapAndFlush(ctx);
852 } catch (Throwable cause) {
853 setHandshakeFailure(ctx, cause);
854 PlatformDependent.throwException(cause);
855 }
856 }
857
858 private void wrapAndFlush(ChannelHandlerContext ctx) throws SSLException {
859 if (pendingUnencryptedWrites.isEmpty()) {
860 // It's important to NOT use a voidPromise here as the user
861 // may want to add a ChannelFutureListener to the ChannelPromise later.
862 //
863 // See https://github.com/netty/netty/issues/3364
864 pendingUnencryptedWrites.add(Unpooled.EMPTY_BUFFER, ctx.newPromise());
865 }
866 if (!handshakePromise.isDone()) {
867 setState(STATE_FLUSHED_BEFORE_HANDSHAKE);
868 }
869 try {
870 wrap(ctx, false);
871 } finally {
872 // We may have written some parts of data before an exception was thrown so ensure we always flush.
873 // See https://github.com/netty/netty/issues/3900#issuecomment-172481830
874 forceFlush(ctx);
875 }
876 }
877
878 // This method will not call setHandshakeFailure(...) !
879 private void wrap(ChannelHandlerContext ctx, boolean inUnwrap) throws SSLException {
880 ByteBuf out = null;
881 ByteBufAllocator alloc = ctx.alloc();
882 try {
883 final int wrapDataSize = this.wrapDataSize;
884 // Only continue to loop if the handler was not removed in the meantime.
885 // See https://github.com/netty/netty/issues/5860
886 outer: while (!ctx.isRemoved() && !pendingUnencryptedWrites.isEmpty()) {
887 ChannelPromise promise = ctx.newPromise();
888 ByteBuf buf = wrapDataSize > 0 ?
889 pendingUnencryptedWrites.remove(alloc, wrapDataSize, promise) :
890 pendingUnencryptedWrites.removeFirst(promise);
891 if (buf == null) {
892 break;
893 }
894
895 SSLEngineResult result;
896
897 try {
898 if (buf.readableBytes() > MAX_PLAINTEXT_LENGTH) {
899 // If we pulled a buffer larger than the supported packet size, we can slice it up and
900 // iteratively, encrypting multiple packets into a single larger buffer. This substantially
901 // saves on allocations for large responses. Here we estimate how large of a buffer we need.
902 // If we overestimate a bit, that's fine. If we underestimate, we'll simply re-enqueue the
903 // remaining buffer and get it on the next outer loop.
904 int readableBytes = buf.readableBytes();
905 int numPackets = readableBytes / MAX_PLAINTEXT_LENGTH;
906 if (readableBytes % MAX_PLAINTEXT_LENGTH != 0) {
907 numPackets += 1;
908 }
909
910 if (out == null) {
911 out = allocateOutNetBuf(ctx, readableBytes, buf.nioBufferCount() + numPackets);
912 }
913 result = wrapMultiple(alloc, engine, buf, out);
914 } else {
915 if (out == null) {
916 out = allocateOutNetBuf(ctx, buf.readableBytes(), buf.nioBufferCount());
917 }
918 result = wrap(alloc, engine, buf, out);
919 }
920 } catch (Throwable e) {
921 // Either wrapMultiple(...), wrap(...) or allocateOutNetBuf(...) did throw.
922 // In this case we need to release the buffer that we removed from pendingUnencryptedWrites
923 // before failing the promise and rethrowing it. Failing to do so would result in a buffer leak.
924 // See https://github.com/netty/netty/issues/14644
925 //
926 // We don't need to release out here as this is done in a finally block already.
927 buf.release();
928 promise.setFailure(e);
929 throw e;
930 }
931
932 if (buf.isReadable()) {
933 pendingUnencryptedWrites.addFirst(buf, promise);
934 // When we add the buffer/promise pair back we need to be sure we don't complete the promise
935 // later. We only complete the promise if the buffer is completely consumed.
936 promise = null;
937 } else {
938 buf.release();
939 }
940
941 // We need to write any data before we invoke any methods which may trigger re-entry, otherwise
942 // writes may occur out of order and TLS sequencing may be off (e.g. SSLV3_ALERT_BAD_RECORD_MAC).
943 if (out.isReadable()) {
944 final ByteBuf b = out;
945 out = null;
946 if (promise != null) {
947 ctx.write(b, promise);
948 } else {
949 ctx.write(b);
950 }
951 } else if (promise != null) {
952 ctx.write(Unpooled.EMPTY_BUFFER, promise);
953 }
954 // else out is not readable we can re-use it and so save an extra allocation
955
956 if (result.getStatus() == Status.CLOSED) {
957 // First check if there is any write left that needs to be failed, if there is none we don't need
958 // to create a new exception or obtain an existing one.
959 if (!pendingUnencryptedWrites.isEmpty()) {
960 // Make a best effort to preserve any exception that way previously encountered from the
961 // handshake or the transport, else fallback to a general error.
962 Throwable exception = handshakePromise.cause();
963 if (exception == null) {
964 exception = sslClosePromise.cause();
965 if (exception == null) {
966 exception = new SslClosedEngineException("SSLEngine closed already");
967 }
968 }
969 pendingUnencryptedWrites.releaseAndFailAll(ctx, exception);
970 }
971
972 return;
973 } else {
974 switch (result.getHandshakeStatus()) {
975 case NEED_TASK:
976 if (!runDelegatedTasks(inUnwrap)) {
977 // We scheduled a task on the delegatingTaskExecutor, so stop processing as we will
978 // resume once the task completes.
979 break outer;
980 }
981 break;
982 case FINISHED:
983 case NOT_HANDSHAKING: // work around for android bug that skips the FINISHED state.
984 setHandshakeSuccess();
985 break;
986 case NEED_WRAP:
987 // If we are expected to wrap again and we produced some data we need to ensure there
988 // is something in the queue to process as otherwise we will not try again before there
989 // was more added. Failing to do so may fail to produce an alert that can be
990 // consumed by the remote peer.
991 if (result.bytesProduced() > 0 && pendingUnencryptedWrites.isEmpty()) {
992 pendingUnencryptedWrites.add(Unpooled.EMPTY_BUFFER);
993 }
994 break;
995 case NEED_UNWRAP:
996 // The underlying engine is starving so we need to feed it with more data.
997 // See https://github.com/netty/netty/pull/5039
998 readIfNeeded(ctx);
999 return;
1000 default:
1001 throw new IllegalStateException(
1002 "Unknown handshake status: " + result.getHandshakeStatus());
1003 }
1004 }
1005 }
1006 } finally {
1007 if (out != null) {
1008 out.release();
1009 }
1010 if (inUnwrap) {
1011 setState(STATE_NEEDS_FLUSH);
1012 }
1013 }
1014 }
1015
1016 /**
1017 * This method will not call
1018 * {@link #setHandshakeFailure(ChannelHandlerContext, Throwable, boolean, boolean, boolean)} or
1019 * {@link #setHandshakeFailure(ChannelHandlerContext, Throwable)}.
1020 * @return {@code true} if this method ends on {@link SSLEngineResult.HandshakeStatus#NOT_HANDSHAKING}.
1021 */
1022 private boolean wrapNonAppData(final ChannelHandlerContext ctx, boolean inUnwrap) throws SSLException {
1023 ByteBuf out = null;
1024 ByteBufAllocator alloc = ctx.alloc();
1025 try {
1026 // Only continue to loop if the handler was not removed in the meantime.
1027 // See https://github.com/netty/netty/issues/5860
1028 outer: while (!ctx.isRemoved()) {
1029 if (out == null) {
1030 // As this is called for the handshake we have no real idea how big the buffer needs to be.
1031 // That said 2048 should give us enough room to include everything like ALPN / NPN data.
1032 // If this is not enough we will increase the buffer in wrap(...).
1033 out = allocateOutNetBuf(ctx, 2048, 1);
1034 }
1035 SSLEngineResult result = wrap(alloc, engine, Unpooled.EMPTY_BUFFER, out);
1036 if (result.bytesProduced() > 0) {
1037 ctx.write(out).addListener(future -> {
1038 Throwable cause = future.cause();
1039 if (cause != null) {
1040 setHandshakeFailureTransportFailure(ctx, cause);
1041 }
1042 });
1043 if (inUnwrap) {
1044 setState(STATE_NEEDS_FLUSH);
1045 }
1046 out = null;
1047 }
1048
1049 HandshakeStatus status = result.getHandshakeStatus();
1050 switch (status) {
1051 case FINISHED:
1052 // We may be here because we read data and discovered the remote peer initiated a renegotiation
1053 // and this write is to complete the new handshake. The user may have previously done a
1054 // writeAndFlush which wasn't able to wrap data due to needing the pending handshake, so we
1055 // attempt to wrap application data here if any is pending.
1056 if (setHandshakeSuccess() && inUnwrap && !pendingUnencryptedWrites.isEmpty()) {
1057 wrap(ctx, true);
1058 }
1059 return false;
1060 case NEED_TASK:
1061 if (!runDelegatedTasks(inUnwrap)) {
1062 // We scheduled a task on the delegatingTaskExecutor, so stop processing as we will
1063 // resume once the task completes.
1064 break outer;
1065 }
1066 break;
1067 case NEED_UNWRAP:
1068 if (inUnwrap || unwrapNonAppData(ctx) <= 0) {
1069 // If we asked for a wrap, the engine requested an unwrap, and we are in unwrap there is
1070 // no use in trying to call wrap again because we have already attempted (or will after we
1071 // return) to feed more data to the engine.
1072 return false;
1073 }
1074 break;
1075 case NEED_WRAP:
1076 break;
1077 case NOT_HANDSHAKING:
1078 if (setHandshakeSuccess() && inUnwrap && !pendingUnencryptedWrites.isEmpty()) {
1079 wrap(ctx, true);
1080 }
1081 // Workaround for TLS False Start problem reported at:
1082 // https://github.com/netty/netty/issues/1108#issuecomment-14266970
1083 if (!inUnwrap) {
1084 unwrapNonAppData(ctx);
1085 }
1086 return true;
1087 default:
1088 throw new IllegalStateException("Unknown handshake status: " + result.getHandshakeStatus());
1089 }
1090
1091 // Check if did not produce any bytes and if so break out of the loop, but only if we did not process
1092 // a task as last action. It's fine to not produce any data as part of executing a task.
1093 if (result.bytesProduced() == 0 && status != HandshakeStatus.NEED_TASK) {
1094 break;
1095 }
1096 }
1097 } finally {
1098 if (out != null) {
1099 out.release();
1100 }
1101 }
1102 return false;
1103 }
1104
1105 private SSLEngineResult wrapMultiple(ByteBufAllocator alloc, SSLEngine engine, ByteBuf in, ByteBuf out)
1106 throws SSLException {
1107 SSLEngineResult result = null;
1108
1109 do {
1110 int nextSliceSize = Math.min(MAX_PLAINTEXT_LENGTH, in.readableBytes());
1111 // This call over-estimates, because we are slicing and not every nioBuffer will be part of
1112 // every slice. We could improve the estimate by having an nioBufferCount(offset, length).
1113 int nextOutSize = engineType.calculateRequiredOutBufSpace(this, nextSliceSize, in.nioBufferCount());
1114
1115 if (!out.isWritable(nextOutSize)) {
1116 if (result != null) {
1117 // We underestimated the space needed to encrypt the entire in buf. Break out, and
1118 // upstream will re-enqueue the buffer for later.
1119 break;
1120 }
1121 // This shouldn't happen, as the out buf was properly sized for at least packetLength
1122 // prior to calling wrap.
1123 out.ensureWritable(nextOutSize);
1124 }
1125
1126 ByteBuf wrapBuf = in.readSlice(nextSliceSize);
1127 result = wrap(alloc, engine, wrapBuf, out);
1128
1129 if (result.getStatus() == Status.CLOSED) {
1130 // If the engine gets closed, we can exit out early. Otherwise, we'll do a full handling of
1131 // possible results once finished.
1132 break;
1133 }
1134
1135 if (wrapBuf.isReadable()) {
1136 // There may be some left-over, in which case we can just pick it up next loop, so reset the original
1137 // reader index so its included again in the next slice.
1138 in.readerIndex(in.readerIndex() - wrapBuf.readableBytes());
1139 }
1140
1141 if (result.bytesConsumed() == 0 && result.bytesProduced() == 0) {
1142 // The engine did neither consume nor produce anything, for example because the handshake is not done
1143 // yet. Break out so the caller handles the handshake status, as otherwise we would loop forever.
1144 break;
1145 }
1146 } while (in.readableBytes() > 0);
1147
1148 return result;
1149 }
1150
1151 private SSLEngineResult wrap(ByteBufAllocator alloc, SSLEngine engine, ByteBuf in, ByteBuf out)
1152 throws SSLException {
1153 ByteBuf newDirectIn = null;
1154 try {
1155 int readerIndex = in.readerIndex();
1156 int readableBytes = in.readableBytes();
1157
1158 // We will call SslEngine.wrap(ByteBuffer[], ByteBuffer) to allow efficient handling of
1159 // CompositeByteBuf without force an extra memory copy when CompositeByteBuffer.nioBuffer() is called.
1160 final ByteBuffer[] in0;
1161 if (in.isDirect() || !engineType.wantsDirectBuffer) {
1162 // As CompositeByteBuf.nioBufferCount() can be expensive (as it needs to check all composed ByteBuf
1163 // to calculate the count) we will just assume a CompositeByteBuf contains more then 1 ByteBuf.
1164 // The worst that can happen is that we allocate an extra ByteBuffer[] in CompositeByteBuf.nioBuffers()
1165 // which is better then walking the composed ByteBuf in most cases.
1166 if (!(in instanceof CompositeByteBuf) && in.nioBufferCount() == 1) {
1167 in0 = singleBuffer;
1168 // We know its only backed by 1 ByteBuffer so use internalNioBuffer to keep object allocation
1169 // to a minimum.
1170 in0[0] = in.internalNioBuffer(readerIndex, readableBytes);
1171 } else {
1172 in0 = in.nioBuffers();
1173 }
1174 } else {
1175 // We could even go further here and check if its a CompositeByteBuf and if so try to decompose it and
1176 // only replace the ByteBuffer that are not direct. At the moment we just will replace the whole
1177 // CompositeByteBuf to keep the complexity to a minimum
1178 newDirectIn = alloc.directBuffer(readableBytes);
1179 newDirectIn.writeBytes(in, readerIndex, readableBytes);
1180 in0 = singleBuffer;
1181 in0[0] = newDirectIn.internalNioBuffer(newDirectIn.readerIndex(), readableBytes);
1182 }
1183
1184 for (;;) {
1185 // Use toByteBuffer(...) which might be able to return the internal ByteBuffer and so reduce
1186 // allocations.
1187 ByteBuffer out0 = toByteBuffer(out, out.writerIndex(), out.writableBytes());
1188 SSLEngineResult result = engine.wrap(in0, out0);
1189 in.skipBytes(result.bytesConsumed());
1190 out.writerIndex(out.writerIndex() + result.bytesProduced());
1191
1192 if (result.getStatus() == Status.BUFFER_OVERFLOW) {
1193 out.ensureWritable(engine.getSession().getPacketBufferSize());
1194 } else {
1195 return result;
1196 }
1197 }
1198 } finally {
1199 // Null out to allow GC of ByteBuffer
1200 singleBuffer[0] = null;
1201
1202 if (newDirectIn != null) {
1203 newDirectIn.release();
1204 }
1205 }
1206 }
1207
1208 @Override
1209 public void channelInactive(ChannelHandlerContext ctx) throws Exception {
1210 boolean handshakeFailed = handshakePromise.cause() != null;
1211
1212 // Channel closed, we will generate 'ClosedChannelException' now.
1213 StacklessClosedChannelException exception = StacklessClosedChannelException.newInstance(
1214 SslHandler.class, "channelInactive(ChannelHandlerContext)");
1215
1216 // Add a suppressed exception if the handshake was not completed yet.
1217 if (isStateSet(STATE_HANDSHAKE_STARTED) && !handshakePromise.isDone()) {
1218 ThrowableUtil.addSuppressed(exception, StacklessSSLHandshakeException.newInstance(
1219 "Connection closed while SSL/TLS handshake was in progress",
1220 SslHandler.class, "channelInactive"));
1221 }
1222
1223 // Make sure to release SSLEngine,
1224 // and notify the handshake future if the connection has been closed during handshake.
1225 setHandshakeFailure(ctx, exception, !isStateSet(STATE_OUTBOUND_CLOSED), isStateSet(STATE_HANDSHAKE_STARTED),
1226 false);
1227
1228 // Ensure we always notify the sslClosePromise as well
1229 notifyClosePromise(exception);
1230
1231 try {
1232 super.channelInactive(ctx);
1233 } catch (DecoderException e) {
1234 if (!handshakeFailed || !(e.getCause() instanceof SSLException)) {
1235 // We only rethrow the exception if the handshake did not fail before channelInactive(...) was called
1236 // as otherwise this may produce duplicated failures as super.channelInactive(...) will also call
1237 // channelRead(...).
1238 //
1239 // See https://github.com/netty/netty/issues/10119
1240 throw e;
1241 }
1242 }
1243 }
1244
1245 @Override
1246 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
1247 if (ignoreException(cause)) {
1248 // It is safe to ignore the 'connection reset by peer' or
1249 // 'broken pipe' error after sending close_notify.
1250 if (logger.isDebugEnabled()) {
1251 logger.debug(
1252 "{} Swallowing a harmless 'connection reset by peer / broken pipe' error that occurred " +
1253 "while writing close_notify in response to the peer's close_notify", ctx.channel(), cause);
1254 }
1255
1256 // Close the connection explicitly just in case the transport
1257 // did not close the connection automatically.
1258 if (ctx.channel().isActive()) {
1259 ctx.close();
1260 }
1261 } else {
1262 ctx.fireExceptionCaught(cause);
1263 }
1264 }
1265
1266 /**
1267 * Checks if the given {@link Throwable} can be ignore and just "swallowed"
1268 *
1269 * When an ssl connection is closed a close_notify message is sent.
1270 * After that the peer also sends close_notify however, it's not mandatory to receive
1271 * the close_notify. The party who sent the initial close_notify can close the connection immediately
1272 * then the peer will get connection reset error.
1273 *
1274 */
1275 private boolean ignoreException(Throwable t) {
1276 if (!(t instanceof SSLException) && t instanceof IOException && sslClosePromise.isDone()) {
1277 String message = t.getMessage();
1278
1279 // first try to match connection reset / broke peer based on the regex. This is the fastest way
1280 // but may fail on different jdk impls or OS's
1281 if (message != null && IGNORABLE_ERROR_MESSAGE.matcher(message).matches()) {
1282 return true;
1283 }
1284
1285 // Inspect the StackTraceElements to see if it was a connection reset / broken pipe or not
1286 StackTraceElement[] elements = t.getStackTrace();
1287 for (StackTraceElement element: elements) {
1288 String classname = element.getClassName();
1289 String methodname = element.getMethodName();
1290
1291 // skip all classes that belong to the io.netty package
1292 if (classname.startsWith("io.netty.")) {
1293 continue;
1294 }
1295
1296 // check if the method name is read if not skip it
1297 if (!"read".equals(methodname)) {
1298 continue;
1299 }
1300
1301 if (isIgnorableClassInStack(classname)) {
1302 return true;
1303 }
1304
1305 try {
1306 // No match by now. Try to load the class via classloader and inspect it.
1307 // This is mainly done as other JDK implementations may differ in name of
1308 // the impl.
1309 Class<?> clazz = PlatformDependent.getClassLoader(getClass()).loadClass(classname);
1310
1311 if (SocketChannel.class.isAssignableFrom(clazz)
1312 || DatagramChannel.class.isAssignableFrom(clazz)) {
1313 return true;
1314 }
1315
1316 // also match against SctpChannel via String matching as it may not present.
1317 if ("com.sun.nio.sctp.SctpChannel".equals(clazz.getSuperclass().getName())) {
1318 return true;
1319 }
1320 } catch (Throwable cause) {
1321 if (logger.isDebugEnabled()) {
1322 logger.debug("Unexpected exception while loading class {} classname {}",
1323 getClass(), classname, cause);
1324 }
1325 }
1326 }
1327 }
1328
1329 return false;
1330 }
1331
1332 private static boolean isIgnorableClassInStack(String classname) {
1333 return classname.contains("SocketChannel") ||
1334 classname.contains("DatagramChannel") ||
1335 classname.contains("SctpChannel") ||
1336 classname.contains("UdtChannel");
1337 }
1338
1339 /**
1340 * Returns {@code true} if the given {@link ByteBuf} is encrypted. Be aware that this method
1341 * will not increase the readerIndex of the given {@link ByteBuf}.
1342 *
1343 * @param buffer
1344 * The {@link ByteBuf} to read from. Be aware that it must have at least 5 bytes to read,
1345 * otherwise it will throw an {@link IllegalArgumentException}.
1346 * @return encrypted
1347 * {@code true} if the {@link ByteBuf} is encrypted, {@code false} otherwise.
1348 * @throws IllegalArgumentException
1349 * Is thrown if the given {@link ByteBuf} has not at least 5 bytes to read.
1350 * @deprecated use {@link #isEncrypted(ByteBuf, boolean)}.
1351 */
1352 @Deprecated
1353 public static boolean isEncrypted(ByteBuf buffer) {
1354 return isEncrypted(buffer, false);
1355 }
1356
1357 /**
1358 * Returns {@code true} if the given {@link ByteBuf} is encrypted. Be aware that this method
1359 * will not increase the readerIndex of the given {@link ByteBuf}.
1360 *
1361 * @param buffer
1362 * The {@link ByteBuf} to read from. Be aware that it must have at least 5 bytes to read,
1363 * otherwise it will throw an {@link IllegalArgumentException}.
1364 * @return encrypted
1365 * {@code true} if the {@link ByteBuf} is encrypted, {@code false} otherwise.
1366 * @param probeSSLv2
1367 * {@code true} if the input {@code buffer} might be SSLv2. If {@code true} is used this
1368 * methods might produce false-positives in some cases so it's strongly suggested to
1369 * use {@code false}.
1370 * @throws IllegalArgumentException
1371 * Is thrown if the given {@link ByteBuf} has not at least 5 bytes to read.
1372 */
1373 public static boolean isEncrypted(ByteBuf buffer, boolean probeSSLv2) {
1374 if (buffer.readableBytes() < SslUtils.SSL_RECORD_HEADER_LENGTH) {
1375 throw new IllegalArgumentException(
1376 "buffer must have at least " + SslUtils.SSL_RECORD_HEADER_LENGTH + " readable bytes");
1377 }
1378 return getEncryptedPacketLength(buffer, buffer.readerIndex(), probeSSLv2) != SslUtils.NOT_ENCRYPTED;
1379 }
1380
1381 private void decodeJdkCompatible(ChannelHandlerContext ctx, ByteBuf in) throws NotSslRecordException {
1382 int packetLength = this.packetLength;
1383 // If we calculated the length of the current SSL record before, use that information.
1384 if (packetLength > 0) {
1385 if (in.readableBytes() < packetLength) {
1386 return;
1387 }
1388 } else {
1389 // Get the packet length and wait until we get a packets worth of data to unwrap.
1390 final int readableBytes = in.readableBytes();
1391 if (readableBytes < SslUtils.SSL_RECORD_HEADER_LENGTH) {
1392 return;
1393 }
1394 packetLength = getEncryptedPacketLength(in, in.readerIndex(), true);
1395 if (packetLength == SslUtils.NOT_ENCRYPTED) {
1396 // Not an SSL/TLS packet
1397 NotSslRecordException e = new NotSslRecordException(
1398 "not an SSL/TLS record: " + ByteBufUtil.hexDump(in));
1399 in.skipBytes(in.readableBytes());
1400
1401 // First fail the handshake promise as we may need to have access to the SSLEngine which may
1402 // be released because the user will remove the SslHandler in an exceptionCaught(...) implementation.
1403 setHandshakeFailure(ctx, e);
1404
1405 throw e;
1406 }
1407 if (packetLength == NOT_ENOUGH_DATA) {
1408 return;
1409 }
1410 assert packetLength > 0;
1411 if (packetLength > readableBytes) {
1412 // wait until the whole packet can be read
1413 this.packetLength = packetLength;
1414 return;
1415 }
1416 }
1417
1418 // Reset the state of this class so we can get the length of the next packet. We assume the entire packet will
1419 // be consumed by the SSLEngine.
1420 this.packetLength = 0;
1421 try {
1422 final int bytesConsumed = unwrap(ctx, in, packetLength);
1423 if (bytesConsumed != packetLength && !engine.isInboundDone()) {
1424 // The JDK equivalent of getEncryptedPacketLength has some optimizations and can behave slightly
1425 // differently to ours, but this should always be a sign of bad input data.
1426 throw new NotSslRecordException();
1427 }
1428 } catch (Throwable cause) {
1429 handleUnwrapThrowable(ctx, cause);
1430 }
1431 }
1432
1433 private void decodeNonJdkCompatible(ChannelHandlerContext ctx, ByteBuf in) {
1434 try {
1435 unwrap(ctx, in, in.readableBytes());
1436 } catch (Throwable cause) {
1437 handleUnwrapThrowable(ctx, cause);
1438 }
1439 }
1440
1441 private void handleUnwrapThrowable(ChannelHandlerContext ctx, Throwable cause) {
1442 try {
1443 // We should attempt to notify the handshake failure before writing any pending data. If we are in unwrap
1444 // and failed during the handshake process, and we attempt to wrap, then promises will fail, and if
1445 // listeners immediately close the Channel then we may end up firing the handshake event after the Channel
1446 // has been closed.
1447 if (handshakePromise.tryFailure(cause)) {
1448 ctx.fireUserEventTriggered(new SslHandshakeCompletionEvent(cause));
1449 }
1450
1451 // Let's check if the handler was removed in the meantime and so pendingUnencryptedWrites is null.
1452 if (pendingUnencryptedWrites != null) {
1453 // We need to flush one time as there may be an alert that we should send to the remote peer because
1454 // of the SSLException reported here.
1455 wrapAndFlush(ctx);
1456 }
1457 } catch (SSLException ex) {
1458 logger.debug("SSLException during trying to call SSLEngine.wrap(...)" +
1459 " because of an previous SSLException, ignoring...", ex);
1460 } finally {
1461 // ensure we always flush and close the channel.
1462 setHandshakeFailure(ctx, cause, true, false, true);
1463 }
1464 PlatformDependent.throwException(cause);
1465 }
1466
1467 @Override
1468 protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws SSLException {
1469 if (isStateSet(STATE_PROCESS_TASK)) {
1470 return;
1471 }
1472 if (jdkCompatibilityMode) {
1473 decodeJdkCompatible(ctx, in);
1474 } else {
1475 decodeNonJdkCompatible(ctx, in);
1476 }
1477 }
1478
1479 @Override
1480 public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
1481 channelReadComplete0(ctx);
1482 }
1483
1484 private void channelReadComplete0(ChannelHandlerContext ctx) {
1485 // Discard bytes of the cumulation buffer if needed.
1486 discardSomeReadBytes();
1487
1488 flushIfNeeded(ctx);
1489 readIfNeeded(ctx);
1490
1491 clearState(STATE_FIRE_CHANNEL_READ);
1492 ctx.fireChannelReadComplete();
1493 }
1494
1495 private void readIfNeeded(ChannelHandlerContext ctx) {
1496 // If handshake is not finished yet, we need more data.
1497 if (!ctx.channel().config().isAutoRead() &&
1498 (!isStateSet(STATE_FIRE_CHANNEL_READ) || !handshakePromise.isDone())) {
1499 // No auto-read used and no message passed through the ChannelPipeline or the handshake was not complete
1500 // yet, which means we need to trigger the read to ensure we not encounter any stalls.
1501 ctx.read();
1502 }
1503 }
1504
1505 private void flushIfNeeded(ChannelHandlerContext ctx) {
1506 if (isStateSet(STATE_NEEDS_FLUSH)) {
1507 forceFlush(ctx);
1508 }
1509 }
1510
1511 /**
1512 * Calls {@link SSLEngine#unwrap(ByteBuffer, ByteBuffer)} with an empty buffer to handle handshakes, etc.
1513 */
1514 private int unwrapNonAppData(ChannelHandlerContext ctx) throws SSLException {
1515 return unwrap(ctx, Unpooled.EMPTY_BUFFER, 0);
1516 }
1517
1518 /**
1519 * Unwraps inbound SSL records.
1520 */
1521 private int unwrap(ChannelHandlerContext ctx, ByteBuf packet, int length) throws SSLException {
1522 final int originalLength = length;
1523 boolean wrapLater = false;
1524 boolean notifyClosure = false;
1525 boolean executedRead = false;
1526 ByteBuf decodeOut = allocate(ctx, engineType.calculateInitialUnwrapBufSize(this, length));
1527 try {
1528 // Only continue to loop if the handler was not removed in the meantime.
1529 // See https://github.com/netty/netty/issues/5860
1530 do {
1531 final SSLEngineResult result = engineType.unwrap(this, packet, length, decodeOut);
1532 final Status status = result.getStatus();
1533 final HandshakeStatus handshakeStatus = result.getHandshakeStatus();
1534 final int produced = result.bytesProduced();
1535 final int consumed = result.bytesConsumed();
1536
1537 // Skip bytes now in case unwrap is called in a re-entry scenario. For example LocalChannel.read()
1538 // may entry this method in a re-entry fashion and if the peer is writing into a shared buffer we may
1539 // unwrap the same data multiple times.
1540 packet.skipBytes(consumed);
1541 length -= consumed;
1542
1543 // The expected sequence of events is:
1544 // 1. Notify of handshake success
1545 // 2. fireChannelRead for unwrapped data
1546 if (handshakeStatus == HandshakeStatus.FINISHED || handshakeStatus == HandshakeStatus.NOT_HANDSHAKING) {
1547 wrapLater |= (decodeOut.isReadable() ?
1548 setHandshakeSuccessUnwrapMarkReentry() : setHandshakeSuccess()) ||
1549 handshakeStatus == HandshakeStatus.FINISHED ||
1550 // We need to check if pendingUnecryptedWrites is null as the SslHandler
1551 // might have been removed in the meantime.
1552 (pendingUnencryptedWrites != null && !pendingUnencryptedWrites.isEmpty());
1553 }
1554
1555 // Dispatch decoded data after we have notified of handshake success. If this method has been invoked
1556 // in a re-entry fashion we execute a task on the executor queue to process after the stack unwinds
1557 // to preserve order of events.
1558 if (decodeOut.isReadable()) {
1559 setState(STATE_FIRE_CHANNEL_READ);
1560 if (isStateSet(STATE_UNWRAP_REENTRY)) {
1561 executedRead = true;
1562 executeChannelRead(ctx, decodeOut);
1563 } else {
1564 ctx.fireChannelRead(decodeOut);
1565 }
1566 decodeOut = null;
1567 }
1568
1569 if (status == Status.CLOSED) {
1570 notifyClosure = true; // notify about the CLOSED state of the SSLEngine. See #137
1571 } else if (status == Status.BUFFER_OVERFLOW) {
1572 if (decodeOut != null) {
1573 decodeOut.release();
1574 }
1575 final int applicationBufferSize = engine.getSession().getApplicationBufferSize();
1576 // Allocate a new buffer which can hold all the rest data and loop again.
1577 // It may happen that applicationBufferSize < produced while there is still more to unwrap, in this
1578 // case we will just allocate a new buffer with the capacity of applicationBufferSize and call
1579 // unwrap again.
1580 decodeOut = allocate(ctx, engineType.calculatePendingData(this, applicationBufferSize < produced ?
1581 applicationBufferSize : applicationBufferSize - produced));
1582 continue;
1583 }
1584
1585 if (handshakeStatus == HandshakeStatus.NEED_TASK) {
1586 boolean pending = runDelegatedTasks(true);
1587 if (!pending) {
1588 // We scheduled a task on the delegatingTaskExecutor, so stop processing as we will
1589 // resume once the task completes.
1590 //
1591 // We break out of the loop only and do NOT return here as we still may need to notify
1592 // about the closure of the SSLEngine.
1593 wrapLater = false;
1594 break;
1595 }
1596 } else if (handshakeStatus == HandshakeStatus.NEED_WRAP) {
1597 // If the wrap operation transitions the status to NOT_HANDSHAKING and there is no more data to
1598 // unwrap then the next call to unwrap will not produce any data. We can avoid the potentially
1599 // costly unwrap operation and break out of the loop.
1600 if (wrapNonAppData(ctx, true) && length == 0) {
1601 break;
1602 }
1603 }
1604
1605 if (status == Status.BUFFER_UNDERFLOW ||
1606 // If we processed NEED_TASK we should try again even we did not consume or produce anything.
1607 handshakeStatus != HandshakeStatus.NEED_TASK && (consumed == 0 && produced == 0 ||
1608 (length == 0 && handshakeStatus == HandshakeStatus.NOT_HANDSHAKING))) {
1609 if (handshakeStatus == HandshakeStatus.NEED_UNWRAP) {
1610 // The underlying engine is starving so we need to feed it with more data.
1611 // See https://github.com/netty/netty/pull/5039
1612 readIfNeeded(ctx);
1613 }
1614
1615 break;
1616 } else if (decodeOut == null) {
1617 decodeOut = allocate(ctx, engineType.calculateInitialUnwrapBufSize(this, length));
1618 }
1619 } while (!ctx.isRemoved());
1620
1621 if (isStateSet(STATE_FLUSHED_BEFORE_HANDSHAKE) && handshakePromise.isDone()) {
1622 // We need to call wrap(...) in case there was a flush done before the handshake completed to ensure
1623 // we do not stale.
1624 //
1625 // See https://github.com/netty/netty/pull/2437
1626 clearState(STATE_FLUSHED_BEFORE_HANDSHAKE);
1627 wrapLater = true;
1628 }
1629
1630 if (wrapLater) {
1631 wrap(ctx, true);
1632 }
1633 } finally {
1634 if (decodeOut != null) {
1635 decodeOut.release();
1636 }
1637
1638 if (notifyClosure) {
1639 if (executedRead) {
1640 executeNotifyClosePromise(ctx);
1641 } else {
1642 notifyClosePromise(null);
1643 }
1644 }
1645 }
1646 return originalLength - length;
1647 }
1648
1649 private boolean setHandshakeSuccessUnwrapMarkReentry() throws SSLException {
1650 // setHandshakeSuccess calls out to external methods which may trigger re-entry. We need to preserve ordering of
1651 // fireChannelRead for decodeOut relative to re-entry data.
1652 final boolean setReentryState = !isStateSet(STATE_UNWRAP_REENTRY);
1653 if (setReentryState) {
1654 setState(STATE_UNWRAP_REENTRY);
1655 }
1656 try {
1657 return setHandshakeSuccess();
1658 } finally {
1659 // It is unlikely this specific method will be re-entry because handshake completion is infrequent, but just
1660 // in case we only clear the state if we set it in the first place.
1661 if (setReentryState) {
1662 clearState(STATE_UNWRAP_REENTRY);
1663 }
1664 }
1665 }
1666
1667 private void executeNotifyClosePromise(final ChannelHandlerContext ctx) {
1668 try {
1669 ctx.executor().execute(() -> notifyClosePromise(null));
1670 } catch (RejectedExecutionException e) {
1671 notifyClosePromise(e);
1672 }
1673 }
1674
1675 private void executeChannelRead(final ChannelHandlerContext ctx, final ByteBuf decodedOut) {
1676 try {
1677 ctx.executor().execute(() -> ctx.fireChannelRead(decodedOut));
1678 } catch (RejectedExecutionException e) {
1679 decodedOut.release();
1680 throw e;
1681 }
1682 }
1683
1684 private static ByteBuffer toByteBuffer(ByteBuf out, int index, int len) {
1685 return out.nioBufferCount() == 1 ? out.internalNioBuffer(index, len) :
1686 out.nioBuffer(index, len);
1687 }
1688
1689 private static boolean inEventLoop(Executor executor) {
1690 return executor instanceof EventExecutor && ((EventExecutor) executor).inEventLoop();
1691 }
1692
1693 /**
1694 * Will either run the delegated task directly calling {@link Runnable#run()} and return {@code true} or will
1695 * offload the delegated task using {@link Executor#execute(Runnable)} and return {@code false}.
1696 *
1697 * If the task is offloaded it will take care to resume its work on the {@link EventExecutor} once there are no
1698 * more tasks to process.
1699 */
1700 private boolean runDelegatedTasks(boolean inUnwrap) {
1701 if (delegatedTaskExecutor == ImmediateExecutor.INSTANCE || inEventLoop(delegatedTaskExecutor)) {
1702 // We should run the task directly in the EventExecutor thread and not offload at all. As we are on the
1703 // EventLoop we can just run all tasks at once.
1704 for (;;) {
1705 Runnable task = engine.getDelegatedTask();
1706 if (task == null) {
1707 return true;
1708 }
1709 setState(STATE_PROCESS_TASK);
1710 if (task instanceof AsyncRunnable) {
1711 // Let's set the task to processing task before we try to execute it.
1712 boolean pending = false;
1713 try {
1714 AsyncRunnable asyncTask = (AsyncRunnable) task;
1715 AsyncTaskCompletionHandler completionHandler = new AsyncTaskCompletionHandler(inUnwrap);
1716 asyncTask.run(completionHandler);
1717 pending = completionHandler.resumeLater();
1718 if (pending) {
1719 return false;
1720 }
1721 } finally {
1722 if (!pending) {
1723 // The task has completed, lets clear the state. If it is not completed we will clear the
1724 // state once it is.
1725 clearState(STATE_PROCESS_TASK);
1726 }
1727 }
1728 } else {
1729 try {
1730 task.run();
1731 } finally {
1732 clearState(STATE_PROCESS_TASK);
1733 }
1734 }
1735 }
1736 } else {
1737 executeDelegatedTask(inUnwrap);
1738 return false;
1739 }
1740 }
1741
1742 private SslTasksRunner getTaskRunner(boolean inUnwrap) {
1743 return inUnwrap ? sslTaskRunnerForUnwrap : sslTaskRunner;
1744 }
1745
1746 private void executeDelegatedTask(boolean inUnwrap) {
1747 executeDelegatedTask(getTaskRunner(inUnwrap));
1748 }
1749
1750 private void executeDelegatedTask(SslTasksRunner task) {
1751 setState(STATE_PROCESS_TASK);
1752 try {
1753 delegatedTaskExecutor.execute(task);
1754 } catch (RejectedExecutionException e) {
1755 clearState(STATE_PROCESS_TASK);
1756 throw e;
1757 }
1758 }
1759
1760 private final class AsyncTaskCompletionHandler implements Runnable {
1761 private final boolean inUnwrap;
1762 boolean didRun;
1763 boolean resumeLater;
1764
1765 AsyncTaskCompletionHandler(boolean inUnwrap) {
1766 this.inUnwrap = inUnwrap;
1767 }
1768
1769 @Override
1770 public void run() {
1771 didRun = true;
1772 if (resumeLater) {
1773 getTaskRunner(inUnwrap).runComplete();
1774 }
1775 }
1776
1777 boolean resumeLater() {
1778 if (!didRun) {
1779 resumeLater = true;
1780 return true;
1781 }
1782 return false;
1783 }
1784 }
1785
1786 /**
1787 * {@link Runnable} that will be scheduled on the {@code delegatedTaskExecutor} and will take care
1788 * of resume work on the {@link EventExecutor} once the task was executed.
1789 */
1790 private final class SslTasksRunner implements Runnable {
1791 private final boolean inUnwrap;
1792 private final Runnable runCompleteTask = this::runComplete;
1793
1794 SslTasksRunner(boolean inUnwrap) {
1795 this.inUnwrap = inUnwrap;
1796 }
1797
1798 // Handle errors which happened during task processing.
1799 private void taskError(Throwable e) {
1800 if (inUnwrap) {
1801 // As the error happened while the task was scheduled as part of unwrap(...) we also need to ensure
1802 // we fire it through the pipeline as inbound error to be consistent with what we do in decode(...).
1803 //
1804 // This will also ensure we fail the handshake future and flush all produced data.
1805 try {
1806 handleUnwrapThrowable(ctx, e);
1807 } catch (Throwable cause) {
1808 safeExceptionCaught(cause);
1809 }
1810 } else {
1811 setHandshakeFailure(ctx, e);
1812 forceFlush(ctx);
1813 }
1814 }
1815
1816 // Try to call exceptionCaught(...)
1817 private void safeExceptionCaught(Throwable cause) {
1818 try {
1819 exceptionCaught(ctx, wrapIfNeeded(cause));
1820 } catch (Throwable error) {
1821 ctx.fireExceptionCaught(error);
1822 }
1823 }
1824
1825 private Throwable wrapIfNeeded(Throwable cause) {
1826 if (!inUnwrap) {
1827 // If we are not in unwrap(...) we can just rethrow without wrapping at all.
1828 return cause;
1829 }
1830 // As the exception would have been triggered by an inbound operation we will need to wrap it in a
1831 // DecoderException to mimic what a decoder would do when decode(...) throws.
1832 return cause instanceof DecoderException ? cause : new DecoderException(cause);
1833 }
1834
1835 private void tryDecodeAgain() {
1836 try {
1837 channelRead(ctx, Unpooled.EMPTY_BUFFER);
1838 } catch (Throwable cause) {
1839 safeExceptionCaught(cause);
1840 } finally {
1841 // As we called channelRead(...) we also need to call channelReadComplete(...) which
1842 // will ensure we either call ctx.fireChannelReadComplete() or will trigger a ctx.read() if
1843 // more data is needed.
1844 channelReadComplete0(ctx);
1845 }
1846 }
1847
1848 /**
1849 * Executed after the wrapped {@code task} was executed via {@code delegatedTaskExecutor} to resume work
1850 * on the {@link EventExecutor}.
1851 */
1852 private void resumeOnEventExecutor() {
1853 assert ctx.executor().inEventLoop();
1854 clearState(STATE_PROCESS_TASK);
1855 try {
1856 HandshakeStatus status = engine.getHandshakeStatus();
1857 switch (status) {
1858 // There is another task that needs to be executed and offloaded to the delegatingTaskExecutor as
1859 // a result of this. Let's reschedule....
1860 case NEED_TASK:
1861 executeDelegatedTask(this);
1862
1863 break;
1864
1865 // The handshake finished, lets notify about the completion of it and resume processing.
1866 case FINISHED:
1867 // Not handshaking anymore, lets notify about the completion if not done yet and resume processing.
1868 case NOT_HANDSHAKING:
1869 setHandshakeSuccess(); // NOT_HANDSHAKING -> workaround for android skipping FINISHED state.
1870 try {
1871 // Lets call wrap to ensure we produce the alert if there is any pending and also to
1872 // ensure we flush any queued data..
1873 wrap(ctx, inUnwrap);
1874 } catch (Throwable e) {
1875 taskError(e);
1876 return;
1877 }
1878 if (inUnwrap) {
1879 // If we were in the unwrap call when the task was processed we should also try to unwrap
1880 // non app data first as there may not anything left in the inbound buffer to process.
1881 unwrapNonAppData(ctx);
1882 }
1883
1884 // Flush now as we may have written some data as part of the wrap call.
1885 forceFlush(ctx);
1886
1887 tryDecodeAgain();
1888 break;
1889
1890 // We need more data so lets try to unwrap first and then call decode again which will feed us
1891 // with buffered data (if there is any).
1892 case NEED_UNWRAP:
1893 try {
1894 unwrapNonAppData(ctx);
1895 } catch (SSLException e) {
1896 handleUnwrapThrowable(ctx, e);
1897 return;
1898 }
1899 tryDecodeAgain();
1900 break;
1901
1902 // To make progress we need to call SSLEngine.wrap(...) which may produce more output data
1903 // that will be written to the Channel.
1904 case NEED_WRAP:
1905 try {
1906 if (!wrapNonAppData(ctx, false) && inUnwrap) {
1907 // The handshake finished in wrapNonAppData(...), we need to try call
1908 // unwrapNonAppData(...) as we may have some alert that we should read.
1909 //
1910 // This mimics what we would do when we are calling this method while in unwrap(...).
1911 unwrapNonAppData(ctx);
1912 }
1913
1914 // Flush now as we may have written some data as part of the wrap call.
1915 forceFlush(ctx);
1916 } catch (Throwable e) {
1917 taskError(e);
1918 return;
1919 }
1920
1921 // Now try to feed in more data that we have buffered.
1922 tryDecodeAgain();
1923 break;
1924
1925 default:
1926 // Should never reach here as we handle all cases.
1927 throw new AssertionError();
1928 }
1929 } catch (Throwable cause) {
1930 safeExceptionCaught(cause);
1931 }
1932 }
1933
1934 void runComplete() {
1935 EventExecutor executor = ctx.executor();
1936 if (executor.isShuttingDown()) {
1937 // The executor is already shutting down, just return.
1938 return;
1939 }
1940 // Jump back on the EventExecutor. We do this even if we are already on the EventLoop to guard against
1941 // reentrancy issues. Failing to do so could lead to the situation of tryDecode(...) be called and so
1942 // channelRead(...) while still in the decode loop. In this case channelRead(...) might release the input
1943 // buffer if its empty which would then result in an IllegalReferenceCountException when we try to continue
1944 // decoding.
1945 //
1946 // See https://github.com/netty/netty-tcnative/issues/680
1947 executor.execute(this::resumeOnEventExecutor);
1948 }
1949
1950 @Override
1951 public void run() {
1952 try {
1953 Runnable task = engine.getDelegatedTask();
1954 if (task == null) {
1955 // The task was processed in the meantime. Let's just return.
1956 return;
1957 }
1958 if (task instanceof AsyncRunnable) {
1959 AsyncRunnable asyncTask = (AsyncRunnable) task;
1960 asyncTask.run(runCompleteTask);
1961 } else {
1962 task.run();
1963 runComplete();
1964 }
1965 } catch (final Throwable cause) {
1966 handleException(cause);
1967 }
1968 }
1969
1970 private void handleException(final Throwable cause) {
1971 EventExecutor executor = ctx.executor();
1972 if (executor.inEventLoop()) {
1973 clearState(STATE_PROCESS_TASK);
1974 safeExceptionCaught(cause);
1975 } else {
1976 try {
1977 executor.execute(() -> {
1978 clearState(STATE_PROCESS_TASK);
1979 safeExceptionCaught(cause);
1980 });
1981 } catch (RejectedExecutionException ignore) {
1982 clearState(STATE_PROCESS_TASK);
1983 // the context itself will handle the rejected exception when try to schedule the operation so
1984 // ignore the RejectedExecutionException
1985 ctx.fireExceptionCaught(cause);
1986 }
1987 }
1988 }
1989 }
1990
1991 /**
1992 * Notify all the handshake futures about the successfully handshake
1993 * @return {@code true} if {@link #handshakePromise} was set successfully and a {@link SslHandshakeCompletionEvent}
1994 * was fired. {@code false} otherwise.
1995 */
1996 private boolean setHandshakeSuccess() throws SSLException {
1997 // Our control flow may invoke this method multiple times for a single FINISHED event. For example
1998 // wrapNonAppData may drain pendingUnencryptedWrites in wrap which transitions to handshake from FINISHED to
1999 // NOT_HANDSHAKING which invokes setHandshakeSuccess, and then wrapNonAppData also directly invokes this method.
2000 final SSLSession session = engine.getSession();
2001 if (resumptionController != null && !handshakePromise.isDone()) {
2002 try {
2003 if (resumptionController.validateResumeIfNeeded(engine) && logger.isDebugEnabled()) {
2004 logger.debug("{} Resumed and reauthenticated session", ctx.channel());
2005 }
2006 } catch (CertificateException e) {
2007 SSLHandshakeException exception = new SSLHandshakeException(e.getMessage());
2008 exception.initCause(e);
2009 throw exception;
2010 }
2011 }
2012 final boolean notified = !handshakePromise.isDone() && handshakePromise.trySuccess(ctx.channel());
2013 if (notified) {
2014 if (logger.isDebugEnabled()) {
2015 logger.debug(
2016 "{} HANDSHAKEN: protocol:{} cipher suite:{}",
2017 ctx.channel(),
2018 session.getProtocol(),
2019 session.getCipherSuite());
2020 }
2021 ctx.fireUserEventTriggered(SslHandshakeCompletionEvent.SUCCESS);
2022 }
2023 if (isStateSet(STATE_READ_DURING_HANDSHAKE)) {
2024 clearState(STATE_READ_DURING_HANDSHAKE);
2025 if (!ctx.channel().config().isAutoRead()) {
2026 ctx.read();
2027 }
2028 }
2029 return notified;
2030 }
2031
2032 /**
2033 * Notify all the handshake futures about the failure during the handshake.
2034 */
2035 private void setHandshakeFailure(ChannelHandlerContext ctx, Throwable cause) {
2036 setHandshakeFailure(ctx, cause, true, true, false);
2037 }
2038
2039 /**
2040 * Notify all the handshake futures about the failure during the handshake.
2041 */
2042 private void setHandshakeFailure(ChannelHandlerContext ctx, Throwable cause, boolean closeInbound,
2043 boolean notify, boolean alwaysFlushAndClose) {
2044 try {
2045 // Release all resources such as internal buffers that SSLEngine is managing.
2046 setState(STATE_OUTBOUND_CLOSED);
2047 engine.closeOutbound();
2048
2049 if (closeInbound) {
2050 try {
2051 engine.closeInbound();
2052 } catch (SSLException e) {
2053 if (logger.isDebugEnabled()) {
2054 // only log in debug mode as it most likely harmless and latest chrome still trigger
2055 // this all the time.
2056 //
2057 // See https://github.com/netty/netty/issues/1340
2058 String msg = e.getMessage();
2059 if (msg == null || !(msg.contains("possible truncation attack") ||
2060 msg.contains("closing inbound before receiving peer's close_notify"))) {
2061 logger.debug("{} SSLEngine.closeInbound() raised an exception.", ctx.channel(), e);
2062 }
2063 }
2064 }
2065 }
2066 if (handshakePromise.tryFailure(cause) || alwaysFlushAndClose) {
2067 SslUtils.handleHandshakeFailure(ctx, cause, notify);
2068 }
2069 } finally {
2070 // Ensure we remove and fail all pending writes in all cases and so release memory quickly.
2071 releaseAndFailAll(ctx, cause);
2072 }
2073 }
2074
2075 private void setHandshakeFailureTransportFailure(ChannelHandlerContext ctx, Throwable cause) {
2076 // If TLS control frames fail to write we are in an unknown state and may become out of
2077 // sync with our peer. We give up and close the channel. This will also take care of
2078 // cleaning up any outstanding state (e.g. handshake promise, queued unencrypted data).
2079 try {
2080 SSLException transportFailure = new SSLException("failure when writing TLS control frames", cause);
2081 releaseAndFailAll(ctx, transportFailure);
2082 if (handshakePromise.tryFailure(transportFailure)) {
2083 ctx.fireUserEventTriggered(new SslHandshakeCompletionEvent(transportFailure));
2084 }
2085 } finally {
2086 ctx.close();
2087 }
2088 }
2089
2090 private void releaseAndFailAll(ChannelHandlerContext ctx, Throwable cause) {
2091 if (resumptionController != null &&
2092 (!engine.getSession().isValid() || cause instanceof SSLHandshakeException)) {
2093 resumptionController.remove(engine());
2094 }
2095 if (pendingUnencryptedWrites != null) {
2096 pendingUnencryptedWrites.releaseAndFailAll(ctx, cause);
2097 }
2098 }
2099
2100 private void notifyClosePromise(Throwable cause) {
2101 if (cause == null) {
2102 if (sslClosePromise.trySuccess(ctx.channel())) {
2103 ctx.fireUserEventTriggered(SslCloseCompletionEvent.SUCCESS);
2104 }
2105 } else {
2106 if (sslClosePromise.tryFailure(cause)) {
2107 ctx.fireUserEventTriggered(new SslCloseCompletionEvent(cause));
2108 }
2109 }
2110 }
2111
2112 private void closeOutboundAndChannel(
2113 final ChannelHandlerContext ctx, final ChannelPromise promise, boolean disconnect) throws Exception {
2114 setState(STATE_OUTBOUND_CLOSED);
2115 engine.closeOutbound();
2116
2117 if (!ctx.channel().isActive()) {
2118 if (disconnect) {
2119 ctx.disconnect(promise);
2120 } else {
2121 ctx.close(promise);
2122 }
2123 return;
2124 }
2125
2126 ChannelPromise closeNotifyPromise = ctx.newPromise();
2127 try {
2128 flush(ctx, closeNotifyPromise);
2129 } finally {
2130 if (!isStateSet(STATE_CLOSE_NOTIFY)) {
2131 setState(STATE_CLOSE_NOTIFY);
2132 // It's important that we do not pass the original ChannelPromise to safeClose(...) as when flush(....)
2133 // throws an Exception it will be propagated to the AbstractChannelHandlerContext which will try
2134 // to fail the promise because of this. This will then fail as it was already completed by
2135 // safeClose(...). We create a new ChannelPromise and try to notify the original ChannelPromise
2136 // once it is complete. If we fail to do so we just ignore it as in this case it was failed already
2137 // because of a propagated Exception.
2138 //
2139 // See https://github.com/netty/netty/issues/5931
2140 safeClose(ctx, closeNotifyPromise, PromiseNotifier.cascade(false, ctx.newPromise(), promise));
2141 } else {
2142 /// We already handling the close_notify so just attach the promise to the sslClosePromise.
2143 sslClosePromise.addListener(f -> promise.setSuccess());
2144 }
2145 }
2146 }
2147
2148 private void flush(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
2149 if (pendingUnencryptedWrites != null) {
2150 pendingUnencryptedWrites.add(Unpooled.EMPTY_BUFFER, promise);
2151 } else {
2152 promise.setFailure(newPendingWritesNullException());
2153 }
2154 flush(ctx);
2155 }
2156
2157 @Override
2158 public void handlerAdded(final ChannelHandlerContext ctx) throws Exception {
2159 this.ctx = ctx;
2160 Channel channel = ctx.channel();
2161 pendingUnencryptedWrites = new SslHandlerCoalescingBufferQueue(channel, 16, engineType.wantsDirectBuffer) {
2162 @Override
2163 protected int wrapDataSize() {
2164 return SslHandler.this.wrapDataSize;
2165 }
2166 };
2167
2168 setOpensslEngineSocketFd(channel);
2169 boolean fastOpen = Boolean.TRUE.equals(channel.config().getOption(ChannelOption.TCP_FASTOPEN_CONNECT));
2170 boolean active = channel.isActive();
2171 if (active || fastOpen) {
2172 // Explicitly flush the handshake only if the channel is already active.
2173 // With TCP Fast Open, we write to the outbound buffer before the TCP connect is established.
2174 // The buffer will then be flushed as part of establishing the connection, saving us a round-trip.
2175 startHandshakeProcessing(active);
2176 // If we weren't able to include client_hello in the TCP SYN (e.g. no token, disabled at the OS) we have to
2177 // flush pending data in the outbound buffer later in channelActive().
2178 final ChannelOutboundBuffer outboundBuffer;
2179 if (fastOpen && ((outboundBuffer = channel.unsafe().outboundBuffer()) == null ||
2180 outboundBuffer.totalPendingWriteBytes() > 0)) {
2181 setState(STATE_NEEDS_FLUSH);
2182 }
2183 }
2184 }
2185
2186 private void startHandshakeProcessing(boolean flushAtEnd) {
2187 if (!isStateSet(STATE_HANDSHAKE_STARTED)) {
2188 setState(STATE_HANDSHAKE_STARTED);
2189 if (engine.getUseClientMode()) {
2190 // Begin the initial handshake.
2191 // channelActive() event has been fired already, which means this.channelActive() will
2192 // not be invoked. We have to initialize here instead.
2193 handshake(flushAtEnd);
2194 }
2195 applyHandshakeTimeout();
2196 } else if (isStateSet(STATE_NEEDS_FLUSH)) {
2197 forceFlush(ctx);
2198 }
2199 }
2200
2201 /**
2202 * Performs TLS renegotiation.
2203 */
2204 public Future<Channel> renegotiate() {
2205 ChannelHandlerContext ctx = this.ctx;
2206 if (ctx == null) {
2207 throw new IllegalStateException();
2208 }
2209
2210 return renegotiate(ctx.executor().newPromise());
2211 }
2212
2213 /**
2214 * Performs TLS renegotiation.
2215 */
2216 public Future<Channel> renegotiate(final Promise<Channel> promise) {
2217 ObjectUtil.checkNotNull(promise, "promise");
2218
2219 ChannelHandlerContext ctx = this.ctx;
2220 if (ctx == null) {
2221 throw new IllegalStateException();
2222 }
2223
2224 EventExecutor executor = ctx.executor();
2225 if (!executor.inEventLoop()) {
2226 executor.execute(() -> renegotiateOnEventLoop(promise));
2227 return promise;
2228 }
2229
2230 renegotiateOnEventLoop(promise);
2231 return promise;
2232 }
2233
2234 private void renegotiateOnEventLoop(final Promise<Channel> newHandshakePromise) {
2235 final Promise<Channel> oldHandshakePromise = handshakePromise;
2236 if (!oldHandshakePromise.isDone()) {
2237 // There's no need to handshake because handshake is in progress already.
2238 // Merge the new promise into the old one.
2239 PromiseNotifier.cascade(oldHandshakePromise, newHandshakePromise);
2240 } else {
2241 handshakePromise = newHandshakePromise;
2242 handshake(true);
2243 applyHandshakeTimeout();
2244 }
2245 }
2246
2247 /**
2248 * Performs TLS (re)negotiation.
2249 * @param flushAtEnd Set to {@code true} if the outbound buffer should be flushed (written to the network) at the
2250 * end. Set to {@code false} if the handshake will be flushed later, e.g. as part of TCP Fast Open
2251 * connect.
2252 */
2253 private void handshake(boolean flushAtEnd) {
2254 if (engine.getHandshakeStatus() != HandshakeStatus.NOT_HANDSHAKING) {
2255 // Not all SSLEngine implementations support calling beginHandshake multiple times while a handshake
2256 // is in progress. See https://github.com/netty/netty/issues/4718.
2257 return;
2258 }
2259 if (handshakePromise.isDone()) {
2260 // If the handshake is done already lets just return directly as there is no need to trigger it again.
2261 // This can happen if the handshake(...) was triggered before we called channelActive(...) by a
2262 // flush() that was triggered by a ChannelFutureListener that was added to the ChannelFuture returned
2263 // from the connect(...) method. In this case we will see the flush() happen before we had a chance to
2264 // call fireChannelActive() on the pipeline.
2265 return;
2266 }
2267
2268 // Begin handshake.
2269 final ChannelHandlerContext ctx = this.ctx;
2270 try {
2271 engine.beginHandshake();
2272 wrapNonAppData(ctx, false);
2273 } catch (Throwable e) {
2274 setHandshakeFailure(ctx, e);
2275 } finally {
2276 if (flushAtEnd) {
2277 forceFlush(ctx);
2278 }
2279 }
2280 }
2281
2282 private void applyHandshakeTimeout() {
2283 final Promise<Channel> localHandshakePromise = this.handshakePromise;
2284
2285 // Set timeout if necessary.
2286 final long handshakeTimeoutMillis = this.handshakeTimeoutMillis;
2287 if (handshakeTimeoutMillis <= 0 || localHandshakePromise.isDone()) {
2288 return;
2289 }
2290
2291 final Future<?> timeoutFuture = ctx.executor().schedule(() -> {
2292 if (localHandshakePromise.isDone()) {
2293 return;
2294 }
2295 SSLException exception =
2296 new SslHandshakeTimeoutException("handshake timed out after " + handshakeTimeoutMillis + "ms");
2297 try {
2298 if (localHandshakePromise.tryFailure(exception)) {
2299 SslUtils.handleHandshakeFailure(ctx, exception, true);
2300 }
2301 } finally {
2302 releaseAndFailAll(ctx, exception);
2303 }
2304 }, handshakeTimeoutMillis, TimeUnit.MILLISECONDS);
2305
2306 // Cancel the handshake timeout when handshake is finished.
2307 localHandshakePromise.addListener(f -> timeoutFuture.cancel(false));
2308 }
2309
2310 private void forceFlush(ChannelHandlerContext ctx) {
2311 clearState(STATE_NEEDS_FLUSH);
2312 ctx.flush();
2313 }
2314
2315 private void setOpensslEngineSocketFd(Channel c) {
2316 if (c instanceof UnixChannel && engine instanceof ReferenceCountedOpenSslEngine) {
2317 ((ReferenceCountedOpenSslEngine) engine).bioSetFd(((UnixChannel) c).fd().intValue());
2318 }
2319 }
2320
2321 /**
2322 * Issues an initial TLS handshake once connected when used in client-mode
2323 */
2324 @Override
2325 public void channelActive(final ChannelHandlerContext ctx) throws Exception {
2326 setOpensslEngineSocketFd(ctx.channel());
2327 if (!startTls) {
2328 startHandshakeProcessing(true);
2329 }
2330 ctx.fireChannelActive();
2331 }
2332
2333 private void safeClose(
2334 final ChannelHandlerContext ctx, final ChannelFuture flushFuture,
2335 final ChannelPromise promise) {
2336 if (!ctx.channel().isActive()) {
2337 ctx.close(promise);
2338 return;
2339 }
2340
2341 final Future<?> timeoutFuture;
2342 if (!flushFuture.isDone()) {
2343 long closeNotifyTimeout = closeNotifyFlushTimeoutMillis;
2344 if (closeNotifyTimeout > 0) {
2345 // Force-close the connection if close_notify is not fully sent in time.
2346 timeoutFuture = ctx.executor().schedule(() -> {
2347 // May be done in the meantime as cancel(...) is only best effort.
2348 if (!flushFuture.isDone()) {
2349 logger.warn("{} Last write attempt timed out; force-closing the connection.",
2350 ctx.channel());
2351 addCloseListener(ctx.close(ctx.newPromise()), promise);
2352 }
2353 }, closeNotifyTimeout, TimeUnit.MILLISECONDS);
2354 } else {
2355 timeoutFuture = null;
2356 }
2357 } else {
2358 timeoutFuture = null;
2359 }
2360
2361 // Close the connection if close_notify is sent in time.
2362 flushFuture.addListener(f -> {
2363 if (timeoutFuture != null) {
2364 timeoutFuture.cancel(false);
2365 }
2366 final long closeNotifyReadTimeout = closeNotifyReadTimeoutMillis;
2367 if (closeNotifyReadTimeout <= 0) {
2368 // Trigger the close in all cases to make sure the promise is notified
2369 // See https://github.com/netty/netty/issues/2358
2370 addCloseListener(ctx.close(ctx.newPromise()), promise);
2371 } else {
2372 final Future<?> closeNotifyReadTimeoutFuture;
2373
2374 if (!sslClosePromise.isDone()) {
2375 closeNotifyReadTimeoutFuture = ctx.executor().schedule(() -> {
2376 if (!sslClosePromise.isDone()) {
2377 logger.debug(
2378 "{} did not receive close_notify in {}ms; force-closing the connection.",
2379 ctx.channel(), closeNotifyReadTimeout);
2380
2381 // Do the close now...
2382 addCloseListener(ctx.close(ctx.newPromise()), promise);
2383 }
2384 }, closeNotifyReadTimeout, TimeUnit.MILLISECONDS);
2385 } else {
2386 closeNotifyReadTimeoutFuture = null;
2387 }
2388
2389 // Do the close once the we received the close_notify.
2390 sslClosePromise.addListener(future -> {
2391 if (closeNotifyReadTimeoutFuture != null) {
2392 closeNotifyReadTimeoutFuture.cancel(false);
2393 }
2394 addCloseListener(ctx.close(ctx.newPromise()), promise);
2395 });
2396 }
2397 });
2398 }
2399
2400 private static void addCloseListener(ChannelFuture future, ChannelPromise promise) {
2401 // We notify the promise in the ChannelPromiseNotifier as there is a "race" where the close(...) call
2402 // by the timeoutFuture and the close call in the flushFuture listener will be called. Because of
2403 // this we need to use trySuccess() and tryFailure(...) as otherwise we can cause an
2404 // IllegalStateException.
2405 // Also we not want to log if the notification happens as this is expected in some cases.
2406 // See https://github.com/netty/netty/issues/5598
2407 PromiseNotifier.cascade(false, future, promise);
2408 }
2409
2410 /**
2411 * Always prefer a direct buffer when it's pooled, so that we reduce the number of memory copies
2412 * in {@link OpenSslEngine}.
2413 */
2414 private ByteBuf allocate(ChannelHandlerContext ctx, int capacity) {
2415 ByteBufAllocator alloc = ctx.alloc();
2416 if (engineType.wantsDirectBuffer) {
2417 return alloc.directBuffer(capacity);
2418 } else {
2419 return alloc.buffer(capacity);
2420 }
2421 }
2422
2423 /**
2424 * Allocates an outbound network buffer for {@link SSLEngine#wrap(ByteBuffer, ByteBuffer)} which can encrypt
2425 * the specified amount of pending bytes.
2426 */
2427 private ByteBuf allocateOutNetBuf(ChannelHandlerContext ctx, int pendingBytes, int numComponents) {
2428 return engineType.allocateWrapBuffer(this, ctx.alloc(), pendingBytes, numComponents);
2429 }
2430
2431 private boolean isStateSet(int bit) {
2432 return (state & bit) == bit;
2433 }
2434
2435 private void setState(int bit) {
2436 state |= bit;
2437 }
2438
2439 private void clearState(int bit) {
2440 state &= ~bit;
2441 }
2442
2443 private final class LazyChannelPromise extends DefaultPromise<Channel> {
2444
2445 @Override
2446 protected EventExecutor executor() {
2447 if (ctx == null) {
2448 throw new IllegalStateException();
2449 }
2450 return ctx.executor();
2451 }
2452
2453 @Override
2454 protected void checkDeadLock() {
2455 if (ctx == null) {
2456 // If ctx is null the handlerAdded(...) callback was not called, in this case the checkDeadLock()
2457 // method was called from another Thread then the one that is used by ctx.executor(). We need to
2458 // guard against this as a user can see a race if handshakeFuture().sync() is called but the
2459 // handlerAdded(..) method was not yet as it is called from the EventExecutor of the
2460 // ChannelHandlerContext. If we not guard against this super.checkDeadLock() would cause an
2461 // IllegalStateException when trying to call executor().
2462 return;
2463 }
2464 super.checkDeadLock();
2465 }
2466 }
2467 }