SERVER-32978 Advertise SCRAM-SHA-256 authentication for the internal user

45 files changed, 825 insertions, 721 deletions

diff --git a/src/mongo/client/SConscript b/src/mongo/client/SConscript
index f6c04996994..0b76fa83d1b 100644
--- a/src/mongo/client/SConscript
+++ b/src/mongo/client/SConscript
@@ -170,7 +170,6 @@ clientDriverEnv.Library(
         'connection_string',
     ],
     LIBDEPS_PRIVATE=[
-        '$BUILD_DIR/mongo/db/auth/internal_user_auth',
         '$BUILD_DIR/mongo/util/net/ssl_manager',
     ],
 )
@@ -243,7 +242,6 @@ env.Library(
         'authentication',
     ],
     LIBDEPS_PRIVATE=[
-        '$BUILD_DIR/mongo/db/auth/internal_user_auth',
         '$BUILD_DIR/mongo/db/commands/test_commands_enabled',
         '$BUILD_DIR/mongo/executor/egress_tag_closer_manager',
         '$BUILD_DIR/mongo/transport/message_compressor',
@@ -259,9 +257,6 @@ env.Library(
     LIBDEPS=[
         'clientdriver_network',
     ],
-    LIBDEPS_PRIVATE=[
-        '$BUILD_DIR/mongo/db/auth/internal_user_auth',
-    ],
 )
 
 env.Library(
diff --git a/src/mongo/client/async_client.cpp b/src/mongo/client/async_client.cpp
index a7ec71b5dd7..520e69606bc 100644
--- a/src/mongo/client/async_client.cpp
+++ b/src/mongo/client/async_client.cpp
@@ -37,7 +37,6 @@
 #include "mongo/bson/bsonobjbuilder.h"
 #include "mongo/client/authenticate.h"
 #include "mongo/config.h"
-#include "mongo/db/auth/internal_user_auth.h"
 #include "mongo/db/commands/test_commands_enabled.h"
 #include "mongo/db/server_options.h"
 #include "mongo/db/wire_version.h"
@@ -67,11 +66,12 @@ Future<AsyncDBClient::Handle> AsyncDBClient::connect(const HostAndPort& peer,
         });
 }
 
-BSONObj AsyncDBClient::_buildIsMasterRequest(const std::string& appName) {
+BSONObj AsyncDBClient::_buildIsMasterRequest(const std::string& appName,
+                                             executor::NetworkConnectionHook* hook) {
     BSONObjBuilder bob;
 
     bob.append("isMaster", 1);
-    bob.append("hangUpOnStepDown", false);
+
     const auto versionString = VersionInfoInterface::instance().version();
     ClientMetadata::serialize(appName, versionString, &bob);
 
@@ -90,7 +90,11 @@ BSONObj AsyncDBClient::_buildIsMasterRequest(const std::string& appName) {
         WireSpec::appendInternalClientWireVersion(WireSpec::instance().outgoing, &bob);
     }
 
-    return bob.obj();
+    if (hook) {
+        return hook->augmentIsMasterRequest(bob.obj());
+    } else {
+        return bob.obj();
+    }
 }
 
 void AsyncDBClient::_parseIsMasterResponse(BSONObj request,
@@ -127,13 +131,20 @@ void AsyncDBClient::_parseIsMasterResponse(BSONObj request,
     _compressorManager.clientFinish(responseBody);
 }
 
-Future<void> AsyncDBClient::authenticate(const BSONObj& params) {
-    // This check is sufficient to see if auth is enabled on the system,
-    // and avoids creating dependencies on deeper, less accessible auth code.
-    if (!isInternalAuthSet()) {
-        return Future<void>::makeReady();
-    }
+auth::RunCommandHook AsyncDBClient::_makeAuthRunCommandHook() {
+    return [this](OpMsgRequest request) {
+        return runCommand(std::move(request)).then([](rpc::UniqueReply reply) -> Future<BSONObj> {
+            auto status = getStatusFromCommandResult(reply->getCommandReply());
+            if (!status.isOK()) {
+                return status;
+            } else {
+                return reply->getCommandReply();
+            }
+        });
+    };
+}
 
+Future<void> AsyncDBClient::authenticate(const BSONObj& params) {
     // We will only have a valid clientName if SSL is enabled.
     std::string clientName;
 #ifdef MONGO_CONFIG_SSL
@@ -142,37 +153,28 @@ Future<void> AsyncDBClient::authenticate(const BSONObj& params) {
     }
 #endif
 
-    auto pf = makePromiseFuture<void>();
-    auto authCompleteCb = [promise = pf.promise.share()](auth::AuthResponse response) mutable {
-        if (response.isOK()) {
-            promise.emplaceValue();
-        } else {
-            promise.setError(response.status);
-        }
-    };
-
-    auto doAuthCb = [this](executor::RemoteCommandRequest request,
-                           auth::AuthCompletionHandler handler) {
-
-        runCommandRequest(request).getAsync([handler = std::move(handler)](
-            StatusWith<executor::RemoteCommandResponse> response) {
-            if (!response.isOK()) {
-                handler(executor::RemoteCommandResponse(response.getStatus()));
-            } else {
-                handler(std::move(response.getValue()));
-            }
-        });
-    };
+    return auth::authenticateClient(params, remote(), clientName, _makeAuthRunCommandHook());
+}
 
-    auth::authenticateClient(
-        params, remote(), clientName, std::move(doAuthCb), std::move(authCompleteCb));
+Future<void> AsyncDBClient::authenticateInternal(boost::optional<std::string> mechanismHint) {
+    // If no internal auth information is set, don't bother trying to authenticate.
+    if (!auth::isInternalAuthSet()) {
+        return Future<void>::makeReady();
+    }
+    // We will only have a valid clientName if SSL is enabled.
+    std::string clientName;
+#ifdef MONGO_CONFIG_SSL
+    if (getSSLManager()) {
+        clientName = getSSLManager()->getSSLConfiguration().clientSubjectName.toString();
+    }
+#endif
 
-    return std::move(pf.future);
+    return auth::authenticateInternalClient(clientName, mechanismHint, _makeAuthRunCommandHook());
 }
 
 Future<void> AsyncDBClient::initWireVersion(const std::string& appName,
                                             executor::NetworkConnectionHook* const hook) {
-    auto requestObj = _buildIsMasterRequest(appName);
+    auto requestObj = _buildIsMasterRequest(appName, hook);
     // We use a legacy request to create our ismaster request because we may
     // have to communicate with servers that do not support other protocols.
     auto requestMsg =
diff --git a/src/mongo/client/async_client.h b/src/mongo/client/async_client.h
index b02eb5506bc..356f4e0f0ca 100644
--- a/src/mongo/client/async_client.h
+++ b/src/mongo/client/async_client.h
@@ -32,6 +32,7 @@
 
 #include <memory>
 
+#include "mongo/client/authenticate.h"
 #include "mongo/db/service_context.h"
 #include "mongo/executor/network_connection_hook.h"
 #include "mongo/executor/remote_command_request.h"
@@ -67,6 +68,8 @@ public:
 
     Future<void> authenticate(const BSONObj& params);
 
+    Future<void> authenticateInternal(boost::optional<std::string> mechanismHint);
+
     Future<void> initWireVersion(const std::string& appName,
                                  executor::NetworkConnectionHook* const hook);
 
@@ -81,9 +84,11 @@ public:
 
 private:
     Future<Message> _call(Message request, const transport::BatonHandle& baton = nullptr);
-    BSONObj _buildIsMasterRequest(const std::string& appName);
+    BSONObj _buildIsMasterRequest(const std::string& appName,
+                                  executor::NetworkConnectionHook* hook);
     void _parseIsMasterResponse(BSONObj request,
                                 const std::unique_ptr<rpc::ReplyInterface>& response);
+    auth::RunCommandHook _makeAuthRunCommandHook();
 
     const HostAndPort _peer;
     transport::SessionHandle _session;
diff --git a/src/mongo/client/authenticate.cpp b/src/mongo/client/authenticate.cpp
index e32164a9228..6dc0337d627 100644
--- a/src/mongo/client/authenticate.cpp
+++ b/src/mongo/client/authenticate.cpp
@@ -40,9 +40,12 @@
 #include "mongo/bson/util/bson_extract.h"
 #include "mongo/client/sasl_client_authenticate.h"
 #include "mongo/config.h"
+#include "mongo/db/auth/authorization_manager.h"
 #include "mongo/db/auth/sasl_command_constants.h"
 #include "mongo/db/server_options.h"
 #include "mongo/rpc/get_status_from_command_result.h"
+#include "mongo/rpc/op_msg_rpc_impls.h"
+#include "mongo/stdx/mutex.h"
 #include "mongo/util/log.h"
 #include "mongo/util/net/ssl_manager.h"
 #include "mongo/util/net/ssl_options.h"
@@ -56,21 +59,11 @@ using executor::RemoteCommandResponse;
 
 using AuthRequest = StatusWith<RemoteCommandRequest>;
 
-const char* const kMechanismMongoCR = "MONGODB-CR";
-const char* const kMechanismMongoX509 = "MONGODB-X509";
-const char* const kMechanismSaslPlain = "PLAIN";
-const char* const kMechanismGSSAPI = "GSSAPI";
-const char* const kMechanismScramSha1 = "SCRAM-SHA-1";
-
 namespace {
 
 const char* const kUserSourceFieldName = "userSource";
 const BSONObj kGetNonceCmd = BSON("getnonce" << 1);
 
-bool isOk(const BSONObj& o) {
-    return getStatusFromCommandResult(o).isOK();
-}
-
 StatusWith<std::string> extractDBField(const BSONObj& params) {
     std::string db;
     if (params.hasField(kUserSourceFieldName)) {
@@ -90,15 +83,16 @@ StatusWith<std::string> extractDBField(const BSONObj& params) {
 // MONGODB-CR
 //
 
-void authMongoCRImpl(RunCommandHook cmd, const BSONObj& params, AuthCompletionHandler handler) {
-    handler({ErrorCodes::AuthenticationFailed, "MONGODB-CR support was removed in MongoDB 3.8"});
+Future<void> authMongoCRImpl(RunCommandHook cmd, const BSONObj& params) {
+    return Status(ErrorCodes::AuthenticationFailed,
+                  "MONGODB-CR support was removed in MongoDB 4.0");
 }
 
 //
 // X-509
 //
 
-AuthRequest createX509AuthCmd(const BSONObj& params, StringData clientName) {
+StatusWith<OpMsgRequest> createX509AuthCmd(const BSONObj& params, StringData clientName) {
     if (clientName.empty()) {
         return {ErrorCodes::AuthenticationFailed,
                 "Please enable SSL on the client-side to use the MONGODB-X509 authentication "
@@ -108,9 +102,6 @@ AuthRequest createX509AuthCmd(const BSONObj& params, StringData clientName) {
     if (!db.isOK())
         return std::move(db.getStatus());
 
-    auto request = RemoteCommandRequest();
-    request.dbname = db.getValue();
-
     std::string username;
     auto response = bsonExtractStringFieldWithDefault(
         params, saslCommandUserFieldName, clientName.toString(), &username);
@@ -126,115 +117,196 @@ AuthRequest createX509AuthCmd(const BSONObj& params, StringData clientName) {
         return {ErrorCodes::AuthenticationFailed, message.str()};
     }
 
-    request.cmdObj = BSON("authenticate" << 1 << "mechanism"
-                                         << "MONGODB-X509"
-                                         << "user"
-                                         << username);
-    return std::move(request);
+    return OpMsgRequest::fromDBAndBody(db.getValue(),
+                                       BSON("authenticate" << 1 << "mechanism"
+                                                           << "MONGODB-X509"
+                                                           << "user"
+                                                           << username));
 }
 
 // Use the MONGODB-X509 protocol to authenticate as "username." The certificate details
 // have already been communicated automatically as part of the connect call.
-void authX509(RunCommandHook runCommand,
-              const BSONObj& params,
-              StringData clientName,
-              AuthCompletionHandler handler) {
+Future<void> authX509(RunCommandHook runCommand, const BSONObj& params, StringData clientName) {
     invariant(runCommand);
-    invariant(handler);
 
     // Just 1 step: send authenticate command, receive response
     auto authRequest = createX509AuthCmd(params, clientName);
     if (!authRequest.isOK())
-        return handler(std::move(authRequest.getStatus()));
+        return authRequest.getStatus();
 
-    runCommand(authRequest.getValue(), handler);
+    // The runCommand hook checks whether the command returned { ok: 1.0 }, and we don't need to
+    // extract anything from the command payload, so this is just turning a Future<BSONObj>
+    // into a Future<void>
+    return runCommand(authRequest.getValue()).then([](BSONObj obj) { return Status::OK(); });
 }
 
+}  // namespace
 //
 // General Auth
 //
 
-bool isFailedAuthOk(const AuthResponse& response) {
-    return (response.status == ErrorCodes::AuthenticationFailed &&
-            serverGlobalParams.transitionToAuth);
-}
-
-void auth(RunCommandHook runCommand,
-          const BSONObj& params,
-          const HostAndPort& hostname,
-          StringData clientName,
-          AuthCompletionHandler handler) {
+Future<void> authenticateClient(const BSONObj& params,
+                                const HostAndPort& hostname,
+                                const std::string& clientName,
+                                RunCommandHook runCommand) {
     std::string mechanism;
-    auto authCompletionHandler = [handler](AuthResponse response) {
-        if (isFailedAuthOk(response)) {
+    auto errorHandler = [](Status status) {
+        if (serverGlobalParams.transitionToAuth && !status.isA<ErrorCategory::NetworkError>()) {
             // If auth failed in transitionToAuth, just pretend it succeeded.
             log() << "Failed to authenticate in transitionToAuth, falling back to no "
                      "authentication.";
 
-            // We need to mock a successful AuthResponse.
-            return handler(AuthResponse(RemoteCommandResponse(BSON("ok" << 1), Milliseconds(0))));
+            return Status::OK();
         }
 
-        // otherwise, call handler
-        return handler(std::move(response));
+        return status;
     };
     auto response = bsonExtractStringField(params, saslCommandMechanismFieldName, &mechanism);
     if (!response.isOK())
-        return handler(std::move(response));
+        return response;
 
     if (params.hasField(saslCommandUserDBFieldName) && params.hasField(kUserSourceFieldName)) {
-        return handler({ErrorCodes::AuthenticationFailed,
-                        "You cannot specify both 'db' and 'userSource'. Please use only 'db'."});
+        return Status(ErrorCodes::AuthenticationFailed,
+                      "You cannot specify both 'db' and 'userSource'. Please use only 'db'.");
     }
 
     if (mechanism == kMechanismMongoCR)
-        return authMongoCR(runCommand, params, authCompletionHandler);
+        return authMongoCR(runCommand, params).onError(errorHandler);
 
 #ifdef MONGO_CONFIG_SSL
     else if (mechanism == kMechanismMongoX509)
-        return authX509(runCommand, params, clientName, authCompletionHandler);
+        return authX509(runCommand, params, clientName).onError(errorHandler);
 #endif
 
     else if (saslClientAuthenticate != nullptr)
-        return saslClientAuthenticate(runCommand, hostname, params, authCompletionHandler);
+        return saslClientAuthenticate(runCommand, hostname, params).onError(errorHandler);
 
-    return handler({ErrorCodes::AuthenticationFailed,
-                    mechanism + " mechanism support not compiled into client library."});
+    return Status(ErrorCodes::AuthenticationFailed,
+                  mechanism + " mechanism support not compiled into client library.");
 };
 
-void asyncAuth(RunCommandHook runCommand,
-               const BSONObj& params,
-               const HostAndPort& hostname,
-               StringData clientName,
-               AuthCompletionHandler handler) {
-    auth(runCommand, params, hostname, clientName, std::move(handler));
+AuthMongoCRHandler authMongoCR = authMongoCRImpl;
+
+static stdx::mutex internalAuthKeysMutex;
+static bool internalAuthSet = false;
+static std::vector<std::string> internalAuthKeys;
+static BSONObj internalAuthParams;
+
+void setInternalAuthKeys(const std::vector<std::string>& keys) {
+    stdx::lock_guard<stdx::mutex> lk(internalAuthKeysMutex);
+
+    internalAuthKeys = keys;
+    fassert(50996, internalAuthKeys.size() > 0);
+    internalAuthSet = true;
 }
 
-}  // namespace
+void setInternalUserAuthParams(BSONObj obj) {
+    stdx::lock_guard<stdx::mutex> lk(internalAuthKeysMutex);
+    internalAuthParams = obj.getOwned();
+    internalAuthKeys.clear();
+    internalAuthSet = true;
+}
 
-AuthMongoCRHandler authMongoCR = authMongoCRImpl;
+bool hasMultipleInternalAuthKeys() {
+    stdx::lock_guard<stdx::mutex> lk(internalAuthKeysMutex);
+    return internalAuthSet && internalAuthKeys.size() > 1;
+}
 
-void authenticateClient(const BSONObj& params,
-                        const HostAndPort& hostname,
-                        StringData clientName,
-                        RunCommandHook runCommand,
-                        AuthCompletionHandler handler) {
-    if (handler) {
-        // Run asynchronously
-        return asyncAuth(std::move(runCommand), params, hostname, clientName, std::move(handler));
-    } else {
-        // Run synchronously through async framework
-        // NOTE: this assumes that runCommand executes synchronously.
-        asyncAuth(runCommand, params, hostname, clientName, [](AuthResponse response) {
-            // DBClient expects us to throw in case of an auth error.
-            uassertStatusOK(response.status);
-
-            auto serverResponse = response.data;
-            uassert(ErrorCodes::AuthenticationFailed,
-                    serverResponse["errmsg"].str(),
-                    isOk(serverResponse));
-        });
+bool isInternalAuthSet() {
+    stdx::lock_guard<stdx::mutex> lk(internalAuthKeysMutex);
+    return internalAuthSet;
+}
+
+BSONObj getInternalAuthParams(size_t idx, const std::string& mechanism) {
+    stdx::lock_guard<stdx::mutex> lk(internalAuthKeysMutex);
+    if (!internalAuthSet) {
+        return BSONObj();
+    }
+
+    // If we've set a specific BSONObj as the internal auth pararms, return it if the index
+    // is zero (there are no alternate credentials if we've set a BSONObj explicitly).
+    if (!internalAuthParams.isEmpty()) {
+        return idx == 0 ? internalAuthParams : BSONObj();
+    }
+
+    // If the index is larger than the number of keys we know about then return an empty
+    // BSONObj.
+    if (idx + 1 > internalAuthKeys.size()) {
+        return BSONObj();
     }
+
+    auto password = internalAuthKeys.at(idx);
+    if (mechanism == kMechanismScramSha1) {
+        password = mongo::createPasswordDigest(
+            internalSecurity.user->getName().getUser().toString(), password);
+    }
+
+    return BSON(saslCommandMechanismFieldName << mechanism << saslCommandUserDBFieldName
+                                              << internalSecurity.user->getName().getDB()
+                                              << saslCommandUserFieldName
+                                              << internalSecurity.user->getName().getUser()
+                                              << saslCommandPasswordFieldName
+                                              << password
+                                              << saslCommandDigestPasswordFieldName
+                                              << false);
+}
+
+Future<std::string> negotiateSaslMechanism(RunCommandHook runCommand,
+                                           const UserName& username,
+                                           boost::optional<std::string> mechanismHint) {
+    if (mechanismHint && !mechanismHint->empty()) {
+        return Future<std::string>::makeReady(*mechanismHint);
+    }
+
+    const auto request =
+        BSON("ismaster" << 1 << "saslSupportedMechs" << username.getUnambiguousName());
+    return runCommand(OpMsgRequest::fromDBAndBody("admin"_sd, std::move(request)))
+        .then([](BSONObj reply) -> Future<std::string> {
+            auto mechsArrayObj = reply.getField("saslSupportedMechs");
+            if (mechsArrayObj.type() != Array) {
+                return Status{ErrorCodes::BadValue, "Expected array of SASL mechanism names"};
+            }
+
+            auto obj = mechsArrayObj.Obj();
+            std::vector<std::string> availableMechanisms;
+            for (const auto elem : obj) {
+                if (elem.type() != String) {
+                    return Status{ErrorCodes::BadValue, "Expected array of SASL mechanism names"};
+                }
+                availableMechanisms.push_back(elem.checkAndGetStringData().toString());
+                // The drivers spec says that if SHA-256 is available then it MUST be selected
+                // as the SASL mech.
+                if (availableMechanisms.back() == kMechanismScramSha256) {
+                    return availableMechanisms.back();
+                }
+            }
+
+            return availableMechanisms.empty() ? kInternalAuthFallbackMechanism.toString()
+                                               : availableMechanisms.front();
+        });
+}
+
+Future<void> authenticateInternalClient(const std::string& clientSubjectName,
+                                        boost::optional<std::string> mechanismHint,
+                                        RunCommandHook runCommand) {
+    return negotiateSaslMechanism(runCommand, internalSecurity.user->getName(), mechanismHint)
+        .then([runCommand, clientSubjectName](std::string mechanism) -> Future<void> {
+            auto params = getInternalAuthParams(0, mechanism);
+            if (params.isEmpty()) {
+                return Status(ErrorCodes::BadValue,
+                              "Missing authentication parameters for internal user auth");
+            }
+            return authenticateClient(params, HostAndPort(), clientSubjectName, runCommand)
+                .onError<ErrorCodes::AuthenticationFailed>(
+                    [runCommand, clientSubjectName, mechanism](Status status) -> Future<void> {
+                        auto altCreds = getInternalAuthParams(1, mechanism);
+                        if (!altCreds.isEmpty()) {
+                            return authenticateClient(
+                                altCreds, HostAndPort(), clientSubjectName, runCommand);
+                        }
+                        return status;
+                    });
+        });
 }
 
 BSONObj buildAuthParams(StringData dbname,
diff --git a/src/mongo/client/authenticate.h b/src/mongo/client/authenticate.h
index 846b2659fc6..a2f42eab052 100644
--- a/src/mongo/client/authenticate.h
+++ b/src/mongo/client/authenticate.h
@@ -34,10 +34,14 @@
 
 #include "mongo/base/status_with.h"
 #include "mongo/base/string_data.h"
-#include "mongo/executor/remote_command_request.h"
+#include "mongo/bson/bsonobj.h"
+#include "mongo/db/auth/user_name.h"
 #include "mongo/executor/remote_command_response.h"
+#include "mongo/rpc/op_msg.h"
 #include "mongo/stdx/functional.h"
+#include "mongo/util/future.h"
 #include "mongo/util/md5.h"
+#include "mongo/util/net/hostandport.h"
 
 namespace mongo {
 
@@ -45,25 +49,23 @@ class BSONObj;
 
 namespace auth {
 
-using AuthResponse = executor::RemoteCommandResponse;
-using AuthCompletionHandler = stdx::function<void(AuthResponse)>;
-using RunCommandResultHandler = AuthCompletionHandler;
-using RunCommandHook =
-    stdx::function<void(executor::RemoteCommandRequest, RunCommandResultHandler)>;
+using RunCommandHook = stdx::function<Future<BSONObj>(OpMsgRequest request)>;
 
 /* Hook for legacy MONGODB-CR support provided by shell client only */
-using AuthMongoCRHandler =
-    stdx::function<void(RunCommandHook, const BSONObj&, AuthCompletionHandler)>;
+using AuthMongoCRHandler = stdx::function<Future<void>(RunCommandHook, const BSONObj&)>;
 extern AuthMongoCRHandler authMongoCR;
 
 /**
  * Names for supported authentication mechanisms.
  */
 
-extern const char* const kMechanismMongoX509;
-extern const char* const kMechanismSaslPlain;
-extern const char* const kMechanismGSSAPI;
-extern const char* const kMechanismScramSha1;
+constexpr auto kMechanismMongoCR = "MONGODB-CR"_sd;
+constexpr auto kMechanismMongoX509 = "MONGODB-X509"_sd;
+constexpr auto kMechanismSaslPlain = "PLAIN"_sd;
+constexpr auto kMechanismGSSAPI = "GSSAPI"_sd;
+constexpr auto kMechanismScramSha1 = "SCRAM-SHA-1"_sd;
+constexpr auto kMechanismScramSha256 = "SCRAM-SHA-256"_sd;
+constexpr auto kInternalAuthFallbackMechanism = kMechanismScramSha1;
 
 /**
  * Authenticate a user.
@@ -89,19 +91,51 @@ extern const char* const kMechanismScramSha1;
  * Other fields in "params" are silently ignored. A "params" object can be constructed
  * using the buildAuthParams() method.
  *
- * If a "handler" is provided, this call will execute asynchronously and "handler" will be
- * invoked when authentication has completed.  If no handler is provided, authenticateClient
- * will run synchronously.
+ * This function will return a future that will be filled with the final result of the
+ * authentication command on success or a Status on error.
+ */
+Future<void> authenticateClient(const BSONObj& params,
+                                const HostAndPort& hostname,
+                                const std::string& clientSubjectName,
+                                RunCommandHook runCommand);
+
+/**
+ * Authenticate as the __system user. All parameters are the same as authenticateClient above,
+ * but the __system user's credentials will be filled in automatically.
  *
- * Returns normally on success, and throws on error.  Throws a DBException with getCode() ==
- * ErrorCodes::AuthenticationFailed if authentication is rejected.  All other exceptions are
- * tantamount to authentication failure, but may also indicate more serious problems.
+ * The "mechanismHint" parameter will force authentication with a specific mechanism
+ * (e.g. SCRAM-SHA-256). If it is boost::none, then an isMaster will be called to negotiate
+ * a SASL mechanism with the server.
+ *
+ * Because this may retry during cluster keyfile rollover, this may call the RunCommandHook more
+ * than once, but will only call the AuthCompletionHandler once.
+ */
+Future<void> authenticateInternalClient(const std::string& clientSubjectName,
+                                        boost::optional<std::string> mechanismHint,
+                                        RunCommandHook runCommand);
+
+/**
+ * Sets the keys used by authenticateInternalClient - these should be a vector of raw passwords,
+ * they will be digested and prepped appropriately by authenticateInternalClient depending
+ * on what mechanism is used.
+ */
+void setInternalAuthKeys(const std::vector<std::string>& keys);
+
+/**
+ * Sets the parameters for non-password based internal authentication.
  */
-void authenticateClient(const BSONObj& params,
-                        const HostAndPort& hostname,
-                        StringData clientSubjectName,
-                        RunCommandHook runCommand,
-                        AuthCompletionHandler handler = AuthCompletionHandler());
+void setInternalUserAuthParams(BSONObj obj);
+
+/**
+ * Returns whether there are multiple keys that will be tried while authenticating an internal
+ * client (used for logging a startup warning).
+ */
+bool hasMultipleInternalAuthKeys();
+
+/**
+ * Returns whether there are any internal auth data set.
+ */
+bool isInternalAuthSet();
 
 /**
  * Build a BSONObject representing parameters to be passed to authenticateClient(). Takes
@@ -118,6 +152,13 @@ BSONObj buildAuthParams(StringData dbname,
                         bool digestPassword);
 
 /**
+ * Run an isMaster exchange to negotiate a SASL mechanism for authentication.
+ */
+Future<std::string> negotiateSaslMechanism(RunCommandHook runCommand,
+                                           const UserName& username,
+                                           boost::optional<std::string> mechanismHint);
+
+/**
  * Return the field name for the database containing credential information.
  */
 StringData getSaslCommandUserDBFieldName();
diff --git a/src/mongo/client/authenticate_test.cpp b/src/mongo/client/authenticate_test.cpp
index 0916770b134..c036d0173d8 100644
--- a/src/mongo/client/authenticate_test.cpp
+++ b/src/mongo/client/authenticate_test.cpp
@@ -45,10 +45,6 @@
 namespace {
 
 using namespace mongo;
-using executor::RemoteCommandRequest;
-using executor::RemoteCommandResponse;
-
-using auth::RunCommandResultHandler;
 
 /**
  * Utility class to support tests in this file.  Allows caller to load
@@ -65,9 +61,8 @@ public:
           _nonce("7ca422a24f326f2a"),
           _requests(),
           _responses() {
-        _runCommandCallback = [this](RemoteCommandRequest request,
-                                     RunCommandResultHandler handler) {
-            runCommand(std::move(request), handler);
+        _runCommandCallback = [this](OpMsgRequest request) {
+            return runCommand(std::move(request));
         };
 
         // create our digest
@@ -84,18 +79,19 @@ public:
     }
 
     // protected:
-    void runCommand(RemoteCommandRequest request, RunCommandResultHandler handler) {
+    Future<BSONObj> runCommand(OpMsgRequest request) {
         // Validate the received request
         ASSERT(!_requests.empty());
-        RemoteCommandRequest expected = _requests.front();
-        ASSERT(expected.dbname == request.dbname);
-        ASSERT_BSONOBJ_EQ(expected.cmdObj, request.cmdObj);
+        auto& expected = _requests.front();
+        ASSERT_EQ(expected.getDatabase(), request.getDatabase());
+        ASSERT_BSONOBJ_EQ(expected.body, request.body);
         _requests.pop();
 
         // Then pop a response and call the handler
         ASSERT(!_responses.empty());
-        handler(_responses.front());
+        auto ret = _responses.front();
         _responses.pop();
+        return ret;
     }
 
     void reset() {
@@ -105,11 +101,11 @@ public:
     }
 
     void pushResponse(const BSONObj& cmd) {
-        _responses.emplace(cmd, _millis);
+        _responses.emplace(cmd);
     }
 
     void pushRequest(StringData dbname, const BSONObj& cmd) {
-        _requests.emplace(_mockHost, dbname.toString(), cmd, nullptr);
+        _requests.emplace(OpMsgRequest::fromDBAndBody(dbname, cmd));
     }
 
     BSONObj loadMongoCRConversation() {
@@ -175,8 +171,8 @@ public:
     std::string _digest;
     std::string _nonce;
 
-    std::queue<RemoteCommandRequest> _requests;
-    std::queue<RemoteCommandResponse> _responses;
+    std::queue<OpMsgRequest> _requests;
+    std::queue<BSONObj> _responses;
 };
 
 TEST_F(AuthClientTest, MongoCR) {
@@ -185,7 +181,7 @@ TEST_F(AuthClientTest, MongoCR) {
     // jstests exist to ensure MONGODB-CR continues to work from the client.
     auto params = loadMongoCRConversation();
     ASSERT_THROWS(
-        auth::authenticateClient(std::move(params), HostAndPort(), "", _runCommandCallback),
+        auth::authenticateClient(std::move(params), HostAndPort(), "", _runCommandCallback).get(),
         DBException);
 }
 
@@ -193,26 +189,23 @@ TEST_F(AuthClientTest, asyncMongoCR) {
     // As with the sync version above, we expect authentication to fail
     // since this test was built without MONGODB-CR support.
     auto params = loadMongoCRConversation();
-    auth::authenticateClient(std::move(params),
-                             HostAndPort(),
-                             "",
-                             _runCommandCallback,
-                             [this](auth::AuthResponse response) { ASSERT(!response.isOK()); });
+    ASSERT_NOT_OK(
+        auth::authenticateClient(std::move(params), HostAndPort(), "", _runCommandCallback)
+            .getNoThrow());
 }
 
 #ifdef MONGO_CONFIG_SSL
 TEST_F(AuthClientTest, X509) {
     auto params = loadX509Conversation();
-    auth::authenticateClient(std::move(params), HostAndPort(), _username, _runCommandCallback);
+    auth::authenticateClient(std::move(params), HostAndPort(), _username, _runCommandCallback)
+        .get();
 }
 
 TEST_F(AuthClientTest, asyncX509) {
     auto params = loadX509Conversation();
-    auth::authenticateClient(std::move(params),
-                             HostAndPort(),
-                             _username,
-                             _runCommandCallback,
-                             [this](auth::AuthResponse response) { ASSERT(response.isOK()); });
+    ASSERT_OK(
+        auth::authenticateClient(std::move(params), HostAndPort(), _username, _runCommandCallback)
+            .getNoThrow());
 }
 #endif
 
diff --git a/src/mongo/client/connection_pool.cpp b/src/mongo/client/connection_pool.cpp
index ae04449ebcb..09d4e2faf43 100644
--- a/src/mongo/client/connection_pool.cpp
+++ b/src/mongo/client/connection_pool.cpp
@@ -32,9 +32,9 @@
 
 #include "mongo/client/connection_pool.h"
 
+#include "mongo/client/authenticate.h"
 #include "mongo/client/connpool.h"
 #include "mongo/client/mongo_uri.h"
-#include "mongo/db/auth/internal_user_auth.h"
 #include "mongo/executor/network_connection_hook.h"
 #include "mongo/executor/remote_command_request.h"
 #include "mongo/executor/remote_command_response.h"
@@ -192,8 +192,8 @@ ConnectionPool::ConnectionList::iterator ConnectionPool::acquireConnection(
     uassertStatusOK(conn->connect(target, StringData()));
     conn->setTags(_messagingPortTags);
 
-    if (isInternalAuthSet()) {
-        conn->auth(getInternalUserAuthParams());
+    if (auth::isInternalAuthSet()) {
+        uassertStatusOK(conn->authenticateInternalUser());
     }
 
     if (_hook) {
diff --git a/src/mongo/client/dbclient_base.cpp b/src/mongo/client/dbclient_base.cpp
index ef23af9aebe..0b53b56bf06 100644
--- a/src/mongo/client/dbclient_base.cpp
+++ b/src/mongo/client/dbclient_base.cpp
@@ -47,7 +47,7 @@
 #include "mongo/client/constants.h"
 #include "mongo/client/dbclient_cursor.h"
 #include "mongo/config.h"
-#include "mongo/db/auth/internal_user_auth.h"
+#include "mongo/db/auth/authorization_manager.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/json.h"
 #include "mongo/db/namespace_string.h"
@@ -449,6 +449,21 @@ private:
 };
 }  // namespace
 
+auth::RunCommandHook DBClientBase::_makeAuthRunCommandHook() {
+    return [this](OpMsgRequest request) -> Future<BSONObj> {
+        try {
+            auto ret = runCommand(std::move(request));
+            auto status = getStatusFromCommandResult(ret->getCommandReply());
+            if (!status.isOK()) {
+                return status;
+            }
+            return Future<BSONObj>::makeReady(std::move(ret->getCommandReply()));
+        } catch (const DBException& e) {
+            return Future<BSONObj>::makeReady(e.toStatus());
+        }
+    };
+}
+
 void DBClientBase::_auth(const BSONObj& params) {
     ScopedMetadataWriterRemover remover{this};
 
@@ -460,71 +475,46 @@ void DBClientBase::_auth(const BSONObj& params) {
     }
 #endif
 
-    auth::authenticateClient(
-        params,
-        HostAndPort(getServerAddress()),
-        clientName,
-        [this](RemoteCommandRequest request, auth::AuthCompletionHandler handler) {
-            BSONObj info;
-            auto start = Date_t::now();
-
-            try {
-                auto reply = runCommand(
-                    OpMsgRequest::fromDBAndBody(request.dbname, request.cmdObj, request.metadata));
-
-                BSONObj data = reply->getCommandReply().getOwned();
-                Milliseconds millis(Date_t::now() - start);
-
-                // Hand control back to authenticateClient()
-                handler({data, millis});
-
-            } catch (...) {
-                handler(exceptionToStatus());
-            }
-        });
+    HostAndPort remote(getServerAddress());
+    auth::authenticateClient(params, remote, clientName, _makeAuthRunCommandHook())
+        .onError([](Status status) {
+            // for some reason, DBClient transformed all errors into AuthenticationFailed errors
+            return Status(ErrorCodes::AuthenticationFailed, status.reason());
+        })
+        .get();
 }
 
-bool DBClientBase::authenticateInternalUser() {
-    if (!isInternalAuthSet()) {
+Status DBClientBase::authenticateInternalUser() {
+    ScopedMetadataWriterRemover remover{this};
+    if (!auth::isInternalAuthSet()) {
         if (!serverGlobalParams.quiet.load()) {
             log() << "ERROR: No authentication parameters set for internal user";
         }
-        return false;
+        return {ErrorCodes::AuthenticationFailed,
+                "No authentication parameters set for internal user"};
     }
 
-    Status authStatus(ErrorCodes::InternalError, "Status was not set after authentication");
-    auto attemptAuth = [&](const BSONObj& params) {
-        if (params.isEmpty()) {
-            return;
-        }
-
-        try {
-            auth(params);
-            authStatus = Status::OK();
-        } catch (const AssertionException& ex) {
-            authStatus = ex.toStatus();
-        }
-    };
-
-    // First we attempt to authenticate with the default authentication parameters.
-    attemptAuth(getInternalUserAuthParams());
-
-    // If we're in the middle of keyfile rollover, we try to authenticate again with the alternate
-    // credentials in the keyfile.
-    if (authStatus == ErrorCodes::AuthenticationFailed) {
-        attemptAuth(getInternalUserAuthParams(1));
+    // We will only have a client name if SSL is enabled
+    std::string clientName = "";
+#ifdef MONGO_CONFIG_SSL
+    if (sslManager() != nullptr) {
+        clientName = sslManager()->getSSLConfiguration().clientSubjectName.toString();
     }
+#endif
 
-    if (authStatus.isOK()) {
-        return true;
+    auto status =
+        auth::authenticateInternalClient(clientName, boost::none, _makeAuthRunCommandHook())
+            .getNoThrow();
+    if (status.isOK()) {
+        return status;
     }
 
     if (serverGlobalParams.quiet.load()) {
         log() << "can't authenticate to " << toString()
-              << " as internal user, error: " << authStatus.reason();
+              << " as internal user, error: " << status.reason();
     }
 
-    return false;
+    return status;
 }
 
 void DBClientBase::auth(const BSONObj& params) {
diff --git a/src/mongo/client/dbclient_base.h b/src/mongo/client/dbclient_base.h
index ed9074e5491..6f0458e95ee 100644
--- a/src/mongo/client/dbclient_base.h
+++ b/src/mongo/client/dbclient_base.h
@@ -33,6 +33,7 @@
 #include <cstdint>
 
 #include "mongo/base/string_data.h"
+#include "mongo/client/authenticate.h"
 #include "mongo/client/connection_string.h"
 #include "mongo/client/dbclient_cursor.h"
 #include "mongo/client/index_spec.h"
@@ -293,10 +294,9 @@ public:
 
     /**
     * Authenticates to another cluster member using appropriate authentication data.
-    * Uses getInternalUserAuthParams() to retrive authentication parameters.
-    * @return true if the authentication was succesful
+    * @return true if the authentication was successful
     */
-    bool authenticateInternalUser();
+    virtual Status authenticateInternalUser();
 
     /**
      * Authenticate a user.
@@ -714,6 +714,8 @@ protected:
     long long _connectionId;  // unique connection id for this connection
 
 private:
+    auth::RunCommandHook _makeAuthRunCommandHook();
+
     /**
      * The rpc protocols this client supports.
      *
diff --git a/src/mongo/client/dbclient_connection.cpp b/src/mongo/client/dbclient_connection.cpp
index 4e6170588e8..65d267a30d1 100644
--- a/src/mongo/client/dbclient_connection.cpp
+++ b/src/mongo/client/dbclient_connection.cpp
@@ -48,7 +48,6 @@
 #include "mongo/client/dbclient_cursor.h"
 #include "mongo/client/replica_set_monitor.h"
 #include "mongo/config.h"
-#include "mongo/db/auth/internal_user_auth.h"
 #include "mongo/db/client.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/commands/test_commands_enabled.h"
@@ -176,6 +175,14 @@ void DBClientConnection::_auth(const BSONObj& params) {
     DBClientBase::_auth(params);
 }
 
+Status DBClientConnection::authenticateInternalUser() {
+    if (autoReconnect) {
+        _internalAuthOnReconnect = true;
+    }
+
+    return DBClientBase::authenticateInternalUser();
+}
+
 bool DBClientConnection::connect(const HostAndPort& server,
                                  StringData applicationName,
                                  std::string& errmsg) {
@@ -341,6 +348,7 @@ Status DBClientConnection::connectSocketOnly(const HostAndPort& serverAddress) {
 
 void DBClientConnection::logout(const string& dbname, BSONObj& info) {
     authCache.erase(dbname);
+    _internalAuthOnReconnect = false;
     runCommand(dbname, BSON("logout" << 1), info);
 }
 
@@ -470,16 +478,18 @@ void DBClientConnection::_checkConnection() {
     }
 
     LOG(_logLevel) << "reconnect " << toString() << " ok" << endl;
-    for (map<string, BSONObj>::const_iterator i = authCache.begin(); i != authCache.end(); i++) {
-        try {
-            DBClientConnection::_auth(i->second);
-        } catch (AssertionException& ex) {
-            if (ex.code() != ErrorCodes::AuthenticationFailed)
-                throw;
-            LOG(_logLevel) << "reconnect: auth failed "
-                           << i->second[auth::getSaslCommandUserDBFieldName()]
-                           << i->second[auth::getSaslCommandUserFieldName()] << ' ' << ex.what()
-                           << std::endl;
+    if (_internalAuthOnReconnect) {
+        uassertStatusOK(authenticateInternalUser());
+    } else {
+        for (const auto& kv : authCache) {
+            try {
+                DBClientConnection::_auth(kv.second);
+            } catch (ExceptionFor<ErrorCodes::AuthenticationFailed>& ex) {
+                LOG(_logLevel) << "reconnect: auth failed "
+                               << kv.second[auth::getSaslCommandUserDBFieldName()]
+                               << kv.second[auth::getSaslCommandUserFieldName()] << ' ' << ex.what()
+                               << std::endl;
+            }
         }
     }
 }
diff --git a/src/mongo/client/dbclient_connection.h b/src/mongo/client/dbclient_connection.h
index c218d739de6..ddadf5c0230 100644
--- a/src/mongo/client/dbclient_connection.h
+++ b/src/mongo/client/dbclient_connection.h
@@ -280,6 +280,8 @@ public:
         return _isMongos;
     }
 
+    Status authenticateInternalUser() override;
+
 protected:
     int _minWireVersion{0};
     int _maxWireVersion{0};
@@ -310,6 +312,7 @@ protected:
 
     void _checkConnection();
 
+    bool _internalAuthOnReconnect = false;
     std::map<std::string, BSONObj> authCache;
 
     static AtomicInt32 _numConnections;
diff --git a/src/mongo/client/dbclient_rs.cpp b/src/mongo/client/dbclient_rs.cpp
index 065fbf0a1f4..9eafdc8e9cf 100644
--- a/src/mongo/client/dbclient_rs.cpp
+++ b/src/mongo/client/dbclient_rs.cpp
@@ -361,6 +361,14 @@ bool DBClientReplicaSet::checkLastHost(const ReadPreferenceSetting* readPref) {
 }
 
 void DBClientReplicaSet::_authConnection(DBClientConnection* conn) {
+    if (_internalAuthRequested) {
+        auto status = conn->authenticateInternalUser();
+        if (!status.isOK()) {
+            warning() << "cached auth failed for set " << _setName << ": " << status;
+        }
+        return;
+    }
+
     for (map<string, BSONObj>::const_iterator i = _auths.begin(); i != _auths.end(); ++i) {
         try {
             conn->auth(i->second);
@@ -373,6 +381,7 @@ void DBClientReplicaSet::_authConnection(DBClientConnection* conn) {
 }
 
 void DBClientReplicaSet::logoutAll(DBClientConnection* conn) {
+    _internalAuthRequested = false;
     for (map<string, BSONObj>::const_iterator i = _auths.begin(); i != _auths.end(); ++i) {
         BSONObj response;
         try {
@@ -406,33 +415,26 @@ bool DBClientReplicaSet::connect() {
     return _getMonitor()->getHostOrRefresh(anyUpHost).getNoThrow().isOK();
 }
 
-static bool isAuthenticationException(const DBException& ex) {
-    return ex.code() == ErrorCodes::AuthenticationFailed;
-}
-
-void DBClientReplicaSet::_auth(const BSONObj& params) {
+template <typename Authenticate>
+Status DBClientReplicaSet::_runAuthLoop(Authenticate authCb) {
     // We prefer to authenticate against a primary, but otherwise a secondary is ok too
     // Empty tag matches every secondary
-    shared_ptr<ReadPreferenceSetting> readPref(
-        new ReadPreferenceSetting(ReadPreference::PrimaryPreferred, TagSet()));
+    const auto readPref =
+        std::make_shared<ReadPreferenceSetting>(ReadPreference::PrimaryPreferred, TagSet());
 
-    LOG(3) << "dbclient_rs authentication of " << _getMonitor()->getName() << endl;
+    LOG(3) << "dbclient_rs authentication of " << _getMonitor()->getName();
 
     // NOTE that we retry MAX_RETRY + 1 times, since we're always primary preferred we don't
     // fallback to the primary.
     Status lastNodeStatus = Status::OK();
     for (size_t retry = 0; retry < MAX_RETRY + 1; retry++) {
         try {
-            DBClientConnection* conn = selectNodeUsingTags(readPref);
-
-            if (conn == NULL) {
+            auto conn = selectNodeUsingTags(readPref);
+            if (conn == nullptr) {
                 break;
             }
 
-            conn->auth(params);
-
-            // Cache the new auth information since we now validated it's good
-            _auths[params[saslCommandUserDBFieldName].str()] = params.getOwned();
+            authCb(conn);
 
             // Ensure the only child connection open is the one we authenticated against - other
             // child connections may not have full authentication information.
@@ -445,27 +447,46 @@ void DBClientReplicaSet::_auth(const BSONObj& params) {
                 resetMaster();
             }
 
-            return;
-        } catch (const DBException& ex) {
-            // We care if we can't authenticate (i.e. bad password) in credential params.
-            if (isAuthenticationException(ex)) {
-                throw;
+            return Status::OK();
+        } catch (const DBException& e) {
+            auto status = e.toStatus();
+            if (status == ErrorCodes::AuthenticationFailed) {
+                return status;
             }
 
             lastNodeStatus =
-                ex.toStatus(str::stream() << "can't authenticate against replica set node "
-                                          << _lastSlaveOkHost);
+                status.withContext(str::stream() << "can't authenticate against replica set node "
+                                                 << _lastSlaveOkHost);
             _invalidateLastSlaveOkCache(lastNodeStatus);
         }
     }
 
     if (lastNodeStatus.isOK()) {
-        StringBuilder assertMsgB;
-        assertMsgB << "Failed to authenticate, no good nodes in " << _getMonitor()->getName();
-        uasserted(ErrorCodes::HostNotFound, assertMsgB.str());
+        return Status(ErrorCodes::HostNotFound,
+                      str::stream() << "Failed to authenticate, no good nodes in "
+                                    << _getMonitor()->getName());
     } else {
-        uassertStatusOK(lastNodeStatus);
+        return lastNodeStatus;
+    }
+}
+
+Status DBClientReplicaSet::authenticateInternalUser() {
+    if (!auth::isInternalAuthSet()) {
+        return {ErrorCodes::AuthenticationFailed,
+                "No authentication parameters set for internal user"};
     }
+
+    _internalAuthRequested = true;
+    return _runAuthLoop(
+        [&](DBClientConnection* conn) { uassertStatusOK(conn->authenticateInternalUser()); });
+}
+
+void DBClientReplicaSet::_auth(const BSONObj& params) {
+    uassertStatusOK(_runAuthLoop([&](DBClientConnection* conn) {
+        conn->auth(params);
+        // Cache the new auth information since we now validated it's good
+        _auths[params[saslCommandUserDBFieldName].str()] = params.getOwned();
+    }));
 }
 
 void DBClientReplicaSet::logout(const string& dbname, BSONObj& info) {
diff --git a/src/mongo/client/dbclient_rs.h b/src/mongo/client/dbclient_rs.h
index ac26b43659a..bf6dbe84172 100644
--- a/src/mongo/client/dbclient_rs.h
+++ b/src/mongo/client/dbclient_rs.h
@@ -74,6 +74,8 @@ public:
      */
     bool connect();
 
+    Status authenticateInternalUser() override;
+
     /**
      * Logs out the connection for the given database.
      *
@@ -256,6 +258,9 @@ private:
 
     DBClientConnection* checkMaster();
 
+    template <typename Authenticate>
+    Status _runAuthLoop(Authenticate authCb);
+
     /**
      * Helper method for selecting a node based on the read preference. Will advance
      * the tag tags object if it cannot find a node that matches the current tag.
@@ -328,6 +333,7 @@ private:
     // we can re-auth
     // this could be a security issue, as the password is stored in memory
     // not sure if/how we should handle
+    bool _internalAuthRequested = false;
     std::map<std::string, BSONObj> _auths;  // dbName -> auth parameters
 
     MongoURI _uri;
diff --git a/src/mongo/client/sasl_client_authenticate.cpp b/src/mongo/client/sasl_client_authenticate.cpp
index 9cb84f02cda..6cdf55e0bec 100644
--- a/src/mongo/client/sasl_client_authenticate.cpp
+++ b/src/mongo/client/sasl_client_authenticate.cpp
@@ -42,10 +42,9 @@ namespace mongo {
 
 using namespace mongoutils;
 
-void (*saslClientAuthenticate)(auth::RunCommandHook runCommand,
-                               const HostAndPort& hostname,
-                               const BSONObj& saslParameters,
-                               auth::AuthCompletionHandler handler) = nullptr;
+Future<void> (*saslClientAuthenticate)(auth::RunCommandHook runCommand,
+                                       const HostAndPort& hostname,
+                                       const BSONObj& saslParameters) = nullptr;
 
 Status saslExtractPayload(const BSONObj& cmdObj, std::string* payload, BSONType* type) {
     BSONElement payloadElement;
diff --git a/src/mongo/client/sasl_client_authenticate.h b/src/mongo/client/sasl_client_authenticate.h
index 3d4401bd5c9..fb497514382 100644
--- a/src/mongo/client/sasl_client_authenticate.h
+++ b/src/mongo/client/sasl_client_authenticate.h
@@ -70,10 +70,9 @@ class BSONObj;
  * rejected.  Other failures, all of which are tantamount to authentication failure, may also be
  * returned.
  */
-extern void (*saslClientAuthenticate)(auth::RunCommandHook runCommand,
-                                      const HostAndPort& hostname,
-                                      const BSONObj& saslParameters,
-                                      auth::AuthCompletionHandler handler);
+extern Future<void> (*saslClientAuthenticate)(auth::RunCommandHook runCommand,
+                                              const HostAndPort& hostname,
+                                              const BSONObj& saslParameters);
 
 /**
  * Extracts the payload field from "cmdObj", and store it into "*payload".
diff --git a/src/mongo/client/sasl_client_authenticate_impl.cpp b/src/mongo/client/sasl_client_authenticate_impl.cpp
index 6623966f612..2940e0eb809 100644
--- a/src/mongo/client/sasl_client_authenticate_impl.cpp
+++ b/src/mongo/client/sasl_client_authenticate_impl.cpp
@@ -167,19 +167,18 @@ Status configureSession(SaslClientSession* session,
     return session->initialize();
 }
 
-void asyncSaslConversation(auth::RunCommandHook runCommand,
-                           const std::shared_ptr<SaslClientSession>& session,
-                           const BSONObj& saslCommandPrefix,
-                           const BSONObj& inputObj,
-                           std::string targetDatabase,
-                           int saslLogLevel,
-                           auth::AuthCompletionHandler handler) {
+Future<void> asyncSaslConversation(auth::RunCommandHook runCommand,
+                                   const std::shared_ptr<SaslClientSession>& session,
+                                   const BSONObj& saslCommandPrefix,
+                                   const BSONObj& inputObj,
+                                   std::string targetDatabase,
+                                   int saslLogLevel) {
     // Extract payload from previous step
     std::string payload;
     BSONType type;
     auto status = saslExtractPayload(inputObj, &payload, &type);
     if (!status.isOK())
-        return handler(std::move(status));
+        return status;
 
     LOG(saslLogLevel) << "sasl client input: " << base64::encode(payload) << endl;
 
@@ -187,7 +186,7 @@ void asyncSaslConversation(auth::RunCommandHook runCommand,
     std::string responsePayload;
     status = session->step(payload, &responsePayload);
     if (!status.isOK())
-        return handler(std::move(status));
+        return status;
 
     LOG(saslLogLevel) << "sasl client output: " << base64::encode(responsePayload) << endl;
 
@@ -202,41 +201,31 @@ void asyncSaslConversation(auth::RunCommandHook runCommand,
     if (!conversationId.eoo())
         commandBuilder.append(conversationId);
 
-    auto request = RemoteCommandRequest();
-    request.dbname = targetDatabase;
-    request.cmdObj = commandBuilder.obj();
-
     // Asynchronously continue the conversation
-    runCommand(
-        request,
-        [runCommand, session, targetDatabase, saslLogLevel, handler](auth::AuthResponse response) {
-            if (!response.isOK()) {
-                return handler(std::move(response));
-            }
-
-            auto serverResponse = response.data.getOwned();
+    return runCommand(OpMsgRequest::fromDBAndBody(targetDatabase, commandBuilder.obj()))
+        .then([runCommand, session, targetDatabase, saslLogLevel](
+                  BSONObj serverResponse) -> Future<void> {
             auto status = getStatusFromCommandResult(serverResponse);
             if (!status.isOK()) {
-                return handler(status);
+                return status;
             }
 
             // Exit if we have finished
             if (session->isDone()) {
                 bool isServerDone = serverResponse[saslCommandDoneFieldName].trueValue();
                 if (!isServerDone) {
-                    return handler({ErrorCodes::ProtocolError, "Client finished before server."});
+                    return Status(ErrorCodes::ProtocolError, "Client finished before server.");
                 }
-                return handler(std::move(response));
+                return Status::OK();
             }
 
-            BSONObj saslFollowupCommandPrefix = BSON(saslContinueCommandName << 1);
-            asyncSaslConversation(runCommand,
-                                  session,
-                                  std::move(saslFollowupCommandPrefix),
-                                  std::move(serverResponse),
-                                  std::move(targetDatabase),
-                                  saslLogLevel,
-                                  handler);
+            static const BSONObj saslFollowupCommandPrefix = BSON(saslContinueCommandName << 1);
+            return asyncSaslConversation(runCommand,
+                                         session,
+                                         std::move(saslFollowupCommandPrefix),
+                                         std::move(serverResponse),
+                                         std::move(targetDatabase),
+                                         saslLogLevel);
         });
 }
 
@@ -244,26 +233,25 @@ void asyncSaslConversation(auth::RunCommandHook runCommand,
  * Driver for the client side of a sasl authentication session, conducted synchronously over
  * "client".
  */
-void saslClientAuthenticateImpl(auth::RunCommandHook runCommand,
-                                const HostAndPort& hostname,
-                                const BSONObj& saslParameters,
-                                auth::AuthCompletionHandler handler) {
+Future<void> saslClientAuthenticateImpl(auth::RunCommandHook runCommand,
+                                        const HostAndPort& hostname,
+                                        const BSONObj& saslParameters) {
     int saslLogLevel = getSaslClientLogLevel(saslParameters);
     std::string targetDatabase;
     try {
         Status status = bsonExtractStringFieldWithDefault(
             saslParameters, saslCommandUserDBFieldName, saslDefaultDBName, &targetDatabase);
         if (!status.isOK())
-            return handler(std::move(status));
+            return status;
     } catch (const DBException& ex) {
-        return handler(ex.toStatus());
+        return ex.toStatus();
     }
 
     std::string mechanism;
     Status status =
         bsonExtractStringField(saslParameters, saslCommandMechanismFieldName, &mechanism);
     if (!status.isOK()) {
-        return handler(std::move(status));
+        return status;
     }
 
     // NOTE: this must be a shared_ptr so that we can capture it in a lambda later on.
@@ -272,19 +260,18 @@ void saslClientAuthenticateImpl(auth::RunCommandHook runCommand,
 
     status = configureSession(session.get(), hostname, targetDatabase, saslParameters);
     if (!status.isOK())
-        return handler(std::move(status));
+        return status;
 
     BSONObj saslFirstCommandPrefix =
         BSON(saslStartCommandName << 1 << saslCommandMechanismFieldName
                                   << session->getParameter(SaslClientSession::parameterMechanism));
     BSONObj inputObj = BSON(saslCommandPayloadFieldName << "");
-    asyncSaslConversation(runCommand,
-                          session,
-                          std::move(saslFirstCommandPrefix),
-                          std::move(inputObj),
-                          targetDatabase,
-                          saslLogLevel,
-                          handler);
+    return asyncSaslConversation(runCommand,
+                                 session,
+                                 std::move(saslFirstCommandPrefix),
+                                 std::move(inputObj),
+                                 targetDatabase,
+                                 saslLogLevel);
 }
 
 MONGO_INITIALIZER(SaslClientAuthenticateFunction)(InitializerContext* context) {
diff --git a/src/mongo/db/SConscript b/src/mongo/db/SConscript
index a81c75a0d18..788417390f2 100644
--- a/src/mongo/db/SConscript
+++ b/src/mongo/db/SConscript
@@ -169,7 +169,7 @@ env.Library(
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/base',
-        '$BUILD_DIR/mongo/db/auth/internal_user_auth',
+        '$BUILD_DIR/mongo/client/authentication',
         '$BUILD_DIR/mongo/util/net/ssl_manager',
     ]
 )
@@ -452,7 +452,6 @@ env.Library(
         "auth/authorization_manager_global",
     ],
     LIBDEPS_PRIVATE=[
-        "$BUILD_DIR/mongo/db/auth/internal_user_auth",
         "$BUILD_DIR/mongo/db/auth/security_key",
     ],
 )
@@ -907,7 +906,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/catalog/multi_index_block',
     ],
     LIBDEPS_PRIVATE=[
-        '$BUILD_DIR/mongo/db/auth/internal_user_auth',
+        '$BUILD_DIR/mongo/client/authentication',
         '$BUILD_DIR/mongo/db/commands/list_collections_filter',
     ],
 )
@@ -1378,15 +1377,13 @@ env.Library(
     ],
     LIBDEPS=[
         '$BUILD_DIR/mongo/base',
+        '$BUILD_DIR/mongo/client/authentication',
         '$BUILD_DIR/mongo/client/remote_command_targeter',
         '$BUILD_DIR/mongo/db/concurrency/lock_manager',
         '$BUILD_DIR/mongo/db/repl/repl_coordinator_interface',
         'dbdirectclient',
         'sessions_collection',
     ],
-    LIBDEPS_PRIVATE=[
-        '$BUILD_DIR/mongo/db/auth/internal_user_auth',
-    ],
 )
 
 env.Library(
diff --git a/src/mongo/db/auth/SConscript b/src/mongo/db/auth/SConscript
index b5ed9fc9f87..bc823be0c3e 100644
--- a/src/mongo/db/auth/SConscript
+++ b/src/mongo/db/auth/SConscript
@@ -179,16 +179,6 @@ env.Library(
 )
 
 env.Library(
-    target='internal_user_auth',
-    source=[
-        'internal_user_auth.cpp',
-    ],
-    LIBDEPS=[
-        '$BUILD_DIR/mongo/base',
-    ],
-)
-
-env.Library(
     target='authorization_manager_global',
     source=[
         'authorization_manager_global.cpp',
@@ -209,11 +199,11 @@ env.Library(
         '$BUILD_DIR/mongo/base',
     ],
     LIBDEPS_PRIVATE=[
-        'internal_user_auth',
         'sasl_options',
         'security_file',
         'user',
         '$BUILD_DIR/mongo/base/secure_allocator',
+        '$BUILD_DIR/mongo/client/authentication',
         '$BUILD_DIR/mongo/crypto/sha_block_${MONGO_CRYPTO}',
         '$BUILD_DIR/mongo/util/icu',
         '$BUILD_DIR/mongo/util/md5',
@@ -241,7 +231,6 @@ env.Library(
     ],
     LIBDEPS=[
         'auth',
-        'internal_user_auth',
         'auth_impl_internal',
         'authorization_manager_global',
         'saslauth',
diff --git a/src/mongo/db/auth/internal_user_auth.cpp b/src/mongo/db/auth/internal_user_auth.cpp
deleted file mode 100644
index 4c2598072dc..00000000000
--- a/src/mongo/db/auth/internal_user_auth.cpp
+++ /dev/null
@@ -1,95 +0,0 @@
-
-/**
- *    Copyright (C) 2018-present MongoDB, Inc.
- *
- *    This program is free software: you can redistribute it and/or modify
- *    it under the terms of the Server Side Public License, version 1,
- *    as published by MongoDB, Inc.
- *
- *    This program is distributed in the hope that it will be useful,
- *    but WITHOUT ANY WARRANTY; without even the implied warranty of
- *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *    Server Side Public License for more details.
- *
- *    You should have received a copy of the Server Side Public License
- *    along with this program. If not, see
- *    <http://www.mongodb.com/licensing/server-side-public-license>.
- *
- *    As a special exception, the copyright holders give permission to link the
- *    code of portions of this program with the OpenSSL library under certain
- *    conditions as described in each individual source file and distribute
- *    linked combinations including the program with the OpenSSL library. You
- *    must comply with the Server Side Public License in all respects for
- *    all of the code used other than as permitted herein. If you modify file(s)
- *    with this exception, you may extend this exception to your version of the
- *    file(s), but you are not obligated to do so. If you do not wish to do so,
- *    delete this exception statement from your version. If you delete this
- *    exception statement from all source files in the program, then also delete
- *    it in the license file.
- */
-
-#define MONGO_LOG_DEFAULT_COMPONENT ::mongo::logger::LogComponent::kAccessControl
-
-#include "mongo/platform/basic.h"
-
-#include "mongo/db/auth/internal_user_auth.h"
-
-#include "mongo/bson/bsonobj.h"
-#include "mongo/util/log.h"
-
-namespace mongo {
-namespace {
-
-// not guarded by the authParams mutex never changed in
-// multi-threaded operation
-static bool authParamsSet = false;
-
-// Store default authentication parameters for internal authentication to cluster members,
-// guarded by the authParams mutex
-static std::vector<BSONObj> authParams;
-
-static stdx::mutex authParamMutex;
-
-template <typename Container>
-void setInternalUserAuthParamsFromContainer(Container&& params) {
-    if (!isInternalAuthSet()) {
-        authParamsSet = true;
-    }
-
-    stdx::lock_guard<stdx::mutex> lk(authParamMutex);
-    authParams.clear();
-    std::transform(params.begin(),
-                   params.end(),
-                   std::back_inserter(authParams),
-                   [](const BSONObj& obj) { return obj.getOwned(); });
-}
-
-}  // namespace
-
-bool isInternalAuthSet() {
-    return authParamsSet;
-}
-
-void setInternalUserAuthParams(BSONObj authParamsIn) {
-    setInternalUserAuthParamsFromContainer(std::initializer_list<BSONObj>{authParamsIn});
-}
-
-void setInternalUserAuthParams(const std::vector<BSONObj>& keys) {
-    setInternalUserAuthParamsFromContainer(keys);
-}
-
-size_t getInternalUserAuthParamsCount() {
-    stdx::lock_guard<stdx::mutex> lk(authParamMutex);
-    return authParams.size();
-}
-
-BSONObj getInternalUserAuthParams(size_t idx) {
-    if (!authParamsSet) {
-        return BSONObj();
-    }
-
-    stdx::lock_guard<stdx::mutex> lk(authParamMutex);
-    return (idx < authParams.size()) ? authParams.at(idx) : BSONObj();
-}
-
-}  // namespace mongo
diff --git a/src/mongo/db/auth/internal_user_auth.h b/src/mongo/db/auth/internal_user_auth.h
deleted file mode 100644
index 9b8106f9333..00000000000
--- a/src/mongo/db/auth/internal_user_auth.h
+++ /dev/null
@@ -1,72 +0,0 @@
-
-/**
- *    Copyright (C) 2018-present MongoDB, Inc.
- *
- *    This program is free software: you can redistribute it and/or modify
- *    it under the terms of the Server Side Public License, version 1,
- *    as published by MongoDB, Inc.
- *
- *    This program is distributed in the hope that it will be useful,
- *    but WITHOUT ANY WARRANTY; without even the implied warranty of
- *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *    Server Side Public License for more details.
- *
- *    You should have received a copy of the Server Side Public License
- *    along with this program. If not, see
- *    <http://www.mongodb.com/licensing/server-side-public-license>.
- *
- *    As a special exception, the copyright holders give permission to link the
- *    code of portions of this program with the OpenSSL library under certain
- *    conditions as described in each individual source file and distribute
- *    linked combinations including the program with the OpenSSL library. You
- *    must comply with the Server Side Public License in all respects for
- *    all of the code used other than as permitted herein. If you modify file(s)
- *    with this exception, you may extend this exception to your version of the
- *    file(s), but you are not obligated to do so. If you do not wish to do so,
- *    delete this exception statement from your version. If you delete this
- *    exception statement from all source files in the program, then also delete
- *    it in the license file.
- */
-
-#pragma once
-
-#include <string>
-#include <vector>
-
-namespace mongo {
-class BSONObj;
-
-/**
- * @return true if internal authentication parameters has been set up. Note this does not
- * imply that auth is enabled. For instance, with the --transitionToAuth flag this will
- * be set and auth will be disabled.
- */
-bool isInternalAuthSet();
-
-/**
- * This method initializes the authParams from a list of keys. It defaults to generating
- * SCRAM-SHA-1 credentials only.
- */
-void setInternalUserAuthParams(const std::vector<BSONObj>& keys);
-
-/**
- * This method initializes the authParams object with authentication
- * credentials to be used by authenticateInternalUser.
- */
-void setInternalUserAuthParams(BSONObj obj);
-
-/**
- * Returns a BSON object containing user info to be used by authenticateInternalUser
- *
- * The format of the return object is { authparams, fallbackParams:params}
- *
- * If SCRAM-SHA-1 is the internal auth mechanism the fallbackParams sub document is
- * for MONGODB-CR auth is included. For MONGODB-XC509 no fallbackParams document is
- * returned.
- *
- * If no internal auth information has been set then this returns an empty BSONObj.
- **/
-
-BSONObj getInternalUserAuthParams(size_t idx = 0);
-
-}  // namespace mongo
diff --git a/src/mongo/db/auth/sasl_authentication_session_test.cpp b/src/mongo/db/auth/sasl_authentication_session_test.cpp
index 2724f1833d9..f4812d09bb3 100644
--- a/src/mongo/db/auth/sasl_authentication_session_test.cpp
+++ b/src/mongo/db/auth/sasl_authentication_session_test.cpp
@@ -95,6 +95,7 @@ SaslConversation::SaslConversation(std::string mech)
           std::unique_ptr<AuthzManagerExternalState>(authManagerExternalState),
           AuthorizationManagerImpl::InstallMockForTestingOrAuthImpl{})),
       authSession(authManager->makeAuthorizationSession()),
+      registry({"SCRAM-SHA-1", "SCRAM-SHA-256", "PLAIN"}),
       mechanism(mech) {
 
     AuthorizationManager::set(getServiceContext(),
@@ -273,12 +274,6 @@ TEST_F(SaslIllegalConversation, IllegalClientMechanism) {
     ASSERT(!client->initialize().isOK() || !client->step(serverMessage, &clientMessage).isOK());
 }
 
-TEST_F(SaslIllegalConversation, IllegalServerMechanism) {
-    SASLServerMechanismRegistry registry;
-    auto swServer = registry.getServerMechanism("FAKE", "test");
-    ASSERT_NOT_OK(swServer.getStatus());
-}
-
 }  // namespace
 
 }  // namespace mongo
diff --git a/src/mongo/db/auth/sasl_mechanism_policies.h b/src/mongo/db/auth/sasl_mechanism_policies.h
index 27c972253d6..059a7737434 100644
--- a/src/mongo/db/auth/sasl_mechanism_policies.h
+++ b/src/mongo/db/auth/sasl_mechanism_policies.h
@@ -43,6 +43,12 @@ struct PLAINPolicy {
     static SecurityPropertySet getProperties() {
         return SecurityPropertySet{};
     }
+    static int securityLevel() {
+        return 0;
+    }
+    static constexpr bool isInternalAuthMech() {
+        return false;
+    }
 };
 
 struct SCRAMSHA1Policy {
@@ -54,6 +60,12 @@ struct SCRAMSHA1Policy {
     static SecurityPropertySet getProperties() {
         return SecurityPropertySet{SecurityProperty::kNoPlainText, SecurityProperty::kMutualAuth};
     }
+    static int securityLevel() {
+        return 1;
+    }
+    static constexpr bool isInternalAuthMech() {
+        return true;
+    }
 };
 
 struct SCRAMSHA256Policy {
@@ -65,6 +77,12 @@ struct SCRAMSHA256Policy {
     static SecurityPropertySet getProperties() {
         return SecurityPropertySet{SecurityProperty::kNoPlainText, SecurityProperty::kMutualAuth};
     }
+    static int securityLevel() {
+        return true;
+    }
+    static constexpr bool isInternalAuthMech() {
+        return true;
+    }
 };
 
 struct GSSAPIPolicy {
@@ -74,6 +92,12 @@ struct GSSAPIPolicy {
     static SecurityPropertySet getProperties() {
         return SecurityPropertySet{SecurityProperty::kNoPlainText, SecurityProperty::kMutualAuth};
     }
+    static int securityLevel() {
+        return 3;
+    }
+    static constexpr bool isInternalAuthMech() {
+        return false;
+    }
 };
 
 
diff --git a/src/mongo/db/auth/sasl_mechanism_registry.cpp b/src/mongo/db/auth/sasl_mechanism_registry.cpp
index c6993210ea7..73847bd37b8 100644
--- a/src/mongo/db/auth/sasl_mechanism_registry.cpp
+++ b/src/mongo/db/auth/sasl_mechanism_registry.cpp
@@ -60,13 +60,22 @@ void SASLServerMechanismRegistry::set(ServiceContext* service,
     getSASLServerMechanismRegistry(service) = std::move(registry);
 }
 
+SASLServerMechanismRegistry::SASLServerMechanismRegistry(std::vector<std::string> enabledMechanisms)
+    : _enabledMechanisms(std::move(enabledMechanisms)) {}
+
+void SASLServerMechanismRegistry::setEnabledMechanisms(std::vector<std::string> enabledMechanisms) {
+    _enabledMechanisms = std::move(enabledMechanisms);
+}
+
 StatusWith<std::unique_ptr<ServerMechanismBase>> SASLServerMechanismRegistry::getServerMechanism(
     StringData mechanismName, std::string authenticationDatabase) {
-    auto& map = _getMapRef(authenticationDatabase);
+    auto& mechList = _getMapRef(authenticationDatabase);
 
-    auto it = map.find(mechanismName.toString());
-    if (it != map.end()) {
-        return it->second->create(std::move(authenticationDatabase));
+    auto it = std::find_if(mechList.begin(), mechList.end(), [&](const auto& mech) {
+        return (mech->mechanismName() == mechanismName);
+    });
+    if (it != mechList.end()) {
+        return (*it)->create(std::move(authenticationDatabase));
     }
 
     return Status(ErrorCodes::BadValue,
@@ -100,18 +109,23 @@ void SASLServerMechanismRegistry::advertiseMechanismNamesForUser(OperationContex
 
         user = std::move(swUser.getValue());
         BSONArrayBuilder mechanismsBuilder;
-        auto& map = _getMapRef(userName.getDB());
+        const auto& mechList = _getMapRef(userName.getDB());
 
-        for (const auto& factoryIt : map) {
-            SecurityPropertySet properties = factoryIt.second->properties();
+        for (const auto& factoryIt : mechList) {
+            SecurityPropertySet properties = factoryIt->properties();
             if (!properties.hasAllProperties(SecurityPropertySet{SecurityProperty::kNoPlainText,
                                                                  SecurityProperty::kMutualAuth}) &&
                 userName.getDB() != "$external") {
                 continue;
             }
 
-            if (factoryIt.second->canMakeMechanismForUser(user.get())) {
-                mechanismsBuilder << factoryIt.first;
+            auto mechanismEnabled = _mechanismSupportedByConfig(factoryIt->mechanismName());
+            if (!mechanismEnabled && userName == internalSecurity.user->getName()) {
+                mechanismEnabled = factoryIt->isInternalAuthMech();
+            }
+
+            if (mechanismEnabled && factoryIt->canMakeMechanismForUser(user.get())) {
+                mechanismsBuilder << factoryIt->mechanismName();
             }
         }
 
@@ -119,8 +133,8 @@ void SASLServerMechanismRegistry::advertiseMechanismNamesForUser(OperationContex
     }
 }
 
-bool SASLServerMechanismRegistry::_mechanismSupportedByConfig(StringData mechName) {
-    return sequenceContains(saslGlobalParams.authenticationMechanisms, mechName);
+bool SASLServerMechanismRegistry::_mechanismSupportedByConfig(StringData mechName) const {
+    return sequenceContains(_enabledMechanisms, mechName);
 }
 
 namespace {
@@ -128,7 +142,9 @@ ServiceContext::ConstructorActionRegisterer SASLServerMechanismRegistryInitializ
     "CreateSASLServerMechanismRegistry",
     {"EndStartupOptionStorage"},
     [](ServiceContext* service) {
-        SASLServerMechanismRegistry::set(service, std::make_unique<SASLServerMechanismRegistry>());
+        SASLServerMechanismRegistry::set(service,
+                                         std::make_unique<SASLServerMechanismRegistry>(
+                                             saslGlobalParams.authenticationMechanisms));
     }};
 }  // namespace
 
diff --git a/src/mongo/db/auth/sasl_mechanism_registry.h b/src/mongo/db/auth/sasl_mechanism_registry.h
index 4c1b8546bbc..e479b0a5177 100644
--- a/src/mongo/db/auth/sasl_mechanism_registry.h
+++ b/src/mongo/db/auth/sasl_mechanism_registry.h
@@ -102,6 +102,24 @@ public:
     virtual ~SaslServerCommonBase() = default;
     virtual StringData mechanismName() const = 0;
     virtual SecurityPropertySet properties() const = 0;
+
+    /**
+     * This returns a number that represents the "amount" of security provided by this mechanism
+     * to determine the order in which it is offered to clients in the isMaster
+     * saslSupportedMechs response.
+     *
+     * The value of securityLevel is arbitrary so long as the more secure mechanisms return a
+     * higher value than the less secure mechanisms.
+     *
+     * For example, SCRAM-SHA-256 > SCRAM-SHA-1 > PLAIN
+     */
+    virtual int securityLevel() const = 0;
+
+    /**
+     * Returns true if the mechanism can be used for internal cluster authentication.
+     * Currently only SCRAM-SHA-1/SCRAM-SHA-256 return true here.
+     */
+    virtual bool isInternalAuthMech() const = 0;
 };
 
 /**
@@ -229,6 +247,14 @@ public:
     SecurityPropertySet properties() const final {
         return policy_type::getProperties();
     }
+
+    int securityLevel() const final {
+        return policy_type::securityLevel();
+    }
+
+    bool isInternalAuthMech() const final {
+        return policy_type::isInternalAuthMech();
+    }
 };
 
 /** Instantiates a class which provides runtime access to Policy properties. */
@@ -254,6 +280,14 @@ public:
     SecurityPropertySet properties() const final {
         return policy_type::getProperties();
     }
+
+    int securityLevel() const final {
+        return policy_type::securityLevel();
+    }
+
+    bool isInternalAuthMech() const final {
+        return policy_type::isInternalAuthMech();
+    }
 };
 
 /**
@@ -268,6 +302,16 @@ public:
     static void set(ServiceContext* service, std::unique_ptr<SASLServerMechanismRegistry> registry);
 
     /**
+     * Intialize the registry with a list of enabled mechanisms.
+     */
+    explicit SASLServerMechanismRegistry(std::vector<std::string> enabledMechanisms);
+
+    /**
+     * Sets a new list of enabled mechanisms - used in testing.
+     */
+    void setEnabledMechanisms(std::vector<std::string> enabledMechanisms);
+
+    /**
      * Produces a list of SASL mechanisms which can be used to authenticate as a user.
      * If isMasterCmd contains a field with a username called 'saslSupportedMechs',
      * will populate 'builder' with an Array called saslSupportedMechs containing each mechanism the
@@ -302,38 +346,42 @@ public:
         using policy_type = typename T::policy_type;
         auto mechName = policy_type::getName();
 
-        // Always allow SCRAM-SHA-1 to pass to the first sasl step since we need to
-        // handle internal user authentication, SERVER-16534
         if (validateGlobalConfig &&
-            (mechName != "SCRAM-SHA-1" && !_mechanismSupportedByConfig(mechName))) {
+            (!policy_type::isInternalAuthMech() && !_mechanismSupportedByConfig(mechName))) {
             return false;
         }
 
-        invariant(
-            _getMapRef(T::isInternal).emplace(mechName.toString(), std::make_unique<T>()).second);
+        auto& list = _getMapRef(T::isInternal);
+        list.emplace_back(std::make_unique<T>());
+        std::stable_sort(list.begin(), list.end(), [](const auto& a, const auto& b) {
+            return (a->securityLevel() >= b->securityLevel());
+        });
+
         return true;
     }
 
 private:
-    stdx::unordered_map<std::string, std::unique_ptr<ServerFactoryBase>>& _getMapRef(
-        StringData dbName) {
+    using MechList = std::vector<std::unique_ptr<ServerFactoryBase>>;
+
+    MechList& _getMapRef(StringData dbName) {
         return _getMapRef(dbName != "$external"_sd);
     }
 
-    stdx::unordered_map<std::string, std::unique_ptr<ServerFactoryBase>>& _getMapRef(
-        bool internal) {
+    MechList& _getMapRef(bool internal) {
         if (internal) {
-            return _internalMap;
+            return _internalMechs;
         }
-        return _externalMap;
+        return _externalMechs;
     }
 
-    bool _mechanismSupportedByConfig(StringData mechName);
+    bool _mechanismSupportedByConfig(StringData mechName) const;
 
     // Stores factories which make mechanisms for all databases other than $external
-    stdx::unordered_map<std::string, std::unique_ptr<ServerFactoryBase>> _internalMap;
+    MechList _internalMechs;
     // Stores factories which make mechanisms exclusively for $external
-    stdx::unordered_map<std::string, std::unique_ptr<ServerFactoryBase>> _externalMap;
+    MechList _externalMechs;
+
+    std::vector<std::string> _enabledMechanisms;
 };
 
 template <typename Factory>
diff --git a/src/mongo/db/auth/sasl_mechanism_registry_test.cpp b/src/mongo/db/auth/sasl_mechanism_registry_test.cpp
index 2d5b9012022..26c2d3766a0 100644
--- a/src/mongo/db/auth/sasl_mechanism_registry_test.cpp
+++ b/src/mongo/db/auth/sasl_mechanism_registry_test.cpp
@@ -64,38 +64,55 @@ TEST(SecurityProperty, mutualAndPlainHasAllSubsets) {
         SecurityPropertySet{SecurityProperty::kMutualAuth, SecurityProperty::kNoPlainText}));
 }
 
+template <typename Policy>
+class BaseMockMechanism : public MakeServerMechanism<Policy> {
+public:
+    explicit BaseMockMechanism(std::string authenticationDatabase)
+        : MakeServerMechanism<Policy>(std::move(authenticationDatabase)) {}
+
+protected:
+    StatusWith<std::tuple<bool, std::string>> stepImpl(OperationContext* opCtx,
+                                                       StringData input) final {
+        return std::make_tuple(true, std::string());
+    }
+};
+
+template <typename Policy, bool argIsInternal>
+class BaseMockMechanismFactory : public MakeServerFactory<Policy> {
+public:
+    static constexpr bool isInternal = argIsInternal;
+    bool canMakeMechanismForUser(const User* user) const final {
+        return true;
+    }
+};
+
 // Policy for a hypothetical "FOO" SASL mechanism.
 struct FooPolicy {
     static constexpr StringData getName() {
         return "FOO"_sd;
     }
 
+    static constexpr int securityLevel() {
+        return 0;
+    }
+
+    static constexpr bool isInternalAuthMech() {
+        return false;
+    }
+
     // This mech is kind of dangerous, it sends plaintext passwords across the wire.
     static SecurityPropertySet getProperties() {
         return SecurityPropertySet{SecurityProperty::kMutualAuth};
     }
 };
 
-class FooMechanism : public MakeServerMechanism<FooPolicy> {
+class FooMechanism : public BaseMockMechanism<FooPolicy> {
 public:
-    explicit FooMechanism(std::string authenticationDatabase)
-        : MakeServerMechanism<FooPolicy>(std::move(authenticationDatabase)) {}
-
-protected:
-    StatusWith<std::tuple<bool, std::string>> stepImpl(OperationContext* opCtx,
-                                                       StringData input) final {
-        return std::make_tuple(true, std::string());
-    }
+    using BaseMockMechanism<FooPolicy>::BaseMockMechanism;
 };
 
 template <bool argIsInternal>
-class FooMechanismFactory : public MakeServerFactory<FooMechanism> {
-public:
-    static constexpr bool isInternal = argIsInternal;
-    bool canMakeMechanismForUser(const User* user) const final {
-        return true;
-    }
-};
+class FooMechanismFactory : public BaseMockMechanismFactory<FooMechanism, argIsInternal> {};
 
 // Policy for a hypothetical "BAR" SASL mechanism.
 struct BarPolicy {
@@ -103,32 +120,53 @@ struct BarPolicy {
         return "BAR"_sd;
     }
 
+    static constexpr int securityLevel() {
+        return 1;
+    }
+
+    static constexpr bool isInternalAuthMech() {
+        return false;
+    }
+
     static SecurityPropertySet getProperties() {
         return SecurityPropertySet{SecurityProperty::kMutualAuth, SecurityProperty::kNoPlainText};
     }
 };
 
-class BarMechanism : public MakeServerMechanism<BarPolicy> {
+class BarMechanism : public BaseMockMechanism<BarPolicy> {
 public:
-    explicit BarMechanism(std::string authenticationDatabase)
-        : MakeServerMechanism<BarPolicy>(std::move(authenticationDatabase)) {}
-
-protected:
-    StatusWith<std::tuple<bool, std::string>> stepImpl(OperationContext* opCtx,
-                                                       StringData input) final {
-        return std::make_tuple(true, std::string());
-    }
+    using BaseMockMechanism<BarPolicy>::BaseMockMechanism;
 };
 
 template <bool argIsInternal>
-class BarMechanismFactory : public MakeServerFactory<BarMechanism> {
-public:
-    static constexpr bool isInternal = argIsInternal;
-    bool canMakeMechanismForUser(const User* user) const final {
+class BarMechanismFactory : public BaseMockMechanismFactory<BarMechanism, argIsInternal> {};
+
+// Policy for a hypothetical "InternalAuth" SASL mechanism.
+struct InternalAuthPolicy {
+    static constexpr StringData getName() {
+        return "InternalAuth"_sd;
+    }
+
+    static constexpr int securityLevel() {
+        return 2;
+    }
+
+    static constexpr bool isInternalAuthMech() {
         return true;
     }
+
+    static SecurityPropertySet getProperties() {
+        return SecurityPropertySet{SecurityProperty::kMutualAuth, SecurityProperty::kNoPlainText};
+    }
 };
 
+class InternalAuthMechanism : public BaseMockMechanism<InternalAuthPolicy> {
+public:
+    using BaseMockMechanism<InternalAuthPolicy>::BaseMockMechanism;
+};
+
+class InternalAuthMechanismFactory : public BaseMockMechanismFactory<InternalAuthMechanism, true> {
+};
 
 class MechanismRegistryTest : public ServiceContextTest {
 public:
@@ -137,7 +175,9 @@ public:
           authManagerExternalState(new AuthzManagerExternalStateMock()),
           authManager(new AuthorizationManagerImpl(
               std::unique_ptr<AuthzManagerExternalStateMock>(authManagerExternalState),
-              AuthorizationManagerImpl::InstallMockForTestingOrAuthImpl{})) {
+              AuthorizationManagerImpl::InstallMockForTestingOrAuthImpl{})),
+          // By default the registry is initialized with all mechanisms enabled.
+          registry({"FOO", "BAR", "InternalAuth"}) {
         AuthorizationManager::set(getServiceContext(),
                                   std::unique_ptr<AuthorizationManager>(authManager));
 
@@ -180,6 +220,17 @@ public:
                                                         << "roles"
                                                         << BSONArray()),
                                                    BSONObj()));
+
+        internalSecurity.user = std::make_shared<User>(UserName("__system", "local"));
+    }
+
+    BSONObj getMechsFor(const UserName user) {
+        BSONObjBuilder builder;
+        registry.advertiseMechanismNamesForUser(
+            opCtx.get(),
+            BSON("isMaster" << 1 << "saslSupportedMechs" << user.getUnambiguousName()),
+            &builder);
+        return builder.obj();
     }
 
     ServiceContext::UniqueOperationContext opCtx;
@@ -187,6 +238,9 @@ public:
     AuthorizationManager* authManager;
 
     SASLServerMechanismRegistry registry;
+
+    const UserName internalSajack = {"sajack"_sd, "test"_sd};
+    const UserName externalSajack = {"sajack"_sd, "$external"_sd};
 };
 
 TEST_F(MechanismRegistryTest, acquireInternalMechanism) {
@@ -226,14 +280,7 @@ TEST_F(MechanismRegistryTest, invalidUserCantAdvertiseMechs) {
     registry.registerFactory<FooMechanismFactory<true>>(
         SASLServerMechanismRegistry::kNoValidateGlobalMechanisms);
 
-    BSONObjBuilder builder;
-
-    registry.advertiseMechanismNamesForUser(opCtx.get(),
-                                            BSON("isMaster" << 1 << "saslSupportedMechs"
-                                                            << "test.noSuchUser"),
-                                            &builder);
-
-    ASSERT_BSONOBJ_EQ(BSONObj(), builder.obj());
+    ASSERT_BSONOBJ_EQ(BSONObj(), getMechsFor(UserName("noSuchUser"_sd, "test"_sd)));
 }
 
 TEST_F(MechanismRegistryTest, strongMechCanAdvertise) {
@@ -242,56 +289,42 @@ TEST_F(MechanismRegistryTest, strongMechCanAdvertise) {
     registry.registerFactory<BarMechanismFactory<false>>(
         SASLServerMechanismRegistry::kNoValidateGlobalMechanisms);
 
-    BSONObjBuilder builder;
-    registry.advertiseMechanismNamesForUser(opCtx.get(),
-                                            BSON("isMaster" << 1 << "saslSupportedMechs"
-                                                            << "test.sajack"),
-                                            &builder);
-
-    BSONObj obj = builder.done();
-    ASSERT_BSONOBJ_EQ(BSON("saslSupportedMechs" << BSON_ARRAY("BAR")), obj);
-
-    BSONObjBuilder builderExternal;
-    registry.advertiseMechanismNamesForUser(opCtx.get(),
-                                            BSON("isMaster" << 1 << "saslSupportedMechs"
-                                                            << "$external.sajack"),
-                                            &builderExternal);
-
-    BSONObj objExternal = builderExternal.done();
-    ASSERT_BSONOBJ_EQ(BSON("saslSupportedMechs" << BSON_ARRAY("BAR")), objExternal);
+    ASSERT_BSONOBJ_EQ(BSON("saslSupportedMechs" << BSON_ARRAY("BAR")), getMechsFor(internalSajack));
+    ASSERT_BSONOBJ_EQ(BSON("saslSupportedMechs" << BSON_ARRAY("BAR")), getMechsFor(externalSajack));
 }
 
 TEST_F(MechanismRegistryTest, weakMechCannotAdvertiseOnInternal) {
     registry.registerFactory<FooMechanismFactory<true>>(
         SASLServerMechanismRegistry::kNoValidateGlobalMechanisms);
 
-    BSONObjBuilder builder;
-    registry.advertiseMechanismNamesForUser(opCtx.get(),
-                                            BSON("isMaster" << 1 << "saslSupportedMechs"
-                                                            << "test.sajack"),
-                                            &builder);
-
-
-    BSONObj obj = builder.done();
-
-    ASSERT_BSONOBJ_EQ(BSON("saslSupportedMechs" << BSONArray()), obj);
+    ASSERT_BSONOBJ_EQ(BSON("saslSupportedMechs" << BSONArray()), getMechsFor(internalSajack));
 }
 
 TEST_F(MechanismRegistryTest, weakMechCanAdvertiseOnExternal) {
     registry.registerFactory<FooMechanismFactory<false>>(
         SASLServerMechanismRegistry::kNoValidateGlobalMechanisms);
 
-    BSONObjBuilder builder;
-    registry.advertiseMechanismNamesForUser(opCtx.get(),
-                                            BSON("isMaster" << 1 << "saslSupportedMechs"
-                                                            << "$external.sajack"),
-                                            &builder);
+    ASSERT_BSONOBJ_EQ(BSON("saslSupportedMechs" << BSON_ARRAY("FOO")), getMechsFor(externalSajack));
+}
 
-    BSONObj obj = builder.done();
+TEST_F(MechanismRegistryTest, internalAuth) {
+    registry.setEnabledMechanisms({"BAR"});
 
-    ASSERT_BSONOBJ_EQ(BSON("saslSupportedMechs" << BSON_ARRAY("FOO")), obj);
+    registry.registerFactory<BarMechanismFactory<true>>(
+        SASLServerMechanismRegistry::kValidateGlobalMechanisms);
+    registry.registerFactory<InternalAuthMechanismFactory>(
+        SASLServerMechanismRegistry::kValidateGlobalMechanisms);
+
+    ASSERT_BSONOBJ_EQ(BSON("saslSupportedMechs" << BSON_ARRAY("BAR")), getMechsFor(internalSajack));
+    ASSERT_BSONOBJ_EQ(BSON("saslSupportedMechs" << BSON_ARRAY("InternalAuth"
+                                                              << "BAR")),
+                      getMechsFor(internalSecurity.user->getName()));
+
+    registry.setEnabledMechanisms({"BAR", "InternalAuth"});
+    ASSERT_BSONOBJ_EQ(BSON("saslSupportedMechs" << BSON_ARRAY("InternalAuth"
+                                                              << "BAR")),
+                      getMechsFor(internalSajack));
 }
 
-
 }  // namespace
 }  // namespace mongo
diff --git a/src/mongo/db/auth/sasl_scram_server_conversation.cpp b/src/mongo/db/auth/sasl_scram_server_conversation.cpp
index 0e11a6238bf..5d91a26466b 100644
--- a/src/mongo/db/auth/sasl_scram_server_conversation.cpp
+++ b/src/mongo/db/auth/sasl_scram_server_conversation.cpp
@@ -180,16 +180,17 @@ StatusWith<std::tuple<bool, std::string>> SaslSCRAMServerMechanism<Policy>::_fir
     }
     const auto clientNonce = input[1].substr(2);
 
+    UserName user(ServerMechanismBase::ServerMechanismBase::_principalName,
+                  ServerMechanismBase::getAuthenticationDatabase());
 
-    // SERVER-16534, SCRAM-SHA-1 must be enabled for authenticating the internal user, so that
+    // SERVER-16534, some mechanisms must be enabled for authenticating the internal user, so that
     // cluster members may communicate with each other. Hence ignore disabled auth mechanism
     // for the internal user.
-    UserName user(ServerMechanismBase::ServerMechanismBase::_principalName,
-                  ServerMechanismBase::getAuthenticationDatabase());
-    if (Policy::getName() == "SCRAM-SHA-1"_sd &&
-        !sequenceContains(saslGlobalParams.authenticationMechanisms, "SCRAM-SHA-1") &&
+    if (Policy::isInternalAuthMech() &&
+        !sequenceContains(saslGlobalParams.authenticationMechanisms, Policy::getName()) &&
         user != internalSecurity.user->getName()) {
-        return Status(ErrorCodes::BadValue, "SCRAM-SHA-1 authentication is disabled");
+        return Status(ErrorCodes::BadValue,
+                      str::stream() << Policy::getName() << " authentication is disabled");
     }
 
     // The authentication database is also the source database for the user.
diff --git a/src/mongo/db/auth/sasl_scram_server_conversation.h b/src/mongo/db/auth/sasl_scram_server_conversation.h
index 5a398f242ea..5c1e56f8e64 100644
--- a/src/mongo/db/auth/sasl_scram_server_conversation.h
+++ b/src/mongo/db/auth/sasl_scram_server_conversation.h
@@ -38,7 +38,7 @@
 namespace mongo {
 
 /**
- *  Server side authentication session for SASL SCRAM-SHA-1.
+ *  Server side authentication session for SASL SCRAM-SHA-1/256.
  */
 template <typename Policy>
 class SaslSCRAMServerMechanism : public MakeServerMechanism<Policy> {
diff --git a/src/mongo/db/auth/security_key.cpp b/src/mongo/db/auth/security_key.cpp
index beec7b11c3b..00344494374 100644
--- a/src/mongo/db/auth/security_key.cpp
+++ b/src/mongo/db/auth/security_key.cpp
@@ -39,13 +39,13 @@
 #include <vector>
 
 #include "mongo/base/status_with.h"
+#include "mongo/client/authenticate.h"
 #include "mongo/crypto/mechanism_scram.h"
 #include "mongo/crypto/sha1_block.h"
 #include "mongo/crypto/sha256_block.h"
 #include "mongo/db/auth/action_set.h"
 #include "mongo/db/auth/action_type.h"
 #include "mongo/db/auth/authorization_manager.h"
-#include "mongo/db/auth/internal_user_auth.h"
 #include "mongo/db/auth/privilege.h"
 #include "mongo/db/auth/sasl_command_constants.h"
 #include "mongo/db/auth/sasl_options.h"
@@ -69,8 +69,7 @@ public:
           _salt256(scram::Presecrets<SHA256Block>::generateSecureRandomSalt()),
           _filename(filename) {}
 
-    boost::optional<std::pair<User::CredentialData, BSONObj>> generate(
-        const std::string& password) {
+    boost::optional<User::CredentialData> generate(const std::string& password) {
         if (password.size() < kMinKeyLength || password.size() > kMaxKeyLength) {
             error() << " security key in " << _filename << " has length " << password.size()
                     << ", must be between 6 and 1024 chars";
@@ -100,17 +99,7 @@ public:
                                   saslGlobalParams.scramSHA256IterationCount.load())))
             return boost::none;
 
-        auto internalAuthParams =
-            BSON(saslCommandMechanismFieldName << "SCRAM-SHA-1" << saslCommandUserDBFieldName
-                                               << internalSecurity.user->getName().getDB()
-                                               << saslCommandUserFieldName
-                                               << internalSecurity.user->getName().getUser()
-                                               << saslCommandPasswordFieldName
-                                               << passwordDigest
-                                               << saslCommandDigestPasswordFieldName
-                                               << false);
-
-        return std::make_pair(std::move(credentials), std::move(internalAuthParams));
+        return credentials;
     }
 
 private:
@@ -152,15 +141,13 @@ bool setUpSecurityKey(const string& filename) {
         return false;
     }
 
-    std::vector<BSONObj> internalAuthParams;
     CredentialsGenerator generator(filename);
     auto credentials = generator.generate(keyStrings.front());
     if (!credentials) {
         return false;
     }
 
-    internalSecurity.user->setCredentials(std::move(credentials->first));
-    internalAuthParams.push_back(std::move(credentials->second));
+    internalSecurity.user->setCredentials(std::move(*credentials));
 
     if (keyStrings.size() == 2) {
         credentials = generator.generate(keyStrings[1]);
@@ -168,14 +155,13 @@ bool setUpSecurityKey(const string& filename) {
             return false;
         }
 
-        internalSecurity.alternateCredentials = std::move(credentials->first);
-        internalAuthParams.push_back(std::move(credentials->second));
+        internalSecurity.alternateCredentials = std::move(*credentials);
     }
 
     int clusterAuthMode = serverGlobalParams.clusterAuthMode.load();
     if (clusterAuthMode == ServerGlobalParams::ClusterAuthMode_keyFile ||
         clusterAuthMode == ServerGlobalParams::ClusterAuthMode_sendKeyFile) {
-        setInternalUserAuthParams(internalAuthParams);
+        auth::setInternalAuthKeys(keyStrings);
     }
 
     return true;
diff --git a/src/mongo/db/auth/user_name.h b/src/mongo/db/auth/user_name.h
index afb8c70844a..54c653a98ee 100644
--- a/src/mongo/db/auth/user_name.h
+++ b/src/mongo/db/auth/user_name.h
@@ -88,6 +88,13 @@ public:
     }
 
     /**
+     * Gets the full unambiguous unique name of a user as a string, formatted as "db.user"
+     */
+    std::string getUnambiguousName() const {
+        return str::stream() << getDB() << "." << getUser();
+    }
+
+    /**
      * Stringifies the object, for logging/debugging.
      */
     std::string toString() const {
diff --git a/src/mongo/db/cloner.cpp b/src/mongo/db/cloner.cpp
index 1fbc4699d72..2768cb0b578 100644
--- a/src/mongo/db/cloner.cpp
+++ b/src/mongo/db/cloner.cpp
@@ -38,7 +38,7 @@
 #include "mongo/bson/unordered_fields_bsonobj_comparator.h"
 #include "mongo/bson/util/bson_extract.h"
 #include "mongo/bson/util/builder.h"
-#include "mongo/db/auth/internal_user_auth.h"
+#include "mongo/client/authenticate.h"
 #include "mongo/db/catalog/collection.h"
 #include "mongo/db/catalog/collection_catalog_entry.h"
 #include "mongo/db/catalog/collection_options.h"
@@ -705,9 +705,11 @@ Status Cloner::copyDb(OperationContext* opCtx,
                 return Status(ErrorCodes::HostUnreachable, errmsg);
             }
 
-            if (isInternalAuthSet() && !con->authenticateInternalUser()) {
-                return Status(ErrorCodes::AuthenticationFailed,
-                              "Unable to authenticate as internal user");
+            if (auth::isInternalAuthSet()) {
+                auto authStatus = con->authenticateInternalUser();
+                if (!authStatus.isOK()) {
+                    return authStatus;
+                }
             }
 
             _conn = std::move(con);
diff --git a/src/mongo/db/commands/parameters.cpp b/src/mongo/db/commands/parameters.cpp
index d1ee7290269..846a0f3f464 100644
--- a/src/mongo/db/commands/parameters.cpp
+++ b/src/mongo/db/commands/parameters.cpp
@@ -41,7 +41,6 @@
 #include "mongo/client/replica_set_monitor.h"
 #include "mongo/config.h"
 #include "mongo/db/auth/authorization_manager.h"
-#include "mongo/db/auth/internal_user_auth.h"
 #include "mongo/db/command_generic_argument.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/server_parameters.h"
diff --git a/src/mongo/db/initialize_server_global_state.cpp b/src/mongo/db/initialize_server_global_state.cpp
index 0c7a70b67b9..069e4163692 100644
--- a/src/mongo/db/initialize_server_global_state.cpp
+++ b/src/mongo/db/initialize_server_global_state.cpp
@@ -47,9 +47,9 @@
 #endif
 
 #include "mongo/base/init.h"
+#include "mongo/client/authenticate.h"
 #include "mongo/config.h"
 #include "mongo/db/auth/authorization_manager.h"
-#include "mongo/db/auth/internal_user_auth.h"
 #include "mongo/db/auth/sasl_command_constants.h"
 #include "mongo/db/auth/security_key.h"
 #include "mongo/db/server_options.h"
@@ -407,7 +407,7 @@ bool initializeServerGlobalState(ServiceContext* service) {
 #ifdef MONGO_CONFIG_SSL
     if (clusterAuthMode == ServerGlobalParams::ClusterAuthMode_x509 ||
         clusterAuthMode == ServerGlobalParams::ClusterAuthMode_sendX509) {
-        setInternalUserAuthParams(
+        auth::setInternalUserAuthParams(
             BSON(saslCommandMechanismFieldName
                  << "MONGODB-X509"
                  << saslCommandUserDBFieldName
diff --git a/src/mongo/db/repl/SConscript b/src/mongo/db/repl/SConscript
index f22ce3144ca..f3c78dcd285 100644
--- a/src/mongo/db/repl/SConscript
+++ b/src/mongo/db/repl/SConscript
@@ -49,7 +49,6 @@ env.Library(
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/base',
         '$BUILD_DIR/mongo/client/clientdriver_network',
-        '$BUILD_DIR/mongo/db/auth/internal_user_auth',
         '$BUILD_DIR/mongo/db/auth/authorization_manager_global',
         '$BUILD_DIR/mongo/util/net/network',
     ],
@@ -548,7 +547,6 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/db/auth/authorization_manager_global',
-        '$BUILD_DIR/mongo/db/auth/internal_user_auth',
     ],
 )
 
diff --git a/src/mongo/db/repl/isself.cpp b/src/mongo/db/repl/isself.cpp
index dd3cc4e7423..a8f77bc6378 100644
--- a/src/mongo/db/repl/isself.cpp
+++ b/src/mongo/db/repl/isself.cpp
@@ -38,10 +38,10 @@
 
 #include "mongo/base/init.h"
 #include "mongo/bson/util/builder.h"
+#include "mongo/client/authenticate.h"
 #include "mongo/client/dbclient_connection.h"
 #include "mongo/db/auth/action_set.h"
 #include "mongo/db/auth/action_type.h"
-#include "mongo/db/auth/internal_user_auth.h"
 #include "mongo/db/auth/privilege.h"
 #include "mongo/db/commands.h"
 #include "mongo/db/service_context.h"
@@ -202,7 +202,7 @@ bool isSelf(const HostAndPort& hostAndPort, ServiceContext* const ctx) {
             return false;
         }
 
-        if (isInternalAuthSet() && !conn.authenticateInternalUser()) {
+        if (auth::isInternalAuthSet() && !conn.authenticateInternalUser().isOK()) {
             return false;
         }
         BSONObj out;
diff --git a/src/mongo/db/repl/oplogreader.cpp b/src/mongo/db/repl/oplogreader.cpp
index 356ffa68afe..86e0d342bc9 100644
--- a/src/mongo/db/repl/oplogreader.cpp
+++ b/src/mongo/db/repl/oplogreader.cpp
@@ -36,9 +36,9 @@
 
 #include <string>
 
+#include "mongo/client/authenticate.h"
 #include "mongo/db/auth/authorization_manager.h"
 #include "mongo/db/auth/authorization_session.h"
-#include "mongo/db/auth/internal_user_auth.h"
 #include "mongo/executor/network_interface.h"
 #include "mongo/util/log.h"
 
@@ -58,8 +58,8 @@ AuthorizationManager* getGlobalAuthorizationManager() {
 }  // namespace
 
 bool replAuthenticate(DBClientBase* conn) {
-    if (isInternalAuthSet())
-        return conn->authenticateInternalUser();
+    if (auth::isInternalAuthSet())
+        return conn->authenticateInternalUser().isOK();
     if (getGlobalAuthorizationManager()->isAuthEnabled())
         return false;
     return true;
diff --git a/src/mongo/db/sessions_collection_rs.cpp b/src/mongo/db/sessions_collection_rs.cpp
index 98e6f5ffe1c..de03a1d65f4 100644
--- a/src/mongo/db/sessions_collection_rs.cpp
+++ b/src/mongo/db/sessions_collection_rs.cpp
@@ -35,11 +35,11 @@
 #include <boost/optional.hpp>
 #include <utility>
 
+#include "mongo/client/authenticate.h"
 #include "mongo/client/connection_string.h"
 #include "mongo/client/query.h"
 #include "mongo/client/read_preference.h"
 #include "mongo/client/remote_command_targeter_factory_impl.h"
-#include "mongo/db/auth/internal_user_auth.h"
 #include "mongo/db/concurrency/d_concurrency.h"
 #include "mongo/db/dbdirectclient.h"
 #include "mongo/db/operation_context.h"
@@ -75,8 +75,8 @@ Status makePrimaryConnection(OperationContext* opCtx, boost::optional<ScopedDbCo
     // Make a connection to the primary, auth, then send
     try {
         conn->emplace(hostname);
-        if (isInternalAuthSet()) {
-            (*conn)->get()->auth(getInternalUserAuthParams());
+        if (auth::isInternalAuthSet()) {
+            uassertStatusOK((*conn)->get()->authenticateInternalUser());
         }
         return Status::OK();
     } catch (...) {
diff --git a/src/mongo/db/startup_warnings_common.cpp b/src/mongo/db/startup_warnings_common.cpp
index 938c66f1320..cfb73e01ae0 100644
--- a/src/mongo/db/startup_warnings_common.cpp
+++ b/src/mongo/db/startup_warnings_common.cpp
@@ -37,8 +37,8 @@
 #include <boost/filesystem/operations.hpp>
 #include <fstream>
 
+#include "mongo/client/authenticate.h"
 #include "mongo/config.h"
-#include "mongo/db/auth/internal_user_auth.h"
 #include "mongo/db/server_options.h"
 #include "mongo/util/log.h"
 #include "mongo/util/net/ssl_options.h"
@@ -137,7 +137,7 @@ void logCommonStartupWarnings(const ServerGlobalParams& serverParams) {
         warned = true;
     }
 
-    if (!getInternalUserAuthParams(1).isEmpty()) {
+    if (auth::hasMultipleInternalAuthKeys()) {
         log() << startupWarningsLog;
         log() << "** WARNING: Multiple keys specified in security key file. If cluster key file"
               << startupWarningsLog;
diff --git a/src/mongo/executor/SConscript b/src/mongo/executor/SConscript
index af9be6312b1..aab6a91626f 100644
--- a/src/mongo/executor/SConscript
+++ b/src/mongo/executor/SConscript
@@ -138,7 +138,6 @@ env.Library(
         '$BUILD_DIR/mongo/transport/transport_layer',
     ],
     LIBDEPS_PRIVATE=[
-        '$BUILD_DIR/mongo/db/auth/internal_user_auth',
         '$BUILD_DIR/mongo/db/commands/test_commands_enabled',
         '$BUILD_DIR/mongo/transport/transport_layer_manager',
         'connection_pool_executor',
diff --git a/src/mongo/executor/connection_pool_tl.cpp b/src/mongo/executor/connection_pool_tl.cpp
index 7dd943649c1..624e3fe3e6b 100644
--- a/src/mongo/executor/connection_pool_tl.cpp
+++ b/src/mongo/executor/connection_pool_tl.cpp
@@ -34,7 +34,8 @@
 
 #include "mongo/executor/connection_pool_tl.h"
 
-#include "mongo/db/auth/internal_user_auth.h"
+#include "mongo/client/authenticate.h"
+#include "mongo/db/auth/authorization_manager.h"
 #include "mongo/util/log.h"
 
 namespace mongo {
@@ -157,6 +158,67 @@ void TLConnection::cancelTimeout() {
     _timer->cancelTimeout();
 }
 
+class TLConnectionSetupHook : public executor::NetworkConnectionHook {
+public:
+    explicit TLConnectionSetupHook(executor::NetworkConnectionHook* hookToWrap)
+        : _wrappedHook(hookToWrap) {}
+
+    BSONObj augmentIsMasterRequest(BSONObj cmdObj) override {
+        BSONObjBuilder bob(std::move(cmdObj));
+        bob.append("hangUpOnStepDown", false);
+        if (internalSecurity.user) {
+            bob.append("saslSupportedMechs", internalSecurity.user->getName().getUnambiguousName());
+        }
+
+        return bob.obj();
+    }
+
+    Status validateHost(const HostAndPort& remoteHost,
+                        const BSONObj& isMasterRequest,
+                        const RemoteCommandResponse& isMasterReply) override try {
+        const auto saslMechsElem = isMasterReply.data.getField("saslSupportedMechs");
+        if (saslMechsElem.type() == Array) {
+            auto array = saslMechsElem.Array();
+            for (const auto& elem : array) {
+                _saslMechsForInternalAuth.push_back(elem.checkAndGetStringData().toString());
+            }
+        }
+
+        if (!_wrappedHook) {
+            return Status::OK();
+        } else {
+            return _wrappedHook->validateHost(remoteHost, isMasterRequest, isMasterReply);
+        }
+    } catch (const DBException& e) {
+        return e.toStatus();
+    }
+
+    StatusWith<boost::optional<RemoteCommandRequest>> makeRequest(
+        const HostAndPort& remoteHost) final {
+        if (_wrappedHook) {
+            return _wrappedHook->makeRequest(remoteHost);
+        } else {
+            return boost::none;
+        }
+    }
+
+    Status handleReply(const HostAndPort& remoteHost, RemoteCommandResponse&& response) final {
+        if (_wrappedHook) {
+            return _wrappedHook->handleReply(remoteHost, std::move(response));
+        } else {
+            return Status::OK();
+        }
+    }
+
+    const std::vector<std::string>& saslMechsForInternalAuth() const {
+        return _saslMechsForInternalAuth;
+    }
+
+private:
+    std::vector<std::string> _saslMechsForInternalAuth;
+    executor::NetworkConnectionHook* const _wrappedHook = nullptr;
+};
+
 void TLConnection::setup(Milliseconds timeout, SetupCallback cb) {
     auto anchor = shared_from_this();
 
@@ -180,28 +242,21 @@ void TLConnection::setup(Milliseconds timeout, SetupCallback cb) {
         }
     });
 
+    auto isMasterHook = std::make_shared<TLConnectionSetupHook>(_onConnectHook);
+
     AsyncDBClient::connect(_peer, transport::kGlobalSSLMode, _serviceContext, _reactor, timeout)
         .onError([](StatusWith<AsyncDBClient::Handle> swc) -> StatusWith<AsyncDBClient::Handle> {
             return Status(ErrorCodes::HostUnreachable, swc.getStatus().reason());
         })
-        .then([this](AsyncDBClient::Handle client) {
+        .then([this, isMasterHook](AsyncDBClient::Handle client) {
             _client = std::move(client);
-            return _client->initWireVersion("NetworkInterfaceTL", _onConnectHook);
+            return _client->initWireVersion("NetworkInterfaceTL", isMasterHook.get());
         })
-        .then([this] {
-            // Try to authenticate with the default system credentials
-            return _client->authenticate(getInternalUserAuthParams())
-                .onError([this](Status status) -> Future<void> {
-                    // If we're in the middle of a keyfile rollover, there may be alternate
-                    // credentials to try.
-                    const auto altParams = getInternalUserAuthParams(1);
-                    if (!altParams.isEmpty() && status == ErrorCodes::AuthenticationFailed) {
-                        return _client->authenticate(altParams);
-                    } else {
-                        // If there weren't alternate credentials, the original error stands.
-                        return status;
-                    }
-                });
+        .then([this, isMasterHook] {
+            boost::optional<std::string> mechanism;
+            if (!isMasterHook->saslMechsForInternalAuth().empty())
+                mechanism = isMasterHook->saslMechsForInternalAuth().front();
+            return _client->authenticateInternal(std::move(mechanism));
         })
         .then([this] {
             if (!_onConnectHook) {
diff --git a/src/mongo/executor/network_connection_hook.h b/src/mongo/executor/network_connection_hook.h
index 51d18fceb84..c84f4054cf2 100644
--- a/src/mongo/executor/network_connection_hook.h
+++ b/src/mongo/executor/network_connection_hook.h
@@ -55,6 +55,15 @@ public:
     virtual ~NetworkConnectionHook() = default;
 
     /**
+     * Optionally augments the isMaster request sent while initializing the wire protocol.
+     *
+     * By default this will just return the cmdObj passed in unaltered.
+     */
+    virtual BSONObj augmentIsMasterRequest(BSONObj cmdObj) {
+        return cmdObj;
+    }
+
+    /**
      * Runs optional validation logic on an isMaster reply from a remote host. If a non-OK
      * Status is returned, it will be propagated up to the completion handler for the command
      * that initiated the request that caused this connection to be created. This will
diff --git a/src/mongo/s/SConscript b/src/mongo/s/SConscript
index f23302d8ed4..d1a922521e3 100644
--- a/src/mongo/s/SConscript
+++ b/src/mongo/s/SConscript
@@ -106,7 +106,6 @@ env.Library(
         'sharding_legacy_api',
     ],
     LIBDEPS_PRIVATE=[
-        '$BUILD_DIR/mongo/db/auth/internal_user_auth',
         '$BUILD_DIR/mongo/executor/thread_pool_task_executor',
         'coreshard',
         'sharding_task_executor',
diff --git a/src/mongo/s/client/sharding_connection_hook.cpp b/src/mongo/s/client/sharding_connection_hook.cpp
index ad3260b3732..1ca0d1b4e02 100644
--- a/src/mongo/s/client/sharding_connection_hook.cpp
+++ b/src/mongo/s/client/sharding_connection_hook.cpp
@@ -37,7 +37,7 @@
 #include <string>
 
 #include "mongo/bson/util/bson_extract.h"
-#include "mongo/db/auth/internal_user_auth.h"
+#include "mongo/client/authenticate.h"
 #include "mongo/db/client.h"
 #include "mongo/rpc/get_status_from_command_result.h"
 #include "mongo/s/client/version_manager.h"
@@ -58,14 +58,12 @@ void ShardingConnectionHook::onCreate(DBClientBase* conn) {
 
     // Authenticate as the first thing we do
     // NOTE: Replica set authentication allows authentication against *any* online host
-    if (isInternalAuthSet()) {
+    if (auth::isInternalAuthSet()) {
         LOG(2) << "calling onCreate auth for " << conn->toString();
 
-        bool result = conn->authenticateInternalUser();
-
-        uassert(15847,
-                str::stream() << "can't authenticate to server " << conn->getServerAddress(),
-                result);
+        uassertStatusOKWithContext(conn->authenticateInternalUser(),
+                                   str::stream() << "can't authenticate to server "
+                                                 << conn->getServerAddress());
     }
 
     // Delegate the metadata hook logic to the egress hook; use lambdas to pass the arguments in
diff --git a/src/mongo/shell/mongodbcr.cpp b/src/mongo/shell/mongodbcr.cpp
index dd11da22587..fb31b310df8 100644
--- a/src/mongo/shell/mongodbcr.cpp
+++ b/src/mongo/shell/mongodbcr.cpp
@@ -38,10 +38,11 @@
 #include "mongo/base/string_data.h"
 #include "mongo/bson/util/bson_extract.h"
 #include "mongo/db/auth/sasl_command_constants.h"
+#include "mongo/rpc/get_status_from_command_result.h"
+#include "mongo/rpc/op_msg.h"
+#include "mongo/rpc/unique_message.h"
 #include "mongo/util/password_digest.h"
 
-using mongo::executor::RemoteCommandRequest;
-
 namespace mongo {
 namespace auth {
 namespace {
@@ -64,20 +65,16 @@ StatusWith<std::string> extractDBField(const BSONObj& params) {
     return std::move(db);
 }
 
-StatusWith<RemoteCommandRequest> createMongoCRGetNonceCmd(const BSONObj& params) {
+StatusWith<OpMsgRequest> createMongoCRGetNonceCmd(const BSONObj& params) {
     auto db = extractDBField(params);
     if (!db.isOK()) {
         return std::move(db.getStatus());
     }
 
-    auto request = RemoteCommandRequest();
-    request.cmdObj = kGetNonceCmd;
-    request.dbname = db.getValue();
-
-    return std::move(request);
+    return OpMsgRequest::fromDBAndBody(db.getValue(), kGetNonceCmd);
 }
 
-RemoteCommandRequest createMongoCRAuthenticateCmd(const BSONObj& params, StringData nonce) {
+OpMsgRequest createMongoCRAuthenticateCmd(const BSONObj& params, StringData nonce) {
     std::string username;
     uassertStatusOK(bsonExtractStringField(params, saslCommandUserFieldName, &username));
 
@@ -93,9 +90,6 @@ RemoteCommandRequest createMongoCRAuthenticateCmd(const BSONObj& params, StringD
         digested = createPasswordDigest(username, password);
     }
 
-    auto request = RemoteCommandRequest();
-    request.dbname = uassertStatusOK(extractDBField(params));
-
     BSONObjBuilder b;
     {
         b << "authenticate" << 1 << "nonce" << nonce << "user" << username;
@@ -109,41 +103,38 @@ RemoteCommandRequest createMongoCRAuthenticateCmd(const BSONObj& params, StringD
             md5_finish(&st, d);
         }
         b << "key" << digestToString(d);
-        request.cmdObj = b.obj();
     }
-    return request;
+
+    return OpMsgRequest::fromDBAndBody(uassertStatusOK(extractDBField(params)), b.obj());
 }
 
-void authMongoCRImpl(RunCommandHook runCommand,
-                     const BSONObj& params,
-                     AuthCompletionHandler handler) {
+Future<void> authMongoCRImpl(RunCommandHook runCommand, const BSONObj& params) {
     invariant(runCommand);
-    invariant(handler);
 
     // Step 1: send getnonce command, receive nonce
     auto nonceRequest = createMongoCRGetNonceCmd(params);
-    if (!nonceRequest.isOK())
-        return handler(std::move(nonceRequest.getStatus()));
+    if (!nonceRequest.isOK()) {
+        return nonceRequest.getStatus();
+    }
 
-    runCommand(nonceRequest.getValue(), [runCommand, params, handler](AuthResponse response) {
-        if (!response.isOK())
-            return handler(std::move(response));
+    return runCommand(nonceRequest.getValue())
+        .then([runCommand, params](BSONObj nonceResponse) -> Future<void> {
+            auto status = getStatusFromCommandResult(nonceResponse);
+            if (!status.isOK()) {
+                return status;
+            }
 
-        try {
             // Ensure response was valid
             std::string nonce;
-            BSONObj nonceResponse = response.data;
             auto valid = bsonExtractStringField(nonceResponse, "nonce", &nonce);
             if (!valid.isOK())
-                return handler({ErrorCodes::AuthenticationFailed,
-                                "Invalid nonce response: " + nonceResponse.toString()});
+                return Status(ErrorCodes::AuthenticationFailed,
+                              "Invalid nonce response: " + nonceResponse.toString());
 
-            // Step 2: send authenticate command, receive response
-            runCommand(createMongoCRAuthenticateCmd(params, nonce), handler);
-        } catch (const DBException& e) {
-            return handler(e.toStatus());
-        }
-    });
+            return runCommand(createMongoCRAuthenticateCmd(params, nonce)).then([](BSONObj reply) {
+                return getStatusFromCommandResult(reply);
+            });
+        });
 }
 
 MONGO_INITIALIZER(RegisterAuthMongoCR)(InitializerContext* context) {
diff --git a/src/mongo/util/net/SConscript b/src/mongo/util/net/SConscript
index bcbd108005a..9141e6eea81 100644
--- a/src/mongo/util/net/SConscript
+++ b/src/mongo/util/net/SConscript
@@ -121,8 +121,8 @@ env.Library(
     ],
     LIBDEPS_PRIVATE=[
         '$BUILD_DIR/mongo/base/secure_allocator',
+        '$BUILD_DIR/mongo/client/authentication',
         '$BUILD_DIR/mongo/crypto/sha_block_${MONGO_CRYPTO}',
-        '$BUILD_DIR/mongo/db/auth/internal_user_auth',
         '$BUILD_DIR/mongo/db/commands/server_status',
         '$BUILD_DIR/mongo/db/server_options_core',
         '$BUILD_DIR/mongo/db/service_context',
diff --git a/src/mongo/util/net/ssl_parameters.cpp b/src/mongo/util/net/ssl_parameters.cpp
index 2172cab7913..e8cfc506b1c 100644
--- a/src/mongo/util/net/ssl_parameters.cpp
+++ b/src/mongo/util/net/ssl_parameters.cpp
@@ -30,8 +30,8 @@
 
 #include "mongo/platform/basic.h"
 
+#include "mongo/client/authenticate.h"
 #include "mongo/config.h"
-#include "mongo/db/auth/internal_user_auth.h"
 #include "mongo/db/auth/sasl_command_constants.h"
 #include "mongo/db/server_options.h"
 #include "mongo/util/net/ssl_options.h"
@@ -204,9 +204,9 @@ mongo::Status mongo::setClusterAuthModeFromString(StringData strMode) {
         }
         serverGlobalParams.clusterAuthMode.store(mode);
 #ifdef MONGO_CONFIG_SSL
-        setInternalUserAuthParams(BSON(saslCommandMechanismFieldName << "MONGODB-X509"
-                                                                     << saslCommandUserDBFieldName
-                                                                     << "$external"));
+        auth::setInternalUserAuthParams(
+            BSON(saslCommandMechanismFieldName << "MONGODB-X509" << saslCommandUserDBFieldName
+                                               << "$external"));
 #endif
     } else if ((mode == ServerGlobalParams::ClusterAuthMode_x509) &&
                (oldMode == ServerGlobalParams::ClusterAuthMode_sendX509)) {