diff --git a/backend/api/posters.py b/backend/api/posters.py index 5448bcce..b44d229f 100644 --- a/backend/api/posters.py +++ b/backend/api/posters.py @@ -28,9 +28,22 @@ ) from backend.modules.sync_gdrive import SyncGDrive from backend.modules.unmatched_assets import UnmatchedAssets +from backend.util.asset_candidates import rank_candidates from backend.util.config import ConfigError from backend.util.database import ChubDB +from backend.util.database.poster_cache import ARTWORK_IMAGE_TYPES from backend.util.helper import get_static_dir +from backend.util.poster_cleanarr_settings import ( + build_cleanup_overrides, + get_excluded_libraries, + get_plex_path, +) +from backend.util.poster_images import ( + build_thumbnail, + optimize_poster_files, + resolve_format, + transcode_poster, +) router = APIRouter( prefix="/api/posters", @@ -985,174 +998,29 @@ async def optimize_posters( quality = max(1, min(100, body.get("quality", 85))) mode = body.get("mode", "report") - format_map = {"jpeg": "JPEG", "jpg": "JPEG", "webp": "WEBP", "png": "PNG"} - pil_format = format_map.get(target_format, "JPEG") - ext_map = {"JPEG": ".jpg", "WEBP": ".webp", "PNG": ".png"} - target_ext = ext_map.get(pil_format, ".jpg") + pil_format, target_ext = resolve_format(target_format) # The full-cache read + per-poster PIL resize/convert loop is heavy and # CPU-bound; run it off the event loop so it doesn't freeze the server. - return await run_in_threadpool( - _optimize_posters_sync, - db, - logger, - max_width, - max_height, - pil_format, - target_ext, - quality, - mode, - ) - - -def _optimize_posters_sync( - db, - logger, - max_width, - max_height, - pil_format, - target_ext, - quality, - mode, -) -> JSONResponse: - """Blocking body of optimize_posters — runs in a worker thread.""" - from PIL import Image - try: - # Get all posters from cache - posters = db.poster.get_all() - if not posters: - return ok( - "No posters found to optimize", - { - "processed": 0, - "skipped": 0, - "failed": 0, - "bytes_saved": 0, - "mode": mode, - }, - ) - - processed = 0 - skipped = 0 - failed = 0 - bytes_saved = 0 - details = [] - - for poster in posters: - file_path = poster.get("file", "") - folder = poster.get("folder", "") - full_path = os.path.join(folder, file_path) if folder else file_path - - if not full_path or not os.path.isfile(full_path): - skipped += 1 - continue - - try: - original_size = os.path.getsize(full_path) - - with Image.open(full_path) as img: - w, h = img.size - needs_resize = w > max_width or h > max_height - needs_convert = not full_path.lower().endswith(target_ext) - - if not needs_resize and not needs_convert: - skipped += 1 - continue - - if mode == "report": - details.append( - { - "file": full_path, - "size": original_size, - "dimensions": f"{w}x{h}", - "needs_resize": needs_resize, - "needs_convert": needs_convert, - } - ) - processed += 1 - continue - - # Actually optimize - if needs_resize: - img.thumbnail((max_width, max_height), Image.LANCZOS) - - img = img.convert("RGB") if pil_format in ("JPEG",) else img - - # Save to temp file, then replace - import tempfile - - with tempfile.NamedTemporaryFile( - suffix=target_ext, delete=False, dir=os.path.dirname(full_path) - ) as tmp: - save_kwargs = {"format": pil_format} - if pil_format in ("JPEG", "WEBP"): - save_kwargs["quality"] = quality - save_kwargs["optimize"] = True - img.save(tmp.name, **save_kwargs) - new_size = os.path.getsize(tmp.name) - - if new_size < original_size: - import shutil - - # Convert changes the container, so write the new - # extension and drop the original — else a .png would - # hold JPEG bytes. - base, _ = os.path.splitext(full_path) - dest_path = ( - base + target_ext if needs_convert else full_path - ) - # Don't clobber a different pre-existing file at the - # converted extension (e.g. a Kometa .jpg beside a - # Plex .png) — leave both, mirror poster_self_heal. - if dest_path != full_path and os.path.exists(dest_path): - os.unlink(tmp.name) - logger.warning( - f"optimize target exists, skipped to avoid " - f"clobber: {dest_path}" - ) - skipped += 1 - continue - shutil.move(tmp.name, dest_path) - if dest_path != full_path: - if os.path.exists(full_path): - os.remove(full_path) - poster["file"] = os.path.basename(dest_path) - try: - db.poster.upsert(poster) - except Exception as ce: - logger.warning( - f"optimized {dest_path}; cache update " - f"failed: {ce}" - ) - bytes_saved += original_size - new_size - processed += 1 - else: - os.unlink(tmp.name) - skipped += 1 - - except Exception as e: - logger.warning(f"Failed to optimize {full_path}: {e}") - failed += 1 - - saved_mb = round(bytes_saved / (1024 * 1024), 1) - result_data = { - "processed": processed, - "skipped": skipped, - "failed": failed, - "bytes_saved": bytes_saved, - "mode": mode, - } - if mode == "report" and details: - result_data["candidates"] = details[:100] + from backend.util.config import load_config - msg = ( - f"Found {processed} posters to optimize" - if mode == "report" - else f"Optimized {processed} posters, saved {saved_mb} MB" + # Loaded once here so the loop confines every cache row it rewrites. + config = load_config() + message, data = await run_in_threadpool( + optimize_poster_files, + db, + logger, + config, + max_width, + max_height, + pil_format, + target_ext, + quality, + mode, ) - return ok(msg, result_data) - + except ConfigError: + raise except Exception as e: logger.error(f"Error optimizing posters: {e}", exc_info=True) return error( @@ -1160,6 +1028,7 @@ def _optimize_posters_sync( code="OPTIMIZE_ERROR", status_code=500, ) + return ok(message, data) @router.get( @@ -1424,10 +1293,9 @@ async def ignore_artwork( db: ChubDB = Depends(get_database), ) -> JSONResponse: """Toggle the per-(media, image_type) ignore flag in media_asset_matches.""" - allowed = {"logo", "background", "squareart"} - if image_type not in allowed: + if image_type not in ARTWORK_IMAGE_TYPES: return error( - f"image_type must be one of {sorted(allowed)}, got '{image_type}'", + f"image_type must be one of {sorted(ARTWORK_IMAGE_TYPES)}, got '{image_type}'", code="INVALID_IMAGE_TYPE", status_code=400, ) @@ -1451,9 +1319,6 @@ async def ignore_artwork( ) -_ARTWORK_IMAGE_TYPES = {"logo", "background", "squareart"} - - @router.get( "/match/{media_id}/artwork/{image_type}/candidates", summary="Candidate artwork files for one (media, image_type)", @@ -1469,19 +1334,13 @@ async def get_artwork_candidates( logger: Any = Depends(get_logger), db: ChubDB = Depends(get_database), ) -> JSONResponse: - if image_type not in _ARTWORK_IMAGE_TYPES: + if image_type not in ARTWORK_IMAGE_TYPES: return error( - f"image_type must be one of {sorted(_ARTWORK_IMAGE_TYPES)}, got '{image_type}'", + f"image_type must be one of {sorted(ARTWORK_IMAGE_TYPES)}, got '{image_type}'", code="INVALID_IMAGE_TYPE", status_code=400, ) try: - import difflib - import json as _json - - from backend.util.helper import is_match - from backend.util.normalization import normalize_titles - row = ( db.collection.get_by_id(media_id) if kind == "collection" @@ -1491,66 +1350,24 @@ async def get_artwork_candidates( return error("Media row not found", code="NOT_FOUND", status_code=404) asset_type = "collection" if kind == "collection" else row.get("asset_type") - 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: - 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: # bound the pool before scoring - break - - # Same scoring as the poster picker: rank real matches first, then by - # title similarity, and drop same-prefix-but-unrelated noise. - 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) - - candidates = [] - for matched, sim, c, reason in scored: - if not matched and sim < 0.6: - continue - candidates.append( - { - "poster_id": c.get("id"), - "title": c.get("title"), - "year": c.get("year"), - "season_number": c.get("season_number"), - "style": c.get("style"), - "image_type": c.get("image_type"), - "owner": os.path.basename(os.path.dirname(c.get("file") or "")), - "would_match": matched, - "similarity": round(sim, 2), - "reason": reason or "title/year/season did not satisfy the matcher", - } + # Same scoring as the poster picker, scoped to one artwork type. + candidates = [ + { + "poster_id": c.get("id"), + "title": c.get("title"), + "year": c.get("year"), + "season_number": c.get("season_number"), + "style": c.get("style"), + "image_type": c.get("image_type"), + "owner": os.path.basename(os.path.dirname(c.get("file") or "")), + "would_match": matched, + "similarity": round(sim, 2), + "reason": reason or "title/year/season did not satisfy the matcher", + } + for c, matched, sim, reason in rank_candidates( + db, row, asset_type, image_type=image_type, limit=limit ) - if len(candidates) >= limit: - break + ] return ok( f"{len(candidates)} candidate {image_type} files", @@ -1559,7 +1376,7 @@ async def get_artwork_candidates( "media": { "title": row.get("title"), "year": row.get("year"), - "season_number": season_number, + "season_number": row.get("season_number"), "type": asset_type, "image_type": image_type, }, @@ -1597,9 +1414,9 @@ def apply_artwork( db: ChubDB = Depends(get_database), ) -> JSONResponse: """Link one artwork file to a media/collection row, apply it, and lock it.""" - if image_type not in _ARTWORK_IMAGE_TYPES: + if image_type not in ARTWORK_IMAGE_TYPES: return error( - f"image_type must be one of {sorted(_ARTWORK_IMAGE_TYPES)}, got '{image_type}'", + f"image_type must be one of {sorted(ARTWORK_IMAGE_TYPES)}, got '{image_type}'", code="INVALID_IMAGE_TYPE", status_code=400, ) @@ -1669,9 +1486,9 @@ async def unlock_artwork( logger: Any = Depends(get_logger), db: ChubDB = Depends(get_database), ) -> JSONResponse: - if image_type not in _ARTWORK_IMAGE_TYPES: + if image_type not in ARTWORK_IMAGE_TYPES: return error( - f"image_type must be one of {sorted(_ARTWORK_IMAGE_TYPES)}, got '{image_type}'", + f"image_type must be one of {sorted(ARTWORK_IMAGE_TYPES)}, got '{image_type}'", code="INVALID_IMAGE_TYPE", status_code=400, ) @@ -1814,13 +1631,6 @@ async def get_match_candidates( db: ChubDB = Depends(get_database), ) -> JSONResponse: try: - import json as _json - - import difflib - - from backend.util.helper import is_match - from backend.util.normalization import normalize_titles - row = ( db.collection.get_by_id(media_id) if kind == "collection" @@ -1830,70 +1640,22 @@ async def get_match_candidates( return error("Media row not found", code="NOT_FOUND", status_code=404) asset_type = "collection" if kind == "collection" else row.get("asset_type") - 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: - for c in db.poster.get_candidates_by_prefix( - st or "", asset_type=asset_type - ): - f = c.get("file") - if f and f not in seen: - seen.add(f) - gathered.append(c) - if len(gathered) >= 800: # bound the pool before scoring - 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) - - candidates = [] - 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 - candidates.append( - { - "poster_id": c.get("id"), - "title": c.get("title"), - "year": c.get("year"), - "season_number": c.get("season_number"), - "style": c.get("style"), - "owner": os.path.basename(os.path.dirname(c.get("file") or "")), - "would_match": matched, - "similarity": round(sim, 2), - "reason": reason or "title/year/season did not satisfy the matcher", - } + candidates = [ + { + "poster_id": c.get("id"), + "title": c.get("title"), + "year": c.get("year"), + "season_number": c.get("season_number"), + "style": c.get("style"), + "owner": os.path.basename(os.path.dirname(c.get("file") or "")), + "would_match": matched, + "similarity": round(sim, 2), + "reason": reason or "title/year/season did not satisfy the matcher", + } + for c, matched, sim, reason in rank_candidates( + db, row, asset_type, limit=limit ) - if len(candidates) >= limit: - break + ] return ok( f"{len(candidates)} candidate posters", @@ -1902,7 +1664,7 @@ async def get_match_candidates( "media": { "title": row.get("title"), "year": row.get("year"), - "season_number": season_number, + "season_number": row.get("season_number"), "type": asset_type, }, }, @@ -2599,7 +2361,7 @@ async def preview_poster_file( # Restrict the served file to configured allowed roots — otherwise # an authenticated caller could point at arbitrary dirs (/etc, /root). from backend.util.config import load_config - from backend.util.path_safety import is_path_allowed, resolve_confined + from backend.util.path_safety import resolve_under_root try: config = load_config() @@ -2608,57 +2370,15 @@ async def preview_poster_file( except Exception: # noqa: S110 — fail closed below config = None - if config is None: + # Fail closed: without config there are no allowed roots to check against. + file_path = resolve_under_root(location, path, config) if config else None + if file_path is None: return error( "Access denied - path outside allowed directory", code="PATH_TRAVERSAL_DENIED", status_code=403, ) - path_obj = Path(path) - if path_obj.is_absolute(): - # Frontend already gave us a concrete absolute path (e.g. the - # Assets Search grid passes item.file straight through, with - # item.folder in location as an owner label, not a root). - # Validate the file path itself against allowed roots instead - # of demanding that `location` is a root. resolve_confined - # authorizes the RESOLVED path (symlink escapes included). - file_path = resolve_confined(path, config) - if file_path is None: - return error( - "Access denied - path outside allowed directory", - code="PATH_TRAVERSAL_DENIED", - status_code=403, - ) - else: - # Relative path — `location` must be an allowed root and the - # resolved result must stay inside it (is_relative_to avoids - # the str.startswith bypass where `/posters_evil/x` slipped - # past a `/posters` prefix). - if not is_path_allowed(location, config): - return error( - "Access denied - path outside allowed directory", - code="PATH_TRAVERSAL_DENIED", - status_code=403, - ) - base_dir = Path(location).resolve() - # Re-confine the resolved root too — location may itself be a link. - if not is_path_allowed(str(base_dir), config): - return error( - "Access denied - path outside allowed directory", - code="PATH_TRAVERSAL_DENIED", - status_code=403, - ) - file_path = (base_dir / path).resolve() - try: - file_path.relative_to(base_dir) - except ValueError: - return error( - "Access denied - path outside allowed directory", - code="PATH_TRAVERSAL_DENIED", - status_code=403, - ) - if not file_path.exists() or not file_path.is_file(): return error( "Poster file not found", @@ -2998,50 +2718,6 @@ async def list_applied_media_by_style( # --------------------------------------------------------------------------- -def _get_plex_path(request: Request) -> 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_cleanarr_excluded_libraries(request: Request) -> 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 []) - - @router.get("/plex-metadata/by-media") async def list_plex_metadata_by_media( request: Request, @@ -3076,7 +2752,7 @@ async def list_plex_metadata_by_media( get_cached_transcoder, ) - plex_path = _get_plex_path(request) + plex_path = get_plex_path() if not plex_path: # Expected state on every Poster Cleanarr page mount before the # user has configured plex_path — not an operational failure, so @@ -3130,7 +2806,7 @@ async def list_plex_metadata_by_media( # so what the user sees matches what a run would touch. The in-use set is # unaffected — this is display only. excluded_libs = { - (n or "").strip().lower() for n in _get_cleanarr_excluded_libraries(request) + (n or "").strip().lower() for n in get_excluded_libraries() } if excluded_libs: bundles = [ @@ -3190,7 +2866,7 @@ async def list_plex_metadata_bloat( try: from backend.util.plex_metadata import bloat_flat_from_scan, get_cached_scan - plex_path = _get_plex_path(request) + plex_path = get_plex_path() if not plex_path: return error( "Plex path is not configured", code="PLEX_PATH_UNSET", status_code=400 @@ -3231,48 +2907,6 @@ async def list_plex_metadata_bloat( ) -def _build_cleanup_overrides(body: dict) -> dict: - """Assemble the poster_cleanarr job overrides from a cleanup request body. - Raises ValueError on an invalid mode (the route maps it to a 400). Bloat - accepts "nothing" so the UI can run stale/orphan cleanup with bloat off.""" - mode = (body.get("mode") or "report").lower() - 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"] = bool(body.get("orphan_assets_enabled")) - orphan_mode = body.get("orphan_assets_mode") - if isinstance(orphan_mode, str): - orphan_mode = orphan_mode.lower() - if orphan_mode not in ("report", "move", "remove"): - raise ValueError(f"Invalid orphan_assets_mode '{orphan_mode}'") - overrides["orphan_assets_mode"] = orphan_mode - - if "stale_duplicates_enabled" in body: - overrides["stale_duplicates_enabled"] = bool( - body.get("stale_duplicates_enabled") - ) - stale_mode = body.get("stale_duplicates_mode") - if isinstance(stale_mode, str): - stale_mode = stale_mode.lower() - if stale_mode not in ("report", "move", "remove"): - raise ValueError(f"Invalid stale_duplicates_mode '{stale_mode}'") - overrides["stale_duplicates_mode"] = stale_mode - - asset_dirs = body.get("asset_dirs") - if isinstance(asset_dirs, list): - overrides["asset_dirs"] = [str(p) for p in asset_dirs] - - if "overlays_only" in body: - overrides["overlays_only"] = bool(body.get("overlays_only")) - return overrides - - @router.post("/plex-metadata/cleanup") async def run_plex_metadata_cleanup( request: Request, @@ -3299,8 +2933,11 @@ async def run_plex_metadata_cleanup( body = await read_json_object(request) if body is BODY_TOO_LARGE: return body_too_large_error() + from backend.util.config import load_config + + config = load_config() try: - overrides = _build_cleanup_overrides(body) + overrides = build_cleanup_overrides(body, config) except ValueError as ve: logger.error(f"Invalid cleanup request: {ve}") return error("Invalid cleanup mode", code="INVALID_MODE", status_code=400) @@ -3321,6 +2958,8 @@ async def run_plex_metadata_cleanup( return error( "Failed to enqueue cleanup", code="ENQUEUE_FAILED", status_code=500 ) + except ConfigError: + raise except Exception as e: logger.error(f"Error enqueuing cleanup: {e}") return error( @@ -3345,7 +2984,7 @@ async def delete_plex_metadata_variant( path = body.get("path") if not isinstance(path, str) or not path: return error("Missing 'path'", code="MISSING_PATH", status_code=400) - plex_path = _get_plex_path(request) + plex_path = get_plex_path() if not plex_path: return error( "Plex path is not configured", code="PLEX_PATH_UNSET", status_code=400 @@ -3385,7 +3024,7 @@ async def set_plex_metadata_active( variant stays on disk and becomes bloat (cleanable from the same UI). """ try: - from backend.util.plex_metadata import invalidate_cache + from backend.util.plex_metadata import invalidate_cache, resolve_in_metadata_dir body = await read_json_object(request) if body is BODY_TOO_LARGE: @@ -3402,14 +3041,13 @@ async def set_plex_metadata_active( # Path-injection guard: only allow files inside Plex's Metadata dir. # Without this, a caller could upload ANY file on the server as a # Plex poster by passing an arbitrary filesystem path. - plex_path = _get_plex_path(request) + plex_path = get_plex_path() if not plex_path: return error( "Plex path is not configured", code="PLEX_PATH_UNSET", status_code=400 ) - metadata_dir = os.path.realpath(os.path.join(plex_path, "Metadata")) - safe_path = os.path.realpath(path) - if not safe_path.startswith(metadata_dir + os.sep): + safe_path = resolve_in_metadata_dir(path, plex_path) + if safe_path is None: return error( "Path outside Plex metadata dir", code="INVALID_PATH", status_code=400 ) @@ -3478,14 +3116,15 @@ async def get_plex_variant_thumbnail( file extension so we send it as image/jpeg (Plex posters are JPEGs). Validates the path stays within Plex's Metadata/ dir. """ - plex_path = _get_plex_path(request) + from backend.util.plex_metadata import resolve_in_metadata_dir + + plex_path = get_plex_path() if not plex_path: return error( "Plex path is not configured", code="PLEX_PATH_UNSET", status_code=400 ) - metadata_dir = os.path.realpath(os.path.join(plex_path, "Metadata")) - real = os.path.realpath(path) - if not real.startswith(metadata_dir + os.sep) or not os.path.isfile(real): + real = resolve_in_metadata_dir(path, plex_path) + if real is None or not os.path.isfile(real): return error("Invalid path", code="INVALID_PATH", status_code=400) return FileResponse(real, media_type="image/jpeg") @@ -3533,7 +3172,7 @@ async def enqueue_plex_metadata_scan( (GET /jobs/{id}/log-tail) before re-fetching /by-media. Duplicate enqueues collapse to the in-flight scan.""" try: - plex_path = _get_plex_path(request) + plex_path = get_plex_path() if not plex_path: return error( "Plex path is not configured", code="PLEX_PATH_UNSET", status_code=400 @@ -3689,38 +3328,37 @@ def get_poster_thumbnail( "Poster file not found on disk", code="FILE_NOT_FOUND", status_code=404 ) - # Resolve to a real path to prevent path traversal - full_path = os.path.realpath(raw_path) + # Confine the served path to configured roots — realpath normalizes + # but authorizes nothing, so a poisoned row could point anywhere. + from backend.util.config import load_config + from backend.util.path_safety import resolve_confined + + try: + config = load_config() + except ConfigError: + raise + except Exception: # noqa: S110 — fail closed below + config = None + + real = resolve_confined(raw_path, config) if config is not None else None + if real is None: + return error( + "Access denied - path outside allowed directory", + code="PATH_TRAVERSAL_DENIED", + status_code=403, + ) + full_path = str(real) if not os.path.isfile(full_path): return error( "Poster file not found on disk", code="FILE_NOT_FOUND", status_code=404 ) - # Build thumbnail cache path using resolved directory - safe_dir = os.path.dirname(full_path) - thumb_dir = os.path.join(safe_dir, ".thumbnails") - thumb_name = f"{poster_id}_w{width}.jpg" - thumb_path = os.path.join(thumb_dir, thumb_name) - - # Serve from cache if fresh - if os.path.isfile(thumb_path): - src_mtime = os.path.getmtime(full_path) - thumb_mtime = os.path.getmtime(thumb_path) - if thumb_mtime >= src_mtime: - return FileResponse(thumb_path, media_type="image/jpeg") - - # Generate thumbnail - from PIL import Image - - os.makedirs(thumb_dir, exist_ok=True) - with Image.open(full_path) as img: - aspect = img.height / img.width - target_height = int(width * aspect) - img.thumbnail((width, target_height), Image.LANCZOS) - img.convert("RGB").save(thumb_path, "JPEG", quality=60, optimize=True) - - return FileResponse(thumb_path, media_type="image/jpeg") + return FileResponse( + build_thumbnail(full_path, poster_id, width), media_type="image/jpeg" + ) + except ConfigError: + raise except Exception as e: logger.error(f"Error generating thumbnail for poster {poster_id}: {e}") return error( @@ -3790,8 +3428,26 @@ def download_poster( "No file path for poster", code="NO_FILE_PATH", status_code=404 ) - # Resolve to a real path to prevent path traversal - full_path = os.path.realpath(raw_path) + # Confine the served path to configured roots — realpath normalizes + # but authorizes nothing, so a poisoned row could point anywhere. + from backend.util.config import load_config + from backend.util.path_safety import resolve_confined + + try: + config = load_config() + except ConfigError: + raise + except Exception: # noqa: S110 — fail closed below + config = None + + real = resolve_confined(raw_path, config) if config is not None else None + if real is None: + return error( + "Access denied - path outside allowed directory", + code="PATH_TRAVERSAL_DENIED", + status_code=403, + ) + full_path = str(real) if not os.path.exists(full_path): return error( @@ -3803,37 +3459,20 @@ def download_poster( return FileResponse(full_path) # Process the image before serving - from PIL import Image - import tempfile - - format_map = {"jpeg": "JPEG", "jpg": "JPEG", "webp": "WEBP", "png": "PNG"} - pil_format = format_map.get((format or "").lower(), "JPEG") - ext_map = {"JPEG": ".jpg", "WEBP": ".webp", "PNG": ".png"} - target_ext = ext_map.get(pil_format, ".jpg") - media_types = {"JPEG": "image/jpeg", "WEBP": "image/webp", "PNG": "image/png"} - - with Image.open(full_path) as img: - if size: - img.thumbnail((size, size), Image.LANCZOS) - if pil_format in ("JPEG",): - img = img.convert("RGB") - - tmp = tempfile.NamedTemporaryFile(suffix=target_ext, delete=False) - save_kwargs = {"format": pil_format} - if quality and pil_format in ("JPEG", "WEBP"): - save_kwargs["quality"] = quality - save_kwargs["optimize"] = True - img.save(tmp.name, **save_kwargs) - tmp.close() + tmp_path, media_type, target_ext = transcode_poster( + full_path, size=size, image_format=format, quality=quality + ) return FileResponse( - tmp.name, - media_type=media_types.get(pil_format, "image/jpeg"), + tmp_path, + media_type=media_type, filename=f"poster_{poster_id}{target_ext}", # FileResponse doesn't delete what it serves — clean up the temp file. - background=BackgroundTask(os.unlink, tmp.name), + background=BackgroundTask(os.unlink, tmp_path), ) + except ConfigError: + raise except Exception as e: logger.error(f"Error downloading poster {poster_id}: {e}") return error( diff --git a/backend/modules/poster_cleanarr.py b/backend/modules/poster_cleanarr.py index 23730766..0f8212ab 100644 --- a/backend/modules/poster_cleanarr.py +++ b/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,40 @@ 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. + # Never self.full_config: that snapshot is taken at construction, so a + # root removed since would still authorize. Lazy import keeps it patchable. + 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 +918,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 +1179,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} diff --git a/backend/util/asset_candidates.py b/backend/util/asset_candidates.py new file mode 100644 index 00000000..283e237a --- /dev/null +++ b/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 diff --git a/backend/util/path_safety.py b/backend/util/path_safety.py index c578fac5..3c195c38 100644 --- a/backend/util/path_safety.py +++ b/backend/util/path_safety.py @@ -214,11 +214,9 @@ 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 @@ -226,16 +224,16 @@ def is_path_allowed(path: str, config: ChubConfig) -> bool: 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) diff --git a/backend/util/plex_metadata.py b/backend/util/plex_metadata.py index 576d0987..d53d5ce6 100644 --- a/backend/util/plex_metadata.py +++ b/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 `/Contents/…`. Refuse. if f"{os.sep}Contents{os.sep}" in real: diff --git a/backend/util/poster_cleanarr_settings.py b/backend/util/poster_cleanarr_settings.py new file mode 100644 index 00000000..049d9d63 --- /dev/null +++ b/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 diff --git a/backend/util/poster_images.py b/backend/util/poster_images.py new file mode 100644 index 00000000..caeabb58 --- /dev/null +++ b/backend/util/poster_images.py @@ -0,0 +1,232 @@ +""" +Poster image file operations — the PIL resize/convert/thumbnail work behind +the optimize, thumbnail and download endpoints. + +Every function returns plain values so the API layer keeps sole ownership of +HTTP shapes (backend/util never imports backend.api). +""" + +import os +from typing import Any, Dict, Optional, Tuple + +from backend.util.config import ChubConfig +from backend.util.path_safety import resolve_confined + +_PIL_FORMATS = {"jpeg": "JPEG", "jpg": "JPEG", "webp": "WEBP", "png": "PNG"} +_FORMAT_EXTENSIONS = {"JPEG": ".jpg", "WEBP": ".webp", "PNG": ".png"} +_FORMAT_MEDIA_TYPES = {"JPEG": "image/jpeg", "WEBP": "image/webp", "PNG": "image/png"} + + +def resolve_format(name: Optional[str]) -> Tuple[str, str]: + """Map a requested format name to its (PIL format, file extension) pair.""" + pil_format = _PIL_FORMATS.get((name or "").lower(), "JPEG") + return pil_format, _FORMAT_EXTENSIONS.get(pil_format, ".jpg") + + +def optimize_poster_files( + db, + logger, + config: ChubConfig, + max_width, + max_height, + pil_format, + target_ext, + quality, + mode, +) -> Tuple[str, Dict[str, Any]]: + """Resize/convert every cached poster in place; returns (message, result data).""" + from PIL import Image + + # Get all posters from cache + posters = db.poster.get_all() + if not posters: + return "No posters found to optimize", { + "processed": 0, + "skipped": 0, + "failed": 0, + "bytes_saved": 0, + "mode": mode, + } + + processed = 0 + skipped = 0 + failed = 0 + bytes_saved = 0 + details = [] + + for poster in posters: + file_path = poster.get("file", "") + folder = poster.get("folder", "") + raw_path = os.path.join(folder, file_path) if folder else file_path + + if not raw_path: + skipped += 1 + continue + + # A poisoned cache row must never reach the shutil.move/os.remove below. + real = resolve_confined(raw_path, config) + if real is None: + logger.warning(f"Skipping poster outside allowed roots: {raw_path}") + skipped += 1 + continue + + full_path = str(real) + if not os.path.isfile(full_path): + skipped += 1 + continue + + try: + original_size = os.path.getsize(full_path) + + with Image.open(full_path) as img: + w, h = img.size + needs_resize = w > max_width or h > max_height + needs_convert = not full_path.lower().endswith(target_ext) + + if not needs_resize and not needs_convert: + skipped += 1 + continue + + if mode == "report": + details.append( + { + "file": full_path, + "size": original_size, + "dimensions": f"{w}x{h}", + "needs_resize": needs_resize, + "needs_convert": needs_convert, + } + ) + processed += 1 + continue + + # Actually optimize + if needs_resize: + img.thumbnail((max_width, max_height), Image.LANCZOS) + + img = img.convert("RGB") if pil_format in ("JPEG",) else img + + # Save to temp file, then replace + import tempfile + + with tempfile.NamedTemporaryFile( + suffix=target_ext, delete=False, dir=os.path.dirname(full_path) + ) as tmp: + save_kwargs = {"format": pil_format} + if pil_format in ("JPEG", "WEBP"): + save_kwargs["quality"] = quality + save_kwargs["optimize"] = True + img.save(tmp.name, **save_kwargs) + new_size = os.path.getsize(tmp.name) + + if new_size < original_size: + import shutil + + # Convert changes the container, so write the new + # extension and drop the original — else a .png would + # hold JPEG bytes. + base, _ = os.path.splitext(full_path) + dest_path = base + target_ext if needs_convert else full_path + # Don't clobber a different pre-existing file at the + # converted extension (e.g. a Kometa .jpg beside a + # Plex .png) — leave both, mirror poster_self_heal. + if dest_path != full_path and os.path.exists(dest_path): + os.unlink(tmp.name) + logger.warning( + f"optimize target exists, skipped to avoid " + f"clobber: {dest_path}" + ) + skipped += 1 + continue + shutil.move(tmp.name, dest_path) + if dest_path != full_path: + if os.path.exists(full_path): + os.remove(full_path) + poster["file"] = os.path.basename(dest_path) + try: + db.poster.upsert(poster) + except Exception as ce: + logger.warning( + f"optimized {dest_path}; cache update failed: {ce}" + ) + bytes_saved += original_size - new_size + processed += 1 + else: + os.unlink(tmp.name) + skipped += 1 + + except Exception as e: + logger.warning(f"Failed to optimize {full_path}: {e}") + failed += 1 + + saved_mb = round(bytes_saved / (1024 * 1024), 1) + result_data = { + "processed": processed, + "skipped": skipped, + "failed": failed, + "bytes_saved": bytes_saved, + "mode": mode, + } + if mode == "report" and details: + result_data["candidates"] = details[:100] + + msg = ( + f"Found {processed} posters to optimize" + if mode == "report" + else f"Optimized {processed} posters, saved {saved_mb} MB" + ) + return msg, result_data + + +def build_thumbnail(source_path: str, poster_id: int, width: int) -> str: + """Return the cached JPEG thumbnail path for a poster, rendering it when stale.""" + # Cache beside the resolved source so the thumbnail follows the poster. + thumb_dir = os.path.join(os.path.dirname(source_path), ".thumbnails") + thumb_path = os.path.join(thumb_dir, f"{poster_id}_w{width}.jpg") + + # Serve from cache if fresh + if os.path.isfile(thumb_path): + src_mtime = os.path.getmtime(source_path) + thumb_mtime = os.path.getmtime(thumb_path) + if thumb_mtime >= src_mtime: + return thumb_path + + from PIL import Image + + os.makedirs(thumb_dir, exist_ok=True) + with Image.open(source_path) as img: + aspect = img.height / img.width + target_height = int(width * aspect) + img.thumbnail((width, target_height), Image.LANCZOS) + img.convert("RGB").save(thumb_path, "JPEG", quality=60, optimize=True) + + return thumb_path + + +def transcode_poster( + source_path: str, + size: Optional[int] = None, + image_format: Optional[str] = None, + quality: Optional[int] = None, +) -> Tuple[str, str, str]: + """Write a resized/converted copy to a temp file; returns (path, media type, ext).""" + from PIL import Image + import tempfile + + pil_format, target_ext = resolve_format(image_format) + + with Image.open(source_path) as img: + if size: + img.thumbnail((size, size), Image.LANCZOS) + if pil_format in ("JPEG",): + img = img.convert("RGB") + + tmp = tempfile.NamedTemporaryFile(suffix=target_ext, delete=False) + save_kwargs = {"format": pil_format} + if quality and pil_format in ("JPEG", "WEBP"): + save_kwargs["quality"] = quality + save_kwargs["optimize"] = True + img.save(tmp.name, **save_kwargs) + tmp.close() + + return tmp.name, _FORMAT_MEDIA_TYPES.get(pil_format, "image/jpeg"), target_ext diff --git a/tests/test_path_safety.py b/tests/test_path_safety.py index d431240e..87f626cf 100644 --- a/tests/test_path_safety.py +++ b/tests/test_path_safety.py @@ -7,6 +7,7 @@ get_browse_roots, is_path_allowed, resolve_confined, + resolve_under_root, ) @@ -270,3 +271,50 @@ def test_resolve_confined_denies_traversal_and_unusable_input(config_with_roots) assert resolve_confined(escape, config) is None assert resolve_confined("", config) is None assert resolve_confined(None, config) is None # type: ignore[arg-type] + + +def test_resolve_under_root_confines_relative_paths_to_the_base(config_with_roots): + """The preview resolver: relative stays under its root, `..` and non-roots deny.""" + config, tmp_path = config_with_roots + base = tmp_path / "posters_src" + inside = base / "a.jpg" + inside.write_text("x") + + assert resolve_under_root(str(base), "a.jpg", config) == inside.resolve() + assert resolve_under_root(str(base), "../secret.jpg", config) is None + assert resolve_under_root(str(tmp_path), "a.jpg", config) is None + + +def test_resolve_confined_denies_a_prefix_sibling_of_a_root(config_with_roots): + """`/posters_src_evil` must not pass as inside `/posters_src` on a bare prefix.""" + config, tmp_path = config_with_roots + sibling = tmp_path / "posters_src_evil" + sibling.mkdir() + victim = sibling / "x.jpg" + victim.write_text("x") + + assert resolve_confined(str(victim), config) is None + + +def test_resolve_under_root_denies_a_prefix_sibling_of_the_base(config_with_roots): + """`/posters_evil/x` must not slip past a `/posters` base via a bare prefix check.""" + config, tmp_path = config_with_roots + base = tmp_path / "posters_src" + sibling = tmp_path / "posters_src_evil" + sibling.mkdir() + (sibling / "x.jpg").write_text("x") + + assert resolve_under_root(str(base), "../posters_src_evil/x.jpg", config) is None + + +def test_resolve_under_root_validates_an_absolute_path_on_its_own(config_with_roots): + """An absolute path is authorized by its own resolution, not by `location`.""" + config, tmp_path = config_with_roots + inside = tmp_path / "posters_src" / "a.jpg" + inside.write_text("x") + outside = tmp_path / "secret.jpg" + outside.write_text("x") + + assert resolve_under_root("owner-label", str(inside), config) == inside.resolve() + base = str(tmp_path / "posters_src") + assert resolve_under_root(base, str(outside), config) is None diff --git a/tests/test_plex_metadata.py b/tests/test_plex_metadata.py index c84d089d..b45ec573 100644 --- a/tests/test_plex_metadata.py +++ b/tests/test_plex_metadata.py @@ -13,6 +13,7 @@ get_in_use_hashes, get_plex_metadata_dir, invalidate_cache, + resolve_in_metadata_dir, scan_bundles, ) @@ -50,6 +51,20 @@ def test_get_plex_metadata_dir_joins(): assert get_plex_metadata_dir("/plex") == "/plex/Metadata" +def test_resolve_in_metadata_dir_confines_to_the_metadata_tree(tmp_path): + """The one confinement check behind delete_variant, set-active and thumbnails.""" + inside = tmp_path / "Metadata" / "Movies" / "x" + inside.parent.mkdir(parents=True) + inside.write_bytes(b"x") + + assert resolve_in_metadata_dir(str(inside), str(tmp_path)) == str(inside) + assert resolve_in_metadata_dir(str(tmp_path / "Metadata"), str(tmp_path)) is None + assert ( + resolve_in_metadata_dir(str(tmp_path / "Metadata" / ".." / "etc"), str(tmp_path)) + is None + ) + + # --- copy_plex_db --- diff --git a/tests/test_poster_cleanarr_duplicates.py b/tests/test_poster_cleanarr_duplicates.py index 15857e51..97162287 100644 --- a/tests/test_poster_cleanarr_duplicates.py +++ b/tests/test_poster_cleanarr_duplicates.py @@ -7,7 +7,9 @@ import pytest from backend.modules.poster_cleanarr import ORPHAN_RESTORE_DIR_NAME, PosterCleanarr +from backend.util.config import ChubConfig, ConfigError from backend.util.database import ChubDB +from backend.util.path_safety import resolve_confined def _logger(): @@ -20,12 +22,31 @@ def _logger(): ) -def _make(): +def _make(*allowed_roots): + """Bare instance; `allowed_roots` seed the construction-time snapshot only.""" m = object.__new__(PosterCleanarr) m.logger = _logger() + m.full_config = ChubConfig() + m.full_config.poster_renamerr.source_dirs = [str(r) for r in allowed_roots] return m +def _live(monkeypatch, *allowed_roots): + """Point load_config at a config whose allowed roots are `allowed_roots`.""" + cfg = ChubConfig() + cfg.poster_renamerr.source_dirs = [str(r) for r in allowed_roots] + monkeypatch.setattr("backend.util.config.load_config", lambda: cfg) + return cfg + + +def _collecting_logger(): + """(logger, errors) — the logger appends every error message to the list.""" + errors = [] + logger = _logger() + logger.error = errors.append + return logger, errors + + @pytest.fixture def db(tmp_path): with ChubDB(_logger(), db_path=str(tmp_path / "chub.db")) as database: @@ -123,8 +144,9 @@ def test_scan_stale_marks_canonical_present(tmp_path): assert stale[0]["canonical_present"] is True -def test_execute_stale_remove_deletes_old_folder(tmp_path): +def test_execute_stale_remove_deletes_old_folder(tmp_path, monkeypatch): m = _make() + _live(monkeypatch, tmp_path) root = tmp_path / "assets" (root / "Dune Prophecy (2024) {tvdb-1}").mkdir(parents=True) # canonical present old = root / "Dune - Prophecy (2024) {tvdb-1}" @@ -146,10 +168,11 @@ def test_execute_stale_remove_deletes_old_folder(tmp_path): assert not old.exists() -def test_execute_stale_remove_keeps_only_copy(tmp_path): +def test_execute_stale_remove_keeps_only_copy(tmp_path, monkeypatch): """If the canonical folder is NOT on disk yet, removing the stale dup would delete the only staged copy — keep it and report instead.""" m = _make() + _live(monkeypatch, tmp_path) # authorized, so the only-copy guard is what keeps it root = tmp_path / "assets" old = root / "Dune - Prophecy (2024) {tvdb-1}" old.mkdir(parents=True) @@ -170,8 +193,9 @@ def test_execute_stale_remove_keeps_only_copy(tmp_path): assert old.exists() # only copy preserved -def test_execute_stale_report_deletes_nothing(tmp_path): +def test_execute_stale_report_deletes_nothing(tmp_path, monkeypatch): m = _make() + _live(monkeypatch, tmp_path) root = tmp_path / "assets" old = root / "Dune - Prophecy (2024) {tvdb-1}" old.mkdir(parents=True) @@ -192,10 +216,13 @@ def test_execute_stale_report_deletes_nothing(tmp_path): assert old.exists() -def test_execute_stale_move_moves_folder(tmp_path): +def test_execute_stale_move_moves_folder(tmp_path, monkeypatch): """move mode: stale folder with canonical present is moved under - // and the original is gone.""" + // and the original is gone. + The restore dir doesn't exist yet, so this also pins that the dest guard + confines the parent makedirs creates rather than the leaf.""" m = _make() + _live(monkeypatch, tmp_path) root = tmp_path / "assets" canonical_name = "Show (2024) {tvdb-1}" (root / canonical_name).mkdir(parents=True) # canonical present on disk @@ -245,8 +272,9 @@ def test_scan_stale_both_ids_resolves_via_tvdb(db, tmp_path): assert entry["canonical"] == tvdb_canonical -def test_run_stale_pass_aborts_when_no_instances(db, tmp_path): - m = _make() +def test_run_stale_pass_aborts_when_no_instances(db, tmp_path, monkeypatch): + m = _make(tmp_path) + _live(monkeypatch, tmp_path) # authorized, so the instances guard is what aborts res = m._run_stale_pass( db=db, instances=[], @@ -257,8 +285,9 @@ def test_run_stale_pass_aborts_when_no_instances(db, tmp_path): assert res["count"] == 0 -def test_run_stale_pass_reports(db, tmp_path): - m = _make() +def test_run_stale_pass_reports(db, tmp_path, monkeypatch): + m = _make(tmp_path) + _live(monkeypatch, tmp_path) _seed(db, "sonarr", "Dune Prophecy (2024) {tvdb-367118}", tvdb=367118) root = tmp_path / "assets" old = root / "Dune - Prophecy (2024) {tvdb-367118}" @@ -274,6 +303,123 @@ def test_run_stale_pass_reports(db, tmp_path): assert res["count"] == 1 +def _seed_stale_pair(root, canonical="Dune Prophecy (2024) {tvdb-367118}"): + """A stale-named folder plus its canonical sibling, so remove mode is allowed.""" + (root / canonical).mkdir(parents=True) + old = root / "Dune - Prophecy (2024) {tvdb-367118}" + old.mkdir(parents=True) + (old / "poster.jpg").write_bytes(b"x") + return old + + +def test_run_stale_pass_skips_asset_dir_outside_allowed_roots( + db, tmp_path, monkeypatch +): + """The stale pass refuses an unconfined dir at execution time; the allowed one still cleans.""" + allowed = tmp_path / "allowed" + outside = tmp_path / "outside" + confined_stale = _seed_stale_pair(allowed) + outside_stale = _seed_stale_pair(outside) + m = _make(allowed) + cfg = _live(monkeypatch, allowed) + assert resolve_confined(str(outside), cfg) is None # control + _seed(db, "sonarr", "Dune Prophecy (2024) {tvdb-367118}", tvdb=367118) + + res = m._run_stale_pass( + db=db, + instances=["sonarr"], + asset_dirs=[str(allowed), str(outside)], + mode="remove", + logger=_logger(), + ) + + assert res["count"] == 1 + assert not confined_stale.exists() # allowed root still processed + assert outside_stale.exists() # unauthorized root untouched + + +def test_run_stale_pass_fails_closed_when_config_unavailable(db, tmp_path, monkeypatch): + """Without a loadable config nothing can be authorized, so nothing is removed.""" + m = object.__new__(PosterCleanarr) # shim entry point's shape: no full_config + m.logger = _logger() + + def _boom(): + raise ConfigError("corrupt config") + + monkeypatch.setattr("backend.util.config.load_config", _boom) + stale = _seed_stale_pair(tmp_path / "assets") + _seed(db, "sonarr", "Dune Prophecy (2024) {tvdb-367118}", tvdb=367118) + errors = [] + logger = _logger() + logger.error = errors.append + + res = m._run_stale_pass( + db=db, + instances=["sonarr"], + asset_dirs=[str(tmp_path / "assets")], + mode="remove", + logger=logger, + ) + + assert res["count"] == 0 + assert stale.exists() # nothing removed on an unverifiable authorization + assert errors # and the refusal is logged, not silent + + +def test_stale_pass_ignores_the_construction_time_config_snapshot( + db, tmp_path, monkeypatch +): + """A root removed from config after the module was built must stop + authorizing: the worker reads the live config, never self.full_config.""" + m = _make(tmp_path) # the construction-time snapshot still carries the root + _live(monkeypatch) # ...but the live config no longer allows anything + stale = _seed_stale_pair(tmp_path / "assets") + _seed(db, "sonarr", "Dune Prophecy (2024) {tvdb-367118}", tvdb=367118) + logger, errors = _collecting_logger() + + assert m._authorized_asset_dirs([str(tmp_path / "assets")], logger) == [] + res = m._run_stale_pass( + db=db, + instances=["sonarr"], + asset_dirs=[str(tmp_path / "assets")], + mode="remove", + logger=logger, + ) + + assert res["count"] == 0 + assert stale.exists() # the de-authorized root is not cleaned + assert errors # and the refusal is logged, not silent + + +def _swap_parent_for_link(allowed, outside): + """Stale entry under `allowed` whose parent then becomes a link to `outside`.""" + parent = allowed / "shows" + parent.mkdir(parents=True) + scanned = parent / "Dune - Prophecy (2024) {tvdb-1}" + scanned.mkdir() + entry = { + "folder": str(scanned), + "asset_dir": str(allowed), + "name": scanned.name, + "canonical": "Dune Prophecy (2024) {tvdb-1}", + "canonical_present": True, + "id": ("tvdb", 1), + "size": 1, + } + scanned.rmdir() + parent.rmdir() + parent.symlink_to(outside, target_is_directory=True) + return entry + + +def _victim_folder(outside): + """A folder outside the roots that the swapped-in link points at.""" + victim = outside / "Dune - Prophecy (2024) {tvdb-1}" + victim.mkdir(parents=True) + (victim / "poster.jpg").write_bytes(b"x") + return victim + + def test_run_invokes_orphan_and_stale_passes(monkeypatch, tmp_path): """run() with mode='nothing' (skips Plex/bloat) must still invoke BOTH the orphan and stale passes when their config flags are set — the path a diff --git a/tests/test_poster_cleanarr_orphans.py b/tests/test_poster_cleanarr_orphans.py index e65155a0..ae55add2 100644 --- a/tests/test_poster_cleanarr_orphans.py +++ b/tests/test_poster_cleanarr_orphans.py @@ -10,9 +10,11 @@ import pytest -from backend.modules.poster_cleanarr import PosterCleanarr +from backend.modules.poster_cleanarr import ORPHAN_RESTORE_DIR_NAME, PosterCleanarr +from backend.util.config import ChubConfig, ConfigError from backend.util.database import ChubDB from backend.util.normalization import normalize_titles +from backend.util.path_safety import resolve_confined def _logger(): @@ -25,12 +27,31 @@ def _logger(): ) -def _make(): +def _make(*allowed_roots): + """Bare instance; `allowed_roots` seed the construction-time snapshot only.""" m = object.__new__(PosterCleanarr) m.logger = _logger() + m.full_config = ChubConfig() + m.full_config.poster_renamerr.source_dirs = [str(r) for r in allowed_roots] return m +def _live(monkeypatch, *allowed_roots): + """Point load_config at a config whose allowed roots are `allowed_roots`.""" + cfg = ChubConfig() + cfg.poster_renamerr.source_dirs = [str(r) for r in allowed_roots] + monkeypatch.setattr("backend.util.config.load_config", lambda: cfg) + return cfg + + +def _collecting_logger(): + """(logger, errors) — the logger appends every error message to the list.""" + errors = [] + logger = _logger() + logger.error = errors.append + return logger, errors + + @pytest.fixture def db(tmp_path): with ChubDB(_logger(), db_path=str(tmp_path / "chub.db")) as database: @@ -45,11 +66,12 @@ def _seed_media(db, identity_key, instance_name, normalized_title, alt=None): ) -def test_orphan_pass_aborts_when_title_set_empty(db, tmp_path): +def test_orphan_pass_aborts_when_title_set_empty(db, tmp_path, monkeypatch): """If media_cache is empty for the configured instances, the comparison set is empty and the pass must abort WITHOUT deleting anything — the last line of defense against wiping every asset.""" - m = _make() + m = _make(tmp_path) + _live(monkeypatch, tmp_path) # authorized, so the title-set guard is what aborts orphan = tmp_path / "Some Movie (2020).png" orphan.write_bytes(b"x") @@ -61,6 +83,77 @@ def test_orphan_pass_aborts_when_title_set_empty(db, tmp_path): assert orphan.exists() # nothing deleted +def test_orphan_pass_skips_asset_dir_outside_allowed_roots(db, tmp_path, monkeypatch): + """A dir outside the allowed roots is refused at execution time; its neighbour still cleans.""" + allowed = tmp_path / "allowed" + outside = tmp_path / "outside" + allowed.mkdir() + outside.mkdir() + m = _make(allowed) + cfg = _live(monkeypatch, allowed) + assert resolve_confined(str(outside), cfg) is None # control + _seed_media(db, "k1", "radarr1", normalize_titles("Keeper (2020)")) + confined_orphan = allowed / "Gone Movie (2019).png" + confined_orphan.write_bytes(b"x") + outside_orphan = outside / "Gone Movie (2019).png" + outside_orphan.write_bytes(b"x") + + res = m._run_orphan_pass( + db, + ["radarr1"], + [str(allowed), str(outside)], + "remove", + False, + _logger(), + ) + + assert res["count"] == 1 + assert not confined_orphan.exists() # allowed root still processed + assert outside_orphan.exists() # unauthorized root untouched + + +def test_orphan_pass_works_via_the_shim_shape_with_a_live_config(db, tmp_path, monkeypatch): + """poster_renamerr's post-rename cleanup builds a bare instance — it must still clean.""" + m = object.__new__(PosterCleanarr) # shim entry point's shape: no full_config + m.logger = _logger() + + live = ChubConfig() + live.poster_renamerr.destination_dir = str(tmp_path) + monkeypatch.setattr("backend.util.config.load_config", lambda: live) + + _seed_media(db, "k1", "radarr1", normalize_titles("Keeper (2020)")) + orphan = tmp_path / "Gone Movie (2019).png" + orphan.write_bytes(b"x") + + res = m._run_orphan_pass(db, ["radarr1"], [str(tmp_path)], "remove", False, _logger()) + + assert res["count"] == 1, res + assert not orphan.exists() + + +def test_orphan_pass_fails_closed_when_config_unavailable(db, tmp_path, monkeypatch): + """No config means no authorization, so the pass must delete nothing.""" + m = object.__new__(PosterCleanarr) # shim entry point's shape: no full_config + m.logger = _logger() + + def _boom(): + raise ConfigError("corrupt config") + + monkeypatch.setattr("backend.util.config.load_config", _boom) + _seed_media(db, "k1", "radarr1", normalize_titles("Keeper (2020)")) + orphan = tmp_path / "Gone Movie (2019).png" + orphan.write_bytes(b"x") + errors = [] + logger = _logger() + logger.error = errors.append + + res = m._run_orphan_pass(db, ["radarr1"], [str(tmp_path)], "remove", False, logger) + + assert res["count"] == 0 + assert orphan.exists() # nothing deleted on an unverifiable authorization + assert errors # and the refusal is logged, not silent + + def test_build_title_set_filters_by_instance_and_absorbs_alternates(db): m = _make() _seed_media(db, "k1", "radarr1", "dune", '["dunepartone"]') @@ -73,9 +166,10 @@ def test_build_title_set_filters_by_instance_and_absorbs_alternates(db): assert "shouldbeignored" not in titles # other instance excluded -def test_execute_orphan_mode_report_then_remove(tmp_path): +def test_execute_orphan_mode_report_then_remove(tmp_path, monkeypatch): """report counts but deletes nothing; remove actually deletes the file.""" m = _make() + _live(monkeypatch, tmp_path) f = tmp_path / "orphan.png" f.write_bytes(b"x") orphan = { @@ -94,8 +188,11 @@ def test_execute_orphan_mode_report_then_remove(tmp_path): assert not f.exists() # remove deletes -def test_execute_orphan_mode_move_relocates(tmp_path): +def test_execute_orphan_mode_move_relocates(tmp_path, monkeypatch): + """The restore dir doesn't exist yet: the dest guard must authorize the + parent makedirs creates, not the leaf, or moves silently stop happening.""" m = _make() + _live(monkeypatch, tmp_path) f = tmp_path / "orphan.png" f.write_bytes(b"x") orphan = { @@ -108,9 +205,48 @@ def test_execute_orphan_mode_move_relocates(tmp_path): res = m._execute_orphan_mode([orphan], "move", _logger()) assert res["count"] == 1 assert not f.exists() # moved out of its original location + assert (tmp_path / ORPHAN_RESTORE_DIR_NAME / "orphan.png").exists() + + +# ── Live-config authorization + per-target re-confinement ──────────────────── -# ── ID-based matching (spare-only) ─────────────────────────────────────────── +def test_orphan_pass_ignores_the_construction_time_config_snapshot( + db, tmp_path, monkeypatch +): + """A root removed from config after the module was built must stop + authorizing: the worker reads the live config, never self.full_config.""" + m = _make(tmp_path) # the construction-time snapshot still carries the root + _live(monkeypatch) # ...but the live config no longer allows anything + _seed_media(db, "k1", "radarr1", normalize_titles("Keeper (2020)")) + orphan = tmp_path / "Gone Movie (2019).png" + orphan.write_bytes(b"x") + logger, errors = _collecting_logger() + + assert m._authorized_asset_dirs([str(tmp_path)], logger) == [] + res = m._run_orphan_pass(db, ["radarr1"], [str(tmp_path)], "remove", False, logger) + + assert res["count"] == 0 + assert orphan.exists() # the de-authorized root is not cleaned + assert errors # and the refusal is logged, not silent + + +def _swap_parent_for_link(allowed, outside): + """Orphan item under `allowed` whose parent then becomes a link to `outside`.""" + show = allowed / "Show" + show.mkdir() + scanned = show / "poster.png" + scanned.write_bytes(b"x") + item = { + "path": str(scanned), + "size": 1, + "parsed": "show", + "asset_dir": str(allowed), + } + scanned.unlink() + show.rmdir() + show.symlink_to(outside, target_is_directory=True) + return item def _write(tmp_path, name): diff --git a/tests/test_posters_api_kometa_scan.py b/tests/test_posters_api_kometa_scan.py index 6fd6f554..6185cbef 100644 --- a/tests/test_posters_api_kometa_scan.py +++ b/tests/test_posters_api_kometa_scan.py @@ -1,5 +1,7 @@ +import os from types import SimpleNamespace +from backend.util.config import ChubConfig from backend.util.database import ChubDB @@ -38,14 +40,15 @@ def test_resolve_plex_match_via_plex_mapping(tmp_path): def test_cleanup_overrides_parse_stale(): - from backend.api.posters import _build_cleanup_overrides + from backend.util.poster_cleanarr_settings import build_cleanup_overrides - ov = _build_cleanup_overrides( + ov = build_cleanup_overrides( { "mode": "remove", "stale_duplicates_enabled": True, "stale_duplicates_mode": "move", - } + }, + ChubConfig(), ) assert ov["mode"] == "remove" assert ov["stale_duplicates_enabled"] is True @@ -53,20 +56,99 @@ def test_cleanup_overrides_parse_stale(): def test_cleanup_overrides_parse_overlays_only(): - from backend.api.posters import _build_cleanup_overrides + from backend.util.poster_cleanarr_settings import build_cleanup_overrides - assert _build_cleanup_overrides({"overlays_only": True})["overlays_only"] is True - assert _build_cleanup_overrides({"overlays_only": False})["overlays_only"] is False + cfg = ChubConfig() + assert ( + build_cleanup_overrides({"overlays_only": True}, cfg)["overlays_only"] is True + ) + assert ( + build_cleanup_overrides({"overlays_only": False}, cfg)["overlays_only"] is False + ) # absent -> not in overrides (module keeps its saved overlays_only) - assert "overlays_only" not in _build_cleanup_overrides({"mode": "report"}) + assert "overlays_only" not in build_cleanup_overrides({"mode": "report"}, cfg) def test_cleanup_overrides_allows_nothing_and_rejects_bad_stale_mode(): import pytest - from backend.api.posters import _build_cleanup_overrides + from backend.util.poster_cleanarr_settings import build_cleanup_overrides + cfg = ChubConfig() # 'nothing' is allowed for the bloat mode (UI runs stale/orphan with bloat off) - assert _build_cleanup_overrides({"mode": "nothing"})["mode"] == "nothing" + assert build_cleanup_overrides({"mode": "nothing"}, cfg)["mode"] == "nothing" + with pytest.raises(ValueError): + build_cleanup_overrides( + {"mode": "report", "stale_duplicates_mode": "nuke"}, cfg + ) + + +def test_cleanup_overrides_rejects_non_string_mode(): + """A non-string mode is a 400, not an AttributeError escaping as a 500.""" + import pytest + + from backend.util.poster_cleanarr_settings import build_cleanup_overrides + + cfg = ChubConfig() + with pytest.raises(ValueError): + build_cleanup_overrides({"mode": 5}, cfg) + # absent / None still default to report + assert build_cleanup_overrides({"mode": None}, cfg)["mode"] == "report" + assert build_cleanup_overrides({}, cfg)["mode"] == "report" + + +def test_cleanup_overrides_rejects_non_bool_flags(): + """bool("false") is True — a malformed flag must never enable a deleting pass.""" + import pytest + + from backend.util.poster_cleanarr_settings import build_cleanup_overrides + + cfg = ChubConfig() + for key in ( + "orphan_assets_enabled", + "stale_duplicates_enabled", + "overlays_only", + ): + with pytest.raises(ValueError): + build_cleanup_overrides({"mode": "remove", key: "false"}, cfg) + + +def test_cleanup_overrides_rejects_non_string_sub_modes(): + """A malformed sub-mode is rejected instead of silently falling back to config.""" + import pytest + + from backend.util.poster_cleanarr_settings import build_cleanup_overrides + + cfg = ChubConfig() + for key in ("orphan_assets_mode", "stale_duplicates_mode"): + with pytest.raises(ValueError): + build_cleanup_overrides({"mode": "remove", key: 5}, cfg) + # None keeps "not specified" — the module's saved mode applies + assert key not in build_cleanup_overrides({"mode": "remove", key: None}, cfg) + + +def test_cleanup_overrides_confines_asset_dirs(tmp_path): + """A configured asset_dir survives the guard and comes back resolved.""" + from backend.util.poster_cleanarr_settings import build_cleanup_overrides + + assets = tmp_path / "assets" + assets.mkdir() + cfg = ChubConfig() + cfg.poster_renamerr.source_dirs = [str(tmp_path)] + + ov = build_cleanup_overrides({"asset_dirs": [str(assets) + "/."]}, cfg) + + assert ov["asset_dirs"] == [os.path.realpath(str(assets))] + + +def test_cleanup_overrides_rejects_asset_dirs_outside_roots(tmp_path): + """An outside dir raises before it can reach the deleting cleanup passes.""" + import pytest + + from backend.util.poster_cleanarr_settings import build_cleanup_overrides + + cfg = ChubConfig() + cfg.poster_renamerr.source_dirs = [str(tmp_path)] + with pytest.raises(ValueError): - _build_cleanup_overrides({"mode": "report", "stale_duplicates_mode": "nuke"}) + build_cleanup_overrides({"asset_dirs": ["/etc"]}, cfg) diff --git a/tests/test_posters_queries.py b/tests/test_posters_queries.py index 80b1d809..e504416c 100644 --- a/tests/test_posters_queries.py +++ b/tests/test_posters_queries.py @@ -13,8 +13,9 @@ sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) -from backend.util.config import ConfigError # noqa: E402 +from backend.util.config import ChubConfig, ConfigError # noqa: E402 from backend.util.database import ChubDB # noqa: E402 +from backend.util.path_safety import is_path_allowed # noqa: E402 class _StubLog: @@ -460,3 +461,208 @@ def test_delete_poster_route_leaves_lookalike_filenames_matched(db): assert body["data"]["media_unmatched"] == 1 assert db.media.get_by_id(literal)["matched"] == 0 assert db.media.get_by_id(lookalike)["matched"] == 1 + + +# --- Stage 2: helpers that moved out of posters.py -------------------------- + + +def test_rank_candidates_ranks_matches_first_and_drops_prefix_noise(db): + """The picker's shared ranker: real match first, same-prefix strangers dropped.""" + from backend.util.asset_candidates import rank_candidates + + _seed_poster(db, "Dune", file="/src/Dune (2021).jpg") + _seed_poster(db, "Dungeons and Dragons Honor Among Thieves") + row = db.media.get_by_id(_seed_media(db, "Dune")) + + ranked = rank_candidates(db, row, "movie") + + assert [c["title"] for c, *_ in ranked] == ["Dune"] + assert ranked[0][1] is True + + +def test_rank_candidates_caps_the_pool_across_alternate_titles(db): + """The 800 cap is absolute: extra alternate titles must not each add another 800.""" + from backend.util.asset_candidates import rank_candidates + + calls = [] + + class _Poster: + def get_candidates_by_prefix(self, prefix, **kw): + calls.append(prefix) + return [ + {"file": f"/src/{prefix}-{i}.jpg", "title": "Dune", "id": i} + for i in range(800) + ] + + class _DB: + poster = _Poster() + + row = { + "title": "Dune", + "normalized_title": "dune", + "alternate_titles": '["Duna", "Dyuna"]', + } + + rank_candidates(_DB(), row, "movie") + + # One title fills the pool; the remaining two must never be queried. + assert len(calls) == 1, calls + + +def test_rank_candidates_scopes_to_the_requested_image_type(db): + """An artwork lookup must not surface the poster row for the same title.""" + from backend.util.asset_candidates import rank_candidates + + _seed_poster(db, "Dune", file="/src/Dune (2021).jpg") + _seed_poster(db, "Dune", file="/src/Dune (2021) - Logo.png", image_type="logo") + row = db.media.get_by_id(_seed_media(db, "Dune")) + + ranked = rank_candidates(db, row, "movie", image_type="logo") + + assert [c["file"] for c, *_ in ranked] == ["/src/Dune (2021) - Logo.png"] + + +def test_resolve_format_maps_aliases_and_falls_back_to_jpeg(): + """Every download/optimize call resolves its target through this one map.""" + from backend.util.poster_images import resolve_format + + assert resolve_format("jpg") == ("JPEG", ".jpg") + assert resolve_format("WEBP") == ("WEBP", ".webp") + assert resolve_format(None) == ("JPEG", ".jpg") + + +def test_thumbnail_and_download_refuse_rows_outside_allowed_roots( + db, monkeypatch, tmp_path +): + """A poisoned row pointing outside every configured root is 403, not served.""" + victim = tmp_path / "elsewhere" / "secret.jpg" + victim.parent.mkdir() + victim.write_bytes(b"x") + pid = _seed_poster(db, "Evil", file=str(victim), folder=str(victim.parent)) + + monkeypatch.setattr("backend.util.config.load_config", ChubConfig) + # Precondition: the 403s below must come from confinement, not from a + # tmp_path that happened to be unreachable. + assert not is_path_allowed(str(victim), ChubConfig()) + client = _client(db) + + thumb = client.get(f"/api/posters/{pid}/thumbnail") + dl = client.post(f"/api/posters/{pid}/download") + assert thumb.status_code == 403, thumb.text + assert dl.status_code == 403, dl.text + + +def _config_error(): + """Stand-in load_config for a malformed config file.""" + raise ConfigError("corrupt config") + + +def test_thumbnail_surfaces_config_error_as_config_invalid(db, monkeypatch): + """A malformed config reaches main.py's handler, not the generic 500.""" + pid = _seed_poster(db, "Dune") + monkeypatch.setattr("backend.util.config.load_config", _config_error) + + resp = _client(db).get(f"/api/posters/{pid}/thumbnail") + + assert resp.status_code == 500 + assert resp.json()["error_code"] == "CONFIG_INVALID" + + +def test_download_surfaces_config_error_as_config_invalid(db, monkeypatch): + """Same for the download route — CONFIG_INVALID, not POSTER_DOWNLOAD_ERROR.""" + pid = _seed_poster(db, "Dune") + monkeypatch.setattr("backend.util.config.load_config", _config_error) + + resp = _client(db).post(f"/api/posters/{pid}/download") + + assert resp.status_code == 500 + assert resp.json()["error_code"] == "CONFIG_INVALID" + + +def test_optimize_surfaces_config_error_as_config_invalid(db, monkeypatch): + """Optimize loads config to confine its writes; a bad one is CONFIG_INVALID.""" + monkeypatch.setattr("backend.util.config.load_config", _config_error) + + resp = _client(db).post("/api/posters/optimize", json={"mode": "report"}) + + assert resp.status_code == 500 + assert resp.json()["error_code"] == "CONFIG_INVALID" + + +def test_cleanup_route_refuses_asset_dirs_outside_allowed_roots(db, monkeypatch): + """asset_dirs feed a deleting job — an outside dir is a 400, never enqueued.""" + import backend.api.posters as posters + + monkeypatch.setattr("backend.util.config.load_config", ChubConfig) + client = _client(db) + # The real dependency builds a file-backed module logger. + client.app.dependency_overrides[posters.get_cleanarr_logger] = _StubLog + + resp = client.post( + "/api/posters/plex-metadata/cleanup", + json={"mode": "remove", "asset_dirs": ["/etc"]}, + ) + + assert resp.status_code == 400, resp.text + assert resp.json()["error_code"] == "INVALID_MODE" + + +def test_cleanup_route_rejects_malformed_field_types(db, monkeypatch): + """A non-string mode or non-bool flag is a 400, not a 500 and not a silent enable.""" + import backend.api.posters as posters + + monkeypatch.setattr("backend.util.config.load_config", ChubConfig) + client = _client(db) + client.app.dependency_overrides[posters.get_cleanarr_logger] = _StubLog + + for body in ({"mode": 5}, {"mode": "report", "orphan_assets_enabled": "false"}): + resp = client.post("/api/posters/plex-metadata/cleanup", json=body) + assert resp.status_code == 400, resp.text + assert resp.json()["error_code"] == "INVALID_MODE" + + +def test_cleanup_route_surfaces_config_error_as_config_invalid(db, monkeypatch): + """The cleanup route loads config too — malformed must not read as INVALID_MODE.""" + import backend.api.posters as posters + + monkeypatch.setattr("backend.util.config.load_config", _config_error) + client = _client(db) + client.app.dependency_overrides[posters.get_cleanarr_logger] = _StubLog + + resp = client.post( + "/api/posters/plex-metadata/cleanup", json={"mode": "remove"} + ) + + assert resp.status_code == 500 + assert resp.json()["error_code"] == "CONFIG_INVALID" + + +def test_optimize_poster_files_skips_rows_outside_allowed_roots(db, tmp_path): + """A poisoned row is skipped and left on disk; the confined poster still runs.""" + from PIL import Image + + from backend.util.poster_images import optimize_poster_files + + allowed = tmp_path / "posters" + allowed.mkdir() + outside = tmp_path / "elsewhere" + outside.mkdir() + good = allowed / "Dune.jpg" + evil = outside / "secret.jpg" + for path in (good, evil): + Image.new("RGB", (2000, 3000)).save(path) + untouched = evil.read_bytes() + _seed_poster(db, "Dune", file=str(good), folder=str(allowed)) + _seed_poster(db, "Evil", file=str(evil), folder=str(outside)) + + config = ChubConfig() + config.poster_renamerr.source_dirs = [str(allowed)] + + _msg, data = optimize_poster_files( + db, _StubLog(), config, 1000, 1500, "JPEG", ".jpg", 85, "optimize" + ) + + assert (data["processed"], data["skipped"], data["failed"]) == (1, 1, 0) + assert evil.read_bytes() == untouched + with Image.open(good) as img: + assert img.size == (1000, 1500)