Harden qBittorrent fallback path joins (#973)

shelfmark/download/clients/qbittorrent.py

@@ -2,9 +2,10 @@
 
 from __future__ import annotations
 
+import os
 import time
 from http import HTTPStatus
-from pathlib import Path
+from pathlib import Path, PurePosixPath, PureWindowsPath
 from types import SimpleNamespace
 from typing import NoReturn, TypedDict
 
@@ -46,6 +47,13 @@ _HTTP_STATUS_NOT_FOUND = HTTPStatus.NOT_FOUND
 _ONE_WEEK_IN_SECONDS = 604800
 
 
+class _UnsafeQBittorrentPath:
+    pass
+
+
+_UNSAFE_QBITTORRENT_PATH = _UnsafeQBittorrentPath()
+
+
 class _QBittorrentAddKwargs(TypedDict, total=False):
     rename: str
     category: str
@@ -136,6 +144,24 @@ def _is_explicit_add_failure(raw_result: object) -> bool:
     return normalized in {"fail", "fails", "error", "errors"}
 
 
+def _build_qbittorrent_child_path(base_path: object, child_path: object) -> str | None:
+    """Build a qBittorrent-reported child path without allowing escape from base."""
+    if not isinstance(base_path, str) or not base_path:
+        return None
+    if not isinstance(child_path, str) or not child_path:
+        return None
+
+    child = child_path.replace("\\", "/")
+    posix_child = PurePosixPath(child)
+    windows_child = PureWindowsPath(child_path)
+    if posix_child.is_absolute() or windows_child.is_absolute() or windows_child.drive:
+        return None
+    if any(part == ".." for part in posix_child.parts):
+        return None
+
+    return os.path.normpath(str(Path(base_path) / child))
+
+
 @register_client("torrent")
 class QBittorrentClient(DownloadClient):
     """qBittorrent download client."""
@@ -629,16 +655,18 @@ class QBittorrentClient(DownloadClient):
         download_id = getattr(torrent, "hash", "")
         if isinstance(download_id, str) and download_id:
             derived = self._derive_download_path_from_files(download_id)
-            if derived:
+            if derived and not isinstance(derived, _UnsafeQBittorrentPath):
                 return derived
 
         # Legacy fallback: save_path + name (for older clients/emulators)
-        return self._build_path(
+        return _build_qbittorrent_child_path(
             getattr(torrent, "save_path", ""),
             getattr(torrent, "name", ""),
         )
 
-    def _derive_download_path_from_files(self, download_id: str) -> str | None:
+    def _derive_download_path_from_files(
+        self, download_id: str
+    ) -> str | _UnsafeQBittorrentPath | None:
         """Derive completed download path using `/torrents/properties` + `/torrents/files`.
 
         This mirrors how common automation apps derive the path when
@@ -685,9 +713,12 @@ class QBittorrentClient(DownloadClient):
             first_name_norm = first_name.replace("\\", "/")
             top_level = first_name_norm.split("/", 1)[0]
             if not top_level:
-                return None
+                return _UNSAFE_QBITTORRENT_PATH
 
-            return os.path.normpath(str(Path(save_path) / top_level))
+            derived = _build_qbittorrent_child_path(save_path, top_level)
+            if derived is None:
+                return _UNSAFE_QBITTORRENT_PATH
+            return os.path.normpath(derived)
         except _QBITTORRENT_CLIENT_ERRORS as e:
             logger.debug(
                 "qBittorrent could not derive path from files: %s: %s",

tests/prowlarr/test_qbittorrent_client.py

@@ -406,6 +406,158 @@ class TestQBittorrentClientGetStatus:
             assert status.complete is True
             assert status.file_path == "/downloads/Some Torrent"
 
+    @pytest.mark.parametrize(
+        "file_name",
+        ["/escape/book.epub", "../escape/book.epub", "C:\\escape\\book.epub"],
+    )
+    def test_get_status_falls_back_after_unsafe_derived_file_path(self, monkeypatch, file_name):
+        """Unsafe file-derived paths do not block the safe legacy fallback."""
+        config_values = {
+            "QBITTORRENT_URL": "http://localhost:8080",
+            "QBITTORRENT_USERNAME": "admin",
+            "QBITTORRENT_PASSWORD": "password",
+            "QBITTORRENT_CATEGORY": "test",
+        }
+        monkeypatch.setattr(
+            "shelfmark.download.clients.qbittorrent.config.get",
+            lambda key, default="": config_values.get(key, default),
+        )
+
+        mock_torrent = MockTorrent(
+            hash_val="abc123",
+            progress=1.0,
+            state="uploading",
+            content_path="/downloads",
+            name="Some Torrent",
+        )
+        info_payload = mock_torrent.to_dict() | {"save_path": "/downloads"}
+
+        def response(kind: str):
+            r = MagicMock()
+            r.status_code = 200
+            r.raise_for_status = MagicMock()
+            if kind == "info":
+                r.json.return_value = [info_payload]
+            elif kind == "properties":
+                r.json.return_value = {"save_path": "/downloads"}
+            elif kind == "files":
+                r.json.return_value = [{"name": file_name}]
+            else:
+                raise AssertionError("unknown")
+            return r
+
+        mock_client_instance = MagicMock()
+
+        def get_side_effect(url, params=None, timeout=None):
+            if url.endswith("/api/v2/torrents/info"):
+                return response("info")
+            if url.endswith("/api/v2/torrents/properties"):
+                return response("properties")
+            if url.endswith("/api/v2/torrents/files"):
+                return response("files")
+            raise AssertionError(f"unexpected url: {url}")
+
+        mock_client_instance._session.get.side_effect = get_side_effect
+        mock_client_class = MagicMock(return_value=mock_client_instance)
+
+        with patch.dict("sys.modules", {"qbittorrentapi": MagicMock(Client=mock_client_class)}):
+            import importlib
+
+            import shelfmark.download.clients.qbittorrent as qb_module
+
+            importlib.reload(qb_module)
+
+            client = qb_module.QBittorrentClient()
+            status = client.get_status("abc123")
+            path = client.get_download_path("abc123")
+
+            assert status.complete is True
+            assert status.file_path == "/downloads/Some Torrent"
+            assert path == "/downloads/Some Torrent"
+
+    @pytest.mark.parametrize("torrent_name", ["/escape", "../escape", "C:\\escape"])
+    def test_get_status_rejects_unsafe_legacy_name_path(self, monkeypatch, torrent_name):
+        """Legacy save_path/name fallback must reject absolute and traversal names."""
+        config_values = {
+            "QBITTORRENT_URL": "http://localhost:8080",
+            "QBITTORRENT_USERNAME": "admin",
+            "QBITTORRENT_PASSWORD": "password",
+            "QBITTORRENT_CATEGORY": "test",
+        }
+        monkeypatch.setattr(
+            "shelfmark.download.clients.qbittorrent.config.get",
+            lambda key, default="": config_values.get(key, default),
+        )
+
+        mock_torrent = MockTorrent(
+            hash_val="abc123",
+            progress=1.0,
+            state="uploading",
+            content_path="/downloads",
+            name=torrent_name,
+        )
+        info_payload = mock_torrent.to_dict() | {"save_path": "/downloads"}
+        mock_client_instance = MagicMock()
+        mock_client_instance._session.get.side_effect = [
+            create_mock_session_response([info_payload], status_code=200),
+            create_mock_session_response([], status_code=404),
+        ]
+        mock_client_class = MagicMock(return_value=mock_client_instance)
+
+        with patch.dict("sys.modules", {"qbittorrentapi": MagicMock(Client=mock_client_class)}):
+            import importlib
+
+            import shelfmark.download.clients.qbittorrent as qb_module
+
+            importlib.reload(qb_module)
+
+            client = qb_module.QBittorrentClient()
+            status = client.get_status("abc123")
+
+            assert status.complete is True
+            assert status.file_path is None
+
+    def test_get_status_accepts_safe_legacy_name_path(self, monkeypatch):
+        """Normal relative torrent names still build under save_path."""
+        config_values = {
+            "QBITTORRENT_URL": "http://localhost:8080",
+            "QBITTORRENT_USERNAME": "admin",
+            "QBITTORRENT_PASSWORD": "password",
+            "QBITTORRENT_CATEGORY": "test",
+        }
+        monkeypatch.setattr(
+            "shelfmark.download.clients.qbittorrent.config.get",
+            lambda key, default="": config_values.get(key, default),
+        )
+
+        mock_torrent = MockTorrent(
+            hash_val="abc123",
+            progress=1.0,
+            state="uploading",
+            content_path="/downloads",
+            name="Some Torrent",
+        )
+        info_payload = mock_torrent.to_dict() | {"save_path": "/downloads"}
+        mock_client_instance = MagicMock()
+        mock_client_instance._session.get.side_effect = [
+            create_mock_session_response([info_payload], status_code=200),
+            create_mock_session_response([], status_code=404),
+        ]
+        mock_client_class = MagicMock(return_value=mock_client_instance)
+
+        with patch.dict("sys.modules", {"qbittorrentapi": MagicMock(Client=mock_client_class)}):
+            import importlib
+
+            import shelfmark.download.clients.qbittorrent as qb_module
+
+            importlib.reload(qb_module)
+
+            client = qb_module.QBittorrentClient()
+            status = client.get_status("abc123")
+
+            assert status.complete is True
+            assert status.file_path == "/downloads/Some Torrent"
+
     def test_get_status_not_found(self, monkeypatch):
         """Test status for non-existent torrent."""
         config_values = {