[FEATURE] Built-in secure transports support

Signed-off-by: Andriy Redko <andriy.redko@aiven.io>

CHANGELOG.md

@@ -105,6 +105,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Support for returning scores in matched queries ([#11626](https://github.com/opensearch-project/OpenSearch/pull/11626))
 - Add shard id property to SearchLookup for use in field types provided by plugins ([#1063](https://github.com/opensearch-project/OpenSearch/pull/1063))
 - Add kuromoji_completion analyzer and filter ([#4835](https://github.com/opensearch-project/OpenSearch/issues/4835))
+- Built-in secure transports support ([#12435](https://github.com/opensearch-project/OpenSearch/pull/12435))
 
 ### Dependencies
 - Bump `peter-evans/find-comment` from 2 to 3 ([#12288](https://github.com/opensearch-project/OpenSearch/pull/12288))

modules/transport-netty4/src/main/java/org/opensearch/http/netty4/ssl/SecureNetty4HttpServerTransport.java

@@ -0,0 +1,164 @@
+/*
+ * Copyright 2015-2017 floragunn GmbH
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+/*
+ * 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.
+ *
+ * Modifications Copyright OpenSearch Contributors. See
+ * GitHub history for details.
+ */
+
+package org.opensearch.http.netty4.ssl;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.opensearch.common.network.NetworkService;
+import org.opensearch.common.settings.ClusterSettings;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.common.util.BigArrays;
+import org.opensearch.core.xcontent.NamedXContentRegistry;
+import org.opensearch.http.HttpChannel;
+import org.opensearch.http.HttpHandlingSettings;
+import org.opensearch.http.netty4.Netty4HttpChannel;
+import org.opensearch.http.netty4.Netty4HttpServerTransport;
+import org.opensearch.plugins.SecureSettingProvider;
+import org.opensearch.telemetry.tracing.Tracer;
+import org.opensearch.threadpool.ThreadPool;
+import org.opensearch.transport.SharedGroupFactory;
+
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.DecoderException;
+import io.netty.handler.ssl.ApplicationProtocolNames;
+import io.netty.handler.ssl.ApplicationProtocolNegotiationHandler;
+import io.netty.handler.ssl.SslHandler;
+
+public class SecureNetty4HttpServerTransport extends Netty4HttpServerTransport {
+    private static final Logger logger = LogManager.getLogger(SecureNetty4HttpServerTransport.class);
+    private final SecureSettingProvider ssp;
+    private final SecureSettingProvider.ServerExceptionHandler exceptionHandler;
+
+    public SecureNetty4HttpServerTransport(
+        final Settings settings,
+        final NetworkService networkService,
+        final BigArrays bigArrays,
+        final ThreadPool threadPool,
+        final NamedXContentRegistry namedXContentRegistry,
+        final Dispatcher dispatcher,
+        final ClusterSettings clusterSettings,
+        final SharedGroupFactory sharedGroupFactory,
+        final SecureSettingProvider ssp,
+        final Tracer tracer
+    ) {
+        super(
+            settings,
+            networkService,
+            bigArrays,
+            threadPool,
+            namedXContentRegistry,
+            dispatcher,
+            clusterSettings,
+            sharedGroupFactory,
+            tracer
+        );
+        this.ssp = ssp;
+        this.exceptionHandler = ssp.buildHttpServerExceptionHandler(settings, this)
+            .orElse(SecureSettingProvider.ServerExceptionHandler.NOOP);
+    }
+
+    @Override
+    public ChannelHandler configureServerChannelHandler() {
+        return new SslHttpChannelHandler(this, handlingSettings);
+    }
+
+    @Override
+    public void onException(HttpChannel channel, Exception cause0) {
+        Throwable cause = cause0;
+
+        if (cause0 instanceof DecoderException && cause0 != null) {
+            cause = cause0.getCause();
+        }
+
+        exceptionHandler.onError(cause);
+        logger.error("Exception during establishing a SSL connection: " + cause, cause);
+        super.onException(channel, cause0);
+    }
+
+    protected class SslHttpChannelHandler extends Netty4HttpServerTransport.HttpChannelHandler {
+        /**
+         * Application negotiation handler to select either HTTP 1.1 or HTTP 2 protocol, based
+         * on client/server ALPN negotiations.
+         */
+        private class Http2OrHttpHandler extends ApplicationProtocolNegotiationHandler {
+            protected Http2OrHttpHandler() {
+                super(ApplicationProtocolNames.HTTP_1_1);
+            }
+
+            @Override
+            protected void configurePipeline(ChannelHandlerContext ctx, String protocol) throws Exception {
+                if (ApplicationProtocolNames.HTTP_2.equals(protocol)) {
+                    configureDefaultHttp2Pipeline(ctx.pipeline());
+                } else if (ApplicationProtocolNames.HTTP_1_1.equals(protocol)) {
+                    configureDefaultHttpPipeline(ctx.pipeline());
+                } else {
+                    throw new IllegalStateException("Unknown application protocol: " + protocol);
+                }
+            }
+
+            @Override
+            public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+                super.exceptionCaught(ctx, cause);
+                Netty4HttpChannel channel = ctx.channel().attr(HTTP_CHANNEL_KEY).get();
+                if (channel != null) {
+                    if (cause instanceof Error) {
+                        onException(channel, new Exception(cause));
+                    } else {
+                        onException(channel, (Exception) cause);
+                    }
+                }
+            }
+        }
+
+        protected SslHttpChannelHandler(final Netty4HttpServerTransport transport, final HttpHandlingSettings handlingSettings) {
+            super(transport, handlingSettings);
+        }
+
+        @Override
+        protected void initChannel(Channel ch) throws Exception {
+            super.initChannel(ch);
+
+            final SSLEngine sslEngine = ssp.buildSecureHttpServerEngine(settings, SecureNetty4HttpServerTransport.this)
+                .orElseGet(SSLContext.getDefault()::createSSLEngine);
+
+            final SslHandler sslHandler = new SslHandler(sslEngine);
+            ch.pipeline().addFirst("ssl_http", sslHandler);
+        }
+
+        @Override
+        protected void configurePipeline(Channel ch) {
+            ch.pipeline().addLast(new Http2OrHttpHandler());
+        }
+    }
+}

modules/transport-netty4/src/main/java/org/opensearch/transport/Netty4ModulePlugin.java

@@ -46,11 +46,14 @@
 import org.opensearch.core.xcontent.NamedXContentRegistry;
 import org.opensearch.http.HttpServerTransport;
 import org.opensearch.http.netty4.Netty4HttpServerTransport;
+import org.opensearch.http.netty4.ssl.SecureNetty4HttpServerTransport;
 import org.opensearch.plugins.NetworkPlugin;
 import org.opensearch.plugins.Plugin;
+import org.opensearch.plugins.SecureSettingProvider;
 import org.opensearch.telemetry.tracing.Tracer;
 import org.opensearch.threadpool.ThreadPool;
 import org.opensearch.transport.netty4.Netty4Transport;
+import org.opensearch.transport.netty4.ssl.SecureNetty4ServerTransport;
 
 import java.util.Arrays;
 import java.util.Collections;
@@ -61,7 +64,9 @@
 public class Netty4ModulePlugin extends Plugin implements NetworkPlugin {
 
     public static final String NETTY_TRANSPORT_NAME = "netty4";
+    public static final String NETTY_SECURE_TRANSPORT_NAME = "netty4-secure";
     public static final String NETTY_HTTP_TRANSPORT_NAME = "netty4";
+    public static final String NETTY_SECURE_HTTP_TRANSPORT_NAME = "netty4-secure";
 
     private final SetOnce<SharedGroupFactory> groupFactory = new SetOnce<>();
 
@@ -144,6 +149,65 @@ public Map<String, Supplier<HttpServerTransport>> getHttpTransports(
         );
     }
 
+    @Override
+    public Map<String, Supplier<HttpServerTransport>> getSecureHttpTransports(
+        Settings settings,
+        ThreadPool threadPool,
+        BigArrays bigArrays,
+        PageCacheRecycler pageCacheRecycler,
+        CircuitBreakerService circuitBreakerService,
+        NamedXContentRegistry xContentRegistry,
+        NetworkService networkService,
+        HttpServerTransport.Dispatcher dispatcher,
+        ClusterSettings clusterSettings,
+        SecureSettingProvider ssp,
+        Tracer tracer
+    ) {
+        return Collections.singletonMap(
+            NETTY_SECURE_HTTP_TRANSPORT_NAME,
+            () -> new SecureNetty4HttpServerTransport(
+                settings,
+                networkService,
+                bigArrays,
+                threadPool,
+                xContentRegistry,
+                dispatcher,
+                clusterSettings,
+                getSharedGroupFactory(settings),
+                ssp,
+                tracer
+            )
+        );
+    }
+
+    @Override
+    public Map<String, Supplier<Transport>> getSecureTransports(
+        Settings settings,
+        ThreadPool threadPool,
+        PageCacheRecycler pageCacheRecycler,
+        CircuitBreakerService circuitBreakerService,
+        NamedWriteableRegistry namedWriteableRegistry,
+        NetworkService networkService,
+        SecureSettingProvider ssp,
+        Tracer tracer
+    ) {
+        return Collections.singletonMap(
+            NETTY_SECURE_TRANSPORT_NAME,
+            () -> new SecureNetty4ServerTransport(
+                settings,
+                Version.CURRENT,
+                threadPool,
+                networkService,
+                pageCacheRecycler,
+                namedWriteableRegistry,
+                circuitBreakerService,
+                getSharedGroupFactory(settings),
+                ssp,
+                tracer
+            )
+        );
+    }
+
     SharedGroupFactory getSharedGroupFactory(Settings settings) {
         SharedGroupFactory groupFactory = this.groupFactory.get();
         if (groupFactory != null) {

modules/transport-netty4/src/main/java/org/opensearch/transport/netty4/ssl/DualModeSslHandler.java

@@ -0,0 +1,101 @@
+/*
+ * 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.
+ *
+ * Modifications Copyright OpenSearch Contributors. See
+ * GitHub history for details.
+ */
+package org.opensearch.transport.netty4.ssl;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.plugins.SecureSettingProvider;
+import org.opensearch.transport.TcpTransport;
+
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
+import javax.net.ssl.SSLException;
+
+import java.nio.charset.StandardCharsets;
+import java.security.NoSuchAlgorithmException;
+import java.util.List;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPipeline;
+import io.netty.handler.codec.ByteToMessageDecoder;
+import io.netty.handler.ssl.SslHandler;
+
+/**
+ * Modifies the current pipeline dynamically to enable TLS
+ */
+public class DualModeSslHandler extends ByteToMessageDecoder {
+
+    private static final Logger logger = LogManager.getLogger(DualModeSslHandler.class);
+    private final Settings settings;
+    private final SecureSettingProvider ssp;
+    private final TcpTransport transport;
+    private final SslHandler providedSSLHandler;
+
+    public DualModeSslHandler(final Settings settings, final SecureSettingProvider ssp, final TcpTransport transport) {
+        this(settings, ssp, transport, null);
+    }
+
+    protected DualModeSslHandler(
+        final Settings settings,
+        final SecureSettingProvider ssp,
+        final TcpTransport transport,
+        SslHandler providedSSLHandler
+    ) {
+        this.settings = settings;
+        this.ssp = ssp;
+        this.transport = transport;
+        this.providedSSLHandler = providedSSLHandler;
+    }
+
+    @Override
+    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
+        // Will use the first six bytes to detect a protocol.
+        if (in.readableBytes() < 6) {
+            return;
+        }
+        int offset = in.readerIndex();
+        if (in.getCharSequence(offset, 6, StandardCharsets.UTF_8).equals(SecureConnectionTestUtil.DUAL_MODE_CLIENT_HELLO_MSG)) {
+            logger.debug("Received DualSSL Client Hello message");
+            ByteBuf responseBuffer = Unpooled.buffer(6);
+            responseBuffer.writeCharSequence(SecureConnectionTestUtil.DUAL_MODE_SERVER_HELLO_MSG, StandardCharsets.UTF_8);
+            ctx.writeAndFlush(responseBuffer).addListener(ChannelFutureListener.CLOSE);
+            return;
+        }
+
+        if (SslUtils.isTLS(in)) {
+            logger.debug("Identified request as SSL request");
+            enableSsl(ctx);
+        } else {
+            logger.debug("Identified request as non SSL request, running in HTTP mode as dual mode is enabled");
+            ctx.pipeline().remove(this);
+        }
+    }
+
+    private void enableSsl(ChannelHandlerContext ctx) throws SSLException, NoSuchAlgorithmException {
+        final SSLEngine sslEngine = ssp.buildSecureServerTransportEngine(settings, transport)
+            .orElseGet(SSLContext.getDefault()::createSSLEngine);
+
+        SslHandler sslHandler;
+        if (providedSSLHandler != null) {
+            sslHandler = providedSSLHandler;
+        } else {
+            sslHandler = new SslHandler(sslEngine);
+        }
+        ChannelPipeline p = ctx.pipeline();
+        p.addAfter("port_unification_handler", "ssl_server", sslHandler);
+        p.remove(this);
+        logger.debug("Removed port unification handler and added SSL handler as incoming request is SSL");
+    }
+}