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 

13 files changed, 63 insertions, 39 deletions

diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestPutStoredScriptAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestPutStoredScriptAction.java
index cee46eba04..358c3656ce 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestPutStoredScriptAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/cluster/RestPutStoredScriptAction.java
@@ -65,7 +65,7 @@ public class RestPutStoredScriptAction extends BaseRestHandler {
                 "specifying lang [" + lang + "] as part of the url path is deprecated, use request content instead");
         }
 
-        PutStoredScriptRequest putRequest = new PutStoredScriptRequest(id, lang, content);
+        PutStoredScriptRequest putRequest = new PutStoredScriptRequest(id, lang, content, request.getXContentType());
         return channel -> client.admin().cluster().putStoredScript(putRequest, new AcknowledgedRestListener<>(channel));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestCreateIndexAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestCreateIndexAction.java
index ea0cf7bf0c..8c314efee6 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestCreateIndexAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestCreateIndexAction.java
@@ -42,7 +42,7 @@ public class RestCreateIndexAction extends BaseRestHandler {
     public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         CreateIndexRequest createIndexRequest = new CreateIndexRequest(request.param("index"));
         if (request.hasContent()) {
-            createIndexRequest.source(request.content());
+            createIndexRequest.source(request.content(), request.getXContentType());
         }
         createIndexRequest.updateAllTypes(request.paramAsBoolean("update_all_types", false));
         createIndexRequest.timeout(request.paramAsTime("timeout", createIndexRequest.timeout()));
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestPutIndexTemplateAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestPutIndexTemplateAction.java
index c48ed3716d..b376f3dab3 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestPutIndexTemplateAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestPutIndexTemplateAction.java
@@ -57,7 +57,7 @@ public class RestPutIndexTemplateAction extends BaseRestHandler {
         putRequest.masterNodeTimeout(request.paramAsTime("master_timeout", putRequest.masterNodeTimeout()));
         putRequest.create(request.paramAsBoolean("create", false));
         putRequest.cause(request.param("cause", ""));
-        putRequest.source(request.content());
+        putRequest.source(request.content(), request.getXContentType());
         return channel -> client.admin().indices().putTemplate(putRequest, new AcknowledgedRestListener<>(channel));
     }
 
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestPutMappingAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestPutMappingAction.java
index 60c434c6b0..06fe679264 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestPutMappingAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestPutMappingAction.java
@@ -20,7 +20,6 @@
 package org.elasticsearch.rest.action.admin.indices;
 
 import org.elasticsearch.action.admin.indices.mapping.put.PutMappingRequest;
-import org.elasticsearch.action.admin.indices.mapping.put.PutMappingResponse;
 import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.Strings;
@@ -65,11 +64,11 @@ public class RestPutMappingAction extends BaseRestHandler {
     public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
         PutMappingRequest putMappingRequest = putMappingRequest(Strings.splitStringByCommaToArray(request.param("index")));
         putMappingRequest.type(request.param("type"));
-        putMappingRequest.source(request.content().utf8ToString());
+        putMappingRequest.source(request.content(), request.getXContentType());
         putMappingRequest.updateAllTypes(request.paramAsBoolean("update_all_types", false));
         putMappingRequest.timeout(request.paramAsTime("timeout", putMappingRequest.timeout()));
         putMappingRequest.masterNodeTimeout(request.paramAsTime("master_timeout", putMappingRequest.masterNodeTimeout()));
         putMappingRequest.indicesOptions(IndicesOptions.fromRequest(request, putMappingRequest.indicesOptions()));
-        return channel -> client.admin().indices().putMapping(putMappingRequest, new AcknowledgedRestListener<PutMappingResponse>(channel));
+        return channel -> client.admin().indices().putMapping(putMappingRequest, new AcknowledgedRestListener<>(channel));
     }
 }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestUpdateSettingsAction.java b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestUpdateSettingsAction.java
index 9e38fb5e07..47037460e2 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestUpdateSettingsAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/admin/indices/RestUpdateSettingsAction.java
@@ -66,7 +66,9 @@ public class RestUpdateSettingsAction extends BaseRestHandler {
         Settings.Builder updateSettings = Settings.builder();
         String bodySettingsStr = request.content().utf8ToString();
         if (Strings.hasText(bodySettingsStr)) {
-            Settings buildSettings = Settings.builder().loadFromSource(bodySettingsStr).build();
+            Settings buildSettings = Settings.builder()
+                .loadFromSource(bodySettingsStr, request.getXContentType())
+                .build();
             for (Map.Entry<String, String> entry : buildSettings.getAsMap().entrySet()) {
                 String key = entry.getKey();
                 String value = entry.getValue();
diff --git a/core/src/main/java/org/elasticsearch/rest/action/document/RestBulkAction.java b/core/src/main/java/org/elasticsearch/rest/action/document/RestBulkAction.java
index af0faac553..8fdf2792a4 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/document/RestBulkAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/document/RestBulkAction.java
@@ -93,7 +93,7 @@ public class RestBulkAction extends BaseRestHandler {
         bulkRequest.timeout(request.paramAsTime("timeout", BulkShardRequest.DEFAULT_TIMEOUT));
         bulkRequest.setRefreshPolicy(request.param("refresh"));
         bulkRequest.add(request.content(), defaultIndex, defaultType, defaultRouting, defaultFields,
-            defaultFetchSourceContext, defaultPipeline, null, allowExplicitIndex);
+            defaultFetchSourceContext, defaultPipeline, null, allowExplicitIndex, request.getXContentType());
 
         return channel -> client.bulk(bulkRequest, new RestBuilderListener<BulkResponse>(channel) {
             @Override
diff --git a/core/src/main/java/org/elasticsearch/rest/action/document/RestGetSourceAction.java b/core/src/main/java/org/elasticsearch/rest/action/document/RestGetSourceAction.java
index 728e1ff599..83d424ed74 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/document/RestGetSourceAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/document/RestGetSourceAction.java
@@ -66,7 +66,7 @@ public class RestGetSourceAction extends BaseRestHandler {
                 client.get(getRequest, new RestResponseListener<GetResponse>(channel) {
                     @Override
                     public RestResponse buildResponse(GetResponse response) throws Exception {
-                        XContentBuilder builder = channel.newBuilder(response.getSourceInternal(), false);
+                        XContentBuilder builder = channel.newBuilder(request.getXContentType(), false);
                         if (response.isSourceEmpty()) { // check if doc source (or doc itself) is missing
                             return new BytesRestResponse(NOT_FOUND, builder);
                         } else {
diff --git a/core/src/main/java/org/elasticsearch/rest/action/document/RestIndexAction.java b/core/src/main/java/org/elasticsearch/rest/action/document/RestIndexAction.java
index 042b0d57b5..ddaf226875 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/document/RestIndexAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/document/RestIndexAction.java
@@ -65,7 +65,7 @@ public class RestIndexAction extends BaseRestHandler {
         indexRequest.routing(request.param("routing"));
         indexRequest.parent(request.param("parent")); // order is important, set it after routing, so it will set the routing
         indexRequest.setPipeline(request.param("pipeline"));
-        indexRequest.source(request.content());
+        indexRequest.source(request.content(), request.getXContentType());
         indexRequest.timeout(request.paramAsTime("timeout", IndexRequest.DEFAULT_TIMEOUT));
         indexRequest.setRefreshPolicy(request.param("refresh"));
         indexRequest.version(RestActions.parseVersion(request));
diff --git a/core/src/main/java/org/elasticsearch/rest/action/ingest/RestPutPipelineAction.java b/core/src/main/java/org/elasticsearch/rest/action/ingest/RestPutPipelineAction.java
index b7a8e6366b..2496c9b4a2 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/ingest/RestPutPipelineAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/ingest/RestPutPipelineAction.java
@@ -21,7 +21,10 @@ package org.elasticsearch.rest.action.ingest;
 
 import org.elasticsearch.action.ingest.PutPipelineRequest;
 import org.elasticsearch.client.node.NodeClient;
+import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
@@ -38,7 +41,8 @@ public class RestPutPipelineAction extends BaseRestHandler {
 
     @Override
     public RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient client) throws IOException {
-        PutPipelineRequest request = new PutPipelineRequest(restRequest.param("id"), restRequest.contentOrSourceParam());
+        Tuple<XContentType, BytesReference> sourceTuple = restRequest.contentOrSourceParam();
+        PutPipelineRequest request = new PutPipelineRequest(restRequest.param("id"), sourceTuple.v2(), sourceTuple.v1());
         request.masterNodeTimeout(restRequest.paramAsTime("master_timeout", request.masterNodeTimeout()));
         request.timeout(restRequest.paramAsTime("timeout", request.timeout()));
         return channel -> client.admin().cluster().putPipeline(request, new AcknowledgedRestListener<>(channel));
diff --git a/core/src/main/java/org/elasticsearch/rest/action/ingest/RestSimulatePipelineAction.java b/core/src/main/java/org/elasticsearch/rest/action/ingest/RestSimulatePipelineAction.java
index 5885dc9df6..9dbe1808a8 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/ingest/RestSimulatePipelineAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/ingest/RestSimulatePipelineAction.java
@@ -21,7 +21,10 @@ package org.elasticsearch.rest.action.ingest;
 
 import org.elasticsearch.action.ingest.SimulatePipelineRequest;
 import org.elasticsearch.client.node.NodeClient;
+import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestController;
 import org.elasticsearch.rest.RestRequest;
@@ -40,7 +43,8 @@ public class RestSimulatePipelineAction extends BaseRestHandler {
 
     @Override
     public RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient client) throws IOException {
-        SimulatePipelineRequest request = new SimulatePipelineRequest(restRequest.contentOrSourceParam());
+        Tuple<XContentType, BytesReference> sourceTuple = restRequest.contentOrSourceParam();
+        SimulatePipelineRequest request = new SimulatePipelineRequest(sourceTuple.v2(), sourceTuple.v1());
         request.setId(restRequest.param("id"));
         request.setVerbose(restRequest.paramAsBoolean("verbose", false));
         return channel -> client.admin().cluster().simulatePipeline(request, new RestToXContentListener<>(channel));
diff --git a/core/src/main/java/org/elasticsearch/rest/action/search/RestClearScrollAction.java b/core/src/main/java/org/elasticsearch/rest/action/search/RestClearScrollAction.java
index e7ee7c7397..47252f5a10 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/search/RestClearScrollAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/search/RestClearScrollAction.java
@@ -20,12 +20,10 @@
 package org.elasticsearch.rest.action.search;
 
 import org.elasticsearch.action.search.ClearScrollRequest;
-import org.elasticsearch.action.search.ClearScrollResponse;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestController;
@@ -50,26 +48,34 @@ public class RestClearScrollAction extends BaseRestHandler {
         String scrollIds = request.param("scroll_id");
         ClearScrollRequest clearRequest = new ClearScrollRequest();
         clearRequest.setScrollIds(Arrays.asList(splitScrollIds(scrollIds)));
-        BytesReference body = request.contentOrSourceParam();
-        if (body.length() > 0) {
-            if (XContentFactory.xContentType(body) == null) {
-                scrollIds = body.utf8ToString();
-                clearRequest.setScrollIds(Arrays.asList(splitScrollIds(scrollIds)));
+        request.withContentOrSourceParamParserOrNullLenient((xContentParser -> {
+            if (xContentParser == null) {
+                if (request.hasContent()) {
+                    // TODO: why do we accept this plain text value? maybe we can just use the scroll params?
+                    BytesReference body = request.content();
+                    String bodyScrollIds = body.utf8ToString();
+                    clearRequest.setScrollIds(Arrays.asList(splitScrollIds(bodyScrollIds)));
+                }
             } else {
                 // NOTE: if rest request with xcontent body has request parameters, these parameters does not override xcontent value
                 clearRequest.setScrollIds(null);
-                try (XContentParser parser = request.contentOrSourceParamParser()) {
-                    buildFromContent(parser, clearRequest);
+                try {
+                    buildFromContent(xContentParser, clearRequest);
                 } catch (IOException e) {
                     throw new IllegalArgumentException("Failed to parse request body", e);
                 }
             }
-        }
+        }));
 
-        return channel -> client.clearScroll(clearRequest, new RestStatusToXContentListener<ClearScrollResponse>(channel));
+        return channel -> client.clearScroll(clearRequest, new RestStatusToXContentListener<>(channel));
+    }
+
+    @Override
+    public boolean supportsPlainText() {
+        return true;
     }
 
-    public static String[] splitScrollIds(String scrollIds) {
+    private static String[] splitScrollIds(String scrollIds) {
         if (scrollIds == null) {
             return Strings.EMPTY_ARRAY;
         }
diff --git a/core/src/main/java/org/elasticsearch/rest/action/search/RestMultiSearchAction.java b/core/src/main/java/org/elasticsearch/rest/action/search/RestMultiSearchAction.java
index 567823a6e2..a803958618 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/search/RestMultiSearchAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/search/RestMultiSearchAction.java
@@ -26,10 +26,11 @@ import org.elasticsearch.action.support.IndicesOptions;
 import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.collect.Tuple;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContent;
-import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.index.query.QueryParseContext;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestController;
@@ -93,7 +94,7 @@ public class RestMultiSearchAction extends BaseRestHandler {
     }
 
     /**
-     * Parses a multi-line {@link RestRequest} body, instanciating a {@link SearchRequest} for each line and applying the given consumer.
+     * Parses a multi-line {@link RestRequest} body, instantiating a {@link SearchRequest} for each line and applying the given consumer.
      */
     public static void parseMultiLineRequest(RestRequest request, IndicesOptions indicesOptions, boolean allowExplicitIndex,
             BiConsumer<SearchRequest, XContentParser> consumer) throws IOException {
@@ -103,9 +104,10 @@ public class RestMultiSearchAction extends BaseRestHandler {
         String searchType = request.param("search_type");
         String routing = request.param("routing");
 
-        final BytesReference data = request.contentOrSourceParam();
+        final Tuple<XContentType, BytesReference> sourceTuple = request.contentOrSourceParam();
+        final XContent xContent = sourceTuple.v1().xContent();
+        final BytesReference data = sourceTuple.v2();
 
-        XContent xContent = XContentFactory.xContent(data);
         int from = 0;
         int length = data.length();
         byte marker = xContent.streamSeparator();
@@ -176,7 +178,7 @@ public class RestMultiSearchAction extends BaseRestHandler {
                 break;
             }
             BytesReference bytes = data.slice(from, nextMarker - from);
-            try (XContentParser parser = XContentFactory.xContent(bytes).createParser(request.getXContentRegistry(), bytes)) {
+            try (XContentParser parser = xContent.createParser(request.getXContentRegistry(), bytes)) {
                 consumer.accept(searchRequest, parser);
             }
             // move pointers
diff --git a/core/src/main/java/org/elasticsearch/rest/action/search/RestSearchScrollAction.java b/core/src/main/java/org/elasticsearch/rest/action/search/RestSearchScrollAction.java
index 87a6f464cc..2a60fc6317 100644
--- a/core/src/main/java/org/elasticsearch/rest/action/search/RestSearchScrollAction.java
+++ b/core/src/main/java/org/elasticsearch/rest/action/search/RestSearchScrollAction.java
@@ -24,7 +24,6 @@ import org.elasticsearch.client.node.NodeClient;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.TimeValue;
-import org.elasticsearch.common.xcontent.XContentFactory;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.rest.BaseRestHandler;
 import org.elasticsearch.rest.RestController;
@@ -58,25 +57,33 @@ public class RestSearchScrollAction extends BaseRestHandler {
             searchScrollRequest.scroll(new Scroll(parseTimeValue(scroll, null, "scroll")));
         }
 
-        BytesReference body = request.contentOrSourceParam();
-        if (body.length() > 0) {
-            if (XContentFactory.xContentType(body) == null) {
-                if (scrollId == null) {
-                    scrollId = body.utf8ToString();
-                    searchScrollRequest.scrollId(scrollId);
+        request.withContentOrSourceParamParserOrNull(xContentParser -> {
+            if (xContentParser == null) {
+                if (request.hasContent()) {
+                    // TODO: why do we accept this plain text value? maybe we can just use the scroll params?
+                    BytesReference body = request.getContentOrSourceParamOnly();
+                    if (scrollId == null) {
+                        String bodyScrollId = body.utf8ToString();
+                        searchScrollRequest.scrollId(bodyScrollId);
+                    }
                 }
             } else {
                 // NOTE: if rest request with xcontent body has request parameters, these parameters override xcontent values
-                try (XContentParser parser = request.contentOrSourceParamParser()) {
-                    buildFromContent(parser, searchScrollRequest);
+                try {
+                    buildFromContent(xContentParser, searchScrollRequest);
                 } catch (IOException e) {
                     throw new IllegalArgumentException("Failed to parse request body", e);
                 }
             }
-        }
+        });
         return channel -> client.searchScroll(searchScrollRequest, new RestStatusToXContentListener<>(channel));
     }
 
+    @Override
+    public boolean supportsPlainText() {
+        return true;
+    }
+
     public static void buildFromContent(XContentParser parser, SearchScrollRequest searchScrollRequest) throws IOException {
         if (parser.nextToken() != XContentParser.Token.START_OBJECT) {
             throw new IllegalArgumentException("Malformed content, must start with an object");