Security: constrain local file access paths

backend/app/routes/projects.py

@@ -22,6 +22,7 @@
     create_project,
     delete_project,
     duplicate_project,
+    find_existing_child_file,
     get_project_bible,
     get_project_workspace,
     get_project_export_dir,
@@ -33,6 +34,10 @@
     save_project_workspace,
     update_project,
 )
+from backend.app.services.path_authorization_service import (
+    approve_desktop_path,
+    require_approved_desktop_path,
+)
 from backend.app.services.project_portability_service import (
     create_project_backup,
     migrate_legacy_project,
@@ -95,7 +100,9 @@ def get_project_storage() -> ApiResponse:
 def choose_project_folder() -> ApiResponse:
     try:
         path = _desktop_dialog("folder")
-    except RuntimeError as exc:
+        if path:
+            path = approve_desktop_path(path, "folder")
+    except (FileNotFoundError, RuntimeError, ValueError) as exc:
         raise HTTPException(status_code=409, detail=str(exc)) from exc
     return ApiResponse(ok=bool(path), message="Folder selected." if path else "Folder selection cancelled.", data={"path": path})
 
@@ -104,7 +111,9 @@ def choose_project_folder() -> ApiResponse:
 def choose_backup_file() -> ApiResponse:
     try:
         path = _desktop_dialog("backup")
-    except RuntimeError as exc:
+        if path:
+            path = approve_desktop_path(path, "backup")
+    except (FileNotFoundError, RuntimeError, ValueError) as exc:
         raise HTTPException(status_code=409, detail=str(exc)) from exc
     return ApiResponse(ok=bool(path), message="Backup selected." if path else "Backup selection cancelled.", data={"path": path})
 
@@ -151,25 +160,32 @@ def save_desktop_export(
 @router.post("/open")
 def open_project(payload: ProjectPathRequest) -> ApiResponse:
     try:
-        project = open_project_directory(payload.path)
+        approved_path = require_approved_desktop_path(payload.path, "folder")
+        project = open_project_directory(approved_path)
     except FileNotFoundError as exc:
         raise HTTPException(status_code=404, detail=str(exc)) from exc
-    except (OSError, ValueError, FileExistsError) as exc:
+    except (OSError, ValueError, FileExistsError, PermissionError) as exc:
         raise HTTPException(status_code=409, detail=str(exc)) from exc
     return ApiResponse(ok=True, message="Project opened.", data=project)
 
 
 @router.post("/restore")
 def restore_project(payload: ProjectRestoreRequest) -> ApiResponse:
     try:
-        project = restore_project_backup(
-            payload.backup_path,
+        approved_backup = require_approved_desktop_path(payload.backup_path, "backup")
+        approved_destination = require_approved_desktop_path(
             payload.destination_parent,
+            "folder",
+            default=project_service.PROJECTS_HOME,
+        )
+        project = restore_project_backup(
+            approved_backup,
+            approved_destination,
             payload.restored_name,
         )
     except FileNotFoundError as exc:
         raise HTTPException(status_code=404, detail=str(exc)) from exc
-    except (OSError, ValueError, FileExistsError) as exc:
+    except (OSError, ValueError, FileExistsError, PermissionError) as exc:
         raise HTTPException(status_code=409, detail=str(exc)) from exc
     return ApiResponse(ok=True, message="Project restored as a new project.", data=project)
 
@@ -182,7 +198,14 @@ def get_projects() -> ApiResponse:
 @router.post("")
 def post_project(payload: ProjectCreate) -> ApiResponse:
     try:
-        project = create_project(payload)
+        approved_parent = require_approved_desktop_path(
+            payload.parent_directory,
+            "folder",
+            default=project_service.PROJECTS_HOME,
+        )
+        project = create_project(payload.model_copy(update={"parent_directory": approved_parent}))
+    except (PermissionError, ValueError) as exc:
+        raise HTTPException(status_code=409, detail=str(exc)) from exc
     except OSError as exc:
         raise HTTPException(status_code=500, detail=str(exc)) from exc
     return ApiResponse(ok=True, message="Project created.", data=project)
@@ -213,25 +236,36 @@ def post_duplicate_project(project_id: str) -> ApiResponse:
 @router.post("/{project_id}/migrate")
 def migrate_project(project_id: str, payload: ProjectMigrationRequest) -> ApiResponse:
     try:
-        result = migrate_legacy_project(project_id, payload.destination_parent)
+        approved_destination = require_approved_desktop_path(
+            payload.destination_parent,
+            "folder",
+            default=project_service.PROJECTS_HOME,
+        )
+        result = migrate_legacy_project(project_id, approved_destination)
     except FileNotFoundError as exc:
         raise HTTPException(status_code=404, detail=str(exc)) from exc
-    except (OSError, ValueError, FileExistsError) as exc:
+    except (OSError, ValueError, FileExistsError, PermissionError) as exc:
         raise HTTPException(status_code=409, detail=str(exc)) from exc
     return ApiResponse(ok=True, message="Legacy project migrated without modifying its source.", data=result)
 
 
 @router.post("/{project_id}/backup")
 def backup_project(project_id: str, payload: ProjectBackupRequest) -> ApiResponse:
     try:
+        approved_destination = require_approved_desktop_path(
+            payload.destination_directory,
+            "folder",
+            default=project_service.get_project_backup_dir(project_id),
+        )
         result = create_project_backup(
             project_id,
-            payload.destination_directory,
+            approved_destination,
             payload.include_exports,
         )
+        approve_desktop_path(result["backup_path"], "backup")
     except FileNotFoundError as exc:
         raise HTTPException(status_code=404, detail=str(exc)) from exc
-    except (OSError, ValueError) as exc:
+    except (OSError, ValueError, PermissionError) as exc:
         raise HTTPException(status_code=409, detail=str(exc)) from exc
     return ApiResponse(ok=True, message="Project backup created and verified.", data=result)
 
@@ -246,13 +280,10 @@ def get_project_file(project_id: str, category: str, filename: str):
             "references": get_project_reference_dir,
             "exports": get_project_export_dir,
         }
-        if category not in bases or Path(filename).name != filename:
+        if category not in bases:
             raise ValueError("Invalid project file path.")
         base = bases[category](project_id).resolve()
-        path = (base / filename).resolve()
-        path.relative_to(base)
-        if not path.is_file():
-            raise FileNotFoundError("Project file not found.")
+        path = find_existing_child_file(base, filename)
     except FileNotFoundError as exc:
         raise HTTPException(status_code=404, detail=str(exc)) from exc
     except ValueError as exc:

backend/app/services/image_service.py

@@ -5,6 +5,7 @@
 import mimetypes
 import re
 import sys
+import uuid
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from datetime import datetime, timezone
 from threading import BoundedSemaphore, Lock
@@ -28,6 +29,7 @@
     output_language_name,
 )
 from backend.app.services.project_service import (
+    find_existing_child_file,
     get_project_bible,
     get_project_dir,
     get_project_generated_dir,
@@ -185,8 +187,11 @@ def _generation_lock(project_id: str) -> Lock:
 def _reference_paths(project_id: str, filenames: list[str] | None) -> list[Path]:
     paths: list[Path] = []
     for filename in filenames or []:
-        path = get_project_reference_dir(project_id) / Path(filename).name
-        if path.exists() and path.is_file():
+        try:
+            path = find_existing_child_file(get_project_reference_dir(project_id), filename)
+        except (FileNotFoundError, ValueError):
+            continue
+        if path.is_file():
             paths.append(path)
     return paths
 
@@ -210,7 +215,7 @@ def _direct_reference_prompt_note(project_id: str, reference_paths: list[Path])
             ]
             if value
         )
-        identity_notes.append(f"{category}: {description}")
+        identity_notes.append(f"{category} asset {path.stem}: {description}")
     if not identity_notes:
         return ""
     return (
@@ -552,6 +557,7 @@ def _generate_shot_image_locked(
     shot = get_shot(project_id, shot_id)
     if not shot:
         raise FileNotFoundError(f"Shot not found: {shot_id}")
+    shot_id = str(shot["shot_id"])
     if is_offline_mode():
         return {
             "project_id": project_id,
@@ -678,7 +684,7 @@ def _generate_shot_image_locked(
     version_stamp = _generation_stamp()
     for size_item in generation_sizes:
         suffix = "" if size_item["id"] == "default" else f"-{size_item['id']}"
-        filename = f"{shot_id}{suffix}-{version_stamp}.png"
+        filename = f"panel-{version_stamp}-{uuid.uuid4().hex}{suffix}.png"
         output_path = output_dir / filename
         size_prompt = f"{prompt}\n\n{_target_aspect_note(size_item)}"
         provider_size = size_item.get("generation_size") or size_item["size"]
@@ -947,7 +953,7 @@ def save_reference_image(project_id: str, filename: str, content: bytes) -> dict
         raise ValueError("Reference image is empty.")
     if len(content) > MAX_REFERENCE_BYTES:
         raise ValueError("Reference image exceeds the 20MB limit.")
-    original_name = Path(filename).name
+    original_name = str(filename or "").strip()
     if not original_name:
         raise ValueError("Reference filename is missing.")
     try:
@@ -959,30 +965,29 @@ def save_reference_image(project_id: str, filename: str, content: bytes) -> dict
     extension = REFERENCE_FORMAT_EXTENSIONS.get(image_format)
     if not extension:
         raise ValueError("Reference image must be PNG, JPEG, or WebP.")
-    safe_stem = re.sub(r"[^\w.-]+", "-", Path(original_name).stem, flags=re.UNICODE).strip(".-")
-    safe_stem = safe_stem[:100] or "reference"
     output_dir = get_project_reference_dir(project_id)
     output_dir.mkdir(parents=True, exist_ok=True)
-    safe_name = f"{safe_stem}{extension}"
-    if (output_dir / safe_name).exists():
-        safe_name = f"{safe_stem}-{_generation_stamp()}{extension}"
+    safe_name = f"reference-{uuid.uuid4().hex}{extension}"
     output_path = output_dir / safe_name
     output_path.write_bytes(content)
+    display_name = Path(original_name).name[:255] or "reference"
+    _record_reference_metadata(
+        project_id,
+        [{"filename": safe_name, "name": display_name, "category": "uploaded"}],
+    )
     return {
         "filename": safe_name,
+        "original_filename": display_name,
+        "name": display_name,
         "path": to_project_relative_path(project_id, output_path),
         "url": project_file_url(project_id, "references", safe_name),
     }
 
 
 def delete_reference_image(project_id: str, filename: str) -> dict:
     require_project(project_id)
-    safe_name = Path(filename).name
-    if not safe_name or safe_name != filename:
-        raise ValueError("Invalid reference filename.")
-    path = get_project_reference_dir(project_id) / safe_name
-    if not path.exists() or not path.is_file():
-        raise FileNotFoundError(f"Reference image not found: {safe_name}")
+    path = find_existing_child_file(get_project_reference_dir(project_id), filename)
+    safe_name = path.name
     from backend.app.services.project_service import get_project_workspace, save_project_workspace
 
     workspace = get_project_workspace(project_id)
@@ -1226,9 +1231,10 @@ def analyze_reference_style(
     output_language: str = "source",
 ) -> dict:
     require_project(project_id)
-    reference_path = get_project_reference_dir(project_id) / Path(reference_filename).name
-    if not reference_path.exists() or not reference_path.is_file():
-        raise FileNotFoundError(f"Reference image not found: {reference_filename}")
+    reference_path = find_existing_child_file(
+        get_project_reference_dir(project_id),
+        reference_filename,
+    )
 
     metrics = _image_style_metrics(reference_path)
     local_analysis = _local_style_analysis_prompt(metrics)
@@ -1367,6 +1373,7 @@ def set_shot_visual_anchor(project_id: str, shot_id: str, output_language: str =
     shot = get_shot(project_id, shot_id)
     if not shot:
         raise FileNotFoundError(f"Shot not found: {shot_id}")
+    shot_id = str(shot["shot_id"])
     output_path = str(shot.get("output_path") or "")
     if not output_path:
         raise ValueError("The panel must have a locally saved selected image before it can become the visual master.")

backend/app/services/path_authorization_service.py

@@ -0,0 +1,66 @@
+import os
+from pathlib import Path
+from threading import RLock
+
+_APPROVED_PATHS: dict[tuple[str, str], Path] = {}
+_APPROVAL_LOCK = RLock()
+_VALID_KINDS = {"folder", "backup"}
+
+
+def _canonical_path(value: str | Path) -> Path:
+    return Path(os.path.realpath(os.path.expanduser(os.fspath(value))))
+
+
+def approve_desktop_path(value: str | Path, kind: str) -> str:
+    """Remember a path returned by a native desktop file dialog."""
+    if kind not in _VALID_KINDS:
+        raise ValueError("Unsupported desktop path approval type.")
+    path = _canonical_path(value)
+    if kind == "folder" and not path.is_dir():
+        raise FileNotFoundError("The selected folder no longer exists.")
+    if kind == "backup" and not path.is_file():
+        raise FileNotFoundError("The selected backup no longer exists.")
+    with _APPROVAL_LOCK:
+        _APPROVED_PATHS[(kind, os.path.normcase(os.fspath(path)))] = path
+    return str(path)
+
+
+def require_approved_desktop_path(
+    value: str,
+    kind: str,
+    *,
+    default: str | Path | None = None,
+) -> str:
+    """Resolve only paths approved by the native dialog or a controlled test root."""
+    if kind not in _VALID_KINDS:
+        raise ValueError("Unsupported desktop path approval type.")
+    if not str(value or "").strip():
+        return str(_canonical_path(default)) if default is not None else ""
+
+    requested_text = os.path.realpath(os.path.expanduser(os.fspath(value)))
+    requested = Path(requested_text)
+    key = (kind, os.path.normcase(os.fspath(requested)))
+    with _APPROVAL_LOCK:
+        approved = _APPROVED_PATHS.get(key)
+    if approved is not None:
+        return str(approved)
+
+    if default is not None:
+        approved_default = _canonical_path(default)
+        if os.path.normcase(os.fspath(requested)) == os.path.normcase(os.fspath(approved_default)):
+            return str(approved_default)
+
+    test_root_value = os.getenv("PANELOOM_DESKTOP_TEST_ROOT", "").strip()
+    if test_root_value:
+        full_path = os.path.normpath(requested_text)
+        test_root_text = os.path.normpath(os.path.realpath(os.path.expanduser(test_root_value)))
+        test_root_prefix = test_root_text.rstrip("\\/") + os.sep
+        if not full_path.startswith(test_root_prefix):
+            raise PermissionError("The test path is outside the controlled desktop test root.")
+        if kind == "backup" and not Path(full_path).is_file():
+            raise FileNotFoundError("The selected backup no longer exists.")
+        return full_path
+
+    raise PermissionError(
+        "For safety, select this path with Paneloom's Windows file or folder picker."
+    )