Merge pull request ceph#55320 from rishabh-d-dave/mdscaps-update-issues

cephfs,mon: fix bugs related to updating MDS caps

Reviewed-by: Venky Shankar <[email protected]>

Author: rishabh-d-dave

qa/tasks/cephfs/caps_helper.py

@@ -249,7 +249,7 @@ def run_mds_cap_tests(self, perm, mntpt=None):
         Run test for read perm and, for write perm, run positive test if it
         is present and run negative test if not.
         """
-        if mntpt:
+        if mntpt and mntpt != '/':
             # beacaue we want to value of mntpt from test_set.path along with
             # slash that precedes it.
             mntpt = '/' + mntpt if mntpt[0] != '/' else mntpt

qa/tasks/cephfs/test_admin.py

@@ -1519,7 +1519,98 @@ def test_multiple_path_rw(self):
 
         self._remount_and_run_tests(FS_AUTH_CAPS, keyring)
 
+    def test_when_MDS_caps_needs_update_but_others_dont(self):
+        '''
+        Test that the command "ceph fs authorize" successfully updates MDS
+        caps even when MON and OSD caps don't need an update.
+
+        Tests: https://tracker.ceph.com/issues/64182
+
+        In this test we run -
+
+            ceph fs authorize cephfs1 client.x / rw
+            ceph fs authorize cephfs2 client.x / rw
+            ceph fs authorize cephfs2 client.x /dir1 rw
+
+        The first command will create the keyring by adding the MDS cap for
+        root path & MON and OSD caps with name of the FS name (say "cephfs1").
+        Second command will update the all of client's caps -- MON, OSD and
+        MDS caps to add caps for 2nd CephFS. The third command doesn't need
+        to add MON and OSD caps since cap for "cephfs2" has been granted
+        already. Thus, third command only need to update the MDS cap, thus
+        testing the bug under consideration here.
+        '''
+        PERM = 'rw'
+        DIR = 'dir1'
+
+        self.fs2 = self.mds_cluster.newfs(name='cephfs2', create=True)
+        self.mount_b.remount(cephfs_name=self.fs2.name)
+        self.mount_b.run_shell(f'mkdir {DIR}')
+        self.captesters = (CapTester(self.mount_a, '/'),
+                           CapTester(self.mount_b, '/'),
+                           CapTester(self.mount_b, f'/{DIR}'))
+
+        FS_AUTH_CAPS = (('/', PERM), ('/', PERM), (DIR, PERM))
+        keyring = self.fs.authorize(self.client_id, FS_AUTH_CAPS[0])
+        keyring = self.fs2.authorize(self.client_id, FS_AUTH_CAPS[1])
+
+        # if the following block of code pass it implies that
+        # https://tracker.ceph.com/issues/64182 has been fixed
+        keyring = self.fs2.authorize(self.client_id, FS_AUTH_CAPS[2])
+        mdscaps = ('caps mds = "'
+                   f'allow {PERM} fsname={self.fs.name}, '
+                   f'allow {PERM} fsname={self.fs2.name}, '
+                   f'allow {PERM} fsname={self.fs2.name} path={DIR}')
+        self.assertIn(mdscaps, keyring)
+
+        self._remount_and_run_tests_for_cap(
+            FS_AUTH_CAPS[0], self.captesters[0], self.fs.name, self.mount_a,
+            keyring)
+        self._remount_and_run_tests_for_cap(
+            FS_AUTH_CAPS[1], self.captesters[1], self.fs2.name, self.mount_b,
+            keyring)
+        self._remount_and_run_tests_for_cap(
+            FS_AUTH_CAPS[2], self.captesters[2], self.fs2.name, self.mount_b,
+            keyring)
+
+    def test_adding_multiple_caps(self):
+        '''
+        Test that the command "ceph fs authorize" is successful in updating
+        the entity's caps when multiple caps are passed to it in one go.
+
+        Tests: https://tracker.ceph.com/issues/64127
+        Tests: https://tracker.ceph.com/issues/64417
+        '''
+        DIR = 'dir1'
+        self.mount_a.run_shell(f'mkdir {DIR}')
+        self.captesters = (CapTester(self.mount_a, '/'),
+                           CapTester(self.mount_a, f'/{DIR}'))
+        self.fs.authorize(self.client_id, ('/', 'rw'))
+
+        FS_AUTH_CAPS = ('/', 'rw', 'root_squash'), (f'/{DIR}', 'rw' )
+        # The fact that following line passes means
+        # https://tracker.ceph.com/issues/64127 has been fixed
+        keyring = self.fs.authorize(self.client_id, FS_AUTH_CAPS)
+
+        # The fact that following lines passes means
+        # https://tracker.ceph.com/issues/64417 has been fixed.
+        mdscaps = (f'allow rw fsname={self.fs.name} root_squash, '
+                   f'allow rw fsname={self.fs.name} path={DIR}')
+        self.assertIn(mdscaps, keyring)
+
+        self._remount_and_run_tests(FS_AUTH_CAPS, keyring)
+
     def _remount_and_run_tests(self, fs_auth_caps, keyring):
+        '''
+        This method is specifically designed to meet needs of most of the
+        test case in this class. Following are assumptions made here -
+
+        1. CephFS to be mounted is self.fs
+        2. Mount object to be used is self.mount_a
+        3. Keyring file will be created on the host specified in self.mount_a.
+        4. CephFS dir that will serve as root is PATH component of particular
+           cap from FS_AUTH_CAPS.
+        '''
         for i, c in enumerate(fs_auth_caps):
             self.assertIn(i, (0, 1))
             PATH = c[0]
@@ -1529,18 +1620,30 @@ def _remount_and_run_tests(self, fs_auth_caps, keyring):
             self.captesters[i].run_cap_tests(self.fs, self.client_id, PERM,
                                              PATH)
 
-    def tearDown(self):
-        self.mount_a.umount_wait()
-        self.run_ceph_cmd(f'auth rm {self.client_name}')
-
-        super(type(self), self).tearDown()
-
     def _remount(self, keyring, path='/'):
         keyring_path = self.mount_a.client_remote.mktemp(data=keyring)
         self.mount_a.remount(client_id=self.client_id,
                              client_keyring_path=keyring_path,
                              cephfs_mntpt=path)
 
+    def _remount_and_run_tests_for_cap(self, cap, captester, fsname, mount,
+                                       keyring):
+        PATH = cap[0]
+        PERM = cap[1]
+
+        cephfs_mntpt = os_path_join('/', PATH)
+        keyring_path = mount.client_remote.mktemp(data=keyring)
+        mount.remount(client_id=self.client_id, cephfs_mntpt=cephfs_mntpt,
+                      cephfs_name=fsname, client_keyring_path=keyring_path)
+
+        captester.run_cap_tests(self.fs, self.client_id, PERM, PATH)
+
+    def tearDown(self):
+        self.mount_a.umount_wait()
+        self.run_ceph_cmd(f'auth rm {self.client_name}')
+
+        super(type(self), self).tearDown()
+
 
 class TestFsAuthorizeUpdate(CephFSTestCase):
     client_id = 'testuser'

src/mds/MDSAuthCaps.cc

@@ -382,25 +382,46 @@ bool MDSAuthCaps::parse(string_view str, ostream *err)
   }
 }
 
-bool MDSAuthCaps::merge(MDSAuthCaps newcap)
+/* Check if the "cap grant" is already present in this cap object. If it is,
+ * return false. If not, add it and return true.
+ *
+ * ng = new grant, new mds cap grant.
+ */
+bool MDSAuthCaps::merge_one_cap_grant(MDSCapGrant ng)
 {
-  ceph_assert(newcap.grants.size() == 1);
-  auto ng = newcap.grants[0];
-
+  // check if "ng" is already present in this cap object.
   for (auto& g : grants) {
     if (g.match.fs_name == ng.match.fs_name && g.match.path == ng.match.path) {
-      if (g.spec.get_caps() == ng.spec.get_caps()) {
-	// no update required. maintaining idempotency.
+      if (g.spec.get_caps() == ng.spec.get_caps() &&
+	  g.match.root_squash == ng.match.root_squash) {
+	// Since all components of MDS caps (fsname, path, perm/spec and
+	// root_squash) matched, it means cap same as "ng" is present in MDS
+	// cap grant list. No need to look further in MDS cap grant list.
+	// No update is required. Maintain idempotency.
 	return false;
-       } else {
-	// cap for given fs name is present, let's update it.
+       }
+
+      // fsname and path match but perm/spec is different. update the cap
+      // with new perm/spec.
+      if (g.spec.get_caps() != ng.spec.get_caps()) {
 	g.spec.set_caps(ng.spec.get_caps());
-	return true;
       }
+
+      // fsname and path match but value of root_squash is different. update
+      // its value.
+      if (g.match.root_squash != ng.match.root_squash) {
+	g.match.root_squash = ng.match.root_squash;
+      }
+
+      // Since fsname and path matched and either perm/spec or root_squash
+      // or both has been updated, cap from "ng" has been incorporated
+      // into this cap grant list. Time to return.
+      return true;
     }
   }
 
-  // cap for given fs name and/or path is absent, let's add a new cap for it.
+  // Since a cap grant like "ng" is absent in this cap object's grant list,
+  // add "ng" to the cap grant list.
   grants.push_back(MDSCapGrant(
     MDSCapSpec(ng.spec.get_caps()),
     MDSCapMatch(ng.match.fs_name, ng.match.path, ng.match.root_squash),
@@ -409,6 +430,29 @@ bool MDSAuthCaps::merge(MDSAuthCaps newcap)
   return true;
 }
 
+/* User can pass one or MDS caps that it wishes to add to entity's keyring.
+ * Merge all of these caps one by one. Return value indicates whether or not
+ * AuthMonitor must update the entity's keyring.
+ *
+ * If all caps do not merge (that is, underlying helper method returns false
+ * after attempting merge), no update is required. Return false so that
+ * AuthMonitor doesn't run the update procedure for caps.
+ *
+ * If even one cap is merged (that is, underlying method returns true even
+ * once), an update to the entity's keyring is required. Return true so that
+ * AuthMonitor runs the update procedure.
+ */
+bool MDSAuthCaps::merge(MDSAuthCaps newcaps)
+{
+  bool were_caps_merged = false;
+
+  for (auto& ng : newcaps.grants) {
+      were_caps_merged |= merge_one_cap_grant(ng);
+  }
+
+  return were_caps_merged;
+}
+
 string MDSCapMatch::to_string()
 {
   string str = "";

src/mds/MDSAuthCaps.h

@@ -259,7 +259,8 @@ class MDSAuthCaps
 
   void set_allow_all();
   bool parse(std::string_view str, std::ostream *err);
-  bool merge(MDSAuthCaps newcap);
+  bool merge_one_cap_grant(MDSCapGrant ng);
+  bool merge(MDSAuthCaps newcaps);
 
   bool allow_all() const;
   bool is_capable(std::string_view inode_path,

src/mon/AuthMonitor.cc

@@ -1867,6 +1867,9 @@ AuthMonitor::caps_update AuthMonitor::_gen_wanted_caps(EntityAuth& e_auth,
   map<string, string>& newcaps, ostream& out)
 {
   caps_update is_caps_update_reqd = CAPS_UPDATE_NOT_REQD;
+  caps_update is_caps_update_reqd_mon = CAPS_UPDATE_NOT_REQD;
+  caps_update is_caps_update_reqd_osd = CAPS_UPDATE_NOT_REQD;
+  caps_update is_caps_update_reqd_mds = CAPS_UPDATE_NOT_REQD;
 
   if (e_auth.caps.empty()) {
     return CAPS_UPDATE_REQD;
@@ -1888,15 +1891,29 @@ AuthMonitor::caps_update AuthMonitor::_gen_wanted_caps(EntityAuth& e_auth,
     }
 
     if (cap_entity == "mon") {
-      is_caps_update_reqd = _merge_caps<MonCap>(cap_entity, new_cap_str,
+      is_caps_update_reqd_mon = _merge_caps<MonCap>(cap_entity, new_cap_str,
 	cur_cap_str, newcaps, out);
     } else if (cap_entity == "osd") {
-      is_caps_update_reqd = _merge_caps<OSDCap>(cap_entity, new_cap_str,
+      is_caps_update_reqd_osd = _merge_caps<OSDCap>(cap_entity, new_cap_str,
 	cur_cap_str, newcaps, out);
     } else if (cap_entity == "mds") {
-      is_caps_update_reqd = _merge_caps<MDSAuthCaps>(cap_entity, new_cap_str,
-	cur_cap_str, newcaps, out);
-    }
+      is_caps_update_reqd_mds = _merge_caps<MDSAuthCaps>(cap_entity,
+	new_cap_str, cur_cap_str, newcaps, out);
+    }
+  }
+
+  // if any one of MON, OSD or MDS caps failed to parse, it is pointless
+  // to run the update procedure.
+  if (is_caps_update_reqd_mon == CAPS_PARSING_ERR ||
+      is_caps_update_reqd_osd == CAPS_PARSING_ERR ||
+      is_caps_update_reqd_mds == CAPS_PARSING_ERR) {
+    is_caps_update_reqd = CAPS_PARSING_ERR;
+  // even if any one of MON, OSD or MDS caps needs an update, the update
+  // procedure needs to be executed.
+  } else if (is_caps_update_reqd_mon == CAPS_UPDATE_REQD ||
+      is_caps_update_reqd_osd == CAPS_UPDATE_REQD ||
+      is_caps_update_reqd_mds == CAPS_UPDATE_REQD) {
+    is_caps_update_reqd = CAPS_UPDATE_REQD;
   }
 
   return is_caps_update_reqd;