[ML] Reject headers supplied directly in the put datafeed body (elastic/x-pack-elasticsearch#3853)

Since elastic/x-pack-elasticsearch#3254 security headers have been stored in datafeed cluster state
to allow the datafeed to run searches using the credentials of the user
who created/updated it.  As a result the parser was changed to read the
"headers" field so that cluster state could be reloaded.  However, this
meant that datafeed configs could be submitted with a "headers" field.
No security loophole arose from this, as subsequent code overwrites the
contents of any supplied headers.  But it could be confusing that an
erroneously supplied field did not cause a parse failure as it usually
would.

This change makes the config parser for datafeeds reject a "headers"
field.  Now only the metadata parser used for reloading cluster state
will read a "headers" field.

Original commit: elastic/x-pack-elasticsearch@afa503275f

plugin/core/src/main/java/org/elasticsearch/xpack/core/ml/datafeed/DatafeedConfig.java

@@ -119,8 +119,10 @@ public class DatafeedConfig extends AbstractDiffable<DatafeedConfig> implements
             // TODO this is to read former _source field. Remove in v7.0.0
             parser.declareBoolean((builder, value) -> {}, SOURCE);
             parser.declareObject(Builder::setChunkingConfig, ChunkingConfig.PARSERS.get(parserType), CHUNKING_CONFIG);
-            parser.declareObject(Builder::setHeaders, (p, c) -> p.mapStrings(), HEADERS);
         }
+        // Headers are only parsed by the metadata parser, so headers supplied in the _body_ of a REST request will be rejected.
+        // (For config headers are explicitly transferred from the auth headers by code in the put/update datafeed actions.)
+        METADATA_PARSER.declareObject(Builder::setHeaders, (p, c) -> p.mapStrings(), HEADERS);
     }
 
     private final String id;
@@ -328,8 +330,7 @@ public class DatafeedConfig extends AbstractDiffable<DatafeedConfig> implements
         if (chunkingConfig != null) {
             builder.field(CHUNKING_CONFIG.getPreferredName(), chunkingConfig);
         }
-        if (headers != null && headers.isEmpty() == false
-                && params.paramAsBoolean(ToXContentParams.FOR_CLUSTER_STATE, false) == true) {
+        if (headers.isEmpty() == false && params.paramAsBoolean(ToXContentParams.FOR_CLUSTER_STATE, false) == true) {
             builder.field(HEADERS.getPreferredName(), headers);
         }
         return builder;
@@ -477,7 +478,7 @@ public class DatafeedConfig extends AbstractDiffable<DatafeedConfig> implements
         }
 
         public void setHeaders(Map<String, String> headers) {
-            this.headers = headers;
+            this.headers = ExceptionsHelper.requireNonNull(headers, HEADERS.getPreferredName());
         }
 
         public void setIndices(List<String> indices) {

plugin/src/test/resources/rest-api-spec/test/ml/datafeeds_crud.yml

@@ -81,6 +81,19 @@ setup:
             "types":["type-bar"]
           }
 
+---
+"Test put datafeed with security headers in the body":
+  - do:
+      catch: /unknown field \[headers\], parser not found/
+      xpack.ml.put_datafeed:
+        datafeed_id: test-datafeed-1
+        body:  >
+          {
+            "job_id":"datafeeds-crud-1",
+            "indices":["index-foo"],
+            "headers":{ "a_security_header" : "secret" }
+          }
+
 ---
 "Test put datafeed referring to existing job_id":
   - do:

qa/smoke-test-ml-with-security/build.gradle

@@ -32,6 +32,7 @@ integTestRunner {
     'ml/datafeeds_crud/Test delete datafeed with missing id',
     'ml/datafeeds_crud/Test put datafeed referring to missing job_id',
     'ml/datafeeds_crud/Test put datafeed with invalid query',
+    'ml/datafeeds_crud/Test put datafeed with security headers in the body',
     'ml/datafeeds_crud/Test update datafeed with missing id',
     'ml/delete_job_force/Test cannot force delete a non-existent job',
     'ml/delete_model_snapshot/Test delete snapshot missing snapshotId',