NETWORKING: Add SSL Handler before other Handlers (#34636)

* NETWORKING: Add SSL Handler before other Handlers

* The only way to run into the issue in #33998 is for `Netty4MessageChannelHandler`
to be in the pipeline while the SslHandler is not. Adding the SslHandler before any
other handlers should ensure correct ordering here even when we handle upstream events
in our own thread pool
* Ensure that channels that were closed concurrently don't trip the assertion
* Closes #33998

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/transport/netty4/SecurityNetty4Transport.java

@@ -157,11 +157,12 @@ public class SecurityNetty4Transport extends Netty4Transport {
 
         @Override
         protected void initChannel(Channel ch) throws Exception {
-            super.initChannel(ch);
             SSLEngine serverEngine = sslService.createSSLEngine(configuration, null, -1);
             serverEngine.setUseClientMode(false);
             final SslHandler sslHandler = new SslHandler(serverEngine);
             ch.pipeline().addFirst("sslhandler", sslHandler);
+            super.initChannel(ch);
+            assert ch.pipeline().first() == sslHandler : "SSL handler must be first handler in pipeline";
         }
     }
 

x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/transport/SSLEngineUtils.java

@@ -6,6 +6,7 @@
 package org.elasticsearch.xpack.security.transport;
 
 import io.netty.channel.Channel;
+import io.netty.channel.ChannelException;
 import io.netty.handler.ssl.SslHandler;
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
@@ -59,7 +60,13 @@ public class SSLEngineUtils {
         if (tcpChannel instanceof Netty4TcpChannel) {
             Channel nettyChannel = ((Netty4TcpChannel) tcpChannel).getNettyChannel();
             SslHandler handler = nettyChannel.pipeline().get(SslHandler.class);
-            assert handler != null : "Must have SslHandler";
+            if (handler == null) {
+                if (nettyChannel.isOpen()) {
+                    assert false : "Must have SslHandler";
+                } else {
+                    throw new ChannelException("Channel is closed.");
+                }
+            }
             return handler.engine();
         } else if (tcpChannel instanceof NioTcpChannel) {
             SocketChannelContext context = ((NioTcpChannel) tcpChannel).getContext();