Deprecate lenient parsing of bulk actions (#78876)

Make sure there are no arbitrary fields after an action declaration and that it gets properly closed by a curly bracket.

Resolves #43774

docs/reference/migration/migrate_8_1.asciidoc

@@ -58,12 +58,26 @@ enable <<deprecation-logging, deprecation logging>>.
 [%collapsible]
 ====
 *Details* +
-Legacy values for the `discovery.type` setting are deprecated and will be
-forbidden in a future version.
+Legacy values for the `discovery.type` setting are deprecated and will be forbidden in a future version.
 
 *Impact* +
 Do not set `discovery.type` to any value except `single-node` or `multi-node`.
 All other values are equivalent to the default discovery type which is
-`multi-node`. Where possible, omit this setting so that {es} uses the default
-discovery type.
+`multi-node`.
+Where possible, omit this setting so that {es} uses the default discovery type.
+====
+
+[discrete]
+[[breaking_8.1_lenient_bulk_action_deprecation]]
+==== Lenient parsing of bulk actions
+
+[[deprecate-lenient-parsing-of-bulk-actions]]
+.Lenient parsing of bulk actions is deprecated.
+[%collapsible]
+====
+*Details* +
+{es} 8.1 parses bulk actions more strictly in order to make sure there are no arbitrary fields after a bulk action declaration and that it gets properly closed with the closing brace.
+
+*Impact* +
+Make sure your bulk actions are valid JSON objects and that you don't have any arbitrary JSON fields after bulk actions.
 ====

server/src/main/java/org/elasticsearch/action/bulk/BulkRequestParser.java

@@ -8,6 +8,8 @@
 
 package org.elasticsearch.action.bulk;
 
+import com.fasterxml.jackson.core.io.JsonEOFException;
+
 import org.elasticsearch.action.DocWriteRequest;
 import org.elasticsearch.action.delete.DeleteRequest;
 import org.elasticsearch.action.index.IndexRequest;
@@ -31,6 +33,7 @@ import org.elasticsearch.xcontent.XContentType;
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Set;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -42,6 +45,8 @@ import static org.elasticsearch.index.seqno.SequenceNumbers.UNASSIGNED_PRIMARY_T
  */
 public final class BulkRequestParser {
     private static final DeprecationLogger deprecationLogger = DeprecationLogger.getLogger(BulkRequestParser.class);
+    private static final Set<String> SUPPORTED_ACTIONS = Set.of("create", "index", "update", "delete");
+    private static final String STRICT_ACTION_PARSING_WARNING_KEY = "bulk_request_strict_action_parsing";
 
     private static final ParseField INDEX = new ParseField("_index");
     private static final ParseField TYPE = new ParseField("_type");
@@ -179,6 +184,14 @@ public final class BulkRequestParser {
                     );
                 }
                 String action = parser.currentName();
+                if (SUPPORTED_ACTIONS.contains(action) == false) {
+                    deprecationLogger.compatibleCritical(
+                        STRICT_ACTION_PARSING_WARNING_KEY,
+                        "Unsupported action: [{}]. Supported values are [create], [delete], [index], and [update]. "
+                            + "Unsupported actions are currently accepted but will be rejected in a future version.",
+                        action
+                    );
+                }
 
                 String index = defaultIndex;
                 String type = null;
@@ -292,6 +305,7 @@ public final class BulkRequestParser {
                             + "]"
                     );
                 }
+                checkBulkActionIsProperlyClosed(parser);
 
                 if ("delete".equals(action)) {
                     if (dynamicTemplates.isEmpty() == false) {
@@ -406,6 +420,35 @@ public final class BulkRequestParser {
         }
     }
 
+    private void checkBulkActionIsProperlyClosed(XContentParser parser) throws IOException {
+        XContentParser.Token token;
+        try {
+            token = parser.nextToken();
+        } catch (JsonEOFException ignore) {
+            deprecationLogger.compatibleCritical(
+                STRICT_ACTION_PARSING_WARNING_KEY,
+                "A bulk action wasn't closed properly with the closing brace. Malformed objects are currently accepted but will be "
+                    + "rejected in a future version."
+            );
+            return;
+        }
+        if (token != XContentParser.Token.END_OBJECT) {
+            deprecationLogger.compatibleCritical(
+                STRICT_ACTION_PARSING_WARNING_KEY,
+                "A bulk action object contained multiple keys. Additional keys are currently ignored but will be rejected in a "
+                    + "future version."
+            );
+            return;
+        }
+        if (parser.nextToken() != null) {
+            deprecationLogger.compatibleCritical(
+                STRICT_ACTION_PARSING_WARNING_KEY,
+                "A bulk action contained trailing data after the closing brace. This is currently ignored but will be rejected in a "
+                    + "future version."
+            );
+        }
+    }
+
     private XContentParser createParser(XContent xContent, BytesReference data) throws IOException {
         if (data.hasArray()) {
             return parseBytesArray(xContent, data, 0, data.length());

server/src/test/java/org/elasticsearch/action/bulk/BulkRequestTests.java

@@ -370,7 +370,7 @@ public class BulkRequestTests extends ESTestCase {
                 + "{ \"field1\" : \"value3\" }\n"
                 +
 
-                "{ \"index\" : {\"dynamic_templates\":{}}\n"
+                "{ \"index\" : {\"dynamic_templates\":{}}}\n"
                 + "{ \"field1\" : \"value3\" }\n"
         );
         BulkRequest bulkRequest = new BulkRequest().add(data, null, XContentType.JSON);
@@ -411,4 +411,48 @@ public class BulkRequestTests extends ESTestCase {
             )
         );
     }
+
+    public void testBulkActionWithoutCurlyBrace() throws Exception {
+        String bulkAction = "{ \"index\":{\"_index\":\"test\",\"_id\":\"1\"} \n" + "{ \"field1\" : \"value1\" }\n";
+        BulkRequest bulkRequest = new BulkRequest();
+        bulkRequest.add(bulkAction.getBytes(StandardCharsets.UTF_8), 0, bulkAction.length(), null, XContentType.JSON);
+
+        assertWarnings(
+            "A bulk action wasn't closed properly with the closing brace. Malformed objects are currently accepted"
+                + " but will be rejected in a future version."
+        );
+    }
+
+    public void testBulkActionWithAdditionalKeys() throws Exception {
+        String bulkAction = "{ \"index\":{\"_index\":\"test\",\"_id\":\"1\"}, \"a\":\"b\"} \n" + "{ \"field1\" : \"value1\" }\n";
+        BulkRequest bulkRequest = new BulkRequest();
+        bulkRequest.add(bulkAction.getBytes(StandardCharsets.UTF_8), 0, bulkAction.length(), null, XContentType.JSON);
+
+        assertWarnings(
+            "A bulk action object contained multiple keys. Additional keys are currently ignored but will be "
+                + "rejected in a future version."
+        );
+    }
+
+    public void testBulkActionWithTrailingData() throws Exception {
+        String bulkAction = "{ \"index\":{\"_index\":\"test\",\"_id\":\"1\"} } {\"a\":\"b\"} \n" + "{ \"field1\" : \"value1\" }\n";
+        BulkRequest bulkRequest = new BulkRequest();
+        bulkRequest.add(bulkAction.getBytes(StandardCharsets.UTF_8), 0, bulkAction.length(), null, XContentType.JSON);
+
+        assertWarnings(
+            "A bulk action contained trailing data after the closing brace. This is currently ignored "
+                + "but will be rejected in a future version."
+        );
+    }
+
+    public void testUnsupportedAction() throws Exception {
+        String bulkAction = "{ \"get\":{\"_index\":\"test\",\"_id\":\"1\"} }\n";
+        BulkRequest bulkRequest = new BulkRequest();
+        bulkRequest.add(bulkAction.getBytes(StandardCharsets.UTF_8), 0, bulkAction.length(), null, XContentType.JSON);
+
+        assertWarnings(
+            "Unsupported action: [get]. Supported values are [create], [delete], [index], and [update]. "
+                + "Unsupported actions are currently accepted but will be rejected in a future version."
+        );
+    }
 }

x-pack/plugin/sql/qa/server/multi-node/src/test/java/org/elasticsearch/xpack/sql/qa/multi_node/RestSqlMultinodeIT.java

@@ -93,7 +93,7 @@ public class RestSqlMultinodeIT extends ESRestTestCase {
             int a = 3 * i;
             int b = a + 1;
             int c = b + 1;
-            bulk.append("{\"index\":{\"_id\":\"" + i + "\"}\n");
+            bulk.append("{\"index\":{\"_id\":\"" + i + "\"}}\n");
             bulk.append("{\"a\": " + a + ", \"b\": " + b + ", \"c\": " + c + "}\n");
         }
         request.setJsonEntity(bulk.toString());

x-pack/plugin/sql/qa/server/security/src/test/java/org/elasticsearch/xpack/sql/qa/security/SqlSecurityTestCase.java

@@ -176,11 +176,11 @@ public abstract class SqlSecurityTestCase extends ESRestTestCase {
         request.addParameter("refresh", "true");
 
         StringBuilder bulk = new StringBuilder();
-        bulk.append("{\"index\":{\"_index\": \"test\", \"_id\":\"1\"}\n");
+        bulk.append("{\"index\":{\"_index\": \"test\", \"_id\":\"1\"}}\n");
         bulk.append("{\"a\": 1, \"b\": 2, \"c\": 3}\n");
-        bulk.append("{\"index\":{\"_index\": \"test\", \"_id\":\"2\"}\n");
+        bulk.append("{\"index\":{\"_index\": \"test\", \"_id\":\"2\"}}\n");
         bulk.append("{\"a\": 4, \"b\": 5, \"c\": 6}\n");
-        bulk.append("{\"index\":{\"_index\": \"bort\", \"_id\":\"1\"}\n");
+        bulk.append("{\"index\":{\"_index\": \"bort\", \"_id\":\"1\"}}\n");
         bulk.append("{\"a\": \"test\"}\n");
         request.setJsonEntity(bulk.toString());
         client().performRequest(request);

x-pack/plugin/sql/qa/server/security/src/test/java/org/elasticsearch/xpack/sql/qa/security/UserFunctionIT.java

@@ -181,7 +181,7 @@ public class UserFunctionIT extends ESRestTestCase {
         request.addParameter("refresh", "true");
         StringBuilder bulk = new StringBuilder();
         for (String doc : docs) {
-            bulk.append("{\"index\":{}\n");
+            bulk.append("{\"index\":{}}\n");
             bulk.append(doc + "\n");
         }
         request.setJsonEntity(bulk.toString());

x-pack/plugin/sql/qa/server/src/main/java/org/elasticsearch/xpack/sql/qa/jdbc/DataLoader.java

@@ -79,7 +79,7 @@ public class DataLoader {
         createEmptyIndex(client, index);
         Request request = new Request("POST", "/" + index + "/_bulk");
         request.addParameter("refresh", "true");
-        request.setJsonEntity("{\"index\":{}\n{}\n" + "{\"index\":{}\n{}\n");
+        request.setJsonEntity("{\"index\":{}}\n{}\n" + "{\"index\":{}}\n{}\n");
         client.performRequest(request);
     }
 

x-pack/plugin/sql/qa/server/src/main/java/org/elasticsearch/xpack/sql/qa/rest/BaseRestSqlTestCase.java

@@ -179,7 +179,7 @@ public abstract class BaseRestSqlTestCase extends RemoteClusterAwareSqlRestTestC
         request.addParameter("refresh", "true");
         StringBuilder bulk = new StringBuilder();
         for (String doc : docs) {
-            bulk.append("{\"index\":{}\n");
+            bulk.append("{\"index\":{}}\n");
             bulk.append(doc + "\n");
         }
         request.setJsonEntity(bulk.toString());