[FIX] fs_storage: Resolve rooted_dir sub path is inside path virtually

Before this fix, a FileNotFoundError would ocassionally arise when
'os.getcwd()' is called due to the rooted_dir _join method to check
whether the sub path is part of the parent path.
This is probably caused by the number and/or combination of workers
and threads and the transactions to the FS storage, having race
issues.

After this commit, the sub path is checked if it is contained
inside the parent path virtually via posixpath and pathlib modules
avoiding 'os.getcwd()' calls and running into this possible race
condition.

Author: hnavarro-kernet

fs_storage/rooted_dir_file_system.py

@@ -1,10 +1,10 @@
 # Copyright 2023 ACSONE SA/NV (https://www.acsone.eu).
 # License LGPL-3.0 or later (http://www.gnu.org/licenses/lgpl).
 
-import os
+import posixpath
+from pathlib import PurePosixPath
 
 from fsspec.implementations.dirfs import DirFileSystem
-from fsspec.implementations.local import make_path_posix
 from fsspec.registry import register_implementation
 
 
@@ -20,18 +20,31 @@ class RootedDirFileSystem(DirFileSystem):
     """
 
     def _join(self, path):
-        path = super()._join(path)
-        # Ensure that the path is a subpath of the root path by resolving
-        # any relative paths.
-        # Since the path separator is not always the same on all systems,
-        # we need to normalize the path separator.
-        path_posix = os.path.normpath(make_path_posix(path))
-        root_posix = os.path.normpath(make_path_posix(self.path))
-        if not path_posix.startswith(root_posix):
+        if path is not None and str(path) != "":
+            target = PurePosixPath(str(path))
+            if target.is_absolute() or (":" in str(target).split("/")[0]):
+                raise PermissionError(
+                    f"Absolute paths are not allowed in rooted_dir; got {path}"
+                )
+
+        joined = super()._join(path)
+
+        root = PurePosixPath(self.path).as_posix()
+        candidate = PurePosixPath(joined or ".").as_posix()
+
+        # Anchor candidate under root
+        candidate = posixpath.join(root, candidate)
+
+        # Normalize both paths to check and enforce containment
+        rnorm = posixpath.normpath(root)
+        cnorm = posixpath.normpath(candidate)
+
+        if not (cnorm == rnorm or cnorm.startswith(rnorm + "/")):
             raise PermissionError(
-                "Path %s is not a subpath of the root path %s" % (path, self.path)
+                f"Path {path} is not a subpath of the root path {self.path}"
             )
-        return path
+
+        return joined
 
 
 register_implementation("rooted_dir", RootedDirFileSystem)