diff --git a/shelfmark/core/path_mappings.py b/shelfmark/core/path_mappings.py index 61578510..d06a16ab 100644 --- a/shelfmark/core/path_mappings.py +++ b/shelfmark/core/path_mappings.py @@ -10,7 +10,7 @@ A mapping rewrites a remote path prefix into a local path prefix. from __future__ import annotations from dataclasses import dataclass -from pathlib import Path +from pathlib import Path, PureWindowsPath from typing import TYPE_CHECKING if TYPE_CHECKING: @@ -50,6 +50,42 @@ def _normalize_host(host: str) -> str: return str(host or "").strip().lower() +def _is_relative_to(path: Path, prefix: Path) -> bool: + try: + path.relative_to(prefix) + except ValueError: + return False + + return True + + +def _join_contained_path(local_prefix: str, remainder: str) -> Path | None: + local_path = Path(local_prefix) + + if remainder: + remainder_path = Path(remainder) + windows_remainder_path = PureWindowsPath(remainder) + + if ( + remainder_path.is_absolute() + or windows_remainder_path.is_absolute() + or ".." in remainder_path.parts + or ".." in windows_remainder_path.parts + ): + return None + + remapped = local_path / remainder_path + else: + remapped = local_path + + resolved_local_path = local_path.resolve(strict=False) + resolved_remapped = remapped.resolve(strict=False) + if not _is_relative_to(resolved_remapped, resolved_local_path): + return None + + return remapped + + def parse_remote_path_mappings(value: object) -> list[RemotePathMapping]: """Parse configured remote-path mapping rows into normalized mappings.""" if not value or not isinstance(value, list): @@ -81,8 +117,12 @@ def remap_remote_to_local_with_match( mappings: Iterable[RemotePathMapping], host: str, remote_path: str | Path, -) -> tuple[Path, bool]: - """Remap a remote path and report whether a configured mapping matched.""" +) -> tuple[Path | None, bool]: + """Remap a remote path and report whether a configured mapping matched. + + Returns ``(None, True)`` when a mapping prefix matched but the remainder was + unsafe to join under the local prefix. + """ host_normalized = _normalize_host(host) remote_normalized = _normalize_prefix(str(remote_path)) @@ -119,7 +159,10 @@ def remap_remote_to_local_with_match( remainder = remainder.removeprefix("/") - remapped = Path(local_prefix) / remainder if remainder else Path(local_prefix) + remapped = _join_contained_path(local_prefix, remainder) + if remapped is None: + return None, True + return remapped, True return Path(remote_normalized), False @@ -134,6 +177,8 @@ def remap_remote_to_local( host=host, remote_path=remote_path, ) + if remapped is None: + return Path(str(remote_path)) return remapped diff --git a/shelfmark/download/clients/base_handler.py b/shelfmark/download/clients/base_handler.py index aaf7720d..8571f79d 100644 --- a/shelfmark/download/clients/base_handler.py +++ b/shelfmark/download/clients/base_handler.py @@ -276,7 +276,18 @@ class ExternalClientHandler(DownloadHandler, ABC): remote_path=source_path_obj, ) - delete_path = remapped if matched_mapping else source_path_obj + if matched_mapping: + if remapped is None: + logger.warning( + "Refusing to delete download data for %s %s because remote path mapping rejected unsafe path: %s", + client.name, + download_id, + source_path_obj, + ) + return + delete_path = remapped + else: + delete_path = source_path_obj if str(delete_path) in ("", "/"): logger.warning( @@ -435,6 +446,19 @@ class ExternalClientHandler(DownloadHandler, ABC): ) if matched_mapping: + if remapped is None: + message = ( + f"Remote path mapping rejected unsafe path '{source_path_obj}'. " + f"Check Settings > Advanced > Remote Path Mappings." + ) + failure_log = "Remote path mapping rejected unsafe path for %s (%s): %s" + failure_args = (client.name, download_id, source_path_obj) + if log_details: + logger.error(failure_log, *failure_args) + else: + logger.debug(failure_log, *failure_args) + return None, message + remapped_exists, remapped_error = _probe_completed_path(remapped) if log_details: diff --git a/tests/core/test_path_mappings.py b/tests/core/test_path_mappings.py new file mode 100644 index 00000000..63073652 --- /dev/null +++ b/tests/core/test_path_mappings.py @@ -0,0 +1,59 @@ +from shelfmark.core.path_mappings import ( + RemotePathMapping, + remap_remote_to_local_with_match, +) + + +def test_remap_rejects_parent_directory_remainder(tmp_path): + mapping = RemotePathMapping( + host="qbittorrent", + remote_path="/remote/downloads", + local_path=str(tmp_path / "local" / "downloads"), + ) + remote_path = "/remote/downloads/../outside/book.epub" + + remapped, matched = remap_remote_to_local_with_match( + mappings=[mapping], + host="qbittorrent", + remote_path=remote_path, + ) + + assert matched is True + assert remapped is None + + +def test_remap_rejects_path_that_resolves_outside_local_prefix(tmp_path): + local_prefix = tmp_path / "local" / "downloads" + mapping = RemotePathMapping( + host="qbittorrent", + remote_path="/remote/downloads", + local_path=str(local_prefix), + ) + remote_path = "/remote/downloads/subdir/../../outside/book.epub" + + remapped, matched = remap_remote_to_local_with_match( + mappings=[mapping], + host="qbittorrent", + remote_path=remote_path, + ) + + assert matched is True + assert remapped is None + + +def test_remap_allows_normal_child_path_under_local_prefix(tmp_path): + local_prefix = tmp_path / "local" / "downloads" + mapping = RemotePathMapping( + host="qbittorrent", + remote_path="/remote/downloads", + local_path=str(local_prefix), + ) + + remapped, matched = remap_remote_to_local_with_match( + mappings=[mapping], + host="qbittorrent", + remote_path="/remote/downloads/author/book.epub", + ) + + assert matched is True + assert remapped == local_prefix / "author" / "book.epub" diff --git a/tests/prowlarr/test_remote_path_mappings.py b/tests/prowlarr/test_remote_path_mappings.py index 27064105..8a7da4a2 100644 --- a/tests/prowlarr/test_remote_path_mappings.py +++ b/tests/prowlarr/test_remote_path_mappings.py @@ -410,3 +410,83 @@ def test_windows_path_case_insensitive_matching(): assert result == str(local_file) assert task.original_download_path == str(local_file) + + +def test_resolve_download_path_rejects_unsafe_mapping_remainder_even_if_original_exists( + tmp_path, +): + remote_dir = tmp_path / "remote" / "downloads" + local_dir = tmp_path / "local" / "downloads" + escaped_file = tmp_path / "escape" / "book.epub" + remote_dir.mkdir(parents=True) + local_dir.mkdir(parents=True) + escaped_file.parent.mkdir(parents=True) + escaped_file.write_text("escaped content") + + raw_path = f"{remote_dir}/../../escape/book.epub" + + mock_client = MagicMock() + mock_client.name = "qbittorrent" + mock_client.get_download_path.return_value = raw_path + + def config_get(key: str, default=""): + if key == "PROWLARR_REMOTE_PATH_MAPPINGS": + return [ + { + "host": "qbittorrent", + "remotePath": str(remote_dir), + "localPath": str(local_dir), + } + ] + return default + + with patch("shelfmark.download.clients.base_handler.config.get", side_effect=config_get): + handler = ProwlarrHandler() + + resolved_path, error = handler._resolve_download_path_once( + mock_client, + "download_id", + log_details=False, + ) + + assert Path(raw_path).exists() + assert resolved_path is None + assert error is not None + assert "rejected unsafe path" in error + + +def test_delete_local_download_data_skips_unsafe_mapping_remainder_even_if_original_exists( + tmp_path, +): + remote_dir = tmp_path / "remote" / "downloads" + local_dir = tmp_path / "local" / "downloads" + escaped_file = tmp_path / "escape" / "book.epub" + remote_dir.mkdir(parents=True) + local_dir.mkdir(parents=True) + escaped_file.parent.mkdir(parents=True) + escaped_file.write_text("escaped content") + + raw_path = f"{remote_dir}/../../escape/book.epub" + + mock_client = MagicMock() + mock_client.name = "nzbget" + mock_client.get_download_path.return_value = raw_path + + def config_get(key: str, default=""): + if key == "PROWLARR_REMOTE_PATH_MAPPINGS": + return [ + { + "host": "nzbget", + "remotePath": str(remote_dir), + "localPath": str(local_dir), + } + ] + return default + + with patch("shelfmark.download.clients.base_handler.config.get", side_effect=config_get): + handler = ProwlarrHandler() + handler._delete_local_download_data(mock_client, "download_id") + + assert Path(raw_path).exists() + assert escaped_file.exists() + assert escaped_file.read_text() == "escaped content"