SERVER-65797 Remove invalid index specs in memory before parsing for `listIndexes`

(cherry picked from commit 142cdf278feee6ce0c4bb1b121fd13d69c61de34)
(cherry picked from commit 788627e58f5d488a833b73739d0dc8d9ff96d5e7)

6 files changed, 173 insertions, 44 deletions

diff --git a/jstests/multiVersion/targetedTestsLastLtsFeatures/invalid_index_options.js b/jstests/multiVersion/targetedTestsLastLtsFeatures/invalid_index_options.js
new file mode 100644
index 00000000000..2b54a868745
--- /dev/null
+++ b/jstests/multiVersion/targetedTestsLastLtsFeatures/invalid_index_options.js
@@ -0,0 +1,51 @@
+/**
+ * Tests that in 5.0 version listIndexes can parse invalid index specs created before 5.0 version.
+ *
+ * @tags: [requires_replication]
+ */
+(function() {
+"use strict";
+
+load('jstests/multiVersion/libs/multi_rs.js');
+
+var nodes = {
+    n1: {binVersion: "last-lts"},
+    n2: {binVersion: "last-lts"},
+};
+
+var rst = new ReplSetTest({nodes: nodes});
+rst.startSet();
+rst.initiate();
+
+const dbName = "test";
+const collName = jsTestName();
+
+let primaryDB = rst.getPrimary().getDB(dbName);
+let primaryColl = primaryDB.getCollection(collName);
+
+// In earlier versions, users were able to add invalid index options when creating an index. The
+// option could still be interpreted accordingly.
+assert.commandWorked(primaryColl.createIndex({x: 1}, {sparse: "yes"}));
+
+// Upgrades from 4.4 to 5.0.
+jsTestLog("Upgrading to version 5.0");
+rst.upgradeSet({binVersion: "latest"});
+const primary = rst.getPrimary();
+assert.commandWorked(primary.adminCommand({setFeatureCompatibilityVersion: latestFCV}));
+
+primaryDB = primary.getDB(dbName);
+
+// Verify listIndexes command can correctly output the repaired index specs.
+assert.commandWorked(primaryDB.runCommand({listIndexes: collName}));
+
+// Add a new node to make sure the initial sync works correctly with the invalid index specs.
+jsTestLog("Bringing up a new node");
+rst.add();
+rst.reInitiate();
+
+jsTestLog("Waiting for new node to be synced.");
+rst.awaitReplication();
+rst.awaitSecondaryNodes();
+
+rst.stopSet();
+})();
\ No newline at end of file
diff --git a/src/mongo/db/catalog/collection_impl.cpp b/src/mongo/db/catalog/collection_impl.cpp
index 4e15e370a20..bcfe8cc156f 100644
--- a/src/mongo/db/catalog/collection_impl.cpp
+++ b/src/mongo/db/catalog/collection_impl.cpp
@@ -1858,7 +1858,7 @@ std::vector<std::string> CollectionImpl::removeInvalidIndexOptions(OperationCont
             }
 
             indexesWithInvalidOptions.push_back(std::string(index.name()));
-            index.spec = index_key_validate::removeUnknownFields(oldSpec);
+            index.spec = index_key_validate::removeUnknownFields(NamespaceString(md.ns), oldSpec);
         }
     });
 
diff --git a/src/mongo/db/catalog/index_key_validate.cpp b/src/mongo/db/catalog/index_key_validate.cpp
index 87ebafdd23a..a22263b7674 100644
--- a/src/mongo/db/catalog/index_key_validate.cpp
+++ b/src/mongo/db/catalog/index_key_validate.cpp
@@ -68,35 +68,6 @@ namespace {
 // specification.
 MONGO_FAIL_POINT_DEFINE(skipIndexCreateFieldNameValidation);
 
-static std::set<StringData> allowedFieldNames = {
-    IndexDescriptor::k2dIndexBitsFieldName,
-    IndexDescriptor::k2dIndexMaxFieldName,
-    IndexDescriptor::k2dIndexMinFieldName,
-    IndexDescriptor::k2dsphereCoarsestIndexedLevel,
-    IndexDescriptor::k2dsphereFinestIndexedLevel,
-    IndexDescriptor::k2dsphereVersionFieldName,
-    IndexDescriptor::kBackgroundFieldName,
-    IndexDescriptor::kCollationFieldName,
-    IndexDescriptor::kDefaultLanguageFieldName,
-    IndexDescriptor::kDropDuplicatesFieldName,
-    IndexDescriptor::kExpireAfterSecondsFieldName,
-    IndexDescriptor::kGeoHaystackBucketSize,
-    IndexDescriptor::kHiddenFieldName,
-    IndexDescriptor::kIndexNameFieldName,
-    IndexDescriptor::kIndexVersionFieldName,
-    IndexDescriptor::kKeyPatternFieldName,
-    IndexDescriptor::kLanguageOverrideFieldName,
-    IndexDescriptor::kNamespaceFieldName,
-    IndexDescriptor::kPartialFilterExprFieldName,
-    IndexDescriptor::kPathProjectionFieldName,
-    IndexDescriptor::kSparseFieldName,
-    IndexDescriptor::kStorageEngineFieldName,
-    IndexDescriptor::kTextVersionFieldName,
-    IndexDescriptor::kUniqueFieldName,
-    IndexDescriptor::kWeightsFieldName,
-    // Index creation under legacy writeMode can result in an index spec with an _id field.
-    "_id"};
-
 static const std::set<StringData> allowedIdIndexFieldNames = {
     IndexDescriptor::kCollationFieldName,
     IndexDescriptor::kIndexNameFieldName,
@@ -120,6 +91,27 @@ Status isIndexVersionAllowedForCreation(IndexVersion indexVersion, const BSONObj
             str::stream() << "Invalid index specification " << indexSpec
                           << "; cannot create an index with v=" << static_cast<int>(indexVersion)};
 }
+
+BSONObj buildRepairedIndexSpec(
+    const NamespaceString& ns,
+    const BSONObj& indexSpec,
+    const std::set<StringData>& allowedFieldNames,
+    std::function<void(const BSONElement&, BSONObjBuilder*)> indexSpecHandleFn) {
+    BSONObjBuilder builder;
+    for (const auto& indexSpecElem : indexSpec) {
+        StringData fieldName = indexSpecElem.fieldNameStringData();
+        if (allowedFieldNames.count(fieldName)) {
+            indexSpecHandleFn(indexSpecElem, &builder);
+        } else {
+            LOGV2_WARNING(23878,
+                          "Removing unknown field from index spec",
+                          "namespace"_attr = redact(ns.toString()),
+                          "fieldName"_attr = redact(fieldName),
+                          "indexSpec"_attr = redact(indexSpec));
+        }
+    }
+    return builder.obj();
+}
 }  // namespace
 
 Status validateKeyPattern(const BSONObj& key, IndexDescriptor::IndexVersion indexVersion) {
@@ -250,21 +242,35 @@ Status validateKeyPattern(const BSONObj& key, IndexDescriptor::IndexVersion inde
     return Status::OK();
 }
 
-BSONObj removeUnknownFields(const BSONObj& indexSpec) {
-    BSONObjBuilder builder;
-    for (const auto& indexSpecElem : indexSpec) {
+BSONObj removeUnknownFields(const NamespaceString& ns, const BSONObj& indexSpec) {
+    auto appendIndexSpecFn = [](const BSONElement& indexSpecElem, BSONObjBuilder* builder) {
+        builder->append(indexSpecElem);
+    };
+    return buildRepairedIndexSpec(ns, indexSpec, allowedFieldNames, appendIndexSpecFn);
+}
+
+BSONObj repairIndexSpec(const NamespaceString& ns,
+                        const BSONObj& indexSpec,
+                        const std::set<StringData>& allowedFieldNames) {
+    auto fixBoolIndexSpecFn = [&indexSpec, &ns](const BSONElement& indexSpecElem,
+                                                BSONObjBuilder* builder) {
         StringData fieldName = indexSpecElem.fieldNameStringData();
-        if (allowedFieldNames.count(fieldName)) {
-            builder.append(indexSpecElem);
-        } else {
-            LOGV2_WARNING(23878,
-                          "Removing field '{fieldName}' from index spec: {indexSpec}",
-                          "Removing unknown field from index spec",
+        if ((IndexDescriptor::kBackgroundFieldName == fieldName ||
+             IndexDescriptor::kUniqueFieldName == fieldName ||
+             IndexDescriptor::kSparseFieldName == fieldName ||
+             IndexDescriptor::kDropDuplicatesFieldName == fieldName) &&
+            !indexSpecElem.isNumber() && !indexSpecElem.isBoolean() && indexSpecElem.trueValue()) {
+            LOGV2_WARNING(6444400,
+                          "Fixing boolean field from index spec",
+                          "namespace"_attr = redact(ns.toString()),
                           "fieldName"_attr = redact(fieldName),
                           "indexSpec"_attr = redact(indexSpec));
+            builder->appendBool(fieldName, true);
+        } else {
+            builder->append(indexSpecElem);
         }
-    }
-    return builder.obj();
+    };
+    return buildRepairedIndexSpec(ns, indexSpec, allowedFieldNames, fixBoolIndexSpecFn);
 }
 
 StatusWith<BSONObj> validateIndexSpec(OperationContext* opCtx, const BSONObj& indexSpec) {
diff --git a/src/mongo/db/catalog/index_key_validate.h b/src/mongo/db/catalog/index_key_validate.h
index b170550b0e7..332572d41d0 100644
--- a/src/mongo/db/catalog/index_key_validate.h
+++ b/src/mongo/db/catalog/index_key_validate.h
@@ -43,6 +43,35 @@ class StatusWith;
 
 namespace index_key_validate {
 
+static std::set<StringData> allowedFieldNames = {
+    IndexDescriptor::k2dIndexBitsFieldName,
+    IndexDescriptor::k2dIndexMaxFieldName,
+    IndexDescriptor::k2dIndexMinFieldName,
+    IndexDescriptor::k2dsphereCoarsestIndexedLevel,
+    IndexDescriptor::k2dsphereFinestIndexedLevel,
+    IndexDescriptor::k2dsphereVersionFieldName,
+    IndexDescriptor::kBackgroundFieldName,
+    IndexDescriptor::kCollationFieldName,
+    IndexDescriptor::kDefaultLanguageFieldName,
+    IndexDescriptor::kDropDuplicatesFieldName,
+    IndexDescriptor::kExpireAfterSecondsFieldName,
+    IndexDescriptor::kGeoHaystackBucketSize,
+    IndexDescriptor::kHiddenFieldName,
+    IndexDescriptor::kIndexNameFieldName,
+    IndexDescriptor::kIndexVersionFieldName,
+    IndexDescriptor::kKeyPatternFieldName,
+    IndexDescriptor::kLanguageOverrideFieldName,
+    IndexDescriptor::kNamespaceFieldName,
+    IndexDescriptor::kPartialFilterExprFieldName,
+    IndexDescriptor::kPathProjectionFieldName,
+    IndexDescriptor::kSparseFieldName,
+    IndexDescriptor::kStorageEngineFieldName,
+    IndexDescriptor::kTextVersionFieldName,
+    IndexDescriptor::kUniqueFieldName,
+    IndexDescriptor::kWeightsFieldName,
+    // Index creation under legacy writeMode can result in an index spec with an _id field.
+    "_id"};
+
 /**
  * Checks if the key is valid for building an index according to the validation rules for the given
  * index version.
@@ -59,7 +88,15 @@ StatusWith<BSONObj> validateIndexSpec(OperationContext* opCtx, const BSONObj& in
 /**
  * Returns a new index spec with any unknown field names removed from 'indexSpec'.
  */
-BSONObj removeUnknownFields(const BSONObj& indexSpec);
+BSONObj removeUnknownFields(const NamespaceString& ns, const BSONObj& indexSpec);
+
+/**
+ * Returns a new index spec with boolean values in correct types and unkown field names removed.
+ */
+BSONObj repairIndexSpec(
+    const NamespaceString& ns,
+    const BSONObj& indexSpec,
+    const std::set<StringData>& allowedFieldNames = index_key_validate::allowedFieldNames);
 
 /**
  * Performs additional validation for _id index specifications. This should be called after
diff --git a/src/mongo/db/commands/create_indexes.cpp b/src/mongo/db/commands/create_indexes.cpp
index 593d525a333..eb7c86f7e13 100644
--- a/src/mongo/db/commands/create_indexes.cpp
+++ b/src/mongo/db/commands/create_indexes.cpp
@@ -109,7 +109,7 @@ std::vector<BSONObj> parseAndValidateIndexSpecs(OperationContext* opCtx,
     for (const auto& index : cmd.getIndexes()) {
         BSONObj parsedIndexSpec = index;
         if (ignoreUnknownIndexOptions) {
-            parsedIndexSpec = index_key_validate::removeUnknownFields(parsedIndexSpec);
+            parsedIndexSpec = index_key_validate::removeUnknownFields(ns, parsedIndexSpec);
         }
 
         auto indexSpecStatus = index_key_validate::validateIndexSpec(opCtx, parsedIndexSpec);
diff --git a/src/mongo/db/commands/list_indexes.cpp b/src/mongo/db/commands/list_indexes.cpp
index 2ca6f292565..33a5a2607dc 100644
--- a/src/mongo/db/commands/list_indexes.cpp
+++ b/src/mongo/db/commands/list_indexes.cpp
@@ -36,6 +36,7 @@
 
 #include "mongo/bson/bsonobjbuilder.h"
 #include "mongo/db/auth/authorization_session.h"
+#include "mongo/db/catalog/index_key_validate.h"
 #include "mongo/db/catalog/list_indexes.h"
 #include "mongo/db/clientcursor.h"
 #include "mongo/db/commands.h"
@@ -63,6 +64,36 @@
 namespace mongo {
 namespace {
 
+// The allowed fields have to be in sync with those defined in 'src/mongo/db/list_indexes.idl'.
+static std::set<StringData> allowedFieldNames = {
+    ListIndexesReplyItem::k2dsphereIndexVersionFieldName,
+    ListIndexesReplyItem::kBackgroundFieldName,
+    ListIndexesReplyItem::kBitsFieldName,
+    ListIndexesReplyItem::kBucketSizeFieldName,
+    ListIndexesReplyItem::kBuildUUIDFieldName,
+    ListIndexesReplyItem::kCoarsestIndexedLevelFieldName,
+    ListIndexesReplyItem::kCollationFieldName,
+    ListIndexesReplyItem::kDefault_languageFieldName,
+    ListIndexesReplyItem::kDropDupsFieldName,
+    ListIndexesReplyItem::kExpireAfterSecondsFieldName,
+    ListIndexesReplyItem::kFinestIndexedLevelFieldName,
+    ListIndexesReplyItem::kHiddenFieldName,
+    ListIndexesReplyItem::kKeyFieldName,
+    ListIndexesReplyItem::kLanguage_overrideFieldName,
+    ListIndexesReplyItem::kMaxFieldName,
+    ListIndexesReplyItem::kMinFieldName,
+    ListIndexesReplyItem::kNameFieldName,
+    ListIndexesReplyItem::kNsFieldName,
+    ListIndexesReplyItem::kPartialFilterExpressionFieldName,
+    ListIndexesReplyItem::kSparseFieldName,
+    ListIndexesReplyItem::kSpecFieldName,
+    ListIndexesReplyItem::kStorageEngineFieldName,
+    ListIndexesReplyItem::kTextIndexVersionFieldName,
+    ListIndexesReplyItem::kUniqueFieldName,
+    ListIndexesReplyItem::kVFieldName,
+    ListIndexesReplyItem::kWeightsFieldName,
+    ListIndexesReplyItem::kWildcardProjectionFieldName};
+
 /**
  * Returns index specs, with resolved namespace, from the catalog for this listIndexes request.
  */
@@ -248,6 +279,7 @@ public:
                     break;
                 }
                 invariant(state == PlanExecutor::ADVANCED);
+                nextDoc = index_key_validate::repairIndexSpec(nss, nextDoc, allowedFieldNames);
 
                 // If we can't fit this result inside the current batch, then we stash it for
                 // later.
@@ -265,7 +297,10 @@ public:
                                 "entry"_attr = nextDoc,
                                 "error"_attr = exc);
                     uasserted(5254501,
-                              "Could not parse catalog entry while replying to listIndexes");
+                              fmt::format("Could not parse catalog entry while replying to "
+                                          "listIndexes. Entry: '{}'. Error: '{}'.",
+                                          nextDoc.toString(),
+                                          exc.toString()));
                 }
                 bytesBuffered += nextDoc.objsize();
             }