mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-29 11:27:43 +08:00
264 lines
8.9 KiB
Python
264 lines
8.9 KiB
Python
"""
|
||
插件实例彻底清理的路径穿越回归测试。
|
||
|
||
彻底清理是唯一会对用户可控标识拼出的目录执行递归删除的入口，而插件数据根是配置根下
|
||
的一层子目录。移植这项功能时原实现把标识直接拼进 ``PLUGIN_DATA_PATH`` 再 ``rmtree``，
|
||
``%2e%2e`` 经 ASGI 解码后就是 ``..``，一次请求即递归删掉整个配置目录。这里从 HTTP 入口
|
||
把这条攻击路径按编码形式逐个钉住。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from fastapi import FastAPI
|
||
from fastapi.testclient import TestClient
|
||
|
||
from app.api.dependencies.auth import get_current_active_superuser
|
||
from app.api.dependencies.plugin import get_plugin_config_command
|
||
from app.api.endpoints import plugininstance as plugininstance_endpoint
|
||
from app.application.plugin.config import PluginConfigCommand, PluginPurgeScope
|
||
from app.runtime.extensions.plugin import datadir as datadir_module
|
||
from app.runtime.extensions.plugin.datadir import remove_plugin_data_directory
|
||
|
||
# 路径段在 URL 里可以写成的形态。服务端拿到的是 ASGI 解码之后的结果，因而这些写法在
|
||
# 处理函数里各自等价于 `..`、`../` 等——只在路由层面过滤字面量 `..` 一个都拦不住。
|
||
TRAVERSAL_URL_SEGMENTS = (
|
||
"..",
|
||
"%2e%2e",
|
||
"%2E%2E",
|
||
"..%5c",
|
||
"%2e%2e%5c",
|
||
".%2e",
|
||
"%2e",
|
||
)
|
||
|
||
# 到达用例层的解码后标识，逐个都要被格式校验挡掉。
|
||
TRAVERSAL_IDS = (
|
||
"..",
|
||
"../",
|
||
"..\\",
|
||
"./..",
|
||
".././..",
|
||
"a/../..",
|
||
"/etc",
|
||
"C:\\Windows",
|
||
"",
|
||
".",
|
||
"Demo Plugin",
|
||
"1Plugin",
|
||
)
|
||
|
||
|
||
class _NullMutation:
|
||
"""
|
||
无准入约束的 mutation 替身。
|
||
"""
|
||
|
||
def __enter__(self) -> None:
|
||
"""
|
||
进入空上下文。
|
||
"""
|
||
|
||
def __exit__(self, *_exc: object) -> bool:
|
||
"""
|
||
退出空上下文且不吞异常。
|
||
"""
|
||
return False
|
||
|
||
|
||
@pytest.fixture(name="config_root")
|
||
def fixture_config_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
|
||
"""
|
||
搭出「配置根 / 插件数据根 / 同级受保护内容」这一真实布局。
|
||
"""
|
||
config_root = tmp_path / "config"
|
||
plugin_root = config_root / "plugins"
|
||
(plugin_root / "DemoPlugin").mkdir(parents=True)
|
||
(plugin_root / "DemoPlugin" / "cache.bin").write_text("x", encoding="utf-8")
|
||
(config_root / "protected").mkdir()
|
||
(config_root / "protected" / "keep.txt").write_text("must survive", encoding="utf-8")
|
||
(config_root / "user.db").write_text("must survive", encoding="utf-8")
|
||
monkeypatch.setattr(
|
||
datadir_module,
|
||
"get_runtime_setting",
|
||
lambda key, default=None: plugin_root if key == "PLUGIN_DATA_PATH" else default,
|
||
)
|
||
return config_root
|
||
|
||
|
||
def _purge_command(calls: list[tuple]) -> PluginConfigCommand:
|
||
"""
|
||
装出一个只有目录删除是真实实现的清理用例。
|
||
|
||
其余端口全部记账：本文件要断言的是「越界目录没有被删」以及「越界请求根本没有走到
|
||
任何删除动作」，把真实存储接进来只会让失败原因变糊。
|
||
"""
|
||
return PluginConfigCommand(
|
||
save_config=lambda *a: True,
|
||
initialize=lambda *a: None,
|
||
stop=lambda plugin_id: calls.append(("stop", plugin_id)),
|
||
delete_config=lambda plugin_id, force: calls.append(("config", plugin_id)),
|
||
delete_data=lambda *a: True,
|
||
reload_runtime=lambda *a: None,
|
||
publish_reset=lambda *a: None,
|
||
refresh_registrations=lambda *a: None,
|
||
mutation=lambda _operation: _NullMutation(),
|
||
delete_plugin_data_rows=lambda plugin_id: calls.append(("rows", plugin_id)),
|
||
destroy_own_database=lambda plugin_id: calls.append(("own_db", plugin_id)),
|
||
delete_data_directory=remove_plugin_data_directory,
|
||
purge_instance=lambda plugin_id: True,
|
||
is_clone=lambda _plugin_id: False,
|
||
)
|
||
|
||
|
||
def _client(calls: list[tuple]) -> TestClient:
|
||
"""
|
||
挂上真实路由与真实目录删除实现的测试客户端。
|
||
"""
|
||
app = FastAPI()
|
||
app.include_router(plugininstance_endpoint.router, prefix="/api/v1/plugin")
|
||
app.dependency_overrides[get_current_active_superuser] = lambda: object()
|
||
app.dependency_overrides[get_plugin_config_command] = lambda: _purge_command(calls)
|
||
return TestClient(app)
|
||
|
||
|
||
def _sentinels_intact(config_root: Path) -> bool:
|
||
"""
|
||
插件数据根之外的内容是否原封不动。
|
||
"""
|
||
return (
|
||
(config_root / "protected" / "keep.txt").is_file()
|
||
and (config_root / "user.db").is_file()
|
||
and (config_root / "plugins" / "DemoPlugin" / "cache.bin").is_file()
|
||
)
|
||
|
||
|
||
def test_percent_encoded_dot_dot_really_reaches_the_handler(
|
||
config_root: Path,
|
||
) -> None:
|
||
"""
|
||
``%2e%2e`` 确实会被路由匹配并交到处理函数手里，服务端才是最后一道防线。
|
||
|
||
客户端与代理通常会把字面量 ``..`` 规范化掉（这里表现为 404），百分号编码则原样
|
||
送达、由 ASGI 解码成 ``..``。把「路由层面过滤 `..`」当成防护是无效的，这条用例
|
||
就是用来防止将来有人据此撤掉服务端校验。
|
||
"""
|
||
calls: list[tuple] = []
|
||
|
||
response = _client(calls).post(
|
||
"/api/v1/plugin/instance/%2e%2e/purge",
|
||
json={"data_directory": True},
|
||
)
|
||
|
||
assert response.status_code == 200
|
||
assert response.json()["success"] is False
|
||
assert calls == []
|
||
assert _sentinels_intact(config_root)
|
||
|
||
|
||
@pytest.mark.parametrize("segment", TRAVERSAL_URL_SEGMENTS)
|
||
def test_no_url_encoding_of_dot_dot_deletes_outside_the_plugin_data_root(
|
||
config_root: Path,
|
||
segment: str,
|
||
) -> None:
|
||
"""
|
||
越界标识的任何 URL 写法都不得删到插件数据根之外，也不得触发任何一步删除。
|
||
|
||
未修版本对 ``%2e%2e`` 会返回 200、``purged: ["data_directory"]``，而配置根下的
|
||
全部内容已被 ``rmtree`` 清空——``PLUGIN_DATA_PATH/..`` 正是配置根。
|
||
"""
|
||
calls: list[tuple] = []
|
||
|
||
response = _client(calls).post(
|
||
f"/api/v1/plugin/instance/{segment}/purge",
|
||
json={"data_directory": True},
|
||
)
|
||
|
||
# 路由不匹配（404）同样算拦住；匹配上就必须由用例层拒绝
|
||
assert response.status_code in (200, 404)
|
||
if response.status_code == 200:
|
||
assert response.json()["success"] is False
|
||
assert calls == []
|
||
assert _sentinels_intact(config_root)
|
||
|
||
|
||
@pytest.mark.parametrize("instance_id", TRAVERSAL_IDS)
|
||
def test_purge_refuses_every_malformed_instance_id(
|
||
config_root: Path,
|
||
instance_id: str,
|
||
) -> None:
|
||
"""
|
||
解码后的非法标识逐个被拒，且在停实例之前就被拒。
|
||
|
||
校验放在最前面，非法标识不该先把配置行删掉再发现路径不对。
|
||
"""
|
||
calls: list[tuple] = []
|
||
|
||
result = _purge_command(calls).purge(
|
||
instance_id,
|
||
PluginPurgeScope(
|
||
config=True,
|
||
plugin_data=True,
|
||
own_database=True,
|
||
data_directory=True,
|
||
),
|
||
)
|
||
|
||
assert result.success is False
|
||
assert result.purged == ()
|
||
assert calls == []
|
||
assert _sentinels_intact(config_root)
|
||
|
||
|
||
@pytest.mark.parametrize("instance_id", ["DemoPlugin", "nullbr_search"])
|
||
def test_purge_deletes_the_directory_for_a_well_formed_instance_id(
|
||
config_root: Path,
|
||
instance_id: str,
|
||
) -> None:
|
||
"""
|
||
普通和含下划线的合法标识照常清理，且只动它自己那一层。
|
||
"""
|
||
calls: list[tuple] = []
|
||
target = config_root / "plugins" / instance_id
|
||
target.mkdir(exist_ok=True)
|
||
(target / "cache.bin").write_text("x", encoding="utf-8")
|
||
|
||
result = _purge_command(calls).purge(
|
||
instance_id, PluginPurgeScope(data_directory=True)
|
||
)
|
||
|
||
assert result.success is True
|
||
assert result.purged == ("data_directory",)
|
||
assert not target.exists()
|
||
assert calls == [("stop", instance_id), ("own_db", instance_id)]
|
||
assert (config_root / "protected" / "keep.txt").is_file()
|
||
assert (config_root / "user.db").is_file()
|
||
|
||
|
||
@pytest.mark.skipif(
|
||
sys.platform.startswith("win"),
|
||
reason="Windows 下建软链需要额外权限，逃逸语义由 POSIX 用例覆盖",
|
||
)
|
||
def test_purge_reports_a_refused_directory_removal_instead_of_raising(
|
||
config_root: Path,
|
||
) -> None:
|
||
"""
|
||
兜底判断拒绝删除时回报失败，而不是让 ValueError 变成 500。
|
||
|
||
实例 ID 合法但目录本身被换成了指向根外的软链，这是运维环境造成的状况而非请求
|
||
非法；把它收成失败结果，调用方才拿得到可读原因。
|
||
"""
|
||
(config_root / "plugins" / "Evil").symlink_to(
|
||
config_root / "protected", target_is_directory=True
|
||
)
|
||
calls: list[tuple] = []
|
||
|
||
result = _purge_command(calls).purge("Evil", PluginPurgeScope(data_directory=True))
|
||
|
||
assert result.success is False
|
||
assert "不在插件数据根" in result.message or "软链" in result.message
|
||
assert (config_root / "protected" / "keep.txt").is_file()
|