Merge PR ceph#59435 into main

* refs/pull/59435/head:
	mgr/volumes: Fix json.loads for test on mon caps
	mgr/volumes: Add test for mon caps if auth key has remaining mds/osd caps
	mgr/volumes: Keep mon caps if auth key has remaining mds/osd caps

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

Author: vshankar

qa/tasks/cephfs/test_volumes.py

@@ -3119,6 +3119,64 @@ def test_authorize_deauthorize_subvolume(self):
         self._fs_cmd("subvolume", "rm", self.volname, subvolume, "--group_name", group)
         self._fs_cmd("subvolumegroup", "rm", self.volname, group)
 
+    def test_subvolume_deauthorize_with_shared_key(self):
+        """
+        That mon caps are preserved when one cephx key authorized on multiple
+        subvolumes is deauthorized on any of those.
+        """
+        subvolume1 = self._gen_subvol_name()
+        subvolume2 = self._gen_subvol_name()
+        group = self._gen_subvol_grp_name()
+        authid = "alice"
+
+        # create group
+        self._fs_cmd("subvolumegroup", "create", self.volname, group)
+
+        # create subvolumes
+        self._fs_cmd("subvolume", "create", self.volname, subvolume1, "--group_name", group)
+        self._fs_cmd("subvolume", "create", self.volname, subvolume2, "--group_name", group)
+
+        # authorize alice authID read-write access to both subvolumes
+        self._fs_cmd("subvolume", "authorize", self.volname, subvolume1, authid,
+                     "--group_name", group)
+        self._fs_cmd("subvolume", "authorize", self.volname, subvolume2, authid,
+                     "--group_name", group)
+
+        # verify autorized-id has access to both subvolumes
+        expected_auth_list = [{'alice': 'rw'}]
+        auth_list1 = json.loads(self._fs_cmd('subvolume', 'authorized_list', self.volname, subvolume1, "--group_name", group))
+        self.assertEqual(expected_auth_list, auth_list1)
+        auth_list2 = json.loads(self._fs_cmd('subvolume', 'authorized_list', self.volname, subvolume2, "--group_name", group))
+        self.assertEqual(expected_auth_list, auth_list2)
+
+        # check mon caps for authid
+        expected_mon_caps = 'allow r'
+        full_caps = json.loads(self._raw_cmd("auth", "get", "client.alice", "--format=json-pretty"))
+        self.assertEqual(expected_mon_caps, full_caps[0]['caps']['mon'])
+
+        # deauthorize guest1 authID
+        self._fs_cmd("subvolume", "deauthorize", self.volname, subvolume2, authid,
+                     "--group_name", group)
+
+        # verify autorized-id has access to subvolume1 only
+        expected_auth_list = [{'alice': 'rw'}]
+        auth_list1 = json.loads(self._fs_cmd('subvolume', 'authorized_list', self.volname, subvolume1, "--group_name", group))
+        self.assertEqual(expected_auth_list, auth_list1)
+        auth_list2 = json.loads(self._fs_cmd('subvolume', 'authorized_list', self.volname, subvolume2, "--group_name", group))
+        self.assertEqual([], auth_list2)
+
+        # check mon caps still hold for authid
+        expected_mon_caps = 'allow r'
+        full_caps = json.loads(self._raw_cmd("auth", "get", "client.alice", "--format=json-pretty"))
+        self.assertEqual(expected_mon_caps, full_caps[0]['caps']['mon'])
+
+        # cleanup
+        self._fs_cmd("subvolume", "deauthorize", self.volname, subvolume1, authid,
+                     "--group_name", group)
+        self._fs_cmd("subvolume", "rm", self.volname, subvolume1, "--group_name", group)
+        self._fs_cmd("subvolume", "rm", self.volname, subvolume2, "--group_name", group)
+        self._fs_cmd("subvolumegroup", "rm", self.volname, group)
+
     def test_multitenant_subvolumes(self):
         """
         That subvolume access can be restricted to a tenant.

src/pybind/mgr/volumes/fs/operations/access.py

@@ -125,7 +125,12 @@ def cap_remove(orig_mds_caps, orig_osd_caps, want_mds_caps, want_osd_caps):
     mds_cap_str, osd_cap_str = cap_remove(orig_mds_caps, orig_osd_caps,
                                           want_mds_caps, want_osd_caps)
 
-    caps_list = prepare_updated_caps_list(cap, mds_cap_str, osd_cap_str, authorize=False)
+    # The same auth key may be used for multiple subvolumes
+    # If upon cap_remove the key still has mds or osd caps, it must also keep
+    # mon caps so that the client is allowed to check in with the mons.
+    auth = True if mds_cap_str or osd_cap_str else False
+
+    caps_list = prepare_updated_caps_list(cap, mds_cap_str, osd_cap_str, authorize=auth)
     if not caps_list:
         mgr.mon_command(
             {