Guard probe_target against creating phantom local dirs when mount_root is unmounted

coe_mirror.py

@@ -81,6 +81,8 @@ PENDING_DIR: Path = Path(
         str(Path.home() / ".local" / "share" / "coe-mirror" / "pending"),
     )
 ).expanduser()
+_MOUNT_ROOT_VAL = _resolve("COE_MIRROR_MOUNT_ROOT", _CFG.get("mount_root"), "")
+MOUNT_ROOT: Optional[Path] = Path(_MOUNT_ROOT_VAL).expanduser() if _MOUNT_ROOT_VAL else None
 LOG_DIR = Path.home() / ".local" / "share" / "coe-mirror"
 LOG_FILE = LOG_DIR / "coe-mirror.log"
 LOCK_NAME = "coe-mirror.lock"
@@ -168,6 +170,14 @@ def release_lock(lock_fd: int) -> None:
 def probe_target() -> bool:
     """Probe target directory health. Write tiny file, read back, delete. Return True if healthy."""
     try:
+        if MOUNT_ROOT is not None:
+            if not MOUNT_ROOT.exists() or not os.path.ismount(str(MOUNT_ROOT)):
+                logger.warning("mount_root %s is not a live mountpoint; target unavailable; queuing locally", MOUNT_ROOT)
+                return False
+            if not (TARGET_DIR == MOUNT_ROOT or TARGET_DIR.is_relative_to(MOUNT_ROOT)):
+                logger.warning("target_dir %s is not under configured mount_root %s; misconfig; queuing locally", TARGET_DIR, MOUNT_ROOT)
+                return False
+
         probe_file = TARGET_DIR / ".write_probe"
         TARGET_DIR.mkdir(parents=True, exist_ok=True)
 
@@ -673,6 +683,11 @@ api_base = "https://git.churchofmalware.org"
 # Where mirrored zips are stored. One subdirectory per owner is created here.
 target_dir = "~/coe-mirror"
 
+# Optional: require this path to be a live mountpoint before writing. If set and
+# not currently mounted, the run queues to pending instead of creating a phantom
+# local directory at the mountpoint. Leave blank to disable.
+# mount_root = "~/pCloudDrive"
+
 # Optional: local staging when target_dir is unavailable.
 # pending_dir = "~/.local/share/coe-mirror/pending"
 

tests/test_coe_mirror.py

@@ -595,6 +595,51 @@ class TestPlaceIntoPloudSameFileIsNoop:
         assert test_file.stat().st_size > 0
 
 
+class TestMountRootGuard:
+    """Test probe_target() mount-root verification (phantom-dir regression guard)."""
+
+    def test_no_mount_root_configured_unchanged_behavior(self, mock_paths, setup_logging, monkeypatch):
+        """MOUNT_ROOT None + writable TARGET_DIR -> True (unchanged behavior)."""
+        monkeypatch.setattr(coe_mirror, "MOUNT_ROOT", None)
+
+        assert coe_mirror.probe_target() is True
+
+    def test_mount_root_live_target_under_it_returns_true(self, mock_paths, setup_logging, monkeypatch, tmp_path):
+        """MOUNT_ROOT set, live mount, TARGET_DIR a subdir of it -> True."""
+        mount_root = tmp_path / "pCloudDrive"
+        mount_root.mkdir()
+        target_dir = mount_root / "Hacking" / "ChurchOfMalware"
+        monkeypatch.setattr(coe_mirror, "MOUNT_ROOT", mount_root)
+        monkeypatch.setattr(coe_mirror, "TARGET_DIR", target_dir)
+        monkeypatch.setattr(coe_mirror.os.path, "ismount", lambda p: True)
+
+        assert coe_mirror.probe_target() is True
+
+    def test_mount_not_live_does_not_create_phantom_dir(self, mock_paths, setup_logging, monkeypatch, tmp_path):
+        """MOUNT_ROOT set but not mounted -> False, and TARGET_DIR is never created."""
+        mount_root = tmp_path / "pCloudDrive"
+        mount_root.mkdir()
+        target_dir = mount_root / "Hacking" / "ChurchOfMalware"
+        monkeypatch.setattr(coe_mirror, "MOUNT_ROOT", mount_root)
+        monkeypatch.setattr(coe_mirror, "TARGET_DIR", target_dir)
+        monkeypatch.setattr(coe_mirror.os.path, "ismount", lambda p: False)
+
+        assert coe_mirror.probe_target() is False
+        assert not target_dir.exists()
+
+    def test_target_dir_not_under_mount_root_returns_false(self, mock_paths, setup_logging, monkeypatch, tmp_path):
+        """MOUNT_ROOT set, live mount, but TARGET_DIR is unrelated -> False, no dir created."""
+        mount_root = tmp_path / "pCloudDrive"
+        mount_root.mkdir()
+        target_dir = tmp_path / "unrelated" / "target"
+        monkeypatch.setattr(coe_mirror, "MOUNT_ROOT", mount_root)
+        monkeypatch.setattr(coe_mirror, "TARGET_DIR", target_dir)
+        monkeypatch.setattr(coe_mirror.os.path, "ismount", lambda p: True)
+
+        assert coe_mirror.probe_target() is False
+        assert not target_dir.exists()
+
+
 # Tests for Matrix notification implementation.
 # Note: Tests requiring monkeypatched subprocess.run must use raising=False because
 # subprocess.run is not a class attribute on the coe_mirror module by default