1 files changed, 86 insertions, 0 deletions
diff --git a/qa/tasks/cephfs/test_admin.py b/qa/tasks/cephfs/test_admin.py
index 4f3100bbe..db0e5660a 100644
--- a/qa/tasks/cephfs/test_admin.py
+++ b/qa/tasks/cephfs/test_admin.py
@@ -1319,6 +1319,92 @@ class TestFsAuthorize(CephFSTestCase):
         self.captester.conduct_neg_test_for_chown_caps()
         self.captester.conduct_neg_test_for_truncate_caps()
 
+    def test_multifs_rootsquash_nofeature(self):
+        """
+        That having root_squash on one fs doesn't prevent access to others.
+        """
+
+        if not isinstance(self.mount_a, FuseMount):
+            self.skipTest("only FUSE client has CEPHFS_FEATURE_MDS_AUTH_CAPS "
+                          "needed to enforce root_squash MDS caps")
+
+        self.fs1 = self.fs
+        self.fs2 = self.mds_cluster.newfs('testcephfs2')
+
+        self.mount_a.umount_wait()
+
+        self.run_ceph_cmd(f'auth caps client.{self.mount_a.client_id} '
+                          f'mon "allow r" '
+                          f'osd "allow rw tag cephfs data={self.fs1.name}, allow rw tag cephfs data={self.fs2.name}" '
+                          f'mds "allow rwp fsname={self.fs1.name}, allow rw fsname={self.fs2.name} root_squash"')
+
+        CEPHFS_FEATURE_MDS_AUTH_CAPS_CHECK = 21
+        # all but CEPHFS_FEATURE_MDS_AUTH_CAPS_CHECK
+        features = ",".join([str(i) for i in range(CEPHFS_FEATURE_MDS_AUTH_CAPS_CHECK)])
+        mntargs = [f"--client_debug_inject_features={features}"]
+
+        # should succeed
+        with self.assert_cluster_log("report clients with broken root_squash", present=False):
+            self.mount_a.remount(mntargs=mntargs, cephfs_name=self.fs1.name)
+
+    def test_rootsquash_nofeature(self):
+        """
+        That having root_squash on an fs without the feature bit raises a HEALTH_ERR warning.
+        """
+
+        if not isinstance(self.mount_a, FuseMount):
+            self.skipTest("only FUSE client has CEPHFS_FEATURE_MDS_AUTH_CAPS "
+                          "needed to enforce root_squash MDS caps")
+
+        self.mount_a.umount_wait()
+
+        FS_AUTH_CAPS = (('/', 'rw', 'root_squash'),)
+        keyring = self.fs.authorize(self.client_id, FS_AUTH_CAPS)
+
+        CEPHFS_FEATURE_MDS_AUTH_CAPS_CHECK = 21
+        # all but CEPHFS_FEATURE_MDS_AUTH_CAPS_CHECK
+        features = ",".join([str(i) for i in range(CEPHFS_FEATURE_MDS_AUTH_CAPS_CHECK)])
+        mntargs = [f"--client_debug_inject_features={features}"]
+
+        # should succeed
+        with self.assert_cluster_log("with broken root_squash implementation"):
+            keyring_path = self.mount_a.client_remote.mktemp(data=keyring)
+            self.mount_a.remount(client_id=self.client_id, client_keyring_path=keyring_path, mntargs=mntargs, cephfs_name=self.fs.name)
+            self.wait_for_health("MDS_CLIENTS_BROKEN_ROOTSQUASH", 60)
+            self.assertFalse(self.mount_a.is_blocked())
+
+        self.mount_a.umount_wait()
+        self.wait_for_health_clear(60)
+
+    def test_rootsquash_nofeature_evict(self):
+        """
+        That having root_squash on an fs without the feature bit can be evicted.
+        """
+
+        if not isinstance(self.mount_a, FuseMount):
+            self.skipTest("only FUSE client has CEPHFS_FEATURE_MDS_AUTH_CAPS "
+                          "needed to enforce root_squash MDS caps")
+
+        self.mount_a.umount_wait()
+
+        FS_AUTH_CAPS = (('/', 'rw', 'root_squash'),)
+        keyring = self.fs.authorize(self.client_id, FS_AUTH_CAPS)
+
+        CEPHFS_FEATURE_MDS_AUTH_CAPS_CHECK = 21
+        # all but CEPHFS_FEATURE_MDS_AUTH_CAPS_CHECK
+        features = ",".join([str(i) for i in range(CEPHFS_FEATURE_MDS_AUTH_CAPS_CHECK)])
+        mntargs = [f"--client_debug_inject_features={features}"]
+
+        # should succeed
+        keyring_path = self.mount_a.client_remote.mktemp(data=keyring)
+        self.mount_a.remount(client_id=self.client_id, client_keyring_path=keyring_path, mntargs=mntargs, cephfs_name=self.fs.name)
+        self.wait_for_health("MDS_CLIENTS_BROKEN_ROOTSQUASH", 60)
+
+        self.fs.required_client_features("add", "client_mds_auth_caps")
+        self.wait_for_health_clear(60)
+        self.assertTrue(self.mount_a.is_blocked())
+
+
     def test_single_path_rootsquash_issue_56067(self):
         """
         That a FS client using root squash MDS caps allows non-root user to write data