Merge PR ceph#52859 into main

* refs/pull/52859/head:
	qa: test cases to make sure invalid paths don't get updated
	mgr/nfs: use helper to validate cephfs path
	mgr/nfs: validate path before updating a cephfs export
	mgr/nfs: add a helper to validate cephfs path

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

Author: vshankar

qa/tasks/cephfs/test_nfs.py

@@ -411,6 +411,21 @@ def _nfs_export_apply(self, cluster, exports, raise_on_error=False):
                                     stdin=json.dumps(exports),
                                     stdout=StringIO(), stderr=StringIO())
 
+    def update_export(self, cluster_id, path, pseudo, fs_name):
+        self.ctx.cluster.run(args=['ceph', 'nfs', 'export', 'apply',
+                                   cluster_id, '-i', '-'],
+                             stdin=json.dumps({
+                                 "path": path,
+                                 "pseudo": pseudo,
+                                 "squash": "none",
+                                 "access_type": "rw",
+                                 "protocols": [4],
+                                 "fsal": {
+                                     "name": "CEPH",
+                                     "fs_name": fs_name
+                                 }
+                             }))
+
     def test_create_and_delete_cluster(self):
         '''
         Test successful creation and deletion of the nfs cluster.
@@ -1138,3 +1153,61 @@ def test_pseudo_path_in_json_response_when_updating_exports_failed(self):
         finally:
             self._delete_cluster_with_fs(self.fs_name, mnt_pt)
             self.ctx.cluster.run(args=['rm', '-rf', f'{mnt_pt}'])
+
+    def test_cephfs_export_update_with_nonexistent_dir(self):
+        """
+        Test that invalid path is not allowed while updating a CephFS
+        export.
+        """
+        self._create_cluster_with_fs(self.fs_name)
+        self._create_export(export_id=1)
+
+        try:
+            self.update_export(self.cluster_id, "/not_existent_dir",
+                               self.pseudo_path, self.fs_name)
+        except CommandFailedError as e:
+            if e.exitstatus != errno.ENOENT:
+                raise
+
+        self._delete_export()
+        self._delete_cluster_with_fs(self.fs_name)
+
+    def test_cephfs_export_update_at_non_dir_path(self):
+        """
+        Test that non-directory path are not allowed while updating a CephFS
+        export.
+        """
+        mnt_pt = '/mnt'
+        preserve_mode = self._sys_cmd(['stat', '-c', '%a', mnt_pt])
+        self._create_cluster_with_fs(self.fs_name, mnt_pt)
+        try:
+            self.ctx.cluster.run(args=['touch', f'{mnt_pt}/testfile'])
+            self._create_export(export_id=1)
+
+            # test at a file path
+            try:
+                self.update_export(self.cluster_id, "/testfile",
+                                   self.pseudo_path, self.fs_name)
+            except CommandFailedError as e:
+                if e.exitstatus != errno.ENOTDIR:
+                    raise
+
+            # test at a symlink path
+            self.ctx.cluster.run(args=['mkdir', f'{mnt_pt}/testdir'])
+            self.ctx.cluster.run(args=['ln', '-s', f'{mnt_pt}/testdir',
+                                       f'{mnt_pt}/testdir_symlink'])
+            try:
+                self.update_export(self.cluster_id, "/testdir_symlink",
+                                   self.pseudo_path, self.fs_name)
+            except CommandFailedError as e:
+                if e.exitstatus != errno.ENOTDIR:
+                    raise
+
+            # verify the path wasn't changed
+            export = json.loads(self._nfs_cmd("export", "ls",
+                                              self.cluster_id, "--detailed"))
+            self.assertEqual(export[0]["pseudo"], "/cephfs")
+
+        finally:
+            self.ctx.cluster.run(args=['rm', '-rf', f'{mnt_pt}/*'])
+            self._delete_cluster_with_fs(self.fs_name, mnt_pt, preserve_mode)

src/pybind/mgr/nfs/export.py

@@ -71,6 +71,17 @@ def normalize_path(path: str) -> str:
     return path
 
 
+def validate_cephfs_path(mgr: 'Module', fs_name: str, path: str) -> None:
+    try:
+        cephfs_path_is_dir(mgr, fs_name, path)
+    except NotADirectoryError:
+        raise NFSException(f"path {path} is not a dir", -errno.ENOTDIR)
+    except cephfs.ObjectNotFound:
+        raise NFSObjectNotFound(f"path {path} does not exist")
+    except cephfs.Error as e:
+        raise NFSException(e.args[1], -e.args[0])
+
+
 class NFSRados:
     def __init__(self, rados: 'Rados', namespace: str) -> None:
         self.rados = rados
@@ -675,6 +686,8 @@ def create_export_from_dict(self,
             if not check_fs(self.mgr, fs_name):
                 raise FSNotFound(fs_name)
 
+            validate_cephfs_path(self.mgr, fs_name, path)
+
             user_id = f"nfs.{cluster_id}.{ex_id}"
             if "user_id" in fsal and fsal["user_id"] != user_id:
                 raise NFSInvalidOperation(f"export FSAL user_id must be '{user_id}'")
@@ -701,14 +714,7 @@ def create_cephfs_export(self,
                              clients: list = [],
                              sectype: Optional[List[str]] = None) -> Dict[str, Any]:
 
-        try:
-            cephfs_path_is_dir(self.mgr, fs_name, path)
-        except NotADirectoryError:
-            raise NFSException(f"path {path} is not a dir", -errno.ENOTDIR)
-        except cephfs.ObjectNotFound:
-            raise NFSObjectNotFound(f"path {path} does not exist")
-        except cephfs.Error as e:
-            raise NFSException(e.args[1], -e.args[0])
+        validate_cephfs_path(self.mgr, fs_name, path)
 
         pseudo_path = normalize_path(pseudo_path)