refactor(api): posters.py stage 2 — non-route logic moves to util owners

backend/modules/poster_cleanarr.py

@@ -18,6 +18,7 @@
 from backend.util.logger import Logger
 from backend.util.notification import NotificationManager
 from backend.util.normalization import normalize_titles, parse_asset_filename
+from backend.util.path_safety import resolve_confined
 
 # EXIF tag id Kometa writes onto its generated overlay images. Used by the
 # overlays_only mode to skip user-uploaded customs (which lack the tag).
@@ -847,6 +848,41 @@ def _resolve_orphan_instances(config: Any) -> List[str]:
             getattr(config, "instances", []) or []
         )
 
+    def _authorized_asset_dirs(
+        self, asset_dirs: List[str], logger: Logger
+    ) -> List[str]:
+        """Existing asset_dirs re-resolved inside the live config's allowed roots."""
+        # Re-authorized here because the API confines at enqueue but a worker
+        # acts later; an unloadable config returns [] so nothing is deleted.
+        config = getattr(self, "full_config", None)
+        if config is None:
+            # Shim entry points (run_orphan_assets_pass) build a bare instance.
+            from backend.util.config import load_config
+
+            try:
+                config = load_config()
+            except Exception as e:
+                logger.error(
+                    f"Cannot authorize asset_dirs against the allowed roots ({e}); "
+                    "skipping cleanup instead of acting on unverified paths."
+                )
+                return []
+
+        authorized: List[str] = []
+        for d in asset_dirs:
+            if not os.path.isdir(d):
+                logger.warning(f"asset_dir does not exist, skipping: {d}")
+                continue
+            real = resolve_confined(d, config)
+            if real is None:
+                logger.error(
+                    f"asset_dir resolves outside the allowed roots, "
+                    f"refusing to clean it: {d}"
+                )
+                continue
+            authorized.append(str(real))
+        return authorized
+
     def _run_orphan_pass(
         self,
         db: ChubDB,
@@ -883,10 +919,7 @@ def _run_orphan_pass(
             )
             return {"count": 0, "total_size": 0, "mode": mode}
 
-        valid_dirs = [d for d in asset_dirs if os.path.isdir(d)]
-        for d in asset_dirs:
-            if not os.path.isdir(d):
-                logger.warning(f"asset_dir does not exist, skipping: {d}")
+        valid_dirs = self._authorized_asset_dirs(asset_dirs, logger)
         if not valid_dirs:
             return {"count": 0, "total_size": 0, "mode": mode}
 
@@ -1147,10 +1180,7 @@ def _run_stale_pass(
             )
             return {"count": 0, "total_size": 0, "mode": mode}
 
-        for d in asset_dirs:
-            if not os.path.isdir(d):
-                logger.warning(f"asset_dir does not exist, skipping: {d}")
-        valid_dirs = [d for d in asset_dirs if os.path.isdir(d)]
+        valid_dirs = self._authorized_asset_dirs(asset_dirs, logger)
         if not valid_dirs:
             return {"count": 0, "total_size": 0, "mode": mode}
 

backend/util/asset_candidates.py

@@ -0,0 +1,78 @@
+"""
+Candidate assets for a media row — the shared gather + rank behind the manual
+poster picker and its artwork counterpart.
+
+Returns scored rows only; each caller owns the JSON shape it renders.
+"""
+
+import difflib
+import json
+from typing import Any, Dict, List, Optional, Tuple
+
+from backend.util.helper import is_match
+from backend.util.normalization import normalize_titles
+
+
+def rank_candidates(
+    db: Any,
+    row: Dict[str, Any],
+    asset_type: Optional[str],
+    image_type: Optional[str] = "poster",
+    limit: int = 24,
+) -> List[Tuple[Dict[str, Any], bool, float, str]]:
+    """Best-first (row, would_match, similarity, reason) tuples for a media row."""
+    season_number = row.get("season_number")
+    try:
+        alts = json.loads(row.get("alternate_titles") or "[]")
+    except (ValueError, TypeError):
+        alts = []
+    search_titles = [row.get("title")] + [a for a in alts if a]
+    row_norm = row.get("normalized_title") or normalize_titles(row.get("title") or "")
+
+    seen = set()
+    gathered = []
+    for st in search_titles:
+        # Checked per title too: an inner break alone lets each alternate
+        # title add another 800 rows before anything is scored.
+        if len(gathered) >= 800:
+            break
+        for c in db.poster.get_candidates_by_prefix(
+            st or "", asset_type=asset_type, image_type=image_type
+        ):
+            f = c.get("file")
+            if f and f not in seen:
+                seen.add(f)
+                gathered.append(c)
+            if len(gathered) >= 800:
+                break
+
+    # Score every candidate by title similarity. The prefix bucket alone is
+    # NOT relevance — without this, the picker showed every poster sharing
+    # the first 3 chars ("str" → Striptease, Strays, …) regardless of the
+    # title. Rank real matches first, then by similarity, and drop posters
+    # that neither match nor resemble the title.
+    scored = []
+    for c in gathered:
+        cs = c.get("season_number")
+        if season_number is not None and cs != season_number:
+            continue
+        if season_number is None and cs is not None:
+            continue
+        matched, reason = is_match(c, row)
+        sim = difflib.SequenceMatcher(
+            None, row_norm, c.get("normalized_title") or ""
+        ).ratio()
+        scored.append((bool(matched), sim, c, reason))
+
+    scored.sort(key=lambda x: (x[0], x[1]), reverse=True)
+
+    ranked: List[Tuple[Dict[str, Any], bool, float, str]] = []
+    for matched, sim, c, reason in scored:
+        # Real matches always show; non-matching extras only if the title
+        # genuinely resembles (drops same-prefix-but-unrelated noise).
+        if not matched and sim < 0.6:
+            continue
+        ranked.append((c, matched, sim, reason))
+        if len(ranked) >= limit:
+            break
+    return ranked

backend/util/path_safety.py

@@ -214,28 +214,26 @@ def get_browse_roots(config: ChubConfig) -> List[Path]:
 def is_path_allowed(path: str, config: ChubConfig) -> bool:
     """
     Check whether *path* falls under one of the allowed roots.
-    Returns True if the resolved path is inside any allowed root.
 
-    The path is resolved to an absolute path and then checked against
-    each allowed root using Path.relative_to(), which is safe against
-    traversal attacks (symlinks, .., etc.) because resolve() normalizes.
+    realpath resolves symlinks and `..` before the comparison, so traversal
+    can't win; the os.sep suffix keeps `/root_evil` out of `/root`.
     """
     if not path or not isinstance(path, str):
         return False
     # Reject null bytes (path injection vector)
     if "\x00" in path:
         return False
     try:
-        target = Path(path).expanduser().resolve()  # noqa: S108 — validated below
+        target = os.path.realpath(os.path.expanduser(path))
     except (ValueError, OSError):
         return False
 
+    # realpath + os.sep prefix: same verdict relative_to gave, in the shape
+    # CodeQL accepts as a traversal barrier. os.sep stops /root_evil.
     for root in get_allowed_roots(config):
-        try:
-            target.relative_to(root)
+        base = str(root)
+        if target == base or target.startswith(base + os.sep):
             return True
-        except ValueError:
-            continue
 
     return False
 
@@ -249,3 +247,25 @@ def resolve_confined(path: str, config: ChubConfig) -> Optional[Path]:
     except (ValueError, OSError):
         return None
     return Path(resolved) if is_path_allowed(resolved, config) else None
+
+
+def resolve_under_root(location: str, path: str, config: ChubConfig) -> Optional[Path]:
+    """Resolve an absolute *path*, or one under *location*, confined to allowed roots."""
+    if Path(path).is_absolute():
+        # Absolute paths (grid passes item.file; location is a label, not a
+        # root) validate on their own — resolve_confined covers symlink escapes.
+        return resolve_confined(path, config)
+
+    # Relative path — `location` must be an allowed root and the resolved
+    # result must stay inside it. os.sep suffix keeps `/posters_evil/x`
+    # from slipping past a `/posters` prefix.
+    if not is_path_allowed(location, config):
+        return None
+    base_dir = os.path.realpath(location)
+    # Re-confine the resolved root too — location may itself be a link.
+    if not is_path_allowed(base_dir, config):
+        return None
+    resolved = os.path.realpath(os.path.join(base_dir, path))
+    if resolved != base_dir and not resolved.startswith(base_dir + os.sep):
+        return None
+    return Path(resolved)

backend/util/plex_metadata.py

@@ -23,6 +23,7 @@
 
 Public surface:
     get_plex_metadata_dir(plex_path) -> str
+    resolve_in_metadata_dir(file_path, plex_path) -> Optional[str]
     get_in_use_hashes(db_path) -> Set[str]
     copy_plex_db(plex_path, dest) -> Optional[str]
     scan_bundles(plex_path, *, force=False) -> Dict
@@ -77,6 +78,15 @@ def get_plex_metadata_dir(plex_path: str) -> str:
     return os.path.join(plex_path, "Metadata")
 
 
+def resolve_in_metadata_dir(file_path: str, plex_path: str) -> Optional[str]:
+    """Resolve a variant path, or None when it lands outside Plex's Metadata dir."""
+    metadata_dir = os.path.realpath(get_plex_metadata_dir(plex_path))
+    real = os.path.realpath(file_path)
+    if not real.startswith(metadata_dir + os.sep):
+        return None
+    return real
+
+
 def copy_plex_db(plex_path: str, dest: str) -> Optional[str]:
     """
     Copy Plex's live SQLite DB to `dest` for read-only querying.
@@ -796,9 +806,8 @@ def delete_variant(file_path: str, *, plex_path: str) -> bool:
       from its metadata agents, and would pollute the scan cache).
     Returns True on success.
     """
-    metadata_dir = os.path.realpath(get_plex_metadata_dir(plex_path))
-    real = os.path.realpath(file_path)
-    if not real.startswith(metadata_dir + os.sep):
+    real = resolve_in_metadata_dir(file_path, plex_path)
+    if real is None:
         return False
     # Plex-sourced variants sit under `<bundle>/Contents/…`. Refuse.
     if f"{os.sep}Contents{os.sep}" in real:

backend/util/poster_cleanarr_settings.py

@@ -0,0 +1,133 @@
+"""
+Poster Cleanarr settings and job-request contract.
+
+Reads the `poster_cleanarr` config section on behalf of the UI/API layer and
+turns a cleanup request body into the overrides a `module_run` job expects.
+"""
+
+from typing import Any, List, Optional
+
+from backend.util.config import ChubConfig, ConfigError
+from backend.util.path_safety import resolve_confined
+
+
+def get_plex_path() -> Optional[str]:
+    """
+    Resolve the Plex filesystem path from config. The poster_cleanarr module
+    config is the canonical place — general config doesn't store this.
+    """
+    try:
+        from backend.util.config import load_config
+
+        cfg = load_config()
+    except ConfigError:
+        raise
+    except Exception:
+        return None
+    section = getattr(cfg, "poster_cleanarr", None)
+    if section is not None:
+        pp = getattr(section, "plex_path", None)
+        if pp:
+            return str(pp)
+    return None
+
+
+def get_excluded_libraries() -> List[str]:
+    """Plex library names the user opted out of in poster_cleanarr config.
+
+    Display-side mirror of the module's deletion-side deny-list — hides excluded
+    libraries from the by-media view and its libraries[] catalog. A malformed
+    config propagates (CONFIG_INVALID); other failures return [] (show
+    everything). This is a UI filter, not a safety guard — the in-use set is
+    global regardless of any opt-out.
+    """
+    try:
+        from backend.util.config import load_config
+
+        cfg = load_config()
+    except ConfigError:
+        raise
+    except Exception:
+        return []
+    section = getattr(cfg, "poster_cleanarr", None)
+    if section is None:
+        return []
+    return list(getattr(section, "excluded_libraries", None) or [])
+
+
+def _require_bool(body: dict, key: str) -> bool:
+    """Return body[key] only when it is a real bool."""
+    # bool("false") is True — coercing would silently enable a deleting pass.
+    value = body.get(key)
+    if not isinstance(value, bool):
+        raise ValueError(f"Invalid {key} type: {type(value).__name__}")
+    return value
+
+
+def _validate_sub_mode(value: Any, key: str) -> Optional[str]:
+    """Lowercase an orphan/stale sub-mode; None when absent, ValueError when malformed."""
+    if value is None:
+        return None
+    if not isinstance(value, str):
+        raise ValueError(f"Invalid {key} type: {type(value).__name__}")
+    mode = value.lower()
+    if mode not in ("report", "move", "remove"):
+        raise ValueError(f"Invalid {key} '{mode}'")
+    return mode
+
+
+def build_cleanup_overrides(body: dict, config: ChubConfig) -> dict:
+    """Assemble the poster_cleanarr job overrides from a cleanup request body.
+    Raises ValueError on a malformed field type, an invalid mode, or an
+    asset_dir outside the allowed roots (the route maps it to a 400). Bloat
+    accepts "nothing" so the UI can run stale/orphan cleanup with bloat off."""
+    raw_mode = body.get("mode")
+    if raw_mode is None:
+        mode = "report"
+    elif isinstance(raw_mode, str):
+        mode = raw_mode.lower() or "report"
+    else:
+        raise ValueError(f"Invalid mode type: {type(raw_mode).__name__}")
+    if mode not in ("report", "move", "remove", "nothing"):
+        raise ValueError(f"Invalid mode '{mode}'")
+    overrides: dict = {"mode": mode}
+
+    target_paths = body.get("target_paths")
+    if isinstance(target_paths, list) and target_paths:
+        overrides["target_paths"] = [str(p) for p in target_paths]
+
+    if "orphan_assets_enabled" in body:
+        overrides["orphan_assets_enabled"] = _require_bool(
+            body, "orphan_assets_enabled"
+        )
+    orphan_mode = _validate_sub_mode(
+        body.get("orphan_assets_mode"), "orphan_assets_mode"
+    )
+    if orphan_mode is not None:
+        overrides["orphan_assets_mode"] = orphan_mode
+
+    if "stale_duplicates_enabled" in body:
+        overrides["stale_duplicates_enabled"] = _require_bool(
+            body, "stale_duplicates_enabled"
+        )
+    stale_mode = _validate_sub_mode(
+        body.get("stale_duplicates_mode"), "stale_duplicates_mode"
+    )
+    if stale_mode is not None:
+        overrides["stale_duplicates_mode"] = stale_mode
+
+    asset_dirs = body.get("asset_dirs")
+    if isinstance(asset_dirs, list):
+        # These reach PosterCleanarr's orphan/stale passes, which DELETE —
+        # confine here so no caller can hand it an unchecked request path.
+        confined = []
+        for p in asset_dirs:
+            real = resolve_confined(str(p), config)
+            if real is None:
+                raise ValueError(f"asset_dirs path is outside the allowed roots: {p}")
+            confined.append(str(real))
+        overrides["asset_dirs"] = confined
+
+    if "overlays_only" in body:
+        overrides["overlays_only"] = _require_bool(body, "overlays_only")
+    return overrides