SERVER-56444 Grant read/write access on time-series buckets collections to admin roles

14 files changed, 882 insertions, 61 deletions

diff --git a/jstests/auth/list_collections_own_collections.js b/jstests/auth/list_collections_own_collections.js
index 78668de3c26..cb55c7b97dc 100644
--- a/jstests/auth/list_collections_own_collections.js
+++ b/jstests/auth/list_collections_own_collections.js
@@ -26,6 +26,61 @@ const resSystemViews = {
     "name": "system.views",
     "type": "collection"
 };
+const resFooTS = {
+    "name": "foo",
+    "type": "timeseries"
+};
+const resBarTS = {
+    "name": "bar",
+    "type": "timeseries"
+};
+const resSBFoo = {
+    "name": "system.buckets.foo",
+    "type": "collection"
+};
+const resSBBar = {
+    "name": "system.buckets.bar",
+    "type": "collection"
+};
+
+function createTestRoleAndUser(db, roleName, privs) {
+    const admin = db.getSiblingDB("admin");
+    assert.commandWorked(admin.runCommand({createRole: roleName, roles: [], privileges: privs}));
+
+    const userName = "user|" + roleName;
+    assert.commandWorked(
+        db.runCommand({createUser: userName, pwd: "pwd", roles: [{role: roleName, db: "admin"}]}));
+}
+
+function runTestOnRole(db, roleName, expectedColls) {
+    jsTestLog(roleName);
+    const userName = "user|" + roleName;
+    assert(db.auth(userName, "pwd"));
+
+    let res;
+
+    res = db.runCommand({listCollections: 1});
+    assert.commandFailed(res);
+    res = db.runCommand({listCollections: 1, nameOnly: true});
+    assert.commandFailed(res);
+    res = db.runCommand({listCollections: 1, authorizedCollections: true});
+    assert.commandFailed(res);
+
+    res = db.runCommand({listCollections: 1, nameOnly: true, authorizedCollections: true});
+    assert.commandWorked(res);
+    assert.eq(expectedColls.sort(nameSort), res.cursor.firstBatch.sort(nameSort));
+
+    res = db.runCommand(
+        {listCollections: 1, nameOnly: true, authorizedCollections: true, filter: {"name": "foo"}});
+    assert.commandWorked(res);
+    if (roleName.indexOf("Buckets") != -1) {
+        assert.eq([resFooTS], res.cursor.firstBatch);
+    } else {
+        assert.eq([resFoo], res.cursor.firstBatch);
+    }
+
+    db.logout();
+}
 
 function runTestOnConnection(conn) {
     const admin = conn.getDB("admin");
@@ -34,51 +89,42 @@ function runTestOnConnection(conn) {
     assert.commandWorked(admin.runCommand({createUser: "root", pwd: "root", roles: ["root"]}));
     assert(admin.auth("root", "root"));
 
-    function createTestRoleAndUser(roleName, privs) {
-        assert.commandWorked(
-            admin.runCommand({createRole: roleName, roles: [], privileges: privs}));
-
-        const userName = "user|" + roleName;
-        assert.commandWorked(db.runCommand(
-            {createUser: userName, pwd: "pwd", roles: [{role: roleName, db: "admin"}]}));
-    }
-
-    createTestRoleAndUser("roleWithExactNamespacePrivileges", [
+    createTestRoleAndUser(db, "roleWithExactNamespacePrivileges", [
         {resource: {db: dbName, collection: "foo"}, actions: ["find"]},
         {resource: {db: dbName, collection: "bar"}, actions: ["find"]}
     ]);
 
-    createTestRoleAndUser("roleWithExactNamespaceAndSystemPrivileges", [
+    createTestRoleAndUser(db, "roleWithExactNamespaceAndSystemPrivileges", [
         {resource: {db: dbName, collection: "foo"}, actions: ["find"]},
         {resource: {db: dbName, collection: "bar"}, actions: ["find"]},
         {resource: {db: dbName, collection: "system.views"}, actions: ["find"]}
     ]);
 
-    createTestRoleAndUser("roleWithCollectionPrivileges", [
+    createTestRoleAndUser(db, "roleWithCollectionPrivileges", [
         {resource: {db: "", collection: "foo"}, actions: ["find"]},
         {resource: {db: "", collection: "bar"}, actions: ["find"]}
     ]);
 
-    createTestRoleAndUser("roleWithCollectionAndSystemPrivileges", [
+    createTestRoleAndUser(db, "roleWithCollectionAndSystemPrivileges", [
         {resource: {db: "", collection: "foo"}, actions: ["find"]},
         {resource: {db: "", collection: "bar"}, actions: ["find"]},
         {resource: {db: "", collection: "system.views"}, actions: ["find"]}
     ]);
 
-    createTestRoleAndUser("roleWithDatabasePrivileges", [
+    createTestRoleAndUser(db, "roleWithDatabasePrivileges", [
         {resource: {db: dbName, collection: ""}, actions: ["find"]},
     ]);
 
-    createTestRoleAndUser("roleWithDatabaseAndSystemPrivileges", [
+    createTestRoleAndUser(db, "roleWithDatabaseAndSystemPrivileges", [
         {resource: {db: dbName, collection: ""}, actions: ["find"]},
         {resource: {db: dbName, collection: "system.views"}, actions: ["find"]}
     ]);
 
-    createTestRoleAndUser("roleWithAnyNormalResourcePrivileges", [
+    createTestRoleAndUser(db, "roleWithAnyNormalResourcePrivileges", [
         {resource: {db: "", collection: ""}, actions: ["find"]},
     ]);
 
-    createTestRoleAndUser("roleWithAnyNormalResourceAndSystemPrivileges", [
+    createTestRoleAndUser(db, "roleWithAnyNormalResourceAndSystemPrivileges", [
         {resource: {db: "", collection: ""}, actions: ["find"]},
         {resource: {db: "", collection: "system.views"}, actions: ["find"]}
     ]);
@@ -95,49 +141,91 @@ function runTestOnConnection(conn) {
 
     admin.logout();
 
-    function runTestOnRole(roleName, expectedColls) {
-        jsTestLog(roleName);
-        const userName = "user|" + roleName;
-        assert(db.auth(userName, "pwd"));
-
-        let res;
-
-        res = db.runCommand({listCollections: 1});
-        assert.commandFailed(res);
-        res = db.runCommand({listCollections: 1, nameOnly: true});
-        assert.commandFailed(res);
-        res = db.runCommand({listCollections: 1, authorizedCollections: true});
-        assert.commandFailed(res);
-
-        res = db.runCommand({listCollections: 1, nameOnly: true, authorizedCollections: true});
-        assert.commandWorked(res);
-        assert.eq(expectedColls.sort(nameSort), res.cursor.firstBatch.sort(nameSort));
-
-        res = db.runCommand({
-            listCollections: 1,
-            nameOnly: true,
-            authorizedCollections: true,
-            filter: {"name": "foo"}
-        });
-        assert.commandWorked(res);
-        assert.eq([resFoo], res.cursor.firstBatch);
-
-        db.logout();
-    }
+    runTestOnRole(db, "roleWithExactNamespacePrivileges", [resFoo, resBar]);
+    runTestOnRole(
+        db, "roleWithExactNamespaceAndSystemPrivileges", [resFoo, resBar, resSystemViews]);
 
-    runTestOnRole("roleWithExactNamespacePrivileges", [resFoo, resBar]);
-    runTestOnRole("roleWithExactNamespaceAndSystemPrivileges", [resFoo, resBar, resSystemViews]);
+    runTestOnRole(db, "roleWithCollectionPrivileges", [resFoo, resBar]);
+    runTestOnRole(db, "roleWithCollectionAndSystemPrivileges", [resFoo, resBar, resSystemViews]);
 
-    runTestOnRole("roleWithCollectionPrivileges", [resFoo, resBar]);
-    runTestOnRole("roleWithCollectionAndSystemPrivileges", [resFoo, resBar, resSystemViews]);
+    runTestOnRole(db, "roleWithDatabasePrivileges", [resFoo, resBar, ...resOther]);
+    runTestOnRole(
+        db, "roleWithDatabaseAndSystemPrivileges", [resFoo, resBar, ...resOther, resSystemViews]);
 
-    runTestOnRole("roleWithDatabasePrivileges", [resFoo, resBar, ...resOther]);
-    runTestOnRole("roleWithDatabaseAndSystemPrivileges",
+    runTestOnRole(db, "roleWithAnyNormalResourcePrivileges", [resFoo, resBar, ...resOther]);
+    runTestOnRole(db,
+                  "roleWithAnyNormalResourceAndSystemPrivileges",
                   [resFoo, resBar, ...resOther, resSystemViews]);
+}
 
-    runTestOnRole("roleWithAnyNormalResourcePrivileges", [resFoo, resBar, ...resOther]);
-    runTestOnRole("roleWithAnyNormalResourceAndSystemPrivileges",
-                  [resFoo, resBar, ...resOther, resSystemViews]);
+function runSystemsBucketsTestOnConnection(conn, isMongod) {
+    const admin = conn.getDB("admin");
+    const db = conn.getDB(dbName);
+
+    assert(admin.auth("root", "root"));
+
+    createTestRoleAndUser(db, "roleWithExactNamespacePrivilegesBuckets", [
+        {resource: {db: dbName, collection: "foo"}, actions: ["find"]},
+    ]);
+
+    createTestRoleAndUser(db, "roleWithExactNamespaceAndSystemPrivilegesBuckets", [
+        {resource: {db: dbName, collection: "foo"}, actions: ["find"]},
+        {resource: {db: dbName, collection: "bar"}, actions: ["find"]},
+        {resource: {db: dbName, collection: "system.buckets.foo"}, actions: ["find"]}
+    ]);
+
+    createTestRoleAndUser(db, "roleWithSystemBucketsInAnyDB", [
+        {resource: {db: "", collection: "foo"}, actions: ["find"]},
+        {resource: {db: "", system_buckets: "foo"}, actions: ["find"]},
+        {resource: {db: "", collection: "bar"}, actions: ["find"]}
+    ]);
+
+    createTestRoleAndUser(db, "roleWithAnySystemBucketsInDB", [
+        {resource: {db: dbName, collection: "foo"}, actions: ["find"]},
+        {resource: {db: dbName, system_buckets: ""}, actions: ["find"]},
+        {resource: {db: dbName, collection: "bar"}, actions: ["find"]},
+    ]);
+
+    createTestRoleAndUser(db, "roleWithAnySystemBuckets", [
+        {resource: {db: dbName, collection: "foo"}, actions: ["find"]},
+        {resource: {db: "", system_buckets: ""}, actions: ["find"]},
+        {resource: {db: dbName, collection: "bar"}, actions: ["find"]},
+    ]);
+
+    // Create the collection and view used by the tests.
+    assert.commandWorked(db.dropDatabase());
+    assert.commandWorked(db.createCollection("foo", {timeseries: {timeField: "date"}}));
+    assert.commandWorked(db.createCollection("bar", {timeseries: {timeField: "date"}}));
+
+    // Create a collection and view that are never granted specific permissions, to ensure
+    // they're only returned by listCollections when the role has access to the whole db/server.
+    assert.commandWorked(db.createCollection("otherCollection"));
+    assert.commandWorked(db.createView("otherView", "otherCollection", []));
+
+    admin.logout();
+
+    // TODO SERVER-57558 - mongod bug
+    if (!isMongod) {
+        runTestOnRole(db, "roleWithExactNamespacePrivilegesBuckets", [resFooTS]);
+    }
+    // TODO SERVER-57558 - mongod bug
+    if (!isMongod) {
+        runTestOnRole(
+            db, "roleWithExactNamespaceAndSystemPrivilegesBuckets", [resFooTS, resBarTS, resSBFoo]);
+    } else {
+        runTestOnRole(db, "roleWithExactNamespaceAndSystemPrivilegesBuckets", [resFooTS, resSBFoo]);
+    }
+
+    // TODO SERVER-57558 - mongod bug
+    if (!isMongod) {
+        runTestOnRole(db, "roleWithSystemBucketsInAnyDB", [resFooTS, resBarTS, resSBFoo]);
+    } else {
+        runTestOnRole(db, "roleWithSystemBucketsInAnyDB", [resFooTS, resSBFoo]);
+    }
+
+    runTestOnRole(db, "roleWithAnySystemBucketsInDB", [resFooTS, resBarTS, resSBFoo, resSBBar]);
+
+    runTestOnRole(db, "roleWithAnySystemBuckets", [resFooTS, resBarTS, resSBFoo, resSBBar]);
 }
 
 function runNoAuthTestOnConnection(conn) {
@@ -147,6 +235,7 @@ function runNoAuthTestOnConnection(conn) {
     assert.commandWorked(db.dropDatabase());
     assert.commandWorked(db.createCollection("foo"));
     assert.commandWorked(db.createView("bar", "foo", []));
+    assert.commandWorked(db.createCollection("ts_foo", {timeseries: {timeField: "date"}}));
 
     var resFull = db.runCommand({listCollections: 1});
     assert.commandWorked(resFull);
@@ -171,11 +260,13 @@ function runNoAuthTestOnConnection(conn) {
 
 const mongod = MongoRunner.runMongod({auth: ''});
 runTestOnConnection(mongod);
+runSystemsBucketsTestOnConnection(mongod, true);
 MongoRunner.stopMongod(mongod);
 
 const st =
     new ShardingTest({shards: 1, mongos: 1, config: 1, other: {keyFile: 'jstests/libs/key1'}});
 runTestOnConnection(st.s0);
+runSystemsBucketsTestOnConnection(st.s0, false);
 st.stop();
 
 const mongodNoAuth = MongoRunner.runMongod();
diff --git a/src/mongo/db/auth/README.md b/src/mongo/db/auth/README.md
index f98840bdc4f..87354dded20 100644
--- a/src/mongo/db/auth/README.md
+++ b/src/mongo/db/auth/README.md
@@ -471,7 +471,7 @@ For users possessing a given set of roles, their effective privileges and
 Each role imparts privileges in the form of a set of `actions` permitted
 against a given `resource`.  The strings in the `actions` list correspond
 1:1 with `ActionType` values as specified [here](https://github.com/mongodb/mongo/blob/92cc84b0171942375ccbd2312a052bc7e9f159dd/src/mongo/db/auth/action_type.h).
-Resources may be specified in any of the following five formats:
+Resources may be specified in any of the following nine formats:
 
 | `resource` | Meaning |
 | --- | --- |
@@ -480,6 +480,10 @@ Resources may be specified in any of the following five formats:
 | { db: '', collection: 'system.views' } | The specific named collection on all DBs |
 | { db: 'test', collection: 'system.view' } | The specific namespace (db+collection) as written |
 | { cluster: true } | Used only by cluster-level actions such as `replsetConfigure`. |
+| { system_bucket: '' } | Any collection with a prefix of `system.buckets.` in any db|
+| { db: '', system_buckets: 'example' } | A collection named `system.buckets.example` in any db|
+| { db: 'test', system_buckets: '' } | Any collection with a prefix of `system.buckets.` in `test` db|
+| { db: 'test', system_buckets: 'example' } | A collected named `system.buckets.example` in `test` db|
 
 #### Normal resources
 
diff --git a/src/mongo/db/auth/action_type.idl b/src/mongo/db/auth/action_type.idl
index ab8effd2b09..68126c5c2b7 100644
--- a/src/mongo/db/auth/action_type.idl
+++ b/src/mongo/db/auth/action_type.idl
@@ -193,3 +193,11 @@ enums:
             kMatchAnyNormalResource : "any_normal"
             # Matches absolutely anything.
             kMatchAnyResource : "any"
+            # Matches a collection named "<db>.system.buckets.<collection>"
+            kMatchExactSystemBucketResource : "system_buckets"
+            # Matches a collection named "system.buckets.<collection>" in any db
+            kMatchSystemBucketInAnyDBResource : "system_buckets_in_any_db"
+            # Matches any collection with a prefix of "system.buckets." in db
+            kMatchAnySystemBucketInDBResource : "any_system_buckets_in_db"
+            # Matches any collection with a prefix of "system.buckets." in any db
+            kMatchAnySystemBucketResource : "any_system_buckets"
diff --git a/src/mongo/db/auth/authorization_session_for_test.cpp b/src/mongo/db/auth/authorization_session_for_test.cpp
index ac993b51b1e..9d2eacbd121 100644
--- a/src/mongo/db/auth/authorization_session_for_test.cpp
+++ b/src/mongo/db/auth/authorization_session_for_test.cpp
@@ -34,6 +34,7 @@
 #include <algorithm>
 #include <memory>
 
+#include "mongo/db/auth/builtin_roles.h"
 #include "mongo/db/auth/privilege.h"
 #include "mongo/db/auth/user.h"
 #include "mongo/db/auth/user_name.h"
@@ -60,6 +61,18 @@ void AuthorizationSessionForTest::assumePrivilegesForDB(PrivilegeVector privileg
     _buildAuthenticatedRolesVector();
 }
 
+
+void AuthorizationSessionForTest::assumePrivilegesForBuiltinRole(const RoleName& roleName) {
+    PrivilegeVector privileges;
+    auth::addPrivilegesForBuiltinRole(roleName, &privileges);
+    StringData db = roleName.getDB();
+    if (db.empty()) {
+        db = "admin"_sd;
+    }
+
+    assumePrivilegesForDB(privileges, db);
+}
+
 void AuthorizationSessionForTest::revokePrivilegesForDB(StringData dbName) {
     _authenticatedUsers.removeByDBName(dbName);
     _testUsers.erase(
diff --git a/src/mongo/db/auth/authorization_session_for_test.h b/src/mongo/db/auth/authorization_session_for_test.h
index b5eee9d1efd..bc6a75c798f 100644
--- a/src/mongo/db/auth/authorization_session_for_test.h
+++ b/src/mongo/db/auth/authorization_session_for_test.h
@@ -77,6 +77,11 @@ public:
      */
     void revokeAllPrivileges();
 
+    /**
+     * Grants this session all privileges for the given builtin role. Do not mix with other methods.
+     */
+    void assumePrivilegesForBuiltinRole(const RoleName& roleName);
+
 private:
     std::vector<UserHandle> _testUsers;
 };
diff --git a/src/mongo/db/auth/authorization_session_impl.cpp b/src/mongo/db/auth/authorization_session_impl.cpp
index d9e3544c109..485af290b74 100644
--- a/src/mongo/db/auth/authorization_session_impl.cpp
+++ b/src/mongo/db/auth/authorization_session_impl.cpp
@@ -71,6 +71,7 @@ auto authorizationSessionCreateRegistration =
     MONGO_WEAK_FUNCTION_REGISTRATION(AuthorizationSession::create, authorizationSessionCreateImpl);
 
 constexpr StringData ADMIN_DBNAME = "admin"_sd;
+constexpr StringData SYSTEM_BUCKETS_PREFIX = "system.buckets."_sd;
 
 bool checkContracts() {
     // Only check contracts in testing modes, invalid contracts should not break customers.
@@ -409,7 +410,7 @@ bool AuthorizationSessionImpl::isAuthorizedForActionsOnNamespace(const Namespace
     return isAuthorizedForPrivilege(Privilege(ResourcePattern::forExactNamespace(ns), actions));
 }
 
-static const int resourceSearchListCapacity = 5;
+static const int resourceSearchListCapacity = 7;
 /**
  * Builds from "target" an exhaustive list of all ResourcePatterns that match "target".
  *
@@ -440,6 +441,14 @@ static const int resourceSearchListCapacity = 5;
  *                  db,
  *                  coll,
  *                  db.coll }
+ * target is a system buckets collection, db.system.buckets.coll:
+ *   searchList = { ResourcePattern::forAnyResource(),
+ *                  ResourcePattern::forAnySystemBuckets(),
+ *                  ResourcePattern::forAnySystemBucketsInDatabase("db"),
+ *                  ResourcePattern::forAnySystemBucketsInAnyDatabase("coll"),
+ *                  ResourcePattern::forExactSystemBucketsCollection("db", "coll"),
+ *                  system.buckets.coll,
+ *                  db.system.buckets.coll }
  * target is a system collection, db.system.coll:
  *   searchList = { ResourcePattern::forAnyResource(),
  *                  system.coll,
@@ -460,6 +469,16 @@ static int buildResourceSearchList(const ResourcePattern& target,
                 resourceSearchList[size++] = ResourcePattern::forAnyNormalResource();
             }
             resourceSearchList[size++] = ResourcePattern::forDatabaseName(target.ns().db());
+        } else if (target.ns().coll().startsWith(SYSTEM_BUCKETS_PREFIX) &&
+                   target.ns().coll().size() > SYSTEM_BUCKETS_PREFIX.size()) {
+            auto bucketColl = target.ns().coll().substr(SYSTEM_BUCKETS_PREFIX.size());
+            resourceSearchList[size++] =
+                ResourcePattern::forExactSystemBucketsCollection(target.ns().db(), bucketColl);
+            resourceSearchList[size++] = ResourcePattern::forAnySystemBuckets();
+            resourceSearchList[size++] =
+                ResourcePattern::forAnySystemBucketsInDatabase(target.ns().db());
+            resourceSearchList[size++] =
+                ResourcePattern::forAnySystemBucketsInAnyDatabase(bucketColl);
         }
 
         // All collections can be matched by a collection resource for their name
@@ -659,6 +678,12 @@ bool AuthorizationSessionImpl::isAuthorizedForAnyActionOnAnyResourceInDB(StringD
             return true;
         }
 
+        // Any resource will match any system_buckets collection in the database
+        if (user->hasActionsForResource(ResourcePattern::forAnySystemBuckets()) ||
+            user->hasActionsForResource(ResourcePattern::forAnySystemBucketsInDatabase(db))) {
+            return true;
+        }
+
         // If the user is authorized for anyNormalResource, then they implicitly have access
         // to most databases.
         if (db != "local" && db != "config" &&
@@ -676,12 +701,25 @@ bool AuthorizationSessionImpl::isAuthorizedForAnyActionOnAnyResourceInDB(StringD
                 return true;
             }
 
+            // User can see system_buckets in any database so we consider them to have permission in
+            // this database
+            if (privilege.first.isAnySystemBucketsCollectionInAnyDB()) {
+                return true;
+            }
+
             // If the user has an exact namespace privilege on a collection in this database, they
             // have access to a resource in this database.
             if (privilege.first.isExactNamespacePattern() &&
                 privilege.first.databaseToMatch() == db) {
                 return true;
             }
+
+            // If the user has an exact namespace privilege on a system.buckets collection in this
+            // database, they have access to a resource in this database.
+            if (privilege.first.isExactSystemBucketsCollection() &&
+                privilege.first.databaseToMatch() == db) {
+                return true;
+            }
         }
     }
 
diff --git a/src/mongo/db/auth/authorization_session_test.cpp b/src/mongo/db/auth/authorization_session_test.cpp
index bbfb17f3f65..a6b54fc11f7 100644
--- a/src/mongo/db/auth/authorization_session_test.cpp
+++ b/src/mongo/db/auth/authorization_session_test.cpp
@@ -1379,6 +1379,257 @@ TEST_F(AuthorizationSessionTest, CanUseUUIDNamespacesWithPrivilege) {
     authzSession->verifyContract(&ac);
 }
 
+class SystemBucketsTest : public AuthorizationSessionTest {
+protected:
+    static constexpr auto sb_db_test = "sb_db_test"_sd;
+    static constexpr auto sb_db_other = "sb_db_other"_sd;
+    static constexpr auto sb_coll_test = "sb_coll_test"_sd;
+
+    static const ResourcePattern testMissingSystemBucketResource;
+    static const ResourcePattern otherMissingSystemBucketResource;
+    static const ResourcePattern otherDbMissingSystemBucketResource;
+
+    static const ResourcePattern testSystemBucketResource;
+    static const ResourcePattern otherSystemBucketResource;
+    static const ResourcePattern otherDbSystemBucketResource;
+
+    static const ResourcePattern testBucketResource;
+    static const ResourcePattern otherBucketResource;
+    static const ResourcePattern otherDbBucketResource;
+};
+
+const ResourcePattern SystemBucketsTest::testMissingSystemBucketResource(
+    ResourcePattern::forExactNamespace(NamespaceString("sb_db_test.sb_coll_test")));
+const ResourcePattern SystemBucketsTest::otherMissingSystemBucketResource(
+    ResourcePattern::forExactNamespace(NamespaceString("sb_db_test.sb_coll_other")));
+const ResourcePattern SystemBucketsTest::otherDbMissingSystemBucketResource(
+    ResourcePattern::forExactNamespace(NamespaceString("sb_db_other.sb_coll_test")));
+
+const ResourcePattern SystemBucketsTest::testSystemBucketResource(
+    ResourcePattern::forExactNamespace(NamespaceString("sb_db_test.system.buckets.sb_coll_test")));
+const ResourcePattern SystemBucketsTest::otherSystemBucketResource(
+    ResourcePattern::forExactNamespace(NamespaceString("sb_db_test.system.buckets.sb_coll_other")));
+const ResourcePattern SystemBucketsTest::otherDbSystemBucketResource(
+    ResourcePattern::forExactNamespace(NamespaceString("sb_db_other.system.buckets.sb_coll_test")));
+
+const ResourcePattern SystemBucketsTest::testBucketResource(
+    ResourcePattern::forExactSystemBucketsCollection("sb_db_test", "sb_coll_test"));
+const ResourcePattern SystemBucketsTest::otherBucketResource(
+    ResourcePattern::forExactSystemBucketsCollection("sb_db_test", "sb_coll_other"));
+const ResourcePattern SystemBucketsTest::otherDbBucketResource(
+    ResourcePattern::forExactSystemBucketsCollection("sb_db_other", "sb_coll_test"));
+
+TEST_F(SystemBucketsTest, CheckExactSystemBucketsCollection) {
+    // If we have a system_buckets exact priv
+    authzSession->assumePrivilegesForDB(Privilege(testBucketResource, ActionType::find));
+
+    ASSERT_FALSE(authzSession->isAuthorizedForActionsOnResource(testSystemBucketResource,
+                                                                ActionType::insert));
+
+    ASSERT_TRUE(
+        authzSession->isAuthorizedForActionsOnResource(testSystemBucketResource, ActionType::find));
+    ASSERT_FALSE(authzSession->isAuthorizedForActionsOnResource(otherSystemBucketResource,
+                                                                ActionType::find));
+    ASSERT_FALSE(authzSession->isAuthorizedForActionsOnResource(otherDbSystemBucketResource,
+                                                                ActionType::find));
+
+
+    ASSERT_FALSE(authzSession->isAuthorizedForActionsOnResource(testMissingSystemBucketResource,
+                                                                ActionType::find));
+    ASSERT_FALSE(authzSession->isAuthorizedForActionsOnResource(otherMissingSystemBucketResource,
+                                                                ActionType::find));
+
+    ASSERT_FALSE(authzSession->isAuthorizedForActionsOnResource(otherDbMissingSystemBucketResource,
+                                                                ActionType::find));
+}
+
+TEST_F(SystemBucketsTest, CheckAnySystemBuckets) {
+    // If we have an any system_buckets priv
+    authzSession->assumePrivilegesForDB(
+        Privilege(ResourcePattern::forAnySystemBuckets(), ActionType::find));
+
+    ASSERT_FALSE(authzSession->isAuthorizedForActionsOnResource(testSystemBucketResource,
+                                                                ActionType::insert));
+
+    ASSERT_TRUE(
+        authzSession->isAuthorizedForActionsOnResource(testSystemBucketResource, ActionType::find));
+    ASSERT_TRUE(authzSession->isAuthorizedForActionsOnResource(otherSystemBucketResource,
+                                                               ActionType::find));
+    ASSERT_TRUE(authzSession->isAuthorizedForActionsOnResource(otherDbSystemBucketResource,
+                                                               ActionType::find));
+
+
+    ASSERT_FALSE(authzSession->isAuthorizedForActionsOnResource(testMissingSystemBucketResource,
+                                                                ActionType::find));
+    ASSERT_FALSE(authzSession->isAuthorizedForActionsOnResource(otherMissingSystemBucketResource,
+                                                                ActionType::find));
+
+    ASSERT_FALSE(authzSession->isAuthorizedForActionsOnResource(otherDbMissingSystemBucketResource,
+                                                                ActionType::find));
+}
+
+TEST_F(SystemBucketsTest, CheckAnySystemBucketsInDatabase) {
+    // If we have a system_buckets in a db priv
+    authzSession->assumePrivilegesForDB(
+        Privilege(ResourcePattern::forAnySystemBucketsInDatabase("sb_db_test"), ActionType::find));
+
+    ASSERT_FALSE(authzSession->isAuthorizedForActionsOnResource(testSystemBucketResource,
+                                                                ActionType::insert));
+
+    ASSERT_TRUE(
+        authzSession->isAuthorizedForActionsOnResource(testSystemBucketResource, ActionType::find));
+    ASSERT_TRUE(authzSession->isAuthorizedForActionsOnResource(otherSystemBucketResource,
+                                                               ActionType::find));
+    ASSERT_FALSE(authzSession->isAuthorizedForActionsOnResource(otherDbSystemBucketResource,
+                                                                ActionType::find));
+
+
+    ASSERT_FALSE(authzSession->isAuthorizedForActionsOnResource(testMissingSystemBucketResource,
+                                                                ActionType::find));
+    ASSERT_FALSE(authzSession->isAuthorizedForActionsOnResource(otherMissingSystemBucketResource,
+                                                                ActionType::find));
+
+    ASSERT_FALSE(authzSession->isAuthorizedForActionsOnResource(otherDbMissingSystemBucketResource,
+                                                                ActionType::find));
+}
+
+TEST_F(SystemBucketsTest, CheckforAnySystemBucketsInAnyDatabase) {
+    // If we have a system_buckets for a coll in any db priv
+    authzSession->assumePrivilegesForDB(Privilege(
+        ResourcePattern::forAnySystemBucketsInAnyDatabase("sb_coll_test"), ActionType::find));
+
+
+    ASSERT_FALSE(authzSession->isAuthorizedForActionsOnResource(testSystemBucketResource,
+                                                                ActionType::insert));
+
+    ASSERT_TRUE(
+        authzSession->isAuthorizedForActionsOnResource(testSystemBucketResource, ActionType::find));
+    ASSERT_FALSE(authzSession->isAuthorizedForActionsOnResource(otherSystemBucketResource,
+                                                                ActionType::find));
+    ASSERT_TRUE(authzSession->isAuthorizedForActionsOnResource(otherDbSystemBucketResource,
+                                                               ActionType::find));
+
+
+    ASSERT_FALSE(authzSession->isAuthorizedForActionsOnResource(testMissingSystemBucketResource,
+                                                                ActionType::find));
+    ASSERT_FALSE(authzSession->isAuthorizedForActionsOnResource(otherMissingSystemBucketResource,
+                                                                ActionType::find));
+
+    ASSERT_FALSE(authzSession->isAuthorizedForActionsOnResource(otherDbMissingSystemBucketResource,
+                                                                ActionType::find));
+}
+
+TEST_F(SystemBucketsTest, CanCheckIfHasAnyPrivilegeOnResourceForSystemBuckets) {
+    // If we have a system.buckets collection privilege, we have actions on that collection
+    authzSession->assumePrivilegesForDB(Privilege(testBucketResource, ActionType::find));
+    ASSERT_TRUE(authzSession->isAuthorizedForAnyActionOnResource(testSystemBucketResource));
+    ASSERT_FALSE(authzSession->isAuthorizedForAnyActionOnResource(
+        ResourcePattern::forDatabaseName(sb_db_test)));
+    ASSERT_FALSE(
+        authzSession->isAuthorizedForAnyActionOnResource(ResourcePattern::forAnyNormalResource()));
+    ASSERT_FALSE(
+        authzSession->isAuthorizedForAnyActionOnResource(ResourcePattern::forAnyResource()));
+
+    // If we have any buckets in a database privilege, we have actions on that database and all
+    // system.buckets collections it contains
+    authzSession->assumePrivilegesForDB(
+        Privilege(ResourcePattern::forAnySystemBucketsInDatabase(sb_db_test), ActionType::find));
+    ASSERT_TRUE(authzSession->isAuthorizedForAnyActionOnResource(
+        ResourcePattern::forAnySystemBucketsInDatabase(sb_db_test)));
+    ASSERT_TRUE(authzSession->isAuthorizedForAnyActionOnResource(testSystemBucketResource));
+    ASSERT_FALSE(authzSession->isAuthorizedForAnyActionOnResource(
+        ResourcePattern::forDatabaseName(sb_db_test)));
+    ASSERT_FALSE(
+        authzSession->isAuthorizedForAnyActionOnResource(ResourcePattern::forAnyNormalResource()));
+    ASSERT_FALSE(
+        authzSession->isAuthorizedForAnyActionOnResource(ResourcePattern::forAnyResource()));
+
+    // If we have a privilege on any systems buckets in any db, we have actions on all databases and
+    // system.buckets.<coll> they contain
+    authzSession->assumePrivilegesForDB(Privilege(
+        ResourcePattern::forAnySystemBucketsInAnyDatabase(sb_coll_test), ActionType::find));
+    ASSERT_TRUE(authzSession->isAuthorizedForAnyActionOnResource(testSystemBucketResource));
+    ASSERT_FALSE(authzSession->isAuthorizedForAnyActionOnResource(
+        ResourcePattern::forDatabaseName(sb_db_test)));
+    ASSERT_FALSE(
+        authzSession->isAuthorizedForAnyActionOnResource(ResourcePattern::forAnyNormalResource()));
+    ASSERT_FALSE(
+        authzSession->isAuthorizedForAnyActionOnResource(ResourcePattern::forAnyResource()));
+}
+
+TEST_F(SystemBucketsTest, CheckBuiltinRolesForSystemBuckets) {
+    // If we have readAnyDatabase, make sure we can read system.buckets anywhere
+    authzSession->assumePrivilegesForBuiltinRole(RoleName("readAnyDatabase", "admin"));
+
+    ASSERT_TRUE(
+        authzSession->isAuthorizedForActionsOnResource(testSystemBucketResource, ActionType::find));
+    ASSERT_TRUE(authzSession->isAuthorizedForActionsOnResource(otherSystemBucketResource,
+                                                               ActionType::find));
+    ASSERT_TRUE(authzSession->isAuthorizedForActionsOnResource(otherDbSystemBucketResource,
+                                                               ActionType::find));
+
+    // If we have readAnyDatabase, make sure we can read and write system.buckets anywhere
+    authzSession->assumePrivilegesForBuiltinRole(RoleName("readWriteAnyDatabase", "admin"));
+
+    ASSERT_TRUE(authzSession->isAuthorizedForActionsOnResource(
+        testSystemBucketResource, {ActionType::find, ActionType::insert}));
+    ASSERT_TRUE(authzSession->isAuthorizedForActionsOnResource(
+        otherSystemBucketResource, {ActionType::find, ActionType::insert}));
+    ASSERT_TRUE(authzSession->isAuthorizedForActionsOnResource(
+        otherDbSystemBucketResource, {ActionType::find, ActionType::insert}));
+
+    // If we have readAnyDatabase, make sure we can do admin stuff on system.buckets anywhere
+    authzSession->assumePrivilegesForBuiltinRole(RoleName("dbAdminAnyDatabase", "admin"));
+
+    ASSERT_TRUE(authzSession->isAuthorizedForActionsOnResource(
+        testSystemBucketResource, ActionType::bypassDocumentValidation));
+    ASSERT_TRUE(authzSession->isAuthorizedForActionsOnResource(
+        otherSystemBucketResource, ActionType::bypassDocumentValidation));
+    ASSERT_TRUE(authzSession->isAuthorizedForActionsOnResource(
+        otherDbSystemBucketResource, ActionType::bypassDocumentValidation));
+
+
+    // If we have readAnyDatabase, make sure we can do restore stuff on system.buckets anywhere
+    authzSession->assumePrivilegesForBuiltinRole(RoleName("restore", "admin"));
+
+    ASSERT_TRUE(authzSession->isAuthorizedForActionsOnResource(
+        testSystemBucketResource, ActionType::bypassDocumentValidation));
+    ASSERT_TRUE(authzSession->isAuthorizedForActionsOnResource(
+        otherSystemBucketResource, ActionType::bypassDocumentValidation));
+    ASSERT_TRUE(authzSession->isAuthorizedForActionsOnResource(
+        otherDbSystemBucketResource, ActionType::bypassDocumentValidation));
+
+    // If we have readAnyDatabase, make sure we can do restore stuff on system.buckets anywhere
+    authzSession->assumePrivilegesForBuiltinRole(RoleName("backup", "admin"));
+
+    ASSERT_TRUE(
+        authzSession->isAuthorizedForActionsOnResource(testSystemBucketResource, ActionType::find));
+    ASSERT_TRUE(authzSession->isAuthorizedForActionsOnResource(otherSystemBucketResource,
+                                                               ActionType::find));
+    ASSERT_TRUE(authzSession->isAuthorizedForActionsOnResource(otherDbSystemBucketResource,
+                                                               ActionType::find));
+}
+
+TEST_F(SystemBucketsTest, CanCheckIfHasAnyPrivilegeInResourceDBForSystemBuckets) {
+    authzSession->assumePrivilegesForDB(Privilege(testBucketResource, ActionType::find));
+    ASSERT_TRUE(authzSession->isAuthorizedForAnyActionOnAnyResourceInDB(sb_db_test));
+    ASSERT_FALSE(authzSession->isAuthorizedForAnyActionOnAnyResourceInDB(sb_db_other));
+
+    authzSession->assumePrivilegesForDB(
+        Privilege(ResourcePattern::forAnySystemBucketsInDatabase(sb_db_test), ActionType::find));
+    ASSERT_TRUE(authzSession->isAuthorizedForAnyActionOnAnyResourceInDB(sb_db_test));
+    ASSERT_FALSE(authzSession->isAuthorizedForAnyActionOnAnyResourceInDB(sb_db_other));
+
+    authzSession->assumePrivilegesForDB(Privilege(
+        ResourcePattern::forAnySystemBucketsInAnyDatabase(sb_coll_test), ActionType::find));
+    ASSERT_TRUE(authzSession->isAuthorizedForAnyActionOnAnyResourceInDB(sb_db_test));
+    ASSERT_TRUE(authzSession->isAuthorizedForAnyActionOnAnyResourceInDB(sb_db_other));
+
+    authzSession->assumePrivilegesForDB(
+        Privilege(ResourcePattern::forAnySystemBuckets(), ActionType::find));
+    ASSERT_TRUE(authzSession->isAuthorizedForAnyActionOnAnyResourceInDB(sb_db_test));
+    ASSERT_TRUE(authzSession->isAuthorizedForAnyActionOnAnyResourceInDB(sb_db_other));
+}
 
 }  // namespace
 }  // namespace mongo
diff --git a/src/mongo/db/auth/builtin_roles.cpp b/src/mongo/db/auth/builtin_roles.cpp
index 3f53b990080..7b505e5f603 100644
--- a/src/mongo/db/auth/builtin_roles.cpp
+++ b/src/mongo/db/auth/builtin_roles.cpp
@@ -324,6 +324,8 @@ void addReadOnlyAnyDbPrivileges(PrivilegeVector* privileges) {
         privileges, Privilege(ResourcePattern::forClusterResource(), ActionType::listDatabases));
     Privilege::addPrivilegeToPrivilegeVector(
         privileges, Privilege(ResourcePattern::forCollectionName("system.js"), readRoleActions));
+    Privilege::addPrivilegeToPrivilegeVector(
+        privileges, Privilege(ResourcePattern::forAnySystemBuckets(), readRoleActions));
 }
 
 void addReadWriteAnyDbPrivileges(PrivilegeVector* privileges) {
@@ -333,6 +335,8 @@ void addReadWriteAnyDbPrivileges(PrivilegeVector* privileges) {
     Privilege::addPrivilegeToPrivilegeVector(
         privileges,
         Privilege(ResourcePattern::forCollectionName("system.js"), readWriteRoleActions));
+    Privilege::addPrivilegeToPrivilegeVector(
+        privileges, Privilege(ResourcePattern::forAnySystemBuckets(), readWriteRoleActions));
 }
 
 void addUserAdminAnyDbPrivileges(PrivilegeVector* privileges) {
@@ -398,6 +402,8 @@ void addDbAdminAnyDbPrivileges(PrivilegeVector* privileges) {
         Privilege(ResourcePattern::forCollectionName("system.profile"), profileActions));
     Privilege::addPrivilegeToPrivilegeVector(
         privileges, Privilege(ResourcePattern::forClusterResource(), ActionType::applyOps));
+    Privilege::addPrivilegeToPrivilegeVector(
+        privileges, Privilege(ResourcePattern::forAnySystemBuckets(), dbAdminRoleActions));
 }
 
 void addClusterMonitorPrivileges(PrivilegeVector* privileges) {
@@ -498,6 +504,8 @@ void addQueryableBackupPrivileges(PrivilegeVector* privileges) {
         privileges, Privilege(ResourcePattern::forAnyResource(), ActionType::listCollections));
     Privilege::addPrivilegeToPrivilegeVector(
         privileges, Privilege(ResourcePattern::forAnyResource(), ActionType::listIndexes));
+    Privilege::addPrivilegeToPrivilegeVector(
+        privileges, Privilege(ResourcePattern::forAnySystemBuckets(), ActionType::find));
 
     ActionSet clusterActions;
     clusterActions << ActionType::getParameter  // To check authSchemaVersion
@@ -591,6 +599,9 @@ void addRestorePrivileges(PrivilegeVector* privileges) {
         privileges, Privilege(ResourcePattern::forDatabaseName("config"), actions));
 
     Privilege::addPrivilegeToPrivilegeVector(
+        privileges, Privilege(ResourcePattern::forAnySystemBuckets(), actions));
+
+    Privilege::addPrivilegeToPrivilegeVector(
         privileges,
         Privilege(ResourcePattern::forExactNamespace(NamespaceString("local", "system.replset")),
                   actions));
diff --git a/src/mongo/db/auth/privilege_parser.cpp b/src/mongo/db/auth/privilege_parser.cpp
index 7fb7ea38008..4d7bb9072c2 100644
--- a/src/mongo/db/auth/privilege_parser.cpp
+++ b/src/mongo/db/auth/privilege_parser.cpp
@@ -45,6 +45,7 @@ using str::stream;
 
 const BSONField<bool> ParsedResource::anyResource("anyResource");
 const BSONField<bool> ParsedResource::cluster("cluster");
+const BSONField<string> ParsedResource::systemBuckets("system_buckets");
 const BSONField<string> ParsedResource::db("db");
 const BSONField<string> ParsedResource::collection("collection");
 
@@ -65,14 +66,20 @@ bool ParsedResource::isValid(std::string* errMsg) const {
         ++numCandidateTypes;
     if (isClusterSet())
         ++numCandidateTypes;
-    if (isDbSet() || isCollectionSet())
+    if (isDbSet() || isCollectionSet() || isSystemBucketsSet())
         ++numCandidateTypes;
 
-    if (isDbSet() != isCollectionSet()) {
+    if (!isSystemBucketsSet() && isDbSet() != isCollectionSet()) {
         *errMsg = stream() << "resource must set both " << db.name() << " and " << collection.name()
                            << " or neither, but not exactly one.";
         return false;
+    } else if (isSystemBucketsSet()) {
+        if (isCollectionSet()) {
+            *errMsg = stream() << "system_buckets and collection cannot both be set";
+            return false;
+        }
     }
+
     if (numCandidateTypes != 1) {
         *errMsg = stream() << "resource must have exactly " << db.name() << " and "
                            << collection.name() << " set, or have only " << cluster.name()
@@ -80,20 +87,24 @@ bool ParsedResource::isValid(std::string* errMsg) const {
                            << " or have only " << anyResource.name() << " set";
         return false;
     }
+
     if (isAnyResourceSet() && !getAnyResource()) {
         *errMsg = stream() << anyResource.name() << " must be true when specified";
         return false;
     }
+
     if (isClusterSet() && !getCluster()) {
         *errMsg = stream() << cluster.name() << " must be true when specified";
         return false;
     }
+
     if (isDbSet() &&
         (!NamespaceString::validDBName(getDb(), NamespaceString::DollarInDbNameBehavior::Allow) &&
          !getDb().empty())) {
         *errMsg = stream() << getDb() << " is not a valid database name";
         return false;
     }
+
     if (isCollectionSet() &&
         (!NamespaceString::validCollectionName(getCollection()) && !getCollection().empty())) {
         // local.oplog.$main is a real collection that the server will create. But, collection
@@ -104,6 +115,7 @@ bool ParsedResource::isValid(std::string* errMsg) const {
             return false;
         }
     }
+
     return true;
 }
 
@@ -116,6 +128,9 @@ BSONObj ParsedResource::toBSON() const {
     if (_isClusterSet)
         builder.append(cluster(), _cluster);
 
+    if (_isSystemBucketsSet)
+        builder.append(systemBuckets(), _systemBuckets);
+
     if (_isDbSet)
         builder.append(db(), _db);
 
@@ -143,6 +158,11 @@ bool ParsedResource::parseBSON(const BSONObj& source, string* errMsg) {
         return false;
     _isClusterSet = fieldState == FieldParser::FIELD_SET;
 
+    fieldState = FieldParser::extract(source, systemBuckets, &_systemBuckets, errMsg);
+    if (fieldState == FieldParser::FIELD_INVALID)
+        return false;
+    _isSystemBucketsSet = fieldState == FieldParser::FIELD_SET;
+
     fieldState = FieldParser::extract(source, db, &_db, errMsg);
     if (fieldState == FieldParser::FIELD_INVALID)
         return false;
@@ -163,6 +183,9 @@ void ParsedResource::clear() {
     _cluster = false;
     _isClusterSet = false;
 
+    _systemBuckets.clear();
+    _isSystemBucketsSet = false;
+
     _db.clear();
     _isDbSet = false;
 
@@ -179,6 +202,9 @@ void ParsedResource::cloneTo(ParsedResource* other) const {
     other->_cluster = _cluster;
     other->_isClusterSet = _isClusterSet;
 
+    other->_systemBuckets = _systemBuckets;
+    other->_isSystemBucketsSet = _isSystemBucketsSet;
+
     other->_db = _db;
     other->_isDbSet = _isDbSet;
 
@@ -226,6 +252,24 @@ bool ParsedResource::getCluster() const {
     return _cluster;
 }
 
+void ParsedResource::setSystemBuckets(StringData collection) {
+    _systemBuckets = collection.toString();
+    _isSystemBucketsSet = true;
+}
+
+void ParsedResource::unsetSystemBuckets() {
+    _isSystemBucketsSet = false;
+}
+
+bool ParsedResource::isSystemBucketsSet() const {
+    return _isSystemBucketsSet;
+}
+
+const std::string& ParsedResource::getSystemBuckets() const {
+    dassert(_isSystemBucketsSet);
+    return _systemBuckets;
+}
+
 void ParsedResource::setDb(StringData db) {
     _db = db.toString();
     _isDbSet = true;
@@ -419,6 +463,40 @@ Status ParsedPrivilege::parsedPrivilegeToPrivilege(const ParsedPrivilege& parsed
         resource = ResourcePattern::forAnyResource();
     } else if (parsedResource.isClusterSet() && parsedResource.getCluster()) {
         resource = ResourcePattern::forClusterResource();
+    } else if (parsedResource.isSystemBucketsSet()) {
+        if (parsedResource.isDbSet()) {
+            if (parsedResource.getDb().empty()) {
+                if (parsedResource.getSystemBuckets().empty()) {
+                    // {db: "", system_buckets: ""} - match any system buckets in any db
+                    resource = ResourcePattern::forAnySystemBuckets();
+                } else {
+                    // {db: "", system_buckets: "<coll>"} - match any system.buckets.<coll> in
+                    // any db
+                    resource = ResourcePattern::forAnySystemBucketsInAnyDatabase(
+                        parsedResource.getSystemBuckets());
+                }
+            } else {
+                if (parsedResource.getSystemBuckets().empty()) {
+                    // {db: "<db>", system_buckets: ""} - match any system buckets in db <db>
+                    resource =
+                        ResourcePattern::forAnySystemBucketsInDatabase(parsedResource.getDb());
+                } else {
+                    // {db: "<db>", system_buckets: "<coll>"} - match <db>.system.buckets.<coll>
+                    resource = ResourcePattern::forExactSystemBucketsCollection(
+                        parsedResource.getDb(), parsedResource.getSystemBuckets());
+                }
+            }
+        } else {
+            if (parsedResource.getSystemBuckets().empty()) {
+                // {system_buckets: ""} - match any system buckets in any db
+                resource = ResourcePattern::forAnySystemBuckets();
+            } else {
+                // {system_buckets: "<coll>"} - match any system.buckets.<coll> in any db
+                resource = ResourcePattern::forAnySystemBucketsInAnyDatabase(
+                    parsedResource.getSystemBuckets());
+            }
+        }
+
     } else {
         if (parsedResource.isDbSet() && !parsedResource.getDb().empty()) {
             if (parsedResource.isCollectionSet() && !parsedResource.getCollection().empty()) {
@@ -458,6 +536,16 @@ bool ParsedPrivilege::privilegeToParsedPrivilege(const Privilege& privilege,
         parsedResource.setCollection("");
     } else if (privilege.getResourcePattern().isClusterResourcePattern()) {
         parsedResource.setCluster(true);
+    } else if (privilege.getResourcePattern().isAnySystemBucketsCollection()) {
+        parsedResource.setSystemBuckets("");
+    } else if (privilege.getResourcePattern().isAnySystemBucketsCollectionInDB()) {
+        parsedResource.setSystemBuckets("");
+        parsedResource.setDb(privilege.getResourcePattern().databaseToMatch());
+    } else if (privilege.getResourcePattern().isAnySystemBucketsCollectionInAnyDB()) {
+        parsedResource.setSystemBuckets(privilege.getResourcePattern().collectionToMatch());
+    } else if (privilege.getResourcePattern().isExactSystemBucketsCollection()) {
+        parsedResource.setDb(privilege.getResourcePattern().databaseToMatch());
+        parsedResource.setSystemBuckets(privilege.getResourcePattern().collectionToMatch());
     } else if (privilege.getResourcePattern().isAnyResourcePattern()) {
         parsedResource.setAnyResource(true);
     } else {
diff --git a/src/mongo/db/auth/privilege_parser.h b/src/mongo/db/auth/privilege_parser.h
index 853a371193e..de315a65a74 100644
--- a/src/mongo/db/auth/privilege_parser.h
+++ b/src/mongo/db/auth/privilege_parser.h
@@ -54,6 +54,7 @@ public:
 
     static const BSONField<bool> anyResource;
     static const BSONField<bool> cluster;
+    static const BSONField<std::string> systemBuckets;
     static const BSONField<std::string> db;
     static const BSONField<std::string> collection;
 
@@ -101,6 +102,11 @@ public:
     bool isCollectionSet() const;
     const std::string& getCollection() const;
 
+    void setSystemBuckets(StringData collection);
+    void unsetSystemBuckets();
+    bool isSystemBucketsSet() const;
+    const std::string& getSystemBuckets() const;
+
 private:
     // Convention: (M)andatory, (O)ptional
 
@@ -112,6 +118,11 @@ private:
     bool _cluster;
     bool _isClusterSet;
 
+    // (O) Only present if the resource is the system.buckets.<collection> or system.buckets.*
+    // resource
+    std::string _systemBuckets;
+    bool _isSystemBucketsSet;
+
     // (O) database portion of the resource
     std::string _db;
     bool _isDbSet;
diff --git a/src/mongo/db/auth/privilege_parser_test.cpp b/src/mongo/db/auth/privilege_parser_test.cpp
index 969360a6f51..fa3f1c5e6bd 100644
--- a/src/mongo/db/auth/privilege_parser_test.cpp
+++ b/src/mongo/db/auth/privilege_parser_test.cpp
@@ -96,6 +96,78 @@ TEST(PrivilegeParserTest, IsValidTest) {
     parsedPrivilege.parseBSON(
         BSON("resource" << BSON("cluster" << true) << "actions" << BSON_ARRAY("find")), &errmsg);
     ASSERT(parsedPrivilege.isValid(&errmsg));
+
+
+    // Works with no db and system_buckets any
+    parsedPrivilege.parseBSON(BSON("resource" << BSON("system_buckets"
+                                                      << "")
+                                              << "actions" << BSON_ARRAY("find")),
+                              &errmsg);
+    ASSERT(parsedPrivilege.isValid(&errmsg));
+
+    // Works with empty db and system_buckets any
+    parsedPrivilege.parseBSON(BSON("resource" << BSON("db"
+                                                      << ""
+                                                      << "system_buckets"
+                                                      << "")
+                                              << "actions" << BSON_ARRAY("find")),
+                              &errmsg);
+    ASSERT(parsedPrivilege.isValid(&errmsg));
+
+    // Works with real db and system_buckets foo
+    parsedPrivilege.parseBSON(BSON("resource" << BSON("db"
+                                                      << "test"
+                                                      << "system_buckets"
+                                                      << "foo")
+                                              << "actions" << BSON_ARRAY("find")),
+                              &errmsg);
+    ASSERT(parsedPrivilege.isValid(&errmsg));
+
+    // Works with real db and system_buckets any
+    parsedPrivilege.parseBSON(BSON("resource" << BSON("db"
+                                                      << "test"
+                                                      << "system_buckets"
+                                                      << "")
+                                              << "actions" << BSON_ARRAY("find")),
+                              &errmsg);
+    ASSERT(parsedPrivilege.isValid(&errmsg));
+
+    // Works with only system_buckets and no db
+    parsedPrivilege.parseBSON(BSON("resource" << BSON("system_buckets"
+                                                      << "foo")
+                                              << "actions" << BSON_ARRAY("find")),
+                              &errmsg);
+    ASSERT(parsedPrivilege.isValid(&errmsg));
+
+    // Fails with real db and system_buckets foo and any
+    parsedPrivilege.parseBSON(BSON("resource" << BSON("db"
+                                                      << "test"
+                                                      << "system_buckets"
+                                                      << "foo"
+                                                      << "anyResource" << true)
+                                              << "actions" << BSON_ARRAY("find")),
+                              &errmsg);
+    ASSERT_FALSE(parsedPrivilege.isValid(&errmsg));
+
+    // Fails with real db and system_buckets foo and any
+    parsedPrivilege.parseBSON(BSON("resource" << BSON("db"
+                                                      << "test"
+                                                      << "system_buckets"
+                                                      << "foo"
+                                                      << "cluster" << true)
+                                              << "actions" << BSON_ARRAY("find")),
+                              &errmsg);
+    ASSERT_FALSE(parsedPrivilege.isValid(&errmsg));
+
+
+    // Fails with real collection and system_buckets foo
+    parsedPrivilege.parseBSON(BSON("resource" << BSON("collection"
+                                                      << "test"
+                                                      << "system_buckets"
+                                                      << "foo")
+                                              << "actions" << BSON_ARRAY("find")),
+                              &errmsg);
+    ASSERT_FALSE(parsedPrivilege.isValid(&errmsg));
 }
 
 TEST(PrivilegeParserTest, ConvertBetweenPrivilegeTest) {
@@ -129,6 +201,7 @@ TEST(PrivilegeParserTest, ConvertBetweenPrivilegeTest) {
     ASSERT(parsedPrivilege.getResource().isCollectionSet());
     ASSERT_EQUALS("", parsedPrivilege.getResource().getDb());
     ASSERT_EQUALS("", parsedPrivilege.getResource().getCollection());
+    ASSERT_FALSE(parsedPrivilege.getResource().isSystemBucketsSet());
     ASSERT(parsedPrivilege.isActionsSet());
     ASSERT(actionsVector == parsedPrivilege.getActions());
 
@@ -156,6 +229,7 @@ TEST(PrivilegeParserTest, ConvertBetweenPrivilegeTest) {
     ASSERT(parsedPrivilege.getResource().isCollectionSet());
     ASSERT_EQUALS("test", parsedPrivilege.getResource().getDb());
     ASSERT_EQUALS("foo", parsedPrivilege.getResource().getCollection());
+    ASSERT_FALSE(parsedPrivilege.getResource().isSystemBucketsSet());
     ASSERT(parsedPrivilege.isActionsSet());
     ASSERT(actionsVector == parsedPrivilege.getActions());
 
@@ -182,6 +256,7 @@ TEST(PrivilegeParserTest, ConvertBetweenPrivilegeTest) {
     ASSERT(parsedPrivilege.getResource().isCollectionSet());
     ASSERT_EQUALS("test", parsedPrivilege.getResource().getDb());
     ASSERT_EQUALS("", parsedPrivilege.getResource().getCollection());
+    ASSERT_FALSE(parsedPrivilege.getResource().isSystemBucketsSet());
     ASSERT(parsedPrivilege.isActionsSet());
     ASSERT(actionsVector == parsedPrivilege.getActions());
 
@@ -208,6 +283,7 @@ TEST(PrivilegeParserTest, ConvertBetweenPrivilegeTest) {
     ASSERT(parsedPrivilege.getResource().isCollectionSet());
     ASSERT_EQUALS("", parsedPrivilege.getResource().getDb());
     ASSERT_EQUALS("foo", parsedPrivilege.getResource().getCollection());
+    ASSERT_FALSE(parsedPrivilege.getResource().isSystemBucketsSet());
     ASSERT(parsedPrivilege.isActionsSet());
     ASSERT(actionsVector == parsedPrivilege.getActions());
 
@@ -229,6 +305,139 @@ TEST(PrivilegeParserTest, ConvertBetweenPrivilegeTest) {
     ASSERT(parsedPrivilege.getResource().getCluster());
     ASSERT_FALSE(parsedPrivilege.getResource().isDbSet());
     ASSERT_FALSE(parsedPrivilege.getResource().isCollectionSet());
+    ASSERT_FALSE(parsedPrivilege.getResource().isSystemBucketsSet());
+    ASSERT(parsedPrivilege.isActionsSet());
+    ASSERT(actionsVector == parsedPrivilege.getActions());
+
+    // Works with any system.buckets resource
+    parsedPrivilege.parseBSON(BSON("resource" << BSON("system_buckets"
+                                                      << "")
+                                              << "actions" << BSON_ARRAY("find")),
+                              &errmsg);
+    ASSERT(parsedPrivilege.isValid(&errmsg));
+    ASSERT_OK(ParsedPrivilege::parsedPrivilegeToPrivilege(
+        parsedPrivilege, &privilege, &unrecognizedActions));
+    ASSERT(unrecognizedActions.empty());
+    ASSERT(privilege.getActions().contains(ActionType::find));
+    ASSERT(!privilege.getActions().contains(ActionType::insert));
+    ASSERT_EQUALS(privilege.getResourcePattern(), ResourcePattern::forAnySystemBuckets());
+
+    ASSERT(ParsedPrivilege::privilegeToParsedPrivilege(privilege, &parsedPrivilege, &errmsg));
+    ASSERT(parsedPrivilege.isValid(&errmsg));
+    ASSERT(parsedPrivilege.isResourceSet());
+    ASSERT_TRUE(parsedPrivilege.getResource().isSystemBucketsSet());
+    ASSERT_EQUALS(parsedPrivilege.getResource().getSystemBuckets(), "");
+    ASSERT_FALSE(parsedPrivilege.getResource().isClusterSet());
+    ASSERT_FALSE(parsedPrivilege.getResource().isDbSet());
+    ASSERT_FALSE(parsedPrivilege.getResource().isCollectionSet());
+    ASSERT(parsedPrivilege.isActionsSet());
+    ASSERT(actionsVector == parsedPrivilege.getActions());
+
+    // Works with any system.buckets resource with empty db
+    parsedPrivilege.parseBSON(BSON("resource" << BSON("db"
+                                                      << ""
+                                                      << "system_buckets"
+                                                      << "")
+                                              << "actions" << BSON_ARRAY("find")),
+                              &errmsg);
+    ASSERT(parsedPrivilege.isValid(&errmsg));
+    ASSERT_OK(ParsedPrivilege::parsedPrivilegeToPrivilege(
+        parsedPrivilege, &privilege, &unrecognizedActions));
+    ASSERT(unrecognizedActions.empty());
+    ASSERT(privilege.getActions().contains(ActionType::find));
+    ASSERT(!privilege.getActions().contains(ActionType::insert));
+    ASSERT_EQUALS(privilege.getResourcePattern(), ResourcePattern::forAnySystemBuckets());
+
+    ASSERT(ParsedPrivilege::privilegeToParsedPrivilege(privilege, &parsedPrivilege, &errmsg));
+    ASSERT(parsedPrivilege.isValid(&errmsg));
+    ASSERT(parsedPrivilege.isResourceSet());
+    ASSERT_TRUE(parsedPrivilege.getResource().isSystemBucketsSet());
+    ASSERT_EQUALS(parsedPrivilege.getResource().getSystemBuckets(), "");
+    ASSERT_FALSE(parsedPrivilege.getResource().isClusterSet());
+    ASSERT_FALSE(parsedPrivilege.getResource().isDbSet());
+    ASSERT_FALSE(parsedPrivilege.getResource().isCollectionSet());
+    ASSERT(parsedPrivilege.isActionsSet());
+    ASSERT(actionsVector == parsedPrivilege.getActions());
+
+    // Works with system.buckets.foo resource in test db
+    parsedPrivilege.parseBSON(BSON("resource" << BSON("db"
+                                                      << "test"
+                                                      << "system_buckets"
+                                                      << "foo")
+                                              << "actions" << BSON_ARRAY("find")),
+                              &errmsg);
+    ASSERT(parsedPrivilege.isValid(&errmsg));
+    ASSERT_OK(ParsedPrivilege::parsedPrivilegeToPrivilege(
+        parsedPrivilege, &privilege, &unrecognizedActions));
+    ASSERT(unrecognizedActions.empty());
+    ASSERT(privilege.getActions().contains(ActionType::find));
+    ASSERT(!privilege.getActions().contains(ActionType::insert));
+    ASSERT_EQUALS(privilege.getResourcePattern(),
+                  ResourcePattern::forExactSystemBucketsCollection("test", "foo"));
+
+    ASSERT(ParsedPrivilege::privilegeToParsedPrivilege(privilege, &parsedPrivilege, &errmsg));
+    ASSERT(parsedPrivilege.isValid(&errmsg));
+    ASSERT(parsedPrivilege.isResourceSet());
+    ASSERT_TRUE(parsedPrivilege.getResource().isSystemBucketsSet());
+    ASSERT_EQUALS(parsedPrivilege.getResource().getSystemBuckets(), "foo");
+    ASSERT_FALSE(parsedPrivilege.getResource().isClusterSet());
+    ASSERT_TRUE(parsedPrivilege.getResource().isDbSet());
+    ASSERT_FALSE(parsedPrivilege.getResource().isCollectionSet());
+    ASSERT_EQUALS("test", parsedPrivilege.getResource().getDb());
+    ASSERT(parsedPrivilege.isActionsSet());
+    ASSERT(actionsVector == parsedPrivilege.getActions());
+
+
+    // Works with any system.buckets resource named foo
+    parsedPrivilege.parseBSON(BSON("resource" << BSON("system_buckets"
+                                                      << "foo")
+                                              << "actions" << BSON_ARRAY("find")),
+                              &errmsg);
+    ASSERT(parsedPrivilege.isValid(&errmsg));
+    ASSERT_OK(ParsedPrivilege::parsedPrivilegeToPrivilege(
+        parsedPrivilege, &privilege, &unrecognizedActions));
+    ASSERT(unrecognizedActions.empty());
+    ASSERT(privilege.getActions().contains(ActionType::find));
+    ASSERT(!privilege.getActions().contains(ActionType::insert));
+    ASSERT_EQUALS(privilege.getResourcePattern(),
+                  ResourcePattern::forAnySystemBucketsInAnyDatabase("foo"));
+
+    ASSERT(ParsedPrivilege::privilegeToParsedPrivilege(privilege, &parsedPrivilege, &errmsg));
+    ASSERT(parsedPrivilege.isValid(&errmsg));
+    ASSERT(parsedPrivilege.isResourceSet());
+    ASSERT_TRUE(parsedPrivilege.getResource().isSystemBucketsSet());
+    ASSERT_EQUALS(parsedPrivilege.getResource().getSystemBuckets(), "foo");
+    ASSERT_FALSE(parsedPrivilege.getResource().isClusterSet());
+    ASSERT_FALSE(parsedPrivilege.getResource().isDbSet());
+    ASSERT_FALSE(parsedPrivilege.getResource().isCollectionSet());
+    ASSERT(parsedPrivilege.isActionsSet());
+    ASSERT(actionsVector == parsedPrivilege.getActions());
+
+
+    // Works with any system.buckets resource in db test
+    parsedPrivilege.parseBSON(BSON("resource" << BSON("db"
+                                                      << "test"
+                                                      << "system_buckets"
+                                                      << "")
+                                              << "actions" << BSON_ARRAY("find")),
+                              &errmsg);
+    ASSERT(parsedPrivilege.isValid(&errmsg));
+    ASSERT_OK(ParsedPrivilege::parsedPrivilegeToPrivilege(
+        parsedPrivilege, &privilege, &unrecognizedActions));
+    ASSERT(unrecognizedActions.empty());
+    ASSERT(privilege.getActions().contains(ActionType::find));
+    ASSERT(!privilege.getActions().contains(ActionType::insert));
+    ASSERT_EQUALS(privilege.getResourcePattern(),
+                  ResourcePattern::forAnySystemBucketsInDatabase("test"));
+    ASSERT(ParsedPrivilege::privilegeToParsedPrivilege(privilege, &parsedPrivilege, &errmsg));
+    ASSERT(parsedPrivilege.isValid(&errmsg));
+    ASSERT(parsedPrivilege.isResourceSet());
+    ASSERT_TRUE(parsedPrivilege.getResource().isSystemBucketsSet());
+    ASSERT_EQUALS(parsedPrivilege.getResource().getSystemBuckets(), "");
+    ASSERT_FALSE(parsedPrivilege.getResource().isClusterSet());
+    ASSERT_TRUE(parsedPrivilege.getResource().isDbSet());
+    ASSERT_FALSE(parsedPrivilege.getResource().isCollectionSet());
+    ASSERT_EQUALS("test", parsedPrivilege.getResource().getDb());
     ASSERT(parsedPrivilege.isActionsSet());
     ASSERT(actionsVector == parsedPrivilege.getActions());
 }
diff --git a/src/mongo/db/auth/resource_pattern.cpp b/src/mongo/db/auth/resource_pattern.cpp
index fe42c22973b..75ede19ebb0 100644
--- a/src/mongo/db/auth/resource_pattern.cpp
+++ b/src/mongo/db/auth/resource_pattern.cpp
@@ -52,6 +52,15 @@ std::string ResourcePattern::toString() const {
             return "<all normal resources>";
         case MatchTypeEnum::kMatchAnyResource:
             return "<all resources>";
+        case MatchTypeEnum::kMatchExactSystemBucketResource:
+            return "<" + _ns.db().toString() + ".system.bucket" + _ns.coll().toString() +
+                " resources>";
+        case MatchTypeEnum::kMatchSystemBucketInAnyDBResource:
+            return "<any system.bucket." + _ns.coll().toString() + ">";
+        case MatchTypeEnum::kMatchAnySystemBucketInDBResource:
+            return "<" + _ns.db().toString() + "system.bucket.*>";
+        case MatchTypeEnum::kMatchAnySystemBucketResource:
+            return "<any system.bucket resources>";
         default:
             return "<unknown resource pattern type>";
     }
diff --git a/src/mongo/db/auth/resource_pattern.h b/src/mongo/db/auth/resource_pattern.h
index d82281d3625..e8678eab848 100644
--- a/src/mongo/db/auth/resource_pattern.h
+++ b/src/mongo/db/auth/resource_pattern.h
@@ -99,6 +99,43 @@ public:
     }
 
     /**
+     * Returns a pattern that matches any collection with the prefix "system.buckets." in any
+     * database.
+     */
+    static ResourcePattern forAnySystemBuckets() {
+        return ResourcePattern(MatchTypeEnum::kMatchAnySystemBucketResource);
+    }
+
+    /**
+     * Returns a pattern that matches any collection with the prefix "system.buckets." in database
+     * "db".
+     */
+    static ResourcePattern forAnySystemBucketsInDatabase(StringData dbName) {
+        return ResourcePattern(MatchTypeEnum::kMatchAnySystemBucketInDBResource,
+                               NamespaceString(dbName, ""));
+    }
+
+    /**
+     * Returns a pattern that matches any collection with the prefix "system.buckets.<collection>"
+     * in any database.
+     */
+    static ResourcePattern forAnySystemBucketsInAnyDatabase(StringData collectionName) {
+        return ResourcePattern(MatchTypeEnum::kMatchSystemBucketInAnyDBResource,
+                               NamespaceString("", collectionName));
+    }
+
+    /**
+     * Returns a pattern that matches a collection with the name
+     * "<dbName>.system.buckets.<collectionName>"
+     */
+    static ResourcePattern forExactSystemBucketsCollection(StringData dbName,
+                                                           StringData collectionName) {
+        invariant(!collectionName.startsWith("system.buckets."));
+        return ResourcePattern(MatchTypeEnum::kMatchExactSystemBucketResource,
+                               NamespaceString(dbName, collectionName));
+    }
+
+    /**
      * Constructs a pattern that never matches.
      */
     ResourcePattern() : _matchType(MatchTypeEnum::kMatchNever) {}
@@ -146,6 +183,34 @@ public:
     }
 
     /**
+     * Returns true if this pattern matches a <db>.system.buckets.<collection name>.
+     */
+    bool isExactSystemBucketsCollection() const {
+        return _matchType == MatchTypeEnum::kMatchExactSystemBucketResource;
+    }
+
+    /**
+     * Returns true if this pattern matches a system.buckets.<collection name> in any db.
+     */
+    bool isAnySystemBucketsCollectionInAnyDB() const {
+        return _matchType == MatchTypeEnum::kMatchSystemBucketInAnyDBResource;
+    }
+
+    /**
+     * Returns true if this pattern matches a system.buckets.* in <db>.
+     */
+    bool isAnySystemBucketsCollectionInDB() const {
+        return _matchType == MatchTypeEnum::kMatchAnySystemBucketInDBResource;
+    }
+
+    /**
+     * Returns true if this pattern matches any collection prefixed with system.buckets
+     */
+    bool isAnySystemBucketsCollection() const {
+        return _matchType == MatchTypeEnum::kMatchAnySystemBucketResource;
+    }
+
+    /**
      * Returns the namespace that this pattern matches.
      *
      * Behavior is undefined unless isExactNamespacePattern() is true.
@@ -158,7 +223,7 @@ public:
      * Returns the database that this pattern matches.
      *
      * Behavior is undefined unless the pattern is of type matchDatabaseName or
-     * matchExactNamespace
+     * matchExactNamespace or matchExactSystemBucketResource or matchAnySystemBucketInDBResource
      */
     StringData databaseToMatch() const {
         return _ns.db();
@@ -168,7 +233,7 @@ public:
      * Returns the collection that this pattern matches.
      *
      * Behavior is undefined unless the pattern is of type matchCollectionName or
-     * matchExactNamespace
+     * matchExactNamespace or matchExactSystemBucketResource
      */
     StringData collectionToMatch() const {
         return _ns.coll();
diff --git a/src/mongo/s/commands/cluster_list_collections_cmd.cpp b/src/mongo/s/commands/cluster_list_collections_cmd.cpp
index 52a144059dc..d7c4ae202eb 100644
--- a/src/mongo/s/commands/cluster_list_collections_cmd.cpp
+++ b/src/mongo/s/commands/cluster_list_collections_cmd.cpp
@@ -43,6 +43,8 @@
 namespace mongo {
 namespace {
 
+constexpr auto systemBucketsDot = "system.buckets."_sd;
+
 bool cursorCommandPassthroughPrimaryShard(OperationContext* opCtx,
                                           StringData dbName,
                                           const CachedDatabaseInfo& dbInfo,
@@ -111,6 +113,16 @@ BSONObj rewriteCommandForListingOwnCollections(OperationContext* opCtx,
         uassertStatusOK(newFilterOr.pushBack(systemCollectionsFilter));
     }
 
+    // system_buckets DB resource grants all system_buckets.* collections so create a filter to
+    // include them
+    if (authzSession->isAuthorizedForAnyActionOnResource(
+            ResourcePattern::forAnySystemBucketsInDatabase(dbName)) ||
+        authzSession->isAuthorizedForAnyActionOnResource(ResourcePattern::forAnySystemBuckets())) {
+        mutablebson::Element systemCollectionsFilter = rewrittenCmdObj.makeElementObject(
+            "", BSON("name" << BSON("$regex" << BSONRegEx("^system\\.buckets\\."))));
+        uassertStatusOK(newFilterOr.pushBack(systemCollectionsFilter));
+    }
+
     // Compute the set of collection names which would be permissible to return.
     std::set<std::string> collectionNames;
     for (UserNameIterator nameIter = authzSession->getAuthenticatedUserNames(); nameIter.more();
@@ -121,6 +133,12 @@ BSONObj rewriteCommandForListingOwnCollections(OperationContext* opCtx,
                 (resource.isExactNamespacePattern() && resource.databaseToMatch() == dbName)) {
                 collectionNames.emplace(resource.collectionToMatch().toString());
             }
+
+            if (resource.isAnySystemBucketsCollectionInAnyDB() ||
+                (resource.isExactSystemBucketsCollection() &&
+                 resource.databaseToMatch() == dbName)) {
+                collectionNames.emplace(systemBucketsDot + resource.collectionToMatch().toString());
+            }
         }
     }