SERVER-25738 authz for views for sharded create/collMod

Merges authz checks for create, collMod and aggregate into AuthorizationSession.

10 files changed, 324 insertions, 288 deletions

diff --git a/jstests/auth/views_authz.js b/jstests/auth/views_authz.js
index 4a851583110..376e85c7e64 100644
--- a/jstests/auth/views_authz.js
+++ b/jstests/auth/views_authz.js
@@ -1,101 +1,145 @@
 /**
  * Tests authorization special cases with views. These are special exceptions that prohibit certain
  * operations on views even if the user has an explicit privilege on that view.
+ *
+ * TODO(SERVER-25526): merge this test into jstests/auth/libs/commands_lib.js.
  */
 (function() {
     "use strict";
 
-    let mongod = MongoRunner.runMongod({auth: "", bind_ip: "127.0.0.1"});
-
-    // Create the admin user.
-    let adminDB = mongod.getDB("admin");
-    assert.commandWorked(adminDB.runCommand({createUser: "admin", pwd: "admin", roles: ["root"]}));
-    assert.eq(1, adminDB.auth("admin", "admin"));
+    function runTest(conn) {
+        // Create the admin user.
+        let adminDB = conn.getDB("admin");
+        assert.commandWorked(
+            adminDB.runCommand({createUser: "admin", pwd: "admin", roles: ["root"]}));
+        assert.eq(1, adminDB.auth("admin", "admin"));
 
-    const viewsDBName = "views_authz";
-    let viewsDB = adminDB.getSiblingDB(viewsDBName);
-    viewsDB.dropAllUsers();
-    viewsDB.logout();
+        const viewsDBName = "views_authz";
+        let viewsDB = adminDB.getSiblingDB(viewsDBName);
+        viewsDB.dropAllUsers();
+        viewsDB.logout();
 
-    // Create a user who can read, create and modify a view 'view' and a read a namespace
-    // 'permitted' but does not have access to 'forbidden'.
-    assert.commandWorked(viewsDB.runCommand({
-        createRole: "readWriteView",
-        privileges: [
-            {
-              resource: {db: viewsDBName, collection: "view"},
-              actions: ["find", "createCollection", "collMod"]
-            },
-            {resource: {db: viewsDBName, collection: "view2"}, actions: ["find"]},
-            {resource: {db: viewsDBName, collection: "permitted"}, actions: ["find"]}
-        ],
-        roles: []
-    }));
-    assert.commandWorked(
-        viewsDB.runCommand({createUser: "viewUser", pwd: "pwd", roles: ["readWriteView"]}));
+        // Create a user who can read, create and modify a view 'view' and a read a namespace
+        // 'permitted' but does not have access to 'forbidden'.
+        assert.commandWorked(viewsDB.runCommand({
+            createRole: "readWriteView",
+            privileges: [
+                {
+                  resource: {db: viewsDBName, collection: "view"},
+                  actions: ["find", "createCollection", "collMod"]
+                },
+                {resource: {db: viewsDBName, collection: "view2"}, actions: ["find"]},
+                {resource: {db: viewsDBName, collection: "permitted"}, actions: ["find"]}
+            ],
+            roles: []
+        }));
+        assert.commandWorked(
+            viewsDB.runCommand({createUser: "viewUser", pwd: "pwd", roles: ["readWriteView"]}));
 
-    adminDB.logout();
-    assert.eq(1, viewsDB.auth("viewUser", "pwd"));
+        adminDB.logout();
+        assert.eq(1, viewsDB.auth("viewUser", "pwd"));
 
-    const lookupStage = {$lookup: {from: "forbidden", localField: "x", foreignField: "x", as: "y"}};
-    const graphLookupStage = {
-        $graphLookup: {
-            from: "forbidden",
-            startWith: [],
-            connectFromField: "x",
-            connectToField: "x",
-            as: "y"
-        }
-    };
+        const lookupStage = {
+            $lookup: {from: "forbidden", localField: "x", foreignField: "x", as: "y"}
+        };
+        const graphLookupStage = {
+            $graphLookup: {
+                from: "forbidden",
+                startWith: [],
+                connectFromField: "x",
+                connectToField: "x",
+                as: "y"
+            }
+        };
 
-    // You cannot create a view if you have both the 'createCollection' and 'find' actions on that
-    // view but not the 'find' action on all of the dependent namespaces.
-    assert.commandFailedWithCode(viewsDB.createView("view", "forbidden", []),
-                                 ErrorCodes.Unauthorized);
-    assert.commandFailedWithCode(viewsDB.createView("view", "permitted", [lookupStage]),
-                                 ErrorCodes.Unauthorized);
-    assert.commandFailedWithCode(viewsDB.createView("view", "permitted", [graphLookupStage]),
-                                 ErrorCodes.Unauthorized);
-    assert.commandFailedWithCode(
-        viewsDB.createView("view", "permitted", [{$facet: {a: [lookupStage]}}]),
-        ErrorCodes.Unauthorized);
-    assert.commandFailedWithCode(
-        viewsDB.createView("view", "permitted", [{$facet: {b: [graphLookupStage]}}]),
-        ErrorCodes.Unauthorized);
+        // You cannot create a view if you have both the 'createCollection' and 'find' actions on
+        // that view but not the 'find' action on all of the dependent namespaces.
+        assert.commandFailedWithCode(viewsDB.createView("view", "forbidden", []),
+                                     ErrorCodes.Unauthorized,
+                                     "created a readable view on an unreadable collection");
+        assert.commandFailedWithCode(
+            viewsDB.createView("view", "permitted", [lookupStage]),
+            ErrorCodes.Unauthorized,
+            "created a readable view on an unreadable collection via $lookup");
+        assert.commandFailedWithCode(
+            viewsDB.createView("view", "permitted", [graphLookupStage]),
+            ErrorCodes.Unauthorized,
+            "created a readable view on an unreadable collection via $graphLookup");
+        assert.commandFailedWithCode(
+            viewsDB.createView("view", "permitted", [{$facet: {a: [lookupStage]}}]),
+            ErrorCodes.Unauthorized,
+            "created a readable view on an unreadable collection via $lookup in a $facet");
+        assert.commandFailedWithCode(
+            viewsDB.createView("view", "permitted", [{$facet: {b: [graphLookupStage]}}]),
+            ErrorCodes.Unauthorized,
+            "created a readable view on an unreadable collection via $graphLookup in a $facet");
 
-    assert.commandWorked(viewsDB.createView("view", "permitted", [{$match: {x: 1}}]));
+        assert.commandWorked(viewsDB.createView("view", "permitted", [{$match: {x: 1}}]));
 
-    // You cannot modify a view if you have both the 'collMod' and 'find' actions on that view but
-    // not the 'find' action on all of the dependent namespaces.
-    assert.commandFailedWithCode(
-        viewsDB.runCommand({collMod: "view", viewOn: "forbidden", pipeline: [{$match: {}}]}),
-        ErrorCodes.Unauthorized);
-    assert.commandFailedWithCode(
-        viewsDB.runCommand({collMod: "view", viewOn: "permitted", pipeline: [lookupStage]}),
-        ErrorCodes.Unauthorized);
-    assert.commandFailedWithCode(
-        viewsDB.runCommand({collMod: "view", viewOn: "permitted", pipeline: [graphLookupStage]}),
-        ErrorCodes.Unauthorized);
-    assert.commandFailedWithCode(
-        viewsDB.runCommand(
-            {collMod: "view", viewOn: "permitted", pipeline: [{$facet: {a: [lookupStage]}}]}),
-        ErrorCodes.Unauthorized);
-    assert.commandFailedWithCode(
-        viewsDB.runCommand(
-            {collMod: "view", viewOn: "permitted", pipeline: [{$facet: {b: [graphLookupStage]}}]}),
-        ErrorCodes.Unauthorized);
+        // You cannot modify a view if you have both the 'collMod' and 'find' actions on that view
+        // but not the 'find' action on all of the dependent namespaces.
+        assert.commandFailedWithCode(
+            viewsDB.runCommand({collMod: "view", viewOn: "forbidden", pipeline: [{$match: {}}]}),
+            ErrorCodes.Unauthorized,
+            "modified a view to read an unreadable collection");
+        assert.commandFailedWithCode(
+            viewsDB.runCommand({collMod: "view", viewOn: "permitted", pipeline: [lookupStage]}),
+            ErrorCodes.Unauthorized,
+            "modified a view to read an unreadable collection via $lookup");
+        assert.commandFailedWithCode(
+            viewsDB.runCommand(
+                {collMod: "view", viewOn: "permitted", pipeline: [graphLookupStage]}),
+            ErrorCodes.Unauthorized,
+            "modified a view to read an unreadable collection via $graphLookup");
+        assert.commandFailedWithCode(
+            viewsDB.runCommand(
+                {collMod: "view", viewOn: "permitted", pipeline: [{$facet: {a: [lookupStage]}}]}),
+            ErrorCodes.Unauthorized,
+            "modified a view to read an unreadable collection via $lookup in a $facet");
+        assert.commandFailedWithCode(
+            viewsDB.runCommand({
+                collMod: "view",
+                viewOn: "permitted",
+                pipeline: [{$facet: {b: [graphLookupStage]}}]
+            }),
+            ErrorCodes.Unauthorized,
+            "modified a view to read an unreadable collection via $graphLookup in a $facet");
 
-    // Performing a find on a readable view returns a cursor that allows us to perform a getMore
-    // even if the underlying collection is unreadable.
-    assert.eq(1, adminDB.auth("admin", "admin"));
-    assert.commandWorked(viewsDB.createView("view2", "forbidden", []));
-    for (let i = 0; i < 10; i++) {
-        assert.writeOK(viewsDB.forbidden.insert({x: 1}));
+        // Performing a find on a readable view returns a cursor that allows us to perform a getMore
+        // even if the underlying collection is unreadable.
+        // TODO(SERVER-24771): getMore does not work yet for sharded clusters
+        assert.eq(1, adminDB.auth("admin", "admin"));
+        let isMaster = adminDB.runCommand({isMaster: 1});
+        assert.commandWorked(isMaster);
+        const isMongos = (isMaster.msg === "isdbgrid");
+        if (!isMongos) {
+            assert.commandWorked(viewsDB.createView("view2", "forbidden", []));
+            for (let i = 0; i < 10; i++) {
+                assert.writeOK(viewsDB.forbidden.insert({x: 1}));
+            }
+            adminDB.logout();
+            assert.commandFailedWithCode(
+                viewsDB.runCommand({find: "forbidden"}),
+                ErrorCodes.Unauthorized,
+                "successfully performed a find on an unreadable namespace");
+            let res = viewsDB.runCommand({find: "view2", batchSize: 1});
+            assert.commandWorked(res, "could not perform a find on a readable view");
+            assert.eq(res.cursor.ns,
+                      "views_authz.view2",
+                      "performing find on a view does not return a cursor on the view namespace");
+            assert.commandWorked(viewsDB.runCommand({getMore: res.cursor.id, collection: "view2"}),
+                                 "could not perform getMore on a readable view");
+        }
     }
-    adminDB.logout();
-    assert.commandFailedWithCode(viewsDB.runCommand({find: "forbidden"}), ErrorCodes.Unauthorized);
-    let res = viewsDB.runCommand({find: "view2", batchSize: 1});
-    assert.commandWorked(res);
-    assert.eq(res.cursor.ns, "views_authz.view2");
-    assert.commandWorked(viewsDB.runCommand({getMore: res.cursor.id, collection: "view2"}));
+
+    // Run the test on a standalone.
+    let mongod = MongoRunner.runMongod({auth: "", bind_ip: "127.0.0.1"});
+    runTest(mongod);
+    MongoRunner.stopMongod(mongod);
+
+    // Run the test on a sharded cluster.
+    let cluster = new ShardingTest(
+        {shards: 1, mongos: 1, keyFile: "jstests/libs/key1", other: {shardOptions: {auth: ""}}});
+    runTest(cluster);
+    cluster.stop();
 }());
diff --git a/src/mongo/db/auth/authorization_session.cpp b/src/mongo/db/auth/authorization_session.cpp
index 990b6a6eb9e..271561cc06a 100644
--- a/src/mongo/db/auth/authorization_session.cpp
+++ b/src/mongo/db/auth/authorization_session.cpp
@@ -43,15 +43,19 @@
 #include "mongo/db/auth/privilege.h"
 #include "mongo/db/auth/security_key.h"
 #include "mongo/db/auth/user_management_commands_parser.h"
+#include "mongo/db/bson/dotted_path_support.h"
+#include "mongo/db/catalog/document_validation.h"
 #include "mongo/db/client.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/db/namespace_string.h"
+#include "mongo/db/pipeline/pipeline.h"
 #include "mongo/util/assert_util.h"
 #include "mongo/util/log.h"
 #include "mongo/util/mongoutils/str.h"
 
 namespace mongo {
 
+namespace dps = ::mongo::dotted_path_support;
 using std::vector;
 
 namespace {
@@ -178,6 +182,89 @@ PrivilegeVector AuthorizationSession::getDefaultPrivileges() {
     return defaultPrivileges;
 }
 
+void AuthorizationSession::_addPrivilegesForStage(const std::string& db,
+                                                  const BSONObj& cmdObj,
+                                                  PrivilegeVector* requiredPrivileges,
+                                                  BSONObj stageSpec,
+                                                  bool haveRecursed) {
+    StringData stageName = stageSpec.firstElementFieldName();
+    if (stageName == "$out" && stageSpec.firstElementType() == BSONType::String) {
+        NamespaceString outputNs(db, stageSpec.firstElement().str());
+        uassert(17139,
+                mongoutils::str::stream() << "Invalid $out target namespace, " << outputNs.ns(),
+                outputNs.isValid());
+
+        ActionSet actions;
+        actions.addAction(ActionType::remove);
+        actions.addAction(ActionType::insert);
+        if (shouldBypassDocumentValidationForCommand(cmdObj)) {
+            actions.addAction(ActionType::bypassDocumentValidation);
+        }
+        Privilege::addPrivilegeToPrivilegeVector(
+            requiredPrivileges, Privilege(ResourcePattern::forExactNamespace(outputNs), actions));
+    } else if (stageName == "$lookup" && stageSpec.firstElementType() == BSONType::Object) {
+        NamespaceString fromNs(db, stageSpec.firstElement()["from"].str());
+        Privilege::addPrivilegeToPrivilegeVector(
+            requiredPrivileges,
+            Privilege(ResourcePattern::forExactNamespace(fromNs), ActionType::find));
+    } else if (stageName == "$graphLookup" && stageSpec.firstElementType() == BSONType::Object) {
+        NamespaceString fromNs(db, stageSpec.firstElement()["from"].str());
+        Privilege::addPrivilegeToPrivilegeVector(
+            requiredPrivileges,
+            Privilege(ResourcePattern::forExactNamespace(fromNs), ActionType::find));
+    } else if (stageName == "$facet" && stageSpec.firstElementType() == BSONType::Object &&
+               !haveRecursed) {
+        // Add privileges of sub-stages, but only if we haven't recursed already. We don't want to
+        // get a stack overflow while checking privileges. If we ever allow a $facet stage inside of
+        // a $facet stage, this code will have to be modified to avoid causing a stack overflow, but
+        // still check all required privileges of nested stages.
+        for (auto&& subPipeline : stageSpec.firstElement().embeddedObject()) {
+            if (subPipeline.type() == BSONType::Array) {
+                for (auto&& subPipeStageSpec : subPipeline.embeddedObject()) {
+                    _addPrivilegesForStage(db,
+                                           cmdObj,
+                                           requiredPrivileges,
+                                           subPipeStageSpec.embeddedObjectUserCheck(),
+                                           true);
+                }
+            }
+        }
+    }
+}
+
+Status AuthorizationSession::checkAuthForAggregate(const NamespaceString& ns,
+                                                   const BSONObj& cmdObj) {
+    std::string db(ns.db().toString());
+    auto inputResource = ResourcePattern::forExactNamespace(ns);
+    uassert(
+        17138, mongoutils::str::stream() << "Invalid input namespace, " << ns.ns(), ns.isValid());
+
+    PrivilegeVector privileges;
+
+    if (dps::extractElementAtPath(cmdObj, "pipeline.0.$indexStats")) {
+        Privilege::addPrivilegeToPrivilegeVector(
+            &privileges,
+            Privilege(ResourcePattern::forAnyNormalResource(), ActionType::indexStats));
+    } else if (dps::extractElementAtPath(cmdObj, "pipeline.0.$collStats")) {
+        Privilege::addPrivilegeToPrivilegeVector(&privileges,
+                                                 Privilege(inputResource, ActionType::collStats));
+    } else {
+        // If no source requiring an alternative permission scheme is specified then default to
+        // requiring find() privileges on the given namespace.
+        Privilege::addPrivilegeToPrivilegeVector(&privileges,
+                                                 Privilege(inputResource, ActionType::find));
+    }
+
+    BSONObj pipeline = cmdObj.getObjectField("pipeline");
+    for (auto&& stageElem : pipeline) {
+        _addPrivilegesForStage(db, cmdObj, &privileges, stageElem.embeddedObjectUserCheck());
+    }
+
+    if (isAuthorizedForPrivileges(privileges))
+        return Status::OK();
+    return Status(ErrorCodes::Unauthorized, "unauthorized");
+}
+
 Status AuthorizationSession::checkAuthForFind(const NamespaceString& ns, bool hasTerm) {
     if (MONGO_unlikely(ns.isCommand())) {
         return Status(ErrorCodes::InternalError,
@@ -330,6 +417,69 @@ Status AuthorizationSession::checkAuthForKillCursors(const NamespaceString& ns,
     return Status::OK();
 }
 
+Status AuthorizationSession::checkAuthForCreate(const NamespaceString& ns, const BSONObj& cmdObj) {
+    if (cmdObj["capped"].trueValue() &&
+        !isAuthorizedForActionsOnNamespace(ns, ActionType::convertToCapped)) {
+        return Status(ErrorCodes::Unauthorized, "unauthorized");
+    }
+
+    const bool hasCreateCollectionAction =
+        isAuthorizedForActionsOnNamespace(ns, ActionType::createCollection);
+
+    // If attempting to create a view, check for additional required privileges.
+    if (cmdObj["viewOn"]) {
+        // You need the createCollection action on this namespace; the insert action is not
+        // sufficient.
+        if (!hasCreateCollectionAction) {
+            return Status(ErrorCodes::Unauthorized, "unauthorized");
+        }
+        return checkAuthForCreateOrModifyView(ns, cmdObj);
+    }
+
+    // To create a regular collection, ActionType::createCollection or ActionType::insert are
+    // both acceptable.
+    if (hasCreateCollectionAction || isAuthorizedForActionsOnNamespace(ns, ActionType::insert)) {
+        return Status::OK();
+    }
+
+    return Status(ErrorCodes::Unauthorized, "unauthorized");
+}
+
+Status AuthorizationSession::checkAuthForCollMod(const NamespaceString& ns, const BSONObj& cmdObj) {
+    if (!isAuthorizedForActionsOnNamespace(ns, ActionType::collMod)) {
+        return Status(ErrorCodes::Unauthorized, "unauthorized");
+    }
+
+    // Check for additional required privileges if attempting to modify a view.
+    if (cmdObj["viewOn"] || cmdObj["pipeline"]) {
+        return checkAuthForCreateOrModifyView(ns, cmdObj);
+    }
+
+    return Status::OK();
+}
+
+Status AuthorizationSession::checkAuthForCreateOrModifyView(const NamespaceString& ns,
+                                                            const BSONObj& cmdObj) {
+    // It's safe to allow a user to create or modify a view if they can't read it anyway.
+    if (!isAuthorizedForActionsOnNamespace(ns, ActionType::find)) {
+        return Status::OK();
+    }
+
+    // The user can read the view they're trying to create/modify, so we must ensure that they also
+    // have the find action on all namespaces in "viewOn" and "pipeline". If "pipeline" is not
+    // specified, default to the empty pipeline.
+    auto viewPipeline =
+        cmdObj.hasField("pipeline") ? BSONArray(cmdObj["pipeline"].Obj()) : BSONArray();
+
+
+    // This check ignores some invalid pipeline specifications. For example, if a user specifies a
+    // view definition with an invalid specification like {$lookup: "blah"}, the authorization check
+    // will succeed but the pipeline will fail to parse later in Command::run().
+    NamespaceString viewOnNss(ns.db(), cmdObj["viewOn"].checkAndGetStringData());
+    return checkAuthForAggregate(
+        viewOnNss, BSON("aggregate" << viewOnNss.coll() << "pipeline" << viewPipeline));
+}
+
 Status AuthorizationSession::checkAuthorizedToGrantPrivilege(const Privilege& privilege) {
     const ResourcePattern& resource = privilege.getResourcePattern();
     if (resource.isDatabasePattern() || resource.isExactNamespacePattern()) {
diff --git a/src/mongo/db/auth/authorization_session.h b/src/mongo/db/auth/authorization_session.h
index f2e6fd0c91c..3e75e0ed381 100644
--- a/src/mongo/db/auth/authorization_session.h
+++ b/src/mongo/db/auth/authorization_session.h
@@ -174,6 +174,25 @@ public:
     // identifier.
     Status checkAuthForKillCursors(const NamespaceString& ns, long long cursorID);
 
+    // Checks if this connection has the privileges necessary to run the aggregation pipeline
+    // specified in 'cmdObj' on the namespace 'ns'.
+    Status checkAuthForAggregate(const NamespaceString& ns, const BSONObj& cmdObj);
+
+    // Checks if this connection has the privileges necessary to create 'ns' with the options
+    // supplied in 'cmdObj'.
+    Status checkAuthForCreate(const NamespaceString& ns, const BSONObj& cmdObj);
+
+    // Checks if this connection has the privileges necessary to modify 'ns' with the options
+    // supplied in 'cmdObj'.
+    Status checkAuthForCollMod(const NamespaceString& ns, const BSONObj& cmdObj);
+
+    // Checks if this connection has the privileges necessary to create or modify the view 'ns'.
+    // Call this function after verifying that the user has the 'createCollection' or 'collMod'
+    // action, respectively.
+    //
+    // 'cmdObj' must have a String field named 'viewOn'.
+    Status checkAuthForCreateOrModifyView(const NamespaceString& ns, const BSONObj& cmdObj);
+
     // Checks if this connection has the privileges necessary to grant the given privilege
     // to a role.
     Status checkAuthorizedToGrantPrivilege(const Privilege& privilege);
@@ -274,6 +293,13 @@ private:
     // lock on the admin database (to update out-of-date user privilege information).
     bool _isAuthorizedForPrivilege(const Privilege& privilege);
 
+    // Helper for recursively checking for privileges in an aggregation pipeline.
+    void _addPrivilegesForStage(const std::string& db,
+                                const BSONObj& cmdObj,
+                                PrivilegeVector* requiredPrivileges,
+                                BSONObj stageSpec,
+                                bool haveRecursed = false);
+
     std::unique_ptr<AuthzSessionExternalState> _externalState;
 
     // All Users who have been authenticated on this connection.
diff --git a/src/mongo/db/commands/dbcommands.cpp b/src/mongo/db/commands/dbcommands.cpp
index 2e8c49c4bf9..0f18c9ad943 100644
--- a/src/mongo/db/commands/dbcommands.cpp
+++ b/src/mongo/db/commands/dbcommands.cpp
@@ -77,7 +77,6 @@
 #include "mongo/db/namespace_string.h"
 #include "mongo/db/op_observer.h"
 #include "mongo/db/ops/insert.h"
-#include "mongo/db/pipeline/pipeline.h"
 #include "mongo/db/query/get_executor.h"
 #include "mongo/db/query/internal_plans.h"
 #include "mongo/db/query/query_planner.h"
@@ -118,41 +117,6 @@ using std::string;
 using std::stringstream;
 using std::unique_ptr;
 
-namespace {
-/**
- * Checks for additional required privileges when creating or modifying a view. Call this function
- * after verifying that the user has the "createCollection" or "collMod" action, respectively.
- *
- * 'cmdObj' must have a String field named 'viewOn'.
- */
-Status canCreateOrModifyView(Client* client,
-                             const std::string& dbname,
-                             const BSONObj& cmdObj,
-                             ResourcePattern resource) {
-    AuthorizationSession* authzSession = AuthorizationSession::get(client);
-
-    // It's safe to allow a user to create or modify a view if they can't read it anyway.
-    if (!authzSession->isAuthorizedForActionsOnResource(resource, ActionType::find)) {
-        return Status::OK();
-    }
-
-    // The user can read the view they're trying to create/modify, so we must ensure that they also
-    // have the find action on all namespaces in "viewOn" and "pipeline". If "pipeline" is not
-    // specified, default to the empty pipeline.
-    auto viewPipeline =
-        cmdObj.hasField("pipeline") ? BSONArray(cmdObj["pipeline"].Obj()) : BSONArray();
-
-    // This check ignores some invalid pipeline specifications. For example, if a user specifies a
-    // view definition with an invalid specification like {$lookup: "blah"}, the authorization check
-    // will succeed but the pipeline will fail to parse later in Command::run().
-    return Pipeline::checkAuthForCommand(
-        client,
-        dbname,
-        BSON("aggregate" << cmdObj["viewOn"].checkAndGetStringData() << "pipeline"
-                         << viewPipeline));
-}
-}  // namespace
-
 class CmdShutdownMongoD : public CmdShutdown {
 public:
     virtual void help(stringstream& help) const {
@@ -558,37 +522,10 @@ public:
     virtual Status checkAuthForCommand(Client* client,
                                        const std::string& dbname,
                                        const BSONObj& cmdObj) {
-        AuthorizationSession* authzSession = AuthorizationSession::get(client);
-        auto cmdNsResource = parseResourcePattern(dbname, cmdObj);
-        if (cmdObj["capped"].trueValue()) {
-            if (!authzSession->isAuthorizedForActionsOnResource(cmdNsResource,
-                                                                ActionType::convertToCapped)) {
-                return Status(ErrorCodes::Unauthorized, "unauthorized");
-            }
-        }
-
-        const bool hasCreateCollectionAction = authzSession->isAuthorizedForActionsOnResource(
-            cmdNsResource, ActionType::createCollection);
-
-        // If attempting to create a view, check for additional required privileges.
-        if (cmdObj["viewOn"]) {
-            // You need the createCollection action on this namespace; the insert action is not
-            // sufficient.
-            if (!hasCreateCollectionAction) {
-                return Status(ErrorCodes::Unauthorized, "unauthorized");
-            }
-            return canCreateOrModifyView(client, dbname, cmdObj, cmdNsResource);
-        }
-
-        // To create a regular collection, ActionType::createCollection or ActionType::insert are
-        // both acceptable.
-        if (hasCreateCollectionAction ||
-            authzSession->isAuthorizedForActionsOnResource(cmdNsResource, ActionType::insert)) {
-            return Status::OK();
-        }
-
-        return Status(ErrorCodes::Unauthorized, "unauthorized");
+        NamespaceString nss(parseNs(dbname, cmdObj));
+        return AuthorizationSession::get(client)->checkAuthForCreate(nss, cmdObj);
     }
+
     virtual bool run(OperationContext* txn,
                      const string& dbname,
                      BSONObj& cmdObj,
@@ -1006,19 +943,8 @@ public:
     virtual Status checkAuthForCommand(Client* client,
                                        const std::string& dbname,
                                        const BSONObj& cmdObj) {
-        AuthorizationSession* authzSession = AuthorizationSession::get(client);
-        if (!authzSession->isAuthorizedForActionsOnResource(parseResourcePattern(dbname, cmdObj),
-                                                            ActionType::collMod)) {
-            return Status(ErrorCodes::Unauthorized, "unauthorized");
-        }
-
-        // Check for additional required privileges if attempting to modify a view.
-        if (cmdObj["viewOn"] || cmdObj["pipeline"]) {
-            return canCreateOrModifyView(
-                client, dbname, cmdObj, parseResourcePattern(dbname, cmdObj));
-        }
-
-        return Status::OK();
+        NamespaceString nss(parseNs(dbname, cmdObj));
+        return AuthorizationSession::get(client)->checkAuthForCollMod(nss, cmdObj);
     }
 
     bool run(OperationContext* txn,
diff --git a/src/mongo/db/commands/pipeline_command.cpp b/src/mongo/db/commands/pipeline_command.cpp
index 20fa4ce7510..af46c6f1d80 100644
--- a/src/mongo/db/commands/pipeline_command.cpp
+++ b/src/mongo/db/commands/pipeline_command.cpp
@@ -36,6 +36,7 @@
 #include "mongo/base/init.h"
 #include "mongo/db/auth/action_set.h"
 #include "mongo/db/auth/action_type.h"
+#include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/auth/privilege.h"
 #include "mongo/db/catalog/database.h"
 #include "mongo/db/client.h"
@@ -306,7 +307,8 @@ public:
     Status checkAuthForCommand(Client* client,
                                const std::string& dbname,
                                const BSONObj& cmdObj) final {
-        return Pipeline::checkAuthForCommand(client, dbname, cmdObj);
+        NamespaceString nss(parseNs(dbname, cmdObj));
+        return AuthorizationSession::get(client)->checkAuthForAggregate(nss, cmdObj);
     }
 
     bool runParsed(OperationContext* txn,
diff --git a/src/mongo/db/pipeline/SConscript b/src/mongo/db/pipeline/SConscript
index b67494e2b2e..9e7e83a5a4d 100644
--- a/src/mongo/db/pipeline/SConscript
+++ b/src/mongo/db/pipeline/SConscript
@@ -255,7 +255,6 @@ env.Library(
         'document_value',
         'expression_context',
         '$BUILD_DIR/mongo/db/auth/authorization_manager_global',
-        '$BUILD_DIR/mongo/db/auth/authcore',
         '$BUILD_DIR/mongo/db/bson/dotted_path_support',
         '$BUILD_DIR/mongo/db/query/collation/collator_interface',
         '$BUILD_DIR/mongo/db/query/collation/collator_factory_interface',
diff --git a/src/mongo/db/pipeline/pipeline.cpp b/src/mongo/db/pipeline/pipeline.cpp
index 9dd627202e4..ac70ef8faf7 100644
--- a/src/mongo/db/pipeline/pipeline.cpp
+++ b/src/mongo/db/pipeline/pipeline.cpp
@@ -33,9 +33,6 @@
 #include "mongo/db/pipeline/pipeline_optimizations.h"
 
 #include "mongo/base/error_codes.h"
-#include "mongo/db/auth/action_set.h"
-#include "mongo/db/auth/authorization_session.h"
-#include "mongo/db/auth/privilege.h"
 #include "mongo/db/bson/dotted_path_support.h"
 #include "mongo/db/catalog/document_validation.h"
 #include "mongo/db/jsobj.h"
@@ -128,93 +125,6 @@ void Pipeline::optimizePipeline() {
     stitch();
 }
 
-namespace {
-
-void addPrivilegesForStage(const std::string& db,
-                           const BSONObj& cmdObj,
-                           PrivilegeVector* requiredPrivileges,
-                           BSONObj stageSpec,
-                           bool haveRecursed = false) {
-    StringData stageName = stageSpec.firstElementFieldName();
-    if (stageName == "$out" && stageSpec.firstElementType() == BSONType::String) {
-        NamespaceString outputNs(db, stageSpec.firstElement().str());
-        uassert(17139,
-                mongoutils::str::stream() << "Invalid $out target namespace, " << outputNs.ns(),
-                outputNs.isValid());
-
-        ActionSet actions;
-        actions.addAction(ActionType::remove);
-        actions.addAction(ActionType::insert);
-        if (shouldBypassDocumentValidationForCommand(cmdObj)) {
-            actions.addAction(ActionType::bypassDocumentValidation);
-        }
-        Privilege::addPrivilegeToPrivilegeVector(
-            requiredPrivileges, Privilege(ResourcePattern::forExactNamespace(outputNs), actions));
-    } else if (stageName == "$lookup" && stageSpec.firstElementType() == BSONType::Object) {
-        NamespaceString fromNs(db, stageSpec.firstElement()["from"].str());
-        Privilege::addPrivilegeToPrivilegeVector(
-            requiredPrivileges,
-            Privilege(ResourcePattern::forExactNamespace(fromNs), ActionType::find));
-    } else if (stageName == "$graphLookup" && stageSpec.firstElementType() == BSONType::Object) {
-        NamespaceString fromNs(db, stageSpec.firstElement()["from"].str());
-        Privilege::addPrivilegeToPrivilegeVector(
-            requiredPrivileges,
-            Privilege(ResourcePattern::forExactNamespace(fromNs), ActionType::find));
-    } else if (stageName == "$facet" && stageSpec.firstElementType() == BSONType::Object &&
-               !haveRecursed) {
-        // Add privileges of sub-stages, but only if we haven't recursed already. We don't want to
-        // get a stack overflow while checking privileges. If we ever allow a $facet stage inside of
-        // a $facet stage, this code will have to be modified to avoid causing a stack overflow, but
-        // still check all required privileges of nested stages.
-        for (auto&& subPipeline : stageSpec.firstElement().embeddedObject()) {
-            if (subPipeline.type() == BSONType::Array) {
-                for (auto&& subPipeStageSpec : subPipeline.embeddedObject()) {
-                    addPrivilegesForStage(db,
-                                          cmdObj,
-                                          requiredPrivileges,
-                                          subPipeStageSpec.embeddedObjectUserCheck(),
-                                          true);
-                }
-            }
-        }
-    }
-}
-
-}  // namespace
-
-Status Pipeline::checkAuthForCommand(Client* client, const std::string& db, const BSONObj& cmdObj) {
-    NamespaceString inputNs(db, cmdObj.firstElement().str());
-    auto inputResource = ResourcePattern::forExactNamespace(inputNs);
-    uassert(17138,
-            mongoutils::str::stream() << "Invalid input namespace, " << inputNs.ns(),
-            inputNs.isValid());
-
-    PrivilegeVector privileges;
-
-    if (dps::extractElementAtPath(cmdObj, "pipeline.0.$indexStats")) {
-        Privilege::addPrivilegeToPrivilegeVector(
-            &privileges,
-            Privilege(ResourcePattern::forAnyNormalResource(), ActionType::indexStats));
-    } else if (dps::extractElementAtPath(cmdObj, "pipeline.0.$collStats")) {
-        Privilege::addPrivilegeToPrivilegeVector(&privileges,
-                                                 Privilege(inputResource, ActionType::collStats));
-    } else {
-        // If no source requiring an alternative permission scheme is specified then default to
-        // requiring find() privileges on the given namespace.
-        Privilege::addPrivilegeToPrivilegeVector(&privileges,
-                                                 Privilege(inputResource, ActionType::find));
-    }
-
-    BSONObj pipeline = cmdObj.getObjectField("pipeline");
-    for (auto&& stageElem : pipeline) {
-        addPrivilegesForStage(db, cmdObj, &privileges, stageElem.embeddedObjectUserCheck());
-    }
-
-    if (AuthorizationSession::get(client)->isAuthorizedForPrivileges(privileges))
-        return Status::OK();
-    return Status(ErrorCodes::Unauthorized, "unauthorized");
-}
-
 bool Pipeline::aggSupportsWriteConcern(const BSONObj& cmd) {
     if (cmd.hasField("pipeline") == false) {
         return false;
diff --git a/src/mongo/db/pipeline/pipeline.h b/src/mongo/db/pipeline/pipeline.h
index 2b4cb3faacd..b75d1ac4c04 100644
--- a/src/mongo/db/pipeline/pipeline.h
+++ b/src/mongo/db/pipeline/pipeline.h
@@ -42,11 +42,10 @@
 namespace mongo {
 class BSONObj;
 class BSONObjBuilder;
-class Client;
 class CollatorInterface;
 class DocumentSource;
-struct ExpressionContext;
 class OperationContext;
+struct ExpressionContext;
 
 /**
  * A Pipeline object represents a list of DocumentSources and is responsible for optimizing the
@@ -75,13 +74,6 @@ public:
         SourceContainer sources, const boost::intrusive_ptr<ExpressionContext>& expCtx);
 
     /**
-     * Helper to implement Command::checkAuthForCommand.
-     */
-    static Status checkAuthForCommand(Client* client,
-                                      const std::string& dbname,
-                                      const BSONObj& cmdObj);
-
-    /**
      * Returns true if the provided aggregation command has a $out stage.
      */
     static bool aggSupportsWriteConcern(const BSONObj& cmd);
diff --git a/src/mongo/s/commands/cluster_pipeline_cmd.cpp b/src/mongo/s/commands/cluster_pipeline_cmd.cpp
index 99420f0ce57..8d14feaceec 100644
--- a/src/mongo/s/commands/cluster_pipeline_cmd.cpp
+++ b/src/mongo/s/commands/cluster_pipeline_cmd.cpp
@@ -37,6 +37,7 @@
 #include <vector>
 
 #include "mongo/base/status.h"
+#include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/client.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/pipeline/aggregation_request.h"
@@ -99,7 +100,8 @@ public:
     Status checkAuthForCommand(Client* client,
                                const std::string& dbname,
                                const BSONObj& cmdObj) final {
-        return Pipeline::checkAuthForCommand(client, dbname, cmdObj);
+        NamespaceString nss(parseNs(dbname, cmdObj));
+        return AuthorizationSession::get(client)->checkAuthForAggregate(nss, cmdObj);
     }
 
     virtual bool run(OperationContext* txn,
diff --git a/src/mongo/s/commands/commands_public.cpp b/src/mongo/s/commands/commands_public.cpp
index cff2718e5c4..4610b4cca67 100644
--- a/src/mongo/s/commands/commands_public.cpp
+++ b/src/mongo/s/commands/commands_public.cpp
@@ -415,12 +415,12 @@ public:
 class CollectionModCmd : public AllShardsCollectionCommand {
 public:
     CollectionModCmd() : AllShardsCollectionCommand("collMod") {}
-    virtual void addRequiredPrivileges(const std::string& dbname,
-                                       const BSONObj& cmdObj,
-                                       std::vector<Privilege>* out) {
-        ActionSet actions;
-        actions.addAction(ActionType::collMod);
-        out->push_back(Privilege(parseResourcePattern(dbname, cmdObj), actions));
+
+    virtual Status checkAuthForCommand(Client* client,
+                                       const std::string& dbname,
+                                       const BSONObj& cmdObj) {
+        NamespaceString nss(parseNs(dbname, cmdObj));
+        return AuthorizationSession::get(client)->checkAuthForCollMod(nss, cmdObj);
     }
 
     virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
@@ -505,23 +505,8 @@ public:
     virtual Status checkAuthForCommand(Client* client,
                                        const std::string& dbname,
                                        const BSONObj& cmdObj) {
-        AuthorizationSession* authzSession = AuthorizationSession::get(client);
-        if (cmdObj["capped"].trueValue()) {
-            if (!authzSession->isAuthorizedForActionsOnResource(
-                    parseResourcePattern(dbname, cmdObj), ActionType::convertToCapped)) {
-                return Status(ErrorCodes::Unauthorized, "unauthorized");
-            }
-        }
-
-        // ActionType::createCollection or ActionType::insert are both acceptable
-        if (authzSession->isAuthorizedForActionsOnResource(parseResourcePattern(dbname, cmdObj),
-                                                           ActionType::createCollection) ||
-            authzSession->isAuthorizedForActionsOnResource(parseResourcePattern(dbname, cmdObj),
-                                                           ActionType::insert)) {
-            return Status::OK();
-        }
-
-        return Status(ErrorCodes::Unauthorized, "unauthorized");
+        NamespaceString nss(parseNs(dbname, cmdObj));
+        return AuthorizationSession::get(client)->checkAuthForCreate(nss, cmdObj);
     }
     virtual bool supportsWriteConcern(const BSONObj& cmd) const override {
         return true;