SERVER-34662 fix $changeStream parsing to check that its argument is an object

Author: Ian Boros

jstests/change_streams/change_stream.js

@@ -16,6 +16,17 @@
     assertDropAndRecreateCollection(db, "t1");
     assertDropAndRecreateCollection(db, "t2");
 
+    // Test that $changeStream only accepts an object as its argument.
+    function checkArgFails(arg) {
+        assert.commandFailedWithCode(
+            db.runCommand({aggregate: "t1", pipeline: [{$changeStream: arg}], cursor: {}}), 50808);
+    }
+
+    checkArgFails(1);
+    checkArgFails("invalid");
+    checkArgFails(false);
+    checkArgFails([1, 2, "invalid", {x: 1}]);
+
     // Test that a change stream cannot be opened on collections in the "admin", "config", or
     // "local" databases.
     assertInvalidChangeStreamNss("admin", "testColl");

jstests/libs/override_methods/implicit_whole_db_changestreams.js

@@ -25,7 +25,8 @@ const ChangeStreamPassthroughHelpers = {
         // Determine whether this command is a valid $changeStream aggregation on a single
         // collection or database.
         if (!(cmdObj && cmdObj.aggregate && Array.isArray(cmdObj.pipeline) &&
-              cmdObj.pipeline.length > 0 && cmdObj.pipeline[0].$changeStream)) {
+              cmdObj.pipeline.length > 0 && typeof cmdObj.pipeline[0].$changeStream == "object" &&
+              cmdObj.pipeline[0].$changeStream.constructor === Object)) {
             return false;
         }
         // Single-collection and whole-db streams cannot be opened on internal databases.

src/mongo/db/pipeline/document_source_change_stream.cpp

@@ -462,6 +462,10 @@ void parseResumeOptions(const intrusive_ptr<ExpressionContext>& expCtx,
 
 list<intrusive_ptr<DocumentSource>> DocumentSourceChangeStream::createFromBson(
     BSONElement elem, const intrusive_ptr<ExpressionContext>& expCtx) {
+    uassert(50808,
+            "$changeStream stage expects a document as argument.",
+            elem.type() == BSONType::Object);
+
     // A change stream is a tailable + awaitData cursor.
     expCtx->tailableMode = TailableModeEnum::kTailableAndAwaitData;
 

src/mongo/db/pipeline/document_source_change_stream_test.cpp

@@ -268,6 +268,20 @@ class ChangeStreamStageTest : public ChangeStreamStageTestNoSetup {
     }
 };
 
+TEST_F(ChangeStreamStageTest, ShouldRejectNonObjectArg) {
+    auto expCtx = getExpCtx();
+
+    ASSERT_THROWS_CODE(DSChangeStream::createFromBson(
+                           BSON(DSChangeStream::kStageName << "invalid").firstElement(), expCtx),
+                       AssertionException,
+                       50808);
+
+    ASSERT_THROWS_CODE(DSChangeStream::createFromBson(
+                           BSON(DSChangeStream::kStageName << 12345).firstElement(), expCtx),
+                       AssertionException,
+                       50808);
+}
+
 TEST_F(ChangeStreamStageTest, ShouldRejectUnrecognizedOption) {
     auto expCtx = getExpCtx();