SERVER-32395 Make killCursors work against pinned cursors on mongos when auth is enabled

17 files changed, 427 insertions, 288 deletions

diff --git a/src/mongo/db/auth/user_name.h b/src/mongo/db/auth/user_name.h
index 260604d6a18..f55d3500dfb 100644
--- a/src/mongo/db/auth/user_name.h
+++ b/src/mongo/db/auth/user_name.h
@@ -192,4 +192,13 @@ UserNameIterator makeUserNameIteratorForContainer(const Container& container) {
     return makeUserNameIterator(container.begin(), container.end());
 }
 
+template <typename Container>
+Container userNameIteratorToContainer(UserNameIterator it) {
+    Container container;
+    while (it.more()) {
+        container.emplace_back(it.next());
+    }
+    return container;
+}
+
 }  // namespace mongo
diff --git a/src/mongo/db/repl/collection_cloner.cpp b/src/mongo/db/repl/collection_cloner.cpp
index eec47e5e02a..159a664ccb8 100644
--- a/src/mongo/db/repl/collection_cloner.cpp
+++ b/src/mongo/db/repl/collection_cloner.cpp
@@ -620,10 +620,7 @@ void CollectionCloner::_establishCollectionCursorsCallback(const RemoteCommandCa
             ShardId("CollectionClonerSyncSource"), _source, std::move(cursorResponse));
     }
 
-    // An empty list of authenticated users is passed into the cluster parameters
-    // as user information is not used in the ARM in context of collection cloning.
-    _clusterClientCursorParams =
-        stdx::make_unique<ClusterClientCursorParams>(_sourceNss, UserNameIterator());
+    _clusterClientCursorParams = stdx::make_unique<ClusterClientCursorParams>(_sourceNss);
     _clusterClientCursorParams->remotes = std::move(remoteCursors);
     if (_collectionCloningBatchSize > 0)
         _clusterClientCursorParams->batchSize = _collectionCloningBatchSize;
diff --git a/src/mongo/s/commands/cluster_aggregate.cpp b/src/mongo/s/commands/cluster_aggregate.cpp
index f187d95d677..154108b279a 100644
--- a/src/mongo/s/commands/cluster_aggregate.cpp
+++ b/src/mongo/s/commands/cluster_aggregate.cpp
@@ -520,10 +520,7 @@ BSONObj establishMergingMongosCursor(OperationContext* opCtx,
                                      std::unique_ptr<Pipeline, PipelineDeleter> pipelineForMerging,
                                      std::vector<ClusterClientCursorParams::RemoteCursor> cursors) {
 
-    ClusterClientCursorParams params(
-        requestedNss,
-        AuthorizationSession::get(opCtx->getClient())->getAuthenticatedUserNames(),
-        ReadPreferenceSetting::get(opCtx));
+    ClusterClientCursorParams params(requestedNss, ReadPreferenceSetting::get(opCtx));
 
     params.tailableMode = pipelineForMerging->getContext()->tailableMode;
     params.mergePipeline = std::move(pipelineForMerging);
@@ -595,12 +592,14 @@ BSONObj establishMergingMongosCursor(OperationContext* opCtx,
     CursorId clusterCursorId = 0;
 
     if (cursorState == ClusterCursorManager::CursorState::NotExhausted) {
+        auto authUsers = AuthorizationSession::get(opCtx->getClient())->getAuthenticatedUserNames();
         clusterCursorId = uassertStatusOK(Grid::get(opCtx)->getCursorManager()->registerCursor(
             opCtx,
             ccc.releaseCursor(),
             requestedNss,
             ClusterCursorManager::CursorType::MultiTarget,
-            ClusterCursorManager::CursorLifetime::Mortal));
+            ClusterCursorManager::CursorLifetime::Mortal,
+            authUsers));
     }
 
     responseBuilder.done(clusterCursorId, requestedNss.ns());
diff --git a/src/mongo/s/commands/cluster_killcursors_cmd.cpp b/src/mongo/s/commands/cluster_killcursors_cmd.cpp
index 6f4bc9c296e..4dbf279b9e4 100644
--- a/src/mongo/s/commands/cluster_killcursors_cmd.cpp
+++ b/src/mongo/s/commands/cluster_killcursors_cmd.cpp
@@ -42,17 +42,12 @@ public:
 
 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, ClusterCursorManager::kNoCheckSession);
-        if (!ccPin.isOK()) {
-            return ccPin.getStatus();
-        }
-
-        return as->checkAuthForKillCursors(nss, ccPin.getValue().getAuthenticatedUsers());
+        auto authzSession = AuthorizationSession::get(client);
+        auto authChecker = [&authzSession, &nss](UserNameIterator userNames) -> Status {
+            return authzSession->checkAuthForKillCursors(nss, userNames);
+        };
+        return grid.getCursorManager()->checkAuthForKillCursors(
+            client->getOperationContext(), nss, cursorId, authChecker);
     }
 
     Status _killCursor(OperationContext* opCtx,
diff --git a/src/mongo/s/commands/strategy.cpp b/src/mongo/s/commands/strategy.cpp
index 519aab14d0e..d1fda2dcfeb 100644
--- a/src/mongo/s/commands/strategy.cpp
+++ b/src/mongo/s/commands/strategy.cpp
@@ -542,7 +542,6 @@ void Strategy::killCursors(OperationContext* opCtx, DbMessage* dbm) {
     ConstDataCursor cursors(dbm->getArray(numCursors));
 
     Client* const client = opCtx->getClient();
-    AuthorizationSession* const authSession = AuthorizationSession::get(client);
     ClusterCursorManager* const manager = Grid::get(opCtx)->getCursorManager();
 
     for (int i = 0; i < numCursors; ++i) {
@@ -554,28 +553,16 @@ void Strategy::killCursors(OperationContext* opCtx, DbMessage* dbm) {
             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;
-            }
+        auto authzSession = AuthorizationSession::get(client);
+        auto authChecker = [&authzSession, &nss](UserNameIterator userNames) -> Status {
+            return authzSession->checkAuthForKillCursors(*nss, userNames);
+        };
+        auto authzStatus = manager->checkAuthForKillCursors(opCtx, *nss, cursorId, authChecker);
+        audit::logKillCursorsAuthzCheck(client, *nss, cursorId, authzStatus.code());
+        if (!authzStatus.isOK()) {
+            LOG(3) << "Not authorized to kill cursor.  Namespace: '" << *nss
+                   << "', cursor id: " << cursorId << ".";
+            continue;
         }
 
         Status killCursorStatus = manager->killCursor(*nss, cursorId);
diff --git a/src/mongo/s/query/async_results_merger_test.cpp b/src/mongo/s/query/async_results_merger_test.cpp
index 68778c64e7f..ccbce1d5517 100644
--- a/src/mongo/s/query/async_results_merger_test.cpp
+++ b/src/mongo/s/query/async_results_merger_test.cpp
@@ -117,7 +117,7 @@ protected:
         boost::optional<BSONObj> findCmd = boost::none,
         boost::optional<long long> getMoreBatchSize = boost::none,
         ReadPreferenceSetting readPref = ReadPreferenceSetting(ReadPreference::PrimaryOnly)) {
-        _params = stdx::make_unique<ClusterClientCursorParams>(_nss, UserNameIterator(), readPref);
+        _params = stdx::make_unique<ClusterClientCursorParams>(_nss, readPref);
         _params->remotes = std::move(remoteCursors);
 
         if (findCmd) {
@@ -1522,8 +1522,7 @@ TEST_F(AsyncResultsMergerTest, GetMoreRequestIncludesMaxTimeMS) {
 }
 
 TEST_F(AsyncResultsMergerTest, SortedTailableCursorNotReadyIfOneOrMoreRemotesHasNoOplogTimestamp) {
-    auto params =
-        stdx::make_unique<ClusterClientCursorParams>(_nss, UserNameIterator(), boost::none);
+    auto params = stdx::make_unique<ClusterClientCursorParams>(_nss, boost::none);
     std::vector<ClusterClientCursorParams::RemoteCursor> cursors;
     cursors.emplace_back(kTestShardIds[0], kTestShardHosts[0], CursorResponse(_nss, 123, {}));
     cursors.emplace_back(kTestShardIds[1], kTestShardHosts[1], CursorResponse(_nss, 456, {}));
@@ -1587,8 +1586,7 @@ TEST_F(AsyncResultsMergerTest, SortedTailableCursorNotReadyIfOneOrMoreRemotesHas
 
 TEST_F(AsyncResultsMergerTest,
        SortedTailableCursorNotReadyIfOneOrMoreRemotesHasNullOplogTimestamp) {
-    auto params =
-        stdx::make_unique<ClusterClientCursorParams>(_nss, UserNameIterator(), boost::none);
+    auto params = stdx::make_unique<ClusterClientCursorParams>(_nss, boost::none);
     std::vector<ClusterClientCursorParams::RemoteCursor> cursors;
     cursors.emplace_back(
         kTestShardIds[0],
@@ -1636,8 +1634,7 @@ TEST_F(AsyncResultsMergerTest,
 }
 
 TEST_F(AsyncResultsMergerTest, SortedTailableCursorNotReadyIfOneRemoteHasLowerOplogTime) {
-    auto params =
-        stdx::make_unique<ClusterClientCursorParams>(_nss, UserNameIterator(), boost::none);
+    auto params = stdx::make_unique<ClusterClientCursorParams>(_nss, boost::none);
     std::vector<ClusterClientCursorParams::RemoteCursor> cursors;
     Timestamp tooLow = Timestamp(1, 2);
     cursors.emplace_back(
@@ -1671,8 +1668,7 @@ TEST_F(AsyncResultsMergerTest, SortedTailableCursorNotReadyIfOneRemoteHasLowerOp
 }
 
 TEST_F(AsyncResultsMergerTest, SortedTailableCursorNewShardOrderedAfterExisting) {
-    auto params =
-        stdx::make_unique<ClusterClientCursorParams>(_nss, UserNameIterator(), boost::none);
+    auto params = stdx::make_unique<ClusterClientCursorParams>(_nss, boost::none);
     std::vector<ClusterClientCursorParams::RemoteCursor> cursors;
     cursors.emplace_back(kTestShardIds[0], kTestShardHosts[0], CursorResponse(_nss, 123, {}));
     params->remotes = std::move(cursors);
@@ -1743,8 +1739,7 @@ TEST_F(AsyncResultsMergerTest, SortedTailableCursorNewShardOrderedAfterExisting)
 }
 
 TEST_F(AsyncResultsMergerTest, SortedTailableCursorNewShardOrderedBeforeExisting) {
-    auto params =
-        stdx::make_unique<ClusterClientCursorParams>(_nss, UserNameIterator(), boost::none);
+    auto params = stdx::make_unique<ClusterClientCursorParams>(_nss, boost::none);
     std::vector<ClusterClientCursorParams::RemoteCursor> cursors;
     cursors.emplace_back(kTestShardIds[0], kTestShardHosts[0], CursorResponse(_nss, 123, {}));
     params->remotes = std::move(cursors);
diff --git a/src/mongo/s/query/cluster_client_cursor.h b/src/mongo/s/query/cluster_client_cursor.h
index 742d294f107..c3bd8579bbb 100644
--- a/src/mongo/s/query/cluster_client_cursor.h
+++ b/src/mongo/s/query/cluster_client_cursor.h
@@ -100,11 +100,6 @@ public:
     virtual bool isTailableAndAwaitData() const = 0;
 
     /**
-     * Returns the set of authenticated users when this cursor was created.
-     */
-    virtual UserNameIterator getAuthenticatedUsers() const = 0;
-
-    /**
      * Returns the number of result documents returned so far by this cursor via the next() method.
      */
     virtual long long getNumReturnedSoFar() const = 0;
diff --git a/src/mongo/s/query/cluster_client_cursor_impl.cpp b/src/mongo/s/query/cluster_client_cursor_impl.cpp
index d027d57c8d0..80e6a0b454f 100644
--- a/src/mongo/s/query/cluster_client_cursor_impl.cpp
+++ b/src/mongo/s/query/cluster_client_cursor_impl.cpp
@@ -125,11 +125,6 @@ bool ClusterClientCursorImpl::isTailableAndAwaitData() const {
     return _params.tailableMode == TailableMode::kTailableAndAwaitData;
 }
 
-UserNameIterator ClusterClientCursorImpl::getAuthenticatedUsers() const {
-    return makeUserNameIterator(_params.authenticatedUsers.begin(),
-                                _params.authenticatedUsers.end());
-}
-
 long long ClusterClientCursorImpl::getNumReturnedSoFar() const {
     return _numReturnedSoFar;
 }
diff --git a/src/mongo/s/query/cluster_client_cursor_impl.h b/src/mongo/s/query/cluster_client_cursor_impl.h
index 67e334a7f47..879280e8189 100644
--- a/src/mongo/s/query/cluster_client_cursor_impl.h
+++ b/src/mongo/s/query/cluster_client_cursor_impl.h
@@ -101,8 +101,6 @@ public:
 
     bool isTailableAndAwaitData() const final;
 
-    UserNameIterator getAuthenticatedUsers() const final;
-
     long long getNumReturnedSoFar() const final;
 
     void queueResult(const ClusterQueryResult& result) final;
diff --git a/src/mongo/s/query/cluster_client_cursor_mock.cpp b/src/mongo/s/query/cluster_client_cursor_mock.cpp
index f2327f861fa..c1ddc2d8c90 100644
--- a/src/mongo/s/query/cluster_client_cursor_mock.cpp
+++ b/src/mongo/s/query/cluster_client_cursor_mock.cpp
@@ -83,14 +83,6 @@ bool ClusterClientCursorMock::isTailableAndAwaitData() const {
     return false;
 }
 
-namespace {
-const std::vector<UserName> emptyAuthenticatedUsers{};
-}  // namespace
-
-UserNameIterator ClusterClientCursorMock::getAuthenticatedUsers() const {
-    return makeUserNameIterator(emptyAuthenticatedUsers.begin(), emptyAuthenticatedUsers.end());
-}
-
 void ClusterClientCursorMock::queueResult(const ClusterQueryResult& result) {
     _resultsQueue.push({result});
 }
diff --git a/src/mongo/s/query/cluster_client_cursor_mock.h b/src/mongo/s/query/cluster_client_cursor_mock.h
index c63a2654cda..55dee0d3e41 100644
--- a/src/mongo/s/query/cluster_client_cursor_mock.h
+++ b/src/mongo/s/query/cluster_client_cursor_mock.h
@@ -59,8 +59,6 @@ public:
 
     bool isTailableAndAwaitData() const final;
 
-    UserNameIterator getAuthenticatedUsers() const final;
-
     long long getNumReturnedSoFar() const final;
 
     void queueResult(const ClusterQueryResult& result) final;
diff --git a/src/mongo/s/query/cluster_client_cursor_params.h b/src/mongo/s/query/cluster_client_cursor_params.h
index d47507bd429..3bc5da6c9c0 100644
--- a/src/mongo/s/query/cluster_client_cursor_params.h
+++ b/src/mongo/s/query/cluster_client_cursor_params.h
@@ -85,12 +85,8 @@ struct ClusterClientCursorParams {
     };
 
     ClusterClientCursorParams(NamespaceString nss,
-                              UserNameIterator authenticatedUsersIter,
                               boost::optional<ReadPreferenceSetting> readPref = boost::none)
         : nsString(std::move(nss)) {
-        while (authenticatedUsersIter.more()) {
-            authenticatedUsers.emplace_back(authenticatedUsersIter.next());
-        }
         if (readPref) {
             readPreference = std::move(readPref.get());
         }
@@ -99,9 +95,6 @@ struct ClusterClientCursorParams {
     // Namespace against which the cursors exist.
     NamespaceString nsString;
 
-    // The set of authenticated users when this cursor was created.
-    std::vector<UserName> authenticatedUsers;
-
     // Per-remote node data.
     std::vector<RemoteCursor> remotes;
 
diff --git a/src/mongo/s/query/cluster_cursor_manager.cpp b/src/mongo/s/query/cluster_cursor_manager.cpp
index 253d9f84652..92f6f76f29c 100644
--- a/src/mongo/s/query/cluster_cursor_manager.cpp
+++ b/src/mongo/s/query/cluster_cursor_manager.cpp
@@ -30,7 +30,6 @@
 
 #include "mongo/platform/basic.h"
 
-#include "mongo/db/auth/authorization_session.h"
 #include "mongo/s/query/cluster_cursor_manager.h"
 
 #include <set>
@@ -145,11 +144,6 @@ boost::optional<ReadPreferenceSetting> ClusterCursorManager::PinnedCursor::getRe
     return _cursor->getReadPreference();
 }
 
-UserNameIterator ClusterCursorManager::PinnedCursor::getAuthenticatedUsers() const {
-    invariant(_cursor);
-    return _cursor->getAuthenticatedUsers();
-}
-
 void ClusterCursorManager::PinnedCursor::returnCursor(CursorState cursorState) {
     invariant(_cursor);
     // Note that unpinning a cursor transfers ownership of the underlying ClusterClientCursor object
@@ -221,7 +215,8 @@ StatusWith<CursorId> ClusterCursorManager::registerCursor(
     std::unique_ptr<ClusterClientCursor> cursor,
     const NamespaceString& nss,
     CursorType cursorType,
-    CursorLifetime cursorLifetime) {
+    CursorLifetime cursorLifetime,
+    UserNameIterator authenticatedUsers) {
     // Read the clock out of the lock.
     const auto now = _clockSource->now();
 
@@ -268,8 +263,9 @@ StatusWith<CursorId> ClusterCursorManager::registerCursor(
     } while (cursorId == 0 || entryMap.count(cursorId) > 0);
 
     // Create a new CursorEntry and register it in the CursorEntryContainer's map.
-    auto emplaceResult =
-        entryMap.emplace(cursorId, CursorEntry(std::move(cursor), cursorType, cursorLifetime, now));
+    auto emplaceResult = entryMap.emplace(
+        cursorId,
+        CursorEntry(std::move(cursor), cursorType, cursorLifetime, now, authenticatedUsers));
     invariant(emplaceResult.second);
 
     return cursorId;
@@ -279,6 +275,7 @@ StatusWith<ClusterCursorManager::PinnedCursor> ClusterCursorManager::checkOutCur
     const NamespaceString& nss,
     CursorId cursorId,
     OperationContext* opCtx,
+    AuthzCheckFn authChecker,
     AuthCheck checkSessionAuth) {
     stdx::lock_guard<stdx::mutex> lk(_mutex);
 
@@ -295,6 +292,14 @@ StatusWith<ClusterCursorManager::PinnedCursor> ClusterCursorManager::checkOutCur
         return cursorNotFoundStatus(nss, cursorId);
     }
 
+    // Check if the user is coauthorized to access this cursor.
+    auto authCheckStatus = authChecker(entry->getAuthenticatedUsers());
+    if (!authCheckStatus.isOK()) {
+        return authCheckStatus.withContext(
+            str::stream() << "cursor id " << cursorId
+                          << " was not created by the authenticated user");
+    }
+
     if (checkSessionAuth == kCheckSession) {
         const auto cursorPrivilegeStatus = checkCursorSessionPrivilege(opCtx, entry->getLsid());
         if (!cursorPrivilegeStatus.isOK()) {
@@ -302,13 +307,14 @@ StatusWith<ClusterCursorManager::PinnedCursor> ClusterCursorManager::checkOutCur
         }
     }
 
-    std::unique_ptr<ClusterClientCursor> cursor = entry->releaseCursor();
-    if (!cursor) {
+    if (entry->getOperationUsingCursor()) {
         return cursorInUseStatus(nss, cursorId);
     }
+
     // Note: due to SERVER-31138, despite putting this in a unique_ptr, it's actually not safe to
-    // return before the end of this function.  Be careful to avoid any early returns/throws after
+    // return before the end of this function. Be careful to avoid any early returns/throws after
     // this point.
+    auto cursor = entry->releaseCursor(opCtx);
 
     // We use pinning of a cursor as a proxy for active, user-initiated use of a cursor.  Therefore,
     // we pass down to the logical session cache and vivify the record (updating last use).
@@ -316,8 +322,6 @@ StatusWith<ClusterCursorManager::PinnedCursor> ClusterCursorManager::checkOutCur
         LogicalSessionCache::get(opCtx)->vivify(opCtx, cursor->getLsid().get());
     }
 
-    // Note that pinning a cursor transfers ownership of the underlying ClusterClientCursor object
-    // to the pin; the CursorEntry is left with a null ClusterClientCursor.
     return PinnedCursor(this, std::move(cursor), nss, cursorId);
 }
 
@@ -361,6 +365,22 @@ void ClusterCursorManager::checkInCursor(std::unique_ptr<ClusterClientCursor> cu
     detachedCursor.getValue().reset();
 }
 
+Status ClusterCursorManager::checkAuthForKillCursors(OperationContext* opCtx,
+                                                     const NamespaceString& nss,
+                                                     CursorId cursorId,
+                                                     AuthzCheckFn authChecker) {
+    stdx::lock_guard<stdx::mutex> lk(_mutex);
+    auto entry = _getEntry(lk, nss, cursorId);
+
+    if (!entry) {
+        return cursorNotFoundStatus(nss, cursorId);
+    }
+
+    // Note that getAuthenticatedUsers() is thread-safe, so it's okay to call even if there's
+    // an operation using the cursor.
+    return authChecker(entry->getAuthenticatedUsers());
+}
+
 Status ClusterCursorManager::killCursor(const NamespaceString& nss, CursorId cursorId) {
     stdx::lock_guard<stdx::mutex> lk(_mutex);
 
@@ -380,8 +400,8 @@ void ClusterCursorManager::killMortalCursorsInactiveSince(Date_t cutoff) {
     for (auto& nsContainerPair : _namespaceToContainerMap) {
         for (auto& cursorIdEntryPair : nsContainerPair.second.entryMap) {
             CursorEntry& entry = cursorIdEntryPair.second;
-            if (entry.getLifetimeType() == CursorLifetime::Mortal && entry.isCursorOwned() &&
-                entry.getLastActive() <= cutoff) {
+            if (entry.getLifetimeType() == CursorLifetime::Mortal &&
+                !entry.getOperationUsingCursor() && entry.getLastActive() <= cutoff) {
                 entry.setInactive();
                 log() << "Marking cursor id " << cursorIdEntryPair.first
                       << " for deletion, idle since " << entry.getLastActive().toString();
@@ -468,7 +488,7 @@ ClusterCursorManager::Stats ClusterCursorManager::stats() const {
                 continue;
             }
 
-            if (!entry.isCursorOwned()) {
+            if (entry.getOperationUsingCursor()) {
                 ++stats.cursorsPinned;
             }
 
@@ -602,11 +622,14 @@ StatusWith<std::unique_ptr<ClusterClientCursor>> ClusterCursorManager::_detachCu
         return cursorNotFoundStatus(nss, cursorId);
     }
 
-    std::unique_ptr<ClusterClientCursor> cursor = entry->releaseCursor();
-    if (!cursor) {
+    if (entry->getOperationUsingCursor()) {
         return cursorInUseStatus(nss, cursorId);
     }
 
+    // Transfer ownership away from the entry.
+    std::unique_ptr<ClusterClientCursor> cursor = entry->releaseCursor(nullptr);
+
+    // Destroy the entry.
     auto nsToContainerIt = _namespaceToContainerMap.find(nss);
     invariant(nsToContainerIt != _namespaceToContainerMap.end());
     CursorEntryMap& entryMap = nsToContainerIt->second.entryMap;
diff --git a/src/mongo/s/query/cluster_cursor_manager.h b/src/mongo/s/query/cluster_cursor_manager.h
index 8357117099f..923e6914953 100644
--- a/src/mongo/s/query/cluster_cursor_manager.h
+++ b/src/mongo/s/query/cluster_cursor_manager.h
@@ -115,6 +115,11 @@ public:
         size_t cursorsPinned = 0;
     };
 
+    // Represents a function that may be passed into a ClusterCursorManager method which checks
+    // whether the current client is authorized to perform the operation in question. The function
+    // will be passed the list of users authorized to use the cursor.
+    using AuthzCheckFn = std::function<Status(UserNameIterator)>;
+
     /**
      * PinnedCursor is a moveable, non-copyable class representing ownership of a cursor that has
      * been leased from a ClusterCursorManager.
@@ -186,12 +191,6 @@ public:
         bool isTailableAndAwaitData() const;
 
         /**
-         * Returns the set of authenticated users when this cursor was created. Cannot be called
-         * after returnCursor() is called.  A cursor must be owned.
-         */
-        UserNameIterator getAuthenticatedUsers() const;
-
-        /**
          * Transfers ownership of the underlying cursor back to the manager.  A cursor must be
          * owned, and a cursor will no longer be owned after this method completes.
          *
@@ -299,7 +298,8 @@ public:
                                         std::unique_ptr<ClusterClientCursor> cursor,
                                         const NamespaceString& nss,
                                         CursorType cursorType,
-                                        CursorLifetime cursorLifetime);
+                                        CursorLifetime cursorLifetime,
+                                        UserNameIterator authenticatedUsers);
 
     /**
      * Moves the given cursor to the 'pinned' state, and transfers ownership of the cursor to the
@@ -310,6 +310,10 @@ public:
      * returns an error Status with code CursorInUse.  If the given cursor is not registered or has
      * a pending kill, returns an error Status with code CursorNotFound.
      *
+     * 'authChecker' is function that will be called with the list of users authorized to use this
+     * cursor. This function should check whether the current client is also authorized to use this
+     * cursor, and if not, return an error status, which will cause checkOutCursor to fail.
+     *
      * This method updates the 'last active' time associated with the cursor to the current time.
      *
      * Does not block.
@@ -318,9 +322,20 @@ public:
     StatusWith<PinnedCursor> checkOutCursor(const NamespaceString& nss,
                                             CursorId cursorId,
                                             OperationContext* opCtx,
+                                            AuthzCheckFn authChecker,
                                             AuthCheck checkSessionAuth = kCheckSession);
 
     /**
+     * This method will find the given cursor, and if it exists, call 'authChecker', passing the
+     * list of users authorized to use the cursor. Will propagate the return value of authChecker.
+     */
+    Status checkAuthForKillCursors(OperationContext* opCtx,
+                                   const NamespaceString& nss,
+                                   CursorId cursorId,
+                                   AuthzCheckFn authChecker);
+
+
+    /**
      * Informs the manager that the given cursor should be killed.  The cursor need not necessarily
      * be in the 'idle' state, and the lifetime type of the cursor is ignored.
      *
@@ -462,12 +477,15 @@ private:
         CursorEntry(std::unique_ptr<ClusterClientCursor> cursor,
                     CursorType cursorType,
                     CursorLifetime cursorLifetime,
-                    Date_t lastActive)
+                    Date_t lastActive,
+                    UserNameIterator authenticatedUsersIter)
             : _cursor(std::move(cursor)),
               _cursorType(cursorType),
               _cursorLifetime(cursorLifetime),
               _lastActive(lastActive),
-              _lsid(_cursor->getLsid()) {
+              _lsid(_cursor->getLsid()),
+              _authenticatedUsers(
+                  userNameIteratorToContainer<std::vector<UserName>>(authenticatedUsersIter)) {
             invariant(_cursor);
         }
 
@@ -494,29 +512,38 @@ private:
             return _lastActive;
         }
 
-        bool isCursorOwned() const {
-            return static_cast<bool>(_cursor);
-        }
-
         boost::optional<LogicalSessionId> getLsid() const {
             return _lsid;
         }
 
         /**
-         * Releases the cursor from this entry.  If the cursor has already been released, returns
-         * null.
+         * Returns the cursor owned by this CursorEntry for an operation to use. Only one operation
+         * may use the cursor at a time, so callers should check that getOperationUsingCursor()
+         * returns null before using this function. Callers may pass nullptr, but only if the
+         * released cursor is going to be deleted.
          */
-        std::unique_ptr<ClusterClientCursor> releaseCursor() {
+        std::unique_ptr<ClusterClientCursor> releaseCursor(OperationContext* opCtx) {
+            invariant(!_operationUsingCursor);
+            invariant(_cursor);
+            _operationUsingCursor = opCtx;
             return std::move(_cursor);
         }
 
+        OperationContext* getOperationUsingCursor() const {
+            return _operationUsingCursor;
+        }
+
         /**
-         * Transfers ownership of the given released cursor back to this entry.
+         * Indicate that the cursor is no longer in use by an operation. Once this is called,
+         * another operation may check the cursor out.
          */
         void returnCursor(std::unique_ptr<ClusterClientCursor> cursor) {
             invariant(cursor);
             invariant(!_cursor);
+            invariant(_operationUsingCursor);
+
             _cursor = std::move(cursor);
+            _operationUsingCursor = nullptr;
         }
 
         void setKillPending() {
@@ -531,6 +558,10 @@ private:
             _lastActive = lastActive;
         }
 
+        UserNameIterator getAuthenticatedUsers() const {
+            return makeUserNameIterator(_authenticatedUsers.begin(), _authenticatedUsers.end());
+        }
+
     private:
         std::unique_ptr<ClusterClientCursor> _cursor;
         bool _killPending = false;
@@ -539,6 +570,12 @@ private:
         CursorLifetime _cursorLifetime = CursorLifetime::Mortal;
         Date_t _lastActive;
         boost::optional<LogicalSessionId> _lsid;
+
+        // Current operation using the cursor. Non-null if the cursor is checked out.
+        OperationContext* _operationUsingCursor = nullptr;
+
+        // The set of users authorized to use this cursor.
+        const std::vector<UserName> _authenticatedUsers;
     };
 
     /**
diff --git a/src/mongo/s/query/cluster_cursor_manager_test.cpp b/src/mongo/s/query/cluster_cursor_manager_test.cpp
index 398cd62a4ac..626f585144b 100644
--- a/src/mongo/s/query/cluster_cursor_manager_test.cpp
+++ b/src/mongo/s/query/cluster_cursor_manager_test.cpp
@@ -56,6 +56,14 @@ protected:
     ServiceContext::UniqueOperationContext _opCtx;
     Client* _client;
 
+    static Status successAuthChecker(UserNameIterator userNames) {
+        return Status::OK();
+    };
+
+    static Status failAuthChecker(UserNameIterator userNames) {
+        return {ErrorCodes::Unauthorized, "Unauthorized"};
+    };
+
     /**
      * Returns an unowned pointer to the manager owned by this test fixture.
      */
@@ -135,12 +143,14 @@ TEST_F(ClusterCursorManagerTest, RegisterCursor) {
     auto cursor = allocateMockCursor();
     cursor->queueResult(BSON("a" << 1));
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                std::move(cursor),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
-    auto pinnedCursor = getManager()->checkOutCursor(nss, cursorId, _opCtx.get());
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
+    auto pinnedCursor =
+        getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), successAuthChecker);
     ASSERT_OK(pinnedCursor.getStatus());
     auto nextResult = pinnedCursor.getValue().next(RouterExecStage::ExecContext::kInitialFind);
     ASSERT_OK(nextResult.getStatus());
@@ -154,11 +164,12 @@ TEST_F(ClusterCursorManagerTest, RegisterCursor) {
 // Test that registering a cursor returns a non-zero cursor id.
 TEST_F(ClusterCursorManagerTest, RegisterCursorReturnsNonZeroId) {
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
     ASSERT_NE(0, cursorId);
 }
 
@@ -167,12 +178,14 @@ TEST_F(ClusterCursorManagerTest, CheckOutCursorBasic) {
     auto cursor = allocateMockCursor();
     cursor->queueResult(BSON("a" << 1));
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                std::move(cursor),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
-    auto checkedOutCursor = getManager()->checkOutCursor(nss, cursorId, _opCtx.get());
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
+    auto checkedOutCursor =
+        getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), successAuthChecker);
     ASSERT_OK(checkedOutCursor.getStatus());
     ASSERT_EQ(cursorId, checkedOutCursor.getValue().getCursorId());
     auto nextResult = checkedOutCursor.getValue().next(RouterExecStage::ExecContext::kInitialFind);
@@ -193,14 +206,16 @@ TEST_F(ClusterCursorManagerTest, CheckOutCursorMultipleCursors) {
         auto cursor = allocateMockCursor();
         cursor->queueResult(BSON("a" << i));
         cursorIds[i] =
-            assertGet(getManager()->registerCursor(nullptr,
+            assertGet(getManager()->registerCursor(_opCtx.get(),
                                                    std::move(cursor),
                                                    nss,
                                                    ClusterCursorManager::CursorType::SingleTarget,
-                                                   ClusterCursorManager::CursorLifetime::Mortal));
+                                                   ClusterCursorManager::CursorLifetime::Mortal,
+                                                   UserNameIterator()));
     }
     for (int i = 0; i < numCursors; ++i) {
-        auto pinnedCursor = getManager()->checkOutCursor(nss, cursorIds[i], _opCtx.get());
+        auto pinnedCursor =
+            getManager()->checkOutCursor(nss, cursorIds[i], _opCtx.get(), successAuthChecker);
         ASSERT_OK(pinnedCursor.getStatus());
         auto nextResult = pinnedCursor.getValue().next(RouterExecStage::ExecContext::kInitialFind);
         ASSERT_OK(nextResult.getStatus());
@@ -215,34 +230,39 @@ TEST_F(ClusterCursorManagerTest, CheckOutCursorMultipleCursors) {
 // Test that checking out a pinned cursor returns an error with code ErrorCodes::CursorInUse.
 TEST_F(ClusterCursorManagerTest, CheckOutCursorPinned) {
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
-    auto pinnedCursor = getManager()->checkOutCursor(nss, cursorId, _opCtx.get());
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
+    auto pinnedCursor =
+        getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), successAuthChecker);
     ASSERT_OK(pinnedCursor.getStatus());
-    ASSERT_EQ(ErrorCodes::CursorInUse,
-              getManager()->checkOutCursor(nss, cursorId, _opCtx.get()).getStatus());
+    ASSERT_EQ(
+        ErrorCodes::CursorInUse,
+        getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), successAuthChecker).getStatus());
 }
 
 // Test that checking out a killed cursor returns an error with code ErrorCodes::CursorNotFound.
 TEST_F(ClusterCursorManagerTest, CheckOutCursorKilled) {
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
     ASSERT_OK(getManager()->killCursor(nss, cursorId));
-    ASSERT_EQ(ErrorCodes::CursorNotFound,
-              getManager()->checkOutCursor(nss, cursorId, _opCtx.get()).getStatus());
+    ASSERT_EQ(
+        ErrorCodes::CursorNotFound,
+        getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), successAuthChecker).getStatus());
 }
 
 // Test that checking out an unknown cursor returns an error with code ErrorCodes::CursorNotFound.
 TEST_F(ClusterCursorManagerTest, CheckOutCursorUnknown) {
     ASSERT_EQ(ErrorCodes::CursorNotFound,
-              getManager()->checkOutCursor(nss, 5, nullptr).getStatus());
+              getManager()->checkOutCursor(nss, 5, nullptr, successAuthChecker).getStatus());
 }
 
 // Test that checking out a unknown cursor returns an error with code ErrorCodes::CursorNotFound,
@@ -251,40 +271,48 @@ TEST_F(ClusterCursorManagerTest, CheckOutCursorWrongNamespace) {
     const NamespaceString correctNamespace("test.correct");
     const NamespaceString incorrectNamespace("test.incorrect");
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                correctNamespace,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
     ASSERT_EQ(ErrorCodes::CursorNotFound,
-              getManager()->checkOutCursor(incorrectNamespace, cursorId, nullptr).getStatus());
+              getManager()
+                  ->checkOutCursor(incorrectNamespace, cursorId, nullptr, successAuthChecker)
+                  .getStatus());
 }
 
 // Test that checking out a unknown cursor returns an error with code ErrorCodes::CursorNotFound,
 // even if there is an existing cursor with the same namespace but a different cursor id.
 TEST_F(ClusterCursorManagerTest, CheckOutCursorWrongCursorId) {
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
     ASSERT_EQ(ErrorCodes::CursorNotFound,
-              getManager()->checkOutCursor(nss, cursorId + 1, _opCtx.get()).getStatus());
+              getManager()
+                  ->checkOutCursor(nss, cursorId + 1, _opCtx.get(), successAuthChecker)
+                  .getStatus());
 }
 
 // Test that checking out a cursor updates the 'last active' time associated with the cursor to the
 // current time.
 TEST_F(ClusterCursorManagerTest, CheckOutCursorUpdateActiveTime) {
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
     Date_t cursorRegistrationTime = getClockSource()->now();
     getClockSource()->advance(Milliseconds(1));
-    auto checkedOutCursor = getManager()->checkOutCursor(nss, cursorId, _opCtx.get());
+    auto checkedOutCursor =
+        getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), successAuthChecker);
     ASSERT_OK(checkedOutCursor.getStatus());
     checkedOutCursor.getValue().returnCursor(ClusterCursorManager::CursorState::NotExhausted);
     getManager()->killMortalCursorsInactiveSince(cursorRegistrationTime);
@@ -293,17 +321,33 @@ TEST_F(ClusterCursorManagerTest, CheckOutCursorUpdateActiveTime) {
     ASSERT(!isMockCursorKilled(0));
 }
 
+TEST_F(ClusterCursorManagerTest, CheckOutCursorAuthFails) {
+    auto cursorId =
+        assertGet(getManager()->registerCursor(_opCtx.get(),
+                                               allocateMockCursor(),
+                                               nss,
+                                               ClusterCursorManager::CursorType::SingleTarget,
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
+    auto checkedOutCursor =
+        getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), failAuthChecker);
+    ASSERT_EQ(checkedOutCursor.getStatus(), ErrorCodes::Unauthorized);
+}
+
+
 // Test that checking in a cursor updates the 'last active' time associated with the cursor to the
 // current time.
 TEST_F(ClusterCursorManagerTest, ReturnCursorUpdateActiveTime) {
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
     Date_t cursorCheckOutTime = getClockSource()->now();
-    auto checkedOutCursor = getManager()->checkOutCursor(nss, cursorId, _opCtx.get());
+    auto checkedOutCursor =
+        getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), successAuthChecker);
     ASSERT_OK(checkedOutCursor.getStatus());
     getClockSource()->advance(Milliseconds(1));
     checkedOutCursor.getValue().returnCursor(ClusterCursorManager::CursorState::NotExhausted);
@@ -315,12 +359,14 @@ TEST_F(ClusterCursorManagerTest, ReturnCursorUpdateActiveTime) {
 // Test that killing a pinned cursor by id successfully kills the cursor.
 TEST_F(ClusterCursorManagerTest, KillCursorBasic) {
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
-    auto pinnedCursor = getManager()->checkOutCursor(nss, cursorId, _opCtx.get());
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
+    auto pinnedCursor =
+        getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), successAuthChecker);
     ASSERT_OK(pinnedCursor.getStatus());
     ASSERT_OK(getManager()->killCursor(nss, pinnedCursor.getValue().getCursorId()));
     pinnedCursor.getValue().returnCursor(ClusterCursorManager::CursorState::NotExhausted);
@@ -337,11 +383,12 @@ TEST_F(ClusterCursorManagerTest, KillCursorMultipleCursors) {
     // Register cursors and populate 'cursorIds' with the returned cursor ids.
     for (size_t i = 0; i < numCursors; ++i) {
         cursorIds[i] =
-            assertGet(getManager()->registerCursor(nullptr,
+            assertGet(getManager()->registerCursor(_opCtx.get(),
                                                    allocateMockCursor(),
                                                    nss,
                                                    ClusterCursorManager::CursorType::SingleTarget,
-                                                   ClusterCursorManager::CursorLifetime::Mortal));
+                                                   ClusterCursorManager::CursorLifetime::Mortal,
+                                                   UserNameIterator()));
     }
     // Kill each cursor and verify that it was successfully killed.
     for (size_t i = 0; i < numCursors; ++i) {
@@ -364,11 +411,12 @@ TEST_F(ClusterCursorManagerTest, KillCursorWrongNamespace) {
     const NamespaceString correctNamespace("test.correct");
     const NamespaceString incorrectNamespace("test.incorrect");
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                correctNamespace,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
     Status killResult = getManager()->killCursor(incorrectNamespace, cursorId);
     ASSERT_EQ(ErrorCodes::CursorNotFound, killResult);
 }
@@ -377,22 +425,24 @@ TEST_F(ClusterCursorManagerTest, KillCursorWrongNamespace) {
 // even if there is an existing cursor with the same namespace but a different cursor id.
 TEST_F(ClusterCursorManagerTest, KillCursorWrongCursorId) {
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
     Status killResult = getManager()->killCursor(nss, cursorId + 1);
     ASSERT_EQ(ErrorCodes::CursorNotFound, killResult);
 }
 
 // Test that killing all mortal expired cursors correctly kills a mortal expired cursor.
 TEST_F(ClusterCursorManagerTest, KillMortalCursorsInactiveSinceBasic) {
-    ASSERT_OK(getManager()->registerCursor(nullptr,
+    ASSERT_OK(getManager()->registerCursor(_opCtx.get(),
                                            allocateMockCursor(),
                                            nss,
                                            ClusterCursorManager::CursorType::SingleTarget,
-                                           ClusterCursorManager::CursorLifetime::Mortal));
+                                           ClusterCursorManager::CursorLifetime::Mortal,
+                                           UserNameIterator()));
     getManager()->killMortalCursorsInactiveSince(getClockSource()->now());
     ASSERT(!isMockCursorKilled(0));
     getManager()->reapZombieCursors(nullptr);
@@ -403,11 +453,12 @@ TEST_F(ClusterCursorManagerTest, KillMortalCursorsInactiveSinceBasic) {
 TEST_F(ClusterCursorManagerTest, KillMortalCursorsInactiveSinceSkipUnexpired) {
     Date_t timeBeforeCursorCreation = getClockSource()->now();
     getClockSource()->advance(Milliseconds(1));
-    ASSERT_OK(getManager()->registerCursor(nullptr,
+    ASSERT_OK(getManager()->registerCursor(_opCtx.get(),
                                            allocateMockCursor(),
                                            nss,
                                            ClusterCursorManager::CursorType::SingleTarget,
-                                           ClusterCursorManager::CursorLifetime::Mortal));
+                                           ClusterCursorManager::CursorLifetime::Mortal,
+                                           UserNameIterator()));
     getManager()->killMortalCursorsInactiveSince(timeBeforeCursorCreation);
     ASSERT(!isMockCursorKilled(0));
     getManager()->reapZombieCursors(nullptr);
@@ -416,11 +467,12 @@ TEST_F(ClusterCursorManagerTest, KillMortalCursorsInactiveSinceSkipUnexpired) {
 
 // Test that killing all mortal expired cursors does not kill a cursor that is immortal.
 TEST_F(ClusterCursorManagerTest, KillMortalCursorsInactiveSinceSkipImmortal) {
-    ASSERT_OK(getManager()->registerCursor(nullptr,
+    ASSERT_OK(getManager()->registerCursor(_opCtx.get(),
                                            allocateMockCursor(),
                                            nss,
                                            ClusterCursorManager::CursorType::SingleTarget,
-                                           ClusterCursorManager::CursorLifetime::Immortal));
+                                           ClusterCursorManager::CursorLifetime::Immortal,
+                                           UserNameIterator()));
     getManager()->killMortalCursorsInactiveSince(getClockSource()->now());
     ASSERT(!isMockCursorKilled(0));
     getManager()->reapZombieCursors(nullptr);
@@ -431,12 +483,14 @@ TEST_F(ClusterCursorManagerTest, KillMortalCursorsInactiveSinceSkipImmortal) {
 // pinned.
 TEST_F(ClusterCursorManagerTest, ShouldNotKillPinnedCursors) {
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
-    auto pin = assertGet(getManager()->checkOutCursor(nss, cursorId, _opCtx.get()));
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
+    auto pin =
+        assertGet(getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), successAuthChecker));
     getManager()->killMortalCursorsInactiveSince(getClockSource()->now());
     ASSERT(!isMockCursorKilled(0));
     getManager()->reapZombieCursors(nullptr);
@@ -458,11 +512,12 @@ TEST_F(ClusterCursorManagerTest, KillMortalCursorsInactiveSinceMultipleCursors)
         if (i < numKilledCursorsExpected) {
             cutoff = getClockSource()->now();
         }
-        ASSERT_OK(getManager()->registerCursor(nullptr,
+        ASSERT_OK(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
         getClockSource()->advance(Milliseconds(1));
     }
     getManager()->killMortalCursorsInactiveSince(cutoff);
@@ -483,11 +538,12 @@ TEST_F(ClusterCursorManagerTest, KillMortalCursorsInactiveSinceMultipleCursors)
 TEST_F(ClusterCursorManagerTest, KillAllCursors) {
     const size_t numCursors = 10;
     for (size_t i = 0; i < numCursors; ++i) {
-        ASSERT_OK(getManager()->registerCursor(nullptr,
+        ASSERT_OK(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
     }
     getManager()->killAllCursors();
     for (size_t i = 0; i < numCursors; ++i) {
@@ -503,11 +559,12 @@ TEST_F(ClusterCursorManagerTest, KillAllCursors) {
 // cursor.
 TEST_F(ClusterCursorManagerTest, ReapZombieCursorsBasic) {
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
     ASSERT_OK(getManager()->killCursor(nss, cursorId));
     ASSERT(!isMockCursorKilled(0));
     getManager()->reapZombieCursors(nullptr);
@@ -518,12 +575,14 @@ TEST_F(ClusterCursorManagerTest, ReapZombieCursorsBasic) {
 // that is still pinned.
 TEST_F(ClusterCursorManagerTest, ReapZombieCursorsSkipPinned) {
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
-    auto pinnedCursor = getManager()->checkOutCursor(nss, cursorId, _opCtx.get());
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
+    auto pinnedCursor =
+        getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), successAuthChecker);
     ASSERT(!isMockCursorKilled(0));
     getManager()->reapZombieCursors(nullptr);
     ASSERT(!isMockCursorKilled(0));
@@ -532,11 +591,12 @@ TEST_F(ClusterCursorManagerTest, ReapZombieCursorsSkipPinned) {
 // Test that reaping does not call kill() on the underlying ClusterClientCursor for cursors that
 // haven't been killed.
 TEST_F(ClusterCursorManagerTest, ReapZombieCursorsSkipNonZombies) {
-    ASSERT_OK(getManager()->registerCursor(nullptr,
+    ASSERT_OK(getManager()->registerCursor(_opCtx.get(),
                                            allocateMockCursor(),
                                            nss,
                                            ClusterCursorManager::CursorType::SingleTarget,
-                                           ClusterCursorManager::CursorLifetime::Mortal));
+                                           ClusterCursorManager::CursorLifetime::Mortal,
+                                           UserNameIterator()));
     ASSERT(!isMockCursorKilled(0));
     getManager()->reapZombieCursors(nullptr);
     ASSERT(!isMockCursorKilled(0));
@@ -551,33 +611,37 @@ TEST_F(ClusterCursorManagerTest, StatsInitAsZero) {
 
 // Test that registering a sharded cursor updates the corresponding counter in stats().
 TEST_F(ClusterCursorManagerTest, StatsRegisterShardedCursor) {
-    ASSERT_OK(getManager()->registerCursor(nullptr,
+    ASSERT_OK(getManager()->registerCursor(_opCtx.get(),
                                            allocateMockCursor(),
                                            nss,
                                            ClusterCursorManager::CursorType::MultiTarget,
-                                           ClusterCursorManager::CursorLifetime::Mortal));
+                                           ClusterCursorManager::CursorLifetime::Mortal,
+                                           UserNameIterator()));
     ASSERT_EQ(1U, getManager()->stats().cursorsMultiTarget);
 }
 
 // Test that registering a not-sharded cursor updates the corresponding counter in stats().
 TEST_F(ClusterCursorManagerTest, StatsRegisterNotShardedCursor) {
-    ASSERT_OK(getManager()->registerCursor(nullptr,
+    ASSERT_OK(getManager()->registerCursor(_opCtx.get(),
                                            allocateMockCursor(),
                                            nss,
                                            ClusterCursorManager::CursorType::SingleTarget,
-                                           ClusterCursorManager::CursorLifetime::Mortal));
+                                           ClusterCursorManager::CursorLifetime::Mortal,
+                                           UserNameIterator()));
     ASSERT_EQ(1U, getManager()->stats().cursorsSingleTarget);
 }
 
 // Test that checking out a cursor updates the pinned counter in stats().
 TEST_F(ClusterCursorManagerTest, StatsPinCursor) {
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::MultiTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
-    auto pinnedCursor = getManager()->checkOutCursor(nss, cursorId, _opCtx.get());
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
+    auto pinnedCursor =
+        getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), successAuthChecker);
     ASSERT_EQ(1U, getManager()->stats().cursorsPinned);
 }
 
@@ -586,21 +650,23 @@ TEST_F(ClusterCursorManagerTest, StatsPinCursor) {
 TEST_F(ClusterCursorManagerTest, StatsRegisterMultipleCursors) {
     const size_t numShardedCursors = 10;
     for (size_t i = 0; i < numShardedCursors; ++i) {
-        ASSERT_OK(getManager()->registerCursor(nullptr,
+        ASSERT_OK(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::MultiTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
         ASSERT_EQ(i + 1, getManager()->stats().cursorsMultiTarget);
         ASSERT_EQ(0U, getManager()->stats().cursorsSingleTarget);
     }
     const size_t numNotShardedCursors = 10;
     for (size_t i = 0; i < numNotShardedCursors; ++i) {
-        ASSERT_OK(getManager()->registerCursor(nullptr,
+        ASSERT_OK(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
         ASSERT_EQ(numShardedCursors, getManager()->stats().cursorsMultiTarget);
         ASSERT_EQ(i + 1, getManager()->stats().cursorsSingleTarget);
     }
@@ -609,11 +675,12 @@ TEST_F(ClusterCursorManagerTest, StatsRegisterMultipleCursors) {
 // Test that killing a sharded cursor decrements the corresponding counter in stats().
 TEST_F(ClusterCursorManagerTest, StatsKillShardedCursor) {
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::MultiTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
     ASSERT_EQ(1U, getManager()->stats().cursorsMultiTarget);
     ASSERT_OK(getManager()->killCursor(nss, cursorId));
     ASSERT_EQ(0U, getManager()->stats().cursorsMultiTarget);
@@ -622,11 +689,12 @@ TEST_F(ClusterCursorManagerTest, StatsKillShardedCursor) {
 // Test that killing a not-sharded cursor decrements the corresponding counter in stats().
 TEST_F(ClusterCursorManagerTest, StatsKillNotShardedCursor) {
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
     ASSERT_EQ(1U, getManager()->stats().cursorsSingleTarget);
     ASSERT_OK(getManager()->killCursor(nss, cursorId));
     ASSERT_EQ(0U, getManager()->stats().cursorsSingleTarget);
@@ -635,12 +703,14 @@ TEST_F(ClusterCursorManagerTest, StatsKillNotShardedCursor) {
 // Test that killing a pinned cursor decrements the corresponding counter in stats().
 TEST_F(ClusterCursorManagerTest, StatsKillPinnedCursor) {
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::MultiTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
-    auto pinnedCursor = getManager()->checkOutCursor(nss, cursorId, _opCtx.get());
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
+    auto pinnedCursor =
+        getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), successAuthChecker);
     ASSERT_EQ(1U, getManager()->stats().cursorsPinned);
     ASSERT_OK(getManager()->killCursor(nss, cursorId));
     ASSERT_EQ(0U, getManager()->stats().cursorsPinned);
@@ -649,12 +719,14 @@ TEST_F(ClusterCursorManagerTest, StatsKillPinnedCursor) {
 // Test that exhausting a sharded cursor decrements the corresponding counter in stats().
 TEST_F(ClusterCursorManagerTest, StatsExhaustShardedCursor) {
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::MultiTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
-    auto pinnedCursor = getManager()->checkOutCursor(nss, cursorId, _opCtx.get());
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
+    auto pinnedCursor =
+        getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), successAuthChecker);
     ASSERT_OK(pinnedCursor.getStatus());
     ASSERT_OK(pinnedCursor.getValue().next(RouterExecStage::ExecContext::kInitialFind).getStatus());
     ASSERT_EQ(1U, getManager()->stats().cursorsMultiTarget);
@@ -665,12 +737,14 @@ TEST_F(ClusterCursorManagerTest, StatsExhaustShardedCursor) {
 // Test that exhausting a not-sharded cursor decrements the corresponding counter in stats().
 TEST_F(ClusterCursorManagerTest, StatsExhaustNotShardedCursor) {
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
-    auto pinnedCursor = getManager()->checkOutCursor(nss, cursorId, _opCtx.get());
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
+    auto pinnedCursor =
+        getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), successAuthChecker);
     ASSERT_OK(pinnedCursor.getStatus());
     ASSERT_OK(pinnedCursor.getValue().next(RouterExecStage::ExecContext::kInitialFind).getStatus());
     ASSERT_EQ(1U, getManager()->stats().cursorsSingleTarget);
@@ -682,12 +756,14 @@ TEST_F(ClusterCursorManagerTest, StatsExhaustNotShardedCursor) {
 // stats().
 TEST_F(ClusterCursorManagerTest, StatsExhaustPinnedCursor) {
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
-    auto pinnedCursor = getManager()->checkOutCursor(nss, cursorId, _opCtx.get());
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
+    auto pinnedCursor =
+        getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), successAuthChecker);
     ASSERT_OK(pinnedCursor.getStatus());
     ASSERT_OK(pinnedCursor.getValue().next(RouterExecStage::ExecContext::kInitialFind).getStatus());
     ASSERT_EQ(1U, getManager()->stats().cursorsPinned);
@@ -699,12 +775,14 @@ TEST_F(ClusterCursorManagerTest, StatsExhaustPinnedCursor) {
 // stats().
 TEST_F(ClusterCursorManagerTest, StatsCheckInWithoutExhaustingPinnedCursor) {
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
-    auto pinnedCursor = getManager()->checkOutCursor(nss, cursorId, _opCtx.get());
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
+    auto pinnedCursor =
+        getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), successAuthChecker);
     ASSERT_OK(pinnedCursor.getStatus());
     ASSERT_OK(pinnedCursor.getValue().next(RouterExecStage::ExecContext::kInitialFind).getStatus());
     ASSERT_EQ(1U, getManager()->stats().cursorsPinned);
@@ -715,11 +793,12 @@ TEST_F(ClusterCursorManagerTest, StatsCheckInWithoutExhaustingPinnedCursor) {
 // Test that getting the namespace for a cursor returns the correct namespace.
 TEST_F(ClusterCursorManagerTest, GetNamespaceForCursorIdBasic) {
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
     boost::optional<NamespaceString> cursorNamespace =
         getManager()->getNamespaceForCursorId(cursorId);
     ASSERT(cursorNamespace);
@@ -733,11 +812,12 @@ TEST_F(ClusterCursorManagerTest, GetNamespaceForCursorIdMultipleCursorsSameNames
     std::vector<CursorId> cursorIds(numCursors);
     for (size_t i = 0; i < numCursors; ++i) {
         cursorIds[i] =
-            assertGet(getManager()->registerCursor(nullptr,
+            assertGet(getManager()->registerCursor(_opCtx.get(),
                                                    allocateMockCursor(),
                                                    nss,
                                                    ClusterCursorManager::CursorType::SingleTarget,
-                                                   ClusterCursorManager::CursorLifetime::Mortal));
+                                                   ClusterCursorManager::CursorLifetime::Mortal,
+                                                   UserNameIterator()));
     }
     for (size_t i = 0; i < numCursors; ++i) {
         boost::optional<NamespaceString> cursorNamespace =
@@ -755,11 +835,12 @@ TEST_F(ClusterCursorManagerTest, GetNamespaceForCursorIdMultipleCursorsDifferent
     for (size_t i = 0; i < numCursors; ++i) {
         NamespaceString cursorNamespace(std::string(str::stream() << "test.collection" << i));
         auto cursorId =
-            assertGet(getManager()->registerCursor(nullptr,
+            assertGet(getManager()->registerCursor(_opCtx.get(),
                                                    allocateMockCursor(),
                                                    cursorNamespace,
                                                    ClusterCursorManager::CursorType::SingleTarget,
-                                                   ClusterCursorManager::CursorLifetime::Mortal));
+                                                   ClusterCursorManager::CursorLifetime::Mortal,
+                                                   UserNameIterator()));
         cursors[i] = {cursorNamespace, cursorId};
     }
     for (size_t i = 0; i < numCursors; ++i) {
@@ -786,18 +867,21 @@ TEST_F(ClusterCursorManagerTest, PinnedCursorDefaultConstructor) {
 // cursor.
 TEST_F(ClusterCursorManagerTest, PinnedCursorReturnCursorNotExhausted) {
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
-    auto registeredCursor = getManager()->checkOutCursor(nss, cursorId, _opCtx.get());
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
+    auto registeredCursor =
+        getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), successAuthChecker);
     ASSERT_OK(registeredCursor.getStatus());
     ASSERT_EQ(cursorId, registeredCursor.getValue().getCursorId());
     ASSERT_NE(0, cursorId);
     registeredCursor.getValue().returnCursor(ClusterCursorManager::CursorState::NotExhausted);
     ASSERT_EQ(0, registeredCursor.getValue().getCursorId());
-    auto checkedOutCursor = getManager()->checkOutCursor(nss, cursorId, _opCtx.get());
+    auto checkedOutCursor =
+        getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), successAuthChecker);
     ASSERT_OK(checkedOutCursor.getStatus());
 }
 
@@ -805,12 +889,14 @@ TEST_F(ClusterCursorManagerTest, PinnedCursorReturnCursorNotExhausted) {
 // cursor, and leaves the pin owning no cursor.
 TEST_F(ClusterCursorManagerTest, PinnedCursorReturnCursorExhausted) {
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
-    auto registeredCursor = getManager()->checkOutCursor(nss, cursorId, _opCtx.get());
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
+    auto registeredCursor =
+        getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), successAuthChecker);
     ASSERT_OK(registeredCursor.getStatus());
     ASSERT_EQ(cursorId, registeredCursor.getValue().getCursorId());
     ASSERT_NE(0, cursorId);
@@ -822,7 +908,8 @@ TEST_F(ClusterCursorManagerTest, PinnedCursorReturnCursorExhausted) {
     // Cursor should have been destroyed without ever being killed. To be sure that the cursor has
     // not been marked kill pending but not yet destroyed (i.e. that the cursor is not a zombie), we
     // reapZombieCursors() and check that the cursor still has not been killed.
-    ASSERT_NOT_OK(getManager()->checkOutCursor(nss, cursorId, _opCtx.get()).getStatus());
+    ASSERT_NOT_OK(
+        getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), successAuthChecker).getStatus());
     ASSERT(!isMockCursorKilled(0));
     getManager()->reapZombieCursors(nullptr);
     ASSERT(!isMockCursorKilled(0));
@@ -838,12 +925,14 @@ TEST_F(ClusterCursorManagerTest, PinnedCursorReturnCursorExhaustedWithNonExhaust
     mockCursor->markRemotesNotExhausted();
 
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                std::move(mockCursor),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
-    auto registeredCursor = getManager()->checkOutCursor(nss, cursorId, _opCtx.get());
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
+    auto registeredCursor =
+        getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), successAuthChecker);
     ASSERT_OK(registeredCursor.getStatus());
     ASSERT_EQ(cursorId, registeredCursor.getValue().getCursorId());
     ASSERT_NE(0, cursorId);
@@ -853,7 +942,8 @@ TEST_F(ClusterCursorManagerTest, PinnedCursorReturnCursorExhaustedWithNonExhaust
     ASSERT_EQ(0, registeredCursor.getValue().getCursorId());
 
     // Cursor should be kill pending, so it will be killed during reaping.
-    ASSERT_NOT_OK(getManager()->checkOutCursor(nss, cursorId, _opCtx.get()).getStatus());
+    ASSERT_NOT_OK(
+        getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), successAuthChecker).getStatus());
     ASSERT(!isMockCursorKilled(0));
     getManager()->reapZombieCursors(nullptr);
     ASSERT(isMockCursorKilled(0));
@@ -863,12 +953,14 @@ TEST_F(ClusterCursorManagerTest, PinnedCursorReturnCursorExhaustedWithNonExhaust
 // been returned.
 TEST_F(ClusterCursorManagerTest, PinnedCursorMoveAssignmentKill) {
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
-    auto pinnedCursor = getManager()->checkOutCursor(nss, cursorId, _opCtx.get());
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
+    auto pinnedCursor =
+        getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), successAuthChecker);
     pinnedCursor = ClusterCursorManager::PinnedCursor();
     ASSERT(!isMockCursorKilled(0));
     getManager()->reapZombieCursors(nullptr);
@@ -879,12 +971,14 @@ TEST_F(ClusterCursorManagerTest, PinnedCursorMoveAssignmentKill) {
 TEST_F(ClusterCursorManagerTest, PinnedCursorDestructorKill) {
     {
         auto cursorId =
-            assertGet(getManager()->registerCursor(nullptr,
+            assertGet(getManager()->registerCursor(_opCtx.get(),
                                                    allocateMockCursor(),
                                                    nss,
                                                    ClusterCursorManager::CursorType::SingleTarget,
-                                                   ClusterCursorManager::CursorLifetime::Mortal));
-        auto pinnedCursor = getManager()->checkOutCursor(nss, cursorId, _opCtx.get());
+                                                   ClusterCursorManager::CursorLifetime::Mortal,
+                                                   UserNameIterator()));
+        auto pinnedCursor =
+            getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), successAuthChecker);
     }
     ASSERT(!isMockCursorKilled(0));
     getManager()->reapZombieCursors(nullptr);
@@ -898,12 +992,14 @@ TEST_F(ClusterCursorManagerTest, RemotesExhausted) {
     ASSERT_FALSE(mockCursor->remotesExhausted());
 
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                std::move(mockCursor),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
-    auto pinnedCursor = getManager()->checkOutCursor(nss, cursorId, _opCtx.get());
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
+    auto pinnedCursor =
+        getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), successAuthChecker);
     ASSERT_OK(pinnedCursor.getStatus());
     ASSERT_FALSE(pinnedCursor.getValue().remotesExhausted());
 }
@@ -911,12 +1007,14 @@ TEST_F(ClusterCursorManagerTest, RemotesExhausted) {
 // Test that killed cursors which are still pinned are not reaped.
 TEST_F(ClusterCursorManagerTest, DoNotReapKilledPinnedCursors) {
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
-    auto pinnedCursor = getManager()->checkOutCursor(nss, cursorId, _opCtx.get());
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
+    auto pinnedCursor =
+        getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), successAuthChecker);
     ASSERT_OK(pinnedCursor.getStatus());
     ASSERT_OK(getManager()->killCursor(nss, cursorId));
     ASSERT(!isMockCursorKilled(0));
@@ -933,11 +1031,12 @@ TEST_F(ClusterCursorManagerTest, DoNotReapKilledPinnedCursors) {
 }
 
 TEST_F(ClusterCursorManagerTest, CannotRegisterCursorDuringShutdown) {
-    ASSERT_OK(getManager()->registerCursor(nullptr,
+    ASSERT_OK(getManager()->registerCursor(_opCtx.get(),
                                            allocateMockCursor(),
                                            nss,
                                            ClusterCursorManager::CursorType::SingleTarget,
-                                           ClusterCursorManager::CursorLifetime::Mortal));
+                                           ClusterCursorManager::CursorLifetime::Mortal,
+                                           UserNameIterator()));
     ASSERT(!isMockCursorKilled(0));
 
     getManager()->shutdown(nullptr);
@@ -945,28 +1044,31 @@ TEST_F(ClusterCursorManagerTest, CannotRegisterCursorDuringShutdown) {
     ASSERT(isMockCursorKilled(0));
 
     ASSERT_EQUALS(ErrorCodes::ShutdownInProgress,
-                  getManager()->registerCursor(nullptr,
+                  getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
 }
 
 TEST_F(ClusterCursorManagerTest, CannotCheckoutCursorDuringShutdown) {
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
     ASSERT(!isMockCursorKilled(0));
 
     getManager()->shutdown(nullptr);
 
     ASSERT(isMockCursorKilled(0));
 
-    ASSERT_EQUALS(ErrorCodes::ShutdownInProgress,
-                  getManager()->checkOutCursor(nss, cursorId, _opCtx.get()).getStatus());
+    ASSERT_EQUALS(
+        ErrorCodes::ShutdownInProgress,
+        getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), successAuthChecker).getStatus());
 }
 
 /**
@@ -974,11 +1076,12 @@ TEST_F(ClusterCursorManagerTest, CannotCheckoutCursorDuringShutdown) {
  */
 TEST_F(ClusterCursorManagerTest, CursorsWithoutSessions) {
     // Add a cursor with no session to the cursor manager.
-    ASSERT_OK(getManager()->registerCursor(nullptr,
+    ASSERT_OK(getManager()->registerCursor(_opCtx.get(),
                                            allocateMockCursor(),
                                            nss,
                                            ClusterCursorManager::CursorType::SingleTarget,
-                                           ClusterCursorManager::CursorLifetime::Mortal));
+                                           ClusterCursorManager::CursorLifetime::Mortal,
+                                           UserNameIterator()));
 
     // Manager should have no active sessions.
     LogicalSessionIdSet lsids;
@@ -993,11 +1096,12 @@ TEST_F(ClusterCursorManagerTest, OneCursorWithASession) {
     // Add a cursor with a session to the cursor manager.
     auto lsid = makeLogicalSessionIdForTest();
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(lsid),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
 
     // Retrieve all sessions active in manager - set should contain just lsid.
     LogicalSessionIdSet lsids;
@@ -1027,14 +1131,15 @@ TEST_F(ClusterCursorManagerTest, GetSessionIdsWhileCheckedOut) {
     // Add a cursor with a session to the cursor manager.
     auto lsid = makeLogicalSessionIdForTest();
     auto cursorId =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(lsid),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
 
     // Check the cursor out, then try to append cursors, see that we get one.
-    auto res = getManager()->checkOutCursor(nss, cursorId, _opCtx.get());
+    auto res = getManager()->checkOutCursor(nss, cursorId, _opCtx.get(), successAuthChecker);
     ASSERT(res.isOK());
 
     auto cursors = getManager()->getCursorsForSession(lsid);
@@ -1048,17 +1153,19 @@ TEST_F(ClusterCursorManagerTest, MultipleCursorsWithSameSession) {
     // Add two cursors on the same session to the cursor manager.
     auto lsid = makeLogicalSessionIdForTest();
     auto cursorId1 =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(lsid),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
     auto cursorId2 =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(lsid),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
 
     // Retrieve all sessions - set should contain just lsid.
     stdx::unordered_set<LogicalSessionId, LogicalSessionIdHash> lsids;
@@ -1096,24 +1203,27 @@ TEST_F(ClusterCursorManagerTest, MultipleCursorsMultipleSessions) {
 
     // Register two cursors with different lsids, and one without.
     CursorId cursor1 =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(lsid1),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
 
     CursorId cursor2 =
-        assertGet(getManager()->registerCursor(nullptr,
+        assertGet(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(lsid2),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
 
-    ASSERT_OK(getManager()->registerCursor(nullptr,
+    ASSERT_OK(getManager()->registerCursor(_opCtx.get(),
                                            allocateMockCursor(),
                                            nss,
                                            ClusterCursorManager::CursorType::SingleTarget,
-                                           ClusterCursorManager::CursorLifetime::Mortal));
+                                           ClusterCursorManager::CursorLifetime::Mortal,
+                                           UserNameIterator()));
 
     // Retrieve all sessions - should be both lsids.
     LogicalSessionIdSet lsids;
@@ -1139,11 +1249,12 @@ TEST_F(ClusterCursorManagerTest, ManyCursorsManySessions) {
     const int count = 10000;
     for (int i = 0; i < count; i++) {
         auto lsid = makeLogicalSessionIdForTest();
-        ASSERT_OK(getManager()->registerCursor(nullptr,
+        ASSERT_OK(getManager()->registerCursor(_opCtx.get(),
                                                allocateMockCursor(lsid),
                                                nss,
                                                ClusterCursorManager::CursorType::SingleTarget,
-                                               ClusterCursorManager::CursorLifetime::Mortal));
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
     }
 
     // Retrieve all sessions.
@@ -1152,6 +1263,24 @@ TEST_F(ClusterCursorManagerTest, ManyCursorsManySessions) {
     ASSERT_EQ(lsids.size(), size_t(count));
 }
 
+TEST_F(ClusterCursorManagerTest, CheckAuthForKillCursors) {
+    auto cursorId =
+        assertGet(getManager()->registerCursor(_opCtx.get(),
+                                               allocateMockCursor(),
+                                               nss,
+                                               ClusterCursorManager::CursorType::SingleTarget,
+                                               ClusterCursorManager::CursorLifetime::Mortal,
+                                               UserNameIterator()));
+
+    ASSERT_EQ(
+        ErrorCodes::CursorNotFound,
+        getManager()->checkAuthForKillCursors(_opCtx.get(), nss, cursorId + 1, successAuthChecker));
+    ASSERT_EQ(ErrorCodes::Unauthorized,
+              getManager()->checkAuthForKillCursors(_opCtx.get(), nss, cursorId, failAuthChecker));
+    ASSERT_OK(
+        getManager()->checkAuthForKillCursors(_opCtx.get(), nss, cursorId, successAuthChecker));
+}
+
 }  // namespace
 
 }  // namespace mongo
diff --git a/src/mongo/s/query/cluster_find.cpp b/src/mongo/s/query/cluster_find.cpp
index a2b5b105ab5..d8f27ea170b 100644
--- a/src/mongo/s/query/cluster_find.cpp
+++ b/src/mongo/s/query/cluster_find.cpp
@@ -192,10 +192,7 @@ CursorId runQueryWithoutRetrying(OperationContext* opCtx,
 
     // Construct the query and parameters.
 
-    ClusterClientCursorParams params(
-        query.nss(),
-        AuthorizationSession::get(opCtx->getClient())->getAuthenticatedUserNames(),
-        readPref);
+    ClusterClientCursorParams params(query.nss(), readPref);
     params.limit = query.getQueryRequest().getLimit();
     params.batchSize = query.getQueryRequest().getEffectiveBatchSize();
     params.skip = query.getQueryRequest().getSkip();
@@ -325,8 +322,10 @@ CursorId runQueryWithoutRetrying(OperationContext* opCtx,
     const auto cursorLifetime = query.getQueryRequest().isNoCursorTimeout()
         ? ClusterCursorManager::CursorLifetime::Immortal
         : ClusterCursorManager::CursorLifetime::Mortal;
+    auto authUsers = AuthorizationSession::get(opCtx->getClient())->getAuthenticatedUserNames();
+
     return uassertStatusOK(cursorManager->registerCursor(
-        opCtx, ccc.releaseCursor(), query.nss(), cursorType, cursorLifetime));
+        opCtx, ccc.releaseCursor(), query.nss(), cursorType, cursorLifetime, authUsers));
 }
 
 }  // namespace
@@ -391,7 +390,15 @@ StatusWith<CursorResponse> ClusterFind::runGetMore(OperationContext* opCtx,
                                                    const GetMoreRequest& request) {
     auto cursorManager = Grid::get(opCtx)->getCursorManager();
 
-    auto pinnedCursor = cursorManager->checkOutCursor(request.nss, request.cursorid, opCtx);
+    auto authzSession = AuthorizationSession::get(opCtx->getClient());
+    auto authChecker = [&authzSession](UserNameIterator userNames) -> Status {
+        return authzSession->isCoauthorizedWith(userNames)
+            ? Status::OK()
+            : Status(ErrorCodes::Unauthorized, "User not authorized to access cursor");
+    };
+
+    auto pinnedCursor =
+        cursorManager->checkOutCursor(request.nss, request.cursorid, opCtx, authChecker);
     if (!pinnedCursor.isOK()) {
         return pinnedCursor.getStatus();
     }
@@ -401,16 +408,6 @@ StatusWith<CursorResponse> ClusterFind::runGetMore(OperationContext* opCtx,
     while (MONGO_FAIL_POINT(keepCursorPinnedDuringGetMore)) {
     }
 
-    // A user can only call getMore on their own cursor. If there were multiple users authenticated
-    // when the cursor was created, then at least one of them must be authenticated in order to run
-    // getMore on the cursor.
-    if (!AuthorizationSession::get(opCtx->getClient())
-             ->isCoauthorizedWith(pinnedCursor.getValue().getAuthenticatedUsers())) {
-        return {ErrorCodes::Unauthorized,
-                str::stream() << "cursor id " << request.cursorid
-                              << " was not created by the authenticated user"};
-    }
-
     if (auto readPref = pinnedCursor.getValue().getReadPreference()) {
         ReadPreferenceSetting::get(opCtx) = *readPref;
     }
diff --git a/src/mongo/s/query/store_possible_cursor.cpp b/src/mongo/s/query/store_possible_cursor.cpp
index f611e612d2a..38347a70a9e 100644
--- a/src/mongo/s/query/store_possible_cursor.cpp
+++ b/src/mongo/s/query/store_possible_cursor.cpp
@@ -61,9 +61,7 @@ StatusWith<BSONObj> storePossibleCursor(OperationContext* opCtx,
         return cmdResult;
     }
 
-    ClusterClientCursorParams params(
-        incomingCursorResponse.getValue().getNSS(),
-        AuthorizationSession::get(opCtx->getClient())->getAuthenticatedUserNames());
+    ClusterClientCursorParams params(incomingCursorResponse.getValue().getNSS());
     params.remotes.emplace_back(shardId,
                                 server,
                                 CursorResponse(incomingCursorResponse.getValue().getNSS(),
@@ -76,12 +74,14 @@ StatusWith<BSONObj> storePossibleCursor(OperationContext* opCtx,
     // We don't expect to use this cursor until a subsequent getMore, so detach from the current
     // OperationContext until then.
     ccc->detachFromOperationContext();
+    auto authUsers = AuthorizationSession::get(opCtx->getClient())->getAuthenticatedUserNames();
     auto clusterCursorId =
         cursorManager->registerCursor(opCtx,
                                       ccc.releaseCursor(),
                                       requestedNss,
                                       ClusterCursorManager::CursorType::SingleTarget,
-                                      ClusterCursorManager::CursorLifetime::Mortal);
+                                      ClusterCursorManager::CursorLifetime::Mortal,
+                                      authUsers);
     if (!clusterCursorId.isOK()) {
         return clusterCursorId.getStatus();
     }