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, 86 insertions, 29 deletions

diff --git a/core/src/main/java/org/elasticsearch/action/index/IndexRequest.java b/core/src/main/java/org/elasticsearch/action/index/IndexRequest.java
index d1a9927c67..7e3f14368d 100644
--- a/core/src/main/java/org/elasticsearch/action/index/IndexRequest.java
+++ b/core/src/main/java/org/elasticsearch/action/index/IndexRequest.java
@@ -47,6 +47,7 @@ import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Objects;
 
 import static org.elasticsearch.action.ValidateActions.addValidationError;
 
@@ -55,10 +56,10 @@ import static org.elasticsearch.action.ValidateActions.addValidationError;
  * created using {@link org.elasticsearch.client.Requests#indexRequest(String)}.
  *
  * The index requires the {@link #index()}, {@link #type(String)}, {@link #id(String)} and
- * {@link #source(byte[])} to be set.
+ * {@link #source(byte[], XContentType)} to be set.
  *
- * The source (content to index) can be set in its bytes form using ({@link #source(byte[])}),
- * its string form ({@link #source(String)}) or using a {@link org.elasticsearch.common.xcontent.XContentBuilder}
+ * The source (content to index) can be set in its bytes form using ({@link #source(byte[], XContentType)}),
+ * its string form ({@link #source(String, XContentType)}) or using a {@link org.elasticsearch.common.xcontent.XContentBuilder}
  * ({@link #source(org.elasticsearch.common.xcontent.XContentBuilder)}).
  *
  * If the {@link #id(String)} is not set, it will be automatically generated.
@@ -83,7 +84,7 @@ public class IndexRequest extends ReplicatedWriteRequest<IndexRequest> implement
     private long version = Versions.MATCH_ANY;
     private VersionType versionType = VersionType.INTERNAL;
 
-    private XContentType contentType = Requests.INDEX_CONTENT_TYPE;
+    private XContentType contentType;
 
     private String pipeline;
 
@@ -103,7 +104,7 @@ public class IndexRequest extends ReplicatedWriteRequest<IndexRequest> implement
 
     /**
      * Constructs a new index request against the specific index. The {@link #type(String)}
-     * {@link #source(byte[])} must be set.
+     * {@link #source(byte[], XContentType)} must be set.
      */
     public IndexRequest(String index) {
         this.index = index;
@@ -140,7 +141,9 @@ public class IndexRequest extends ReplicatedWriteRequest<IndexRequest> implement
         if (source == null) {
             validationException = addValidationError("source is missing", validationException);
         }
-
+        if (contentType == null) {
+            validationException = addValidationError("content type is missing", validationException);
+        }
         final long resolvedVersion = resolveVersionDefaults();
         if (opType() == OpType.CREATE) {
             if (versionType != VersionType.INTERNAL) {
@@ -179,21 +182,14 @@ public class IndexRequest extends ReplicatedWriteRequest<IndexRequest> implement
     }
 
     /**
-     * The content type that will be used when generating a document from user provided objects like Maps.
+     * The content type. This will be used when generating a document from user provided objects like Maps and when parsing the
+     * source at index time
      */
     public XContentType getContentType() {
         return contentType;
     }
 
     /**
-     * Sets the content type that will be used when generating a document from user provided objects (like Map).
-     */
-    public IndexRequest contentType(XContentType contentType) {
-        this.contentType = contentType;
-        return this;
-    }
-
-    /**
      * The type of the indexed document.
      */
     @Override
@@ -284,16 +280,16 @@ public class IndexRequest extends ReplicatedWriteRequest<IndexRequest> implement
     }
 
     public Map<String, Object> sourceAsMap() {
-        return XContentHelper.convertToMap(source, false).v2();
+        return XContentHelper.convertToMap(source, false, contentType).v2();
     }
 
     /**
-     * Index the Map as a {@link org.elasticsearch.client.Requests#INDEX_CONTENT_TYPE}.
+     * Index the Map in {@link Requests#INDEX_CONTENT_TYPE} format
      *
      * @param source The map to index
      */
     public IndexRequest source(Map source) throws ElasticsearchGenerationException {
-        return source(source, contentType);
+        return source(source, Requests.INDEX_CONTENT_TYPE);
     }
 
     /**
@@ -314,24 +310,32 @@ public class IndexRequest extends ReplicatedWriteRequest<IndexRequest> implement
     /**
      * Sets the document source to index.
      *
-     * Note, its preferable to either set it using {@link #source(org.elasticsearch.common.xcontent.XContentBuilder)}
-     * or using the {@link #source(byte[])}.
+     * @deprecated use {@link #source(String, XContentType)}
      */
+    @Deprecated
     public IndexRequest source(String source) {
-        this.source = new BytesArray(source.getBytes(StandardCharsets.UTF_8));
-        return this;
+        return source(new BytesArray(source), XContentFactory.xContentType(source));
+    }
+
+    /**
+     * Sets the document source to index.
+     *
+     * Note, its preferable to either set it using {@link #source(org.elasticsearch.common.xcontent.XContentBuilder)}
+     * or using the {@link #source(byte[], XContentType)}.
+     */
+    public IndexRequest source(String source, XContentType xContentType) {
+        return source(new BytesArray(source), xContentType);
     }
 
     /**
      * Sets the content source to index.
      */
     public IndexRequest source(XContentBuilder sourceBuilder) {
-        source = sourceBuilder.bytes();
-        return this;
+        return source(sourceBuilder.bytes(), sourceBuilder.contentType());
     }
 
     /**
-     * Sets the content source to index.
+     * Sets the content source to index using the default content type ({@link Requests#INDEX_CONTENT_TYPE})
      * <p>
      * <b>Note: the number of objects passed to this method must be an even
      * number. Also the first argument in each pair (the field name) must have a
@@ -339,6 +343,18 @@ public class IndexRequest extends ReplicatedWriteRequest<IndexRequest> implement
      * </p>
      */
     public IndexRequest source(Object... source) {
+        return source(Requests.INDEX_CONTENT_TYPE, source);
+    }
+
+    /**
+     * Sets the content source to index.
+     * <p>
+     * <b>Note: the number of objects passed to this method as varargs must be an even
+     * number. Also the first argument in each pair (the field name) must have a
+     * valid String representation.</b>
+     * </p>
+     */
+    public IndexRequest source(XContentType xContentType, Object... source) {
         if (source.length % 2 != 0) {
             throw new IllegalArgumentException("The number of object passed must be even but was [" + source.length + "]");
         }
@@ -346,7 +362,7 @@ public class IndexRequest extends ReplicatedWriteRequest<IndexRequest> implement
             throw new IllegalArgumentException("you are using the removed method for source with bytes and unsafe flag, the unsafe flag was removed, please just use source(BytesReference)");
         }
         try {
-            XContentBuilder builder = XContentFactory.contentBuilder(contentType);
+            XContentBuilder builder = XContentFactory.contentBuilder(xContentType);
             builder.startObject();
             for (int i = 0; i < source.length; i++) {
                 builder.field(source[i++].toString(), source[i]);
@@ -360,30 +376,63 @@ public class IndexRequest extends ReplicatedWriteRequest<IndexRequest> implement
 
     /**
      * Sets the document to index in bytes form.
+     * @deprecated use {@link #source(BytesReference, XContentType)}
      */
+    @Deprecated
     public IndexRequest source(BytesReference source) {
-        this.source = source;
+        return source(source, XContentFactory.xContentType(source));
+
+    }
+
+    /**
+     * Sets the document to index in bytes form.
+     */
+    public IndexRequest source(BytesReference source, XContentType xContentType) {
+        this.source = Objects.requireNonNull(source);
+        this.contentType = Objects.requireNonNull(xContentType);
         return this;
     }
 
     /**
      * Sets the document to index in bytes form.
+     * @deprecated use {@link #source(byte[], XContentType)}
      */
+    @Deprecated
     public IndexRequest source(byte[] source) {
         return source(source, 0, source.length);
     }
 
     /**
+     * Sets the document to index in bytes form.
+     */
+    public IndexRequest source(byte[] source, XContentType xContentType) {
+        return source(source, 0, source.length, xContentType);
+    }
+
+    /**
      * Sets the document to index in bytes form (assumed to be safe to be used from different
      * threads).
      *
      * @param source The source to index
      * @param offset The offset in the byte array
      * @param length The length of the data
+     * @deprecated use {@link #source(byte[], int, int, XContentType)}
      */
+    @Deprecated
     public IndexRequest source(byte[] source, int offset, int length) {
-        this.source = new BytesArray(source, offset, length);
-        return this;
+        return source(new BytesArray(source, offset, length), XContentFactory.xContentType(source));
+    }
+
+    /**
+     * Sets the document to index in bytes form (assumed to be safe to be used from different
+     * threads).
+     *
+     * @param source The source to index
+     * @param offset The offset in the byte array
+     * @param length The length of the data
+     */
+    public IndexRequest source(byte[] source, int offset, int length, XContentType xContentType) {
+        return source(new BytesArray(source, offset, length), xContentType);
     }
 
     /**
@@ -515,6 +564,11 @@ public class IndexRequest extends ReplicatedWriteRequest<IndexRequest> implement
         pipeline = in.readOptionalString();
         isRetry = in.readBoolean();
         autoGeneratedTimestamp = in.readLong();
+        if (in.getVersion().after(Version.V_5_3_0_UNRELEASED)) { // TODO update to onOrAfter after backporting
+            contentType = in.readOptionalWriteable(XContentType::readFrom);
+        } else {
+            contentType = XContentFactory.xContentType(source);
+        }
     }
 
     @Override
@@ -543,6 +597,9 @@ public class IndexRequest extends ReplicatedWriteRequest<IndexRequest> implement
         out.writeOptionalString(pipeline);
         out.writeBoolean(isRetry);
         out.writeLong(autoGeneratedTimestamp);
+        if (out.getVersion().after(Version.V_5_3_0_UNRELEASED)) { // TODO update to onOrAfter after backporting
+            out.writeOptionalWriteable(contentType);
+        }
     }
 
     @Override