Update x-pack to support TcpChannel (elastic/x-pack-elasticsearch#2983)

This is related to elastic/elasticsearch#27132. This commit updates
x-pack to support signature changes introduced in that PR.

Original commit: elastic/x-pack-elasticsearch@461c96ffe6

plugin/src/main/java/org/elasticsearch/xpack/security/rest/SecurityRestFilter.java

@@ -52,7 +52,7 @@ public class SecurityRestFilter implements RestHandler {
                 Netty4HttpRequest nettyHttpRequest = (Netty4HttpRequest) request;
                 SslHandler handler = nettyHttpRequest.getChannel().pipeline().get(SslHandler.class);
                 assert handler != null;
-                ServerTransportFilter.extactClientCertificates(logger, threadContext, handler.engine(), nettyHttpRequest.getChannel());
+                ServerTransportFilter.extractClientCertificates(logger, threadContext, handler.engine(), nettyHttpRequest.getChannel());
             }
             service.authenticate(maybeWrapRestRequest(request), ActionListener.wrap(
                 authentication -> {

plugin/src/main/java/org/elasticsearch/xpack/security/transport/ServerTransportFilter.java

@@ -24,6 +24,7 @@ import org.elasticsearch.transport.TcpTransportChannel;
 import org.elasticsearch.transport.TransportChannel;
 import org.elasticsearch.transport.TransportRequest;
 import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.transport.netty4.NettyTcpChannel;
 import org.elasticsearch.xpack.security.SecurityContext;
 import org.elasticsearch.xpack.security.action.SecurityActionMapper;
 import org.elasticsearch.xpack.security.authc.Authentication;
@@ -114,12 +115,12 @@ public interface ServerTransportFilter {
             }
 
             if (extractClientCert && (unwrappedChannel instanceof TcpTransportChannel) &&
-                ((TcpTransportChannel) unwrappedChannel).getChannel() instanceof io.netty.channel.Channel) {
+                ((TcpTransportChannel) unwrappedChannel).getChannel() instanceof NettyTcpChannel) {
-                Channel channel = (io.netty.channel.Channel) ((TcpTransportChannel) unwrappedChannel).getChannel();
+                Channel channel = ((NettyTcpChannel) ((TcpTransportChannel) unwrappedChannel).getChannel()).getLowLevelChannel();
                 SslHandler sslHandler = channel.pipeline().get(SslHandler.class);
                 if (channel.isOpen()) {
                     assert sslHandler != null : "channel [" + channel + "] did not have a ssl handler. pipeline " + channel.pipeline();
-                    extactClientCertificates(logger, threadContext, sslHandler.engine(), channel);
+                    extractClientCertificates(logger, threadContext, sslHandler.engine(), channel);
                 }
             }
 
@@ -170,7 +171,7 @@ public interface ServerTransportFilter {
         }
     }
 
-    static void extactClientCertificates(Logger logger, ThreadContext threadContext, SSLEngine sslEngine, Object channel) {
+    static void extractClientCertificates(Logger logger, ThreadContext threadContext, SSLEngine sslEngine, Channel channel) {
         try {
             Certificate[] certs = sslEngine.getSession().getPeerCertificates();
             if (certs instanceof X509Certificate[]) {

plugin/src/main/java/org/elasticsearch/xpack/security/transport/netty4/SecurityNetty4Transport.java

@@ -19,8 +19,10 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.BigArrays;
 import org.elasticsearch.indices.breaker.CircuitBreakerService;
 import org.elasticsearch.threadpool.ThreadPool;
+import org.elasticsearch.transport.TcpChannel;
 import org.elasticsearch.transport.TcpTransport;
 import org.elasticsearch.transport.netty4.Netty4Transport;
+import org.elasticsearch.transport.netty4.NettyTcpChannel;
 import org.elasticsearch.xpack.XPackSettings;
 import org.elasticsearch.xpack.ssl.SSLConfiguration;
 import org.elasticsearch.xpack.ssl.SSLService;
@@ -107,10 +109,10 @@ public class SecurityNetty4Transport extends Netty4Transport {
     }
 
     @Override
-    protected void onException(Channel channel, Exception e) {
+    protected void onException(NettyTcpChannel channel, Exception e) {
         if (!lifecycle.started()) {
             // just close and ignore - we are already stopped and just need to make sure we release all resources
-            closeChannelWhileHandlingExceptions(channel);
+            TcpChannel.closeChannel(channel, false);
         } else if (isNotSslRecordException(e)) {
             if (logger.isTraceEnabled()) {
                 logger.trace(
@@ -118,21 +120,21 @@ public class SecurityNetty4Transport extends Netty4Transport {
             } else {
                 logger.warn("received plaintext traffic on an encrypted channel, closing connection {}", channel);
             }
-            closeChannelWhileHandlingExceptions(channel);
+            TcpChannel.closeChannel(channel, false);
         } else if (isCloseDuringHandshakeException(e)) {
             if (logger.isTraceEnabled()) {
                 logger.trace(new ParameterizedMessage("connection {} closed during ssl handshake", channel), e);
             } else {
                 logger.warn("connection {} closed during handshake", channel);
             }
-            closeChannelWhileHandlingExceptions(channel);
+            TcpChannel.closeChannel(channel, false);
         } else if (isReceivedCertificateUnknownException(e)) {
             if (logger.isTraceEnabled()) {
                 logger.trace(new ParameterizedMessage("client did not trust server's certificate, closing connection {}", channel), e);
             } else {
                 logger.warn("client did not trust this server's certificate, closing connection {}", channel);
             }
-            closeChannelWhileHandlingExceptions(channel);
+            TcpChannel.closeChannel(channel, false);
         } else {
             super.onException(channel, e);
         }