From 65b9c777de9194e40ba29dc17345afc5b4b4b0ba Mon Sep 17 00:00:00 2001 From: jxxghp Date: Fri, 11 Sep 2026 16:59:51 +0800 Subject: [PATCH] fix(security): replace episode offset eval with safe arithmetic --- app/application/formatting.py | 32 ++++++------- app/domain/meta/words.py | 4 +- .../architecture/dependency-baseline.json | 5 +- .../fixtures/architecture/mypy-baseline.json | 2 +- tests/test_episode_format_helper.py | 48 ++++++++++++++++++- 5 files changed, 67 insertions(+), 24 deletions(-) diff --git a/app/application/formatting.py b/app/application/formatting.py index 26b73e0b2..fbf9e5ca5 100644 --- a/app/application/formatting.py +++ b/app/application/formatting.py @@ -3,11 +3,12 @@ from collections import defaultdict from dataclasses import dataclass from functools import lru_cache from pathlib import Path -from typing import Dict, Iterable, List, Match, Optional, Tuple, Union +from typing import Dict, Iterable, List, Match, Optional, Tuple, Union, cast import anitopy from app.application.configuration import get_chain_runtime_config_snapshot +from app.domain.meta.words import calculate_episode_offset from app.domain.metainfo import MetaInfoPath from app.domain.meta.metabase import MetaBase from app.runtime.log import logger @@ -178,31 +179,28 @@ class FormatParser(object): if isinstance(self._start_ep, str): # `details` 格式为 `X-X` s, e = self._start_ep.split("-") - start_ep = self.__offset.replace("EP", s) - end_ep = self.__offset.replace("EP", e) if int(s) == int(e): - return int(eval(start_ep)), None, self.part - return int(eval(start_ep)), int(eval(end_ep)), self.part + return calculate_episode_offset(self.__offset, int(s)), None, self.part + return calculate_episode_offset(self.__offset, int(s)), calculate_episode_offset(self.__offset, int(e)), self.part else: # `details` 格式为 `X` - start_ep = self.__offset.replace("EP", str(self._start_ep)) - return int(eval(start_ep)), None, self.part + return calculate_episode_offset(self.__offset, self._start_ep), None, self.part elif not self._format: # `details` 格式为 `X,X` - start_ep = self.__offset.replace("EP", str(self._start_ep)) - end_ep = self.__offset.replace("EP", str(self._end_ep)) - return int(eval(start_ep)), int(eval(end_ep)), self.part + start_ep: Optional[int] = calculate_episode_offset(self.__offset, cast(int, self._start_ep)) + end_ep: Optional[int] = calculate_episode_offset(self.__offset, cast(int, self._end_ep)) + return start_ep, end_ep, self.part if not self._format: # 未填入`集数定位` 且没有`指定集数` 仅处理`集数偏移` - start_ep = eval(self.__offset.replace("EP", str(file_meta.begin_episode))) if file_meta.begin_episode else None - end_ep = eval(self.__offset.replace("EP", str(file_meta.end_episode))) if file_meta.end_episode else None - return int(start_ep) if start_ep else None, int(end_ep) if end_ep else None, self.part + start_ep = calculate_episode_offset(self.__offset, file_meta.begin_episode) if file_meta.begin_episode else None + end_ep = calculate_episode_offset(self.__offset, file_meta.end_episode) if file_meta.end_episode else None + return start_ep if start_ep else None, end_ep if end_ep else None, self.part else: # 有`集数定位` - s, e = self.__handle_single(file_name) - start_ep = self.__offset.replace("EP", str(s)) if s else None - end_ep = self.__offset.replace("EP", str(e)) if e else None - return int(eval(start_ep)) if start_ep else None, int(eval(end_ep)) if end_ep else None, self.part + matched_start, matched_end = self.__handle_single(file_name) + start_ep = calculate_episode_offset(self.__offset, matched_start) if matched_start else None + end_ep = calculate_episode_offset(self.__offset, matched_end) if matched_end else None + return start_ep if start_ep else None, end_ep if end_ep else None, self.part def __handle_single(self, file: str) -> Tuple[Optional[int], Optional[int]]: """ diff --git a/app/domain/meta/words.py b/app/domain/meta/words.py index 8763de662..8c1e86790 100644 --- a/app/domain/meta/words.py +++ b/app/domain/meta/words.py @@ -51,7 +51,7 @@ def _compile_custom_word_regex(pattern: str): return re.compile(pattern) -def _calculate_episode_offset(offset: str, episode: int) -> int: +def calculate_episode_offset(offset: str, episode: int) -> int: """ 按白名单算术语法计算集数偏移，避免执行任意表达式。 """ @@ -192,7 +192,7 @@ class WordsMatcher(metaclass=Singleton): offset_order_flag = False for episode_num_str in episode_nums_str: episode_num_int = int(cn2an.cn2an(episode_num_str, "smart")) - episode_num_offset_int = _calculate_episode_offset(offset, episode_num_int) + episode_num_offset_int = calculate_episode_offset(offset, episode_num_int) # 向前偏移 if episode_num_int > episode_num_offset_int: offset_order_flag = True diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 9a60a2b92..57d47724c 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -1077,8 +1077,8 @@ "runtime_only": true } }, - "edge_count": 8551, - "edge_sha256": "3a52844425d3330b00935d7d565e468506a85ae62972b5c257b409b5b000206b", + "edge_count": 8552, + "edge_sha256": "750c2b2d5ff50ba8c004011261f52e79bf1978cbaeac7019522cea67e073ca98", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -3356,6 +3356,7 @@ "app.application.formatting -> app.domain", "app.application.formatting -> app.domain.meta", "app.application.formatting -> app.domain.meta.metabase", + "app.application.formatting -> app.domain.meta.words", "app.application.formatting -> app.domain.metainfo", "app.application.formatting -> app.runtime", "app.application.formatting -> app.runtime.log", diff --git a/tests/fixtures/architecture/mypy-baseline.json b/tests/fixtures/architecture/mypy-baseline.json index 5d3b111b4..960440487 100644 --- a/tests/fixtures/architecture/mypy-baseline.json +++ b/tests/fixtures/architecture/mypy-baseline.json @@ -630,7 +630,7 @@ }, "app/application/formatting.py": { "arg-type": 4, - "assignment": 9, + "assignment": 4, "attr-defined": 1, "no-untyped-def": 7, "operator": 3, diff --git a/tests/test_episode_format_helper.py b/tests/test_episode_format_helper.py index 57566587d..624311e02 100644 --- a/tests/test_episode_format_helper.py +++ b/tests/test_episode_format_helper.py @@ -1,11 +1,13 @@ from pathlib import Path from types import SimpleNamespace +from unittest.mock import patch import pytest -from app.chain.transfer import TransferChain from app.application.formatting import EpisodeFormatRuleHelper, FormatParser, _AutoRecommendSample -from app.schemas import EpisodeFormatRule, FileItem +from app.chain.transfer.facade import TransferChain +from app.schemas.file import FileItem +from app.schemas.transfer import EpisodeFormatRule def _make_file(name: str, size: int = 150 * 1024 * 1024) -> FileItem: @@ -32,6 +34,48 @@ def _patch_media_exts(monkeypatch): ) +@pytest.mark.parametrize( + ("offset", "expected"), + [("-1", 2), ("EP*2-1", 5), ("EP+EP-1", 5)], +) +def test_format_parser_applies_safe_episode_offset(offset: str, expected: int): + """剧集偏移仅计算白名单算术表达式，并保留现有偏移语义。""" + parser = FormatParser(eformat="", offset=offset) + file_meta = SimpleNamespace(begin_episode=3, end_episode=3) + + assert parser.split_episode("", file_meta) == (expected, expected, None) + + +def test_format_parser_applies_safe_episode_offset_to_template_episode(): + """带定位模板时也应使用白名单计算集数偏移。""" + parser = FormatParser(eformat="Show - {ep}.mkv", offset="EP*2-1") + file_meta = SimpleNamespace(begin_episode=None, end_episode=None) + + assert parser.split_episode("Show - 03.mkv", file_meta) == (5, None, None) + + +def test_format_parser_applies_safe_episode_offset_to_explicit_range(): + """显式集数区间的起止集数都应使用安全偏移计算。""" + parser = FormatParser(eformat="", details="03-04", offset="EP+1") + file_meta = SimpleNamespace(begin_episode=None, end_episode=None) + + assert parser.split_episode("", file_meta) == (4, 5, None) + + +def test_format_parser_rejects_code_execution_in_episode_offset(): + """剧集偏移中的函数调用必须被拒绝，不能执行系统命令。""" + parser = FormatParser( + eformat="", + offset="EP+__import__('os').system('touch /tmp/moviepilot-format-parser-rce')*0", + ) + file_meta = SimpleNamespace(begin_episode=1, end_episode=1) + + with patch("os.system") as system_call, pytest.raises(ValueError): + parser.split_episode("", file_meta) + + system_call.assert_not_called() + + def test_rule_recommend_supports_range_episode_validation(): helper = EpisodeFormatRuleHelper() rule = EpisodeFormatRule(