SERVER-28260 Check coauth for killCursors and add killAnyCursors

(cherry picked from commit d75b113186e1914a5f2dc6d1983d604324a8f7f1)

16 files changed, 383 insertions, 129 deletions

diff --git a/jstests/auth/basic_role_auth.js b/jstests/auth/basic_role_auth.js
index a2484565755..fc4e39089ba 100644
--- a/jstests/auth/basic_role_auth.js
+++ b/jstests/auth/basic_role_auth.js
@@ -41,7 +41,8 @@ var READ_WRITE_PERM =
     {insert: 1, update: 1, remove: 1, query: 1, index_r: 1, index_w: 1, killCursor: 1};
 var ADMIN_PERM = {index_r: 1, index_w: 1, profile_r: 1};
 var UADMIN_PERM = {user_r: 1, user_w: 1};
-var CLUSTER_PERM = {killOp: 1, currentOp: 1, fsync_unlock: 1, killCursor: 1, profile_r: 1};
+var CLUSTER_PERM =
+    {killOp: 1, currentOp: 1, fsync_unlock: 1, killCursor: 1, killAnyCursor: 1, profile_r: 1};
 
 /**
  * Checks whether an error occurs after running an operation.
@@ -170,7 +171,29 @@ var testOps = function(db, allowedActions) {
         assert(!bsonBinaryEqual({cursorId: cursorId}, {cursorId: NumberLong(0)}),
                "find command didn't return a cursor: " + tojson(cmdRes));
 
-        checkErr(allowedActions.hasOwnProperty('killCursor'), function() {
+        const shouldSucceed = (function() {
+            // admin users can do anything they want.
+            if (allowedActions.hasOwnProperty('killAnyCursor')) {
+                return true;
+            }
+
+            // users can kill their own cursors
+            const users = assert.commandWorked(db.runCommand({connectionStatus: 1}))
+                              .authInfo.authenticatedUsers;
+            const users2 = assert.commandWorked(db2.runCommand({connectionStatus: 1}))
+                               .authInfo.authenticatedUsers;
+            if (!users.length && !users2.length) {
+                // Special case, no-auth
+                return true;
+            }
+            return users.some(function(u) {
+                return users2.some(function(u2) {
+                    return ((u.db === u2.db) && (u.user === u2.user));
+                });
+            });
+        })();
+
+        checkErr(shouldSucceed, function() {
             // Issue killCursor command from db.
             cmdRes = db.runCommand({killCursors: db2.kill_cursor.getName(), cursors: [cursorId]});
             assert.commandWorked(cmdRes);
diff --git a/jstests/auth/kill_cursors.js b/jstests/auth/kill_cursors.js
new file mode 100644
index 00000000000..89523927235
--- /dev/null
+++ b/jstests/auth/kill_cursors.js
@@ -0,0 +1,145 @@
+// Test the killCursors command.
+(function() {
+    'use strict';
+
+    function runTest(mongod) {
+        /**
+         * Open a cursor on `db` while authenticated as `authUsers`.
+         * Then logout, and log back in as `killUsers` and try to kill that cursor.
+         *
+         * @param db - The db to create a cursor on and ultimately kill agains.
+         * @param authUsers - Array of ['username', db] pairs to create the cursor under.
+         * @param killUsers - Array of ['username', dn] pairs to use when killing.
+         * @param shouldWork - Whether we expect success
+         */
+        function tryKill(db, authUsers, killUsers, shouldWork) {
+            function loginAll(users) {
+                users.forEach(function(u) {
+                    assert(u[1].auth(u[0], 'pass'));
+                });
+            }
+
+            function logoutAll() {
+                [testA, testB].forEach(function(d) {
+                    const users = assert.commandWorked(d.runCommand({connectionStatus: 1}))
+                                      .authInfo.authenticatedUsers;
+                    users.forEach(function(u) {
+                        mongod.getDB(u.db).logout();
+                    });
+                });
+            }
+
+            // Create a cursor to be killed later.
+            loginAll(authUsers);
+            const id = assert.commandWorked(db.runCommand({find: db.coll.getName(), batchSize: 2}))
+                           .cursor.id;
+            assert.neq(id, 0, "Invalid cursor ID");
+            logoutAll();
+
+            loginAll(killUsers);
+            const killCmd = db.runCommand({killCursors: db.coll.getName(), cursors: [id]});
+            logoutAll();
+            if (shouldWork) {
+                assert.commandWorked(killCmd, "Unable to kill cursor");
+            } else {
+                assert.commandFailed(killCmd, "Should not have been able to kill cursor");
+            }
+        }
+
+        /**
+         * Create user1/user2 in testA, and user3/user4 in testB.
+         * Create two 101 element collections in testA and testB.
+         * Use various combinations of those users to open cursors,
+         * then (potentially) different combinations of users to kill them.
+         *
+         * A cursor should only be killable if at least one of the users
+         * who created it is trying to kill it.
+         */
+
+        const testA = mongod.getDB('testA');
+        const testB = mongod.getDB('testB');
+        const admin = mongod.getDB('admin');
+
+        // Setup users
+        admin.createUser({user: 'admin', pwd: 'pass', roles: jsTest.adminUserRoles});
+        assert(admin.auth('admin', 'pass'));
+
+        testA.createUser({user: 'user1', pwd: 'pass', roles: jsTest.basicUserRoles});
+        testA.createUser({user: 'user2', pwd: 'pass', roles: jsTest.basicUserRoles});
+        testB.createUser({user: 'user3', pwd: 'pass', roles: jsTest.basicUserRoles});
+        testB.createUser({user: 'user4', pwd: 'pass', roles: jsTest.basicUserRoles});
+        admin.logout();
+
+        // Create a collection with batchable data
+        assert(testA.auth('user1', 'pass'));
+        assert(testB.auth('user3', 'pass'));
+        for (var i = 0; i < 101; ++i) {
+            assert.writeOK(testA.coll.insert({_id: i}));
+            assert.writeOK(testB.coll.insert({_id: i}));
+        }
+        testA.logout();
+        testB.logout();
+
+        // A user can kill their own cursor.
+        tryKill(testA, [['user1', testA]], [['user1', testA]], true);
+        tryKill(testA, [['user2', testA]], [['user2', testA]], true);
+        tryKill(testB, [['user3', testB]], [['user3', testB]], true);
+        tryKill(testB, [['user4', testB]], [['user4', testB]], true);
+
+        // A user cannot kill someone else's cursor.
+        tryKill(testA, [['user1', testA]], [['user2', testA]], false);
+        tryKill(testA, [['user1', testA]], [['user2', testA], ['user3', testB]], false);
+        tryKill(testA, [['user2', testA]], [['user1', testA]], false);
+        tryKill(testA, [['user2', testA]], [['user1', testA], ['user3', testB]], false);
+        tryKill(testB, [['user3', testB]], [['user1', testA], ['user4', testB]], false);
+        tryKill(testB, [['user3', testB]], [['user2', testA], ['user4', testB]], false);
+
+        // A multi-owned cursor can be killed by any/all owner.
+        tryKill(testA, [['user1', testA], ['user3', testB]], [['user1', testA]], true);
+        tryKill(testB, [['user1', testA], ['user3', testB]], [['user3', testB]], true);
+        tryKill(testA,
+                [['user1', testA], ['user3', testB]],
+                [['user1', testA], ['user3', testB]],
+                true);
+        tryKill(testA,
+                [['user1', testA], ['user3', testB]],
+                [['user2', testA], ['user3', testB]],
+                true);
+        tryKill(testB,
+                [['user1', testA], ['user3', testB]],
+                [['user1', testA], ['user3', testB]],
+                true);
+        tryKill(testB,
+                [['user1', testA], ['user3', testB]],
+                [['user1', testA], ['user4', testB]],
+                true);
+
+        // An owned cursor can not be killed by other user(s).
+        tryKill(testA,
+                [['user1', testA], ['user3', testB]],
+                [['user2', testA], ['user4', testB]],
+                false);
+        tryKill(testA, [['user1', testA]], [['user2', testA], ['user3', testB]], false);
+        tryKill(testA,
+                [['user1', testA], ['user3', testB]],
+                [['user2', testA], ['user4', testB]],
+                false);
+
+        // Admin can kill anything.
+        tryKill(testA, [['user1', testA]], [['admin', admin]], true);
+        tryKill(testA, [['user2', testA]], [['admin', admin]], true);
+        tryKill(testB, [['user3', testB]], [['admin', admin]], true);
+        tryKill(testB, [['user4', testB]], [['admin', admin]], true);
+        tryKill(testA, [['user1', testA], ['user3', testB]], [['admin', admin]], true);
+        tryKill(testB, [['user2', testA], ['user4', testB]], [['admin', admin]], true);
+    }
+
+    const mongod = MongoRunner.runMongod({auth: ""});
+    runTest(mongod);
+    MongoRunner.stopMongod(mongod);
+
+    const st =
+        new ShardingTest({shards: 1, mongos: 1, config: 1, other: {keyFile: 'jstests/libs/key1'}});
+    runTest(st.s0);
+    st.stop();
+})();
diff --git a/jstests/auth/lib/commands_lib.js b/jstests/auth/lib/commands_lib.js
index f06d8c980ac..092856a4170 100644
--- a/jstests/auth/lib/commands_lib.js
+++ b/jstests/auth/lib/commands_lib.js
@@ -3877,8 +3877,21 @@ var authCommandsLib = {
         },
         {
           testname: "killCursors",
-          command: {killCursors: "foo", cursors: [NumberLong("123")]},
-          skipSharded: true,  // TODO enable when killCursors command is implemented on mongos
+          setup: function(runOnDb) {
+              return runOnDb;
+          },
+          command: function(runOnDb) {
+              // Don't create the cursor during setup() because we're not logged in yet.
+              // Cursor must be created with same user who tries to kill it.
+              const cmd = runOnDb.runCommand({find: "foo", batchSize: 2});
+              if (cmd.ok === 1) {
+                  return {killCursors: "foo", cursors: [cmd.cursor.id]};
+              } else {
+                  // If we can't even execute a find, then we certainly can't kill it.
+                  // Let it fail/unauthorized via the find command
+                  return {find: "foo", batchSize: 2};
+              }
+          },
           testcases: [
               {
                 runOnDb: firstDbName,
@@ -3888,29 +3901,18 @@ var authCommandsLib = {
                     readWrite: 1,
                     readWriteAnyDatabase: 1,
                     dbOwner: 1,
-                    hostManager: 1,
-                    clusterAdmin: 1,
                     backup: 1,
                     root: 1,
                     __system: 1
                 },
-                privileges:
-                    [{resource: {db: firstDbName, collection: "foo"}, actions: ["killCursors"]}],
+                privileges: [{resource: {db: firstDbName, collection: "foo"}, actions: ["find"]}],
                 expectFail: true
               },
               {
                 runOnDb: secondDbName,
-                roles: {
-                    readAnyDatabase: 1,
-                    readWriteAnyDatabase: 1,
-                    hostManager: 1,
-                    clusterAdmin: 1,
-                    backup: 1,
-                    root: 1,
-                    __system: 1
-                },
-                privileges:
-                    [{resource: {db: secondDbName, collection: "foo"}, actions: ["killCursors"]}],
+                roles:
+                    {readAnyDatabase: 1, readWriteAnyDatabase: 1, backup: 1, root: 1, __system: 1},
+                privileges: [{resource: {db: secondDbName, collection: "foo"}, actions: ["find"]}],
                 expectFail: true
               }
           ]
diff --git a/jstests/core/kill_cursors.js b/jstests/core/kill_cursors.js
index fc0e8b89361..e825c6bd8bf 100644
--- a/jstests/core/kill_cursors.js
+++ b/jstests/core/kill_cursors.js
@@ -106,23 +106,23 @@
         // Currently, pinned cursors that are targeted by a killCursors operation are kept alive on
         // mongod but are killed on mongos (see SERVER-21710).
         cmdRes = db.runCommand({killCursors: coll.getName(), cursors: [NumberLong(123), cursorId]});
-        assert.commandWorked(cmdRes);
-        assert.eq(cmdRes.cursorsNotFound, [NumberLong(123)]);
-        assert.eq(cmdRes.cursorsUnknown, []);
-
-        if (isMongos) {
-            assert.eq(cmdRes.cursorsKilled, [cursorId]);
-            assert.eq(cmdRes.cursorsAlive, []);
-        } else {
-            // If the cursor has already been pinned it will be left alive; otherwise it will be
-            // killed.
-            if (cmdRes.cursorsAlive.length === 1) {
-                assert.eq(cmdRes.cursorsKilled, []);
-                assert.eq(cmdRes.cursorsAlive, [cursorId]);
-            } else {
+        if (cmdRes.ok) {
+            // Cursor wasn't pinned.
+            // If auth is enabled, kill will fail due to 123 not existing (caught in auchCheck).
+            // If auth is not enabled, we'll get past the authCheck, and kill of existing cursor
+            // can succeed for the real cursor.
+            if (cmdRes.cursorsKilled.length) {
                 assert.eq(cmdRes.cursorsKilled, [cursorId]);
                 assert.eq(cmdRes.cursorsAlive, []);
+            } else {
+                assert.eq(cmdRes.cursorsKilled, []);
+                assert.eq(cmdRes.cursorsAlive, [cursorId]);
             }
+            assert.eq(cmdRes.cursorsNotFound, [NumberLong(123)]);
+            assert.eq(cmdRes.cursorsUnknown, []);
+        } else {
+            // Pin released before we got to it.
+            assert.eq(cmdRes.code, ErrorCodes.CursorInUse);
         }
     } finally {
         assert.commandWorked(db.adminCommand({configureFailPoint: failpointName, mode: "off"}));
diff --git a/jstests/core/operation_latency_histogram.js b/jstests/core/operation_latency_histogram.js
index 0e616b8c313..64d6bd6a33a 100644
--- a/jstests/core/operation_latency_histogram.js
+++ b/jstests/core/operation_latency_histogram.js
@@ -67,7 +67,15 @@
     for (var i = 0; i < numRecords - 1; i++) {
         cursors[i].close();
     }
-    lastHistogram = assertHistogramDiffEq(testColl, lastHistogram, 0, 0, numRecords - 1);
+    try {
+        // Each close may result in two commands in latencyStats due to separate
+        // pinning during auth check and execution.
+        lastHistogram = assertHistogramDiffEq(testColl, lastHistogram, 0, 0, 2 * (numRecords - 1));
+    } catch (e) {
+        // Increment last reads to account for extra getstats call
+        ++lastHistogram.reads.ops;
+        lastHistogram = assertHistogramDiffEq(testColl, lastHistogram, 0, 0, numRecords - 1);
+    }
 
     // Remove
     for (var i = 0; i < numRecords; i++) {
diff --git a/src/mongo/db/auth/action_types.txt b/src/mongo/db/auth/action_types.txt
index 77a58a053d1..e3412549c15 100644
--- a/src/mongo/db/auth/action_types.txt
+++ b/src/mongo/db/auth/action_types.txt
@@ -65,6 +65,7 @@
 "insert",
 "internal", # Special action type that represents internal actions
 "invalidateUserCache",
+"killAnyCursor",
 "killAnySession",
 "killCursors",
 "killop",
diff --git a/src/mongo/db/auth/authorization_session.cpp b/src/mongo/db/auth/authorization_session.cpp
index 15ee3480a88..8d990765552 100644
--- a/src/mongo/db/auth/authorization_session.cpp
+++ b/src/mongo/db/auth/authorization_session.cpp
@@ -466,35 +466,57 @@ Status AuthorizationSession::checkAuthForDelete(OperationContext* opCtx,
 }
 
 Status AuthorizationSession::checkAuthForKillCursors(const NamespaceString& ns,
-                                                     long long cursorID) {
-    // See implementation comments in checkAuthForGetMore().  This method looks very similar.
+                                                     UserNameIterator cursorOwner) {
+    if (isAuthorizedForActionsOnResource(ResourcePattern::forClusterResource(),
+                                         ActionType::killAnyCursor)) {
+        return Status::OK();
+    }
 
-    // SERVER-20364 Check for find or killCursor privileges until we have a way of associating
-    // a cursor with an owner.
     if (ns.isListCollectionsCursorNS()) {
-        if (!(isAuthorizedForActionsOnResource(ResourcePattern::forDatabaseName(ns.db()),
-                                               ActionType::killCursors) ||
-              isAuthorizedToListCollections(ns.db()))) {
-            return Status(ErrorCodes::Unauthorized,
-                          str::stream() << "not authorized to kill listCollections cursor on "
-                                        << ns.ns());
+        // listCollections: Target the database being enumerated.
+        if (isAuthorizedForActionsOnResource(ResourcePattern::forDatabaseName(ns.db()),
+                                             ActionType::killAnyCursor)) {
+            return Status::OK();
+        }
+        const bool canKill =
+            isAuthorizedForActionsOnResource(ResourcePattern::forDatabaseName(ns.db()),
+                                             ActionType::killCursors) ||
+            isAuthorizedToListCollections(ns.db());
+        if (canKill && isCoauthorizedWith(cursorOwner)) {
+            return Status::OK();
         }
+        return Status(ErrorCodes::Unauthorized,
+                      str::stream() << "not authorized to kill listCollections cursor on "
+                                    << ns.ns());
     } else if (ns.isListIndexesCursorNS()) {
+
+        // listIndexes: Target the underlying collection.
         NamespaceString targetNS = ns.getTargetNSForListIndexes();
-        if (!(isAuthorizedForActionsOnNamespace(targetNS, ActionType::killCursors) ||
-              isAuthorizedForActionsOnNamespace(targetNS, ActionType::listIndexes))) {
-            return Status(ErrorCodes::Unauthorized,
-                          str::stream() << "not authorized to kill listIndexes cursor on "
-                                        << ns.ns());
+        if (isAuthorizedForActionsOnNamespace(targetNS, ActionType::killAnyCursor)) {
+            return Status::OK();
+        }
+        const bool canKill = isAuthorizedForActionsOnNamespace(targetNS, ActionType::killCursors) ||
+            isAuthorizedForActionsOnNamespace(targetNS, ActionType::listIndexes);
+        if (canKill && isCoauthorizedWith(cursorOwner)) {
+            return Status::OK();
         }
+
+        return Status(ErrorCodes::Unauthorized,
+                      str::stream() << "not authorized to kill listIndexes cursor on " << ns.ns());
     } else {
-        if (!(isAuthorizedForActionsOnNamespace(ns, ActionType::killCursors) ||
-              isAuthorizedForActionsOnNamespace(ns, ActionType::find))) {
-            return Status(ErrorCodes::Unauthorized,
-                          str::stream() << "not authorized to kill cursor on " << ns.ns());
+
+        // Otherwise: Target the collection as named.
+        if (isAuthorizedForActionsOnNamespace(ns, ActionType::killAnyCursor)) {
+            return Status::OK();
         }
+        const bool canKill = isAuthorizedForActionsOnNamespace(ns, ActionType::killCursors) ||
+            isAuthorizedForActionsOnNamespace(ns, ActionType::find);
+        if (canKill && isCoauthorizedWith(cursorOwner)) {
+            return Status::OK();
+        }
+        return Status(ErrorCodes::Unauthorized,
+                      str::stream() << "not authorized to kill cursor on " << ns.ns());
     }
-    return Status::OK();
 }
 
 Status AuthorizationSession::checkAuthForCreate(const NamespaceString& ns,
diff --git a/src/mongo/db/auth/authorization_session.h b/src/mongo/db/auth/authorization_session.h
index a1e159c80de..81cfdc466d0 100644
--- a/src/mongo/db/auth/authorization_session.h
+++ b/src/mongo/db/auth/authorization_session.h
@@ -202,7 +202,7 @@ public:
     // Checks if this connection has the privileges necessary to perform a killCursor on
     // the identified cursor, supposing that cursor is associated with the supplied namespace
     // identifier.
-    Status checkAuthForKillCursors(const NamespaceString& ns, long long cursorID);
+    Status checkAuthForKillCursors(const NamespaceString& cursorNss, UserNameIterator cursorOwner);
 
     // Checks if this connection has the privileges necessary to run the aggregation pipeline
     // specified in 'cmdObj' on the namespace 'ns' either directly on mongoD or via mongoS.
diff --git a/src/mongo/db/auth/role_graph_builtin_roles.cpp b/src/mongo/db/auth/role_graph_builtin_roles.cpp
index 892d30d92c7..ce467c7bc5f 100644
--- a/src/mongo/db/auth/role_graph_builtin_roles.cpp
+++ b/src/mongo/db/auth/role_graph_builtin_roles.cpp
@@ -221,6 +221,7 @@ MONGO_INITIALIZER(AuthorizationBuiltinRoles)(InitializerContext* context) {
         << ActionType::flushRouterConfig  // clusterManager gets this also
         << ActionType::fsync
         << ActionType::invalidateUserCache // userAdminAnyDatabase gets this also
+        << ActionType::killAnyCursor
         << ActionType::killAnySession
         << ActionType::killop
         << ActionType::replSetResizeOplog
diff --git a/src/mongo/db/commands/killcursors_cmd.cpp b/src/mongo/db/commands/killcursors_cmd.cpp
index 7e2aaa91cbc..0810d02797d 100644
--- a/src/mongo/db/commands/killcursors_cmd.cpp
+++ b/src/mongo/db/commands/killcursors_cmd.cpp
@@ -29,6 +29,7 @@
 #include "mongo/platform/basic.h"
 
 #include "mongo/base/disallow_copying.h"
+#include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/commands/killcursors_common.h"
 #include "mongo/db/curop.h"
@@ -47,29 +48,27 @@ public:
     KillCursorsCmd() = default;
 
 private:
+    Status _checkAuth(Client* client, const NamespaceString& nss, CursorId id) const final {
+        auto opCtx = client->getOperationContext();
+        const auto check = [client, opCtx, id](CursorManager* manager) {
+            auto ccPin = manager->pinCursor(opCtx, id);
+            if (!ccPin.isOK()) {
+                return ccPin.getStatus();
+            }
+
+            const auto* cursor = ccPin.getValue().getCursor();
+            AuthorizationSession* as = AuthorizationSession::get(client);
+            return as->checkAuthForKillCursors(cursor->nss(), cursor->getAuthenticatedUsers());
+        };
+
+        return CursorManager::withCursorManager(opCtx, id, nss, check);
+    }
+
     Status _killCursor(OperationContext* opCtx,
                        const NamespaceString& nss,
-                       CursorId cursorId) final {
-        // Cursors come in one of two flavors:
-        // - Cursors owned by the collection cursor manager, such as those generated via the find
-        //   command. For these cursors, we hold the appropriate collection lock for the duration of
-        //   the getMore using AutoGetCollectionForRead. This will automatically update the CurOp
-        //   object appropriately and record execution time via Top upon completion.
-        // - Cursors owned by the global cursor manager, such as those generated via the aggregate
-        //   command. These cursors either hold no collection state or manage their collection state
-        //   internally, so we acquire no locks. In this case we use the AutoStatsTracker object to
-        //   update the CurOp object appropriately and record execution time via Top upon
-        //   completion.
-        //
-        // Thus, exactly one of 'readLock' and 'statsTracker' will be populated as we populate
-        // 'cursorManager'.
-        boost::optional<AutoGetCollectionForReadCommand> readLock;
+                       CursorId id) const final {
         boost::optional<AutoStatsTracker> statsTracker;
-        CursorManager* cursorManager;
-
-        if (CursorManager::isGloballyManagedCursor(cursorId)) {
-            cursorManager = CursorManager::getGlobalCursorManager();
-
+        if (CursorManager::isGloballyManagedCursor(id)) {
             if (auto nssForCurOp = nss.isGloballyManagedNamespace()
                     ? nss.getTargetNSForGloballyManagedNamespace()
                     : nss) {
@@ -77,33 +76,12 @@ private:
                 statsTracker.emplace(
                     opCtx, *nssForCurOp, Top::LockType::NotLocked, dbProfilingLevel);
             }
-
-            // Make sure the namespace of the cursor matches the namespace passed to the killCursors
-            // command so we can be sure we checked the correct privileges.
-            auto ccPin = cursorManager->pinCursor(opCtx, cursorId);
-            if (ccPin.isOK()) {
-                auto cursorNs = ccPin.getValue().getCursor()->nss();
-                if (cursorNs != nss) {
-                    return Status{ErrorCodes::Unauthorized,
-                                  str::stream() << "issued killCursors on namespace '" << nss.ns()
-                                                << "', but cursor with id "
-                                                << cursorId
-                                                << " belongs to a different namespace: "
-                                                << cursorNs.ns()};
-                }
-            }
-        } else {
-            readLock.emplace(opCtx, nss);
-            Collection* collection = readLock->getCollection();
-            if (!collection) {
-                return {ErrorCodes::CursorNotFound,
-                        str::stream() << "collection does not exist: " << nss.ns()};
-            }
-            cursorManager = collection->getCursorManager();
         }
-        invariant(cursorManager);
 
-        return cursorManager->eraseCursor(opCtx, cursorId, true /*shouldAudit*/);
+        return CursorManager::withCursorManager(
+            opCtx, id, nss, [opCtx, id](CursorManager* manager) {
+                return manager->eraseCursor(opCtx, id, true /* shouldAudit */);
+            });
     }
 } killCursorsCmd;
 
diff --git a/src/mongo/db/commands/killcursors_common.cpp b/src/mongo/db/commands/killcursors_common.cpp
index 0dcc262f56f..39edb0a7c6f 100644
--- a/src/mongo/db/commands/killcursors_common.cpp
+++ b/src/mongo/db/commands/killcursors_common.cpp
@@ -42,21 +42,23 @@ namespace mongo {
 Status KillCursorsCmdBase::checkAuthForCommand(Client* client,
                                                const std::string& dbname,
                                                const BSONObj& cmdObj) {
-    auto statusWithRequest = KillCursorsRequest::parseFromBSON(dbname, cmdObj);
+    const auto statusWithRequest = KillCursorsRequest::parseFromBSON(dbname, cmdObj);
     if (!statusWithRequest.isOK()) {
         return statusWithRequest.getStatus();
     }
-    auto killCursorsRequest = std::move(statusWithRequest.getValue());
-
-    AuthorizationSession* as = AuthorizationSession::get(client);
+    const auto killCursorsRequest = statusWithRequest.getValue();
 
+    const auto& nss = killCursorsRequest.nss;
     for (CursorId id : killCursorsRequest.cursorIds) {
-        Status authorizationStatus = as->checkAuthForKillCursors(killCursorsRequest.nss, id);
-
-        if (!authorizationStatus.isOK()) {
-            audit::logKillCursorsAuthzCheck(
-                client, killCursorsRequest.nss, id, ErrorCodes::Unauthorized);
-            return authorizationStatus;
+        const auto status = _checkAuth(client, nss, id);
+        if (!status.isOK()) {
+            if (status.code() == ErrorCodes::CursorNotFound) {
+                // Not found isn't an authorization issue.
+                // run() will raise it as a return value.
+                continue;
+            }
+            audit::logKillCursorsAuthzCheck(client, nss, id, status.code());
+            return status;
         }
     }
 
diff --git a/src/mongo/db/commands/killcursors_common.h b/src/mongo/db/commands/killcursors_common.h
index 43a0443ff80..e34bd5098da 100644
--- a/src/mongo/db/commands/killcursors_common.h
+++ b/src/mongo/db/commands/killcursors_common.h
@@ -26,8 +26,6 @@
  *    it in the license file.
  */
 
-#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kQuery
-
 #include "mongo/db/commands.h"
 #include "mongo/db/cursor_id.h"
 
@@ -78,6 +76,13 @@ public:
 
 private:
     /**
+     * Verify the cursor exists, is unpinned, and can be killed by the current user(s).
+     */
+    virtual Status _checkAuth(Client* client,
+                              const NamespaceString& nss,
+                              CursorId cursorId) const = 0;
+
+    /**
      * Kill the cursor with id 'cursorId' in namespace 'nss'. Use 'opCtx' if necessary.
      *
      * Returns Status::OK() if the cursor was killed, or ErrorCodes::CursorNotFound if there is no
@@ -85,7 +90,7 @@ private:
      */
     virtual Status _killCursor(OperationContext* opCtx,
                                const NamespaceString& nss,
-                               CursorId cursorId) = 0;
+                               CursorId cursorId) const = 0;
 };
 
 }  // namespace mongo
diff --git a/src/mongo/db/cursor_manager.cpp b/src/mongo/db/cursor_manager.cpp
index 0b85ccee4ed..a85830cfaa0 100644
--- a/src/mongo/db/cursor_manager.cpp
+++ b/src/mongo/db/cursor_manager.cpp
@@ -196,10 +196,18 @@ bool GlobalCursorIdCache::eraseCursor(OperationContext* opCtx, CursorId id, bool
 
     // Check if we are authorized to erase this cursor.
     if (checkAuth) {
-        AuthorizationSession* as = AuthorizationSession::get(opCtx->getClient());
-        Status authorizationStatus = as->checkAuthForKillCursors(nss, id);
-        if (!authorizationStatus.isOK()) {
-            audit::logKillCursorsAuthzCheck(opCtx->getClient(), nss, id, ErrorCodes::Unauthorized);
+        auto status = CursorManager::withCursorManager(
+            opCtx, id, nss, [nss, id, opCtx](CursorManager* manager) {
+                auto ccPin = manager->pinCursor(opCtx, id);
+                if (!ccPin.isOK()) {
+                    return ccPin.getStatus();
+                }
+                AuthorizationSession* as = AuthorizationSession::get(opCtx->getClient());
+                auto cursorOwner = ccPin.getValue().getCursor()->getAuthenticatedUsers();
+                return as->checkAuthForKillCursors(nss, cursorOwner);
+            });
+        if (!status.isOK()) {
+            audit::logKillCursorsAuthzCheck(opCtx->getClient(), nss, id, status.code());
             return false;
         }
     }
@@ -351,6 +359,28 @@ bool CursorManager::eraseCursorGlobal(OperationContext* opCtx, CursorId id) {
     return globalCursorIdCache->eraseCursor(opCtx, id, false);
 }
 
+Status CursorManager::withCursorManager(OperationContext* opCtx,
+                                        CursorId id,
+                                        const NamespaceString& nss,
+                                        stdx::function<Status(CursorManager*)> callback) {
+    boost::optional<AutoGetCollectionForReadCommand> readLock;
+    CursorManager* cursorManager = nullptr;
+
+    if (CursorManager::isGloballyManagedCursor(id)) {
+        cursorManager = CursorManager::getGlobalCursorManager();
+    } else {
+        readLock.emplace(opCtx, nss);
+        Collection* collection = readLock->getCollection();
+        if (!collection) {
+            return {ErrorCodes::CursorNotFound,
+                    str::stream() << "collection does not exist: " << nss.ns()};
+        }
+        cursorManager = collection->getCursorManager();
+    }
+    invariant(cursorManager);
+
+    return callback(cursorManager);
+}
 
 // --------------------------
 
@@ -504,7 +534,9 @@ StatusWith<ClientCursorPin> CursorManager::pinCursor(OperationContext* opCtx, Cu
     }
 
     ClientCursor* cursor = it->second;
-    uassert(12051, str::stream() << "cursor id " << id << " is already in use", !cursor->_isPinned);
+    uassert(ErrorCodes::CursorInUse,
+            str::stream() << "cursor id " << id << " is already in use",
+            !cursor->_isPinned);
     if (cursor->getExecutor()->isMarkedAsKilled()) {
         // This cursor was killed while it was idle.
         Status error{ErrorCodes::QueryPlanKilled,
diff --git a/src/mongo/db/cursor_manager.h b/src/mongo/db/cursor_manager.h
index 9683728adab..2736fd181b6 100644
--- a/src/mongo/db/cursor_manager.h
+++ b/src/mongo/db/cursor_manager.h
@@ -227,6 +227,15 @@ public:
      */
     static std::size_t timeoutCursorsGlobal(OperationContext* opCtx, Date_t now);
 
+    /**
+     * Locate the correct cursor manager for a given cursorId and execute the provided callback.
+     * Returns ErrorCodes::CursorNotFound if cursorId does not exist.
+     */
+    static Status withCursorManager(OperationContext* opCtx,
+                                    CursorId id,
+                                    const NamespaceString& nss,
+                                    stdx::function<Status(CursorManager*)> callback);
+
 private:
     static constexpr int kNumPartitions = 16;
     friend class ClientCursorPin;
diff --git a/src/mongo/s/commands/cluster_killcursors_cmd.cpp b/src/mongo/s/commands/cluster_killcursors_cmd.cpp
index e04dad6fc42..901feca75bb 100644
--- a/src/mongo/s/commands/cluster_killcursors_cmd.cpp
+++ b/src/mongo/s/commands/cluster_killcursors_cmd.cpp
@@ -28,6 +28,7 @@
 
 #include "mongo/platform/basic.h"
 
+#include "mongo/db/auth/authorization_session.h"
 #include "mongo/db/commands/killcursors_common.h"
 #include "mongo/s/grid.h"
 #include "mongo/s/query/cluster_cursor_manager.h"
@@ -40,9 +41,22 @@ public:
     ClusterKillCursorsCmd() = default;
 
 private:
+    Status _checkAuth(Client* client, const NamespaceString& nss, CursorId cursorId) const final {
+        auto* as = AuthorizationSession::get(client);
+        invariant(as);
+
+        auto* opCtx = client->getOperationContext();
+        auto ccPin = grid.getCursorManager()->checkOutCursor(nss, cursorId, opCtx);
+        if (!ccPin.isOK()) {
+            return ccPin.getStatus();
+        }
+
+        return as->checkAuthForKillCursors(nss, ccPin.getValue().getAuthenticatedUsers());
+    }
+
     Status _killCursor(OperationContext* opCtx,
                        const NamespaceString& nss,
-                       CursorId cursorId) final {
+                       CursorId cursorId) const final {
         return grid.getCursorManager()->killCursor(nss, cursorId);
     }
 } clusterKillCursorsCmd;
diff --git a/src/mongo/s/commands/strategy.cpp b/src/mongo/s/commands/strategy.cpp
index 148b10f61c4..9b05cad2a4d 100644
--- a/src/mongo/s/commands/strategy.cpp
+++ b/src/mongo/s/commands/strategy.cpp
@@ -547,16 +547,28 @@ void Strategy::killCursors(OperationContext* opCtx, DbMessage* dbm) {
             continue;
         }
 
-        Status authorizationStatus = authSession->checkAuthForKillCursors(*nss, cursorId);
-        audit::logKillCursorsAuthzCheck(client,
-                                        *nss,
-                                        cursorId,
-                                        authorizationStatus.isOK() ? ErrorCodes::OK
-                                                                   : ErrorCodes::Unauthorized);
-        if (!authorizationStatus.isOK()) {
-            LOG(3) << "Not authorized to kill cursor.  Namespace: '" << *nss
-                   << "', cursor id: " << cursorId << ".";
-            continue;
+        {
+            // Block scope ccPin so that it releases our checked out cursor
+            // prior to the killCursor invocation below.
+            auto ccPin = manager->checkOutCursor(*nss, cursorId, opCtx);
+            if (!ccPin.isOK()) {
+                LOG(3) << "Unable to check out cursor for killCursor.  Namespace: '" << *nss
+                       << "', cursor id: " << cursorId << ".";
+                continue;
+            }
+            auto cursorOwners = ccPin.getValue().getAuthenticatedUsers();
+            auto authorizationStatus = authSession->checkAuthForKillCursors(*nss, cursorOwners);
+
+            audit::logKillCursorsAuthzCheck(client,
+                                            *nss,
+                                            cursorId,
+                                            authorizationStatus.isOK() ? ErrorCodes::OK
+                                                                       : ErrorCodes::Unauthorized);
+            if (!authorizationStatus.isOK()) {
+                LOG(3) << "Not authorized to kill cursor.  Namespace: '" << *nss
+                       << "', cursor id: " << cursorId << ".";
+                continue;
+            }
         }
 
         Status killCursorStatus = manager->killCursor(*nss, cursorId);