diff --git a/app/agent/tools/impl/add_download_tasks.py b/app/agent/tools/impl/add_download_tasks.py index 5bc909f1..15e847f9 100644 --- a/app/agent/tools/impl/add_download_tasks.py +++ b/app/agent/tools/impl/add_download_tasks.py @@ -15,7 +15,7 @@ from app.core.config import settings from app.core.context import Context from app.core.metainfo import MetaInfo from app.db.site_oper import SiteOper -from app.helper.directory import DirectoryHelper +from app.helper.directory import DirectoryHelper, validate_download_save_path from app.log import logger from app.schemas import FileURI, TorrentInfo from app.utils.crypto import HashUtils @@ -183,8 +183,8 @@ class AddDownloadTasksTool(MoviePilotTool): @staticmethod def _resolve_direct_download_dir(save_path: Optional[str]) -> Optional[Path]: """解析直接下载使用的目录,优先使用 save_path,其次使用默认下载目录""" - if save_path: - return Path(save_path) + if save_path is not None: + return Path(validate_download_save_path(save_path)) download_dirs = DirectoryHelper().get_download_dirs() if not download_dirs: @@ -225,6 +225,8 @@ class AddDownloadTasksTool(MoviePilotTool): merged_labels: Optional[str], ) -> tuple[Optional[str], Optional[str]]: """同步提交带上下文的下载任务,避免站点下载与下载器调用阻塞事件循环。""" + if save_path is not None: + save_path = validate_download_save_path(save_path) return DownloadChain().download_single( context=context, downloader=downloader, @@ -245,6 +247,12 @@ class AddDownloadTasksTool(MoviePilotTool): if not torrent_inputs: return "错误:torrent_url 不能为空。" + if save_path is not None: + try: + save_path = validate_download_save_path(save_path) + except ValueError as err: + return f"参数错误:save_path {str(err)}" + merged_labels = self._merge_labels_with_system_tag(labels) success_count = 0 failed_messages = [] diff --git a/app/agent/tools/impl/update_download_tasks.py b/app/agent/tools/impl/update_download_tasks.py index 715883e0..7decb511 100644 --- a/app/agent/tools/impl/update_download_tasks.py +++ b/app/agent/tools/impl/update_download_tasks.py @@ -8,6 +8,7 @@ from pydantic import BaseModel, Field from app.agent.tools.base import MoviePilotTool from app.agent.tools.tags import ToolTag from app.chain.download import DownloadChain +from app.helper.directory import validate_download_save_path from app.log import logger @@ -150,6 +151,18 @@ class UpdateDownloadTasksTool(MoviePilotTool): ], } + if save_path is not None: + try: + save_path = validate_download_save_path(save_path) + except ValueError: + return { + "hash": hash_value, + "downloader": resolved_downloader, + "results": [ + cls._build_result("save_path", False, "保存目录不在允许的下载目录范围内") + ], + } + results = [] if tags: tag_result = download_chain.set_torrents_tag( diff --git a/app/chain/download.py b/app/chain/download.py index 6d3ce0d3..18ae12d1 100644 --- a/app/chain/download.py +++ b/app/chain/download.py @@ -18,7 +18,7 @@ from app.core.meta import MetaBase from app.core.metainfo import MetaInfo from app.db.downloadhistory_oper import DownloadHistoryOper from app.db.mediaserver_oper import MediaServerOper -from app.helper.directory import DirectoryHelper +from app.helper.directory import DirectoryHelper, validate_download_save_path from app.helper.thread import ThreadHelper from app.helper.torrent import TorrentHelper from app.log import logger @@ -107,19 +107,27 @@ class DownloadChain(ChainBase): def _resolve_media_download_dir( media_info: MediaInfo, save_path: Optional[str] = None, - ) -> Union[str, Path]: + ) -> Tuple[Optional[str], Optional[Path], str]: """ 根据媒体信息解析下载目录。 """ storage = 'local' - if save_path: - return storage, Path(save_path) + if save_path is not None: + try: + validated_save_path = validate_download_save_path(save_path) + except ValueError as err: + logger.warn(str(err)) + return None, None, str(err) + if re.match(r"^[A-Za-z]:/", validated_save_path): + return storage, Path(validated_save_path), "" + file_uri = FileURI.from_uri(validated_save_path) + return file_uri.storage or storage, Path(file_uri.path), "" dir_info = DirectoryHelper().get_dir(media_info, include_unsorted=True) storage = dir_info.storage if dir_info else storage if not dir_info: logger.error(f"未找到下载目录:{media_info.type.value} {media_info.title_year}") - return None + return None, None, "未找到下载目录" if not dir_info.media_type and dir_info.download_type_folder: download_dir = Path(dir_info.download_path) / media_info.type.value @@ -129,7 +137,7 @@ class DownloadChain(ChainBase): if not dir_info.media_category and dir_info.download_category_folder and media_info.category: download_dir = download_dir / media_info.category - return storage, download_dir + return storage, download_dir, "" @staticmethod def _upload_subtitle_file( @@ -293,12 +301,12 @@ class DownloadChain(ChainBase): if not mediainfo: return False, "无法识别媒体信息", [] - storage, target_dir = self._resolve_media_download_dir( + storage, target_dir, error_msg = self._resolve_media_download_dir( media_info=mediainfo, save_path=save_path, ) if not target_dir: - return False, "未找到下载目录", [] + return False, error_msg or "未找到下载目录", [] request = RequestUtils( cookies=subtitle.site_cookie, @@ -527,9 +535,16 @@ class DownloadChain(ChainBase): f"Reason: {event_data.reason}") return (None, "下载被事件取消") if return_detail else None # 如果事件修改了下载路径,使用新路径 - if event_data.options and event_data.options.get("save_path"): + if event_data.options and "save_path" in event_data.options: save_path = event_data.options.get("save_path") + if save_path is not None: + try: + save_path = validate_download_save_path(save_path) + except ValueError as err: + logger.warn(str(err)) + return (None, str(err)) if return_detail else None + # 补充完整的media数据 if not _media.genre_ids: new_media = self.recognize_media(mtype=_media.type, tmdbid=_media.tmdb_id, @@ -570,7 +585,7 @@ class DownloadChain(ChainBase): storage = 'local' # 下载目录 - if save_path: + if save_path is not None: download_dir = Path(save_path) else: # 根据媒体信息查询下载目录配置 diff --git a/app/helper/directory.py b/app/helper/directory.py index 411943ce..5caaa775 100644 --- a/app/helper/directory.py +++ b/app/helper/directory.py @@ -1,15 +1,17 @@ import re -from pathlib import Path +from pathlib import Path, PurePath, PurePosixPath, PureWindowsPath from typing import List, Optional, Tuple from app import schemas from app.core.context import MediaInfo from app.db.systemconfig_oper import SystemConfigOper from app.log import logger -from app.schemas.types import SystemConfigKey +from app.schemas.types import StorageSchema, SystemConfigKey from app.utils.system import SystemUtils JINJA2_VAR_PATTERN = re.compile(r"\{\{.*?}}", re.DOTALL) +WINDOWS_DRIVE_PATTERN = re.compile(r"^[A-Za-z]:[\\/]") +WINDOWS_DRIVE_PREFIX_PATTERN = re.compile(r"^[A-Za-z]:") class DirectoryHelper: @@ -169,3 +171,118 @@ class DirectoryHelper: # 媒体根路径 media_root = rename_path.parents[rename_format_level - 1] return media_root + + +def _split_file_uri(value: str) -> Tuple[str, str]: + """ + 拆分 FileURI 字符串,保留原始路径用于安全校验。 + """ + for storage in StorageSchema: + protocol = f"{storage.value}:" + if value.startswith(protocol): + return storage.value, value[len(protocol):] + return "local", value + + +def _normalize_safe_posix_path(raw_path: str) -> PurePosixPath: + """ + 规范化保存目录路径,并拒绝跨目录或跨平台歧义写法。 + """ + if not raw_path: + raise ValueError("保存路径不能为空") + if "\\" in raw_path: + raise ValueError("保存路径不能包含反斜杠") + if raw_path.startswith("//"): + raise ValueError("保存路径不能使用 UNC 路径") + if WINDOWS_DRIVE_PATTERN.match(raw_path): + raise ValueError("保存路径不能使用 Windows 盘符路径") + if not raw_path.startswith("/"): + raise ValueError("保存路径必须是绝对路径") + + path = PurePosixPath(raw_path) + parts = [part for part in path.parts if part != "/"] + if ".." in parts: + raise ValueError("保存路径不能包含上级目录") + if parts and re.fullmatch(r"[A-Za-z]:", parts[0]): + raise ValueError("保存路径不能使用 Windows 盘符路径") + return path + + +def _normalize_safe_windows_path(raw_path: str) -> PureWindowsPath: + """ + 规范化已配置的 Windows 盘符路径;UNC 与反斜杠写法不参与下载目录 allowlist。 + """ + if not raw_path: + raise ValueError("保存路径不能为空") + if "\\" in raw_path: + raise ValueError("保存路径不能包含反斜杠") + if raw_path.startswith("//"): + raise ValueError("保存路径不能使用 UNC 路径") + if not WINDOWS_DRIVE_PATTERN.match(raw_path): + raise ValueError("保存路径必须是 Windows 绝对路径") + + path = PureWindowsPath(raw_path) + if ".." in path.parts: + raise ValueError("保存路径不能包含上级目录") + return path + + +def _normalize_download_path(raw_path: str, storage: str) -> Tuple[str, PurePath]: + """ + 按存储类型解析下载路径,本地允许 POSIX 或已配置的 Windows drive,远端保持 FileURI POSIX 语义。 + """ + path_value = str(raw_path or "").strip() + if storage == "local" and WINDOWS_DRIVE_PREFIX_PATTERN.match(path_value): + return "windows", _normalize_safe_windows_path(path_value) + return "posix", _normalize_safe_posix_path(path_value) + + +def _download_path_uri(storage: str, path: PurePath) -> str: + """ + 生成可传给下载器的 save_path,保持 /download/paths 暴露的本地和远端路径风格。 + """ + path_value = path.as_posix() + if storage == "local": + return path_value + return schemas.FileURI(storage=storage, path=path_value).uri + + +def _normalize_download_root(dir_info: schemas.TransferDirectoryConf) -> Optional[Tuple[str, str, PurePath]]: + """ + 读取下载目录配置中的根路径;无效配置不参与用户 save_path allowlist。 + """ + if not dir_info.download_path: + return None + storage = dir_info.storage or "local" + try: + path_style, root_path = _normalize_download_path(dir_info.download_path, storage) + return storage, path_style, root_path + except ValueError as err: + logger.warn(f"跳过无效下载目录配置:{str(err)}") + return None + + +def validate_download_save_path(save_path: str) -> str: + """ + 校验用户传入的下载保存目录,/download/paths 暴露的下载目录配置是允许写入的公共合同。 + + :param save_path: 下载保存目录,支持本地 /path 或远端 :/path + :return: 可直接传给下载接口的规范化保存目录 + """ + value = str(save_path or "").strip() + storage, raw_path = _split_file_uri(value) + target_style, target_path = _normalize_download_path(raw_path, storage) + + for dir_info in DirectoryHelper().get_download_dirs(): + root = _normalize_download_root(dir_info) + if not root: + continue + root_storage, root_style, root_path = root + if storage != root_storage: + continue + if target_style != root_style: + continue + if target_path == root_path or target_path.is_relative_to(root_path): + return _download_path_uri(storage, target_path) + + raise ValueError("保存路径不在允许的下载目录范围内") diff --git a/tests/test_download_save_path_allowlist.py b/tests/test_download_save_path_allowlist.py new file mode 100644 index 00000000..db05a09b --- /dev/null +++ b/tests/test_download_save_path_allowlist.py @@ -0,0 +1,439 @@ +import asyncio +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest + +import app.agent.tools.impl.add_download_tasks as add_tasks_module +import app.agent.tools.impl.update_download_tasks as update_tasks_module +import app.chain.download as download_module +from app.agent.tools.impl.add_download_tasks import AddDownloadTasksTool +from app.agent.tools.impl.update_download_tasks import UpdateDownloadTasksTool +from app.chain.download import DownloadChain +from app.core.context import Context, MediaInfo, SubtitleInfo, TorrentInfo +from app.core.metainfo import MetaInfo +from app.helper.directory import validate_download_save_path +from app.schemas import DownloaderTorrent, TransferDirectoryConf +from app.schemas.types import MediaType + + +def _download_dirs(): + return [ + TransferDirectoryConf( + name="本地下载", + priority=1, + storage="local", + download_path="/downloads", + ), + TransferDirectoryConf( + name="动漫远程下载", + priority=2, + storage="rclone", + download_path="/media/anime", + ), + ] + + +def _windows_download_dirs(): + return [ + TransferDirectoryConf( + name="Windows 下载", + priority=1, + storage="local", + download_path="C:/downloads", + ), + ] + + +@pytest.fixture(autouse=True) +def patch_download_dirs(monkeypatch): + monkeypatch.setattr( + "app.helper.directory.DirectoryHelper.get_download_dirs", + lambda _self: _download_dirs(), + ) + + +@pytest.mark.parametrize( + ("save_path", "expected"), + [ + ("/downloads", "/downloads"), + ("/downloads/movie/demo", "/downloads/movie/demo"), + ("rclone:/media/anime/sub", "rclone:/media/anime/sub"), + ], +) +def test_validate_download_save_path_accepts_configured_roots_and_children(save_path, expected): + assert validate_download_save_path(save_path) == expected + + +@pytest.mark.parametrize( + ("save_path", "expected"), + [ + ("C:/downloads", "C:/downloads"), + ("C:/downloads/movie", "C:/downloads/movie"), + ], +) +def test_validate_download_save_path_accepts_windows_configured_root_and_children( + monkeypatch, + save_path, + expected, +): + monkeypatch.setattr( + "app.helper.directory.DirectoryHelper.get_download_dirs", + lambda _self: _windows_download_dirs(), + ) + + assert validate_download_save_path(save_path) == expected + + +@pytest.mark.parametrize( + "save_path", + [ + "C:/other", + "D:/downloads", + "C:/downloads/../Windows", + "C:\\downloads\\movie", + "\\\\server\\share\\downloads", + ], +) +def test_validate_download_save_path_rejects_windows_paths_outside_configured_root( + monkeypatch, + save_path, +): + monkeypatch.setattr( + "app.helper.directory.DirectoryHelper.get_download_dirs", + lambda _self: _windows_download_dirs(), + ) + + with pytest.raises(ValueError): + validate_download_save_path(save_path) + + +@pytest.mark.parametrize( + "save_path", + [ + "/etc", + "/downloads/../etc", + "/downloads\\..\\etc", + "C:/downloads", + "\\\\server\\share\\downloads", + "//server/share/downloads", + "relative/downloads", + "", + " ", + "rclone:/media/movies", + "smb:/media/anime/sub", + ], +) +def test_validate_download_save_path_rejects_paths_outside_configured_roots(save_path): + with pytest.raises(ValueError): + validate_download_save_path(save_path) + + +def _build_context() -> Context: + return Context( + meta_info=MetaInfo("Demo Movie 2026"), + media_info=MediaInfo( + type=MediaType.MOVIE, + title="Demo Movie", + year="2026", + tmdb_id=1, + genre_ids=[18], + ), + torrent_info=TorrentInfo( + title="Demo Movie 2026", + enclosure="https://example.test/demo.torrent", + site_cookie="uid=1", + site_name="TestSite", + ), + ) + + +def _build_download_chain() -> DownloadChain: + chain = DownloadChain.__new__(DownloadChain) + chain.download = MagicMock() + chain.post_message = MagicMock() + chain.messagehelper = MagicMock() + return chain + + +def test_download_single_rejects_bad_save_path_before_downloader(monkeypatch): + monkeypatch.setattr(download_module.eventmanager, "send_event", lambda *args, **kwargs: None) + chain = _build_download_chain() + + download_id, error_msg = chain.download_single( + context=_build_context(), + torrent_content=b"torrent-content", + save_path="/etc", + return_detail=True, + ) + + assert download_id is None + assert "保存路径" in error_msg + chain.download.assert_not_called() + + +def test_download_single_rejects_event_overridden_bad_save_path_before_downloader(monkeypatch): + event_data = SimpleNamespace(cancel=False, source="plugin", reason="", options={"save_path": "/etc"}) + monkeypatch.setattr( + download_module.eventmanager, + "send_event", + lambda *args, **kwargs: SimpleNamespace(event_data=event_data), + ) + chain = _build_download_chain() + + download_id, error_msg = chain.download_single( + context=_build_context(), + torrent_content=b"torrent-content", + save_path="/downloads", + return_detail=True, + ) + + assert download_id is None + assert "保存路径" in error_msg + chain.download.assert_not_called() + + +@pytest.mark.parametrize("save_path", ["", " "]) +def test_download_single_rejects_explicit_empty_save_path_before_default_fallback(monkeypatch, save_path): + monkeypatch.setattr(download_module.eventmanager, "send_event", lambda *args, **kwargs: None) + monkeypatch.setattr( + download_module.DirectoryHelper, + "get_dir", + lambda *_args, **_kwargs: TransferDirectoryConf(storage="local", download_path="/downloads"), + ) + chain = _build_download_chain() + + download_id, error_msg = chain.download_single( + context=_build_context(), + torrent_content=b"torrent-content", + save_path=save_path, + return_detail=True, + ) + + assert download_id is None + assert "保存路径" in error_msg + chain.download.assert_not_called() + + +def test_download_single_rejects_event_empty_save_path_override_before_downloader(monkeypatch): + event_data = SimpleNamespace(cancel=False, source="plugin", reason="", options={"save_path": ""}) + monkeypatch.setattr( + download_module.eventmanager, + "send_event", + lambda *args, **kwargs: SimpleNamespace(event_data=event_data), + ) + chain = _build_download_chain() + + download_id, error_msg = chain.download_single( + context=_build_context(), + torrent_content=b"torrent-content", + save_path="/downloads", + return_detail=True, + ) + + assert download_id is None + assert "保存路径" in error_msg + chain.download.assert_not_called() + + +def test_resolve_media_download_dir_rejects_bad_subtitle_save_path(): + media_info = MediaInfo( + type=MediaType.MOVIE, + title="Demo Movie", + year="2026", + tmdb_id=1, + ) + + storage, target_dir, error_msg = DownloadChain._resolve_media_download_dir( + media_info=media_info, + save_path="/etc", + ) + + assert storage is None + assert target_dir is None + assert error_msg == "保存路径不在允许的下载目录范围内" + + +def test_download_subtitle_returns_specific_error_for_bad_save_path(): + chain = DownloadChain.__new__(DownloadChain) + chain.recognize_media = MagicMock( + return_value=MediaInfo( + type=MediaType.MOVIE, + title="Demo Movie", + year="2026", + tmdb_id=1, + ) + ) + subtitle = SubtitleInfo( + title="Demo Movie", + enclosure="https://example.test/subtitle.srt", + ) + + success, message, saved_files = chain.download_subtitle( + subtitle=subtitle, + save_path="/etc", + ) + + assert not success + assert message == "保存路径不在允许的下载目录范围内" + assert saved_files == [] + + +@pytest.mark.parametrize("save_path", ["", " "]) +def test_resolve_media_download_dir_rejects_explicit_empty_save_path_before_default_fallback( + monkeypatch, + save_path, +): + monkeypatch.setattr( + download_module.DirectoryHelper, + "get_dir", + lambda *_args, **_kwargs: TransferDirectoryConf(storage="local", download_path="/downloads"), + ) + media_info = MediaInfo( + type=MediaType.MOVIE, + title="Demo Movie", + year="2026", + tmdb_id=1, + ) + + storage, target_dir, error_msg = DownloadChain._resolve_media_download_dir( + media_info=media_info, + save_path=save_path, + ) + + assert storage is None + assert target_dir is None + assert "保存路径" in error_msg + + +def test_add_download_tasks_direct_magnet_rejects_bad_save_path_before_downloader(): + with pytest.raises(ValueError): + AddDownloadTasksTool._resolve_direct_download_dir("/etc") + + +@pytest.mark.parametrize("save_path", ["", " "]) +def test_add_download_tasks_direct_magnet_rejects_explicit_empty_save_path_before_default_fallback(save_path): + with pytest.raises(ValueError): + AddDownloadTasksTool._resolve_direct_download_dir(save_path) + + +def test_add_download_tasks_cached_context_rejects_bad_save_path_before_download_single(monkeypatch): + download_chain = MagicMock() + monkeypatch.setattr(add_tasks_module, "DownloadChain", lambda: download_chain) + + with pytest.raises(ValueError): + AddDownloadTasksTool._download_single_sync( + context=_build_context(), + downloader="qb", + save_path="/etc", + merged_labels=None, + ) + + download_chain.download_single.assert_not_called() + + +@pytest.mark.parametrize("save_path", ["", " "]) +def test_add_download_tasks_cached_context_rejects_explicit_empty_save_path_before_download_single( + monkeypatch, + save_path, +): + download_chain = MagicMock() + monkeypatch.setattr(add_tasks_module, "DownloadChain", lambda: download_chain) + + with pytest.raises(ValueError): + AddDownloadTasksTool._download_single_sync( + context=_build_context(), + downloader="qb", + save_path=save_path, + merged_labels=None, + ) + + download_chain.download_single.assert_not_called() + + +def test_update_download_tasks_rejects_bad_save_path_before_update_torrent(monkeypatch): + hash_value = "a" * 40 + download_chain = MagicMock() + download_chain.list_torrents.return_value = [ + DownloaderTorrent(downloader="qb", hash=hash_value, title="Demo") + ] + monkeypatch.setattr(update_tasks_module, "DownloadChain", lambda: download_chain) + + result = UpdateDownloadTasksTool._update_download_sync( + hash_value=hash_value, + save_path="/etc", + ) + + assert result["downloader"] == "qb" + assert result["results"] == [ + { + "operation": "save_path", + "success": False, + "message": "保存目录不在允许的下载目录范围内", + } + ] + download_chain.update_torrent.assert_not_called() + + +def test_update_download_tasks_passes_normalized_save_path_to_update_torrent(monkeypatch): + hash_value = "b" * 40 + download_chain = MagicMock() + download_chain.list_torrents.return_value = [ + DownloaderTorrent(downloader="qb", hash=hash_value, title="Demo") + ] + download_chain.update_torrent.return_value = {"save_path": True} + monkeypatch.setattr(update_tasks_module, "DownloadChain", lambda: download_chain) + + result = UpdateDownloadTasksTool._update_download_sync( + hash_value=hash_value, + save_path="rclone:/media/anime/sub", + ) + + assert result["results"][0]["success"] is True + download_chain.update_torrent.assert_called_once_with( + hash_string=hash_value, + downloader="qb", + download_limit=None, + upload_limit=None, + tracker_list=None, + save_path="rclone:/media/anime/sub", + category=None, + ratio_limit=None, + seeding_time_limit=None, + ) + + +def test_add_download_tasks_run_rejects_bad_save_path_before_partial_download(monkeypatch): + tool = AddDownloadTasksTool(session_id="session-1", user_id="10001") + download_chain = MagicMock() + monkeypatch.setattr(add_tasks_module, "DownloadChain", lambda: download_chain) + + result = asyncio.run( + tool.run( + torrent_url=["magnet:?xt=urn:btih:123"], + save_path="/etc", + ) + ) + + assert "save_path" in result + download_chain.download.assert_not_called() + + +@pytest.mark.parametrize("save_path", ["", " "]) +def test_add_download_tasks_run_rejects_explicit_empty_save_path_before_partial_download( + monkeypatch, + save_path, +): + tool = AddDownloadTasksTool(session_id="session-1", user_id="10001") + download_chain = MagicMock() + monkeypatch.setattr(add_tasks_module, "DownloadChain", lambda: download_chain) + + result = asyncio.run( + tool.run( + torrent_url=["magnet:?xt=urn:btih:123"], + save_path=save_path, + ) + ) + + assert "save_path" in result + download_chain.download.assert_not_called()