Grpc secure transport

CHANGELOG.md

@@ -11,6 +11,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Introduce a setting to disable download of full cluster state from remote on term mismatch([#16798](https://github.com/opensearch-project/OpenSearch/pull/16798/))
 - Added ability to retrieve value from DocValues in a flat_object filed([#16802](https://github.com/opensearch-project/OpenSearch/pull/16802))
 - Improve performace of NumericTermAggregation by avoiding unnecessary sorting([#17252](https://github.com/opensearch-project/OpenSearch/pull/17252))
+- Add TLS enabled SecureNetty4GrpcServerTransport ([#17406](https://github.com/opensearch-project/OpenSearch/pull/17406))
 
 ### Dependencies
 - Bump `org.awaitility:awaitility` from 4.2.0 to 4.2.2 ([#17230](https://github.com/opensearch-project/OpenSearch/pull/17230))

plugins/arrow-flight-rpc/src/main/java/org/opensearch/arrow/flight/bootstrap/FlightService.java

@@ -134,6 +134,7 @@ public StreamManager getStreamManager() {
      * Retrieves the bound address of the FlightService.
      * @return The BoundTransportAddress instance.
      */
+    @Override
     public BoundTransportAddress getBoundAddress() {
         return serverComponents.getBoundAddress();
     }

plugins/transport-grpc/README.md

@@ -0,0 +1,22 @@
+# transport-grpc
+
+An auxiliary transport which runs in parallel to the REST API.
+The `transport-grpc` plugin initializes a new client/server transport implementing a gRPC protocol on Netty4.
+
+Enable this transport with:
+```
+setting 'aux.transport.types', '[experimental-transport-grpc]'
+setting 'aux.transport.experimental-transport-grpc.port', '9400-9500' //optional
+```
+
+## Testing
+
+### Unit Tests
+
+```
+./gradlew :plugins:transport-grpc:test
+```
+
+### Integration Tests
+
+COMING SOON - Fill this out with PR completion

plugins/transport-grpc/build.gradle

@@ -8,11 +8,21 @@ import org.gradle.api.attributes.java.TargetJvmEnvironment
  * compatible open source license.
  */
 
+apply plugin: 'opensearch.testclusters'
+apply plugin: 'opensearch.internal-cluster-test'
+
 opensearchplugin {
   description = 'gRPC based transport implementation'
   classname = 'org.opensearch.transport.grpc.GrpcPlugin'
 }
 
+testClusters {
+  integTest {
+    plugin(project.path)
+    setting 'aux.transport.types', '[experimental-transport-grpc]'
+  }
+}
+
 dependencies {
   compileOnly "com.google.code.findbugs:jsr305:3.0.2"
   runtimeOnly "com.google.guava:guava:${versions.guava}"
@@ -27,6 +37,9 @@ dependencies {
   implementation "io.grpc:grpc-stub:${versions.grpc}"
   implementation "io.grpc:grpc-util:${versions.grpc}"
   implementation "io.perfmark:perfmark-api:0.26.0"
+
+  testImplementation project(':test:framework')
+  testImplementation project(':libs:opensearch-core')
 }
 
 tasks.named("dependencyLicenses").configure {

plugins/transport-grpc/src/main/java/org/opensearch/transport/grpc/GrpcPlugin.java

@@ -14,8 +14,10 @@
 import org.opensearch.core.indices.breaker.CircuitBreakerService;
 import org.opensearch.plugins.NetworkPlugin;
 import org.opensearch.plugins.Plugin;
+import org.opensearch.plugins.SecureAuxTransportSettingsProvider;
 import org.opensearch.telemetry.tracing.Tracer;
 import org.opensearch.threadpool.ThreadPool;
+import org.opensearch.transport.grpc.ssl.SecureNetty4GrpcServerTransport;
 
 import java.util.Collections;
 import java.util.List;
@@ -55,6 +57,22 @@ public Map<String, Supplier<AuxTransport>> getAuxTransports(
         );
     }
 
+    @Override
+    public Map<String, Supplier<AuxTransport>> getSecureAuxTransports(
+        Settings settings,
+        ThreadPool threadPool,
+        CircuitBreakerService circuitBreakerService,
+        NetworkService networkService,
+        ClusterSettings clusterSettings,
+        SecureAuxTransportSettingsProvider secureAuxTransportSettingsProvider,
+        Tracer tracer
+    ) {
+        return Collections.singletonMap(
+            GRPC_TRANSPORT_SETTING_KEY,
+            () -> new SecureNetty4GrpcServerTransport(settings, Collections.emptyList(), networkService, secureAuxTransportSettingsProvider)
+        );
+    }
+
     @Override
     public List<Setting<?>> getSettings() {
         return List.of(

plugins/transport-grpc/src/main/java/org/opensearch/transport/grpc/Netty4GrpcServerTransport.java

@@ -32,9 +32,9 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
+import java.util.function.UnaryOperator;
 
 import io.grpc.BindableService;
-import io.grpc.InsecureServerCredentials;
 import io.grpc.Server;
 import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder;
 import io.grpc.netty.shaded.io.netty.channel.EventLoopGroup;
@@ -115,14 +115,23 @@ public class Netty4GrpcServerTransport extends NetworkPlugin.AuxTransport {
         Setting.Property.NodeScope
     );
 
-    private final Settings settings;
+    /**
+     * Port range on which servers bind.
+     */
+    protected PortsRange port;
+
+    /**
+     * Settings.
+     */
+    protected final Settings settings;
+
     private final NetworkService networkService;
     private final List<BindableService> services;
-    private final CopyOnWriteArrayList<Server> servers = new CopyOnWriteArrayList<>();
     private final String[] bindHosts;
     private final String[] publishHosts;
-    private final PortsRange port;
     private final int nettyEventLoopThreads;
+    private final CopyOnWriteArrayList<Server> servers = new CopyOnWriteArrayList<>();
+    private final List<UnaryOperator<NettyServerBuilder>> serverBuilderConfigs = new ArrayList<>();
 
     private volatile BoundTransportAddress boundAddress;
     private volatile EventLoopGroup eventLoopGroup;
@@ -152,10 +161,20 @@ public Netty4GrpcServerTransport(Settings settings, List<BindableService> servic
         this.nettyEventLoopThreads = SETTING_GRPC_WORKER_COUNT.get(settings);
     }
 
-    BoundTransportAddress boundAddress() {
+    // public for tests
+    @Override
+    public BoundTransportAddress getBoundAddress() {
         return this.boundAddress;
     }
 
+    /**
+     * Inject a NettyServerBuilder configuration to be applied at server bind and start.
+     * @param configModifier builder configuration to set.
+     */
+    protected void addServerConfig(UnaryOperator<NettyServerBuilder> configModifier) {
+        serverBuilderConfigs.add(configModifier);
+    }
+
     @Override
     protected void doStart() {
         boolean success = false;
@@ -198,7 +217,7 @@ protected void doStop() {
 
     @Override
     protected void doClose() {
-
+        eventLoopGroup.close();
     }
 
     private void bindServer() {
@@ -249,13 +268,18 @@ private TransportAddress bindAddress(InetAddress hostAddress, PortsRange portRan
             try {
 
                 final InetSocketAddress address = new InetSocketAddress(hostAddress, portNumber);
-                final NettyServerBuilder serverBuilder = NettyServerBuilder.forAddress(address, InsecureServerCredentials.create())
+                final NettyServerBuilder serverBuilder = NettyServerBuilder.forAddress(address)
+                    .directExecutor()
                     .bossEventLoopGroup(eventLoopGroup)
                     .workerEventLoopGroup(eventLoopGroup)
                     .channelType(NioServerSocketChannel.class)
                     .addService(new HealthStatusManager().getHealthService())
                     .addService(ProtoReflectionService.newInstance());
 
+                for (UnaryOperator<NettyServerBuilder> op : serverBuilderConfigs) {
+                    op.apply(serverBuilder);
+                }
+
                 services.forEach(serverBuilder::addService);
 
                 Server srv = serverBuilder.build().start();

plugins/transport-grpc/src/main/java/org/opensearch/transport/grpc/ssl/SecureNetty4GrpcServerTransport.java

@@ -0,0 +1,88 @@
+/*
+ * 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.transport.grpc.ssl;
+
+import org.opensearch.common.network.NetworkService;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.plugins.SecureAuxTransportSettingsProvider;
+import org.opensearch.transport.grpc.Netty4GrpcServerTransport;
+
+import javax.net.ssl.SSLException;
+
+import java.util.List;
+import java.util.Locale;
+
+import io.grpc.BindableService;
+import io.grpc.netty.shaded.io.grpc.netty.NettyServerBuilder;
+import io.grpc.netty.shaded.io.netty.handler.ssl.ApplicationProtocolConfig;
+import io.grpc.netty.shaded.io.netty.handler.ssl.ApplicationProtocolNames;
+import io.grpc.netty.shaded.io.netty.handler.ssl.ClientAuth;
+import io.grpc.netty.shaded.io.netty.handler.ssl.SslContext;
+import io.grpc.netty.shaded.io.netty.handler.ssl.SslContextBuilder;
+import io.grpc.netty.shaded.io.netty.handler.ssl.SslProvider;
+
+/**
+ * Netty4GrpcServerTransport with TLS enabled.
+ * Security settings injected through a SecureAuxTransportSettingsProvider.
+ */
+public class SecureNetty4GrpcServerTransport extends Netty4GrpcServerTransport {
+    private final SecureAuxTransportSettingsProvider secureAuxTransportSettingsProvider;
+    private final SslContext sslContext;
+
+    /**
+     * Creates a new SecureNetty4GrpcServerTransport instance.
+     * @param settings the configured settings.
+     * @param services the gRPC compatible services to be registered with the server.
+     * @param networkService the bind/publish addresses.
+     * @param secureTransportSettingsProvider TLS configuration settings.
+     */
+    public SecureNetty4GrpcServerTransport(
+        Settings settings,
+        List<BindableService> services,
+        NetworkService networkService,
+        SecureAuxTransportSettingsProvider secureTransportSettingsProvider
+    ) {
+        super(settings, services, networkService);
+        this.secureAuxTransportSettingsProvider = secureTransportSettingsProvider;
+        this.port = SecureNetty4GrpcServerTransport.SETTING_GRPC_PORT.get(settings);
+
+        try {
+            this.sslContext = buildSslContext();
+        } catch (SSLException e) {
+            throw new RuntimeException(SecureNetty4GrpcServerTransport.class + " failed to build SslContext", e);
+        }
+
+        this.addServerConfig((NettyServerBuilder builder) -> builder.sslContext(this.sslContext));
+    }
+
+    /**
+     * @return io.grpc SslContext from SecureAuxTransportSettingsProvider.
+     */
+    private SslContext buildSslContext() throws SSLException {
+        if (secureAuxTransportSettingsProvider.parameters(settings).isEmpty()) {
+            throw new SSLException("SSLContext could not be built from SecureAuxTransportSettingsProvider: provider empty");
+        }
+        SecureAuxTransportSettingsProvider.SecureTransportParameters params = secureAuxTransportSettingsProvider.parameters(settings).get();
+        return SslContextBuilder.forServer(params.keyManagerFactory())
+            .trustManager(params.trustManagerFactory())
+            .sslProvider(SslProvider.valueOf(params.sslProvider().toUpperCase(Locale.ROOT)))
+            .clientAuth(ClientAuth.valueOf(params.clientAuth().toUpperCase(Locale.ROOT)))
+            .protocols(params.protocols())
+            .ciphers(params.cipherSuites())
+            .applicationProtocolConfig(
+                new ApplicationProtocolConfig(
+                    ApplicationProtocolConfig.Protocol.ALPN,
+                    ApplicationProtocolConfig.SelectorFailureBehavior.NO_ADVERTISE,
+                    ApplicationProtocolConfig.SelectedListenerFailureBehavior.ACCEPT,
+                    ApplicationProtocolNames.HTTP_2
+                )
+            )
+            .build();
+    }
+}

plugins/transport-grpc/src/main/java/org/opensearch/transport/grpc/ssl/package-info.java

@@ -0,0 +1,12 @@
+/*
+ * 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.
+ */
+
+/**
+ * gRPC transport for OpenSearch implementing TLS.
+ */
+package org.opensearch.transport.grpc.ssl;

plugins/transport-grpc/src/test/java/org/opensearch/transport/grpc/Netty4GrpcServerTransportIT.java

@@ -0,0 +1,59 @@
+/*
+ * 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.transport.grpc;
+
+import org.opensearch.action.admin.cluster.health.ClusterHealthResponse;
+import org.opensearch.cluster.health.ClusterHealthStatus;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.core.common.transport.TransportAddress;
+import org.opensearch.plugins.Plugin;
+import org.opensearch.test.OpenSearchIntegTestCase;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+import io.grpc.health.v1.HealthCheckResponse;
+
+import static org.opensearch.plugins.NetworkPlugin.AuxTransport.AUX_TRANSPORT_TYPES_KEY;
+import static org.opensearch.transport.grpc.Netty4GrpcServerTransport.GRPC_TRANSPORT_SETTING_KEY;
+
+public class Netty4GrpcServerTransportIT extends OpenSearchIntegTestCase {
+
+    private TransportAddress randomNetty4GrpcServerTransportAddr() {
+        List<TransportAddress> addresses = new ArrayList<>();
+        for (Netty4GrpcServerTransport transport : internalCluster().getInstances(Netty4GrpcServerTransport.class)) {
+            TransportAddress tAddr = new TransportAddress(transport.getBoundAddress().publishAddress().address());
+            addresses.add(tAddr);
+        }
+        return randomFrom(addresses);
+    }
+
+    @Override
+    protected Settings nodeSettings(int nodeOrdinal) {
+        return Settings.builder().put(super.nodeSettings(nodeOrdinal)).put(AUX_TRANSPORT_TYPES_KEY, GRPC_TRANSPORT_SETTING_KEY).build();
+    }
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return Collections.singleton(GrpcPlugin.class);
+    }
+
+    public void testStartGrpcTransportClusterHealth() throws Exception {
+        // REST api cluster health
+        ClusterHealthResponse healthResponse = client().admin().cluster().prepareHealth().get();
+        assertEquals(ClusterHealthStatus.GREEN, healthResponse.getStatus());
+
+        // gRPC transport service health
+        try (NettyGrpcClient client = new NettyGrpcClient.Builder().setAddress(randomNetty4GrpcServerTransportAddr()).build()) {
+            assertEquals(client.checkHealth(), HealthCheckResponse.ServingStatus.SERVING);
+        }
+    }
+}