Make sure Netty4Http3ServerTransport uses configured HeaderVerifier and Decompressor instances (opensearch-project#6108)

Signed-off-by: Andriy Redko <drreta@gmail.com>
(cherry picked from commit 9d1b44e)
Signed-off-by: Andriy Redko <drreta@gmail.com>

Author: reta

CHANGELOG.md

@@ -26,7 +26,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Fix the issue of unprocessed X-Request-Id ([#5954](https://github.com/opensearch-project/security/pull/5954))
 - Fix audit log `NONE` sentinel not respected for `disabled_rest_categories`, `disabled_transport_categories`, and `ignore_users` in dynamic configuration ([#6021](https://github.com/opensearch-project/security/pull/6021))
 - Improve DLS error message to identify undefined user attributes when query substitution fails ([#5975](https://github.com/opensearch-project/security/pull/5975))
-- Fix span propagation issue for tracing([#6006](https://github.com/opensearch-project/security/pull/6006))
+- Fix span propagation issue for tracing ([#6006](https://github.com/opensearch-project/security/pull/6006))
+- Make sure Netty4Http3ServerTransport uses configured HeaderVerifier and Decompressor instances ([#6121](https://github.com/opensearch-project/security/pull/6121))
 
 ### Refactoring
 

build.gradle

@@ -610,6 +610,8 @@ allprojects {
                 integrationTestImplementation ('com.jayway.jsonpath:json-path:2.10.0') {
                     exclude(group: 'net.minidev', module: 'json-smart')
                 }
+                integrationTestImplementation "io.projectreactor.netty:reactor-netty-core:${versions.reactor_netty}"
+                integrationTestImplementation "io.projectreactor.netty:reactor-netty-http:${versions.reactor_netty}"
             }
         }
     }

src/integrationTest/java/org/opensearch/security/ResourceFocusedTests.java

@@ -13,6 +13,9 @@
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
@@ -31,14 +34,22 @@
 import org.junit.Test;
 
 import org.opensearch.action.index.IndexRequest;
+import org.opensearch.common.collect.Tuple;
+import org.opensearch.common.settings.Settings;
+import org.opensearch.http.HttpTransportSettings;
 import org.opensearch.test.framework.AsyncActions;
 import org.opensearch.test.framework.TestSecurityConfig;
 import org.opensearch.test.framework.TestSecurityConfig.User;
 import org.opensearch.test.framework.cluster.ClusterManager;
 import org.opensearch.test.framework.cluster.LocalCluster;
+import org.opensearch.test.framework.cluster.ReactorHttpClient;
 import org.opensearch.test.framework.cluster.TestRestClient;
 import org.opensearch.transport.client.Client;
 
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import reactor.netty.http.HttpProtocol;
+
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.equalTo;
 import static org.opensearch.action.support.WriteRequest.RefreshPolicy.IMMEDIATE;
@@ -64,11 +75,13 @@ public class ResourceFocusedTests {
             )
             .on("*")
     );
+    private static Map<String, Object> NODE_SETTINGS = Map.of(HttpTransportSettings.SETTING_HTTP_HTTP3_ENABLED.getKey(), true);
 
     @ClassRule
     public static LocalCluster cluster = new LocalCluster.Builder().clusterManager(ClusterManager.THREE_CLUSTER_MANAGERS)
         .authc(AUTHC_HTTPBASIC_INTERNAL)
         .users(ADMIN_USER, LIMITED_USER)
+        .nodeSettings(NODE_SETTINGS)
         .anonymousAuth(false)
         .doNotFailOnForbidden(true)
         .build();
@@ -86,38 +99,41 @@ public void testUnauthenticatedFewBig() {
         // Tweaks:
         final RequestBodySize size = RequestBodySize.XLarge;
         final String requestPath = "/*/_search";
-        final int parrallelism = 5;
+        final int parallelism = 5;
         final int totalNumberOfRequests = 100;
 
-        runResourceTest(size, requestPath, parrallelism, totalNumberOfRequests);
+        runResourceTest(size, requestPath, parallelism, totalNumberOfRequests);
+        runResourceTestWithGenericClient(size, requestPath, parallelism, totalNumberOfRequests);
     }
 
     @Test
     public void testUnauthenticatedManyMedium() {
         // Tweaks:
         final RequestBodySize size = RequestBodySize.Medium;
         final String requestPath = "/*/_search";
-        final int parrallelism = 20;
+        final int parallelism = 20;
         final int totalNumberOfRequests = 10_000;
 
-        runResourceTest(size, requestPath, parrallelism, totalNumberOfRequests);
+        runResourceTest(size, requestPath, parallelism, totalNumberOfRequests);
+        runResourceTestWithGenericClient(size, requestPath, parallelism, totalNumberOfRequests);
     }
 
     @Test
     public void testUnauthenticatedTonsSmall() {
         // Tweaks:
         final RequestBodySize size = RequestBodySize.Small;
         final String requestPath = "/*/_search";
-        final int parrallelism = 100;
+        final int parallelism = 100;
         final int totalNumberOfRequests = 15_000;
 
-        runResourceTest(size, requestPath, parrallelism, totalNumberOfRequests);
+        runResourceTest(size, requestPath, parallelism, totalNumberOfRequests);
+        runResourceTestWithGenericClient(size, requestPath, parallelism, totalNumberOfRequests);
     }
 
     private void runResourceTest(
         final RequestBodySize size,
         final String requestPath,
-        final int parrallelism,
+        final int parallelism,
         final int totalNumberOfRequests
     ) {
         final byte[] compressedRequestBody = createCompressedRequestBody(size);
@@ -127,14 +143,40 @@ private void runResourceTest(
                 post.setEntity(new ByteArrayEntity(compressedRequestBody, ContentType.APPLICATION_JSON));
                 TestRestClient.HttpResponse response = client.executeRequest(post);
                 return response.getStatusCode();
-            }, parrallelism, totalNumberOfRequests);
+            }, parallelism, totalNumberOfRequests);
 
             AsyncActions.getAll(requests, 2, TimeUnit.MINUTES).forEach((responseCode) -> {
                 assertThat(responseCode, equalTo(HttpStatus.SC_UNAUTHORIZED));
             });
         }
     }
 
+    private void runResourceTestWithGenericClient(
+        final RequestBodySize size,
+        final String requestPath,
+        final int parallelism,
+        final int totalNumberOfRequests
+    ) {
+        final byte[] compressedRequestBody = createCompressedRequestBody(size);
+        try (
+            final ReactorHttpClient client = cluster.getGenericClient(
+                HttpProtocol.HTTP3,
+                true,
+                Settings.builder().loadFromMap(NODE_SETTINGS).build()
+            )
+        ) {
+            List<Tuple<String, byte[]>> requestUris = new ArrayList<>();
+            for (int i = 0; i < totalNumberOfRequests; i++) {
+                requestUris.add(Tuple.tuple(requestPath, compressedRequestBody));
+            }
+
+            final Collection<FullHttpResponse> responses = client.post(requestUris, parallelism);
+            responses.stream()
+                .map(FullHttpResponse::status)
+                .forEach(responseCode -> assertThat(responseCode, equalTo(HttpResponseStatus.UNAUTHORIZED)));
+        }
+    }
+
     static enum RequestBodySize {
         Small(1),
         Medium(1_000),

src/integrationTest/java/org/opensearch/test/framework/cluster/OpenSearchClientProvider.java

@@ -65,10 +65,13 @@
 import org.opensearch.client.RestClient;
 import org.opensearch.client.RestClientBuilder;
 import org.opensearch.client.RestHighLevelClient;
+import org.opensearch.common.settings.Settings;
 import org.opensearch.security.support.PemKeyReader;
 import org.opensearch.test.framework.certificate.CertificateData;
 import org.opensearch.test.framework.certificate.TestCertificates;
 
+import reactor.netty.http.HttpProtocol;
+
 import static org.opensearch.test.framework.cluster.TestRestClientConfiguration.getBasicAuthHeader;
 
 /**
@@ -226,13 +229,19 @@ default TestRestClient getRestClient(CertificateData useCertificateData, Header.
         return getRestClient(Arrays.asList(headers), useCertificateData);
     }
 
+    /**
+     * Returns a generic HTTP/1.1/HTTP 2.0/HTTP 3.0 client.
+     */
+    default ReactorHttpClient getGenericClient(HttpProtocol protocol, boolean secure, Settings settings) {
+        return new ReactorHttpClient(true, true, settings, getHttpAddress());
+    }
+
     default TestRestClient getRestClient(Header... headers) {
         return getRestClient((CertificateData) null, headers);
     }
 
     default TestRestClient getRestClient(List<Header> headers) {
         return createGenericClientRestClient(new TestRestClientConfiguration().headers(headers));
-
     }
 
     default TestRestClient getRestClient(List<Header> headers, CertificateData useCertificateData) {