View Javadoc

    1   /*
    2    * Copyright 2009 Red Hat, Inc.
    3    *
    4    * Red Hat licenses this file to you under the Apache License, version 2.0
    5    * (the "License"); you may not use this file except in compliance with the
    6    * License.  You may obtain a copy of the License at:
    7    *
    8    *    http://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 org.jboss.netty.handler.ssl;
    17
    18  import static org.jboss.netty.channel.Channels.*;
    19
    20  import java.io.IOException;
    21  import java.nio.ByteBuffer;
    22  import java.nio.channels.ClosedChannelException;
    23  import java.util.LinkedList;
    24  import java.util.Queue;
    25  import java.util.concurrent.Executor;
    26  import java.util.concurrent.atomic.AtomicBoolean;
    27  import java.util.regex.Pattern;
    28
    29  import javax.net.ssl.SSLEngine;
    30  import javax.net.ssl.SSLEngineResult;
    31  import javax.net.ssl.SSLEngineResult.HandshakeStatus;
    32  import javax.net.ssl.SSLEngineResult.Status;
    33  import javax.net.ssl.SSLException;
    34
    35  import org.jboss.netty.buffer.ChannelBuffer;
    36  import org.jboss.netty.buffer.ChannelBuffers;
    37  import org.jboss.netty.channel.Channel;
    38  import org.jboss.netty.channel.ChannelDownstreamHandler;
    39  import org.jboss.netty.channel.ChannelEvent;
    40  import org.jboss.netty.channel.ChannelFuture;
    41  import org.jboss.netty.channel.ChannelFutureListener;
    42  import org.jboss.netty.channel.ChannelHandlerContext;
    43  import org.jboss.netty.channel.ChannelPipeline;
    44  import org.jboss.netty.channel.ChannelStateEvent;
    45  import org.jboss.netty.channel.Channels;
    46  import org.jboss.netty.channel.DownstreamMessageEvent;
    47  import org.jboss.netty.channel.ExceptionEvent;
    48  import org.jboss.netty.channel.LifeCycleAwareChannelHandler;
    49  import org.jboss.netty.channel.MessageEvent;
    50  import org.jboss.netty.handler.codec.frame.FrameDecoder;
    51  import org.jboss.netty.logging.InternalLogger;
    52  import org.jboss.netty.logging.InternalLoggerFactory;
    53  import org.jboss.netty.util.internal.LinkedTransferQueue;
    54  import org.jboss.netty.util.internal.NonReentrantLock;
    55
    56  /**
    57   * Adds <a href="http://en.wikipedia.org/wiki/Transport_Layer_Security">SSL
    58   * &middot; TLS</a> and StartTLS support to a {@link Channel}.  Please refer
    59   * to the <strong>"SecureChat"</strong> example in the distribution or the web
    60   * site for the detailed usage.
    61   *
    62   * <h3>Beginning the handshake</h3>
    63   * <p>
    64   * You must make sure not to write a message while the
    65   * {@linkplain #handshake() handshake} is in progress unless you are
    66   * renegotiating.  You will be notified by the {@link ChannelFuture} which is
    67   * returned by the {@link #handshake()} method when the handshake
    68   * process succeeds or fails.
    69   *
    70   * <h3>Renegotiation</h3>
    71   * <p>
    72   * TLS renegotiation has been disabled by default due to a known security issue,
    73   * <a href="http://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2009-3555">CVE-2009-3555</a>.
    74   * You can re-enable renegotiation by calling {@link #setEnableRenegotiation(boolean)}
    75   * with {@code true} at your own risk.
    76   * <p>
    77   * If {@link #isEnableRenegotiation() enableRenegotiation} is {@code true} and
    78   * the initial handshake has been done successfully, you can call
    79   * {@link #handshake()} to trigger the renegotiation.
    80   * <p>
    81   * If {@link #isEnableRenegotiation() enableRenegotiation} is {@code false},
    82   * an attempt to trigger renegotiation will result in the connection closure.
    83   *
    84   * <h3>Closing the session</h3>
    85   * <p>
    86   * To close the SSL session, the {@link #close()} method should be
    87   * called to send the {@code close_notify} message to the remote peer.  One
    88   * exception is when you close the {@link Channel} - {@link SslHandler}
    89   * intercepts the close request and send the {@code close_notify} message
    90   * before the channel closure automatically.  Once the SSL session is closed,
    91   * it is not reusable, and consequently you should create a new
    92   * {@link SslHandler} with a new {@link SSLEngine} as explained in the
    93   * following section.
    94   *
    95   * <h3>Restarting the session</h3>
    96   * <p>
    97   * To restart the SSL session, you must remove the existing closed
    98   * {@link SslHandler} from the {@link ChannelPipeline}, insert a new
    99   * {@link SslHandler} with a new {@link SSLEngine} into the pipeline,
    100  * and start the handshake process as described in the first section.
    101  *
    102  * <h3>Implementing StartTLS</h3>
    103  * <p>
    104  * <a href="http://en.wikipedia.org/wiki/STARTTLS">StartTLS</a> is the
    105  * communication pattern that secures the wire in the middle of the plaintext
    106  * connection.  Please note that it is different from SSL &middot; TLS, that
    107  * secures the wire from the beginning of the connection.  Typically, StartTLS
    108  * is composed of three steps:
    109  * <ol>
    110  * <li>Client sends a StartTLS request to server.</li>
    111  * <li>Server sends a StartTLS response to client.</li>
    112  * <li>Client begins SSL handshake.</li>
    113  * </ol>
    114  * If you implement a server, you need to:
    115  * <ol>
    116  * <li>create a new {@link SslHandler} instance with {@code startTls} flag set
    117  *     to {@code true},</li>
    118  * <li>insert the {@link SslHandler} to the {@link ChannelPipeline}, and</li>
    119  * <li>write a StartTLS response.</li>
    120  * </ol>
    121  * Please note that you must insert {@link SslHandler} <em>before</em> sending
    122  * the StartTLS response.  Otherwise the client can send begin SSL handshake
    123  * before {@link SslHandler} is inserted to the {@link ChannelPipeline}, causing
    124  * data corruption.
    125  * <p>
    126  * The client-side implementation is much simpler.
    127  * <ol>
    128  * <li>Write a StartTLS request,</li>
    129  * <li>wait for the StartTLS response,</li>
    130  * <li>create a new {@link SslHandler} instance with {@code startTls} flag set
    131  *     to {@code false},</li>
    132  * <li>insert the {@link SslHandler} to the {@link ChannelPipeline}, and</li>
    133  * <li>Initiate SSL handshake by calling {@link SslHandler#handshake()}.</li>
    134  * </ol>
    135  *
    136  * @author <a href="http://www.jboss.org/netty/">The Netty Project</a>
    137  * @author <a href="http://gleamynode.net/">Trustin Lee</a>
    138  *
    139  * @version $Rev: 2369 $, $Date: 2010-10-19 13:05:28 +0900 (Tue, 19 Oct 2010) $
    140  *
    141  * @apiviz.landmark
    142  * @apiviz.uses org.jboss.netty.handler.ssl.SslBufferPool
    143  */
    144 public class SslHandler extends FrameDecoder
    145                         implements ChannelDownstreamHandler,
    146                                    LifeCycleAwareChannelHandler {
    147
    148     private static final InternalLogger logger =
    149         InternalLoggerFactory.getInstance(SslHandler.class);
    150
    151     private static final ByteBuffer EMPTY_BUFFER = ByteBuffer.allocate(0);
    152
    153     private static final Pattern IGNORABLE_ERROR_MESSAGE = Pattern.compile(
    154             "^.*(?:connection.*reset|connection.*closed|broken.*pipe).*$",
    155             Pattern.CASE_INSENSITIVE);
    156
    157     private static SslBufferPool defaultBufferPool;
    158
    159     /**
    160      * Returns the default {@link SslBufferPool} used when no pool is
    161      * specified in the constructor.
    162      */
    163     public static synchronized SslBufferPool getDefaultBufferPool() {
    164         if (defaultBufferPool == null) {
    165             defaultBufferPool = new SslBufferPool();
    166         }
    167         return defaultBufferPool;
    168     }
    169
    170     private volatile ChannelHandlerContext ctx;
    171     private final SSLEngine engine;
    172     private final SslBufferPool bufferPool;
    173     private final Executor delegatedTaskExecutor;
    174     private final boolean startTls;
    175
    176     private volatile boolean enableRenegotiation;
    177
    178     final Object handshakeLock = new Object();
    179     private boolean handshaking;
    180     private volatile boolean handshaken;
    181     private volatile ChannelFuture handshakeFuture;
    182
    183     private final AtomicBoolean sentFirstMessage = new AtomicBoolean();
    184     private final AtomicBoolean sentCloseNotify = new AtomicBoolean();
    185     int ignoreClosedChannelException;
    186     final Object ignoreClosedChannelExceptionLock = new Object();
    187     private final Queue<PendingWrite> pendingUnencryptedWrites = new LinkedList<PendingWrite>();
    188     private final Queue<MessageEvent> pendingEncryptedWrites = new LinkedTransferQueue<MessageEvent>();
    189     private final NonReentrantLock pendingEncryptedWritesLock = new NonReentrantLock();
    190
    191     /**
    192      * Creates a new instance.
    193      *
    194      * @param engine  the {@link SSLEngine} this handler will use
    195      */
    196     public SslHandler(SSLEngine engine) {
    197         this(engine, getDefaultBufferPool(), ImmediateExecutor.INSTANCE);
    198     }
    199
    200     /**
    201      * Creates a new instance.
    202      *
    203      * @param engine      the {@link SSLEngine} this handler will use
    204      * @param bufferPool  the {@link SslBufferPool} where this handler will
    205      *                    acquire the buffers required by the {@link SSLEngine}
    206      */
    207     public SslHandler(SSLEngine engine, SslBufferPool bufferPool) {
    208         this(engine, bufferPool, ImmediateExecutor.INSTANCE);
    209     }
    210
    211     /**
    212      * Creates a new instance.
    213      *
    214      * @param engine    the {@link SSLEngine} this handler will use
    215      * @param startTls  {@code true} if the first write request shouldn't be
    216      *                  encrypted by the {@link SSLEngine}
    217      */
    218     public SslHandler(SSLEngine engine, boolean startTls) {
    219         this(engine, getDefaultBufferPool(), startTls);
    220     }
    221
    222     /**
    223      * Creates a new instance.
    224      *
    225      * @param engine      the {@link SSLEngine} this handler will use
    226      * @param bufferPool  the {@link SslBufferPool} where this handler will
    227      *                    acquire the buffers required by the {@link SSLEngine}
    228      * @param startTls    {@code true} if the first write request shouldn't be
    229      *                    encrypted by the {@link SSLEngine}
    230      */
    231     public SslHandler(SSLEngine engine, SslBufferPool bufferPool, boolean startTls) {
    232         this(engine, bufferPool, startTls, ImmediateExecutor.INSTANCE);
    233     }
    234
    235     /**
    236      * Creates a new instance.
    237      *
    238      * @param engine
    239      *        the {@link SSLEngine} this handler will use
    240      * @param delegatedTaskExecutor
    241      *        the {@link Executor} which will execute the delegated task
    242      *        that {@link SSLEngine#getDelegatedTask()} will return
    243      */
    244     public SslHandler(SSLEngine engine, Executor delegatedTaskExecutor) {
    245         this(engine, getDefaultBufferPool(), delegatedTaskExecutor);
    246     }
    247
    248     /**
    249      * Creates a new instance.
    250      *
    251      * @param engine
    252      *        the {@link SSLEngine} this handler will use
    253      * @param bufferPool
    254      *        the {@link SslBufferPool} where this handler will acquire
    255      *        the buffers required by the {@link SSLEngine}
    256      * @param delegatedTaskExecutor
    257      *        the {@link Executor} which will execute the delegated task
    258      *        that {@link SSLEngine#getDelegatedTask()} will return
    259      */
    260     public SslHandler(SSLEngine engine, SslBufferPool bufferPool, Executor delegatedTaskExecutor) {
    261         this(engine, bufferPool, false, delegatedTaskExecutor);
    262     }
    263
    264     /**
    265      * Creates a new instance.
    266      *
    267      * @param engine
    268      *        the {@link SSLEngine} this handler will use
    269      * @param startTls
    270      *        {@code true} if the first write request shouldn't be encrypted
    271      *        by the {@link SSLEngine}
    272      * @param delegatedTaskExecutor
    273      *        the {@link Executor} which will execute the delegated task
    274      *        that {@link SSLEngine#getDelegatedTask()} will return
    275      */
    276     public SslHandler(SSLEngine engine, boolean startTls, Executor delegatedTaskExecutor) {
    277         this(engine, getDefaultBufferPool(), startTls, delegatedTaskExecutor);
    278     }
    279
    280     /**
    281      * Creates a new instance.
    282      *
    283      * @param engine
    284      *        the {@link SSLEngine} this handler will use
    285      * @param bufferPool
    286      *        the {@link SslBufferPool} where this handler will acquire
    287      *        the buffers required by the {@link SSLEngine}
    288      * @param startTls
    289      *        {@code true} if the first write request shouldn't be encrypted
    290      *        by the {@link SSLEngine}
    291      * @param delegatedTaskExecutor
    292      *        the {@link Executor} which will execute the delegated task
    293      *        that {@link SSLEngine#getDelegatedTask()} will return
    294      */
    295     public SslHandler(SSLEngine engine, SslBufferPool bufferPool, boolean startTls, Executor delegatedTaskExecutor) {
    296         if (engine == null) {
    297             throw new NullPointerException("engine");
    298         }
    299         if (bufferPool == null) {
    300             throw new NullPointerException("bufferPool");
    301         }
    302         if (delegatedTaskExecutor == null) {
    303             throw new NullPointerException("delegatedTaskExecutor");
    304         }
    305         this.engine = engine;
    306         this.bufferPool = bufferPool;
    307         this.delegatedTaskExecutor = delegatedTaskExecutor;
    308         this.startTls = startTls;
    309     }
    310
    311     /**
    312      * Returns the {@link SSLEngine} which is used by this handler.
    313      */
    314     public SSLEngine getEngine() {
    315         return engine;
    316     }
    317
    318     /**
    319      * Starts an SSL / TLS handshake for the specified channel.
    320      *
    321      * @return a {@link ChannelFuture} which is notified when the handshake
    322      *         succeeds or fails.
    323      */
    324     public ChannelFuture handshake() {
    325         if (handshaken && !isEnableRenegotiation()) {
    326             throw new IllegalStateException("renegotiation disabled");
    327         }
    328
    329         ChannelHandlerContext ctx = this.ctx;
    330         Channel channel = ctx.getChannel();
    331         ChannelFuture handshakeFuture;
    332         synchronized (handshakeLock) {
    333             if (handshaking) {
    334                 return this.handshakeFuture;
    335             } else {
    336                 handshaking = true;
    337                 try {
    338                     engine.beginHandshake();
    339                     runDelegatedTasks();
    340                     handshakeFuture = this.handshakeFuture = future(channel);
    341                 } catch (SSLException e) {
    342                     handshakeFuture = this.handshakeFuture = failedFuture(channel, e);
    343                 }
    344             }
    345         }
    346
    347         try {
    348             wrapNonAppData(ctx, channel);
    349         } catch (SSLException e) {
    350             handshakeFuture.setFailure(e);
    351         }
    352         return handshakeFuture;
    353     }
    354
    355     /**
    356      * @deprecated Use {@link #handshake()} instead.
    357      */
    358     @Deprecated
    359     public ChannelFuture handshake(@SuppressWarnings("unused") Channel channel) {
    360         return handshake();
    361     }
    362
    363     /**
    364      * Sends an SSL {@code close_notify} message to the specified channel and
    365      * destroys the underlying {@link SSLEngine}.
    366      */
    367     public ChannelFuture close() {
    368         ChannelHandlerContext ctx = this.ctx;
    369         Channel channel = ctx.getChannel();
    370         try {
    371             engine.closeOutbound();
    372             return wrapNonAppData(ctx, channel);
    373         } catch (SSLException e) {
    374             return failedFuture(channel, e);
    375         }
    376     }
    377
    378     /**
    379      * @deprecated Use {@link #close()} instead.
    380      */
    381     @Deprecated
    382     public ChannelFuture close(@SuppressWarnings("unused") Channel channel) {
    383         return close();
    384     }
    385
    386     /**
    387      * Returns {@code true} if and only if TLS renegotiation is enabled.
    388      */
    389     public boolean isEnableRenegotiation() {
    390         return enableRenegotiation;
    391     }
    392
    393     /**
    394      * Enables or disables TLS renegotiation.
    395      */
    396     public void setEnableRenegotiation(boolean enableRenegotiation) {
    397         this.enableRenegotiation = enableRenegotiation;
    398     }
    399
    400     public void handleDownstream(
    401             final ChannelHandlerContext context, final ChannelEvent evt) throws Exception {
    402         if (evt instanceof ChannelStateEvent) {
    403             ChannelStateEvent e = (ChannelStateEvent) evt;
    404             switch (e.getState()) {
    405             case OPEN:
    406             case CONNECTED:
    407             case BOUND:
    408                 if (Boolean.FALSE.equals(e.getValue()) || e.getValue() == null) {
    409                     closeOutboundAndChannel(context, e);
    410                     return;
    411                 }
    412             }
    413         }
    414         if (!(evt instanceof MessageEvent)) {
    415             context.sendDownstream(evt);
    416             return;
    417         }
    418
    419         MessageEvent e = (MessageEvent) evt;
    420         if (!(e.getMessage() instanceof ChannelBuffer)) {
    421             context.sendDownstream(evt);
    422             return;
    423         }
    424
    425         // Do not encrypt the first write request if this handler is
    426         // created with startTLS flag turned on.
    427         if (startTls && sentFirstMessage.compareAndSet(false, true)) {
    428             context.sendDownstream(evt);
    429             return;
    430         }
    431
    432         // Otherwise, all messages are encrypted.
    433         ChannelBuffer msg = (ChannelBuffer) e.getMessage();
    434         PendingWrite pendingWrite;
    435
    436         if (msg.readable()) {
    437             pendingWrite = new PendingWrite(evt.getFuture(), msg.toByteBuffer(msg.readerIndex(), msg.readableBytes()));
    438         } else {
    439             pendingWrite = new PendingWrite(evt.getFuture(), null);
    440         }
    441         synchronized (pendingUnencryptedWrites) {
    442             boolean offered = pendingUnencryptedWrites.offer(pendingWrite);
    443             assert offered;
    444         }
    445
    446         wrap(context, evt.getChannel());
    447     }
    448
    449     @Override
    450     public void channelDisconnected(ChannelHandlerContext ctx,
    451             ChannelStateEvent e) throws Exception {
    452
    453         // Make sure the handshake future is notified when a connection has
    454         // been closed during handshake.
    455         synchronized (handshakeLock) {
    456             if (handshaking) {
    457                 handshakeFuture.setFailure(new ClosedChannelException());
    458             }
    459         }
    460
    461         try {
    462             super.channelDisconnected(ctx, e);
    463         } finally {
    464             unwrap(ctx, e.getChannel(), ChannelBuffers.EMPTY_BUFFER, 0, 0);
    465             engine.closeOutbound();
    466             if (!sentCloseNotify.get() && handshaken) {
    467                 try {
    468                     engine.closeInbound();
    469                 } catch (SSLException ex) {
    470                     logger.debug("Failed to clean up SSLEngine.", ex);
    471                 }
    472             }
    473         }
    474     }
    475
    476     @Override
    477     public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)
    478             throws Exception {
    479
    480         Throwable cause = e.getCause();
    481         if (cause instanceof IOException) {
    482             if (cause instanceof ClosedChannelException) {
    483                 synchronized (ignoreClosedChannelExceptionLock) {
    484                     if (ignoreClosedChannelException > 0) {
    485                         ignoreClosedChannelException --;
    486                         logger.debug(
    487                                 "Swallowing an exception raised while " +
    488                                 "writing non-app data", cause);
    489                         return;
    490                     }
    491                 }
    492             } else if (engine.isOutboundDone()) {
    493                 String message = String.valueOf(cause.getMessage()).toLowerCase();
    494                 if (IGNORABLE_ERROR_MESSAGE.matcher(message).matches()) {
    495                     // It is safe to ignore the 'connection reset by peer' or
    496                     // 'broken pipe' error after sending closure_notify.
    497                     logger.debug(
    498                             "Swallowing a 'connection reset by peer / " +
    499                             "broken pipe' error occurred while writing " +
    500                             "'closure_notify'", cause);
    501
    502                     // Close the connection explicitly just in case the transport
    503                     // did not close the connection automatically.
    504                     Channels.close(ctx, succeededFuture(e.getChannel()));
    505                     return;
    506                 }
    507             }
    508         }
    509
    510         ctx.sendUpstream(e);
    511     }
    512
    513     @Override
    514     protected Object decode(
    515             final ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer) throws Exception {
    516
    517         if (buffer.readableBytes() < 5) {
    518             return null;
    519         }
    520
    521         int packetLength = 0;
    522
    523         // SSLv3 or TLS - Check ContentType
    524         boolean tls;
    525         switch (buffer.getUnsignedByte(buffer.readerIndex())) {
    526         case 20:  // change_cipher_spec
    527         case 21:  // alert
    528         case 22:  // handshake
    529         case 23:  // application_data
    530             tls = true;
    531             break;
    532         default:
    533             // SSLv2 or bad data
    534             tls = false;
    535         }
    536
    537         if (tls) {
    538             // SSLv3 or TLS - Check ProtocolVersion
    539             int majorVersion = buffer.getUnsignedByte(buffer.readerIndex() + 1);
    540             if (majorVersion >= 3 && majorVersion < 10) {
    541                 // SSLv3 or TLS
    542                 packetLength = (buffer.getShort(buffer.readerIndex() + 3) & 0xFFFF) + 5;
    543                 if (packetLength <= 5) {
    544                     // Neither SSLv2 or TLSv1 (i.e. SSLv2 or bad data)
    545                     tls = false;
    546                 }
    547             } else {
    548                 // Neither SSLv2 or TLSv1 (i.e. SSLv2 or bad data)
    549                 tls = false;
    550             }
    551         }
    552
    553         if (!tls) {
    554             // SSLv2 or bad data - Check the version
    555             boolean sslv2 = true;
    556             int headerLength = (buffer.getUnsignedByte(
    557                     buffer.readerIndex()) & 0x80) != 0 ? 2 : 3;
    558             int majorVersion = buffer.getUnsignedByte(
    559                     buffer.readerIndex() + headerLength + 1);
    560             if (majorVersion >= 2 && majorVersion < 10) {
    561                 // SSLv2
    562                 if (headerLength == 2) {
    563                     packetLength = (buffer.getShort(buffer.readerIndex()) & 0x7FFF) + 2;
    564                 } else {
    565                     packetLength = (buffer.getShort(buffer.readerIndex()) & 0x3FFF) + 3;
    566                 }
    567                 if (packetLength <= headerLength) {
    568                     sslv2 = false;
    569                 }
    570             } else {
    571                 sslv2 = false;
    572             }
    573
    574             if (!sslv2) {
    575                 // Bad data - discard the buffer and raise an exception.
    576                 SSLException e = new SSLException(
    577                         "not an SSL/TLS record: " + ChannelBuffers.hexDump(buffer));
    578                 buffer.skipBytes(buffer.readableBytes());
    579                 throw e;
    580             }
    581         }
    582
    583         assert packetLength > 0;
    584
    585         if (buffer.readableBytes() < packetLength) {
    586             return null;
    587         }
    588
    589         // We advance the buffer's readerIndex before calling unwrap() because
    590         // unwrap() can trigger FrameDecoder call decode(), this method, recursively.
    591         // The recursive call results in decoding the same packet twice if
    592         // the readerIndex is advanced *after* decode().
    593         //
    594         // Here's an example:
    595         // 1) An SSL packet is received from the wire.
    596         // 2) SslHandler.decode() deciphers the packet and calls the user code.
    597         // 3) The user closes the channel in the same thread.
    598         // 4) The same thread triggers a channelDisconnected() event.
    599         // 5) FrameDecoder.cleanup() is called, and it calls SslHandler.decode().
    600         // 6) SslHandler.decode() will feed the same packet with what was
    601         //    deciphered at the step 2 again if the readerIndex was not advanced
    602         //    before calling the user code.
    603         final int packetOffset = buffer.readerIndex();
    604         buffer.skipBytes(packetLength);
    605         return unwrap(ctx, channel, buffer, packetOffset, packetLength);
    606     }
    607
    608     private ChannelFuture wrap(ChannelHandlerContext context, Channel channel)
    609             throws SSLException {
    610
    611         ChannelFuture future = null;
    612         ChannelBuffer msg;
    613         ByteBuffer outNetBuf = bufferPool.acquire();
    614         boolean success = true;
    615         boolean offered = false;
    616         boolean needsUnwrap = false;
    617         try {
    618             loop:
    619             for (;;) {
    620                 // Acquire a lock to make sure unencrypted data is polled
    621                 // in order and their encrypted counterpart is offered in
    622                 // order.
    623                 synchronized (pendingUnencryptedWrites) {
    624                     PendingWrite pendingWrite = pendingUnencryptedWrites.peek();
    625                     if (pendingWrite == null) {
    626                         break;
    627                     }
    628
    629                     ByteBuffer outAppBuf = pendingWrite.outAppBuf;
    630                     if (outAppBuf == null) {
    631                         // A write request with an empty buffer
    632                         pendingUnencryptedWrites.remove();
    633                         offerEncryptedWriteRequest(
    634                                 new DownstreamMessageEvent(
    635                                         channel, pendingWrite.future,
    636                                         ChannelBuffers.EMPTY_BUFFER,
    637                                         channel.getRemoteAddress()));
    638                         offered = true;
    639                     } else {
    640                         SSLEngineResult result = null;
    641                         try {
    642                             synchronized (handshakeLock) {
    643                                 result = engine.wrap(outAppBuf, outNetBuf);
    644                             }
    645                         } finally {
    646                             if (!outAppBuf.hasRemaining()) {
    647                                 pendingUnencryptedWrites.remove();
    648                             }
    649                         }
    650
    651                         if (result.bytesProduced() > 0) {
    652                             outNetBuf.flip();
    653                             msg = ChannelBuffers.buffer(outNetBuf.remaining());
    654                             msg.writeBytes(outNetBuf.array(), 0, msg.capacity());
    655                             outNetBuf.clear();
    656
    657                             if (pendingWrite.outAppBuf.hasRemaining()) {
    658                                 // pendingWrite's future shouldn't be notified if
    659                                 // only partial data is written.
    660                                 future = succeededFuture(channel);
    661                             } else {
    662                                 future = pendingWrite.future;
    663                             }
    664
    665                             MessageEvent encryptedWrite = new DownstreamMessageEvent(
    666                                     channel, future, msg, channel.getRemoteAddress());
    667                             offerEncryptedWriteRequest(encryptedWrite);
    668                             offered = true;
    669                         } else {
    670                             final HandshakeStatus handshakeStatus = result.getHandshakeStatus();
    671                             handleRenegotiation(handshakeStatus);
    672                             switch (handshakeStatus) {
    673                             case NEED_WRAP:
    674                                 if (outAppBuf.hasRemaining()) {
    675                                     break;
    676                                 } else {
    677                                     break loop;
    678                                 }
    679                             case NEED_UNWRAP:
    680                                 needsUnwrap = true;
    681                                 break loop;
    682                             case NEED_TASK:
    683                                 runDelegatedTasks();
    684                                 break;
    685                             case FINISHED:
    686                             case NOT_HANDSHAKING:
    687                                 if (handshakeStatus == HandshakeStatus.FINISHED) {
    688                                     setHandshakeSuccess(channel);
    689                                 }
    690                                 if (result.getStatus() == Status.CLOSED) {
    691                                     success = false;
    692                                 }
    693                                 break loop;
    694                             default:
    695                                 throw new IllegalStateException(
    696                                         "Unknown handshake status: " +
    697                                         handshakeStatus);
    698                             }
    699                         }
    700                     }
    701                 }
    702             }
    703         } catch (SSLException e) {
    704             success = false;
    705             setHandshakeFailure(channel, e);
    706             throw e;
    707         } finally {
    708             bufferPool.release(outNetBuf);
    709
    710             if (offered) {
    711                 flushPendingEncryptedWrites(context);
    712             }
    713
    714             if (!success) {
    715                 IllegalStateException cause =
    716                     new IllegalStateException("SSLEngine already closed");
    717                 // Mark all remaining pending writes as failure if anything
    718                 // wrong happened before the write requests are wrapped.
    719                 // Please note that we do not call setFailure while a lock is
    720                 // acquired, to avoid a potential dead lock.
    721                 for (;;) {
    722                     PendingWrite pendingWrite;
    723                     synchronized (pendingUnencryptedWrites) {
    724                         pendingWrite = pendingUnencryptedWrites.poll();
    725                         if (pendingWrite == null) {
    726                             break;
    727                         }
    728                     }
    729
    730                     pendingWrite.future.setFailure(cause);
    731                 }
    732             }
    733         }
    734
    735         if (needsUnwrap) {
    736             unwrap(context, channel, ChannelBuffers.EMPTY_BUFFER, 0, 0);
    737         }
    738
    739         if (future == null) {
    740             future = succeededFuture(channel);
    741         }
    742         return future;
    743     }
    744
    745     private void offerEncryptedWriteRequest(MessageEvent encryptedWrite) {
    746         final boolean locked = pendingEncryptedWritesLock.tryLock();
    747         try {
    748             pendingEncryptedWrites.offer(encryptedWrite);
    749         } finally {
    750             if (locked) {
    751                 pendingEncryptedWritesLock.unlock();
    752             }
    753         }
    754     }
    755
    756     private void flushPendingEncryptedWrites(ChannelHandlerContext ctx) {
    757         // Avoid possible dead lock and data integrity issue
    758         // which is caused by cross communication between more than one channel
    759         // in the same VM.
    760         if (!pendingEncryptedWritesLock.tryLock()) {
    761             return;
    762         }
    763
    764         try {
    765             MessageEvent e;
    766             while ((e = pendingEncryptedWrites.poll()) != null) {
    767                 ctx.sendDownstream(e);
    768             }
    769         } finally {
    770             pendingEncryptedWritesLock.unlock();
    771         }
    772     }
    773
    774     private ChannelFuture wrapNonAppData(ChannelHandlerContext ctx, Channel channel) throws SSLException {
    775         ChannelFuture future = null;
    776         ByteBuffer outNetBuf = bufferPool.acquire();
    777
    778         SSLEngineResult result;
    779         try {
    780             for (;;) {
    781                 synchronized (handshakeLock) {
    782                     result = engine.wrap(EMPTY_BUFFER, outNetBuf);
    783                 }
    784
    785                 if (result.bytesProduced() > 0) {
    786                     outNetBuf.flip();
    787                     ChannelBuffer msg = ChannelBuffers.buffer(outNetBuf.remaining());
    788                     msg.writeBytes(outNetBuf.array(), 0, msg.capacity());
    789                     outNetBuf.clear();
    790
    791                     future = future(channel);
    792                     future.addListener(new ChannelFutureListener() {
    793                         public void operationComplete(ChannelFuture future)
    794                                 throws Exception {
    795                             if (future.getCause() instanceof ClosedChannelException) {
    796                                 synchronized (ignoreClosedChannelExceptionLock) {
    797                                     ignoreClosedChannelException ++;
    798                                 }
    799                             }
    800                         }
    801                     });
    802
    803                     write(ctx, future, msg);
    804                 }
    805
    806                 final HandshakeStatus handshakeStatus = result.getHandshakeStatus();
    807                 handleRenegotiation(handshakeStatus);
    808                 switch (handshakeStatus) {
    809                 case FINISHED:
    810                     setHandshakeSuccess(channel);
    811                     runDelegatedTasks();
    812                     break;
    813                 case NEED_TASK:
    814                     runDelegatedTasks();
    815                     break;
    816                 case NEED_UNWRAP:
    817                     if (!Thread.holdsLock(handshakeLock)) {
    818                         // unwrap shouldn't be called when this method was
    819                         // called by unwrap - unwrap will keep running after
    820                         // this method returns.
    821                         unwrap(ctx, channel, ChannelBuffers.EMPTY_BUFFER, 0, 0);
    822                     }
    823                     break;
    824                 case NOT_HANDSHAKING:
    825                 case NEED_WRAP:
    826                     break;
    827                 default:
    828                     throw new IllegalStateException(
    829                             "Unexpected handshake status: " + handshakeStatus);
    830                 }
    831
    832                 if (result.bytesProduced() == 0) {
    833                     break;
    834                 }
    835             }
    836         } catch (SSLException e) {
    837             setHandshakeFailure(channel, e);
    838             throw e;
    839         } finally {
    840             bufferPool.release(outNetBuf);
    841         }
    842
    843         if (future == null) {
    844             future = succeededFuture(channel);
    845         }
    846
    847         return future;
    848     }
    849
    850     private ChannelBuffer unwrap(
    851             ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer, int offset, int length) throws SSLException {
    852         ByteBuffer inNetBuf = buffer.toByteBuffer(offset, length);
    853         ByteBuffer outAppBuf = bufferPool.acquire();
    854
    855         try {
    856             boolean needsWrap = false;
    857             loop:
    858             for (;;) {
    859                 SSLEngineResult result;
    860                 synchronized (handshakeLock) {
    861                     if (!handshaken && !handshaking &&
    862                         !engine.getUseClientMode() &&
    863                         !engine.isInboundDone() && !engine.isOutboundDone()) {
    864                         handshake();
    865                     }
    866
    867                     try {
    868                         result = engine.unwrap(inNetBuf, outAppBuf);
    869                     } catch (SSLException e) {
    870                         throw e;
    871                     }
    872
    873                     final HandshakeStatus handshakeStatus = result.getHandshakeStatus();
    874                     handleRenegotiation(handshakeStatus);
    875                     switch (handshakeStatus) {
    876                     case NEED_UNWRAP:
    877                         if (inNetBuf.hasRemaining() && !engine.isInboundDone()) {
    878                             break;
    879                         } else {
    880                             break loop;
    881                         }
    882                     case NEED_WRAP:
    883                         wrapNonAppData(ctx, channel);
    884                         break;
    885                     case NEED_TASK:
    886                         runDelegatedTasks();
    887                         break;
    888                     case FINISHED:
    889                         setHandshakeSuccess(channel);
    890                         needsWrap = true;
    891                         break loop;
    892                     case NOT_HANDSHAKING:
    893                         needsWrap = true;
    894                         break loop;
    895                     default:
    896                         throw new IllegalStateException(
    897                                 "Unknown handshake status: " + handshakeStatus);
    898                     }
    899                 }
    900             }
    901
    902             if (needsWrap) {
    903                 // wrap() acquires pendingUnencryptedWrites first and then
    904                 // handshakeLock.  If handshakeLock is already hold by the
    905                 // current thread, calling wrap() will lead to a dead lock
    906                 // i.e. pendingUnencryptedWrites -> handshakeLock vs.
    907                 //      handshakeLock -> pendingUnencryptedLock -> handshakeLock
    908                 //
    909                 // There is also a same issue between pendingEncryptedWrites
    910                 // and pendingUnencryptedWrites.
    911                 if (!Thread.holdsLock(handshakeLock) &&
    912                     !pendingEncryptedWritesLock.isHeldByCurrentThread()) {
    913                     wrap(ctx, channel);
    914                 }
    915             }
    916
    917             outAppBuf.flip();
    918
    919             if (outAppBuf.hasRemaining()) {
    920                 ChannelBuffer frame = ChannelBuffers.buffer(outAppBuf.remaining());
    921                 frame.writeBytes(outAppBuf.array(), 0, frame.capacity());
    922                 return frame;
    923             } else {
    924                 return null;
    925             }
    926         } catch (SSLException e) {
    927             setHandshakeFailure(channel, e);
    928             throw e;
    929         } finally {
    930             bufferPool.release(outAppBuf);
    931         }
    932     }
    933
    934     private void handleRenegotiation(HandshakeStatus handshakeStatus) {
    935         if (handshakeStatus == HandshakeStatus.NOT_HANDSHAKING ||
    936             handshakeStatus == HandshakeStatus.FINISHED) {
    937             // Not handshaking
    938             return;
    939         }
    940
    941         if (!handshaken) {
    942             // Not renegotiation
    943             return;
    944         }
    945
    946         final boolean renegotiate;
    947         synchronized (handshakeLock) {
    948             if (handshaking) {
    949                 // Renegotiation in progress or failed already.
    950                 // i.e. Renegotiation check has been done already below.
    951                 return;
    952             }
    953
    954             if (engine.isInboundDone() || engine.isOutboundDone()) {
    955                 // Not handshaking but closing.
    956                 return;
    957             }
    958
    959             if (isEnableRenegotiation()) {
    960                 // Continue renegotiation.
    961                 renegotiate = true;
    962             } else {
    963                 // Do not renegotiate.
    964                 renegotiate = false;
    965                 // Prevent reentrance of this method.
    966                 handshaking = true;
    967             }
    968         }
    969
    970         if (renegotiate) {
    971             // Renegotiate.
    972             handshake();
    973         } else {
    974             // Raise an exception.
    975             fireExceptionCaught(
    976                     ctx, new SSLException(
    977                             "renegotiation attempted by peer; " +
    978                             "closing the connection"));
    979
    980             // Close the connection to stop renegotiation.
    981             Channels.close(ctx, succeededFuture(ctx.getChannel()));
    982         }
    983     }
    984
    985     private void runDelegatedTasks() {
    986         for (;;) {
    987             final Runnable task;
    988             synchronized (handshakeLock) {
    989                 task = engine.getDelegatedTask();
    990             }
    991
    992             if (task == null) {
    993                 break;
    994             }
    995
    996             delegatedTaskExecutor.execute(new Runnable() {
    997                 public void run() {
    998                     synchronized (handshakeLock) {
    999                         task.run();
    1000                     }
    1001                 }
    1002             });
    1003         }
    1004     }
    1005
    1006     private void setHandshakeSuccess(Channel channel) {
    1007         synchronized (handshakeLock) {
    1008             handshaking = false;
    1009             handshaken = true;
    1010
    1011             if (handshakeFuture == null) {
    1012                 handshakeFuture = future(channel);
    1013             }
    1014         }
    1015
    1016         handshakeFuture.setSuccess();
    1017     }
    1018
    1019     private void setHandshakeFailure(Channel channel, SSLException cause) {
    1020         synchronized (handshakeLock) {
    1021             if (!handshaking) {
    1022                 return;
    1023             }
    1024             handshaking = false;
    1025             handshaken = false;
    1026
    1027             if (handshakeFuture == null) {
    1028                 handshakeFuture = future(channel);
    1029             }
    1030         }
    1031         handshakeFuture.setFailure(cause);
    1032     }
    1033
    1034     private void closeOutboundAndChannel(
    1035             final ChannelHandlerContext context, final ChannelStateEvent e) throws SSLException {
    1036         if (!e.getChannel().isConnected()) {
    1037             context.sendDownstream(e);
    1038             return;
    1039         }
    1040
    1041         unwrap(context, e.getChannel(), ChannelBuffers.EMPTY_BUFFER, 0, 0);
    1042         if (!engine.isInboundDone()) {
    1043             if (sentCloseNotify.compareAndSet(false, true)) {
    1044                 engine.closeOutbound();
    1045                 ChannelFuture closeNotifyFuture = wrapNonAppData(context, e.getChannel());
    1046                 closeNotifyFuture.addListener(
    1047                         new ClosingChannelFutureListener(context, e));
    1048                 return;
    1049             }
    1050         }
    1051
    1052         context.sendDownstream(e);
    1053     }
    1054
    1055     private static final class PendingWrite {
    1056         final ChannelFuture future;
    1057         final ByteBuffer outAppBuf;
    1058
    1059         PendingWrite(ChannelFuture future, ByteBuffer outAppBuf) {
    1060             this.future = future;
    1061             this.outAppBuf = outAppBuf;
    1062         }
    1063     }
    1064
    1065     private static final class ClosingChannelFutureListener implements ChannelFutureListener {
    1066
    1067         private final ChannelHandlerContext context;
    1068         private final ChannelStateEvent e;
    1069
    1070         ClosingChannelFutureListener(
    1071                 ChannelHandlerContext context, ChannelStateEvent e) {
    1072             this.context = context;
    1073             this.e = e;
    1074         }
    1075
    1076         public void operationComplete(ChannelFuture closeNotifyFuture) throws Exception {
    1077             if (!(closeNotifyFuture.getCause() instanceof ClosedChannelException)) {
    1078                 Channels.close(context, e.getFuture());
    1079             }
    1080         }
    1081     }
    1082
    1083     public void beforeAdd(ChannelHandlerContext ctx) throws Exception {
    1084         this.ctx = ctx;
    1085     }
    1086
    1087     public void afterAdd(ChannelHandlerContext ctx) throws Exception {
    1088         // Unused
    1089     }
    1090
    1091     public void beforeRemove(ChannelHandlerContext ctx) throws Exception {
    1092         // Unused
    1093     }
    1094
    1095     public void afterRemove(ChannelHandlerContext ctx) throws Exception {
    1096         // Unused
    1097     }
    1098 }