Implement SslHandler retrival logic for transport-reactor-netty4 plugin

Signed-off-by: Andriy Redko <[email protected]>

Author: reta

plugins/transport-reactor-netty4/src/main/java/org/opensearch/http/reactor/netty4/ReactorNetty4BaseHttpChannel.java

@@ -0,0 +1,39 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.http.reactor.netty4;
+
+import java.util.Optional;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandler;
+import reactor.netty.NettyPipeline;
+import reactor.netty.http.server.HttpServerRequest;
+
+final class ReactorNetty4BaseHttpChannel {
+    private ReactorNetty4BaseHttpChannel() {}
+
+    @SuppressWarnings("unchecked")
+    static <T> Optional<T> get(HttpServerRequest request, String name, Class<T> clazz) {
+        if ("ssl_http".equalsIgnoreCase(name)) {
+            final ChannelHandler[] channels = new ChannelHandler[1];
+            request.withConnection(connection -> {
+                final Channel channel = connection.channel();
+                if (channel.parent() != null) {
+                    channels[0] = channel.parent().pipeline().get(NettyPipeline.SslHandler);
+                } else {
+                    channels[0] = channel.pipeline().get(NettyPipeline.SslHandler);
+                }
+            });
+            if (channels[0] != null && clazz.isInstance(channels[0].getClass()) == true) {
+                return Optional.of((T) channels[0]);
+            }
+        }
+        return Optional.empty();
+    }
+}

plugins/transport-reactor-netty4/src/main/java/org/opensearch/http/reactor/netty4/ReactorNetty4HttpServerTransport.java

@@ -53,6 +53,7 @@
 import io.netty.handler.codec.http.HttpResponseStatus;
 import io.netty.handler.ssl.ApplicationProtocolConfig;
 import io.netty.handler.ssl.ApplicationProtocolNames;
+import io.netty.handler.ssl.ClientAuth;
 import io.netty.handler.ssl.SslContext;
 import io.netty.handler.ssl.SslContextBuilder;
 import io.netty.handler.ssl.SupportedCipherSuiteFilter;
@@ -317,6 +318,8 @@ private HttpServer configure(final HttpServer server) throws Exception {
             parameters.flatMap(SecureHttpTransportParameters::trustManagerFactory).ifPresent(sslContextBuilder::trustManager);
             parameters.map(SecureHttpTransportParameters::cipherSuites)
                 .ifPresent(ciphers -> sslContextBuilder.ciphers(ciphers, SupportedCipherSuiteFilter.INSTANCE));
+            parameters.flatMap(SecureHttpTransportParameters::clientAuth)
+                .ifPresent(clientAuth -> sslContextBuilder.clientAuth(ClientAuth.valueOf(clientAuth)));
 
             final SslContext sslContext = sslContextBuilder.protocols(
                 parameters.map(SecureHttpTransportParameters::protocols).orElseGet(() -> Arrays.asList(SslUtils.DEFAULT_SSL_PROTOCOLS))

plugins/transport-reactor-netty4/src/main/java/org/opensearch/http/reactor/netty4/ReactorNetty4NonStreamingHttpChannel.java

@@ -15,6 +15,7 @@
 import org.opensearch.transport.reactor.netty4.Netty4Utils;
 
 import java.net.InetSocketAddress;
+import java.util.Optional;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import io.netty.handler.codec.http.FullHttpResponse;
@@ -75,6 +76,11 @@ public InetSocketAddress getLocalAddress() {
         return (InetSocketAddress) response.hostAddress();
     }
 
+    @Override
+    public <T> Optional<T> get(String name, Class<T> clazz) {
+        return ReactorNetty4BaseHttpChannel.get(request, name, clazz);
+    }
+
     FullHttpResponse createResponse(HttpResponse response) {
         return (FullHttpResponse) response;
     }

plugins/transport-reactor-netty4/src/main/java/org/opensearch/http/reactor/netty4/ReactorNetty4StreamingHttpChannel.java

@@ -19,6 +19,7 @@
 import java.net.InetSocketAddress;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 
 import io.netty.buffer.Unpooled;
 import io.netty.handler.codec.http.DefaultHttpContent;
@@ -123,6 +124,11 @@ public void subscribe(Subscriber<? super HttpChunk> subscriber) {
         receiver.subscribe(subscriber);
     }
 
+    @Override
+    public <T> Optional<T> get(String name, Class<T> clazz) {
+        return ReactorNetty4BaseHttpChannel.get(request, name, clazz);
+    }
+
     private static HttpContent createContent(HttpResponse response) {
         final FullHttpResponse fullHttpResponse = (FullHttpResponse) response;
         return new DefaultHttpContent(fullHttpResponse.content());