SERVER-61614 Add TenantID to AuthName

10 files changed, 344 insertions, 212 deletions

diff --git a/src/mongo/db/auth/SConscript b/src/mongo/db/auth/SConscript
index e62251887c6..d455ef5333d 100644
--- a/src/mongo/db/auth/SConscript
+++ b/src/mongo/db/auth/SConscript
@@ -116,6 +116,7 @@ env.Library(
         '$BUILD_DIR/mongo/db/catalog/collection_catalog',
         '$BUILD_DIR/mongo/db/concurrency/lock_manager',
         '$BUILD_DIR/mongo/db/db_raii',
+        '$BUILD_DIR/mongo/db/multitenancy',
         'auth_types',
     ]
 )
@@ -466,11 +467,13 @@ env.Library(
         'user_cache_invalidator_job_parameters.idl',
         ],
     LIBDEPS=[
-        '$BUILD_DIR/mongo/s/coreshard',
-        'authservercommon',
+        '$BUILD_DIR/mongo/base',
     ],
     LIBDEPS_PRIVATE=[
+        '$BUILD_DIR/mongo/db/multitenancy',
         '$BUILD_DIR/mongo/idl/server_parameter',
+        '$BUILD_DIR/mongo/s/coreshard',
+        'authservercommon',
     ],
 )
 
diff --git a/src/mongo/db/auth/auth_identifier_test.cpp b/src/mongo/db/auth/auth_identifier_test.cpp
index bb3b6e2f313..55d6fc98a25 100644
--- a/src/mongo/db/auth/auth_identifier_test.cpp
+++ b/src/mongo/db/auth/auth_identifier_test.cpp
@@ -33,6 +33,7 @@
 
 #include "mongo/platform/basic.h"
 
+#include <boost/optional/optional_io.hpp>
 #include <string>
 
 #include "mongo/base/status.h"
@@ -61,13 +62,17 @@ std::string stream(const T& obj) {
 }
 
 template <typename T, typename Name, typename Db>
-void checkValueAssertions(const T& obj, Name name, Db db) {
-    const bool expectEmpty = StringData(name).empty() && StringData(db).empty();
+void checkValueAssertions(const T& obj,
+                          Name name,
+                          Db db,
+                          const boost::optional<OID>& tenant = boost::none) {
+    const bool expectEmpty = StringData(name).empty() && StringData(db).empty() && !tenant;
     ASSERT_EQ(obj.empty(), expectEmpty);
 
     ASSERT_EQ(obj.getDB(), db);
     ASSERT_EQ(obj.getName(), name);
     ASSERT_EQ(getName(obj), name);
+    ASSERT_EQ(obj.getTenant(), tenant);
 
     std::string expectDisplay, expectUnique;
     if (!expectEmpty) {
@@ -78,6 +83,12 @@ void checkValueAssertions(const T& obj, Name name, Db db) {
     ASSERT_EQ(stream<StringBuilder>(obj), expectDisplay);
     ASSERT_EQ(stream<std::ostringstream>(obj), expectDisplay);
     ASSERT_EQ(obj.getUnambiguousName(), expectUnique);
+
+    T same(name, db, tenant);
+    ASSERT_EQ(obj, same);
+
+    T bigger("zzzz", "zzzz", tenant);
+    ASSERT_LT(obj, bigger);
 }
 
 template <typename T>
@@ -101,10 +112,16 @@ TEST(AuthName, ConstructorTest) {
 
 template <typename T, typename Name, typename Db>
 void doBSONParseTest(Name name, Db db) {
+    // Without TenantID.
     auto obj = BSON(T::kFieldName << name << "db" << db);
     checkValueAssertions(T::parseFromBSON(BSON("" << obj).firstElement()), name, db);
-
     checkValueAssertions(T::parseFromBSONObj(obj), name, db);
+
+    // With TenantID.
+    const auto tenant = OID::gen();
+    auto tobj = BSON(T::kFieldName << name << "db" << db << "tenant" << tenant);
+    checkValueAssertions(T::parseFromBSON(BSON("" << tobj).firstElement()), name, db, tenant);
+    checkValueAssertions(T::parseFromBSONObj(tobj), name, db, tenant);
 }
 
 template <typename T, typename Name, typename Db>
diff --git a/src/mongo/db/auth/auth_name.cpp b/src/mongo/db/auth/auth_name.cpp
index dc15c685485..acde60ea5c1 100644
--- a/src/mongo/db/auth/auth_name.cpp
+++ b/src/mongo/db/auth/auth_name.cpp
@@ -33,9 +33,12 @@
 #include "mongo/db/auth/user_name.h"
 
 namespace mongo {
+namespace {
+constexpr auto kTenantFieldName = "tenant"_sd;
+}  // namespace
 
 template <typename T>
-StatusWith<T> AuthName<T>::parse(StringData str) {
+StatusWith<T> AuthName<T>::parse(StringData str, const boost::optional<OID>& tenant) {
     auto split = str.find('.');
 
     if (split == std::string::npos) {
@@ -44,51 +47,62 @@ StatusWith<T> AuthName<T>::parse(StringData str) {
                                     << T::kFieldName << " pair");
     }
 
-    return T(str.substr(split + 1), str.substr(0, split));
+    return T(str.substr(split + 1), str.substr(0, split), tenant);
 }
 
 template <typename T>
-T AuthName<T>::parseFromVariant(const stdx::variant<std::string, BSONObj>& name) {
+T AuthName<T>::parseFromVariant(const stdx::variant<std::string, BSONObj>& name,
+                                const boost::optional<OID>& tenant) {
     if (stdx::holds_alternative<std::string>(name)) {
         return uassertStatusOK(parse(stdx::get<std::string>(name)));
     }
 
-    return parseFromBSONObj(stdx::get<BSONObj>(name));
+    return parseFromBSONObj(stdx::get<BSONObj>(name), tenant);
 }
 
 template <typename T>
-T AuthName<T>::parseFromBSONObj(const BSONObj& obj) {
-    std::bitset<2> usedFields;
+T AuthName<T>::parseFromBSONObj(const BSONObj& obj, const boost::optional<OID>& activeTenant) {
+    std::bitset<3> usedFields;
     constexpr size_t kNameFieldBit = 0;
     constexpr size_t kDbFieldBit = 1;
+    constexpr size_t kTenantFieldBit = 2;
     StringData name, db;
+    boost::optional<OID> tenant = activeTenant;
+
+    const auto validateField = [&](const BSONElement& elem, const size_t bit, BSONType expType) {
+        const auto fieldName = elem.fieldNameStringData();
+        uassert(ErrorCodes::BadValue,
+                str::stream() << T::kName << " must contain a " << typeName(expType)
+                              << " field named: " << fieldName,
+                elem.type() == expType);
+        uassert(ErrorCodes::BadValue,
+                str::stream() << T::kName << " has more than one field named: " << fieldName,
+                !usedFields[bit]);
+        usedFields.set(bit);
+    };
 
     for (const auto& element : obj) {
         const auto fieldName = element.fieldNameStringData();
 
         if (fieldName == T::kFieldName) {
-            uassert(ErrorCodes::BadValue,
-                    str::stream() << T::kName
-                                  << " must contain a string field named: " << T::kFieldName,
-                    element.type() == String);
-            uassert(ErrorCodes::BadValue,
-                    str::stream() << T::kName
-                                  << " has more than one field named: " << T::kFieldName,
-                    !usedFields[kNameFieldBit]);
-
-            usedFields.set(kNameFieldBit);
+            validateField(element, kNameFieldBit, String);
             name = element.valueStringData();
+
         } else if (fieldName == "db"_sd) {
-            uassert(ErrorCodes::BadValue,
-                    str::stream() << T::kName
-                                  << " must contain a string field named: " << T::kFieldName,
-                    element.type() == String);
-            uassert(ErrorCodes::BadValue,
-                    str::stream() << T::kName << " has more than one field named: db",
-                    !usedFields[kDbFieldBit]);
-
-            usedFields.set(kDbFieldBit);
+            validateField(element, kDbFieldBit, String);
             db = element.valueStringData();
+
+        } else if (fieldName == kTenantFieldName) {
+            validateField(element, kTenantFieldBit, jstOID);
+            tenant = element.OID();
+            if (activeTenant) {
+                uassert(ErrorCodes::BadValue,
+                        str::stream()
+                            << T::kName
+                            << " contains a TenantID which does not match the active tenant",
+                        tenant == activeTenant);
+            }
+
         } else if constexpr (std::is_same_v<UserName, T>) {
             // Only UserName is strict, RoleName is non-strict.
             uasserted(ErrorCodes::BadValue,
@@ -105,16 +119,16 @@ T AuthName<T>::parseFromBSONObj(const BSONObj& obj) {
             str::stream() << T::kName << " must contain a field named: db",
             usedFields[kDbFieldBit]);
 
-    return T(name, db);
+    return T(name, db, tenant);
 }
 
 template <typename T>
-T AuthName<T>::parseFromBSON(const BSONElement& elem) {
+T AuthName<T>::parseFromBSON(const BSONElement& elem, const boost::optional<OID>& activeTenant) {
     if (elem.type() == String) {
-        return uassertStatusOK(parse(elem.valueStringData()));
+        return uassertStatusOK(parse(elem.valueStringData(), activeTenant));
     } else if (elem.type() == Object) {
         const auto obj = elem.embeddedObject();
-        return parseFromBSONObj(obj);
+        return parseFromBSONObj(obj, activeTenant);
     } else {
         uasserted(ErrorCodes::BadValue,
                   str::stream() << T::kName << " must be either a string or an object");
@@ -134,14 +148,17 @@ void AuthName<T>::serializeToBSON(BSONArrayBuilder* bob) const {
 }
 
 template <typename T>
-void AuthName<T>::appendToBSON(BSONObjBuilder* bob) const {
+void AuthName<T>::appendToBSON(BSONObjBuilder* bob, bool encodeTenant) const {
     *bob << T::kFieldName << getName() << "db"_sd << getDB();
+    if (encodeTenant && _tenant) {
+        *bob << kTenantFieldName << _tenant.get();
+    }
 }
 
 template <typename T>
-BSONObj AuthName<T>::toBSON() const {
+BSONObj AuthName<T>::toBSON(bool encodeTenant) const {
     BSONObjBuilder bob;
-    appendToBSON(&bob);
+    appendToBSON(&bob, encodeTenant);
     return bob.obj();
 }
 
diff --git a/src/mongo/db/auth/auth_name.h b/src/mongo/db/auth/auth_name.h
index f00300fec36..db6027e53d2 100644
--- a/src/mongo/db/auth/auth_name.h
+++ b/src/mongo/db/auth/auth_name.h
@@ -29,16 +29,17 @@
 
 #pragma once
 
+#include <boost/optional.hpp>
 #include <iosfwd>
 #include <memory>
 #include <string>
 
-
 #include "mongo/base/clonable_ptr.h"
 #include "mongo/base/status_with.h"
 #include "mongo/base/string_data.h"
 #include "mongo/bson/bsonelement.h"
 #include "mongo/bson/bsonobjbuilder.h"
+#include "mongo/bson/oid.h"
 #include "mongo/stdx/variant.h"
 
 namespace mongo {
@@ -54,7 +55,7 @@ public:
     AuthName() = default;
 
     template <typename Name, typename DB>
-    AuthName(Name name, DB db) {
+    AuthName(Name name, DB db, boost::optional<OID> tenant = boost::none) {
         if constexpr (std::is_same_v<Name, std::string>) {
             _name = std::move(name);
         } else {
@@ -66,23 +67,27 @@ public:
         } else {
             _db = StringData(db).toString();
         }
+
+        _tenant = std::move(tenant);
     }
 
     /**
-     * Parses a string of the form "db.name" into an AuthName object.
+     * Parses a string of the form "db.name" into an AuthName object with an optional tenant.
      */
-    static StatusWith<T> parse(StringData str);
+    static StatusWith<T> parse(StringData str, const boost::optional<OID>& tenant = boost::none);
 
     /**
      * These methods support parsing usernames from IDL
      */
-    static T parseFromVariant(const stdx::variant<std::string, mongo::BSONObj>& name);
-    static T parseFromBSONObj(const BSONObj& obj);
-    static T parseFromBSON(const BSONElement& elem);
+    static T parseFromVariant(const stdx::variant<std::string, mongo::BSONObj>& name,
+                              const boost::optional<OID>& tenant = boost::none);
+    static T parseFromBSONObj(const BSONObj& obj, const boost::optional<OID>& tenant = boost::none);
+    static T parseFromBSON(const BSONElement& elem,
+                           const boost::optional<OID>& tenant = boost::none);
     void serializeToBSON(StringData fieldName, BSONObjBuilder* bob) const;
     void serializeToBSON(BSONArrayBuilder* bob) const;
-    void appendToBSON(BSONObjBuilder* bob) const;
-    BSONObj toBSON() const;
+    void appendToBSON(BSONObjBuilder* bob, bool encodeTenant = false) const;
+    BSONObj toBSON(bool encodeTenant = false) const;
 
     /**
      * Gets the name part of a AuthName.
@@ -99,6 +104,13 @@ public:
     }
 
     /**
+     * Gets the TenantID, if any, associated with this AuthName.
+     */
+    const boost::optional<OID>& getTenant() const {
+        return _tenant;
+    }
+
+    /**
      * Gets the full unique name of a user as a string, formatted as "name@db".
      */
     std::string getDisplayName() const {
@@ -129,14 +141,14 @@ public:
     }
 
     /**
-     * True if the username and dbname have not been set.
+     * True if the username, dbname, and tenant have not been set.
      */
     bool empty() const {
-        return _db.empty() && _name.empty();
+        return _db.empty() && _name.empty() && !_tenant;
     }
 
     bool operator==(const AuthName& rhs) const {
-        return (_name == rhs._name) && (_db == rhs._db);
+        return (_name == rhs._name) && (_db == rhs._db) && (_tenant == rhs._tenant);
     }
 
     bool operator!=(const AuthName& rhs) const {
@@ -144,17 +156,28 @@ public:
     }
 
     bool operator<(const AuthName& rhs) const {
-        return (_name < rhs._name) || ((_name == rhs._name) && (_db < rhs._db));
+        if (_tenant != rhs._tenant) {
+            return _tenant < rhs._tenant;
+        } else if (_db != rhs._db) {
+            return _db < rhs._db;
+        } else {
+            return _name < rhs._name;
+        }
     }
 
     template <typename H>
     friend H AbslHashValue(H h, const AuthName& name) {
-        return H::combine(std::move(h), name._db, '.', name._name);
+        auto state = std::move(h);
+        if (name._tenant) {
+            state = H::combine(std::move(state), OID::Hasher()(name._tenant.get()), '_');
+        }
+        return H::combine(std::move(state), name._db, '.', name._name);
     }
 
 private:
     std::string _name;
     std::string _db;
+    boost::optional<OID> _tenant;
 };
 
 template <typename Stream, typename T>
diff --git a/src/mongo/db/auth/authz_manager_external_state_local.cpp b/src/mongo/db/auth/authz_manager_external_state_local.cpp
index 3b71f8dce4d..63a06bcdfde 100644
--- a/src/mongo/db/auth/authz_manager_external_state_local.cpp
+++ b/src/mongo/db/auth/authz_manager_external_state_local.cpp
@@ -40,6 +40,7 @@
 #include "mongo/db/auth/auth_types_gen.h"
 #include "mongo/db/auth/privilege_parser.h"
 #include "mongo/db/auth/user_document_parser.h"
+#include "mongo/db/multitenancy.h"
 #include "mongo/db/operation_context.h"
 #include "mongo/db/server_options.h"
 #include "mongo/db/storage/snapshot_manager.h"
@@ -88,6 +89,25 @@ Status AuthzManagerExternalStateLocal::getStoredAuthorizationVersion(OperationCo
 }
 
 namespace {
+
+// Temporary placeholder pending availability of NamespaceWithTenant.
+NamespaceString getNamespaceWithTenant(const NamespaceString& nss,
+                                       const boost::optional<OID>& tenant) {
+    if (tenant) {
+        return NamespaceString(str::stream() << tenant.get() << '_' << nss.db(), nss.coll());
+    } else {
+        return nss;
+    }
+}
+
+NamespaceString getUsersCollection(const boost::optional<OID>& tenant) {
+    return getNamespaceWithTenant(AuthorizationManager::usersCollectionNamespace, tenant);
+}
+
+NamespaceString getRolesCollection(const boost::optional<OID>& tenant) {
+    return getNamespaceWithTenant(AuthorizationManager::rolesCollectionNamespace, tenant);
+}
+
 void serializeResolvedRoles(BSONObjBuilder* user,
                             const AuthzManagerExternalState::ResolvedRoleData& data,
                             boost::optional<const BSONObj&> roleDoc = boost::none) {
@@ -225,6 +245,8 @@ void handleAuthLocalGetUserFailPoint(const std::vector<RoleName>& directRoles) {
 }
 }  // namespace
 
+// We ignore tenant-specific collections here, since hasAnyPrivilegeDocuments
+// only impacts localhost auth bypass which by definition will be a local user.
 bool AuthzManagerExternalStateLocal::hasAnyPrivilegeDocuments(OperationContext* opCtx) {
     if (_hasAnyPrivilegeDocuments.load()) {
         return true;
@@ -250,14 +272,15 @@ bool AuthzManagerExternalStateLocal::hasAnyPrivilegeDocuments(OperationContext*
     return false;
 }
 
-AuthzManagerExternalStateLocal::RolesLocks::RolesLocks(OperationContext* opCtx) {
+AuthzManagerExternalStateLocal::RolesLocks::RolesLocks(OperationContext* opCtx,
+                                                       const boost::optional<OID>& tenant) {
     if (!storageGlobalParams.disableLockFreeReads) {
         _readLockFree = std::make_unique<AutoReadLockFree>(opCtx);
     } else {
         _adminLock =
             std::make_unique<Lock::DBLock>(opCtx, NamespaceString::kAdminDb, LockMode::MODE_IS);
         _rolesLock = std::make_unique<Lock::CollectionLock>(
-            opCtx, AuthorizationManager::rolesCollectionNamespace, LockMode::MODE_S);
+            opCtx, getRolesCollection(tenant), LockMode::MODE_S);
     }
 }
 
@@ -268,8 +291,8 @@ AuthzManagerExternalStateLocal::RolesLocks::~RolesLocks() {
 }
 
 AuthzManagerExternalStateLocal::RolesLocks AuthzManagerExternalStateLocal::_lockRoles(
-    OperationContext* opCtx) {
-    return AuthzManagerExternalStateLocal::RolesLocks(opCtx);
+    OperationContext* opCtx, const boost::optional<OID>& tenant) {
+    return AuthzManagerExternalStateLocal::RolesLocks(opCtx, tenant);
 }
 
 StatusWith<User> AuthzManagerExternalStateLocal::getUserObject(OperationContext* opCtx,
@@ -278,13 +301,13 @@ StatusWith<User> AuthzManagerExternalStateLocal::getUserObject(OperationContext*
     std::vector<RoleName> directRoles;
     User user(userReq.name);
 
-    auto rolesLock = _lockRoles(opCtx);
+    auto rolesLock = _lockRoles(opCtx, userName.getTenant());
 
     if (!userReq.roles) {
         // Normal path: Acquire a user from the local store by UserName.
         BSONObj userDoc;
-        auto status = findOne(
-            opCtx, AuthorizationManager::usersCollectionNamespace, userName.toBSON(), &userDoc);
+        auto status =
+            findOne(opCtx, getUsersCollection(userName.getTenant()), userName.toBSON(), &userDoc);
         if (!status.isOK()) {
             if (status == ErrorCodes::NoMatchingDocument) {
                 return {ErrorCodes::UserNotFound,
@@ -294,7 +317,9 @@ StatusWith<User> AuthzManagerExternalStateLocal::getUserObject(OperationContext*
             return status;
         }
 
-        uassertStatusOK(V2UserDocumentParser().initializeUserFromUserDocument(userDoc, &user));
+        V2UserDocumentParser userDocParser;
+        userDocParser.setTenantID(userReq.name.getTenant());
+        uassertStatusOK(userDocParser.initializeUserFromUserDocument(userDoc, &user));
         for (auto iter = user.getRoles(); iter.more();) {
             directRoles.push_back(iter.next());
         }
@@ -336,12 +361,12 @@ Status AuthzManagerExternalStateLocal::getUserDescription(OperationContext* opCt
     std::vector<RoleName> directRoles;
     BSONObjBuilder resultBuilder;
 
-    auto rolesLock = _lockRoles(opCtx);
+    auto rolesLock = _lockRoles(opCtx, userName.getTenant());
 
     if (!userReq.roles) {
         BSONObj userDoc;
-        auto status = findOne(
-            opCtx, AuthorizationManager::usersCollectionNamespace, userName.toBSON(), &userDoc);
+        auto status =
+            findOne(opCtx, getUsersCollection(userName.getTenant()), userName.toBSON(), &userDoc);
         if (!status.isOK()) {
             if (status == ErrorCodes::NoMatchingDocument) {
                 return {ErrorCodes::UserNotFound,
@@ -353,6 +378,10 @@ Status AuthzManagerExternalStateLocal::getUserDescription(OperationContext* opCt
 
         directRoles = filterAndMapRole(&resultBuilder, userDoc, ResolveRoleOption::kAll, false);
     } else {
+        uassert(ErrorCodes::BadValue,
+                "Illegal combination of pre-defined roles with tenant identifier",
+                userName.getTenant() == boost::none);
+
         // We are able to artifically construct the external user from the request
         resultBuilder.append("_id", str::stream() << userName.getDB() << '.' << userName.getUser());
         resultBuilder.append("user", userName.getUser());
@@ -385,7 +414,7 @@ Status AuthzManagerExternalStateLocal::rolesExist(OperationContext* opCtx,
     stdx::unordered_set<RoleName> unknownRoles;
     for (const auto& roleName : roleNames) {
         if (!auth::isBuiltinRole(roleName) &&
-            !hasOne(opCtx, AuthorizationManager::rolesCollectionNamespace, roleName.toBSON())) {
+            !hasOne(opCtx, getRolesCollection(roleName.getTenant()), roleName.toBSON())) {
             unknownRoles.insert(roleName);
         }
     }
@@ -407,6 +436,7 @@ StatusWith<ResolvedRoleData> AuthzManagerExternalStateLocal::resolveRoles(
     const bool processRests = option & ResolveRoleOption::kRestrictions;
     const bool walkIndirect = (option & ResolveRoleOption::kDirectOnly) == 0;
 
+    auto optTenant = getActiveTenant(opCtx);
     RoleNameSet inheritedRoles;
     PrivilegeVector inheritedPrivileges;
     RestrictionDocuments::sequence_type inheritedRestrictions;
@@ -426,8 +456,8 @@ StatusWith<ResolvedRoleData> AuthzManagerExternalStateLocal::resolveRoles(
             }
 
             BSONObj roleDoc;
-            auto status = findOne(
-                opCtx, AuthorizationManager::rolesCollectionNamespace, role.toBSON(), &roleDoc);
+            auto status =
+                findOne(opCtx, getRolesCollection(role.getTenant()), role.toBSON(), &roleDoc);
             if (!status.isOK()) {
                 if (status.code() == ErrorCodes::NoMatchingDocument) {
                     LOGV2(5029200, "Role does not exist", "role"_attr = role);
@@ -445,7 +475,7 @@ StatusWith<ResolvedRoleData> AuthzManagerExternalStateLocal::resolveRoles(
                                 << "', expected an array but found " << typeName(elem.type())};
                 }
                 for (const auto& subroleElem : elem.Obj()) {
-                    auto subrole = RoleName::parseFromBSON(subroleElem);
+                    auto subrole = RoleName::parseFromBSON(subroleElem, optTenant);
                     if (visited.count(subrole) || nextFrontier.count(subrole)) {
                         continue;
                     }
@@ -571,8 +601,8 @@ Status AuthzManagerExternalStateLocal::getRolesDescription(
 
                 roleDoc = builtinBuilder.obj();
             } else {
-                auto status = findOne(
-                    opCtx, AuthorizationManager::rolesCollectionNamespace, role.toBSON(), &roleDoc);
+                auto status =
+                    findOne(opCtx, getRolesCollection(role.getTenant()), role.toBSON(), &roleDoc);
                 if (status.code() == ErrorCodes::NoMatchingDocument) {
                     continue;
                 }
@@ -648,7 +678,7 @@ Status AuthzManagerExternalStateLocal::getRoleDescriptionsForDB(
     }
 
     return query(opCtx,
-                 AuthorizationManager::rolesCollectionNamespace,
+                 getRolesCollection(getActiveTenant(opCtx)),
                  BSON(AuthorizationManager::ROLE_DB_FIELD_NAME << dbname),
                  BSONObj(),
                  [&](const BSONObj& roleDoc) {
@@ -680,27 +710,112 @@ Status AuthzManagerExternalStateLocal::getRoleDescriptionsForDB(
  */
 
 namespace {
-enum class AuthzCollection {
-    kNone,
-    kUsers,
-    kRoles,
-    kVersion,
-    kAdmin,
-};
-AuthzCollection _parseAuthzCollection(const NamespaceString& nss) {
-    if (nss == AuthorizationManager::usersCollectionNamespace) {
-        return AuthzCollection::kUsers;
-    } else if (nss == AuthorizationManager::rolesCollectionNamespace) {
-        return AuthzCollection::kRoles;
-    } else if (nss == AuthorizationManager::versionCollectionNamespace) {
-        return AuthzCollection::kVersion;
-    } else if (nss == AuthorizationManager::adminCommandNamespace) {
-        return AuthzCollection::kAdmin;
-    } else {
-        // Some collection we don't care about.
-        return AuthzCollection::kNone;
+class AuthzCollection {
+public:
+    enum class AuthzCollectionType {
+        kNone,
+        kUsers,
+        kRoles,
+        kVersion,
+        kAdmin,
+    };
+
+    AuthzCollection() = default;
+    explicit AuthzCollection(const NamespaceString& nss) {
+        // System-only collections.
+        if (nss == AuthorizationManager::versionCollectionNamespace) {
+            _type = AuthzCollectionType::kVersion;
+            return;
+        }
+
+        if (nss == AuthorizationManager::adminCommandNamespace) {
+            _type = AuthzCollectionType::kAdmin;
+            return;
+        }
+
+        auto db = nss.db();
+        auto coll = nss.coll();
+        if (coll == NamespaceString::kSystemUsers) {
+            if (db == NamespaceString::kAdminDb) {
+                // admin.system.users
+                _type = AuthzCollectionType::kUsers;
+                return;
+            } else if (auto tenant = isAdminDBWithTenant(db)) {
+                // {tenantID}_admin.system.users
+                _type = AuthzCollectionType::kUsers;
+                _tenant = std::move(tenant);
+                return;
+            }
+            return;  // none
+        }
+
+        if (nss == AuthorizationManager::rolesCollectionNamespace) {
+            if (db == NamespaceString::kAdminDb) {
+                // admin.system.roles
+                _type = AuthzCollectionType::kRoles;
+                return;
+            } else if (auto tenant = isAdminDBWithTenant(db)) {
+                // {tenantID}_admin.system.roles
+                _type = AuthzCollectionType::kRoles;
+                _tenant = std::move(tenant);
+            }
+            return;  // none
+        }
     }
-}
+
+    operator bool() const {
+        return _type != AuthzCollectionType::kNone;
+    }
+
+    bool isPrivilegeCollection() const {
+        return (_type == AuthzCollectionType::kUsers) || (_type == AuthzCollectionType::kRoles);
+    }
+
+    AuthzCollectionType getType() const {
+        return _type;
+    }
+
+    const boost::optional<OID>& getTenant() const {
+        return _tenant;
+    }
+
+private:
+    /**
+     * Attempt to parse "{tenant}_admin" into an OID.
+     * Returns boost::none if the db is not in the above format.
+     *
+     * Temporary fixture pending availability of NamespaceWithTenant.
+     */
+    static boost::optional<OID> isAdminDBWithTenant(StringData db) {
+        constexpr std::size_t len =
+            (OID::kOIDSize * 2) + 1 /* '_' */ + NamespaceString::kAdminDb.size();
+        if (db.size() != len) {
+            // Not requisite size.
+            return boost::none;
+        }
+
+        if (db.substr((OID::kOIDSize * 2) + 1) != NamespaceString::kAdminDb) {
+            // Doesn't end with "admin"
+            return boost::none;
+        }
+
+        if (db[OID::kOIDSize * 2] != '_') {
+            // Not delimited by an underscore
+            return boost::none;
+        }
+
+        auto swTenant = OID::parse(db.substr(0, OID::kOIDSize * 2));
+        if (!swTenant.isOK()) {
+            // Not a valid OID
+            return boost::none;
+        }
+
+        return swTenant.getValue();
+    }
+
+    AuthzCollectionType _type = AuthzCollectionType::kNone;
+    boost::optional<OID> _tenant;
+};
 
 constexpr auto kOpInsert = "i"_sd;
 constexpr auto kOpUpdate = "u"_sd;
@@ -712,7 +827,7 @@ void _invalidateUserCache(OperationContext* opCtx,
                           AuthzCollection coll,
                           const BSONObj& o,
                           const BSONObj* o2) {
-    if ((coll == AuthzCollection::kUsers) &&
+    if ((coll.getType() == AuthzCollection::AuthzCollectionType::kUsers) &&
         ((op == kOpInsert) || (op == kOpUpdate) || (op == kOpDelete))) {
         const BSONObj* src = (op == kOpUpdate) ? o2 : &o;
         auto id = (*src)["_id"].str();
@@ -729,7 +844,7 @@ void _invalidateUserCache(OperationContext* opCtx,
             authzManager->invalidateUserCache(opCtx);
             return;
         }
-        UserName userName(id.substr(splitPoint + 1), id.substr(0, splitPoint));
+        UserName userName(id.substr(splitPoint + 1), id.substr(0, splitPoint), coll.getTenant());
         authzManager->invalidateUserByName(opCtx, userName);
     } else {
         authzManager->invalidateUserCache(opCtx);
@@ -743,15 +858,14 @@ void AuthzManagerExternalStateLocal::logOp(OperationContext* opCtx,
                                            const NamespaceString& nss,
                                            const BSONObj& o,
                                            const BSONObj* o2) {
-    auto coll = _parseAuthzCollection(nss);
-    if (coll == AuthzCollection::kNone) {
+    AuthzCollection coll(nss);
+    if (!coll) {
         return;
     }
 
     _invalidateUserCache(opCtx, authzManager, op, coll, o, o2);
 
-    if (((coll == AuthzCollection::kUsers) || (coll == AuthzCollection::kRoles)) &&
-        (op == kOpInsert)) {
+    if (coll.isPrivilegeCollection() && !coll.getTenant() && (op == kOpInsert)) {
         _hasAnyPrivilegeDocuments.store(true);
     }
 }
diff --git a/src/mongo/db/auth/authz_manager_external_state_local.h b/src/mongo/db/auth/authz_manager_external_state_local.h
index 4f2807b6b57..dea8bd7228c 100644
--- a/src/mongo/db/auth/authz_manager_external_state_local.h
+++ b/src/mongo/db/auth/authz_manager_external_state_local.h
@@ -137,7 +137,7 @@ protected:
     class RolesLocks {
     public:
         RolesLocks() = default;
-        RolesLocks(OperationContext*);
+        RolesLocks(OperationContext*, const boost::optional<OID>&);
         ~RolesLocks();
 
     private:
@@ -152,7 +152,7 @@ protected:
      *
      * virtual to allow Mock to not lock anything.
      */
-    virtual RolesLocks _lockRoles(OperationContext* opCtx);
+    virtual RolesLocks _lockRoles(OperationContext* opCtx, const boost::optional<OID>&);
 
 private:
     /**
diff --git a/src/mongo/db/auth/authz_manager_external_state_mock.h b/src/mongo/db/auth/authz_manager_external_state_mock.h
index 19390add9a3..df719d2757a 100644
--- a/src/mongo/db/auth/authz_manager_external_state_mock.h
+++ b/src/mongo/db/auth/authz_manager_external_state_mock.h
@@ -115,7 +115,7 @@ public:
     std::vector<BSONObj> getCollectionContents(const NamespaceString& collectionName);
 
 protected:
-    RolesLocks _lockRoles(OperationContext* opCtx) override {
+    RolesLocks _lockRoles(OperationContext* opCtx, const boost::optional<OID>&) override {
         return RolesLocks();
     }
 
diff --git a/src/mongo/db/auth/authz_manager_external_state_s.cpp b/src/mongo/db/auth/authz_manager_external_state_s.cpp
index 47cf83499bc..6753dffd28c 100644
--- a/src/mongo/db/auth/authz_manager_external_state_s.cpp
+++ b/src/mongo/db/auth/authz_manager_external_state_s.cpp
@@ -39,7 +39,7 @@
 #include "mongo/db/auth/user_document_parser.h"
 #include "mongo/db/auth/user_management_commands_parser.h"
 #include "mongo/db/auth/user_name.h"
-#include "mongo/db/jsobj.h"
+#include "mongo/db/multitenancy.h"
 #include "mongo/db/operation_context.h"
 #include "mongo/rpc/get_status_from_command_result.h"
 #include "mongo/s/grid.h"
@@ -119,7 +119,9 @@ StatusWith<User> AuthzManagerExternalStateMongos::getUserObject(OperationContext
     }
 
     User user(userReq.name);
-    status = V2UserDocumentParser().initializeUserFromUserDocument(userDoc, &user);
+    V2UserDocumentParser dp;
+    dp.setTenantID(getActiveTenant(opCtx));
+    status = dp.initializeUserFromUserDocument(userDoc, &user);
     if (!status.isOK()) {
         return status;
     }
@@ -132,13 +134,10 @@ Status AuthzManagerExternalStateMongos::getUserDescription(OperationContext* opC
                                                            BSONObj* result) {
     const UserName& userName = user.name;
     if (!user.roles) {
-        BSONObj usersInfoCmd = BSON(
-            "usersInfo" << BSON_ARRAY(BSON(AuthorizationManager::USER_NAME_FIELD_NAME
-                                           << userName.getUser()
-                                           << AuthorizationManager::USER_DB_FIELD_NAME
-                                           << userName.getDB()))
-                        << "showPrivileges" << true << "showCredentials" << true
-                        << "showAuthenticationRestrictions" << true << "showCustomData" << false);
+        BSONObj usersInfoCmd = BSON("usersInfo" << userName.toBSON(true /* serialize tenant */)
+                                                << "showPrivileges" << true << "showCredentials"
+                                                << true << "showAuthenticationRestrictions" << true
+                                                << "showCustomData" << false);
         BSONObjBuilder builder;
         const bool ok = Grid::get(opCtx)->catalogClient()->runUserManagementReadCommand(
             opCtx, "admin", usersInfoCmd, &builder);
diff --git a/src/mongo/db/auth/user_document_parser.cpp b/src/mongo/db/auth/user_document_parser.cpp
index a2510755661..d8320632bf5 100644
--- a/src/mongo/db/auth/user_document_parser.cpp
+++ b/src/mongo/db/auth/user_document_parser.cpp
@@ -105,26 +105,38 @@ bool parseSCRAMCredentials(const BSONElement& credentialsElement,
     return true;
 }
 
-}  // namespace
-
-Status _checkV2RolesArray(const BSONElement& rolesElement) {
+Status _checkV2RolesArray(const BSONElement& rolesElement) try {
     if (rolesElement.eoo()) {
         return _badValue("User document needs 'roles' field to be provided");
     }
     if (rolesElement.type() != Array) {
         return _badValue("'roles' field must be an array");
     }
-    for (BSONObjIterator iter(rolesElement.embeddedObject()); iter.more(); iter.next()) {
-        if ((*iter).type() != Object) {
-            return _badValue("Elements in 'roles' array must objects");
-        }
-        Status status = V2UserDocumentParser::checkValidRoleObject((*iter).Obj());
-        if (!status.isOK())
-            return status;
+    for (const auto& elem : rolesElement.Array()) {
+        uassert(ErrorCodes::UnsupportedFormat,
+                "User document needs values in 'roles' array to be a sub-documents",
+                elem.type() == Object);
+        RoleName::parseFromBSONObj(elem.Obj());
     }
     return Status::OK();
+} catch (const DBException& ex) {
+    return ex.toStatus();
+}
+
+User::UserId extractUserIDFromUserDocument(const BSONObj& doc) {
+    auto userId = doc[AuthorizationManager::USERID_FIELD_NAME];
+    if (userId.isBinData(BinDataType::newUUID)) {
+        auto id = userId.uuid();
+        User::UserId ret;
+        std::copy(id.begin(), id.end(), std::back_inserter(ret));
+        return ret;
+    }
+
+    return User::UserId();
 }
 
+}  // namespace
+
 Status V2UserDocumentParser::checkValidUserDocument(const BSONObj& doc) const {
     auto userIdElement = doc[AuthorizationManager::USERID_FIELD_NAME];
     auto userElement = doc[AuthorizationManager::USER_NAME_FIELD_NAME];
@@ -220,22 +232,6 @@ Status V2UserDocumentParser::checkValidUserDocument(const BSONObj& doc) const {
     return Status::OK();
 }
 
-User::UserId V2UserDocumentParser::extractUserIDFromUserDocument(const BSONObj& doc) const {
-    auto userId = doc[AuthorizationManager::USERID_FIELD_NAME];
-    if (userId.isBinData(BinDataType::newUUID)) {
-        auto id = userId.uuid();
-        User::UserId ret;
-        std::copy(id.begin(), id.end(), std::back_inserter(ret));
-        return ret;
-    }
-
-    return User::UserId();
-}
-
-std::string V2UserDocumentParser::extractUserNameFromUserDocument(const BSONObj& doc) const {
-    return doc[AuthorizationManager::USER_NAME_FIELD_NAME].str();
-}
-
 Status V2UserDocumentParser::initializeUserCredentialsFromUserDocument(
     User* user, const BSONObj& privDoc) const {
     User::CredentialData credentials;
@@ -301,39 +297,6 @@ static Status _extractRoleDocumentElements(const BSONObj& roleObject,
     return Status::OK();
 }
 
-Status V2UserDocumentParser::checkValidRoleObject(const BSONObj& roleObject) {
-    BSONElement roleNameElement;
-    BSONElement roleSourceElement;
-    return _extractRoleDocumentElements(roleObject, &roleNameElement, &roleSourceElement);
-}
-
-Status V2UserDocumentParser::parseRoleName(const BSONObj& roleObject, RoleName* result) {
-    BSONElement roleNameElement;
-    BSONElement roleSourceElement;
-    Status status = _extractRoleDocumentElements(roleObject, &roleNameElement, &roleSourceElement);
-    if (!status.isOK())
-        return status;
-    *result = RoleName(roleNameElement.str(), roleSourceElement.str());
-    return status;
-}
-
-Status V2UserDocumentParser::parseRoleVector(const BSONArray& rolesArray,
-                                             std::vector<RoleName>* result) {
-    std::vector<RoleName> roles;
-    for (BSONObjIterator it(rolesArray); it.more(); it.next()) {
-        if ((*it).type() != Object) {
-            return Status(ErrorCodes::TypeMismatch, "Roles must be objects.");
-        }
-        RoleName role;
-        Status status = parseRoleName((*it).Obj(), &role);
-        if (!status.isOK())
-            return status;
-        roles.push_back(role);
-    }
-    std::swap(*result, roles);
-    return Status::OK();
-}
-
 Status V2UserDocumentParser::initializeAuthenticationRestrictionsFromUserDocument(
     const BSONObj& privDoc, User* user) const {
 
@@ -391,7 +354,7 @@ Status V2UserDocumentParser::initializeAuthenticationRestrictionsFromUserDocumen
 }
 
 Status V2UserDocumentParser::initializeUserRolesFromUserDocument(const BSONObj& privDoc,
-                                                                 User* user) const {
+                                                                 User* user) const try {
     BSONElement rolesElement = privDoc[ROLES_FIELD_NAME];
 
     if (rolesElement.type() != Array) {
@@ -399,27 +362,24 @@ Status V2UserDocumentParser::initializeUserRolesFromUserDocument(const BSONObj&
                       "User document needs 'roles' field to be an array");
     }
 
+    auto rolesArray = rolesElement.Array();
     std::vector<RoleName> roles;
-    for (BSONObjIterator it(rolesElement.Obj()); it.more(); it.next()) {
-        if ((*it).type() != Object) {
-            return Status(ErrorCodes::UnsupportedFormat,
-                          "User document needs values in 'roles' array to be a sub-documents");
-        }
-        BSONObj roleObject = (*it).Obj();
-
-        RoleName role;
-        Status status = parseRoleName(roleObject, &role);
-        if (!status.isOK()) {
-            return status;
-        }
-        roles.push_back(role);
-    }
+    std::transform(
+        rolesArray.begin(), rolesArray.end(), std::back_inserter(roles), [this](const auto& elem) {
+            uassert(ErrorCodes::UnsupportedFormat,
+                    "User document needs values in 'roles' array to be a sub-documents",
+                    elem.type() == Object);
+            return RoleName::parseFromBSONObj(elem.Obj(), this->_tenant);
+        });
     user->setRoles(makeRoleNameIteratorForContainer(roles));
+
     return Status::OK();
+} catch (const DBException& ex) {
+    return ex.toStatus();
 }
 
 Status V2UserDocumentParser::initializeUserIndirectRolesFromUserDocument(const BSONObj& privDoc,
-                                                                         User* user) const {
+                                                                         User* user) const try {
     BSONElement indirectRolesElement = privDoc[INHERITED_ROLES_FIELD_NAME];
 
     if (!indirectRolesElement) {
@@ -431,24 +391,23 @@ Status V2UserDocumentParser::initializeUserIndirectRolesFromUserDocument(const B
                       "User document needs 'inheritedRoles' field to be an array");
     }
 
+    auto rolesArray = indirectRolesElement.Array();
     std::vector<RoleName> indirectRoles;
-    for (BSONObjIterator it(indirectRolesElement.Obj()); it.more(); it.next()) {
-        if ((*it).type() != Object) {
-            return Status(ErrorCodes::UnsupportedFormat,
-                          "User document needs values in 'inheritedRoles'"
-                          " array to be a sub-documents");
-        }
-        BSONObj indirectRoleObject = (*it).Obj();
-
-        RoleName indirectRole;
-        Status status = parseRoleName(indirectRoleObject, &indirectRole);
-        if (!status.isOK()) {
-            return status;
-        }
-        indirectRoles.push_back(indirectRole);
-    }
+    std::transform(
+        rolesArray.begin(),
+        rolesArray.end(),
+        std::back_inserter(indirectRoles),
+        [this](const auto& elem) {
+            uassert(ErrorCodes::UnsupportedFormat,
+                    "User document needs values in 'inheritedRoles' array to be a sub-documents",
+                    elem.type() == Object);
+            return RoleName::parseFromBSONObj(elem.Obj(), this->_tenant);
+        });
     user->setIndirectRoles(makeRoleNameIteratorForContainer(indirectRoles));
+
     return Status::OK();
+} catch (const DBException& ex) {
+    return ex.toStatus();
 }
 
 Status V2UserDocumentParser::initializeUserPrivilegesFromUserDocument(const BSONObj& doc,
@@ -507,7 +466,7 @@ Status V2UserDocumentParser::initializeUserPrivilegesFromUserDocument(const BSON
 
 Status V2UserDocumentParser::initializeUserFromUserDocument(const BSONObj& privDoc,
                                                             User* user) const try {
-    auto userName = extractUserNameFromUserDocument(privDoc);
+    auto userName = privDoc[AuthorizationManager::USER_NAME_FIELD_NAME].str();
     uassert(ErrorCodes::BadValue,
             str::stream() << "User name from privilege document \"" << userName
                           << "\" doesn't match name of provided User \""
diff --git a/src/mongo/db/auth/user_document_parser.h b/src/mongo/db/auth/user_document_parser.h
index 2a7e86fd4f9..18eb6d88ece 100644
--- a/src/mongo/db/auth/user_document_parser.h
+++ b/src/mongo/db/auth/user_document_parser.h
@@ -30,9 +30,8 @@
 #pragma once
 
 #include "mongo/base/status.h"
-#include "mongo/db/auth/action_set.h"
+#include "mongo/bson/oid.h"
 #include "mongo/db/auth/user.h"
-#include "mongo/db/jsobj.h"
 
 namespace mongo {
 
@@ -42,29 +41,30 @@ class V2UserDocumentParser {
 
 public:
     V2UserDocumentParser() {}
-    Status checkValidUserDocument(const BSONObj& doc) const;
 
     /**
-     * Returns Status::OK() iff the given BSONObj describes a valid element from a roles array.
+     * Apply a tenant identifier to every tenant aware object during parsing.
      */
-    static Status checkValidRoleObject(const BSONObj& roleObject);
-
-    static Status parseRoleName(const BSONObj& roleObject, RoleName* result);
+    void setTenantID(boost::optional<OID> tenant) {
+        _tenant = std::move(tenant);
+    }
 
-    static Status parseRoleVector(const BSONArray& rolesArray, std::vector<RoleName>* result);
+    Status checkValidUserDocument(const BSONObj& doc) const;
+    Status initializeUserFromUserDocument(const BSONObj& privDoc, User* user) const;
 
-    std::string extractUserNameFromUserDocument(const BSONObj& doc) const;
-    User::UserId extractUserIDFromUserDocument(const BSONObj& doc) const;
+private:
+    Status initializeUserIndirectRolesFromUserDocument(const BSONObj& doc, User* user) const;
+    Status initializeUserPrivilegesFromUserDocument(const BSONObj& doc, User* user) const;
 
+public:
+    // public for unit testing only.
     Status initializeUserCredentialsFromUserDocument(User* user, const BSONObj& privDoc) const;
-
     Status initializeUserRolesFromUserDocument(const BSONObj& doc, User* user) const;
-    Status initializeUserIndirectRolesFromUserDocument(const BSONObj& doc, User* user) const;
-    Status initializeUserPrivilegesFromUserDocument(const BSONObj& doc, User* user) const;
     Status initializeAuthenticationRestrictionsFromUserDocument(const BSONObj& doc,
                                                                 User* user) const;
 
-    Status initializeUserFromUserDocument(const BSONObj& privDoc, User* user) const;
+private:
+    boost::optional<OID> _tenant;
 };
 
 }  // namespace mongo