Allow customization of netty channel handles before and during decompression

CHANGELOG.md

@@ -92,6 +92,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Add Doc Status Counter for Indexing Engine ([#4562](https://github.com/opensearch-project/OpenSearch/issues/4562))
 - Add unreferenced file cleanup count to merge stats ([#10204](https://github.com/opensearch-project/OpenSearch/pull/10204))
 - [Remote Store] Add support to restrict creation & deletion if system repository and mutation of immutable settings of system repository ([#9839](https://github.com/opensearch-project/OpenSearch/pull/9839))
+- Improve compressed request handling ([#10261](https://github.com/opensearch-project/OpenSearch/pull/10261))
 
 ### Dependencies
 - Bump `peter-evans/create-or-update-comment` from 2 to 3 ([#9575](https://github.com/opensearch-project/OpenSearch/pull/9575))

modules/transport-netty4/src/internalClusterTest/java/org/opensearch/http/netty4/Netty4BlockingPlugin.java

@@ -0,0 +1,128 @@
+/*
+ * 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.netty4;
+
+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.common.util.PageCacheRecycler;
+import org.opensearch.core.indices.breaker.CircuitBreakerService;
+import org.opensearch.core.xcontent.NamedXContentRegistry;
+import org.opensearch.http.HttpServerTransport;
+import org.opensearch.telemetry.tracing.Tracer;
+import org.opensearch.threadpool.ThreadPool;
+import org.opensearch.transport.Netty4ModulePlugin;
+import org.opensearch.transport.SharedGroupFactory;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.Map;
+import java.util.function.Supplier;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.handler.codec.http.DefaultFullHttpResponse;
+import io.netty.handler.codec.http.DefaultHttpRequest;
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.HttpRequest;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.netty.util.ReferenceCountUtil;
+
+public class Netty4BlockingPlugin extends Netty4ModulePlugin {
+
+    public class Netty4BlockingHttpServerTransport extends Netty4HttpServerTransport {
+
+        public Netty4BlockingHttpServerTransport(
+            Settings settings,
+            NetworkService networkService,
+            BigArrays bigArrays,
+            ThreadPool threadPool,
+            NamedXContentRegistry xContentRegistry,
+            Dispatcher dispatcher,
+            ClusterSettings clusterSettings,
+            SharedGroupFactory sharedGroupFactory,
+            Tracer tracer
+        ) {
+            super(
+                settings,
+                networkService,
+                bigArrays,
+                threadPool,
+                xContentRegistry,
+                dispatcher,
+                clusterSettings,
+                sharedGroupFactory,
+                tracer
+            );
+        }
+
+        @Override
+        protected ChannelInboundHandlerAdapter createHeaderVerifier() {
+            return new ExampleBlockingNetty4HeaderVerifier();
+        }
+    }
+
+    @Override
+    public Map<String, Supplier<HttpServerTransport>> getHttpTransports(
+        Settings settings,
+        ThreadPool threadPool,
+        BigArrays bigArrays,
+        PageCacheRecycler pageCacheRecycler,
+        CircuitBreakerService circuitBreakerService,
+        NamedXContentRegistry xContentRegistry,
+        NetworkService networkService,
+        HttpServerTransport.Dispatcher dispatcher,
+        ClusterSettings clusterSettings,
+        Tracer tracer
+    ) {
+        return Collections.singletonMap(
+            NETTY_HTTP_TRANSPORT_NAME,
+            () -> new Netty4BlockingHttpServerTransport(
+                settings,
+                networkService,
+                bigArrays,
+                threadPool,
+                xContentRegistry,
+                dispatcher,
+                clusterSettings,
+                getSharedGroupFactory(settings),
+                tracer
+            )
+        );
+    }
+
+    /** POC for how an external header verifier would be implemented */
+    public class ExampleBlockingNetty4HeaderVerifier extends SimpleChannelInboundHandler<DefaultHttpRequest> {
+
+        @Override
+        public void channelRead0(ChannelHandlerContext ctx, DefaultHttpRequest msg) throws Exception {
+            ReferenceCountUtil.retain(msg);
+            if (isBlocked(msg)) {
+                ByteBuf buf = Unpooled.copiedBuffer("Hit header_verifier".getBytes(StandardCharsets.UTF_8));
+                final FullHttpResponse response = new DefaultFullHttpResponse(msg.protocolVersion(), HttpResponseStatus.UNAUTHORIZED, buf);
+                ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
+                ReferenceCountUtil.release(msg);
+            } else {
+                // Lets the request pass to the next channel handler
+                ctx.fireChannelRead(msg);
+            }
+        }
+
+        private boolean isBlocked(HttpRequest request) {
+            final boolean shouldBlock = request.headers().contains("blockme");
+
+            return shouldBlock;
+        }
+    }
+}

modules/transport-netty4/src/internalClusterTest/java/org/opensearch/http/netty4/Netty4HeaderVerifierIT.java

@@ -0,0 +1,74 @@
+/*
+ * 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.netty4;
+
+import org.opensearch.OpenSearchNetty4IntegTestCase;
+import org.opensearch.core.common.transport.TransportAddress;
+import org.opensearch.http.HttpServerTransport;
+import org.opensearch.plugins.Plugin;
+import org.opensearch.test.OpenSearchIntegTestCase.ClusterScope;
+import org.opensearch.test.OpenSearchIntegTestCase.Scope;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+import io.netty.buffer.ByteBufUtil;
+import io.netty.handler.codec.http.DefaultFullHttpRequest;
+import io.netty.handler.codec.http.FullHttpRequest;
+import io.netty.handler.codec.http.FullHttpResponse;
+import io.netty.handler.codec.http.HttpMethod;
+import io.netty.handler.codec.http.HttpVersion;
+import io.netty.handler.codec.http2.HttpConversionUtil;
+import io.netty.util.ReferenceCounted;
+
+import static org.hamcrest.CoreMatchers.containsString;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static io.netty.handler.codec.http.HttpHeaderNames.HOST;
+
+@ClusterScope(scope = Scope.TEST, supportsDedicatedMasters = false, numDataNodes = 1)
+public class Netty4HeaderVerifierIT extends OpenSearchNetty4IntegTestCase {
+
+    @Override
+    protected boolean addMockHttpTransport() {
+        return false; // enable http
+    }
+
+    @Override
+    protected Collection<Class<? extends Plugin>> nodePlugins() {
+        return Collections.singletonList(Netty4BlockingPlugin.class);
+    }
+
+    public void testThatNettyHttpServerRequestBlockedWithHeaderVerifier() throws Exception {
+        HttpServerTransport httpServerTransport = internalCluster().getInstance(HttpServerTransport.class);
+        TransportAddress[] boundAddresses = httpServerTransport.boundAddress().boundAddresses();
+        TransportAddress transportAddress = randomFrom(boundAddresses);
+
+        final FullHttpRequest blockedRequest = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, "/");
+        blockedRequest.headers().add("blockme", "Not Allowed");
+        blockedRequest.headers().add(HOST, "localhost");
+        blockedRequest.headers().add(HttpConversionUtil.ExtensionHeaderNames.SCHEME.text(), "http");
+
+        final List<FullHttpResponse> responses = new ArrayList<>();
+        try (Netty4HttpClient nettyHttpClient = Netty4HttpClient.http2()) {
+            try {
+                FullHttpResponse blockedResponse = nettyHttpClient.send(transportAddress.address(), blockedRequest);
+                responses.add(blockedResponse);
+                String blockedResponseContent = new String(ByteBufUtil.getBytes(blockedResponse.content()), StandardCharsets.UTF_8);
+                assertThat(blockedResponseContent, containsString("Hit header_verifier"));
+                assertThat(blockedResponse.status().code(), equalTo(401));
+            } finally {
+                responses.forEach(ReferenceCounted::release);
+            }
+        }
+    }
+
+}

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

@@ -0,0 +1,142 @@
+/*
+ * 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.netty4;
+
+import org.opensearch.rest.RestRequest;
+
+import java.util.AbstractMap;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import io.netty.handler.codec.http.HttpHeaders;
+import io.netty.handler.codec.http.HttpMethod;
+import io.netty.handler.codec.http.HttpRequest;
+
+public abstract class AbstractNetty4HttpRequest {
+
+    protected HttpHeadersMap headers;
+    protected Exception inboundException;
+
+    protected RestRequest.Method getHttpMethod(HttpRequest request) {
+        HttpMethod httpMethod = request.method();
+        if (httpMethod == HttpMethod.GET) return RestRequest.Method.GET;
+
+        if (httpMethod == HttpMethod.POST) return RestRequest.Method.POST;
+
+        if (httpMethod == HttpMethod.PUT) return RestRequest.Method.PUT;
+
+        if (httpMethod == HttpMethod.DELETE) return RestRequest.Method.DELETE;
+
+        if (httpMethod == HttpMethod.HEAD) {
+            return RestRequest.Method.HEAD;
+        }
+
+        if (httpMethod == HttpMethod.OPTIONS) {
+            return RestRequest.Method.OPTIONS;
+        }
+
+        if (httpMethod == HttpMethod.PATCH) {
+            return RestRequest.Method.PATCH;
+        }
+
+        if (httpMethod == HttpMethod.TRACE) {
+            return RestRequest.Method.TRACE;
+        }
+
+        if (httpMethod == HttpMethod.CONNECT) {
+            return RestRequest.Method.CONNECT;
+        }
+
+        throw new IllegalArgumentException("Unexpected http method: " + httpMethod);
+    }
+
+    /**
+     * A wrapper of {@link HttpHeaders} that implements a map to prevent copying unnecessarily. This class does not support modifications
+     * and due to the underlying implementation, it performs case insensitive lookups of key to values.
+     *
+     * It is important to note that this implementation does have some downsides in that each invocation of the
+     * {@link #values()} and {@link #entrySet()} methods will perform a copy of the values in the HttpHeaders rather than returning a
+     * view of the underlying values.
+     */
+    protected static class HttpHeadersMap implements Map<String, List<String>> {
+
+        private final HttpHeaders httpHeaders;
+
+        HttpHeadersMap(HttpHeaders httpHeaders) {
+            this.httpHeaders = httpHeaders;
+        }
+
+        @Override
+        public int size() {
+            return httpHeaders.size();
+        }
+
+        @Override
+        public boolean isEmpty() {
+            return httpHeaders.isEmpty();
+        }
+
+        @Override
+        public boolean containsKey(Object key) {
+            return key instanceof String && httpHeaders.contains((String) key);
+        }
+
+        @Override
+        public boolean containsValue(Object value) {
+            return value instanceof List && httpHeaders.names().stream().map(httpHeaders::getAll).anyMatch(value::equals);
+        }
+
+        @Override
+        public List<String> get(Object key) {
+            return key instanceof String ? httpHeaders.getAll((String) key) : null;
+        }
+
+        @Override
+        public List<String> put(String key, List<String> value) {
+            throw new UnsupportedOperationException("modifications are not supported");
+        }
+
+        @Override
+        public List<String> remove(Object key) {
+            throw new UnsupportedOperationException("modifications are not supported");
+        }
+
+        @Override
+        public void putAll(Map<? extends String, ? extends List<String>> m) {
+            throw new UnsupportedOperationException("modifications are not supported");
+        }
+
+        @Override
+        public void clear() {
+            throw new UnsupportedOperationException("modifications are not supported");
+        }
+
+        @Override
+        public Set<String> keySet() {
+            return httpHeaders.names();
+        }
+
+        @Override
+        public Collection<List<String>> values() {
+            return httpHeaders.names().stream().map(k -> Collections.unmodifiableList(httpHeaders.getAll(k))).collect(Collectors.toList());
+        }
+
+        @Override
+        public Set<Entry<String, List<String>>> entrySet() {
+            return httpHeaders.names()
+                .stream()
+                .map(k -> new AbstractMap.SimpleImmutableEntry<>(k, httpHeaders.getAll(k)))
+                .collect(Collectors.toSet());
+        }
+    }
+}