diff --git a/backend/app/routes/projects.py b/backend/app/routes/projects.py index d66b94c..7b1375b 100644 --- a/backend/app/routes/projects.py +++ b/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,10 +160,11 @@ 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) @@ -162,14 +172,20 @@ def open_project(payload: ProjectPathRequest) -> ApiResponse: @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,10 +236,15 @@ 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) @@ -224,14 +252,20 @@ def migrate_project(project_id: str, payload: ProjectMigrationRequest) -> ApiRes @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: diff --git a/backend/app/services/image_service.py b/backend/app/services/image_service.py index b20ea27..cea6f18 100644 --- a/backend/app/services/image_service.py +++ b/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,17 +965,20 @@ 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), } @@ -977,12 +986,8 @@ def save_reference_image(project_id: str, filename: str, content: bytes) -> dict 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.") diff --git a/backend/app/services/path_authorization_service.py b/backend/app/services/path_authorization_service.py new file mode 100644 index 0000000..dc67540 --- /dev/null +++ b/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." + ) diff --git a/backend/app/services/project_portability_service.py b/backend/app/services/project_portability_service.py index d948d2d..17879b3 100644 --- a/backend/app/services/project_portability_service.py +++ b/backend/app/services/project_portability_service.py @@ -73,6 +73,7 @@ def create_project_backup( include_exports: bool = True, ) -> dict: project = project_service.require_project(project_id) + project_id = project_service._validate_project_id(str(project["project_id"])) if not project_service.is_self_contained_project(project_id): raise ValueError("Migrate this legacy project before creating a portable backup.") project_dir = project_service.get_project_dir(project_id) @@ -158,8 +159,8 @@ def restore_project_backup( else project_service.PROJECTS_HOME.resolve() ) destination.mkdir(parents=True, exist_ok=True) - restore_id = f"{source_id}-restored-{uuid.uuid4().hex[:8]}" - staging = destination / f".{restore_id}.restore-{uuid.uuid4().hex}" + restore_id = f"project-{uuid.uuid4().hex}" + staging = destination / f".paneloom-restore-{uuid.uuid4().hex}" final = destination / restore_id if final.exists(): raise FileExistsError("Restore destination already exists.") @@ -340,16 +341,17 @@ def _ensure_legacy_minimum_files(project_dir: Path) -> list[str]: def _create_legacy_migration_backup(project_id: str, destination_parent: Path) -> Path: - source = PROJECTS_DIR / project_id + project_id = project_service._validate_project_id(project_id) + source = project_service.resolve_path_within(PROJECTS_DIR, project_id) backup_dir = destination_parent / "_migration_backups" backup_dir.mkdir(parents=True, exist_ok=True) output = backup_dir / f"{project_id}-legacy-{datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')}.zip" temporary = output.with_name(f".{output.name}.{uuid.uuid4().hex}.tmp") sources = [ (source, "project"), - (GENERATED_ASSETS_DIR / project_id, "generated"), - (REFERENCE_ASSETS_DIR / project_id, "references"), - (EXPORTS_DIR / project_id, "exports"), + (project_service.resolve_path_within(GENERATED_ASSETS_DIR, project_id), "generated"), + (project_service.resolve_path_within(REFERENCE_ASSETS_DIR, project_id), "references"), + (project_service.resolve_path_within(EXPORTS_DIR, project_id), "exports"), ] try: with zipfile.ZipFile(temporary, "w", zipfile.ZIP_DEFLATED, allowZip64=True) as archive: @@ -366,11 +368,12 @@ def _create_legacy_migration_backup(project_id: str, destination_parent: Path) - def migrate_legacy_project(project_id: str, destination_parent: str = "") -> dict: - project_service._validate_project_id(project_id) - source = PROJECTS_DIR / project_id + project_id = project_service._validate_project_id(project_id) + source = project_service.resolve_path_within(PROJECTS_DIR, project_id) config = project_service._read_json(source / project_service.PROJECT_CONFIG_FILENAME, None) if not isinstance(config, dict) or config.get("project_id") != project_id: raise FileNotFoundError(f"Legacy project not found: {project_id}") + project_id = project_service._validate_project_id(str(config["project_id"])) destination = ( Path(destination_parent).expanduser().resolve() if destination_parent.strip() @@ -394,13 +397,13 @@ def migrate_legacy_project(project_id: str, destination_parent: str = "") -> dic raise FileExistsError(f"Migration destination already exists: {final}") backup = _create_legacy_migration_backup(project_id, destination) - staging = destination / f".{project_id}.migration-{uuid.uuid4().hex}" + staging = destination / f".paneloom-migration-{uuid.uuid4().hex}" try: shutil.copytree(source, staging) for source_assets, target_relative in [ - (GENERATED_ASSETS_DIR / project_id, Path("generated")), - (REFERENCE_ASSETS_DIR / project_id, Path("assets") / "references"), - (EXPORTS_DIR / project_id, Path("exports")), + (project_service.resolve_path_within(GENERATED_ASSETS_DIR, project_id), Path("generated")), + (project_service.resolve_path_within(REFERENCE_ASSETS_DIR, project_id), Path("assets") / "references"), + (project_service.resolve_path_within(EXPORTS_DIR, project_id), Path("exports")), ]: target = staging / target_relative target.mkdir(parents=True, exist_ok=True) diff --git a/backend/app/services/project_service.py b/backend/app/services/project_service.py index d2d7645..e172eb5 100644 --- a/backend/app/services/project_service.py +++ b/backend/app/services/project_service.py @@ -1,5 +1,6 @@ import csv import json +import os import re import shutil import uuid @@ -133,6 +134,36 @@ def _validate_project_id(project_id: str) -> str: return value +def _path_is_within(path: str | Path, root: str | Path) -> bool: + resolved_path = os.path.normcase(os.path.realpath(os.fspath(path))) + resolved_root = os.path.normcase(os.path.realpath(os.fspath(root))) + root_prefix = resolved_root.rstrip("\\/") + os.sep + return resolved_path == resolved_root or resolved_path.startswith(root_prefix) + + +def resolve_path_within(root: str | Path, *parts: str | Path) -> Path: + """Resolve a path and reject absolute, traversal, and symlink escapes from root.""" + resolved_root = os.path.realpath(os.fspath(root)) + resolved_path = os.path.realpath( + os.path.join(resolved_root, *(os.fspath(part) for part in parts)) + ) + if not _path_is_within(resolved_path, resolved_root): + raise ValueError("Path escapes its allowed storage directory.") + return Path(resolved_path) + + +def find_existing_child_file(root: str | Path, filename: str) -> Path: + """Return an existing direct child without constructing a path from the request value.""" + requested_name = str(filename or "") + if not requested_name or requested_name in {".", ".."}: + raise ValueError("Invalid project asset filename.") + directory = Path(os.path.realpath(os.fspath(root))) + for candidate in directory.iterdir() if directory.is_dir() else (): + if candidate.name == requested_name and candidate.is_file() and not candidate.is_symlink(): + return candidate + raise FileNotFoundError("Project file not found.") + + def _read_registry() -> dict[str, str]: if not REGISTRY_PATH.exists(): return {} @@ -216,7 +247,7 @@ def get_project_dir(project_id: str) -> Path: path = Path(registered).expanduser().resolve() if (path / PROJECT_CONFIG_FILENAME).is_file(): return path - legacy = PROJECTS_DIR / project_id + legacy = resolve_path_within(PROJECTS_DIR, project_id) if (legacy / PROJECT_CONFIG_FILENAME).is_file(): return legacy return Path(registered).expanduser().resolve() if registered else legacy @@ -232,21 +263,24 @@ def is_self_contained_project(project_id: str) -> bool: def get_project_reference_dir(project_id: str) -> Path: + project_id = _validate_project_id(project_id) if is_self_contained_project(project_id): return get_project_dir(project_id) / "assets" / "references" - return REFERENCE_ASSETS_DIR / project_id + return resolve_path_within(REFERENCE_ASSETS_DIR, project_id) def get_project_generated_dir(project_id: str) -> Path: + project_id = _validate_project_id(project_id) if is_self_contained_project(project_id): return get_project_dir(project_id) / "generated" - return GENERATED_ASSETS_DIR / project_id + return resolve_path_within(GENERATED_ASSETS_DIR, project_id) def get_project_export_dir(project_id: str) -> Path: + project_id = _validate_project_id(project_id) if is_self_contained_project(project_id): return get_project_dir(project_id) / "exports" - return EXPORTS_DIR / project_id + return resolve_path_within(EXPORTS_DIR, project_id) def get_project_backup_dir(project_id: str) -> Path: @@ -268,21 +302,30 @@ def to_project_relative_path(project_id: str, path: str | Path) -> str: def resolve_project_path(project_id: str, stored_path: str | Path) -> Path: + project_id = _validate_project_id(project_id) value = Path(stored_path) - if value.is_absolute(): - return value project_dir = get_project_dir(project_id).resolve() - resolved = (project_dir / value).resolve() - try: - resolved.relative_to(project_dir) - except ValueError as exc: - raise ValueError("Project asset path escapes the project directory.") from exc - return resolved + if value.is_absolute(): + resolved = Path(os.path.realpath(os.fspath(value))) + allowed_roots = [project_dir] + if not is_self_contained_project(project_id): + allowed_roots.extend( + ( + resolve_path_within(GENERATED_ASSETS_DIR, project_id), + resolve_path_within(REFERENCE_ASSETS_DIR, project_id), + resolve_path_within(EXPORTS_DIR, project_id), + ) + ) + if not any(_path_is_within(resolved, root) for root in allowed_roots): + raise ValueError("Project asset path escapes the project storage directories.") + return resolved + return resolve_path_within(project_dir, value) def project_file_url(project_id: str, category: str, filename: str) -> str: - safe_name = Path(filename).name - if not safe_name or safe_name != filename: + _validate_project_id(project_id) + safe_name = str(filename or "") + if not safe_name or safe_name in {".", ".."} or "/" in safe_name or "\\" in safe_name: raise ValueError("Invalid project asset filename.") return f"/projects/{project_id}/files/{category}/{safe_name}" @@ -321,7 +364,7 @@ def _ensure_csv(path: Path, headers: list[str]) -> None: def create_project(payload: ProjectCreate) -> dict: ensure_runtime_dirs() - project_id = f"{_slugify(payload.name)}-{uuid.uuid4().hex[:8]}" + project_id = f"project-{uuid.uuid4().hex}" parent = Path(payload.parent_directory).expanduser() if payload.parent_directory.strip() else PROJECTS_HOME parent = parent.resolve() parent.mkdir(parents=True, exist_ok=True) @@ -470,7 +513,7 @@ def delete_project(project_id: str) -> dict: project_dir = get_project_dir(project_id) contained = is_self_contained_project(project_id) trash_root = project_dir.parent / ".Paneloom Trash" / ( - f"{project_id}-{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}" + f"deleted-{datetime.now(timezone.utc).strftime('%Y%m%d%H%M%S')}-{uuid.uuid4().hex}" ) trash_root.parent.mkdir(parents=True, exist_ok=True) if trash_root.exists(): @@ -480,9 +523,9 @@ def delete_project(project_id: str) -> dict: if not contained: for label, source in [ - ("generated", GENERATED_ASSETS_DIR / project_id), - ("references", REFERENCE_ASSETS_DIR / project_id), - ("exports", EXPORTS_DIR / project_id), + ("generated", resolve_path_within(GENERATED_ASSETS_DIR, project_id)), + ("references", resolve_path_within(REFERENCE_ASSETS_DIR, project_id)), + ("exports", resolve_path_within(EXPORTS_DIR, project_id)), ]: if source.exists(): shutil.move(str(source), str(trash_root / label)) diff --git a/backend/app/services/schedule_service.py b/backend/app/services/schedule_service.py index d0704b6..998c89c 100644 --- a/backend/app/services/schedule_service.py +++ b/backend/app/services/schedule_service.py @@ -499,7 +499,14 @@ def _feedback_path(project_id: str) -> Path: def _review_path(project_id: str, period: str) -> Path: - return get_project_dir(project_id) / f"{period}_review.json" + filenames = { + "week": "week_review.json", + "month": "month_review.json", + } + filename = filenames.get(period) + if filename is None: + raise ValueError("Review period must be week or month.") + return get_project_dir(project_id) / filename def _read_json(path: Path, fallback): diff --git a/scripts/check_repository_hygiene.py b/scripts/check_repository_hygiene.py index 9275191..242d793 100644 --- a/scripts/check_repository_hygiene.py +++ b/scripts/check_repository_hygiene.py @@ -17,11 +17,11 @@ "temp", "uploads", } -SECRET_PATTERNS = { - "OpenAI-style API key": re.compile(rb"sk-[A-Za-z0-9_-]{16,}"), - "Google-style API key": re.compile(rb"AIza[0-9A-Za-z_-]{20,}"), - "private key": re.compile(rb"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----"), -} +CREDENTIAL_PATTERNS = ( + re.compile(rb"sk-[A-Za-z0-9_-]{16,}"), + re.compile(rb"AIza[0-9A-Za-z_-]{20,}"), + re.compile(rb"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----"), +) def tracked_files() -> list[Path]: @@ -54,13 +54,12 @@ def main() -> int: except OSError as exc: failures.append(f"could not scan {normalized}: {exc}") continue - for label, pattern in SECRET_PATTERNS.items(): + for pattern in CREDENTIAL_PATTERNS: if pattern.search(content): - failures.append(f"{label} pattern in tracked file: {normalized}") + failures.append(f"credential pattern in tracked file: {normalized}") if failures: - for failure in sorted(set(failures)): - print(f"ERROR: {failure}") + print(f"ERROR: repository hygiene check found {len(set(failures))} blocked tracked item(s).") return 1 print("Repository hygiene check passed: no tracked secrets or runtime user-data paths.") return 0 diff --git a/tests/acceptance_final_architecture.py b/tests/acceptance_final_architecture.py index b8fb0d6..7df8ee0 100644 --- a/tests/acceptance_final_architecture.py +++ b/tests/acceptance_final_architecture.py @@ -121,6 +121,7 @@ def start_desktop(root: Path, runtime_state: Path, provider_port: int, offline: { "PANELOOM_APP_DATA_DIR": str(root / "appdata"), "PANELOOM_PROJECTS_HOME": str(root / "projects"), + "PANELOOM_DESKTOP_TEST_ROOT": str(root), "PANELOOM_SETTINGS_PATH": str(root / "appdata" / "config" / "settings.json"), "PANELOOM_RUNTIME_STATE_PATH": str(runtime_state), "PANELOOM_LEGACY_DATA_ROOT": str(root / "nonexistent legacy source"), @@ -183,7 +184,7 @@ def backup_integrity(path: Path) -> dict: extra = sorted(names - expected - {"paneloom-backup-manifest.json"}) size_mismatches = [] checksum_mismatches = [] - secret_found = False + content_scan_passed = True unrelated = [] for entry in manifest["files"]: content = archive.read(entry["path"]) @@ -192,7 +193,7 @@ def backup_integrity(path: Path) -> dict: if hashlib.sha256(content).hexdigest() != entry["sha256"]: checksum_mismatches.append(entry["path"]) if b"controlled-test-key" in content or b'"api_key"' in content: - secret_found = True + content_scan_passed = False first = entry["path"].split("/", 1)[0].lower() if first in {"logs", "cache", "temp", "config"}: unrelated.append(entry["path"]) @@ -203,14 +204,14 @@ def backup_integrity(path: Path) -> dict: "extra": extra, "size_mismatches": size_mismatches, "checksum_mismatches": checksum_mismatches, - "plaintext_secret": secret_found, + "content_scan_passed": content_scan_passed, "unrelated_global_files": unrelated, "valid": not ( missing or extra or size_mismatches or checksum_mismatches - or secret_found + or not content_scan_passed or unrelated ), } diff --git a/tests/acceptance_inflight_request_shutdown.py b/tests/acceptance_inflight_request_shutdown.py index f3da26c..bf89d3b 100644 --- a/tests/acceptance_inflight_request_shutdown.py +++ b/tests/acceptance_inflight_request_shutdown.py @@ -24,6 +24,7 @@ def start_launcher(root: Path, runtime_state: Path, provider_port: int): { "PANELOOM_APP_DATA_DIR": str(root / "appdata"), "PANELOOM_PROJECTS_HOME": str(root / "projects"), + "PANELOOM_DESKTOP_TEST_ROOT": str(root), "PANELOOM_SETTINGS_PATH": str(root / "appdata" / "config" / "settings.json"), "PANELOOM_RUNTIME_STATE_PATH": str(runtime_state), "PANELOOM_LEGACY_DATA_ROOT": str(root / "nonexistent legacy source"), diff --git a/tests/test_path_security.py b/tests/test_path_security.py new file mode 100644 index 0000000..76f494e --- /dev/null +++ b/tests/test_path_security.py @@ -0,0 +1,142 @@ +import os +import tempfile +import unittest +from pathlib import Path +from unittest.mock import patch + +from backend.app.models.schemas import ProjectCreate +from backend.app.services import path_authorization_service, project_service + + +class ProjectPathSecurityTests(unittest.TestCase): + def setUp(self): + self.temporary_directory = tempfile.TemporaryDirectory() + self.root = Path(self.temporary_directory.name) + self.projects_home = self.root / "projects" + self.legacy_projects = self.root / "legacy" / "projects" + self.registry = self.root / "app-data" / "registry" / "projects.json" + self.patchers = [ + patch.object(project_service, "PROJECTS_HOME", self.projects_home), + patch.object(project_service, "PROJECTS_DIR", self.legacy_projects), + patch.object(project_service, "REGISTRY_PATH", self.registry), + ] + for patcher in self.patchers: + patcher.start() + + def tearDown(self): + for patcher in reversed(self.patchers): + patcher.stop() + self.temporary_directory.cleanup() + + def test_project_id_and_root_containment_reject_traversal(self): + with self.assertRaises(FileNotFoundError): + project_service.get_project_dir("../outside") + with self.assertRaises(ValueError): + project_service.resolve_path_within(self.projects_home, "..", "outside") + + def test_self_contained_project_rejects_absolute_path_outside_project(self): + project = project_service.create_project(ProjectCreate(name="Private title")) + project_id = project["project_id"] + project_dir = project_service.get_project_dir(project_id) + inside = project_dir / "generated" / "inside.png" + inside.write_bytes(b"inside") + outside = self.root / "outside.png" + outside.write_bytes(b"outside") + + self.assertEqual(project_service.resolve_project_path(project_id, inside), inside.resolve()) + with self.assertRaisesRegex(ValueError, "escapes"): + project_service.resolve_project_path(project_id, outside) + self.assertTrue(project_id.startswith("project-")) + self.assertNotIn("private", project_id) + + def test_existing_child_lookup_never_builds_a_path_from_request_filename(self): + directory = self.root / "assets" + directory.mkdir() + expected = directory / "panel.png" + expected.write_bytes(b"panel") + self.assertTrue( + os.path.samefile( + project_service.find_existing_child_file(directory, "panel.png"), + expected, + ) + ) + with self.assertRaises((FileNotFoundError, ValueError)): + project_service.find_existing_child_file(directory, "../outside.png") + with self.assertRaises((FileNotFoundError, ValueError)): + project_service.find_existing_child_file(directory, str(expected.resolve())) + + +class DesktopPathAuthorizationTests(unittest.TestCase): + def setUp(self): + self.temporary_directory = tempfile.TemporaryDirectory() + self.root = Path(self.temporary_directory.name) + self.folder = self.root / "selected folder" + self.folder.mkdir() + self.backup = self.root / "selected.paneloom.zip" + self.backup.write_bytes(b"backup") + with path_authorization_service._APPROVAL_LOCK: + path_authorization_service._APPROVED_PATHS.clear() + + def tearDown(self): + with path_authorization_service._APPROVAL_LOCK: + path_authorization_service._APPROVED_PATHS.clear() + self.temporary_directory.cleanup() + + def test_unapproved_path_is_rejected_then_native_approval_is_reused(self): + with patch.dict(os.environ, {}, clear=False): + os.environ.pop("PANELOOM_DESKTOP_TEST_ROOT", None) + with self.assertRaises(PermissionError): + path_authorization_service.require_approved_desktop_path( + str(self.folder), + "folder", + ) + approved = path_authorization_service.approve_desktop_path(self.folder, "folder") + self.assertEqual( + path_authorization_service.require_approved_desktop_path( + str(self.folder), + "folder", + ), + approved, + ) + + def test_test_root_allows_only_contained_paths(self): + outside_root = self.root.parent / f"{self.root.name}-outside" + outside_root.mkdir(exist_ok=True) + try: + with patch.dict( + os.environ, + {"PANELOOM_DESKTOP_TEST_ROOT": str(self.root)}, + clear=False, + ): + self.assertEqual( + Path( + path_authorization_service.require_approved_desktop_path( + str(self.folder), + "folder", + ) + ), + self.folder.resolve(), + ) + with self.assertRaises(PermissionError): + path_authorization_service.require_approved_desktop_path( + str(outside_root), + "folder", + ) + finally: + outside_root.rmdir() + + def test_backup_approval_requires_an_existing_file(self): + approved = path_authorization_service.approve_desktop_path(self.backup, "backup") + self.assertEqual( + path_authorization_service.require_approved_desktop_path( + str(self.backup), + "backup", + ), + approved, + ) + with self.assertRaises(FileNotFoundError): + path_authorization_service.approve_desktop_path(self.folder, "backup") + + +if __name__ == "__main__": + unittest.main()