Merge branch 'main' into feat/memory-lifecycle-ci-guards

Author: curdriceaurora

src/file_organizer/api/utils.py

@@ -13,10 +13,14 @@
 
 
 def resolve_path(path_value: str, allowed_paths: Optional[list[str]] = None) -> Path:
-    """Expand and normalize a filesystem path."""
+    """Expand and normalize a filesystem path, then enforce allowed-root policy.
+
+    Uses ``Path.resolve()`` + ``Path.is_relative_to()`` to evaluate containment,
+    which correctly handles symlinks, ``..`` sequences, and Windows drive-qualified
+    paths — unlike a bare ``str.startswith`` or ``os.path.commonpath`` comparison.
+    """
     # Path is validated against allowed roots below.
-    resolved = Path(path_value).expanduser()  # codeql[py/path-injection]
-    resolved_str = os.path.realpath(resolved)
+    resolved = Path(path_value).expanduser().resolve()  # codeql[py/path-injection]
     if not allowed_paths:
         raise ApiError(
             status_code=403,
@@ -26,15 +30,15 @@ def resolve_path(path_value: str, allowed_paths: Optional[list[str]] = None) ->
 
     # Allowed roots are configuration-controlled.
     # codeql[py/path-injection]
-    roots = [os.path.realpath(Path(root).expanduser()) for root in allowed_paths]
+    roots = [Path(root).expanduser().resolve() for root in allowed_paths]
     if not roots:
         raise ApiError(
             status_code=403,
             error="path_not_allowed",
             message="No allowed paths configured for this API instance.",
         )
     try:
-        allowed = any(os.path.commonpath([resolved_str, root]) == root for root in roots)
+        allowed = any(resolved == root or resolved.is_relative_to(root) for root in roots)
     except ValueError:
         allowed = False
     if not allowed:
@@ -44,7 +48,7 @@ def resolve_path(path_value: str, allowed_paths: Optional[list[str]] = None) ->
             message="Path is outside allowed roots.",
         )
 
-    return Path(resolved_str)
+    return resolved
 
 
 def is_hidden(path: Path) -> bool:

src/file_organizer/interfaces/__init__.py

@@ -25,14 +25,15 @@
     BatchProcessorProtocol,
     FileProcessorProtocol,
 )
-from file_organizer.interfaces.storage import CacheProtocol, StorageProtocol
+from file_organizer.interfaces.storage import MISSING, CacheProtocol, StorageProtocol
 
 __all__ = [
     "AudioModelProtocol",
     "BatchProcessorProtocol",
     "CacheProtocol",
     "FileProcessorProtocol",
     "LearnerProtocol",
+    "MISSING",
     "PipelineStage",
     "ScorerProtocol",
     "StageContext",

src/file_organizer/interfaces/pipeline.py

@@ -8,7 +8,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass, field
-from pathlib import Path
+from pathlib import Path, PureWindowsPath
 from typing import Any, Protocol, runtime_checkable
 
 
@@ -51,8 +51,21 @@ class StageContext:
 
     @staticmethod
     def _validate_path_component(field_name: str, value: str) -> str:
-        """Reject traversal sequences and separators in a path component."""
-        if value and (".." in value or "/" in value or "\\" in value):
+        r"""Reject traversal sequences, separators, and Windows drive qualifiers.
+
+        Checks for ``..``, POSIX ``/``, Windows ``\``, and any Windows
+        drive letter or UNC anchor (``PureWindowsPath.drive`` or ``.anchor``
+        being non-empty), so that ``output_dir / value`` cannot escape the
+        intended output directory on any platform.
+        """
+        if not value:
+            return value
+        if ".." in value or "/" in value or "\\" in value:
+            raise ValueError(f"Invalid {field_name}: {value!r}")
+        # Reject Windows drive-qualified values such as "C:" or "C:docs"
+        # which have no slash but still produce an absolute PureWindowsPath.
+        win_path = PureWindowsPath(value)
+        if win_path.drive or win_path.anchor:
             raise ValueError(f"Invalid {field_name}: {value!r}")
         return value
 

src/file_organizer/interfaces/processor.py

@@ -18,6 +18,12 @@ class FileProcessorProtocol(Protocol):
 
     Implementations accept a file path and return a processed-file
     dataclass (``ProcessedFile``, ``ProcessedImage``, etc.).
+
+    The three keyword arguments below are the common subset shared by
+    every concrete processor (``TextProcessor``, ``VisionProcessor``, etc.).
+    Individual implementations may expose additional keyword-only parameters
+    (e.g. ``perform_ocr`` on ``VisionProcessor``) without violating this
+    protocol.
     """
 
     def initialize(self) -> None:
@@ -27,9 +33,24 @@ def initialize(self) -> None:
     def process_file(
         self,
         file_path: str | Path,
-        **kwargs: Any,
+        *,
+        generate_description: bool = True,
+        generate_folder: bool = True,
+        generate_filename: bool = True,
     ) -> Any:
-        """Process a single file and return a result dataclass."""
+        """Process a single file and return a result dataclass.
+
+        Args:
+            file_path: Path to the file to process.
+            generate_description: Whether to generate a natural-language
+                description of the file's content.
+            generate_folder: Whether to suggest a target folder name.
+            generate_filename: Whether to suggest a new filename.
+
+        Returns:
+            A processed-file dataclass (e.g. ``ProcessedFile``,
+            ``ProcessedImage``) containing the generated metadata.
+        """
         ...
 
 

src/file_organizer/interfaces/storage.py

@@ -7,10 +7,14 @@
 from __future__ import annotations
 
 from collections.abc import Callable
-from typing import Any, Protocol, TypeVar, runtime_checkable
+from typing import Any, Final, Protocol, TypeVar, runtime_checkable
 
 T = TypeVar("T")
 
+#: Sentinel used by :meth:`StorageProtocol.get` to distinguish a stored
+#: ``None`` value from a missing key.
+MISSING: Final[object] = object()
+
 
 @runtime_checkable
 class CacheProtocol(Protocol):
@@ -23,9 +27,14 @@ class CacheProtocol(Protocol):
     def get_or_load(
         self,
         key: str,
+        /,
         loader: Callable[[], Any],
     ) -> Any:
-        """Return cached value for *key*, or call *loader* to populate it."""
+        """Return cached value for *key*, or call *loader* to populate it.
+
+        *key* is positional-only to match the ``ModelCache`` implementation
+        and prevent callers from passing it as a keyword argument.
+        """
         ...
 
     def stats(self) -> Any:
@@ -41,8 +50,18 @@ class StorageProtocol(Protocol):
     retrieving serializable data.
     """
 
-    def get(self, key: str) -> Any | None:
-        """Retrieve the value for *key*, or ``None`` if absent."""
+    def get(self, key: str, default: Any = MISSING) -> Any:
+        """Retrieve the value for *key*.
+
+        Returns *default* if *key* is absent.  When *default* is omitted the
+        module-level :data:`MISSING` sentinel is returned on a miss, allowing
+        callers to distinguish a stored ``None`` from an absent key::
+
+            value = storage.get("my_key")
+            if value is MISSING:
+                # key not present
+                ...
+        """
         ...
 
     def put(self, key: str, value: Any) -> None:

src/file_organizer/methodologies/para/ai/file_mover.py

@@ -419,8 +419,10 @@ def _is_already_organized(
         try:
             file_resolved = file_path.resolve()
             expected_resolved = expected_parent.resolve()
-            return str(file_resolved).startswith(str(expected_resolved))
-        except OSError:
+            return file_resolved == expected_resolved or file_resolved.is_relative_to(
+                expected_resolved
+            )
+        except (OSError, ValueError):
             return False
 
     def _resolve_collision(self, destination: Path) -> Path:

src/file_organizer/models/registry.py

@@ -9,6 +9,7 @@
 
 from __future__ import annotations
 
+import copy
 from dataclasses import dataclass
 
 
@@ -34,24 +35,33 @@ class ModelInfo:
 
 
 def get_text_models() -> list[ModelInfo]:
-    """Return all registered text models with domain metadata."""
+    """Return a deep copy of all registered text models with domain metadata.
+
+    Returns a copy so that callers cannot mutate the shared registry state.
+    """
     from file_organizer.models.text_registry import TEXT_MODELS
 
-    return list(TEXT_MODELS)
+    return copy.deepcopy(list(TEXT_MODELS))
 
 
 def get_vision_models() -> list[ModelInfo]:
-    """Return all registered vision models with domain metadata."""
+    """Return a deep copy of all registered vision models with domain metadata.
+
+    Returns a copy so that callers cannot mutate the shared registry state.
+    """
     from file_organizer.models.vision_registry import VISION_MODELS
 
-    return list(VISION_MODELS)
+    return copy.deepcopy(list(VISION_MODELS))
 
 
 def get_audio_models() -> list[ModelInfo]:
-    """Return all registered audio models with domain metadata."""
+    """Return a deep copy of all registered audio models with domain metadata.
+
+    Returns a copy so that callers cannot mutate the shared registry state.
+    """
     from file_organizer.models.audio_registry import AUDIO_MODELS
 
-    return list(AUDIO_MODELS)
+    return copy.deepcopy(list(AUDIO_MODELS))
 
 
 def get_all_models() -> list[ModelInfo]:

tests/api/test_utils.py

@@ -64,6 +64,70 @@ def test_file_inside_allowed_root(self, tmp_path: Path) -> None:
         result = resolve_path(str(f), allowed_paths=[str(tmp_path)])
         assert result == f.resolve()
 
+    def test_path_prefix_attack_blocked(self, tmp_path: Path) -> None:
+        # Regression for #672: str.startswith() would allow "/allowed_dir_suffix"
+        # to pass when "/allowed_dir" is the allowed root.
+        # Path.is_relative_to() correctly rejects this because the directory
+        # boundary is respected (it's not a child path).
+        allowed = tmp_path / "data"
+        allowed.mkdir()
+        attack = tmp_path / "data_extra"
+        attack.mkdir()
+        with pytest.raises(ApiError) as exc_info:
+            resolve_path(str(attack), allowed_paths=[str(allowed)])
+        assert exc_info.value.status_code == 403
+        assert exc_info.value.error == "path_not_allowed"
+
+    def test_dotdot_traversal_via_subdirectory_blocked(self, tmp_path: Path) -> None:
+        # Ensure that a path with embedded ".." that would escape the root is blocked.
+        allowed = tmp_path / "allowed"
+        allowed.mkdir()
+        outside = tmp_path / "outside"
+        outside.mkdir()
+        # Construct a path that goes through the allowed root and back out
+        traversal = str(allowed) + "/../outside"
+        with pytest.raises(ApiError) as exc_info:
+            resolve_path(traversal, allowed_paths=[str(allowed)])
+        assert exc_info.value.status_code == 403
+
+    def test_symlink_escaping_allowed_root_blocked(self, tmp_path: Path) -> None:
+        # A symlink inside the allowed root that points outside must be blocked.
+        allowed = tmp_path / "allowed"
+        allowed.mkdir()
+        outside = tmp_path / "secret"
+        outside.mkdir()
+        # Create a symlink inside allowed that points to outside
+        link = allowed / "escape_link"
+        link.symlink_to(outside)
+        with pytest.raises(ApiError) as exc_info:
+            resolve_path(str(link), allowed_paths=[str(allowed)])
+        assert exc_info.value.status_code == 403
+
+    def test_windows_drive_qualified_path_blocked(self, tmp_path: Path) -> None:
+        # Drive-qualified input must never be accepted when it is outside the
+        # configured allowlist root (Windows and non-Windows behavior alike).
+        allowed = tmp_path / "allowed"
+        allowed.mkdir()
+        with pytest.raises(ApiError) as exc_info:
+            resolve_path(r"C:\Windows\System32", allowed_paths=[str(allowed)])
+        assert exc_info.value.status_code == 403
+        assert exc_info.value.error == "path_not_allowed"
+
+    def test_unc_path_blocked_when_outside_allowlist(self, tmp_path: Path) -> None:
+        # UNC-style paths are treated as out-of-scope unless explicitly under an
+        # allowed root; this guards Windows network-share escape cases.
+        allowed = tmp_path / "allowed"
+        allowed.mkdir()
+        with pytest.raises(ApiError) as exc_info:
+            resolve_path(r"\\server\share\secret.txt", allowed_paths=[str(allowed)])
+        assert exc_info.value.status_code == 403
+        assert exc_info.value.error == "path_not_allowed"
+
+    def test_returns_path_object(self, tmp_path: Path) -> None:
+        result = resolve_path(str(tmp_path), allowed_paths=[str(tmp_path)])
+        assert isinstance(result, Path)
+        assert result.is_absolute()
+
 
 # ---------------------------------------------------------------------------
 # is_hidden

tests/ci/test_path_security_contract.py

@@ -25,11 +25,11 @@
         "path = Path(config_path).expanduser()",
     },
     "api/utils.py": {
-        "resolved = Path(path_value).expanduser()",
-        "roots = [os.path.realpath(Path(root).expanduser()) for root in allowed_paths]",
-        "return Path(resolved_str)",
+        "resolved = Path(path_value).expanduser().resolve()",
+        "roots = [Path(root).expanduser().resolve() for root in allowed_paths]",
+        "return resolved",
         # Multi-line variants (formatter may break long lines)
-        "os.path.realpath(Path(root).expanduser())",
+        "Path(root).expanduser().resolve() for root in allowed_paths",
     },
     "api/routers/system.py": {
         "file_info_from_path(Path(info.path))",
@@ -74,8 +74,8 @@
         "path = Path(config_path).expanduser()",
     },
     "api/utils.py": {
-        "resolved = Path(path_value).expanduser()",
-        "roots = [os.path.realpath(Path(root).expanduser()) for root in allowed_paths]",
+        "resolved = Path(path_value).expanduser().resolve()",
+        "roots = [Path(root).expanduser().resolve() for root in allowed_paths]",
     },
 }
 

tests/interfaces/test_protocol_conformance.py

@@ -8,11 +8,13 @@
 
 from __future__ import annotations
 
+import inspect
 from unittest.mock import MagicMock, patch
 
 import pytest
 
 from file_organizer.interfaces import (
+    MISSING,
     AudioModelProtocol,
     BatchProcessorProtocol,
     CacheProtocol,
@@ -229,8 +231,8 @@ class TestStorageProtocolConformance:
 
     def test_stub_satisfies_storage_protocol(self) -> None:
         class _StubStorage:
-            def get(self, key: str) -> object | None:
-                return None
+            def get(self, key: str, default: object = MISSING) -> object:
+                return default
 
             def put(self, key: str, value: object) -> None:
                 pass
@@ -241,6 +243,12 @@ def delete(self, key: str) -> bool:
             def exists(self, key: str) -> bool:
                 return False
 
+        protocol_params = list(inspect.signature(StorageProtocol.get).parameters.values())
+        stub_params = list(inspect.signature(_StubStorage.get).parameters.values())
+        assert len(stub_params) == len(protocol_params)
+        assert [param.name for param in stub_params] == [param.name for param in protocol_params]
+        assert [param.kind for param in stub_params] == [param.kind for param in protocol_params]
+        assert stub_params[-1].default is MISSING
         assert isinstance(_StubStorage(), StorageProtocol)