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, 36 insertions, 8 deletions

diff --git a/core/src/main/java/org/elasticsearch/ingest/PipelineConfiguration.java b/core/src/main/java/org/elasticsearch/ingest/PipelineConfiguration.java
index 5cb9f6111f..3e874a0fd1 100644
--- a/core/src/main/java/org/elasticsearch/ingest/PipelineConfiguration.java
+++ b/core/src/main/java/org/elasticsearch/ingest/PipelineConfiguration.java
@@ -19,6 +19,7 @@
 
 package org.elasticsearch.ingest;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.cluster.AbstractDiffable;
 import org.elasticsearch.cluster.Diff;
 import org.elasticsearch.common.ParseField;
@@ -29,10 +30,13 @@ import org.elasticsearch.common.xcontent.ContextParser;
 import org.elasticsearch.common.xcontent.ObjectParser;
 import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.common.xcontent.XContentHelper;
+import org.elasticsearch.common.xcontent.XContentType;
 
 import java.io.IOException;
 import java.util.Map;
+import java.util.Objects;
 
 /**
  * Encapsulates a pipeline's id and configuration as a blob
@@ -45,8 +49,9 @@ public final class PipelineConfiguration extends AbstractDiffable<PipelineConfig
         PARSER.declareField((parser, builder, aVoid) -> {
             XContentBuilder contentBuilder = XContentBuilder.builder(parser.contentType().xContent());
             XContentHelper.copyCurrentStructure(contentBuilder.generator(), parser);
-            builder.setConfig(contentBuilder.bytes());
+            builder.setConfig(contentBuilder.bytes(), contentBuilder.contentType());
         }, new ParseField("config"), ObjectParser.ValueType.OBJECT);
+
     }
 
     public static ContextParser<Void, PipelineConfiguration> getParser() {
@@ -56,17 +61,19 @@ public final class PipelineConfiguration extends AbstractDiffable<PipelineConfig
 
         private String id;
         private BytesReference config;
+        private XContentType xContentType;
 
         void setId(String id) {
             this.id = id;
         }
 
-        void setConfig(BytesReference config) {
+        void setConfig(BytesReference config, XContentType xContentType) {
             this.config = config;
+            this.xContentType = xContentType;
         }
 
         PipelineConfiguration build() {
-            return new PipelineConfiguration(id, config);
+            return new PipelineConfiguration(id, config, xContentType);
         }
     }
 
@@ -75,10 +82,12 @@ public final class PipelineConfiguration extends AbstractDiffable<PipelineConfig
     // and the way the map of maps config is read requires a deep copy (it removes instead of gets entries to check for unused options)
     // also the get pipeline api just directly returns this to the caller
     private final BytesReference config;
+    private final XContentType xContentType;
 
-    public PipelineConfiguration(String id, BytesReference config) {
-        this.id = id;
-        this.config = config;
+    public PipelineConfiguration(String id, BytesReference config, XContentType xContentType) {
+        this.id = Objects.requireNonNull(id);
+        this.config = Objects.requireNonNull(config);
+        this.xContentType = Objects.requireNonNull(xContentType);
     }
 
     public String getId() {
@@ -86,7 +95,17 @@ public final class PipelineConfiguration extends AbstractDiffable<PipelineConfig
     }
 
     public Map<String, Object> getConfigAsMap() {
-        return XContentHelper.convertToMap(config, true).v2();
+        return XContentHelper.convertToMap(config, true, xContentType).v2();
+    }
+
+    // pkg-private for tests
+    XContentType getXContentType() {
+        return xContentType;
+    }
+
+    // pkg-private for tests
+    BytesReference getConfig() {
+        return config;
     }
 
     @Override
@@ -99,7 +118,13 @@ public final class PipelineConfiguration extends AbstractDiffable<PipelineConfig
     }
 
     public static PipelineConfiguration readFrom(StreamInput in) throws IOException {
-        return new PipelineConfiguration(in.readString(), in.readBytesReference());
+        if (in.getVersion().after(Version.V_5_3_0_UNRELEASED)) { // TODO update to onOrAfter after backporting
+            return new PipelineConfiguration(in.readString(), in.readBytesReference(), XContentType.readFrom(in));
+        } else {
+            final String id = in.readString();
+            final BytesReference config = in.readBytesReference();
+            return new PipelineConfiguration(id, config, XContentFactory.xContentType(config));
+        }
     }
 
     public static Diff<PipelineConfiguration> readDiffFrom(StreamInput in) throws IOException {
@@ -110,6 +135,9 @@ public final class PipelineConfiguration extends AbstractDiffable<PipelineConfig
     public void writeTo(StreamOutput out) throws IOException {
         out.writeString(id);
         out.writeBytesReference(config);
+        if (out.getVersion().after(Version.V_5_3_0_UNRELEASED)) { // TODO update to onOrAfter after backporting
+            xContentType.writeTo(out);
+        }
     }
 
     @Override