SERVER-31864 applyOps command with UUID containing op must require granular privileges

(cherry picked from commit ec36d4bb79be90b24d81f4dfc2718ea2625cfc75)

2 files changed, 390 insertions, 24 deletions

diff --git a/jstests/auth/lib/commands_lib.js b/jstests/auth/lib/commands_lib.js
index 007bb31a52a..f06d8c980ac 100644
--- a/jstests/auth/lib/commands_lib.js
+++ b/jstests/auth/lib/commands_lib.js
@@ -337,6 +337,36 @@ var authCommandsLib = {
           ]
         },
         {
+          testname: "applyOps_c_create_UUID_failure",
+          command: {
+              applyOps: [{
+                  "ts": Timestamp(1474051004, 1),
+                  "h": NumberLong(0),
+                  "ui": UUID("71f1d1d7-68ca-493e-a7e9-f03c94e2e960"),
+                  "v": 2,
+                  "op": "c",
+                  "ns": firstDbName + ".$cmd",
+                  "o": {
+                      "create": "x",
+                  }
+              }]
+          },
+          skipSharded: true,
+          setup: function(db) {},
+          teardown: function(db) {
+              db.getSisterDB(firstDbName).x.drop();
+          },
+          testcases: [{
+              expectAuthzFailure: true,
+              runOnDb: adminDbName,
+              privileges: [
+                  {resource: {db: firstDbName, collection: "x"}, actions: ["createCollection"]},
+                  // Do not have forceUUID.
+                  {resource: {cluster: true}, actions: ["useUUID"]}
+              ]
+          }]
+        },
+        {
           testname: "applyOps_c_drop",
           command: {
               applyOps: [{
@@ -412,7 +442,48 @@ var authCommandsLib = {
                 roles: {root: 1, restore: 1, __system: 1},
                 privileges: [
                     {resource: {db: firstDbName, collection: "x"}, actions: ["dropCollection"]},
-                    {resource: {cluster: true}, actions: ["useUUID", "forceUUID"]}
+                    {resource: {cluster: true}, actions: ["useUUID"]}
+                ]
+              },
+          ]
+        },
+        {
+          testname: "applyOps_c_drop_UUID_failure",
+          command: function(state) {
+              return {
+                  applyOps: [{
+                      "ts": Timestamp(1474051004, 1),
+                      "h": NumberLong(0),
+                      "v": 2,
+                      "op": "c",
+                      "ui": state.uuid,
+                      "ns": firstDbName + ".$cmd",
+                      "o": {
+                          "drop": "x",
+                      }
+                  }]
+              };
+          },
+          skipSharded: true,
+          setup: function(db) {
+              var sibling = db.getSisterDB(firstDbName);
+              sibling.runCommand({create: "x"});
+
+              return {
+                  collName: sibling.x.getFullName(),
+                  uuid: getUUIDFromListCollections(sibling, sibling.x.getName())
+              };
+          },
+          teardown: function(db) {
+              db.getSisterDB(firstDbName).x.drop();
+          },
+          testcases: [
+              {
+                expectAuthzFailure: true,
+                runOnDb: adminDbName,
+                privileges: [
+                    {resource: {db: firstDbName, collection: "x"}, actions: ["dropCollection"]}
+                    // don't have useUUID privilege.
                 ]
               },
           ]
@@ -553,6 +624,221 @@ var authCommandsLib = {
                 roles: {root: 1, restore: 1, __system: 1},
                 privileges: [
                     {resource: {db: firstDbName, collection: "x"}, actions: ["insert"]},
+                    {resource: {cluster: true}, actions: ["useUUID"]}
+                ],
+              },
+          ]
+        },
+        {
+          testname: "applyOps_insert_with_nonexistent_UUID",
+          command: function(state) {
+              return {
+                  applyOps: [{
+                      "ts": Timestamp(1474051453, 1),
+                      "h": NumberLong(0),
+                      "v": 2,
+                      "op": "i",
+                      "ns": state.collName,
+                      // Given a nonexistent UUID. The command should fail.
+                      "ui": UUID("71f1d1d7-68ca-493e-a7e9-f03c94e2e960"),
+                      "o": {"_id": ObjectId("57dc3d7da4fce4358afa85b8"), "data": 5}
+                  }]
+              };
+          },
+          skipSharded: true,
+          setup: function(db) {
+              var sibling = db.getSisterDB(firstDbName);
+              sibling.runCommand({create: "x"});
+
+              return {
+                  collName: sibling.x.getFullName(),
+                  uuid: getUUIDFromListCollections(sibling, sibling.x.getName())
+              };
+          },
+          teardown: function(db) {
+              db.getSisterDB(firstDbName).x.drop();
+          },
+          testcases: [
+              {
+                // It would be an sanity check failure rather than a auth check
+                // failure.
+                expectFail: true,
+                runOnDb: adminDbName,
+                roles: {root: 1, restore: 1, __system: 1},
+                privileges: [
+                    {resource: {db: firstDbName, collection: "x"}, actions: ["insert"]},
+                    {resource: {cluster: true}, actions: ["useUUID"]}
+                ],
+              },
+          ]
+        },
+        {
+          testname: "applyOps_insert_UUID_failure",
+          command: function(state) {
+              return {
+                  applyOps: [{
+                      "ts": Timestamp(1474051453, 1),
+                      "h": NumberLong(0),
+                      "v": 2,
+                      "op": "i",
+                      "ns": state.collName,
+                      "ui": state.uuid,
+                      "o": {"_id": ObjectId("57dc3d7da4fce4358afa85b8"), "data": 5}
+                  }]
+              };
+          },
+          skipSharded: true,
+          setup: function(db) {
+              var sibling = db.getSisterDB(firstDbName);
+              sibling.runCommand({create: "x"});
+
+              return {
+                  collName: sibling.x.getFullName(),
+                  uuid: getUUIDFromListCollections(sibling, sibling.x.getName())
+              };
+          },
+          teardown: function(db) {
+              db.getSisterDB(firstDbName).x.drop();
+          },
+          testcases: [
+              {
+                expectAuthzFailure: true,
+                runOnDb: adminDbName,
+                privileges: [
+                    {resource: {db: firstDbName, collection: "x"}, actions: ["insert"]},
+                    // Don't have useUUID privilege.
+                ],
+              },
+          ]
+        },
+        {
+          testname: "applyOps_create_and_insert_UUID_failure",
+          command: function(state) {
+              return {
+                  applyOps: [
+                      {
+                        "ts": Timestamp(1474051004, 1),
+                        "h": NumberLong(0),
+                        "ui": UUID("71f1d1d7-68ca-493e-a7e9-f03c94e2e960"),
+                        "v": 2,
+                        "op": "c",
+                        "ns": firstDbName + ".$cmd",
+                        "o": {
+                            "create": "x",
+                        }
+                      },
+                      {
+                        "ts": Timestamp(1474051453, 1),
+                        "h": NumberLong(0),
+                        "v": 2,
+                        "op": "i",
+                        "ns": firstDbName + ".x",
+                        "ui": UUID("71f1d1d7-68ca-493e-a7e9-f03c94e2e960"),
+                        "o": {"_id": ObjectId("57dc3d7da4fce4358afa85b8"), "data": 5}
+                      }
+                  ]
+              };
+          },
+          skipSharded: true,
+          setup: function(db) {
+              db.getSisterDB(firstDbName).x.drop();
+          },
+          teardown: function(db) {
+              db.getSisterDB(firstDbName).x.drop();
+          },
+          testcases: [
+              {
+                // Batching createCollection and insert together is not allowed.
+                expectAuthzFailure: true,
+                runOnDb: adminDbName,
+                privileges: [
+                    {resource: {db: firstDbName, collection: "x"}, actions: ["insert"]},
+                    {resource: {cluster: true}, actions: ["useUUID", "forceUUID"]}
+                    // Require universal privilege set.
+                ],
+              },
+          ]
+        },
+        {
+          testname: "applyOps_insert_UUID_with_wrong_ns",
+          command: function(state) {
+              return {
+                  applyOps: [{
+                      "ts": Timestamp(1474051453, 1),
+                      "h": NumberLong(0),
+                      "v": 2,
+                      "op": "i",
+                      "ns":
+                          firstDbName + ".y",  // Specify wrong name but correct uuid. Should work.
+                      "ui": state.x_uuid,      // The insert should on x
+                      "o": {"_id": ObjectId("57dc3d7da4fce4358afa85b8"), "data": 5}
+                  }]
+              };
+          },
+          skipSharded: true,
+          setup: function(db) {
+              db.getSisterDB(firstDbName).x.drop();
+              db.getSisterDB(firstDbName).y.drop();
+              var sibling = db.getSisterDB(firstDbName);
+              sibling.runCommand({create: "x"});
+              sibling.runCommand({create: "y"});
+              return {x_uuid: getUUIDFromListCollections(sibling, sibling.x.getName())};
+          },
+          teardown: function(db) {
+              db.getSisterDB(firstDbName).x.drop();
+          },
+          testcases: [
+              {
+                runOnDb: adminDbName,
+                privileges: [
+                    {
+                      resource: {db: firstDbName, collection: "x"},
+                      actions: ["createCollection", "insert"]
+                    },
+                    {resource: {db: firstDbName, collection: "y"}, actions: ["createCollection"]},
+                    {resource: {cluster: true}, actions: ["useUUID", "forceUUID"]}
+                ],
+              },
+          ]
+        },
+        {
+          testname: "applyOps_insert_UUID_with_wrong_ns_failure",
+          command: function(state) {
+              return {
+                  applyOps: [{
+                      "ts": Timestamp(1474051453, 1),
+                      "h": NumberLong(0),
+                      "v": 2,
+                      "op": "i",
+                      "ns":
+                          firstDbName + ".y",  // Specify wrong name but correct uuid. Should work.
+                      "ui": state.x_uuid,      // The insert should on x
+                      "o": {"_id": ObjectId("57dc3d7da4fce4358afa85b8"), "data": 5}
+                  }]
+              };
+          },
+          skipSharded: true,
+          setup: function(db) {
+              db.getSisterDB(firstDbName).x.drop();
+              db.getSisterDB(firstDbName).y.drop();
+              var sibling = db.getSisterDB(firstDbName);
+              sibling.runCommand({create: "x"});
+              sibling.runCommand({create: "y"});
+              return {x_uuid: getUUIDFromListCollections(sibling, sibling.x.getName())};
+          },
+          teardown: function(db) {
+              db.getSisterDB(firstDbName).x.drop();
+          },
+          testcases: [
+              {
+                expectAuthzFailure: true,
+                runOnDb: adminDbName,
+                privileges: [
+                    {resource: {db: firstDbName, collection: "x"}, actions: ["createCollection"]},
+                    {
+                      resource: {db: firstDbName, collection: "y"},
+                      actions: ["createCollection", "insert"]
+                    },
                     {resource: {cluster: true}, actions: ["useUUID", "forceUUID"]}
                 ],
               },
@@ -655,12 +941,51 @@ var authCommandsLib = {
                 roles: {root: 1, __system: 1},
                 privileges: [
                     {resource: {db: firstDbName, collection: "x"}, actions: ["update"]},
-                    {resource: {cluster: true}, actions: ["useUUID", "forceUUID"]}
+                    {resource: {cluster: true}, actions: ["useUUID"]}
                 ],
               },
           ]
         },
+        {
+          testname: "applyOps_update_UUID_failure",
+          command: function(state) {
+              return {
+                  applyOps: [{
+                      "ts": Timestamp(1474053682, 1),
+                      "h": NumberLong(0),
+                      "v": 2,
+                      "op": "u",
+                      "ns": state.collName,
+                      "ui": state.uuid,
+                      "o2": {"_id": 1},
+                      "o": {"_id": 1, "data": 8}
+                  }],
+                  alwaysUpsert: false
+              };
+          },
+          skipSharded: true,
+          setup: function(db) {
+              var sibling = db.getSisterDB(firstDbName);
+              sibling.x.save({_id: 1, data: 1});
 
+              return {
+                  collName: sibling.x.getFullName(),
+                  uuid: getUUIDFromListCollections(sibling, sibling.x.getName())
+              };
+          },
+          teardown: function(db) {
+              db.getSisterDB(firstDbName).x.drop();
+          },
+          testcases: [
+              {
+                expectAuthzFailure: true,
+                runOnDb: adminDbName,
+                privileges: [
+                    {resource: {db: firstDbName, collection: "x"}, actions: ["update"]},
+                ],
+              },
+          ]
+        },
         {
           testname: "applyOps_delete",
           command: {
diff --git a/src/mongo/db/commands/apply_ops_cmd.cpp b/src/mongo/db/commands/apply_ops_cmd.cpp
index 8e0c305eca4..2f12d0871f8 100644
--- a/src/mongo/db/commands/apply_ops_cmd.cpp
+++ b/src/mongo/db/commands/apply_ops_cmd.cpp
@@ -53,10 +53,33 @@
 #include "mongo/db/service_context.h"
 #include "mongo/util/log.h"
 #include "mongo/util/scopeguard.h"
+#include "mongo/util/uuid.h"
 
 namespace mongo {
 namespace {
 
+bool checkCOperationType(const BSONObj& opObj, const StringData opName) {
+    BSONElement opTypeElem = opObj["op"];
+    checkBSONType(BSONType::String, opTypeElem);
+    const StringData opType = opTypeElem.checkAndGetStringData();
+
+    if (opType == "c"_sd) {
+        BSONElement oElem = opObj["o"];
+        checkBSONType(BSONType::Object, oElem);
+        BSONObj o = oElem.Obj();
+
+        if (o.firstElement().fieldNameStringData() == opName) {
+            return true;
+        }
+    }
+    return false;
+};
+
+UUID getUUIDFromOplogEntry(const BSONObj& oplogEntry) {
+    BSONElement uiElem = oplogEntry["ui"];
+    return uassertStatusOK(UUID::parse(uiElem));
+};
+
 Status checkOperationAuthorization(OperationContext* opCtx,
                                    const std::string& dbname,
                                    const BSONObj& oplogEntry,
@@ -80,6 +103,14 @@ Status checkOperationAuthorization(OperationContext* opCtx,
     checkBSONType(BSONType::String, nsElem);
     NamespaceString ns(oplogEntry["ns"].checkAndGetStringData());
 
+    if (oplogEntry.hasField("ui"_sd)) {
+        // ns by UUID overrides the ns specified if they are different.
+        auto& uuidCatalog = UUIDCatalog::get(opCtx);
+        NamespaceString uuidCollNS = uuidCatalog.lookupNSSByUUID(getUUIDFromOplogEntry(oplogEntry));
+        if (!uuidCollNS.isEmpty() && uuidCollNS != ns)
+            ns = uuidCollNS;
+    }
+
     BSONElement oElem = oplogEntry["o"];
     checkBSONType(BSONType::Object, oElem);
     BSONObj o = oElem.Obj();
@@ -136,35 +167,20 @@ Status checkOperationAuthorization(OperationContext* opCtx,
     return Status(ErrorCodes::FailedToParse, "Unrecognized opType");
 }
 
-enum class ApplyOpsValidity { kOk, kNeedsForceAndUseUUID, kNeedsSuperuser };
+enum class ApplyOpsValidity { kOk, kNeedsUseUUID, kNeedsForceAndUseUUID, kNeedsSuperuser };
 
 /**
  * Returns kNeedsSuperuser, if the provided applyOps command contains
- * an empty applyOps command. Returns kNeedForceAndUseUUID if an operation contains a UUID,
- * indicating that privileges to manipulate UUIDs are required. Returns kOk if no conditions
+ * an empty applyOps command or createCollection/renameCollection commands are mixed in applyOps
+ * batch. Returns kNeedForceAndUseUUID if an operation contains a UUID, and will create a collection
+ * with the user-specified UUID. Returns
+ * kNeedsUseUUID if the operation contains a UUID. Returns kOk if no conditions
  * which must be specially handled are detected. May throw exceptions if the input is malformed.
  */
 ApplyOpsValidity validateApplyOpsCommand(const BSONObj& cmdObj) {
     const size_t maxApplyOpsDepth = 10;
     std::stack<std::pair<size_t, BSONObj>> toCheck;
 
-    auto operationContainsApplyOps = [](const BSONObj& opObj) {
-        BSONElement opTypeElem = opObj["op"];
-        checkBSONType(BSONType::String, opTypeElem);
-        const StringData opType = opTypeElem.checkAndGetStringData();
-
-        if (opType == "c"_sd) {
-            BSONElement oElem = opObj["o"];
-            checkBSONType(BSONType::Object, oElem);
-            BSONObj o = oElem.Obj();
-
-            if (o.firstElement().fieldNameStringData() == "applyOps"_sd) {
-                return true;
-            }
-        }
-        return false;
-    };
-
     auto operationContainsUUID = [](const BSONObj& opObj) {
         auto anyTopLevelElementIsUUID = [](const BSONObj& opObj) {
             for (const BSONElement opElement : opObj) {
@@ -214,6 +230,20 @@ ApplyOpsValidity validateApplyOpsCommand(const BSONObj& cmdObj) {
             return ApplyOpsValidity::kNeedsSuperuser;
         }
 
+        // createCollection and renameCollection are only allowed to be applied
+        // individually. Ensure there is no create/renameCollection in a batch
+        // of size greater than 1.
+        if (applyOpsObj.firstElement().Array().size() > 1) {
+            for (const BSONElement& e : applyOpsObj.firstElement().Array()) {
+                checkBSONType(BSONType::Object, e);
+                auto oplogEntry = e.Obj();
+                if (checkCOperationType(oplogEntry, "create"_sd) ||
+                    checkCOperationType(oplogEntry, "renameCollection"_sd)) {
+                    return ApplyOpsValidity::kNeedsSuperuser;
+                }
+            }
+        }
+
         // For each applyOps command, iterate the ops.
         for (BSONElement element : applyOpsObj.firstElement().Array()) {
             checkBSONType(BSONType::Object, element);
@@ -229,12 +259,16 @@ ApplyOpsValidity validateApplyOpsCommand(const BSONObj& cmdObj) {
             }
 
             // If the op uses any UUIDs at all then the user must possess extra privileges.
-            if (ret == ApplyOpsValidity::kOk && opHasUUIDs) {
+            if (opHasUUIDs && ret == ApplyOpsValidity::kOk)
+                ret = ApplyOpsValidity::kNeedsUseUUID;
+            if (opHasUUIDs && checkCOperationType(opObj, "create"_sd)) {
+                // If the op is 'c' and forces the server to ingest a collection
+                // with a specific, user defined UUID.
                 ret = ApplyOpsValidity::kNeedsForceAndUseUUID;
             }
 
             // If the op contains a nested applyOps...
-            if (operationContainsApplyOps(opObj)) {
+            if (checkCOperationType(opObj, "applyOps"_sd)) {
                 // And we've recursed too far, then bail out.
                 uassert(ErrorCodes::FailedToParse,
                         "Too many nested applyOps",
@@ -291,6 +325,13 @@ public:
             }
             validity = ApplyOpsValidity::kOk;
         }
+        if (validity == ApplyOpsValidity::kNeedsUseUUID) {
+            if (!authSession->isAuthorizedForActionsOnResource(
+                    ResourcePattern::forClusterResource(), ActionType::useUUID)) {
+                return Status(ErrorCodes::Unauthorized, "Unauthorized");
+            }
+            validity = ApplyOpsValidity::kOk;
+        }
         fassert(40314, validity == ApplyOpsValidity::kOk);
 
         boost::optional<DisableDocumentValidation> maybeDisableValidation;