Fix SnapshotInfo.fromXContentInternal not Fully Consuming Parser (#73268)

The parsing here was causing trouble with the new streaming deserialization because it
did not fully consume the parser so if the internal buffer of the parser was just enough
to finish reading the `"snapshot"` field but missed the closing bracket, then the stream behind
the parser would not have been consumed fully. Also it was strangely lenient and would just read
a broken in-progress `SnapshotInfo` if it ran into SMILE that contained any object field
under any key that isn't "snapshot". I made it a little stricter now to enforce that
we have a "snapshot" field and not just an object field by any name.

server/src/main/java/org/elasticsearch/snapshots/SnapshotInfo.java

@@ -802,76 +802,76 @@ public final class SnapshotInfo implements Comparable<SnapshotInfo>, ToXContent,
             parser.nextToken();
         }
         XContentParser.Token token;
+        XContentParserUtils.ensureFieldName(parser, parser.currentToken(), SNAPSHOT);
         XContentParserUtils.ensureExpectedToken(XContentParser.Token.START_OBJECT, parser.nextToken(), parser);
-        String currentFieldName = parser.currentName();
-        if (SNAPSHOT.equals(currentFieldName)) {
-            while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
-                if (token == XContentParser.Token.FIELD_NAME) {
-                    currentFieldName = parser.currentName();
-                    token = parser.nextToken();
-                    if (token.isValue()) {
-                        if (NAME.equals(currentFieldName)) {
-                            name = parser.text();
-                        } else if (UUID.equals(currentFieldName)) {
-                            uuid = parser.text();
-                        } else if (STATE.equals(currentFieldName)) {
-                            state = SnapshotState.valueOf(parser.text());
-                        } else if (REASON.equals(currentFieldName)) {
-                            reason = parser.text();
-                        } else if (START_TIME.equals(currentFieldName)) {
-                            startTime = parser.longValue();
-                        } else if (END_TIME.equals(currentFieldName)) {
-                            endTime = parser.longValue();
-                        } else if (TOTAL_SHARDS.equals(currentFieldName)) {
-                            totalShards = parser.intValue();
-                        } else if (SUCCESSFUL_SHARDS.equals(currentFieldName)) {
-                            successfulShards = parser.intValue();
-                        } else if (VERSION_ID.equals(currentFieldName)) {
-                            version = Version.fromId(parser.intValue());
-                        } else if (INCLUDE_GLOBAL_STATE.equals(currentFieldName)) {
-                            includeGlobalState = parser.booleanValue();
+        while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
+            if (token == XContentParser.Token.FIELD_NAME) {
+                final String currentFieldName = parser.currentName();
+                token = parser.nextToken();
+                if (token.isValue()) {
+                    if (NAME.equals(currentFieldName)) {
+                        name = parser.text();
+                    } else if (UUID.equals(currentFieldName)) {
+                        uuid = parser.text();
+                    } else if (STATE.equals(currentFieldName)) {
+                        state = SnapshotState.valueOf(parser.text());
+                    } else if (REASON.equals(currentFieldName)) {
+                        reason = parser.text();
+                    } else if (START_TIME.equals(currentFieldName)) {
+                        startTime = parser.longValue();
+                    } else if (END_TIME.equals(currentFieldName)) {
+                        endTime = parser.longValue();
+                    } else if (TOTAL_SHARDS.equals(currentFieldName)) {
+                        totalShards = parser.intValue();
+                    } else if (SUCCESSFUL_SHARDS.equals(currentFieldName)) {
+                        successfulShards = parser.intValue();
+                    } else if (VERSION_ID.equals(currentFieldName)) {
+                        version = Version.fromId(parser.intValue());
+                    } else if (INCLUDE_GLOBAL_STATE.equals(currentFieldName)) {
+                        includeGlobalState = parser.booleanValue();
+                    }
+                } else if (token == XContentParser.Token.START_ARRAY) {
+                    if (DATA_STREAMS.equals(currentFieldName)) {
+                        dataStreams = new ArrayList<>();
+                        while (parser.nextToken() != XContentParser.Token.END_ARRAY) {
+                            dataStreams.add(parser.text());
+                        }
+                    } else if (INDICES.equals(currentFieldName)) {
+                        ArrayList<String> indicesArray = new ArrayList<>();
+                        while (parser.nextToken() != XContentParser.Token.END_ARRAY) {
+                            indicesArray.add(parser.text());
                         }
-                    } else if (token == XContentParser.Token.START_ARRAY) {
-                        if (DATA_STREAMS.equals(currentFieldName)){
-                            dataStreams = new ArrayList<>();
-                            while (parser.nextToken() != XContentParser.Token.END_ARRAY) {
-                                dataStreams.add(parser.text());
-                            }
-                        } else if (INDICES.equals(currentFieldName)) {
-                            ArrayList<String> indicesArray = new ArrayList<>();
-                            while (parser.nextToken() != XContentParser.Token.END_ARRAY) {
-                                indicesArray.add(parser.text());
-                            }
-                            indices = Collections.unmodifiableList(indicesArray);
-                        } else if (FAILURES.equals(currentFieldName)) {
-                            ArrayList<SnapshotShardFailure> shardFailureArrayList = new ArrayList<>();
-                            while (parser.nextToken() != XContentParser.Token.END_ARRAY) {
-                                shardFailureArrayList.add(SnapshotShardFailure.fromXContent(parser));
-                            }
-                            shardFailures = Collections.unmodifiableList(shardFailureArrayList);
-                        } else if (FEATURE_STATES.equals(currentFieldName)) {
-                            ArrayList<SnapshotFeatureInfo> snapshotFeatureInfoArrayList = new ArrayList<>();
-                            while (parser.nextToken() != XContentParser.Token.END_ARRAY) {
-                                snapshotFeatureInfoArrayList.add(SnapshotFeatureInfo.fromXContent(parser));
-                            }
-                            featureStates = Collections.unmodifiableList(snapshotFeatureInfoArrayList);
-                        } else {
-                            // It was probably created by newer version - ignoring
-                            parser.skipChildren();
+                        indices = Collections.unmodifiableList(indicesArray);
+                    } else if (FAILURES.equals(currentFieldName)) {
+                        ArrayList<SnapshotShardFailure> shardFailureArrayList = new ArrayList<>();
+                        while (parser.nextToken() != XContentParser.Token.END_ARRAY) {
+                            shardFailureArrayList.add(SnapshotShardFailure.fromXContent(parser));
                         }
-                    } else if (token == XContentParser.Token.START_OBJECT) {
-                        if (USER_METADATA.equals(currentFieldName)) {
-                            userMetadata = parser.map();
-                        } else if (INDEX_DETAILS.equals(currentFieldName)) {
-                            indexSnapshotDetails = parser.map(HashMap::new, p -> IndexSnapshotDetails.PARSER.parse(p, null));
-                        } else {
-                            // It was probably created by newer version - ignoring
-                            parser.skipChildren();
+                        shardFailures = Collections.unmodifiableList(shardFailureArrayList);
+                    } else if (FEATURE_STATES.equals(currentFieldName)) {
+                        ArrayList<SnapshotFeatureInfo> snapshotFeatureInfoArrayList = new ArrayList<>();
+                        while (parser.nextToken() != XContentParser.Token.END_ARRAY) {
+                            snapshotFeatureInfoArrayList.add(SnapshotFeatureInfo.fromXContent(parser));
                         }
+                        featureStates = Collections.unmodifiableList(snapshotFeatureInfoArrayList);
+                    } else {
+                        // It was probably created by newer version - ignoring
+                        parser.skipChildren();
+                    }
+                } else if (token == XContentParser.Token.START_OBJECT) {
+                    if (USER_METADATA.equals(currentFieldName)) {
+                        userMetadata = parser.map();
+                    } else if (INDEX_DETAILS.equals(currentFieldName)) {
+                        indexSnapshotDetails = parser.map(HashMap::new, p -> IndexSnapshotDetails.PARSER.parse(p, null));
+                    } else {
+                        // It was probably created by newer version - ignoring
+                        parser.skipChildren();
                     }
                 }
             }
         }
+        // closing bracket of the object containing the "snapshot" field should be there
+        XContentParserUtils.ensureExpectedToken(XContentParser.Token.END_OBJECT, parser.nextToken(), parser);
         if (uuid == null) {
             // the old format where there wasn't a UUID
             uuid = name;