SERVER-41164 stop documents from leaking into unrelated changestreams when regex control characters are in db name

2 files changed, 74 insertions, 2 deletions

diff --git a/jstests/change_streams/no_regex_leak.js b/jstests/change_streams/no_regex_leak.js
new file mode 100644
index 00000000000..a0e1b3aae7d
--- /dev/null
+++ b/jstests/change_streams/no_regex_leak.js
@@ -0,0 +1,60 @@
+/*
+ * This test makes sure that regex control characters in the namespace of changestream targets don't
+ * affect what documents appear in a changestream, in response to SERVER-41164.
+ */
+(function() {
+    "use strict";
+
+    load("jstests/libs/change_stream_util.js");
+    load("jstests/libs/collection_drop_recreate.js");
+    function test_no_leak(
+        dbNameUnrelated, collNameUnrelated, dbNameProblematic, collNameProblematic) {
+        const dbUnrelated = db.getSiblingDB(dbNameUnrelated);
+        const cstUnrelated = new ChangeStreamTest(dbUnrelated);
+        assertDropAndRecreateCollection(dbUnrelated, collNameUnrelated);
+
+        const watchUnrelated = cstUnrelated.startWatchingChanges(
+            {pipeline: [{$changeStream: {}}], collection: collNameUnrelated});
+
+        const dbProblematic = db.getSiblingDB(dbNameProblematic);
+        const cstProblematic = new ChangeStreamTest(dbProblematic);
+        assertDropAndRecreateCollection(dbProblematic, collNameProblematic);
+
+        const watchProblematic = cstProblematic.startWatchingChanges(
+            {pipeline: [{$changeStream: {}}], collection: collNameProblematic});
+
+        assert.commandWorked(dbUnrelated.getCollection(collNameUnrelated).insert({_id: 2}));
+        let expected = {
+            documentKey: {_id: 2},
+            fullDocument: {_id: 2},
+            ns: {db: dbNameUnrelated, coll: collNameUnrelated},
+            operationType: "insert",
+        };
+        // Make sure that only the database which was inserted into reflects a change on its
+        // changestream.
+        cstUnrelated.assertNextChangesEqual({cursor: watchUnrelated, expectedChanges: [expected]});
+        // The other DB shouldn't have any changes.
+        cstProblematic.assertNoChange(watchProblematic);
+
+        assert.commandWorked(dbProblematic.getCollection(collNameProblematic).insert({_id: 3}));
+        expected = {
+            documentKey: {_id: 3},
+            fullDocument: {_id: 3},
+            ns: {db: dbNameProblematic, coll: collNameProblematic},
+            operationType: "insert",
+        };
+        cstProblematic.assertNextChangesEqual(
+            {cursor: watchProblematic, expectedChanges: [expected]});
+        cstUnrelated.assertNoChange(watchUnrelated);
+
+        cstUnrelated.cleanUp();
+        cstProblematic.cleanUp();
+    }
+
+    test_no_leak("has_no_pipe", "coll", "has_a_|pipe", "coll");
+    test_no_leak("has_[two]_brackets", "coll", "has_t_brackets", "coll");
+    test_no_leak("test", "dotted.collection", "testadotted", "collection");
+    test_no_leak("carat", "coll", "hasa^carat", "coll");
+    test_no_leak("starssss", "coll", "stars*", "coll");
+    test_no_leak("db1", "coll", "db1", "col*");
+}());
diff --git a/src/mongo/db/pipeline/document_source_change_stream.cpp b/src/mongo/db/pipeline/document_source_change_stream.cpp
index 20c146a73ca..b9225fbe8cf 100644
--- a/src/mongo/db/pipeline/document_source_change_stream.cpp
+++ b/src/mongo/db/pipeline/document_source_change_stream.cpp
@@ -214,15 +214,27 @@ DocumentSourceChangeStream::ChangeStreamType DocumentSourceChangeStream::getChan
 }
 
 std::string DocumentSourceChangeStream::getNsRegexForChangeStream(const NamespaceString& nss) {
+    auto regexEscape = [](const std::string& source) {
+        std::string result = "";
+        std::string escapes = "*+|()^?[]./\\$";
+        for (const char& c : source) {
+            if (escapes.find(c) != std::string::npos) {
+                result.append("\\");
+            }
+            result += c;
+        }
+        return result;
+    };
+
     auto type = getChangeStreamType(nss);
     switch (type) {
         case ChangeStreamType::kSingleCollection:
             // Match the target namespace exactly.
-            return "^" + nss.ns() + "$";
+            return "^" + regexEscape(nss.ns()) + "$";
         case ChangeStreamType::kSingleDatabase:
             // Match all namespaces that start with db name, followed by ".", then NOT followed by
             // '$' or 'system.'
-            return "^" + nss.db() + "\\." + kRegexAllCollections;
+            return "^" + regexEscape(nss.db().toString()) + "\\." + kRegexAllCollections;
         case ChangeStreamType::kAllChangesForCluster:
             // Match all namespaces that start with any db name other than admin, config, or local,
             // followed by ".", then NOT followed by '$' or 'system.'.