From a747584adca8fb4f51eb491416f744785ad9b074 Mon Sep 17 00:00:00 2001
From: chodeus <dhoges90@icloud.com>
Date: Fri, 14 Aug 2026 17:35:24 +0800
Subject: [PATCH 1/4] fix(security): confine and anchor every cleanup delete
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The dirs were authorized once at enqueue, but each destructive op still
resolved its own path, so a symlink swap between scan and mutation could
send shutil.move/os.remove/shutil.rmtree outside the roots. Every sink now
re-confines its target immediately before acting, and the two delete sinks
go further: they open the confined PARENT with O_NOFOLLOW, pin it by
(st_dev, st_ino) the way prune_old_backups does, and operate by name
through that descriptor — so the name cannot be swapped for a link after
the check. rmtree takes dir_fd since CPython 3.11, so recursive delete is
anchored too, no reimplementation.

Moves stay at re-confinement only: shutil.move has no dir_fd and hardening
it would mean reimplementing the cross-filesystem fallback. Deletes can no
longer escape the authorized root; the leaf name is still resolved at the
syscall, which is the honest remaining gap.
---
 backend/modules/poster_cleanarr.py       | 171 ++++++++++++++----
 tests/test_poster_cleanarr_duplicates.py | 128 +++++++++++++-
 tests/test_poster_cleanarr_orphans.py    | 212 ++++++++++++++++++++++-
 3 files changed, 476 insertions(+), 35 deletions(-)

diff --git a/backend/modules/poster_cleanarr.py b/backend/modules/poster_cleanarr.py
index 0f8212ab..a24e833f 100644
--- a/backend/modules/poster_cleanarr.py
+++ b/backend/modules/poster_cleanarr.py
@@ -12,6 +12,7 @@
 from PIL import Image, UnidentifiedImageError
 
 from backend.util.base_module import ChubModule
+from backend.util.config import ChubConfig
 from backend.util.constants import asset_type_regex, tmdb_id_regex, tvdb_id_regex
 from backend.util.database import ChubDB
 from backend.util.helper import create_table
@@ -848,23 +849,108 @@ 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.
+    def _live_config(self, logger: Logger) -> Optional[ChubConfig]:
+        """Config as it is on disk now, or None (logged) so callers fail closed."""
         # 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()
+            return load_config()
         except Exception as e:
             logger.error(
-                f"Cannot authorize asset_dirs against the allowed roots ({e}); "
+                f"Cannot authorize paths against the allowed roots ({e}); "
                 "skipping cleanup instead of acting on unverified paths."
             )
+            return None
+
+    def _confined_target(
+        self, path: str, config: Optional[ChubConfig], logger: Logger
+    ) -> Optional[str]:
+        """Realpath of `path` if inside a live allowed root, else None (logged)."""
+        real = resolve_confined(path, config) if config is not None else None
+        if real is None:
+            logger.error(
+                f"Target resolves outside the allowed roots, refusing to touch: {path}"
+            )
+            return None
+        return str(real)
+
+    def _confined_parent_fd(
+        self, path: str, config: Optional[ChubConfig], logger: Logger
+    ) -> Optional[int]:
+        """Descriptor on `path`'s confined parent, or None (logged); caller closes."""
+        if self._confined_target(path, config, logger) is None:
+            return None
+        # Confine the PARENT, not the leaf: the leaf is what we delete, and
+        # deleting it by name through this descriptor is what closes the race.
+        parent = self._confined_target(os.path.dirname(path), config, logger)
+        if parent is None:
+            return None
+        flags = (
+            os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
+        )
+        try:
+            dir_fd = os.open(parent, flags)
+        except OSError as e:
+            logger.error(f"Refusing to delete {path}: parent not openable ({e})")
+            return None
+        try:
+            # Mirrors maintenance.prune_old_backups: the descriptor must BE the
+            # directory that was authorized, not one swapped in after the check.
+            opened, authorized = os.fstat(dir_fd), os.stat(parent)
+            if (opened.st_dev, opened.st_ino) != (authorized.st_dev, authorized.st_ino):
+                logger.error(f"Refusing to delete {path}: its parent changed under us")
+                os.close(dir_fd)
+                return None
+        except OSError as e:
+            logger.error(f"Refusing to delete {path}: parent unverifiable ({e})")
+            os.close(dir_fd)
+            return None
+        return dir_fd
+
+    def _remove_confined(
+        self, path: str, config: Optional[ChubConfig], logger: Logger
+    ) -> bool:
+        """Unlink `path` by name through a descriptor pinned to its confined parent."""
+        dir_fd = self._confined_parent_fd(path, config, logger)
+        if dir_fd is None:
+            return False
+        try:
+            # NEVER resolve-then-unlink: that deletes a symlink's TARGET. The
+            # name is resolved by the kernel against dir_fd, so it can't escape.
+            os.unlink(os.path.basename(path), dir_fd=dir_fd)
+            return True
+        except OSError as e:
+            logger.error(f"Failed to remove {path}: {e}")
+            return False
+        finally:
+            os.close(dir_fd)
+
+    def _rmtree_confined(
+        self, folder: str, config: Optional[ChubConfig], logger: Logger
+    ) -> bool:
+        """rmtree `folder` by name through a descriptor on its confined parent."""
+        dir_fd = self._confined_parent_fd(folder, config, logger)
+        if dir_fd is None:
+            return False
+        try:
+            shutil.rmtree(os.path.basename(folder), dir_fd=dir_fd)
+            return True
+        except OSError as e:
+            logger.error(f"Failed to remove {folder}: {e}")
+            return False
+        finally:
+            os.close(dir_fd)
+
+    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 = self._live_config(logger)
+        if config is None:
             return []
 
         authorized: List[str] = []
@@ -1113,6 +1199,7 @@ def _execute_stale_mode(
         count = 0
         total_size = 0
         touched: Set[str] = set()
+        config = self._live_config(logger)
         for d in dupes:
             folder = d["folder"]
             size = d.get("size", 0)
@@ -1130,8 +1217,14 @@ def _execute_stale_mode(
             if mode == "move":
                 dest_root = os.path.join(d["asset_dir"], ORPHAN_RESTORE_DIR_NAME)
                 dest = os.path.join(dest_root, d["name"])
+                dest_parent = os.path.dirname(dest)
+                if self._confined_target(folder, config, logger) is None:
+                    continue
                 try:
-                    os.makedirs(os.path.dirname(dest), exist_ok=True)
+                    os.makedirs(dest_parent, exist_ok=True)
+                    # The leaf can't exist yet — confine the parent makedirs just made.
+                    if self._confined_target(dest_parent, config, logger) is None:
+                        continue
                     shutil.move(folder, dest)
                     logger.info(f"  [STALE MOVED] {folder} -> {dest}")
                     count += 1
@@ -1140,15 +1233,13 @@ def _execute_stale_mode(
                 except OSError as e:
                     logger.error(f"Failed to move {folder}: {e}")
             elif mode == "remove":
-                try:
-                    shutil.rmtree(folder)
-                    logger.info(f"  [STALE REMOVED] {folder}")
-                    count += 1
-                    total_size += size
-                    touched.add(d["asset_dir"])
-                except OSError as e:
-                    logger.error(f"Failed to remove {folder}: {e}")
-        empty = sum(self._clean_empty_dirs(d) for d in touched)
+                if not self._rmtree_confined(folder, config, logger):
+                    continue
+                logger.info(f"  [STALE REMOVED] {folder}")
+                count += 1
+                total_size += size
+                touched.add(d["asset_dir"])
+        empty = sum(self._clean_empty_dirs(d, config) for d in touched)
         logger.info(
             f"   → stale duplicates: {count} {mode}d"
             + (f", {empty} empty dir(s) pruned" if empty else "")
@@ -1355,6 +1446,7 @@ def _execute_orphan_mode(
         count = 0
         total_size = 0
         touched_dirs: Set[str] = set()
+        config = self._live_config(logger)
 
         for item in orphans:
             path = item["path"]
@@ -1370,8 +1462,14 @@ def _execute_orphan_mode(
                 dest_root = os.path.join(item["asset_dir"], ORPHAN_RESTORE_DIR_NAME)
                 rel = os.path.relpath(path, item["asset_dir"])
                 dest = os.path.join(dest_root, rel)
+                dest_parent = os.path.dirname(dest)
+                if self._confined_target(path, config, logger) is None:
+                    continue
                 try:
-                    os.makedirs(os.path.dirname(dest), exist_ok=True)
+                    os.makedirs(dest_parent, exist_ok=True)
+                    # The leaf can't exist yet — confine the parent makedirs just made.
+                    if self._confined_target(dest_parent, config, logger) is None:
+                        continue
                     shutil.move(path, dest)
                     # Destructive ops stay at INFO deliberately — audit trail
                     # without debug mode (mirrors the bloat-cleanup pass).
@@ -1384,17 +1482,15 @@ def _execute_orphan_mode(
                 continue
 
             if mode == "remove":
-                try:
-                    os.remove(path)
-                    logger.info(f"  [REMOVED] {path}")
-                    count += 1
-                    total_size += size
-                    touched_dirs.add(item["asset_dir"])
-                except OSError as e:
-                    logger.error(f"Failed to remove {path}: {e}")
+                if not self._remove_confined(path, config, logger):
+                    continue
+                logger.info(f"  [REMOVED] {path}")
+                count += 1
+                total_size += size
+                touched_dirs.add(item["asset_dir"])
 
         # Prune empty dirs left behind by move/remove.
-        empty_dirs = sum(self._clean_empty_dirs(d) for d in touched_dirs)
+        empty_dirs = sum(self._clean_empty_dirs(d, config) for d in touched_dirs)
 
         logger.info(
             f"   → orphan scan: {count} {mode}d"
@@ -1406,16 +1502,25 @@ def _execute_orphan_mode(
     # Empty directory cleanup
     # =========================================================================
 
-    def _clean_empty_dirs(self, base_dir: str) -> int:
-        """Remove empty directories bottom-up."""
+    def _clean_empty_dirs(
+        self, base_dir: str, config: Optional[ChubConfig] = None
+    ) -> int:
+        """Remove empty directories bottom-up; with `config`, confine each dir first."""
         count = 0
         for root, dirs, files in os.walk(base_dir, topdown=False):
             for dir_name in dirs:
                 dir_path = os.path.join(root, dir_name)
                 try:
-                    if not os.listdir(dir_path):
-                        os.rmdir(dir_path)
-                        count += 1
+                    if os.listdir(dir_path):
+                        continue
+                    # Confine at the rmdir, not at the walk: a component swapped
+                    # to a symlink mid-sweep resolves outside the roots.
+                    if config is not None and (
+                        self._confined_target(dir_path, config, self.logger) is None
+                    ):
+                        continue
+                    os.rmdir(dir_path)
+                    count += 1
                 except OSError as e:
                     # Dir vanished, became non-empty, or permissions — skip,
                     # but leave a trace so the count discrepancy is explainable.
diff --git a/tests/test_poster_cleanarr_duplicates.py b/tests/test_poster_cleanarr_duplicates.py
index 97162287..b14c7fe3 100644
--- a/tests/test_poster_cleanarr_duplicates.py
+++ b/tests/test_poster_cleanarr_duplicates.py
@@ -10,7 +10,7 @@
 from backend.util.config import ChubConfig, ConfigError
 from backend.util.database import ChubDB
 from backend.util.path_safety import resolve_confined
-
+import os
 
 def _logger():
     return SimpleNamespace(
@@ -420,6 +420,94 @@ def _victim_folder(outside):
     return victim
 
 
+def test_execute_stale_remove_refuses_a_swapped_parent(tmp_path, monkeypatch):
+    """A parent component swapped to a symlink between scan and rmtree must not delete the tree it now points at."""
+    allowed = tmp_path / "allowed"
+    outside = tmp_path / "outside"
+    allowed.mkdir()
+    victim = _victim_folder(outside)
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+    entry = _swap_parent_for_link(allowed, outside)
+    logger, errors = _collecting_logger()
+
+    res = m._execute_stale_mode([entry], "remove", logger)
+
+    assert res["count"] == 0
+    assert victim.exists()  # the swapped-in target is untouched
+    assert errors
+
+
+def test_execute_stale_move_refuses_a_swapped_parent(tmp_path, monkeypatch):
+    """Same swap on the move path — shutil.move must not relocate an outside folder."""
+    allowed = tmp_path / "allowed"
+    outside = tmp_path / "outside"
+    allowed.mkdir()
+    victim = _victim_folder(outside)
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+    entry = _swap_parent_for_link(allowed, outside)
+    logger, errors = _collecting_logger()
+
+    res = m._execute_stale_mode([entry], "move", logger)
+
+    assert res["count"] == 0
+    assert victim.exists()  # not dragged into the restore dir
+    assert errors
+
+
+def test_execute_stale_remove_rmtrees_through_a_parent_descriptor(
+    tmp_path, monkeypatch
+):
+    """The recursive delete is anchored to a descriptor on the confined parent, so a parent re-pointed the instant after it is opened is refused."""
+    allowed = tmp_path / "allowed"
+    outside = tmp_path / "outside"
+    allowed.mkdir()
+    victim = _victim_folder(outside)
+    shows = allowed / "shows"
+    shows.mkdir()
+    stale = shows / "Dune - Prophecy (2024) {tvdb-1}"
+    stale.mkdir()
+    (stale / "poster.jpg").write_bytes(b"x")
+    (shows / "Dune Prophecy (2024) {tvdb-1}").mkdir()  # canonical present
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+
+    real_open = os.open
+
+    def _swapping_open(path, *a, **kw):
+        """Open normally, then re-point that same path at `outside`."""
+        fd = real_open(path, *a, **kw)
+        if (
+            str(path) == str(shows.resolve())
+            and shows.is_dir()
+            and not shows.is_symlink()
+        ):
+            shows.rename(allowed / "shows_real")
+            shows.symlink_to(outside, target_is_directory=True)
+        return fd
+
+    # `os` here is the same module object poster_cleanarr calls through.
+    monkeypatch.setattr(os, "open", _swapping_open)
+    logger, errors = _collecting_logger()
+    entry = {
+        "folder": str(stale),
+        "asset_dir": str(allowed),
+        "name": stale.name,
+        "canonical": "Dune Prophecy (2024) {tvdb-1}",
+        "canonical_present": True,
+        "id": ("tvdb", 1),
+        "size": 1,
+    }
+
+    res = m._execute_stale_mode([entry], "remove", logger)
+
+    assert res["count"] == 0
+    assert victim.exists()  # the swapped-in tree is never deleted through
+    assert (allowed / "shows_real" / stale.name).exists()  # nor the pinned one
+    assert errors
+
+
 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
@@ -479,3 +567,41 @@ def _sched_logger():
     m.run()
     assert "orphan" in calls
     assert "stale" in calls
+def test_stale_remove_passes_the_descriptor_to_rmtree(tmp_path, monkeypatch):
+    """The delete resolves its name against dir_fd, not a path the kernel re-walks."""
+    import shutil
+
+    allowed = tmp_path / "allowed"
+    shows = allowed / "shows"
+    stale = shows / "Dune - Prophecy (2024) {tvdb-1}"
+    stale.mkdir(parents=True)
+    (stale / "poster.jpg").write_bytes(b"x")
+    (shows / "Dune Prophecy (2024) {tvdb-1}").mkdir()
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+
+    seen = {}
+    real_rmtree = shutil.rmtree
+
+    def _recording_rmtree(path, *a, **kw):
+        seen["path"] = path
+        seen["dir_fd"] = kw.get("dir_fd")
+        return real_rmtree(path, *a, **kw)
+
+    monkeypatch.setattr(shutil, "rmtree", _recording_rmtree)
+    entry = {
+        "folder": str(stale),
+        "asset_dir": str(allowed),
+        "name": stale.name,
+        "canonical": "Dune Prophecy (2024) {tvdb-1}",
+        "canonical_present": True,
+        "id": ("tvdb", 1),
+        "size": 1,
+    }
+
+    res = m._execute_stale_mode([entry], "remove", _collecting_logger()[0])
+
+    assert res["count"] == 1
+    assert seen["dir_fd"] is not None  # anchored, not path-walked
+    assert seen["path"] == stale.name  # a bare name, resolved against the fd
+    assert not stale.exists()
diff --git a/tests/test_poster_cleanarr_orphans.py b/tests/test_poster_cleanarr_orphans.py
index ae55add2..b07b76bc 100644
--- a/tests/test_poster_cleanarr_orphans.py
+++ b/tests/test_poster_cleanarr_orphans.py
@@ -15,7 +15,7 @@
 from backend.util.database import ChubDB
 from backend.util.normalization import normalize_titles
 from backend.util.path_safety import resolve_confined
-
+import os
 
 def _logger():
     return SimpleNamespace(
@@ -249,6 +249,216 @@ def _swap_parent_for_link(allowed, outside):
     return item
 
 
+def test_execute_orphan_mode_remove_refuses_a_swapped_parent(tmp_path, monkeypatch):
+    """A parent component swapped to a symlink between scan and os.remove must not delete the file it now points at."""
+    allowed = tmp_path / "allowed"
+    outside = tmp_path / "outside"
+    allowed.mkdir()
+    outside.mkdir()
+    victim = outside / "poster.png"
+    victim.write_bytes(b"x")
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+    item = _swap_parent_for_link(allowed, outside)
+    logger, errors = _collecting_logger()
+
+    res = m._execute_orphan_mode([item], "remove", logger)
+
+    assert res["count"] == 0
+    assert victim.exists()  # the swapped-in target is untouched
+    assert errors
+
+
+def test_execute_orphan_mode_move_refuses_a_swapped_parent(tmp_path, monkeypatch):
+    """Same swap on the move path — shutil.move must not relocate an outside file."""
+    allowed = tmp_path / "allowed"
+    outside = tmp_path / "outside"
+    allowed.mkdir()
+    outside.mkdir()
+    victim = outside / "poster.png"
+    victim.write_bytes(b"x")
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+    item = _swap_parent_for_link(allowed, outside)
+    logger, errors = _collecting_logger()
+
+    res = m._execute_orphan_mode([item], "move", logger)
+
+    assert res["count"] == 0
+    assert victim.exists()  # not dragged into the restore dir
+    assert errors
+
+
+def test_execute_orphan_mode_move_refuses_a_symlinked_restore_dir(
+    tmp_path, monkeypatch
+):
+    """The move destination is confined too: a restore dir linking outside the roots must not receive the file."""
+    allowed = tmp_path / "allowed"
+    outside = tmp_path / "outside"
+    allowed.mkdir()
+    outside.mkdir()
+    (allowed / ORPHAN_RESTORE_DIR_NAME).symlink_to(outside, target_is_directory=True)
+    f = allowed / "orphan.png"
+    f.write_bytes(b"x")
+    item = {"path": str(f), "size": 1, "parsed": "orphan", "asset_dir": str(allowed)}
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+    logger, errors = _collecting_logger()
+
+    res = m._execute_orphan_mode([item], "move", logger)
+
+    assert res["count"] == 0
+    assert f.exists()  # left in place
+    assert not (outside / "orphan.png").exists()  # nothing written outside
+    assert errors
+
+
+def test_clean_empty_dirs_refuses_a_dir_outside_the_roots(tmp_path, monkeypatch):
+    """The empty-dir sweep re-confines each dir: an asset dir swapped to a link must not have the outside tree pruned through it."""
+    allowed = tmp_path / "allowed"
+    outside = tmp_path / "outside"
+    allowed.mkdir()
+    outside.mkdir()
+    stray = outside / "empty"
+    stray.mkdir()
+    cfg = _live(monkeypatch, allowed)
+    swapped = allowed / "assets"
+    swapped.symlink_to(outside, target_is_directory=True)
+    m = _make(allowed)
+    m.logger, errors = _collecting_logger()
+
+    assert m._clean_empty_dirs(str(swapped), cfg) == 0
+    assert stray.exists()  # pruning stops at the roots
+    assert errors
+
+
+# ── Descriptor-bound deletion (the race re-confinement only narrows) ─────────
+
+
+def _orphan_item(path, asset_dir):
+    """An orphan entry for `path`, as _scan_orphan_assets would emit it."""
+    return {
+        "path": str(path),
+        "size": 1,
+        "parsed": "orphan",
+        "asset_dir": str(asset_dir),
+    }
+
+
+def test_execute_orphan_mode_remove_unlinks_through_a_parent_descriptor(
+    tmp_path, monkeypatch
+):
+    """The delete goes through os.unlink(name, dir_fd=...) on the confined parent — a bare os.remove(path) would resolve the path a second time."""
+    allowed = tmp_path / "allowed"
+    allowed.mkdir()
+    f = allowed / "orphan.png"
+    f.write_bytes(b"x")
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+
+    calls = []
+    real_unlink = os.unlink
+
+    def _recording_unlink(path, *a, dir_fd=None, **kw):
+        """Record the name/descriptor pair, then delegate to the real unlink."""
+        calls.append((path, dir_fd))
+        return real_unlink(path, *a, dir_fd=dir_fd, **kw)
+
+    # `os` here is the same module object poster_cleanarr calls through.
+    monkeypatch.setattr(os, "unlink", _recording_unlink)
+    res = m._execute_orphan_mode([_orphan_item(f, allowed)], "remove", _logger())
+
+    assert res["count"] == 1
+    assert not f.exists()
+    assert [name for name, _fd in calls] == ["orphan.png"]  # the NAME, not the path
+    assert calls[0][1] is not None  # ...resolved against a descriptor
+
+
+def test_execute_orphan_mode_remove_refuses_a_parent_swapped_after_the_check(
+    tmp_path, monkeypatch
+):
+    """The parent is re-pointed at an outside dir the instant after it is opened: the descriptor pin must refuse rather than delete through the new path."""
+    allowed = tmp_path / "allowed"
+    outside = tmp_path / "outside"
+    allowed.mkdir()
+    outside.mkdir()
+    victim = outside / "poster.png"
+    victim.write_bytes(b"x")
+    show = allowed / "Show"
+    show.mkdir()
+    f = show / "poster.png"
+    f.write_bytes(b"x")
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+
+    real_open = os.open
+
+    def _swapping_open(path, *a, **kw):
+        """Open normally, then re-point that same path at `outside`."""
+        fd = real_open(path, *a, **kw)
+        if str(path) == str(show.resolve()) and show.is_dir() and not show.is_symlink():
+            show.rename(allowed / "Show_real")
+            show.symlink_to(outside, target_is_directory=True)
+        return fd
+
+    monkeypatch.setattr(os, "open", _swapping_open)
+    logger, errors = _collecting_logger()
+    res = m._execute_orphan_mode([_orphan_item(f, allowed)], "remove", logger)
+
+    assert res["count"] == 0
+    assert victim.exists()  # the swapped-in dir is never deleted through
+    assert (allowed / "Show_real" / "poster.png").exists()  # nor the pinned one
+    assert errors
+
+
+def test_execute_orphan_mode_remove_deletes_the_link_not_its_target(
+    tmp_path, monkeypatch
+):
+    """An orphaned symlink is unlinked by name; resolving it first would delete the artwork it points at."""
+    allowed = tmp_path / "allowed"
+    allowed.mkdir()
+    target = allowed / "keep.png"
+    target.write_bytes(b"x")
+    link = allowed / "orphan.png"
+    link.symlink_to(target)
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+
+    res = m._execute_orphan_mode([_orphan_item(link, allowed)], "remove", _logger())
+
+    assert res["count"] == 1
+    assert not link.is_symlink()  # the link entry is gone
+    assert target.exists()  # its target survives
+
+
+def test_execute_orphan_mode_remove_sweeps_dead_links_only_inside_the_roots(
+    tmp_path, monkeypatch
+):
+    """Behaviour change to keep visible: confinement resolves a link's TARGET, so the dead-link sweep still clears links into a root but now refuses links pointing at an unconfigured directory."""
+    allowed = tmp_path / "allowed"
+    outside = tmp_path / "outside"
+    allowed.mkdir()
+    outside.mkdir()
+    inside_link = allowed / "gone-inside.png"
+    inside_link.symlink_to(allowed / "missing.png")
+    outside_link = allowed / "gone-outside.png"
+    outside_link.symlink_to(outside / "missing.png")
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+    logger, errors = _collecting_logger()
+
+    res = m._execute_orphan_mode(
+        [_orphan_item(inside_link, allowed), _orphan_item(outside_link, allowed)],
+        "remove",
+        logger,
+    )
+
+    assert res["count"] == 1
+    assert not inside_link.is_symlink()  # dead link into a root: still swept
+    assert outside_link.is_symlink()  # dead link out of the roots: refused
+    assert errors
+
+
 def _write(tmp_path, name):
     f = tmp_path / name
     f.write_bytes(b"x")

From 892de0d9c90bd94070d56d09d22bf8bab7640582 Mon Sep 17 00:00:00 2001
From: chodeus <dhoges90@icloud.com>
Date: Fri, 14 Aug 2026 18:04:58 +0800
Subject: [PATCH 2/4] fix(security): open the delete parent one component at a
 time

O_NOFOLLOW guards only the last component and the inode pin re-resolves the
same string, so an intermediate symlink swapped in after confinement landed
the descriptor outside the roots. Walk down from the containing root
instead. Empty-dir sweep gains a base_dir floor for the bloat pass, which
passes no config. Refusals aggregate to one line per pass.
---
 backend/modules/poster_cleanarr.py       |  93 +++++++++++++++---
 backend/util/path_safety.py              |  14 +++
 tests/test_poster_cleanarr_duplicates.py |   5 +-
 tests/test_poster_cleanarr_orphans.py    | 115 ++++++++++++++++++++++-
 4 files changed, 208 insertions(+), 19 deletions(-)

diff --git a/backend/modules/poster_cleanarr.py b/backend/modules/poster_cleanarr.py
index a24e833f..4749089f 100644
--- a/backend/modules/poster_cleanarr.py
+++ b/backend/modules/poster_cleanarr.py
@@ -19,7 +19,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
+from backend.util.path_safety import containing_root, 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).
@@ -105,6 +105,10 @@ def _bundle_root(path: str) -> Optional[str]:
 
 
 class PosterCleanarr(ChubModule):
+    # Per-pass refusal tally behind _refuse/_report_refusals. Class level because
+    # shim callers build the module with object.__new__, skipping __init__.
+    _refused_paths = 0
+
     def __init__(self, logger: Optional[Logger] = None) -> None:
         super().__init__(logger=logger)
         self.plex_path: str = getattr(self.config, "plex_path", "")
@@ -864,18 +868,68 @@ def _live_config(self, logger: Logger) -> Optional[ChubConfig]:
             )
             return None
 
+    def _refuse(self, logger: Logger, message: str) -> None:
+        """Tally one refusal, keeping its per-path detail at debug."""
+        self._refused_paths += 1
+        logger.debug(message)
+
+    def _report_refusals(self, logger: Logger, what: str) -> None:
+        """Log one ERROR carrying the tallied refusals, then reset the tally."""
+        if self._refused_paths:
+            logger.error(
+                f"{what}: refused {self._refused_paths} path(s) resolving outside "
+                "the allowed roots or their base dir; per-path detail at debug level."
+            )
+        self._refused_paths = 0
+
     def _confined_target(
         self, path: str, config: Optional[ChubConfig], logger: Logger
     ) -> Optional[str]:
-        """Realpath of `path` if inside a live allowed root, else None (logged)."""
+        """Realpath of `path` if inside a live allowed root, else None (tallied)."""
         real = resolve_confined(path, config) if config is not None else None
         if real is None:
-            logger.error(
-                f"Target resolves outside the allowed roots, refusing to touch: {path}"
+            self._refuse(
+                logger,
+                f"Target resolves outside the allowed roots, refusing to touch: {path}",
             )
             return None
         return str(real)
 
+    def _walk_open_dir(
+        self, real_dir: str, config: ChubConfig, logger: Logger
+    ) -> Optional[int]:
+        """Descriptor on `real_dir`, opened one component at a time from its allowed root."""
+        # The root is the trust ANCHOR: its own ancestors are unverifiable from
+        # here, so confinement starts at it and every step below refuses a link.
+        root = containing_root(real_dir, config)
+        if root is None:
+            logger.error(f"Refusing to open {real_dir}: no allowed root contains it")
+            return None
+        flags = (
+            os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
+        )
+        try:
+            dir_fd = os.open(str(root), flags)
+        except OSError as e:
+            logger.error(f"Refusing to open {real_dir}: root {root} not openable ({e})")
+            return None
+        rel = os.path.relpath(real_dir, str(root))
+        for name in [c for c in rel.split(os.sep) if c and c != os.curdir]:
+            try:
+                # NEVER re-open the whole string: O_NOFOLLOW only guards the last
+                # component, so an intermediate one swapped to a link would escape.
+                nxt = os.open(name, flags, dir_fd=dir_fd)
+            except OSError as e:
+                logger.error(
+                    f"Refusing to open {real_dir}: '{name}' is not a plain "
+                    f"directory reached from {root} ({e})"
+                )
+                os.close(dir_fd)
+                return None
+            os.close(dir_fd)
+            dir_fd = nxt
+        return dir_fd
+
     def _confined_parent_fd(
         self, path: str, config: Optional[ChubConfig], logger: Logger
     ) -> Optional[int]:
@@ -885,19 +939,14 @@ def _confined_parent_fd(
         # Confine the PARENT, not the leaf: the leaf is what we delete, and
         # deleting it by name through this descriptor is what closes the race.
         parent = self._confined_target(os.path.dirname(path), config, logger)
-        if parent is None:
+        if parent is None or config is None:
             return None
-        flags = (
-            os.O_RDONLY | getattr(os, "O_DIRECTORY", 0) | getattr(os, "O_NOFOLLOW", 0)
-        )
-        try:
-            dir_fd = os.open(parent, flags)
-        except OSError as e:
-            logger.error(f"Refusing to delete {path}: parent not openable ({e})")
+        dir_fd = self._walk_open_dir(parent, config, logger)
+        if dir_fd is None:
             return None
         try:
-            # Mirrors maintenance.prune_old_backups: the descriptor must BE the
-            # directory that was authorized, not one swapped in after the check.
+            # Kept as a cheap SECONDARY check — the walk above is the barrier;
+            # this still catches the parent being renamed after it was opened.
             opened, authorized = os.fstat(dir_fd), os.stat(parent)
             if (opened.st_dev, opened.st_ino) != (authorized.st_dev, authorized.st_ino):
                 logger.error(f"Refusing to delete {path}: its parent changed under us")
@@ -1239,6 +1288,7 @@ def _execute_stale_mode(
                 count += 1
                 total_size += size
                 touched.add(d["asset_dir"])
+        self._report_refusals(logger, "Stale-duplicate cleanup")
         empty = sum(self._clean_empty_dirs(d, config) for d in touched)
         logger.info(
             f"   → stale duplicates: {count} {mode}d"
@@ -1489,6 +1539,7 @@ def _execute_orphan_mode(
                 total_size += size
                 touched_dirs.add(item["asset_dir"])
 
+        self._report_refusals(logger, "Orphan cleanup")
         # Prune empty dirs left behind by move/remove.
         empty_dirs = sum(self._clean_empty_dirs(d, config) for d in touched_dirs)
 
@@ -1505,14 +1556,25 @@ def _execute_orphan_mode(
     def _clean_empty_dirs(
         self, base_dir: str, config: Optional[ChubConfig] = None
     ) -> int:
-        """Remove empty directories bottom-up; with `config`, confine each dir first."""
+        """Remove empty dirs bottom-up, never escaping `base_dir`; `config` confines further."""
         count = 0
+        # Trailing sep so a sibling like `<base>_old` can't pass the prefix test.
+        floor = os.path.realpath(base_dir).rstrip(os.sep) + os.sep
         for root, dirs, files in os.walk(base_dir, topdown=False):
             for dir_name in dirs:
                 dir_path = os.path.join(root, dir_name)
                 try:
                     if os.listdir(dir_path):
                         continue
+                    # base_dir is the ONLY floor the bloat pass has (it passes no
+                    # config), so it is checked before the allowed-roots check.
+                    if not os.path.realpath(dir_path).startswith(floor):
+                        self._refuse(
+                            self.logger,
+                            f"Refusing to remove {dir_path}: it resolves outside "
+                            f"the tree being pruned ({base_dir})",
+                        )
+                        continue
                     # Confine at the rmdir, not at the walk: a component swapped
                     # to a symlink mid-sweep resolves outside the roots.
                     if config is not None and (
@@ -1525,6 +1587,7 @@ def _clean_empty_dirs(
                     # Dir vanished, became non-empty, or permissions — skip,
                     # but leave a trace so the count discrepancy is explainable.
                     self.logger.debug(f"Could not remove empty dir {dir_path}: {e}")
+        self._report_refusals(self.logger, "Empty-dir sweep")
         return count
 
     # =========================================================================
diff --git a/backend/util/path_safety.py b/backend/util/path_safety.py
index bf135497..86ed15fb 100644
--- a/backend/util/path_safety.py
+++ b/backend/util/path_safety.py
@@ -255,6 +255,20 @@ def is_path_allowed(path: str, config: ChubConfig) -> bool:
     return False
 
 
+def containing_root(path: str, config: ChubConfig) -> Optional[Path]:
+    """Longest allowed root containing the already-resolved *path*, or None."""
+    # Callers walk down from this root, so the LONGEST match is the tightest
+    # anchor. os.sep suffix stops `/root_evil` matching `/root`, as above.
+    best: Optional[Path] = None
+    for root in get_allowed_roots(config):
+        base = str(root)
+        if path != base and not path.startswith(base + os.sep):
+            continue
+        if best is None or len(base) > len(str(best)):
+            best = root
+    return best
+
+
 def resolve_confined(path: str, config: ChubConfig) -> Optional[Path]:
     """Resolve *path* and return it only when the resolved target is inside an allowed root, else None."""
     if not path or not isinstance(path, str):
diff --git a/tests/test_poster_cleanarr_duplicates.py b/tests/test_poster_cleanarr_duplicates.py
index b14c7fe3..284f43c4 100644
--- a/tests/test_poster_cleanarr_duplicates.py
+++ b/tests/test_poster_cleanarr_duplicates.py
@@ -476,10 +476,11 @@ def test_execute_stale_remove_rmtrees_through_a_parent_descriptor(
     real_open = os.open
 
     def _swapping_open(path, *a, **kw):
-        """Open normally, then re-point that same path at `outside`."""
+        """Open normally, then re-point that same dir at `outside`."""
         fd = real_open(path, *a, **kw)
+        # Match on the basename: the walk opens each component by NAME.
         if (
-            str(path) == str(shows.resolve())
+            os.path.basename(str(path)) == shows.name
             and shows.is_dir()
             and not shows.is_symlink()
         ):
diff --git a/tests/test_poster_cleanarr_orphans.py b/tests/test_poster_cleanarr_orphans.py
index b07b76bc..2988f64b 100644
--- a/tests/test_poster_cleanarr_orphans.py
+++ b/tests/test_poster_cleanarr_orphans.py
@@ -332,6 +332,47 @@ def test_clean_empty_dirs_refuses_a_dir_outside_the_roots(tmp_path, monkeypatch)
     assert errors
 
 
+def test_clean_empty_dirs_refuses_a_dir_outside_base_dir_without_a_config(
+    tmp_path, monkeypatch
+):
+    """The bloat pass prunes with no config at all, so base_dir is the only floor: a component swapped mid-sweep must not have rmdir reach past it."""
+    base = tmp_path / "metadata"
+    show = base / "Show"
+    (show / "empty").mkdir(parents=True)
+    outside = tmp_path / "outside"
+    (outside / "empty").mkdir(parents=True)
+    m = _make()
+    m.logger, errors = _collecting_logger()
+
+    real_listdir = os.listdir
+    swapped = []
+
+    def _swapping_listdir(path, *a, **kw):
+        """Swap `Show` for a link to `outside` between the walk and the rmdir."""
+        if not swapped and str(path) == str(show / "empty"):
+            swapped.append(True)
+            show.rename(base / "Show_real")
+            show.symlink_to(outside, target_is_directory=True)
+        return real_listdir(path, *a, **kw)
+
+    monkeypatch.setattr(os, "listdir", _swapping_listdir)
+
+    assert m._clean_empty_dirs(str(base)) == 0
+    assert swapped  # the race window was actually entered
+    assert (outside / "empty").is_dir()  # pruning stops at base_dir
+    assert errors
+
+
+def test_clean_empty_dirs_still_prunes_inside_base_dir_without_a_config(tmp_path):
+    """The floor must not disable the bloat pass, which passes no config."""
+    base = tmp_path / "metadata"
+    (base / "Show" / "empty").mkdir(parents=True)
+    m = _make()
+
+    assert m._clean_empty_dirs(str(base)) == 2
+    assert not (base / "Show").exists()  # pruned bottom-up, base kept
+
+
 # ── Descriptor-bound deletion (the race re-confinement only narrows) ─────────
 
 
@@ -394,9 +435,14 @@ def test_execute_orphan_mode_remove_refuses_a_parent_swapped_after_the_check(
     real_open = os.open
 
     def _swapping_open(path, *a, **kw):
-        """Open normally, then re-point that same path at `outside`."""
+        """Open normally, then re-point that same dir at `outside`."""
         fd = real_open(path, *a, **kw)
-        if str(path) == str(show.resolve()) and show.is_dir() and not show.is_symlink():
+        # Match on the basename: the walk opens each component by NAME.
+        if (
+            os.path.basename(str(path)) == show.name
+            and show.is_dir()
+            and not show.is_symlink()
+        ):
             show.rename(allowed / "Show_real")
             show.symlink_to(outside, target_is_directory=True)
         return fd
@@ -411,6 +457,46 @@ def _swapping_open(path, *a, **kw):
     assert errors
 
 
+def test_execute_orphan_mode_remove_refuses_an_intermediate_component_swapped_after_confinement(
+    tmp_path, monkeypatch
+):
+    """An INTERMEDIATE component becomes a link once confinement has passed."""
+    allowed = tmp_path / "allowed"
+    outside = tmp_path / "outside"
+    lib = allowed / "lib"
+    show = lib / "Show"
+    show.mkdir(parents=True)
+    f = show / "poster.png"
+    f.write_bytes(b"x")
+    victim = outside / "Show" / "poster.png"
+    victim.parent.mkdir(parents=True)
+    victim.write_bytes(b"x")
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+
+    real_open = os.open
+    swapped = []
+
+    def _swapping_open(path, *a, **kw):
+        """Swap `lib` for a link to `outside` in the check→open window."""
+        if not swapped and str(path).startswith(os.path.realpath(allowed)):
+            swapped.append(True)
+            lib.rename(allowed / "lib_real")
+            lib.symlink_to(outside, target_is_directory=True)
+        return real_open(path, *a, **kw)
+
+    monkeypatch.setattr(os, "open", _swapping_open)
+    logger, errors = _collecting_logger()
+
+    res = m._execute_orphan_mode([_orphan_item(f, allowed)], "remove", logger)
+
+    assert swapped  # the race window was actually entered
+    assert res["count"] == 0
+    assert victim.exists()  # the swapped-in tree is never deleted through
+    assert (allowed / "lib_real" / "Show" / "poster.png").exists()  # nor the real one
+    assert errors
+
+
 def test_execute_orphan_mode_remove_deletes_the_link_not_its_target(
     tmp_path, monkeypatch
 ):
@@ -459,6 +545,31 @@ def test_execute_orphan_mode_remove_sweeps_dead_links_only_inside_the_roots(
     assert errors
 
 
+def test_execute_orphan_mode_aggregates_refusals_into_one_error(tmp_path, monkeypatch):
+    """Links into unconfigured storage are retained and re-refused every pass, so the refusals collapse to one summary line; detail moves to debug."""
+    allowed = tmp_path / "allowed"
+    outside = tmp_path / "outside"
+    allowed.mkdir()
+    outside.mkdir()
+    items = []
+    for i in range(5):
+        link = allowed / f"gone-{i}.png"
+        link.symlink_to(outside / f"missing-{i}.png")
+        items.append(_orphan_item(link, allowed))
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+    logger, errors = _collecting_logger()
+    debugs = []
+    logger.debug = debugs.append
+
+    res = m._execute_orphan_mode(items, "remove", logger)
+
+    assert res["count"] == 0
+    assert len(errors) == 1  # one line per pass, not one per link
+    assert "refused 5 path(s)" in errors[0]
+    assert len(debugs) == 5  # per-link detail retained at debug
+
+
 def _write(tmp_path, name):
     f = tmp_path / name
     f.write_bytes(b"x")

From 1881f5b8dde3c07f3d7071bda3933d645475079d Mon Sep 17 00:00:00 2001
From: chodeus <dhoges90@icloud.com>
Date: Fri, 14 Aug 2026 19:12:53 +0800
Subject: [PATCH 3/4] fix(security): anchor moves and rmdir, reload config per
 item

os.rename takes src_dir_fd/dst_dir_fd, so same-filesystem moves are now
anchored at both ends; shutil.move survives only as the cross-device
fallback, the one path that still re-resolves strings. rmdir gets the same
descriptor treatment when a config is available. Both executors re-read
config per destructive item instead of trusting a pre-loop snapshot.
Makefile now gates at 3.11, which is what rmtree(dir_fd=) needs.
---
 Makefile                                 |   4 +-
 backend/modules/poster_cleanarr.py       | 155 ++++++++++++++----
 tests/test_poster_cleanarr_duplicates.py |  80 ++++++++++
 tests/test_poster_cleanarr_orphans.py    | 195 +++++++++++++++++++++++
 4 files changed, 397 insertions(+), 37 deletions(-)

diff --git a/Makefile b/Makefile
index ad15bad1..522d5b59 100755
--- a/Makefile
+++ b/Makefile
@@ -22,8 +22,8 @@ bootstrap: install ui-install ## Setup everything
 install: ## Install backend dependencies
 	@echo "Installing backend..."
 	@test -d $(VENV) || $(PY) -m venv $(VENV)
-	@$(VENV)/bin/python -c 'import sys; sys.exit(sys.version_info < (3, 10))' || \
-		{ echo "ERROR: $(VENV) is $$($(VENV)/bin/python -V); requirements-dev.txt needs Python >= 3.10 (repo targets 3.14). Recreate the venv with a newer PY=."; exit 1; }
+	@$(VENV)/bin/python -c 'import sys; sys.exit(sys.version_info < (3, 11))' || \
+		{ echo "ERROR: $(VENV) is $$($(VENV)/bin/python -V); this repo needs Python >= 3.11 (shutil.rmtree(dir_fd=) is 3.11+); CI and Docker build on 3.14. Recreate the venv with a newer PY=."; exit 1; }
 	@$(VENV)/bin/python -m pip install --upgrade pip
 	@$(VENV)/bin/pip install -r requirements.txt
 	@$(VENV)/bin/pip install -r requirements-dev.txt
diff --git a/backend/modules/poster_cleanarr.py b/backend/modules/poster_cleanarr.py
index 4749089f..11a931b8 100644
--- a/backend/modules/poster_cleanarr.py
+++ b/backend/modules/poster_cleanarr.py
@@ -1,5 +1,6 @@
 # modules/poster_cleanarr.py
 
+import errno
 import glob
 import json
 import os
@@ -936,8 +937,8 @@ def _confined_parent_fd(
         """Descriptor on `path`'s confined parent, or None (logged); caller closes."""
         if self._confined_target(path, config, logger) is None:
             return None
-        # Confine the PARENT, not the leaf: the leaf is what we delete, and
-        # deleting it by name through this descriptor is what closes the race.
+        # Confine the PARENT, not the leaf: the leaf is what we act on, and
+        # naming it through this descriptor is what closes the race.
         parent = self._confined_target(os.path.dirname(path), config, logger)
         if parent is None or config is None:
             return None
@@ -949,11 +950,11 @@ def _confined_parent_fd(
             # this still catches the parent being renamed after it was opened.
             opened, authorized = os.fstat(dir_fd), os.stat(parent)
             if (opened.st_dev, opened.st_ino) != (authorized.st_dev, authorized.st_ino):
-                logger.error(f"Refusing to delete {path}: its parent changed under us")
+                logger.error(f"Refusing to touch {path}: its parent changed under us")
                 os.close(dir_fd)
                 return None
         except OSError as e:
-            logger.error(f"Refusing to delete {path}: parent unverifiable ({e})")
+            logger.error(f"Refusing to touch {path}: parent unverifiable ({e})")
             os.close(dir_fd)
             return None
         return dir_fd
@@ -992,6 +993,69 @@ def _rmtree_confined(
         finally:
             os.close(dir_fd)
 
+    def _rmdir_confined(self, dir_path: str, config: ChubConfig) -> bool:
+        """rmdir `dir_path` by name through a descriptor on its confined parent."""
+        dir_fd = self._confined_parent_fd(dir_path, config, self.logger)
+        if dir_fd is None:
+            return False
+        try:
+            # OSError propagates: the sweep's own handler explains the count gap.
+            os.rmdir(os.path.basename(dir_path), dir_fd=dir_fd)
+            return True
+        finally:
+            os.close(dir_fd)
+
+    def _anchored_rename(
+        self, src: str, dest: str, config: Optional[ChubConfig], logger: Logger
+    ) -> Optional[bool]:
+        """rename `src` to `dest` through both confined parents; None = cross-device."""
+        src_fd = self._confined_parent_fd(src, config, logger)
+        if src_fd is None:
+            return False
+        try:
+            dest_parent = self._confined_target(os.path.dirname(dest), config, logger)
+            if dest_parent is None or config is None:
+                return False
+            dest_fd = self._walk_open_dir(dest_parent, config, logger)
+            if dest_fd is None:
+                return False
+            try:
+                # Both ends resolved by the kernel against a pinned descriptor,
+                # so neither parent can be swapped for a link after the check.
+                os.rename(
+                    os.path.basename(src),
+                    os.path.basename(dest),
+                    src_dir_fd=src_fd,
+                    dst_dir_fd=dest_fd,
+                )
+                return True
+            except OSError as e:
+                if e.errno == errno.EXDEV:
+                    return None
+                logger.error(f"Failed to move {src}: {e}")
+                return False
+            finally:
+                os.close(dest_fd)
+        finally:
+            os.close(src_fd)
+
+    def _move_confined(
+        self, src: str, dest: str, config: Optional[ChubConfig], logger: Logger
+    ) -> bool:
+        """Move `src` to `dest`, anchored to both parents; copies only across devices."""
+        renamed = self._anchored_rename(src, dest, config, logger)
+        if renamed is not None:
+            return renamed
+        # Cross-device has no anchored primitive: this copy+unlink re-resolves
+        # both paths, the one case the descriptor pin cannot cover.
+        logger.debug(f"Anchored rename unavailable across devices: {src} -> {dest}")
+        try:
+            shutil.move(src, dest)
+            return True
+        except OSError as e:
+            logger.error(f"Failed to move {src}: {e}")
+            return False
+
     def _authorized_asset_dirs(
         self, asset_dirs: List[str], logger: Logger
     ) -> List[str]:
@@ -1248,7 +1312,6 @@ def _execute_stale_mode(
         count = 0
         total_size = 0
         touched: Set[str] = set()
-        config = self._live_config(logger)
         for d in dupes:
             folder = d["folder"]
             size = d.get("size", 0)
@@ -1263,24 +1326,28 @@ def _execute_stale_mode(
                     "not staged yet; keeping the only copy"
                 )
                 continue
+            # Re-read per item, not once per batch: a root dropped from config
+            # mid-run must stop the REST of the batch. load_config is mtime-cached.
+            config = self._live_config(logger)
+            if config is None:
+                self._refuse(logger, f"No live config, refusing to {mode}: {folder}")
+                continue
             if mode == "move":
                 dest_root = os.path.join(d["asset_dir"], ORPHAN_RESTORE_DIR_NAME)
                 dest = os.path.join(dest_root, d["name"])
-                dest_parent = os.path.dirname(dest)
                 if self._confined_target(folder, config, logger) is None:
                     continue
                 try:
-                    os.makedirs(dest_parent, exist_ok=True)
-                    # The leaf can't exist yet — confine the parent makedirs just made.
-                    if self._confined_target(dest_parent, config, logger) is None:
-                        continue
-                    shutil.move(folder, dest)
-                    logger.info(f"  [STALE MOVED] {folder} -> {dest}")
-                    count += 1
-                    total_size += size
-                    touched.add(d["asset_dir"])
+                    os.makedirs(os.path.dirname(dest), exist_ok=True)
                 except OSError as e:
                     logger.error(f"Failed to move {folder}: {e}")
+                    continue
+                if not self._move_confined(folder, dest, config, logger):
+                    continue
+                logger.info(f"  [STALE MOVED] {folder} -> {dest}")
+                count += 1
+                total_size += size
+                touched.add(d["asset_dir"])
             elif mode == "remove":
                 if not self._rmtree_confined(folder, config, logger):
                     continue
@@ -1289,7 +1356,13 @@ def _execute_stale_mode(
                 total_size += size
                 touched.add(d["asset_dir"])
         self._report_refusals(logger, "Stale-duplicate cleanup")
-        empty = sum(self._clean_empty_dirs(d, config) for d in touched)
+        empty = 0
+        if touched:
+            # Re-read before the sweep too, and skip it entirely on None: the
+            # weaker base_dir-only floor is for the bloat pass, not for this one.
+            sweep_config = self._live_config(logger)
+            if sweep_config is not None:
+                empty = sum(self._clean_empty_dirs(d, sweep_config) for d in touched)
         logger.info(
             f"   → stale duplicates: {count} {mode}d"
             + (f", {empty} empty dir(s) pruned" if empty else "")
@@ -1496,7 +1569,6 @@ def _execute_orphan_mode(
         count = 0
         total_size = 0
         touched_dirs: Set[str] = set()
-        config = self._live_config(logger)
 
         for item in orphans:
             path = item["path"]
@@ -1508,27 +1580,32 @@ def _execute_orphan_mode(
                 total_size += size
                 continue
 
+            # Re-read per item, not once per batch: a root dropped from config
+            # mid-run must stop the REST of the batch. load_config is mtime-cached.
+            config = self._live_config(logger)
+            if config is None:
+                self._refuse(logger, f"No live config, refusing to {mode}: {path}")
+                continue
+
             if mode == "move":
                 dest_root = os.path.join(item["asset_dir"], ORPHAN_RESTORE_DIR_NAME)
                 rel = os.path.relpath(path, item["asset_dir"])
                 dest = os.path.join(dest_root, rel)
-                dest_parent = os.path.dirname(dest)
                 if self._confined_target(path, config, logger) is None:
                     continue
                 try:
-                    os.makedirs(dest_parent, exist_ok=True)
-                    # The leaf can't exist yet — confine the parent makedirs just made.
-                    if self._confined_target(dest_parent, config, logger) is None:
-                        continue
-                    shutil.move(path, dest)
-                    # Destructive ops stay at INFO deliberately — audit trail
-                    # without debug mode (mirrors the bloat-cleanup pass).
-                    logger.info(f"  [MOVED] {path} -> {dest}")
-                    count += 1
-                    total_size += size
-                    touched_dirs.add(item["asset_dir"])
+                    os.makedirs(os.path.dirname(dest), exist_ok=True)
                 except OSError as e:
                     logger.error(f"Failed to move {path}: {e}")
+                    continue
+                if not self._move_confined(path, dest, config, logger):
+                    continue
+                # Destructive ops stay at INFO deliberately — audit trail
+                # without debug mode (mirrors the bloat-cleanup pass).
+                logger.info(f"  [MOVED] {path} -> {dest}")
+                count += 1
+                total_size += size
+                touched_dirs.add(item["asset_dir"])
                 continue
 
             if mode == "remove":
@@ -1540,8 +1617,15 @@ def _execute_orphan_mode(
                 touched_dirs.add(item["asset_dir"])
 
         self._report_refusals(logger, "Orphan cleanup")
-        # Prune empty dirs left behind by move/remove.
-        empty_dirs = sum(self._clean_empty_dirs(d, config) for d in touched_dirs)
+        # Prune empty dirs left behind by move/remove, on a re-read config: the
+        # loop above can run for minutes, and None must skip the sweep entirely.
+        empty_dirs = 0
+        if touched_dirs:
+            sweep_config = self._live_config(logger)
+            if sweep_config is not None:
+                empty_dirs = sum(
+                    self._clean_empty_dirs(d, sweep_config) for d in touched_dirs
+                )
 
         logger.info(
             f"   → orphan scan: {count} {mode}d"
@@ -1577,11 +1661,12 @@ def _clean_empty_dirs(
                         continue
                     # Confine at the rmdir, not at the walk: a component swapped
                     # to a symlink mid-sweep resolves outside the roots.
-                    if config is not None and (
-                        self._confined_target(dir_path, config, self.logger) is None
-                    ):
+                    if config is None:
+                        # The bloat pass passes no config, so it has no allowed
+                        # root to anchor against — the floor above is its guard.
+                        os.rmdir(dir_path)
+                    elif not self._rmdir_confined(dir_path, config):
                         continue
-                    os.rmdir(dir_path)
                     count += 1
                 except OSError as e:
                     # Dir vanished, became non-empty, or permissions — skip,
diff --git a/tests/test_poster_cleanarr_duplicates.py b/tests/test_poster_cleanarr_duplicates.py
index 284f43c4..7d266d26 100644
--- a/tests/test_poster_cleanarr_duplicates.py
+++ b/tests/test_poster_cleanarr_duplicates.py
@@ -606,3 +606,83 @@ def _recording_rmtree(path, *a, **kw):
     assert seen["dir_fd"] is not None  # anchored, not path-walked
     assert seen["path"] == stale.name  # a bare name, resolved against the fd
     assert not stale.exists()
+
+
+def _stale_entry(folder, asset_dir, canonical):
+    """A stale-duplicate entry for `folder`, as _scan_stale_duplicates emits it."""
+    return {
+        "folder": str(folder),
+        "asset_dir": str(asset_dir),
+        "name": os.path.basename(str(folder)),
+        "canonical": canonical,
+        "canonical_present": True,
+        "id": ("tvdb", 1),
+        "size": 1,
+    }
+
+
+def test_execute_stale_move_renames_through_both_parent_descriptors(
+    tmp_path, monkeypatch
+):
+    """The stale move is an os.rename resolved against a descriptor on each parent, not a path pair shutil re-walks."""
+    allowed = tmp_path / "allowed"
+    canonical = "Dune Prophecy (2024) {tvdb-1}"
+    (allowed / canonical).mkdir(parents=True)  # canonical present
+    stale = allowed / "Dune - Prophecy (2024) {tvdb-1}"
+    stale.mkdir()
+    (stale / "poster.jpg").write_bytes(b"x")
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+
+    calls = []
+    real_rename = os.rename
+
+    def _recording_rename(src, dst, **kw):
+        calls.append((src, dst, kw.get("src_dir_fd"), kw.get("dst_dir_fd")))
+        return real_rename(src, dst, **kw)
+
+    monkeypatch.setattr(os, "rename", _recording_rename)
+
+    res = m._execute_stale_mode(
+        [_stale_entry(stale, allowed, canonical)], "move", _logger()
+    )
+
+    assert res["count"] == 1
+    assert (allowed / ORPHAN_RESTORE_DIR_NAME / stale.name / "poster.jpg").exists()
+    assert len(calls) == 1
+    src, dst, src_fd, dst_fd = calls[0]
+    assert (src, dst) == (stale.name, stale.name)  # bare names, not paths
+    assert src_fd is not None and dst_fd is not None  # ...against both fds
+
+
+def test_execute_stale_mode_stops_when_config_stops_authorizing(tmp_path, monkeypatch):
+    """Config is re-read per item, so a root dropped mid-batch stops the tail while the head — authorized when its turn came — still ran."""
+    allowed = tmp_path / "allowed"
+    entries = []
+    for tag in ("A", "B"):
+        (allowed / f"Show {tag} (2024) {{tvdb-1}}").mkdir(parents=True)
+        old = allowed / f"Show {tag} old (2024) {{tvdb-1}}"
+        old.mkdir()
+        (old / "poster.jpg").write_bytes(b"x")
+        entries.append(_stale_entry(old, allowed, f"Show {tag} (2024) {{tvdb-1}}"))
+    m = _make(allowed)
+    calls = []
+
+    def _load():
+        """Authorize `allowed` for the first load only, nothing afterwards."""
+        calls.append(1)
+        cfg = ChubConfig()
+        if len(calls) == 1:
+            cfg.poster_renamerr.source_dirs = [str(allowed)]
+        return cfg
+
+    monkeypatch.setattr("backend.util.config.load_config", _load)
+    logger, errors = _collecting_logger()
+
+    res = m._execute_stale_mode(entries, "remove", logger)
+
+    assert res["count"] == 1
+    assert not os.path.exists(entries[0]["folder"])
+    assert os.path.exists(entries[1]["folder"])  # de-authorized before its turn
+    assert len(calls) >= 2  # one load per item, not one per batch
+    assert errors
diff --git a/tests/test_poster_cleanarr_orphans.py b/tests/test_poster_cleanarr_orphans.py
index 2988f64b..b278c48c 100644
--- a/tests/test_poster_cleanarr_orphans.py
+++ b/tests/test_poster_cleanarr_orphans.py
@@ -15,6 +15,7 @@
 from backend.util.database import ChubDB
 from backend.util.normalization import normalize_titles
 from backend.util.path_safety import resolve_confined
+import errno
 import os
 
 def _logger():
@@ -739,6 +740,200 @@ def test_resolve_orphan_instances_handles_missing_attrs():
     assert PosterCleanarr._resolve_orphan_instances(cfg) == []
 
 
+# ── Anchored moves, anchored rmdir, per-item authorization ──────────────────
+
+
+def test_execute_orphan_mode_move_renames_through_both_parent_descriptors(
+    tmp_path, monkeypatch
+):
+    """The move is an os.rename resolved against a descriptor on each parent, not a path pair shutil re-walks."""
+    allowed = tmp_path / "allowed"
+    allowed.mkdir()
+    f = allowed / "orphan.png"
+    f.write_bytes(b"x")
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+
+    calls = []
+    real_rename = os.rename
+
+    def _recording_rename(src, dst, **kw):
+        calls.append((src, dst, kw.get("src_dir_fd"), kw.get("dst_dir_fd")))
+        return real_rename(src, dst, **kw)
+
+    monkeypatch.setattr(os, "rename", _recording_rename)
+
+    res = m._execute_orphan_mode([_orphan_item(f, allowed)], "move", _logger())
+
+    assert res["count"] == 1
+    assert (allowed / ORPHAN_RESTORE_DIR_NAME / "orphan.png").exists()
+    assert len(calls) == 1
+    src, dst, src_fd, dst_fd = calls[0]
+    assert (src, dst) == ("orphan.png", "orphan.png")  # bare names, not paths
+    assert src_fd is not None and dst_fd is not None  # ...against both fds
+
+
+def test_execute_orphan_mode_move_refuses_a_parent_swapped_after_the_check(
+    tmp_path, monkeypatch
+):
+    """The source parent becomes a link to an outside dir the instant after confinement passes: the move must refuse, not relocate the outside file."""
+    allowed = tmp_path / "allowed"
+    outside = tmp_path / "outside"
+    allowed.mkdir()
+    outside.mkdir()
+    victim = outside / "poster.png"
+    victim.write_bytes(b"x")
+    show = allowed / "Show"
+    show.mkdir()
+    f = show / "poster.png"
+    f.write_bytes(b"x")
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+    logger, errors = _collecting_logger()
+
+    real_confined = m._confined_target
+    swapped = []
+
+    def _swapping_confined(path, config, log):
+        """Swap `Show` for a link to `outside` once the file itself is confined."""
+        result = real_confined(path, config, log)
+        if not swapped and str(path) == str(f):
+            swapped.append(True)
+            show.rename(allowed / "Show_real")
+            show.symlink_to(outside, target_is_directory=True)
+        return result
+
+    monkeypatch.setattr(m, "_confined_target", _swapping_confined)
+
+    res = m._execute_orphan_mode([_orphan_item(f, allowed)], "move", logger)
+
+    assert swapped  # the race window was actually entered
+    assert res["count"] == 0
+    assert victim.exists()  # never dragged into the restore dir
+    assert (allowed / "Show_real" / "poster.png").exists()  # nor the pinned one
+    assert errors
+
+
+def test_execute_orphan_mode_move_falls_back_across_devices(tmp_path, monkeypatch):
+    """EXDEV is the one case the anchored rename can't serve: it must fall back to the copy path, not drop the move."""
+    allowed = tmp_path / "allowed"
+    allowed.mkdir()
+    f = allowed / "orphan.png"
+    f.write_bytes(b"x")
+    m = _make(allowed)
+    _live(monkeypatch, allowed)
+
+    def _cross_device(*a, **kw):
+        raise OSError(errno.EXDEV, "Invalid cross-device link")
+
+    # shutil.move's own os.rename is patched too, so it takes its copy branch.
+    monkeypatch.setattr(os, "rename", _cross_device)
+    logger, errors = _collecting_logger()
+
+    res = m._execute_orphan_mode([_orphan_item(f, allowed)], "move", logger)
+
+    assert res["count"] == 1
+    assert not f.exists()
+    assert (allowed / ORPHAN_RESTORE_DIR_NAME / "orphan.png").exists()
+    assert not errors  # a cross-device move is normal, not a refusal
+
+
+def test_clean_empty_dirs_rmdirs_through_a_parent_descriptor(tmp_path, monkeypatch):
+    """With a config the prune goes through os.rmdir(name, dir_fd=...) — a bare os.rmdir(path) would re-walk every component."""
+    allowed = tmp_path / "allowed"
+    (allowed / "Show" / "empty").mkdir(parents=True)
+    cfg = _live(monkeypatch, allowed)
+    m = _make(allowed)
+    m.logger, errors = _collecting_logger()
+
+    calls = []
+    real_rmdir = os.rmdir
+
+    def _recording_rmdir(path, *a, dir_fd=None, **kw):
+        calls.append((path, dir_fd))
+        return real_rmdir(path, *a, dir_fd=dir_fd, **kw)
+
+    monkeypatch.setattr(os, "rmdir", _recording_rmdir)
+
+    assert m._clean_empty_dirs(str(allowed), cfg) == 2
+    assert not (allowed / "Show").exists()  # pruned bottom-up, root kept
+    assert [name for name, _fd in calls] == ["empty", "Show"]  # names, not paths
+    assert all(fd is not None for _name, fd in calls)
+    assert not errors
+
+
+def test_clean_empty_dirs_refuses_a_parent_swapped_after_the_check(
+    tmp_path, monkeypatch
+):
+    """A parent swapped to a link in the confine→rmdir window must not have the outside tree pruned through it."""
+    allowed = tmp_path / "allowed"
+    outside = tmp_path / "outside"
+    (outside / "empty").mkdir(parents=True)
+    base = allowed / "assets"
+    show = base / "Show"
+    (show / "empty").mkdir(parents=True)
+    cfg = _live(monkeypatch, allowed)
+    m = _make(allowed)
+    m.logger, errors = _collecting_logger()
+
+    real_confined = m._confined_target
+    swapped = []
+
+    def _swapping_confined(path, config, log):
+        """Swap `Show` for a link to `outside` once its child dir is confined."""
+        result = real_confined(path, config, log)
+        if not swapped and str(path) == str(show / "empty"):
+            swapped.append(True)
+            show.rename(base / "Show_real")
+            show.symlink_to(outside, target_is_directory=True)
+        return result
+
+    monkeypatch.setattr(m, "_confined_target", _swapping_confined)
+
+    assert m._clean_empty_dirs(str(base), cfg) == 0
+    assert swapped  # the race window was actually entered
+    assert (outside / "empty").is_dir()  # pruning never reached through the link
+    assert errors
+
+
+def _tightening_config(monkeypatch, allowed, authorized_calls=1):
+    """load_config authorizes `allowed` for the first N calls, then nothing."""
+    calls = []
+
+    def _load():
+        calls.append(1)
+        cfg = ChubConfig()
+        if len(calls) <= authorized_calls:
+            cfg.poster_renamerr.source_dirs = [str(allowed)]
+        return cfg
+
+    monkeypatch.setattr("backend.util.config.load_config", _load)
+    return calls
+
+
+def test_execute_orphan_mode_stops_when_config_stops_authorizing(tmp_path, monkeypatch):
+    """Config is re-read per item, so a root dropped mid-batch stops the tail while the head — authorized when its turn came — still ran."""
+    allowed = tmp_path / "allowed"
+    allowed.mkdir()
+    head = allowed / "head.png"
+    head.write_bytes(b"x")
+    tail = allowed / "tail.png"
+    tail.write_bytes(b"x")
+    m = _make(allowed)
+    calls = _tightening_config(monkeypatch, allowed)
+    logger, errors = _collecting_logger()
+
+    res = m._execute_orphan_mode(
+        [_orphan_item(head, allowed), _orphan_item(tail, allowed)], "remove", logger
+    )
+
+    assert res["count"] == 1
+    assert not head.exists()
+    assert tail.exists()  # de-authorized before its turn
+    assert len(calls) >= 2  # one load per item, not one per batch
+    assert errors
+
+
 def test_build_library_id_sets_filters_by_instance(db):
     m = _make()
     db.media.execute_query(

From 156706a426897ee936d7ef172ff5a16306e23154 Mon Sep 17 00:00:00 2001
From: chodeus <dhoges90@icloud.com>
Date: Sat, 15 Aug 2026 04:03:59 +0800
Subject: [PATCH 4/4] test: bind de-authorization asserts to the refusing pass

Share the tightening-config stub as a conftest fixture. assert errors also
passed on an empty-dir-sweep refusal; key on the pass label and count.
---
 tests/conftest.py                        | 20 ++++++++++++++++++++
 tests/test_poster_cleanarr_duplicates.py | 23 +++++++++--------------
 tests/test_poster_cleanarr_orphans.py    | 24 ++++++------------------
 3 files changed, 35 insertions(+), 32 deletions(-)

diff --git a/tests/conftest.py b/tests/conftest.py
index 4aed5753..63dcd438 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -74,6 +74,26 @@ def config_with_roots(tmp_path):
     return config, tmp_path
 
 
+@pytest.fixture
+def tightening_config(monkeypatch):
+    """Factory: patch load_config to authorize `allowed` for the first N calls, then nothing."""
+
+    def _install(allowed, authorized_calls=1):
+        calls = []
+
+        def _load():
+            calls.append(1)
+            config = ChubConfig()
+            if len(calls) <= authorized_calls:
+                config.poster_renamerr.source_dirs = [str(allowed)]
+            return config
+
+        monkeypatch.setattr("backend.util.config.load_config", _load)
+        return calls
+
+    return _install
+
+
 @pytest.fixture
 def instances_config():
     return InstancesConfig(
diff --git a/tests/test_poster_cleanarr_duplicates.py b/tests/test_poster_cleanarr_duplicates.py
index 7d266d26..7b429bc9 100644
--- a/tests/test_poster_cleanarr_duplicates.py
+++ b/tests/test_poster_cleanarr_duplicates.py
@@ -2,6 +2,7 @@
 whose {tvdb/tmdb} id matches a live media item but whose name != the item's
 canonical folder (media_cache.folder). Safety: never remove the only copy."""
 
+import os
 from types import SimpleNamespace
 
 import pytest
@@ -10,7 +11,6 @@
 from backend.util.config import ChubConfig, ConfigError
 from backend.util.database import ChubDB
 from backend.util.path_safety import resolve_confined
-import os
 
 def _logger():
     return SimpleNamespace(
@@ -655,7 +655,9 @@ def _recording_rename(src, dst, **kw):
     assert src_fd is not None and dst_fd is not None  # ...against both fds
 
 
-def test_execute_stale_mode_stops_when_config_stops_authorizing(tmp_path, monkeypatch):
+def test_execute_stale_mode_stops_when_config_stops_authorizing(
+    tmp_path, tightening_config
+):
     """Config is re-read per item, so a root dropped mid-batch stops the tail while the head — authorized when its turn came — still ran."""
     allowed = tmp_path / "allowed"
     entries = []
@@ -666,17 +668,7 @@ def test_execute_stale_mode_stops_when_config_stops_authorizing(tmp_path, monkey
         (old / "poster.jpg").write_bytes(b"x")
         entries.append(_stale_entry(old, allowed, f"Show {tag} (2024) {{tvdb-1}}"))
     m = _make(allowed)
-    calls = []
-
-    def _load():
-        """Authorize `allowed` for the first load only, nothing afterwards."""
-        calls.append(1)
-        cfg = ChubConfig()
-        if len(calls) == 1:
-            cfg.poster_renamerr.source_dirs = [str(allowed)]
-        return cfg
-
-    monkeypatch.setattr("backend.util.config.load_config", _load)
+    calls = tightening_config(allowed, authorized_calls=1)
     logger, errors = _collecting_logger()
 
     res = m._execute_stale_mode(entries, "remove", logger)
@@ -685,4 +677,7 @@ def _load():
     assert not os.path.exists(entries[0]["folder"])
     assert os.path.exists(entries[1]["folder"])  # de-authorized before its turn
     assert len(calls) >= 2  # one load per item, not one per batch
-    assert errors
+    # Labelled by pass, so an empty-dir-sweep refusal cannot satisfy it.
+    assert [
+        "Stale-duplicate cleanup: refused 1 path(s)" in e for e in errors
+    ].count(True) == 1
diff --git a/tests/test_poster_cleanarr_orphans.py b/tests/test_poster_cleanarr_orphans.py
index b278c48c..b9a4596e 100644
--- a/tests/test_poster_cleanarr_orphans.py
+++ b/tests/test_poster_cleanarr_orphans.py
@@ -896,22 +896,9 @@ def _swapping_confined(path, config, log):
     assert errors
 
 
-def _tightening_config(monkeypatch, allowed, authorized_calls=1):
-    """load_config authorizes `allowed` for the first N calls, then nothing."""
-    calls = []
-
-    def _load():
-        calls.append(1)
-        cfg = ChubConfig()
-        if len(calls) <= authorized_calls:
-            cfg.poster_renamerr.source_dirs = [str(allowed)]
-        return cfg
-
-    monkeypatch.setattr("backend.util.config.load_config", _load)
-    return calls
-
-
-def test_execute_orphan_mode_stops_when_config_stops_authorizing(tmp_path, monkeypatch):
+def test_execute_orphan_mode_stops_when_config_stops_authorizing(
+    tmp_path, tightening_config
+):
     """Config is re-read per item, so a root dropped mid-batch stops the tail while the head — authorized when its turn came — still ran."""
     allowed = tmp_path / "allowed"
     allowed.mkdir()
@@ -920,7 +907,7 @@ def test_execute_orphan_mode_stops_when_config_stops_authorizing(tmp_path, monke
     tail = allowed / "tail.png"
     tail.write_bytes(b"x")
     m = _make(allowed)
-    calls = _tightening_config(monkeypatch, allowed)
+    calls = tightening_config(allowed)
     logger, errors = _collecting_logger()
 
     res = m._execute_orphan_mode(
@@ -931,7 +918,8 @@ def test_execute_orphan_mode_stops_when_config_stops_authorizing(tmp_path, monke
     assert not head.exists()
     assert tail.exists()  # de-authorized before its turn
     assert len(calls) >= 2  # one load per item, not one per batch
-    assert errors
+    # Labelled by pass, so an empty-dir-sweep refusal cannot satisfy it.
+    assert ["Orphan cleanup: refused 1 path(s)" in e for e in errors].count(True) == 1
 
 
 def test_build_library_id_sets_filters_by_instance(db):