fix(slskd): refuse unsafe peer paths, keep queued downloads alive, report search outages (#1512)

Fixes four defects in the Soulseek (slskd)

A peer names its own files, and Shelfmark works out where slskd stored
one from the last folder and the file name. `remote_directory_name`
returned the whole directory when its last segment was empty. So
`../../../etc//passwd.epub` or `/abs/path//book.epub` resolved outside
the downloads folder, and Shelfmark imported whatever was there. For a
multi-file release it moved those files into staging (`STAGE_MOVE`).

- New `safe_remote_segments` refuses a name with an empty, `.` or `..`
segment (which covers absolute and UNC paths), a NUL, or a drive
anywhere but a Windows share root (`C:\Books\…`, which never reaches a
local path).
- Search (`iter_peer_files`, so `build_releases`) drops such files.
- `download` refuses them before anything is queued, in both modes. The
spec arrives with the queued release payload, so search filtering alone
is not enough.
- `local_path_for` also refuses a path that resolves outside the root
(`resolve()` + `is_relative_to`), for example through a symlinked
folder.

`_poll`, used when isolation is off and as the fallback for slskd <
0.26, never asked for an activity grace. A file "Queued, Remotely" sent
the same status and progress on every poll, so the orchestrator
cancelled it after 300s and `SLSKD_QUEUE_TIMEOUT_MINUTES` never applied.
It now requests, renews and releases the grace the same way isolated
mode does.

`search` caught every exception per query, so slskd being down or
rejecting the API key showed as "no releases". When no query gets an
answer it now raises `SourceUnavailableError`, as Prowlarr does, and the
error reaches the user. One failed query among answered ones is still
not an outage.

A retry reuses the task's folder, and slskd renames a new file when one
is already there. Files from a failed attempt therefore meant
`_folder_complete`'s exact count could never match again. Now:
- a failed or cancelled attempt removes its folder;
- a new attempt removes any finished files an earlier one left (for
example after a restart) before it queues the batch.

`tests/slskd/conftest.py` stubbed `flask_socketio` whenever it had not
been imported yet, even though it is installed. Running `pytest
tests/slskd tests/` therefore broke the Flask app tests in the same
worker (285 errors, also on 10710ab). It now stubs only when the module
is missing.

- 25 new regression tests. All of them fail on 10710ab.
- `make python-checks` passes.
- `uv run pytest tests/slskd tests/ -m "not integration and not e2e"`
passes: 4047 tests on e83bab9.

shelfmark/release_sources/slskd/handler.py

@@ -50,7 +50,7 @@ from shelfmark.release_sources.slskd.api import (
 from shelfmark.release_sources.slskd.source import (
     SOURCE_NAME,
     build_client_from_config,
-    remote_directory_name,
+    safe_remote_segments,
     split_remote_path,
 )
 
@@ -73,7 +73,7 @@ PATH_MAPPING_HOST = "slskd"
 CANCEL_REMOVE_PASSES = 3
 CANCEL_REMOVE_INTERVAL = 0.5
 _BYTES_PER_MB = 1024 * 1024
-# Isolated mode: while a peer holds the files in its queue nothing changes, so the orchestrator's
+# While a peer holds the files in its queue nothing changes, so the orchestrator's
 # stall timer would cancel the download. Ask for a grace instead, renewed while still queued
 # (the orchestrator caps a single grace), and let the peer queue timeout decide when to give up.
 QUEUE_GRACE_SECONDS = 900.0
@@ -124,6 +124,7 @@ class _TransferSnapshot:
     failed_filename: str | None = None
     in_progress: int = 0
     queued_remotely: int = 0
+    queued: int = 0  # requested or queued, locally or at the peer
     queue_positions: list[int] = field(default_factory=list)
 
 
@@ -200,14 +201,23 @@ def local_path_for(root: Path, remote_filename: str) -> Path | None:
     """Compute where slskd stored ``remote_filename`` under ``root``.
 
     slskd keeps only the last remote folder name, so ``@@a\\Books\\Author\\x.epub``
-    lands in ``<root>/Author/x.epub``. Names that would escape ``root`` are refused.
+    lands in ``<root>/Author/x.epub``. A peer chooses that name, so one that is unsafe
+    (see ``safe_remote_segments``) or that resolves outside ``root`` is refused.
     """
-    directory, basename = split_remote_path(remote_filename)
-    folder = remote_directory_name(directory)
-    parts = [p for p in (folder, basename) if p]
-    if not basename or any(p in {".", ".."} for p in parts):
+    segments = safe_remote_segments(remote_filename)
+    if segments is None:
         return None
-    return root.joinpath(*parts)
+    path = root.joinpath(*segments[-2:])
+    try:
+        inside = run_blocking_io(path.resolve).is_relative_to(run_blocking_io(root.resolve))
+    except OSError, RuntimeError, ValueError:
+        return None
+    return path if inside else None
+
+
+def _is_queued(state: object) -> bool:
+    text = str(state or "").lower()
+    return "queued" in text or "requested" in text
 
 
 def _snapshot(transfers: list[dict[str, Any]], spec: SlskdDownloadSpec) -> _TransferSnapshot:
@@ -239,7 +249,10 @@ def _snapshot(transfers: list[dict[str, Any]], spec: SlskdDownloadSpec) -> _Tran
         if state.startswith("InProgress"):
             snap.in_progress += 1
             snap.speed += float(transfer.get("averageSpeed") or 0)
-        elif "Remotely" in state:
+            continue
+        if _is_queued(state):
+            snap.queued += 1
+        if "Remotely" in state:
             snap.queued_remotely += 1
             position = transfer.get("placeInQueue")
             if isinstance(position, int) and position > 0:
@@ -335,6 +348,23 @@ def _folder_complete(folder: Path, expected_sizes: list[int]) -> bool:
     return actual == sorted(expected_sizes)
 
 
+def _remove_files(paths: list[Path]) -> None:
+    for path in paths:
+        try:
+            run_blocking_io(path.unlink, missing_ok=True)
+        except OSError as e:
+            logger.debug("Could not remove slskd file %s: %s", path, e)
+
+
+def _discard_folder(folder: Path) -> None:
+    """Remove a download's own folder (isolated mode) and whatever arrived in it."""
+    try:
+        if run_blocking_io(folder.is_dir):
+            run_blocking_io(lambda: shutil.rmtree(folder))
+    except OSError as e:
+        logger.debug("Could not remove slskd folder %s: %s", folder, e)
+
+
 def _completed_remote_files(events: list[dict[str, Any]]) -> set[str]:
     """Remote filenames slskd's stored events say finished downloading."""
     done: set[str] = set()
@@ -432,6 +462,12 @@ class SlskdHandler(DownloadHandler):
             status_callback("error", "Soulseek file details are missing for this release")
             logger.warning("slskd task %s has no download spec", task.task_id)
             return None
+        unsafe = next((n for n in spec.filenames if safe_remote_segments(n) is None), None)
+        if unsafe is not None:
+            # Search drops these, but the spec arrives with the queued release payload.
+            status_callback("error", f"Refusing unsafe Soulseek file name: {unsafe}")
+            logger.warning("slskd task %s has an unsafe file name: %r", task.task_id, unsafe)
+            return None
 
         client = self._get_client()
         if client is None:
@@ -520,72 +556,91 @@ class SlskdHandler(DownloadHandler):
     ) -> _TransferSnapshot | None:
         poll_interval = self._poll_interval()
         queue_timeout = self._queue_timeout_seconds()
-        started = time.monotonic()
+        started = self._clock()
+        grace_at: float | None = None
         missing_polls = 0
         total = spec.total_size
 
-        while not cancel_flag.is_set():
-            try:
-                transfers = client.list_downloads(spec.username)
-            except _SLSKD_ERRORS as e:
-                logger.warning("slskd status check failed: %s", e)
-                status_callback("resolving", "Waiting for slskd...")
-                if cancel_flag.wait(timeout=poll_interval):
-                    break
-                continue
+        try:
+            while not cancel_flag.is_set():
+                try:
+                    transfers = client.list_downloads(spec.username)
+                except _SLSKD_ERRORS as e:
+                    logger.warning("slskd status check failed: %s", e)
+                    status_callback("resolving", "Waiting for slskd...")
+                    if cancel_flag.wait(timeout=poll_interval):
+                        break
+                    continue
 
-            snap = _snapshot(transfers, spec)
-            if snap.found < len(spec.files):
-                missing_polls += 1
-                if missing_polls >= MAX_MISSING_POLLS:
-                    status_callback("error", "Transfer disappeared from slskd")
-                    # Part of a multi-file release may still be listed and downloading.
+                snap = _snapshot(transfers, spec)
+                if snap.found < len(spec.files):
+                    missing_polls += 1
+                    if missing_polls >= MAX_MISSING_POLLS:
+                        status_callback("error", "Transfer disappeared from slskd")
+                        # Part of a multi-file release may still be listed and downloading.
+                        self._cancel_transfers(client, spec, remove=True)
+                        return None
+                    status_callback("resolving", "Waiting for slskd...")
+                    if cancel_flag.wait(timeout=poll_interval):
+                        break
+                    continue
+                missing_polls = 0
+
+                if snap.failed_state is not None:
+                    outcome = snap.failed_state.split(",", 1)[-1].strip() or snap.failed_state
+                    message = f"Peer transfer failed: {outcome}"
+                    if len(spec.files) > 1 and snap.failed_filename:
+                        message += f" ({split_remote_path(snap.failed_filename)[1]})"
+                    logger.error(
+                        "slskd transfer failed for %s: %s", task.task_id, snap.failed_state
+                    )
+                    status_callback("error", message)
                     self._cancel_transfers(client, spec, remove=True)
                     return None
-                status_callback("resolving", "Waiting for slskd...")
+
+                percent = (snap.bytes_transferred / total * 100) if total > 0 else 0.0
+                percent = max(0.0, min(100.0, percent))
+                progress_callback(percent)
+
+                if snap.all_succeeded:
+                    progress_callback(100.0)
+                    return snap
+
+                # Same as isolated mode: a peer queue repeats the same status, so ask for a
+                # grace and let the queue timeout below decide when to give up.
+                now = self._clock()
+                if snap.queued and (
+                    grace_at is None or now - grace_at >= QUEUE_GRACE_RENEW_SECONDS
+                ):
+                    request_activity_grace(status_callback, QUEUE_GRACE_SECONDS)
+                    grace_at = now
+                elif not snap.queued and grace_at is not None:
+                    release_activity_grace(status_callback)
+                    grace_at = None
+
+                if (
+                    queue_timeout > 0
+                    and snap.bytes_transferred == 0
+                    and now - started > queue_timeout
+                ):
+                    minutes = int(queue_timeout // _SECONDS_PER_MINUTE)
+                    status_callback(
+                        "error", f"Peer did not start the transfer within {minutes} minutes"
+                    )
+                    self._cancel_transfers(client, spec, remove=True)
+                    return None
+
+                status_callback("downloading", _progress_message(snap, spec, percent))
                 if cancel_flag.wait(timeout=poll_interval):
                     break
-                continue
-            missing_polls = 0
 
-            if snap.failed_state is not None:
-                outcome = snap.failed_state.split(",", 1)[-1].strip() or snap.failed_state
-                message = f"Peer transfer failed: {outcome}"
-                if len(spec.files) > 1 and snap.failed_filename:
-                    message += f" ({split_remote_path(snap.failed_filename)[1]})"
-                logger.error("slskd transfer failed for %s: %s", task.task_id, snap.failed_state)
-                status_callback("error", message)
-                self._cancel_transfers(client, spec, remove=True)
-                return None
-
-            percent = (snap.bytes_transferred / total * 100) if total > 0 else 0.0
-            percent = max(0.0, min(100.0, percent))
-            progress_callback(percent)
-
-            if snap.all_succeeded:
-                progress_callback(100.0)
-                return snap
-
-            if (
-                queue_timeout > 0
-                and snap.bytes_transferred == 0
-                and time.monotonic() - started > queue_timeout
-            ):
-                minutes = int(queue_timeout // _SECONDS_PER_MINUTE)
-                status_callback(
-                    "error", f"Peer did not start the transfer within {minutes} minutes"
-                )
-                self._cancel_transfers(client, spec, remove=True)
-                return None
-
-            status_callback("downloading", _progress_message(snap, spec, percent))
-            if cancel_flag.wait(timeout=poll_interval):
-                break
-
-        # Cancelled
-        self._cancel_transfers(client, spec, remove=True)
-        status_callback("cancelled", "Cancelled")
-        return None
+            # Cancelled
+            self._cancel_transfers(client, spec, remove=True)
+            status_callback("cancelled", "Cancelled")
+            return None
+        finally:
+            if grace_at is not None:
+                release_activity_grace(status_callback)
 
     def _current_transfer_ids(self, client: SlskdClient, spec: SlskdDownloadSpec) -> list[str]:
         try:
@@ -735,6 +790,13 @@ class SlskdHandler(DownloadHandler):
             logger.info("slskd: %s already holds the files for %s", folder, task.task_id)
             progress_callback(100.0)
             return self._finish_isolated(task, client, spec.username, [], folder)
+        # An earlier attempt's files (a restart, or a cleanup that failed). They are finished,
+        # since slskd moves a file in only once it is complete, so slskd will fetch them again
+        # and rename the new copies; left here, the folder could never match.
+        leftovers = _folder_files(folder)
+        if leftovers:
+            logger.info("slskd: removing %d file(s) left in %s", len(leftovers), folder)
+            _remove_files(leftovers)
 
         status_callback("resolving", "Sending to slskd")
         try:
@@ -758,12 +820,13 @@ class SlskdHandler(DownloadHandler):
                 first = unresolved[0].get("message") or "unknown reason"
                 status_callback("error", f"slskd refused {len(unresolved)} file(s): {first}")
                 self._cancel_ids(client, spec.username, list(ids.values()))
+                _discard_folder(folder)
                 return None
         if not ids:
             status_callback("error", "slskd did not queue any of the files")
             return None
 
-        return self._poll_isolated(
+        result = self._poll_isolated(
             task,
             spec,
             client,
@@ -774,6 +837,11 @@ class SlskdHandler(DownloadHandler):
             progress_callback,
             status_callback,
         )
+        if result is None:
+            # Failed or cancelled, and the transfers are stopped. A retry reuses this folder,
+            # so the files that did arrive must not be left to break it.
+            _discard_folder(folder)
+        return result
 
     def _rejoin_refused(
         self,
@@ -869,7 +937,7 @@ class SlskdHandler(DownloadHandler):
                 ) + sum(sizes.get(n, 0) for n in gone)
                 percent = max(0.0, min(100.0, (done_bytes / total * 100) if total > 0 else 0.0))
 
-                queued = [r for r in active.values() if self._is_queued(r.get("state"))]
+                queued = [r for r in active.values() if _is_queued(r.get("state"))]
                 if queued and (grace_at is None or now - grace_at >= QUEUE_GRACE_RENEW_SECONDS):
                     request_activity_grace(status_callback, QUEUE_GRACE_SECONDS)
                     grace_at = now
@@ -921,11 +989,6 @@ class SlskdHandler(DownloadHandler):
             if grace_at is not None:
                 release_activity_grace(status_callback)
 
-    @staticmethod
-    def _is_queued(state: object) -> bool:
-        text = str(state or "").lower()
-        return "queued" in text or "requested" in text
-
     @staticmethod
     def _isolated_message(
         active: dict[str, dict[str, Any]], spec: SlskdDownloadSpec, percent: float
@@ -1065,11 +1128,7 @@ class SlskdHandler(DownloadHandler):
                 client.cancel_transfer(username, transfer_id, remove=True)
             except _SLSKD_ERRORS as e:
                 logger.debug("Failed to remove slskd transfer %s: %s", transfer_id, e)
-        try:
-            if run_blocking_io(folder.is_dir):
-                run_blocking_io(lambda: shutil.rmtree(folder))
-        except OSError as e:
-            logger.debug("Could not remove slskd folder %s: %s", folder, e)
+        _discard_folder(folder)
 
     def cancel(self, task_id: str) -> bool:
         """Cancellation is driven by the queue's cancel flag inside ``download``."""

shelfmark/release_sources/slskd/source.py

@@ -34,6 +34,7 @@ from shelfmark.release_sources import (
     ReleaseSource,
     SortOption,
     SourceActionButton,
+    SourceUnavailableError,
     register_source,
 )
 from shelfmark.release_sources.prowlarr.source import _parse_size
@@ -82,8 +83,33 @@ def split_remote_path(filename: str) -> tuple[str, str]:
 
 
 def remote_directory_name(directory: str) -> str:
-    """Return the last segment of a remote directory path."""
-    return split_remote_path(directory)[1] or directory
+    """Return the last non-empty segment of a remote directory path."""
+    return next((part for part in reversed(re.split(r"[\\/]", directory)) if part), "")
+
+
+_DRIVE = re.compile(r"^[A-Za-z]:")
+
+
+def safe_remote_segments(filename: str) -> list[str] | None:
+    """The segments of a peer's file path, or None when the path is unsafe to lay out locally.
+
+    slskd turns the last folder and the file name into a path under its downloads folder, and
+    a peer controls both. A path with an empty, ``.`` or ``..`` segment is refused, which
+    covers one starting at ``/`` or ``\\`` and an empty last folder (``a//b.epub``). A drive
+    (``C:``) is allowed only as the first of three or more segments: Windows peers share from
+    there, and slskd keeps just the last folder, so it never reaches a local path.
+    """
+    text = str(filename or "")
+    if "\0" in text:
+        return None
+    segments = re.split(r"[\\/]", text)
+    if any(not part or part in {".", ".."} for part in segments):
+        return None
+    if any(_DRIVE.match(part) for part in segments[1:]) or (
+        len(segments) < 3 and _DRIVE.match(segments[0])  # the drive would be the local folder
+    ):
+        return None
+    return segments
 
 
 def file_extension(basename: str) -> str:
@@ -149,9 +175,10 @@ def iter_peer_files(response: dict[str, Any]) -> list[PeerFile]:
         filename = str(raw.get("filename") or "").strip()
         if not filename:
             continue
-        directory, basename = split_remote_path(filename)
-        if not basename:
+        if safe_remote_segments(filename) is None:
+            logger.debug("slskd: skipping unsafe file name from %s: %r", username, filename)
             continue
+        directory, basename = split_remote_path(filename)
         peer_files.append(
             PeerFile(
                 username=username,
@@ -502,6 +529,7 @@ class SlskdSource(ReleaseSource):
 
         releases: list[Release] = []
         answered: list[list[dict[str, Any]]] = []
+        last_error: Exception | None = None
         # Expanding keeps going through every variant even after one produced results.
         for index, query in enumerate(queries, start=1):
             if releases and not expand_search:
@@ -516,8 +544,9 @@ class SlskdSource(ReleaseSource):
                     wait_seconds=wait_seconds,
                     response_limit=response_limit,
                 )
-            except Exception:
+            except Exception as e:
                 logger.exception("slskd search failed for '%s'", query)
+                last_error = e
                 continue
 
             answered.append(responses)
@@ -525,6 +554,11 @@ class SlskdSource(ReleaseSource):
                 merge_responses(answered), content_type=content_type, formats=formats
             )
 
+        if not answered and last_error is not None:
+            # slskd being down or rejecting the API key is not "no releases found".
+            msg = f"search failed: {last_error}"
+            raise SourceUnavailableError(msg) from last_error
+
         releases.sort(key=rank_key)
         if releases:
             peers = len({r.indexer for r in releases})

tests/download/test_output_ignored_files.py

@@ -0,0 +1,91 @@
+"""Files the user skipped in the pack review are not emailed or uploaded to BookLore."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from threading import Event
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from shelfmark.core.models import DownloadTask
+
+EMAIL = "shelfmark.download.outputs.email"
+BOOKLORE = "shelfmark.download.outputs.booklore"
+
+
+def _pack(tmp_path: Path) -> tuple[Path, Path, Path]:
+    root = tmp_path / "Series Pack"
+    kept = root / "Book 1" / "one.epub"
+    skipped = root / "Book 2" / "two.epub"
+    for book in (kept, skipped):
+        book.parent.mkdir(parents=True)
+        book.write_bytes(b"epub")
+    return root, kept, skipped
+
+
+def _task(ignored: list[str]) -> DownloadTask:
+    return DownloadTask(
+        task_id="pack",
+        source="prowlarr",
+        title="Series Pack",
+        output_args={"to": "reader@example.com"},
+        ignored_files=ignored,
+    )
+
+
+def _send_by_email(root: Path, task: DownloadTask) -> tuple[list[Path] | None, list[tuple]]:
+    from shelfmark.download.outputs.email import process_email_output
+
+    sent: list[Path] = []
+    statuses: list[tuple] = []
+    with (
+        patch(f"{EMAIL}._get_email_settings", return_value={}),
+        patch(f"{EMAIL}.build_email_smtp_config", return_value=MagicMock()),
+        patch(
+            f"{EMAIL}.compose_email_message",
+            side_effect=lambda _config, **kw: sent.extend(kw["files"]),
+        ),
+        patch(f"{EMAIL}.send_email_message") as send,
+    ):
+        process_email_output(root, task, Event(), lambda *s: statuses.append(s))
+    return (sent if send.called else None), statuses
+
+
+def _upload_to_booklore(root: Path, task: DownloadTask) -> tuple[list[Path] | None, list[tuple]]:
+    from shelfmark.download.outputs.booklore import process_booklore_output
+
+    statuses: list[tuple] = []
+    with (
+        patch(f"{BOOKLORE}._get_booklore_settings", return_value={}),
+        patch(
+            f"{BOOKLORE}.build_booklore_config",
+            return_value=MagicMock(refresh_after_upload=False),
+        ),
+        patch(f"{BOOKLORE}.booklore_login", return_value="token"),
+        patch(f"{BOOKLORE}.booklore_upload_file") as upload,
+    ):
+        process_booklore_output(root, task, Event(), lambda *s: statuses.append(s))
+    uploaded = [c.args[2] for c in upload.call_args_list]
+    return (uploaded or None), statuses
+
+
+@pytest.mark.parametrize("deliver", [_send_by_email, _upload_to_booklore])
+def test_a_skipped_book_is_not_delivered(deliver, tmp_path: Path) -> None:
+    root, kept, _skipped = _pack(tmp_path)
+
+    delivered, _statuses = deliver(root, _task(["Series Pack/Book 2/two.epub"]))
+
+    assert delivered == [kept]
+
+
+@pytest.mark.parametrize("deliver", [_send_by_email, _upload_to_booklore])
+def test_nothing_is_delivered_when_every_file_was_skipped(deliver, tmp_path: Path) -> None:
+    root, _kept, _skipped = _pack(tmp_path)
+
+    delivered, statuses = deliver(
+        root, _task(["Series Pack/Book 1/one.epub", "Series Pack/Book 2/two.epub"])
+    )
+
+    assert delivered is None
+    assert ("error", "No book files found") in statuses

tests/slskd/conftest.py

@@ -6,13 +6,18 @@ in lightweight dev/CI environments. These are only needed by the IRC source,
 which is unrelated to the slskd plugin under test.
 """
 
+import importlib.util
 import sys
 import types
 
 
 def _stub_module(name: str) -> None:
-    """Insert a minimal stub module into sys.modules if not already present."""
-    if name not in sys.modules:
+    """Insert a minimal stub module into sys.modules if the real one is not installed.
+
+    A stub over an installed module would break every later test in the same worker that
+    needs the real one (the Flask app imports ``flask_socketio.emit``).
+    """
+    if name not in sys.modules and importlib.util.find_spec(name) is None:
         stub = types.ModuleType(name)
         # Provide no-op stand-ins for the symbols IRC source imports.
         stub.SocketIO = object

tests/slskd/test_handler.py

@@ -7,6 +7,7 @@ from unittest.mock import MagicMock
 import pytest
 
 from shelfmark.core.models import DownloadTask
+from shelfmark.download.activity import ACTIVITY_GRACE_STATUS
 from shelfmark.release_sources import get_handler
 from shelfmark.release_sources.slskd.api import SlskdError
 from shelfmark.release_sources.slskd.handler import (
@@ -47,17 +48,26 @@ class ProgressRecorder:
     def status_callback(self, status: str, message: str | None):
         self.status_updates.append((status, message))
 
+    @property
+    def real(self) -> list[tuple[str, str | None]]:
+        """Statuses a user would see, without the stall-grace signals."""
+        return [(s, m) for s, m in self.status_updates if s != ACTIVITY_GRACE_STATUS]
+
+    @property
+    def graces(self) -> list[float]:
+        return [float(m) for s, m in self.status_updates if s == ACTIVITY_GRACE_STATUS]
+
     @property
     def last_status(self) -> str | None:
-        return self.status_updates[-1][0] if self.status_updates else None
+        return self.real[-1][0] if self.real else None
 
     @property
     def last_message(self) -> str | None:
-        return self.status_updates[-1][1] if self.status_updates else None
+        return self.real[-1][1] if self.real else None
 
     @property
     def statuses(self) -> list[str]:
-        return [s[0] for s in self.status_updates]
+        return [s[0] for s in self.real]
 
 
 class FakeSlskd:
@@ -246,6 +256,35 @@ class TestLocalPaths:
         assert local_path_for(Path("/dl"), "a\\Author\\..") is None
         assert local_path_for(Path("/dl"), "a\\Author\\") is None
 
+    @pytest.mark.parametrize(
+        "filename",
+        [
+            "../../../etc//passwd.epub",  # an empty last folder used to keep the whole path
+            "/abs/path//book.epub",
+            "/abs/path/book.epub",
+            "\\\\server\\share\\book.epub",
+            "a//Book.epub",
+            "a\\.\\Book.epub",
+            "C:\\Book.epub",
+            "C:Book.epub",
+            "a\\D:\\Book.epub",
+            "a\\Author\\Book\0.epub",
+        ],
+    )
+    def test_local_path_refuses_names_that_could_leave_the_root(self, tmp_path, filename):
+        assert local_path_for(tmp_path / "dl", filename) is None
+
+    def test_a_windows_peers_drive_is_only_its_share_root(self):
+        assert local_path_for(Path("/dl"), "C:\\Books\\Author\\x.epub") == Path("/dl/Author/x.epub")
+
+    def test_local_path_refuses_a_folder_that_links_outside_the_root(self, tmp_path):
+        root = tmp_path / "dl"
+        root.mkdir()
+        (tmp_path / "elsewhere").mkdir()
+        (root / "Author").symlink_to(tmp_path / "elsewhere", target_is_directory=True)
+
+        assert local_path_for(root, FILE_A) is None
+
     def test_root_override_wins(self, config_values):
         config_values["SLSKD_DOWNLOAD_PATH"] = "/mnt/slskd "
         client = MagicMock()
@@ -633,6 +672,91 @@ class TestDownload:
         assert rec.last_status == "error"
         assert "unsafe" in (rec.last_message or "")
 
+    @pytest.mark.parametrize("folder", ["..\\outside\\", "../outside/", "{outside}/"])
+    def test_a_peer_file_name_cannot_reach_files_outside_the_root(
+        self, fast_handler, config_values, tmp_path, folder
+    ):
+        root = tmp_path / "dl"
+        outside = tmp_path / "outside"  # where "../outside" leads from the root
+        root.mkdir()
+        outside.mkdir()
+        names = ["01.mp3", "02.mp3"]
+        for name in names:
+            (outside / name).write_bytes(b"secret")
+        config_values["SLSKD_DOWNLOAD_PATH"] = str(root)
+        # The doubled separator leaves an empty last folder name, which used to let the
+        # whole directory through as the local folder.
+        sep = folder[-1]
+        files = [
+            {"filename": f"{folder.format(outside=outside)}{sep}{name}", "size": 6}
+            for name in names
+        ]
+        done = [_transfer(f["filename"], "Completed, Succeeded", transferred=6) for f in files]
+        fake = FakeSlskd(polls=[done])
+        _mount(fast_handler, fake)
+        rec = ProgressRecorder()
+
+        result = fast_handler.download(
+            _task(files=files), Event(), rec.progress_callback, rec.status_callback
+        )
+
+        assert result is None
+        assert rec.last_status == "error"
+        assert "unsafe" in (rec.last_message or "")
+        assert fake.enqueued == []
+        # A multi-file release is staged with a move: the files must still be there.
+        assert sorted(p.name for p in outside.iterdir()) == names
+
+    def test_a_peer_queue_asks_the_orchestrator_not_to_stall_the_download(
+        self, fast_handler, config_values, tmp_path
+    ):
+        """Queued at the peer, every poll repeats the same status and progress. Without a
+        grace the orchestrator cancels after 5 minutes, long before the queue timeout."""
+        import shelfmark.release_sources.slskd.handler as mod
+
+        config_values["SLSKD_DOWNLOAD_PATH"] = str(tmp_path)
+        queued = _transfer(FILE_A, "Queued, Remotely", placeInQueue=4)
+        fake = FakeSlskd(
+            polls=[
+                [queued],
+                [queued],
+                [_transfer(FILE_A, "InProgress", transferred=500)],
+                [_transfer(FILE_A, "Completed, Succeeded", transferred=1000)],
+            ]
+        )
+        _mount(fast_handler, fake)
+        _write_completed(tmp_path, FILE_A)
+        rec = ProgressRecorder()
+
+        result = fast_handler.download(_task(), Event(), rec.progress_callback, rec.status_callback)
+
+        assert result is not None
+        assert rec.graces, "no activity grace was requested while the peer queued the file"
+        assert rec.graces[0] == mod.QUEUE_GRACE_SECONDS
+        # released once the transfer starts, so the normal stall timer applies again
+        assert rec.graces[-1] == 0.0
+
+    def test_a_long_peer_queue_renews_the_grace_until_the_queue_timeout(
+        self, fast_handler, config_values, monkeypatch
+    ):
+        import shelfmark.release_sources.slskd.handler as mod
+
+        monkeypatch.setattr(SlskdHandler, "_queue_timeout_seconds", lambda self: 3600)
+        clock = iter([0, 0, 700, 1400, 3700])
+        monkeypatch.setattr(SlskdHandler, "_clock", lambda self: next(clock, 3700))
+        fake = FakeSlskd(polls=[[_transfer(FILE_A, "Queued, Remotely", tid="q")]])
+        _mount(fast_handler, fake)
+        rec = ProgressRecorder()
+        cancel = MagicMock(spec=Event)
+        cancel.is_set.side_effect = lambda: fake.poll_count > 10  # never hang
+        cancel.wait.return_value = False
+
+        result = fast_handler.download(_task(), cancel, rec.progress_callback, rec.status_callback)
+
+        assert result is None
+        assert rec.last_message == "Peer did not start the transfer within 60 minutes"
+        assert rec.graces == [mod.QUEUE_GRACE_SECONDS] * 4 + [0.0]
+
     def test_status_check_errors_are_retried(self, fast_handler, config_values, tmp_path):
         config_values["SLSKD_DOWNLOAD_PATH"] = str(tmp_path)
         fake = FakeSlskd(polls=[[_transfer(FILE_A, "Completed, Succeeded", transferred=1000)]])

tests/slskd/test_handler_isolated.py

@@ -100,7 +100,10 @@ class FakeSlskd:
         if state == OK and self.write_on_success and filename not in self._written:
             folder = self.root / self._destination
             folder.mkdir(parents=True, exist_ok=True)
-            (folder / filename.rsplit("\\", 1)[-1]).write_bytes(b"x" * size)
+            target = folder / filename.rsplit("\\", 1)[-1]
+            if target.exists():  # slskd keeps the file already there and renames the new one
+                target = target.with_stem(f"{target.stem}_639000000000000000")
+            target.write_bytes(b"x" * size)
             self._written.add(filename)
         done = size if state == OK else 0
         return {
@@ -395,6 +398,71 @@ class TestRejoinAfterARestart:
         assert "denied" in (rec.real[-1][1] or "")
 
 
+class TestARetryAfterAPartialDownload:
+    """A retry reuses the task's folder, and slskd renames a file that is already there, so
+    files from an attempt that did not finish must not stay behind."""
+
+    def test_a_failed_attempt_leaves_no_files_in_the_folder(self, handler, cfg, tmp_path):
+        fake = FakeSlskd(tmp_path, script={FILE_A: [OK], FILE_B: ["Completed, Errored"]})
+
+        result, _, rec = _run(handler, fake)
+
+        assert result is None
+        assert rec.real[-1][0] == "error"
+        assert not (tmp_path / "shelfmark" / "slskd_abc").exists()
+
+    def test_a_cancelled_attempt_leaves_no_files_in_the_folder(self, handler, cfg, tmp_path):
+        cancel = Event()
+        fake = FakeSlskd(tmp_path, script={FILE_A: [OK], FILE_B: ["InProgress"]})
+        fake.on_get = lambda f, name, state: cancel.set() if name == FILE_B else None
+
+        result, _, rec = _run(handler, fake, cancel=cancel)
+
+        assert result is None
+        assert rec.real[-1][0] == "cancelled"
+        assert not (tmp_path / "shelfmark" / "slskd_abc").exists()
+
+    def test_the_retry_of_a_failed_attempt_completes(self, handler, cfg, tmp_path):
+        first = FakeSlskd(tmp_path, script={FILE_A: [OK], FILE_B: ["Completed, Errored"]})
+        assert _run(handler, first)[0] is None
+
+        result, _, rec = _run(handler, FakeSlskd(tmp_path))
+
+        folder = tmp_path / "shelfmark" / "slskd_abc"
+        assert result == str(folder)
+        assert "error" not in rec.statuses
+        assert sorted(p.name for p in folder.iterdir()) == ["Book One.epub", "Book Two.epub"]
+
+    def test_files_left_by_an_attempt_that_never_cleaned_up_are_replaced(
+        self, handler, cfg, tmp_path
+    ):
+        # Shelfmark restarted after one of the two files had arrived.
+        folder = tmp_path / "shelfmark" / "slskd_abc"
+        folder.mkdir(parents=True)
+        (folder / "Book One.epub").write_bytes(b"x" * 5)
+
+        result, _, rec = _run(handler, FakeSlskd(tmp_path))
+
+        assert result == str(folder)
+        assert "error" not in rec.statuses
+        assert sorted(p.name for p in folder.iterdir()) == ["Book One.epub", "Book Two.epub"]
+
+
+class TestUnsafeFileNames:
+    def test_an_unsafe_peer_file_name_is_refused_before_anything_is_queued(
+        self, handler, cfg, tmp_path
+    ):
+        fake = FakeSlskd(tmp_path)
+        task = _task(files=[{"filename": "../../../etc//passwd.epub", "size": 5}])
+
+        result, _, rec = _run(handler, fake, task=task)
+
+        assert result is None
+        assert rec.real[-1][0] == "error"
+        assert "unsafe" in (rec.real[-1][1] or "")
+        assert fake.batches == []
+
+
 class TestFallbackAndSettings:
     def test_an_old_slskd_without_batches_uses_the_original_enqueue(self, handler, cfg, tmp_path):
         class Old(FakeSlskd):

tests/slskd/test_source.py

@@ -3,10 +3,12 @@
 from unittest.mock import MagicMock
 
 import pytest
+import requests
 
 from shelfmark.core.search_plan import ReleaseSearchPlan, ReleaseSearchVariant
 from shelfmark.metadata_providers import BookMetadata
-from shelfmark.release_sources import ReleaseProtocol, get_source
+from shelfmark.release_sources import ReleaseProtocol, SourceUnavailableError, get_source
+from shelfmark.release_sources.slskd.api import SlskdAuthError
 from shelfmark.release_sources.slskd.source import (
     SlskdSource,
     _plan_queries,
@@ -96,6 +98,10 @@ class TestPathHelpers:
         assert remote_directory_name("Author") == "Author"
         assert remote_directory_name("") == ""
 
+    def test_remote_directory_name_skips_an_empty_last_segment(self):
+        assert remote_directory_name("../../../etc/") == "etc"
+        assert remote_directory_name("@@a\\Books\\Author\\") == "Author"
+
     @pytest.mark.parametrize(
         ("name", "ext"),
         [
@@ -137,6 +143,20 @@ class TestIterPeerFiles:
         response = _response(files=[None, {"filename": ""}, {"filename": "dir\\"}, "junk"])
         assert iter_peer_files(response) == []
 
+    def test_skips_file_names_that_could_leave_the_downloads_folder(self):
+        evil = [
+            "../../../etc//passwd.epub",
+            "/abs/path//book.epub",
+            "/abs/path/book.epub",
+            "a\\..\\book.epub",
+            "C:\\book.epub",
+            "a\\D:\\book.epub",
+        ]
+        response = _response(files=[*(_file(name) for name in evil), _file("C:\\Books\\ok.epub")])
+        assert [pf.filename for pf in iter_peer_files(response)] == ["C:\\Books\\ok.epub"]
+        releases = build_releases([response], content_type="ebook", formats=EBOOK_FORMATS)
+        assert [r.extra["files"][0]["filename"] for r in releases] == ["C:\\Books\\ok.epub"]
+
     def test_missing_username_yields_nothing(self):
         assert iter_peer_files({"files": [_file("a.epub")]}) == []
 
@@ -359,7 +379,7 @@ class TestSlskdSource:
     def _config(self, **overrides):
         values = {"SLSKD_ENABLED": True, "SLSKD_URL": "http://slskd:5030", "SLSKD_API_KEY": "k"}
         values.update(overrides)
-        return lambda k, default=None: values.get(k, default)
+        return lambda k, default=None, **_kwargs: values.get(k, default)
 
     def test_registered(self):
         assert isinstance(get_source("slskd"), SlskdSource)
@@ -461,6 +481,51 @@ class TestSlskdSource:
 
         assert len(SlskdSource().search(_make_book(), _make_plan())) == 1
 
+    @pytest.mark.parametrize(
+        ("error", "expected"),
+        [
+            (requests.ConnectionError("Connection refused"), "Connection refused"),
+            (SlskdAuthError("slskd rejected the API key"), "API key"),
+        ],
+    )
+    def test_search_raises_when_every_query_fails(self, monkeypatch, error, expected):
+        """slskd being down or rejecting the key must not read as "no releases found"."""
+        import shelfmark.release_sources.slskd.source as mod
+
+        monkeypatch.setattr(mod.config, "get", self._config())
+        client = MagicMock()
+        client.search.side_effect = error
+        monkeypatch.setattr(SlskdSource, "_get_client", lambda self: client)
+
+        with pytest.raises(SourceUnavailableError, match=expected):
+            SlskdSource().search(_make_book(), _make_plan())
+        assert client.search.call_count == 2  # every query was tried
+
+    def test_search_failure_is_reported_to_the_user(self, monkeypatch):
+        import shelfmark.release_sources.slskd.source as mod
+        from shelfmark.core.release_search import search_source_releases
+
+        monkeypatch.setattr(mod.config, "get", self._config())
+        client = MagicMock()
+        client.search.side_effect = requests.ConnectionError("Connection refused")
+        monkeypatch.setattr(SlskdSource, "_get_client", lambda self: client)
+
+        _, releases, error = search_source_releases("slskd", _make_book())
+
+        assert releases == []
+        assert error is not None
+        assert "Connection refused" in error
+
+    def test_one_failed_query_among_answered_ones_is_not_an_outage(self, monkeypatch):
+        import shelfmark.release_sources.slskd.source as mod
+
+        monkeypatch.setattr(mod.config, "get", self._config())
+        client = MagicMock()
+        client.search.side_effect = [RuntimeError("boom"), []]
+        monkeypatch.setattr(SlskdSource, "_get_client", lambda self: client)
+
+        assert SlskdSource().search(_make_book(), _make_plan()) == []
+
     def test_search_uses_configured_limits(self, monkeypatch):
         import shelfmark.release_sources.slskd.source as mod