diff --git a/shelfmark/download/clients/qbittorrent.py b/shelfmark/download/clients/qbittorrent.py index 62096efa..6c9e36dc 100644 --- a/shelfmark/download/clients/qbittorrent.py +++ b/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", diff --git a/tests/prowlarr/test_qbittorrent_client.py b/tests/prowlarr/test_qbittorrent_client.py index 202a6028..1715c4a4 100644 --- a/tests/prowlarr/test_qbittorrent_client.py +++ b/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 = {