Implements file duplication with robust path validation and enhanced frontend error handling

Author: daizutabi

frontend/src/components/editor/file-tree/requesting-tree.tsx

@@ -79,6 +79,10 @@ export class RequestingTree {
   async copy(id: string, newName: string): Promise<void> {
     const node = this.delegate.find(id);
     if (!node) {
+      toast({
+        title: "Failed",
+        description: `Node with id ${id} not found in the tree`,
+      });
       return;
     }
     const currentPath = node.data.path as FilePath;
@@ -106,6 +110,10 @@ export class RequestingTree {
   async rename(id: string, name: string): Promise<void> {
     const node = this.delegate.find(id);
     if (!node) {
+      toast({
+        title: "Failed",
+        description: `Node with id ${id} not found in the tree`,
+      });
       return;
     }
     const currentPath = node.data.path as FilePath;
@@ -201,6 +209,10 @@ export class RequestingTree {
   async delete(id: string): Promise<void> {
     const node = this.delegate.find(id);
     if (!node) {
+      toast({
+        title: "Failed",
+        description: `Node with id ${id} not found in the tree`,
+      });
       return;
     }
 

marimo/_server/api/endpoints/file_explorer.py

@@ -159,6 +159,7 @@ async def delete_file_or_directory(
     """
     body = await parse_request(request, cls=FileDeleteRequest)
     try:
+        # TODO: Refactor this side-effect based validation to a dedicated validation.
         file_system.get_details(body.path)
         success = file_system.delete_file_or_directory(body.path)
         return FileDeleteResponse(success=success)
@@ -189,6 +190,7 @@ async def copy_file_or_directory(
     """
     body = await parse_request(request, cls=FileCopyRequest)
     try:
+        # TODO: Refactor this side-effect based validation to a dedicated validation.
         file_system.get_details(body.path)
         info = file_system.copy_file_or_directory(body.path, body.new_path)
         return FileCopyResponse(success=True, info=info)
@@ -219,6 +221,7 @@ async def move_file_or_directory(
     """
     body = await parse_request(request, cls=FileMoveRequest)
     try:
+        # TODO: Refactor this side-effect based validation to a dedicated validation.
         file_system.get_details(body.path)
         info = file_system.move_file_or_directory(body.path, body.new_path)
         return FileMoveResponse(success=True, info=info)
@@ -250,6 +253,7 @@ async def update_file(
     app_state = AppState(request)
     body = await parse_request(request, cls=FileUpdateRequest)
     try:
+        # TODO: Refactor this side-effect based validation to a dedicated validation.
         file_system.get_details(body.path)
         info = file_system.update_file(body.path, body.contents)
 
@@ -285,6 +289,7 @@ async def open_file(
     """
     body = await parse_request(request, cls=FileOpenRequest)
     try:
+        # TODO: Refactor this side-effect based validation to a dedicated validation.
         file_system.get_details(body.path)
         success = file_system.open_in_editor(body.path, body.line_number)
         return SuccessResponse(success=success)

marimo/_server/files/os_file_system.py

@@ -188,22 +188,20 @@ def delete_file_or_directory(self, path: str) -> bool:
 
     def copy_file_or_directory(self, path: str, new_path: str) -> FileInfo:
         new_path = str(_generate_unique_path(new_path))
+        if not _is_allowed_paths(path, new_path):
+            raise ValueError(f"Cannot copy to {new_path}")
         if Path(path).is_dir():
             shutil.copytree(path, new_path)
         else:
             shutil.copy2(path, new_path)
         return self.get_details(new_path).file
 
     def move_file_or_directory(self, path: str, new_path: str) -> FileInfo:
-        file_name = os.path.basename(new_path)
-        # Disallow renaming to . or ..
-        if file_name in DISALLOWED_NAMES:
+        if not _is_allowed_paths(path, new_path):
             raise ValueError(f"Cannot rename to {new_path}")
-        # Disallow moving to an existing path or directory
-        if os.path.exists(new_path) or os.path.isdir(new_path):
-            raise ValueError(
-                f"Destination path {new_path} already exists or is a directory"
-            )
+        # Disallow moving to an existing path
+        if os.path.exists(new_path):
+            raise ValueError(f"Destination path {new_path} already exists")
         safe_move(path, new_path)
         return self.get_details(new_path).file
 
@@ -472,3 +470,13 @@ def _generate_unique_path(new_path: str | Path) -> Path:
         if not new_path.exists():
             return new_path
         i += 1
+
+
+def _is_allowed_paths(path: str | Path, new_path: str | Path) -> bool:
+    file_name = os.path.basename(new_path)
+    if file_name in DISALLOWED_NAMES or not file_name.strip():
+        return False
+
+    src = Path(path).resolve()
+    dst = Path(new_path).resolve()
+    return not (src.is_dir() and dst.is_relative_to(src))

tests/_server/files/test_os_file_system.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import sys
 from pathlib import Path
 from typing import Literal
 
@@ -8,6 +9,7 @@
 from marimo._server.files.os_file_system import (
     OSFileSystem,
     _generate_unique_path,
+    _is_allowed_paths,
 )
 from marimo._server.models.files import FileDetailsResponse
 from marimo._utils.files import natural_sort
@@ -701,3 +703,29 @@ def test_generate_unique_path_existing_path(tmp_path: Path) -> None:
     unique_path = _generate_unique_path(base_path)
     assert not unique_path.exists()
     assert unique_path.name == "file_2.txt"
+
+
+@pytest.mark.parametrize("new_path", [".", "..", ""])
+def test_is_allowed_paths_disallowed_names(new_path: str) -> None:
+    assert not _is_allowed_paths("path/to/file.txt", f"path/to/{new_path}")
+
+
+def test_is_allowed_paths_subdirectory(tmp_path: Path) -> None:
+    base_path = tmp_path / "subdir"
+    base_path.mkdir()
+    assert not _is_allowed_paths(tmp_path, base_path)
+    assert not _is_allowed_paths(base_path, base_path)
+    assert not _is_allowed_paths(base_path, base_path / "../subdir/subsubdir")
+    assert _is_allowed_paths(base_path, base_path / "../subdir2")
+
+
+@pytest.mark.skipif(
+    sys.platform == "win32",
+    reason="Symlinks on Windows require Admin privileges",
+)
+def test_is_allowed_paths_recursive_symlink(tmp_path: Path) -> None:
+    src = tmp_path / "src"
+    src.mkdir()
+    dst_link = tmp_path / "link_to_src"
+    dst_link.symlink_to(src)
+    assert not _is_allowed_paths(src, dst_link / "child")