Enable TLS for Netty4GrpcServerTransport

CHANGELOG.md

@@ -17,6 +17,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add ingestion management APIs for pause, resume and get ingestion state ([#17631](https://github.com/opensearch-project/OpenSearch/pull/17631))
 - [Security Manager Replacement] Enhance Java Agent to intercept System::exit ([#17746](https://github.com/opensearch-project/OpenSearch/pull/17746))
 - Support AutoExpand for SearchReplica ([#17741](https://github.com/opensearch-project/OpenSearch/pull/17741))
+- Add TLS enabled SecureNetty4GrpcServerTransport ([#17406](https://github.com/opensearch-project/OpenSearch/pull/17406))
 
 ### Changed
 - Migrate BC libs to their FIPS counterparts ([#14912](https://github.com/opensearch-project/OpenSearch/pull/14912))

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,42 @@
+# 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
+```
+
+For the secure transport:
+
+```
+setting 'aux.transport.types',                                      '[experimental-secure-transport-grpc]'
+setting 'aux.transport.experimental-secure-transport-grpc.port',    '9400-9500' //optional
+```
+
+Other settings are agnostic as to the gRPC transport type:
+
+```
+setting 'grpc.publish_port',        '9400'
+setting 'grpc.host',                '["0.0.0.0"]'
+setting 'grpc.bind_host',           '["0.0.0.0", "::", "10.0.0.1"]'
+setting 'grpc.publish_host',        '["thisnode.example.com"]'
+setting 'grpc.netty.worker_count',  '2'
+```
+
+## Testing
+
+### Unit Tests
+
+```
+./gradlew :plugins:transport-grpc:test
+```
+
+### Integration Tests
+
+```
+./gradlew :plugins:transport-grpc:internalClusterTest
+```

plugins/transport-grpc/build.gradle

@@ -1,5 +1,3 @@
-import org.gradle.api.attributes.java.TargetJvmEnvironment
-
 /*
  * SPDX-License-Identifier: Apache-2.0
  *
@@ -8,11 +6,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 +35,7 @@ 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')
 }
 
 tasks.named("dependencyLicenses").configure {

plugins/transport-grpc/src/internalClusterTest/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);
+        }
+    }
+}

plugins/transport-grpc/src/internalClusterTest/java/org/opensearch/transport/grpc/SecureNetty4GrpcServerTransportIT.java

@@ -0,0 +1,222 @@
+/*
+ * 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.NetworkPlugin;
+import org.opensearch.plugins.Plugin;
+import org.opensearch.plugins.SecureAuxTransportSettingsProvider;
+import org.opensearch.plugins.SecureHttpTransportSettingsProvider;
+import org.opensearch.plugins.SecureSettingsFactory;
+import org.opensearch.plugins.SecureTransportSettingsProvider;
+import org.opensearch.test.OpenSearchIntegTestCase;
+import org.opensearch.transport.grpc.ssl.SecureNetty4GrpcServerTransport;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Optional;
+
+import io.grpc.health.v1.HealthCheckResponse;
+
+import static org.opensearch.plugins.NetworkPlugin.AuxTransport.AUX_TRANSPORT_TYPES_KEY;
+import static org.opensearch.transport.grpc.SecureSettingsHelpers.getServerClientAuthNone;
+import static org.opensearch.transport.grpc.SecureSettingsHelpers.getServerClientAuthOptional;
+import static org.opensearch.transport.grpc.SecureSettingsHelpers.getServerClientAuthRequired;
+import static org.opensearch.transport.grpc.ssl.SecureNetty4GrpcServerTransport.GRPC_SECURE_TRANSPORT_SETTING_KEY;
+
+public abstract class SecureNetty4GrpcServerTransportIT extends OpenSearchIntegTestCase {
+
+    public static class MockSecurityPlugin extends Plugin implements NetworkPlugin {
+        public MockSecurityPlugin() {}
+
+        static class MockSecureSettingsFactory implements SecureSettingsFactory {
+            @Override
+            public Optional<SecureTransportSettingsProvider> getSecureTransportSettingsProvider(Settings settings) {
+                return Optional.empty();
+            }
+
+            @Override
+            public Optional<SecureHttpTransportSettingsProvider> getSecureHttpTransportSettingsProvider(Settings settings) {
+                return Optional.empty();
+            }
+
+            @Override
+            public Optional<SecureAuxTransportSettingsProvider> getSecureAuxTransportSettingsProvider(Settings settings) {
+                return Optional.empty();
+            }
+        }
+    }
+
+    protected TransportAddress randomNetty4GrpcServerTransportAddr() {
+        List<TransportAddress> addresses = new ArrayList<>();
+        for (SecureNetty4GrpcServerTransport transport : internalCluster().getInstances(SecureNetty4GrpcServerTransport.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_SECURE_TRANSPORT_SETTING_KEY)
+            .build();
+    }
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return List.of(GrpcPlugin.class, MockSecurityPlugin.class);
+    }
+
+    private SecureSettingsHelpers.ConnectExceptions tryConnectClient(NettyGrpcClient client) {
+        try {
+            HealthCheckResponse.ServingStatus status = client.checkHealth();
+            if (status == HealthCheckResponse.ServingStatus.SERVING) {
+                return SecureSettingsHelpers.ConnectExceptions.NONE;
+            } else {
+                throw new RuntimeException("Illegal state - unexpected server status: " + status.toString());
+            }
+        } catch (Exception e) {
+            return SecureSettingsHelpers.ConnectExceptions.get(e);
+        }
+    }
+
+    protected SecureSettingsHelpers.ConnectExceptions plaintextClientConnect() throws Exception {
+        try (NettyGrpcClient client = new NettyGrpcClient.Builder().setAddress(randomNetty4GrpcServerTransportAddr()).build()) {
+            return tryConnectClient(client);
+        }
+    }
+
+    protected SecureSettingsHelpers.ConnectExceptions insecureClientConnect() throws Exception {
+        try (
+            NettyGrpcClient client = new NettyGrpcClient.Builder().setAddress(randomNetty4GrpcServerTransportAddr()).insecure(true).build()
+        ) {
+            return tryConnectClient(client);
+        }
+    }
+
+    protected SecureSettingsHelpers.ConnectExceptions trustedCertClientConnect() throws Exception {
+        try (
+            NettyGrpcClient client = new NettyGrpcClient.Builder().setAddress(randomNetty4GrpcServerTransportAddr())
+                .clientAuth(true)
+                .build()
+        ) {
+            return tryConnectClient(client);
+        }
+    }
+
+    public void testClusterHealth() {
+        ClusterHealthResponse healthResponse = client().admin().cluster().prepareHealth().get();
+        assertEquals(ClusterHealthStatus.GREEN, healthResponse.getStatus());
+    }
+
+    public static class SecureNetty4GrpcServerTransportNoAuthIT extends SecureNetty4GrpcServerTransportIT {
+        public static class NoAuthMockSecurityPlugin extends MockSecurityPlugin {
+            public NoAuthMockSecurityPlugin() {}
+
+            @Override
+            public Optional<SecureSettingsFactory> getSecureSettingFactory(Settings settings) {
+                return Optional.of(new MockSecureSettingsFactory() {
+                    @Override
+                    public Optional<SecureAuxTransportSettingsProvider> getSecureAuxTransportSettingsProvider(Settings settings) {
+                        return Optional.of(getServerClientAuthNone());
+                    }
+                });
+            }
+        }
+
+        @Override
+        protected Collection<Class<? extends Plugin>> nodePlugins() {
+            return List.of(GrpcPlugin.class, NoAuthMockSecurityPlugin.class);
+        }
+
+        public void testPlaintextClientConnect() throws Exception {
+            assertEquals(plaintextClientConnect(), SecureSettingsHelpers.ConnectExceptions.UNAVAILABLE);
+        }
+
+        public void testInsecureClientConnect() throws Exception {
+            assertEquals(insecureClientConnect(), SecureSettingsHelpers.ConnectExceptions.NONE);
+        }
+
+        public void testTrustedClientConnect() throws Exception {
+            assertEquals(trustedCertClientConnect(), SecureSettingsHelpers.ConnectExceptions.NONE);
+        }
+    }
+
+    public static class SecureNetty4GrpcServerTransportOptionalAuthIT extends SecureNetty4GrpcServerTransportIT {
+        public static class OptAuthMockSecurityPlugin extends MockSecurityPlugin {
+            public OptAuthMockSecurityPlugin() {}
+
+            @Override
+            public Optional<SecureSettingsFactory> getSecureSettingFactory(Settings settings) {
+                return Optional.of(new MockSecureSettingsFactory() {
+                    @Override
+                    public Optional<SecureAuxTransportSettingsProvider> getSecureAuxTransportSettingsProvider(Settings settings) {
+                        return Optional.of(getServerClientAuthOptional());
+                    }
+                });
+            }
+        }
+
+        @Override
+        protected Collection<Class<? extends Plugin>> nodePlugins() {
+            return List.of(GrpcPlugin.class, OptAuthMockSecurityPlugin.class);
+        }
+
+        public void testPlaintextClientConnect() throws Exception {
+            assertEquals(plaintextClientConnect(), SecureSettingsHelpers.ConnectExceptions.UNAVAILABLE);
+        }
+
+        public void testInsecureClientConnect() throws Exception {
+            assertEquals(insecureClientConnect(), SecureSettingsHelpers.ConnectExceptions.NONE);
+        }
+
+        public void testTrustedClientConnect() throws Exception {
+            assertEquals(trustedCertClientConnect(), SecureSettingsHelpers.ConnectExceptions.NONE);
+        }
+    }
+
+    public static class SecureNetty4GrpcServerTransportRequiredAuthIT extends SecureNetty4GrpcServerTransportIT {
+        public static class RequireAuthMockSecurityPlugin extends MockSecurityPlugin {
+            public RequireAuthMockSecurityPlugin() {}
+
+            @Override
+            public Optional<SecureSettingsFactory> getSecureSettingFactory(Settings settings) {
+                return Optional.of(new MockSecureSettingsFactory() {
+                    @Override
+                    public Optional<SecureAuxTransportSettingsProvider> getSecureAuxTransportSettingsProvider(Settings settings) {
+                        return Optional.of(getServerClientAuthRequired());
+                    }
+                });
+            }
+        }
+
+        @Override
+        protected Collection<Class<? extends Plugin>> nodePlugins() {
+            return List.of(GrpcPlugin.class, RequireAuthMockSecurityPlugin.class);
+        }
+
+        public void testPlaintextClientConnect() throws Exception {
+            assertEquals(plaintextClientConnect(), SecureSettingsHelpers.ConnectExceptions.UNAVAILABLE);
+        }
+
+        public void testInsecureClientConnect() throws Exception {
+            assertEquals(insecureClientConnect(), SecureSettingsHelpers.ConnectExceptions.BAD_CERT);
+        }
+
+        public void testTrustedClientConnect() throws Exception {
+            assertEquals(trustedCertClientConnect(), SecureSettingsHelpers.ConnectExceptions.NONE);
+        }
+    }
+}