Contain remote path mappings (#974)

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
 
 

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:

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"

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"