Ensures watch definitions are valid json

Currently, watches may be submitted and accepted by the cluster
that do not consist of valid JSON.  This is because in certain
cases, WatchParser.java will not consume enough tokens from
XContentParser to be able to encounter an underlying JSON parse
error.  This patch fixes this behavior and adds a test.

Closes elastic#29746

x-pack/plugin/watcher/src/main/java/org/elasticsearch/xpack/watcher/watch/WatchParser.java

@@ -5,6 +5,7 @@
  */
 package org.elasticsearch.xpack.watcher.watch;
 
+import com.fasterxml.jackson.core.JsonParseException;
 import org.elasticsearch.ElasticsearchParseException;
 import org.elasticsearch.common.Nullable;
 import org.elasticsearch.common.bytes.BytesReference;
@@ -174,6 +175,19 @@ public Watch parse(String id, boolean includeStatus, WatcherXContentParser parse
                 throw new ElasticsearchParseException("could not parse watch [{}]. unexpected field [{}]", id, currentFieldName);
             }
         }
+
+        try {
+            /* Make sure we are at the end of the available input data -- certain types of JSON errors will not manifest
+                until we try to consume additional tokens.
+             */
+            if (parser.nextToken() != null) {
+                throw new ElasticsearchParseException("Received non-null token beyond end of watch definition!");
+            }
+        } catch (JsonParseException|ElasticsearchParseException e) {
+            throw new ElasticsearchParseException("could not parse watch [{}].  unexpected data beyond [line: {}, column: {}]",
+                e, id, parser.getTokenLocation().lineNumber, parser.getTokenLocation().columnNumber);
+        }
+
         if (trigger == null) {
             throw new ElasticsearchParseException("could not parse watch [{}]. missing required field [{}]", id,
                     WatchField.TRIGGER.getPreferredName());

x-pack/plugin/watcher/src/test/java/org/elasticsearch/xpack/watcher/watch/WatchTests.java

@@ -11,6 +11,7 @@
 import org.elasticsearch.action.support.WriteRequest;
 import org.elasticsearch.client.Client;
 import org.elasticsearch.common.ParseField;
+import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.logging.Loggers;
 import org.elasticsearch.common.settings.Settings;
@@ -19,6 +20,7 @@
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.common.xcontent.XContentType;
+import org.elasticsearch.common.xcontent.json.JsonXContent;
 import org.elasticsearch.index.query.MatchAllQueryBuilder;
 import org.elasticsearch.index.query.QueryBuilder;
 import org.elasticsearch.index.query.ScriptQueryBuilder;
@@ -296,6 +298,38 @@ public void testParserBadActions() throws Exception {
         }
     }
 
+    public void testParserConsumesEntireDefinition() throws Exception {
+        WatchParser wp = createWatchparser();
+        String watchDef="{\n" +
+            "  \"trigger\": {\n" +
+            "    \"schedule\": {\n" +
+            "      \"interval\": \"10s\"\n" +
+            "    }\n" +
+            "  },\n" +
+            "  \"input\": {\n" +
+            "    \"simple\": {}\n" +
+            "  }},\n" + /* NOTICE EXTRA CLOSING CURLY BRACE */
+            "  \"condition\": {\n" +
+            "    \"script\": {\n" +
+            "      \"inline\": \"return false\"\n" +
+            "    }\n" +
+            "  },\n" +
+            "  \"actions\": {\n" +
+            "    \"logging\": {\n" +
+            "      \"logging\": {\n" +
+            "        \"text\": \"{{ctx.payload}}\"\n" +
+            "      }\n" +
+            "    }\n" +
+            "  }\n" +
+            "}";
+        try {
+            wp.parseWithSecrets("failure", false, new BytesArray(watchDef), new DateTime(), XContentType.JSON, true);
+            fail("This watch should fail to parse as there is an extra closing curly brace in the middle of the watch");
+        } catch (ElasticsearchParseException pe) {
+            assertThat(pe.getMessage().contains("unexpected data beyond"), is(true));
+        }
+    }
+
     public void testParserDefaults() throws Exception {
         Schedule schedule = randomSchedule();
         ScheduleRegistry scheduleRegistry = registry(schedule);