Optionally require a valid content type for all rest requests with content (#22691)

This change adds a strict mode for xcontent parsing on the rest layer. The strict mode will be off by default for 5.x and in a separate commit will be enabled by default for 6.0. The strict mode, which can be enabled by setting `http.content_type.required: true` in 5.x, will require that all incoming rest requests have a valid and supported content type header before the request is dispatched. In the non-strict mode, the Content-Type header will be inspected and if it is not present or not valid, we will continue with auto detection of content like we have done previously.

The content type header is parsed to the matching XContentType value with the only exception being for plain text requests. This value is then passed on with the content bytes so that we can reduce the number of places where we need to auto-detect the content type.

As part of this, many transport requests and builders were updated to provide methods that
accepted the XContentType along with the bytes and the methods that would rely on auto-detection have been deprecated.

In the non-strict mode, deprecation warnings are issued whenever a request with body doesn't provide the Content-Type header.

See #19388 

1 files changed, 97 insertions, 9 deletions

diff --git a/modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpRequest.java b/modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpRequest.java
index 10a2da21ac..c3b37ef7be 100644
--- a/modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpRequest.java
+++ b/modules/transport-netty4/src/main/java/org/elasticsearch/http/netty4/Netty4HttpRequest.java
@@ -21,6 +21,7 @@ package org.elasticsearch.http.netty4;
 
 import io.netty.channel.Channel;
 import io.netty.handler.codec.http.FullHttpRequest;
+import io.netty.handler.codec.http.HttpHeaders;
 import io.netty.handler.codec.http.HttpMethod;
 
 import org.elasticsearch.common.bytes.BytesArray;
@@ -30,7 +31,12 @@ import org.elasticsearch.rest.RestRequest;
 import org.elasticsearch.transport.netty4.Netty4Utils;
 
 import java.net.SocketAddress;
+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;
 
 public class Netty4HttpRequest extends RestRequest {
 
@@ -39,7 +45,7 @@ public class Netty4HttpRequest extends RestRequest {
     private final BytesReference content;
 
     Netty4HttpRequest(NamedXContentRegistry xContentRegistry, FullHttpRequest request, Channel channel) {
-        super(xContentRegistry, request.uri());
+        super(xContentRegistry, request.uri(), new HttpHeadersMap(request.headers()));
         this.request = request;
         this.channel = channel;
         if (request.content().isReadable()) {
@@ -120,14 +126,96 @@ public class Netty4HttpRequest extends RestRequest {
         return channel;
     }
 
-    @Override
-    public String header(String name) {
-        return request.headers().get(name);
-    }
+    /**
+     * 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.
+     */
+    private static class HttpHeadersMap implements Map<String, List<String>> {
 
-    @Override
-    public Iterable<Map.Entry<String, String>> headers() {
-        return request.headers().entries();
-    }
+        private final HttpHeaders httpHeaders;
+
+        private 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 Entry<String, List<String>>() {
+
+                @Override
+                public String getKey() {
+                    return k;
+                }
+
+                @Override
+                public List<String> getValue() {
+                    return httpHeaders.getAll(k);
+                }
+
+                @Override
+                public List<String> setValue(List<String> value) {
+                    throw new UnsupportedOperationException("modifications are not supported");
+                }
+            }).collect(Collectors.toSet());
+        }
+    }
 }