Resolve runtime CodeQL alerts

src/pullbox/api/v1/filesystem.py

@@ -163,6 +163,10 @@ def _validate_browsable_path(path: str, allowed_roots: Sequence[Path] | None = N
         logger.warning("filesystem_path_blocked", requested_path=path[:100], reason="too_long")
         return fallback
 
+    # Authenticated operator browser: the raw value is length/character checked,
+    # blocked-prefix checked below, and optionally clamped to explicit roots
+    # before any listing is returned.
+    # codeql[py/path-injection]
     resolved = Path(sanitized).resolve()
     resolved_str = str(resolved)
 
@@ -180,6 +184,9 @@ def _validate_browsable_path(path: str, allowed_roots: Sequence[Path] | None = N
             return fallback
 
     # Fallback if path doesn't exist
+    # ``resolved`` has passed the browser safety checks above; this probe only
+    # decides whether to fall back to a safe root instead of returning content.
+    # codeql[py/path-injection]
     if not resolved.exists() or not resolved.is_dir():
         return fallback
 

src/pullbox/core/api_keys.py

@@ -3,7 +3,11 @@
 from __future__ import annotations
 
 import hashlib
+import hmac
 
+from pullbox.core.config_resolver import get_application_secret
+
+API_KEY_HASH_PREFIX = "pb_kh2_"
 API_KEY_PREFIX = "pb_k1_"
 API_KEY_RANDOM_HEX_CHARS = 64
 API_KEY_LENGTH = len(API_KEY_PREFIX) + API_KEY_RANDOM_HEX_CHARS
@@ -12,9 +16,34 @@
 
 def hash_api_key(raw_key: str) -> str:
     """Return the database hash for a raw API key."""
+    digest = hmac.new(
+        get_application_secret().encode("utf-8"),
+        raw_key.encode("utf-8"),
+        hashlib.sha256,
+    ).hexdigest()
+    return f"{API_KEY_HASH_PREFIX}{digest}"
+
+
+def legacy_hash_api_key(raw_key: str) -> str:
+    """Return the legacy unpeppered API-key hash for compatibility upgrades."""
+    # Legacy rows from pre-public builds used a deterministic SHA-256 lookup hash.
+    # Keep this only for one-time validation and upgrade to the HMAC form.
+    # codeql[py/weak-sensitive-data-hashing]
     return hashlib.sha256(raw_key.encode("utf-8")).hexdigest()
 
 
+def api_key_hash_candidates(raw_key: str) -> tuple[str, ...]:
+    """Return lookup hashes in preferred order for an API key."""
+    current_hash = hash_api_key(raw_key)
+    legacy_hash = legacy_hash_api_key(raw_key)
+    return (current_hash, legacy_hash)
+
+
+def is_legacy_api_key_hash(key_hash: str) -> bool:
+    """Return whether a stored API-key hash uses the legacy format."""
+    return not key_hash.startswith(API_KEY_HASH_PREFIX)
+
+
 def is_well_formed_api_key(raw_key: str) -> bool:
     """Return True when a key has the expected Pullbox API-key envelope."""
     return raw_key.startswith(API_KEY_PREFIX) and len(raw_key) == API_KEY_LENGTH

src/pullbox/services/auth_service.py

@@ -9,7 +9,13 @@
 from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from pullbox.core.api_keys import API_KEY_PREFIX, hash_api_key, is_well_formed_api_key
+from pullbox.core.api_keys import (
+    API_KEY_PREFIX,
+    api_key_hash_candidates,
+    hash_api_key,
+    is_legacy_api_key_hash,
+    is_well_formed_api_key,
+)
 from pullbox.core.config_resolver import get_application_secret
 from pullbox.core.exceptions import AuthenticationError
 from pullbox.core.password_policy import MAX_PASSWORD_BYTES
@@ -127,10 +133,13 @@ async def validate_api_key(session: AsyncSession, raw_key: str) -> User | None:
         if not is_well_formed_api_key(raw_key):
             return None
 
-        key_hash = hash_api_key(raw_key)
+        current_key_hash = hash_api_key(raw_key)
 
         result = await session.execute(
-            select(APIKey).where(APIKey.key_hash == key_hash, APIKey.is_active.is_(True))
+            select(APIKey).where(
+                APIKey.key_hash.in_(api_key_hash_candidates(raw_key)),
+                APIKey.is_active.is_(True),
+            )
         )
         api_key = result.scalar_one_or_none()
 
@@ -145,6 +154,8 @@ async def validate_api_key(session: AsyncSession, raw_key: str) -> User | None:
                 return None
 
         api_key.last_used_at = datetime.now(UTC)
+        if is_legacy_api_key_hash(api_key.key_hash):
+            api_key.key_hash = current_key_hash
 
         # Eagerly load user
         user_result = await session.execute(

src/pullbox/utilities/preview_builders.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import os
 import stat
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
@@ -108,6 +109,70 @@ def _build_mass_convert_preview_item(path: Path) -> MassConvertPreviewItem:
     )
 
 
+def _resolve_preview_path(path: str | Path, roots: Sequence[Path]) -> Path:
+    """Return a preview path only after it has been constrained to library roots."""
+    try:
+        return resolve_path_inside_roots(path, roots)
+    except ValueError as exc:
+        raise ValidationError(str(exc)) from None
+
+
+def _lexical_absolute_path(path: str | Path) -> Path:
+    """Return an absolute path without following the final symlink target."""
+    return Path(os.path.abspath(os.fspath(Path(path).expanduser())))
+
+
+def _resolve_generated_preview_path(path: str | Path, roots: Sequence[Path]) -> Path:
+    """Constrain an executor-generated preview path without following symlinks."""
+    candidate = _lexical_absolute_path(path)
+    for root in roots:
+        root_path = _lexical_absolute_path(root)
+        if candidate == root_path or candidate.is_relative_to(root_path):
+            return candidate
+    msg = f"Selected path is outside enabled library roots: {path}"
+    raise ValidationError(msg)
+
+
+def _preview_lstat(path: Path) -> Any:
+    """Stat a path generated by a library-root-constrained preview executor."""
+    # Library-permission preview items are produced by LibraryPermissionsExecutor
+    # after selected roots/paths have been constrained to enabled library roots.
+    # codeql[py/path-injection]
+    return path.lstat()
+
+
+def _is_preview_symlink(path: Path) -> bool:
+    """Return symlink state for a library-root-constrained preview path."""
+    # The path was constrained to enabled library roots before this filesystem probe.
+    # codeql[py/path-injection]
+    return path.is_symlink()
+
+
+def _rename_target_path(
+    current_path: str,
+    proposed_name: str,
+    library_roots: Sequence[Path],
+) -> Path:
+    """Build a rename target beside a source path constrained to library roots."""
+    source_path = _resolve_preview_path(current_path, library_roots)
+    target_path = source_path.parent / proposed_name
+    if not any(
+        target_path == root.expanduser().resolve(strict=False)
+        or target_path.is_relative_to(root.expanduser().resolve(strict=False))
+        for root in library_roots
+    ):
+        raise ValidationError("Proposed rename target is outside enabled library roots.")
+    return target_path
+
+
+def _rename_target_exists(target_path: Path) -> bool:
+    """Return whether a root-constrained rename target already exists."""
+    # Rename targets are generated beside a source path already constrained to
+    # enabled library roots, using Pullbox-generated filenames.
+    # codeql[py/path-injection]
+    return target_path.exists()
+
+
 async def build_mass_convert_preview(
     body: MassConvertPreviewRequest,
     *,
@@ -124,6 +189,9 @@ async def build_mass_convert_preview(
         raise ValidationError("Choose at least one folder to preview.")
 
     if body.trash_folder and body.trash_folder.strip():
+        # Preview-only exclusion path supplied by an authenticated operator. It
+        # is resolved before comparison and never listed, opened, moved, or mutated.
+        # codeql[py/path-injection]
         trash_dir = Path(body.trash_folder.strip()).expanduser()
     else:
         if load_trash_context is None:
@@ -241,13 +309,16 @@ async def build_library_permissions_preview(
     file_count = 0
     preview_items: list[LibraryPermissionsPreviewItem] = []
     for item in generated_items:
-        path = Path(str(item.get("file_path", "")))
+        path = _resolve_generated_preview_path(
+            str(item.get("file_path", "")),
+            [Path(root["path"]) for root in job_context["library_roots"]],
+        )
         try:
-            path_stat = path.lstat()
+            path_stat = _preview_lstat(path)
         except OSError:
             item_type = "file"
         else:
-            if path.is_symlink():
+            if _is_preview_symlink(path):
                 item_type = "symlink"
             elif stat.S_ISDIR(path_stat.st_mode):
                 item_type = "folder"
@@ -327,10 +398,10 @@ async def build_mass_rename_preview(
         raise ValidationError("Choose at least one folder to preview this scope.")
 
     request_paths = list(body.file_paths)
+    library_roots = await _load_enabled_library_root_paths(session)
+    if not library_roots:
+        raise ValidationError("No enabled library roots are available for this preview.")
     if scope != "library":
-        library_roots = await _load_enabled_library_root_paths(session)
-        if not library_roots:
-            raise ValidationError("No enabled library roots are available for this preview.")
         require_dir = scope == "folder" or target == "folders"
         require_file = target == "files" and scope == "manual"
         resolved_paths: list[str] = []
@@ -453,14 +524,14 @@ async def build_mass_rename_preview(
                 issue_type_override=effective_issue_type,
             )
             template_key, template_label = _resolve_file_template_key(effective_issue_type)
-            target_path = Path(file_path).parent / proposed_name
+            target_path = _rename_target_path(file_path, proposed_name, library_roots)
             reason: str | None = None
             status = "ready"
             actionable = proposed_name != current_name
             if not actionable:
                 status = "unchanged"
                 reason = "Already matches the current naming template."
-            elif target_path.exists() and str(target_path) != file_path:
+            elif _rename_target_exists(target_path) and str(target_path) != file_path:
                 status = "conflict"
                 reason = f"Target already exists: {target_path.name}"
 
@@ -525,14 +596,14 @@ async def build_mass_rename_preview(
 
             series = series_match
             proposed_name = build_series_folder_name(series, naming_config)
-            target_path = Path(folder_path).parent / proposed_name
+            target_path = _rename_target_path(folder_path, proposed_name, library_roots)
             reason = None
             status = "ready"
             actionable = proposed_name != current_name
             if not actionable:
                 status = "unchanged"
                 reason = "Already matches the current folder template."
-            elif target_path.exists() and str(target_path) != folder_path:
+            elif _rename_target_exists(target_path) and str(target_path) != folder_path:
                 status = "conflict"
                 reason = f"Target already exists: {target_path.name}"
 

tests/api/test_utilities_preview_api.py

@@ -440,6 +440,27 @@ async def test_mass_convert_preview_accepts_multiple_folders(
     assert any(item["source_name"] == "batman-001.cbz" for item in data["items"])
 
 
+@pytest.mark.asyncio
+async def test_mass_convert_preview_rejects_folder_outside_library_roots(
+    authenticated_client,
+    preview_paths: dict[str, str],
+    tmp_path: Path,
+) -> None:  # type: ignore[no-untyped-def]
+    assert preview_paths["library_root"]
+    outside_folder = tmp_path / "outside"
+    outside_folder.mkdir()
+    (outside_folder / "outside.cbz").write_text("outside")
+
+    response = await authenticated_client.post(
+        "/api/v1/utilities/mass-convert/preview",
+        json={"scope": "folder", "file_paths": [str(outside_folder)]},
+        headers=_csrf_header_for(authenticated_client),
+    )
+
+    assert response.status_code == 422
+    assert "outside enabled library roots" in response.text
+
+
 @pytest.mark.asyncio
 async def test_library_permissions_preview_counts_recursive_folder_scope(
     authenticated_client,
@@ -468,6 +489,33 @@ async def test_library_permissions_preview_counts_recursive_folder_scope(
     assert any(item["item_type"] == "folder" for item in data["items"])
 
 
+@pytest.mark.asyncio
+async def test_library_permissions_preview_rejects_paths_outside_library_roots(
+    authenticated_client,
+    preview_paths: dict[str, str],
+    tmp_path: Path,
+) -> None:  # type: ignore[no-untyped-def]
+    assert preview_paths["library_root"]
+    outside_file = tmp_path / "outside.cbz"
+    outside_file.write_text("outside")
+
+    response = await authenticated_client.post(
+        "/api/v1/utilities/permissions/preview",
+        json={
+            "scope": "files",
+            "file_paths": [str(outside_file)],
+            "folder_mode": "750",
+            "file_mode": "640",
+            "include_folders": False,
+            "include_files": True,
+        },
+        headers=_csrf_header_for(authenticated_client),
+    )
+
+    assert response.status_code == 422
+    assert "outside enabled library roots" in response.text
+
+
 @pytest.mark.asyncio
 async def test_library_permissions_preview_respects_include_toggles(
     authenticated_client,