Harden system nettest SSRF handling

tests/test_system_nettest.py

@@ -0,0 +1,185 @@
+import asyncio
+import sys
+import unittest
+from types import ModuleType, SimpleNamespace
+from unittest.mock import patch
+
+
+def _stub_module(name: str, **attrs):
+    module = sys.modules.get(name)
+    if module is None:
+        module = ModuleType(name)
+        sys.modules[name] = module
+    for key, value in attrs.items():
+        setattr(module, key, value)
+    return module
+
+
+class _Dummy:
+    def __init__(self, *args, **kwargs):
+        pass
+
+
+for _module_name in ("pillow_avif", "aiofiles", "psutil"):
+    _stub_module(_module_name)
+
+_stub_module("app.helper.sites", SitesHelper=_Dummy)
+_stub_module("app.chain.mediaserver", MediaServerChain=_Dummy)
+_stub_module("app.chain.search", SearchChain=_Dummy)
+_stub_module("app.chain.system", SystemChain=_Dummy)
+_stub_module("app.core.event", eventmanager=_Dummy())
+_stub_module("app.core.metainfo", MetaInfo=_Dummy)
+_stub_module("app.core.module", ModuleManager=_Dummy)
+_stub_module(
+    "app.core.security",
+    verify_apitoken=_Dummy,
+    verify_resource_token=_Dummy,
+    verify_token=_Dummy,
+)
+_stub_module("app.db.models", User=_Dummy)
+_stub_module("app.db.systemconfig_oper", SystemConfigOper=_Dummy)
+_stub_module(
+    "app.db.user_oper",
+    get_current_active_superuser=_Dummy,
+    get_current_active_superuser_async=_Dummy,
+    get_current_active_user_async=_Dummy,
+)
+_stub_module("app.helper.llm", LLMHelper=_Dummy)
+_stub_module("app.helper.mediaserver", MediaServerHelper=_Dummy)
+_stub_module("app.helper.message", MessageHelper=_Dummy)
+_stub_module("app.helper.progress", ProgressHelper=_Dummy)
+_stub_module("app.helper.rule", RuleHelper=_Dummy)
+_stub_module("app.helper.subscribe", SubscribeHelper=_Dummy)
+_stub_module("app.helper.system", SystemHelper=_Dummy)
+_stub_module("app.helper.image", ImageHelper=_Dummy)
+_stub_module("app.scheduler", Scheduler=_Dummy)
+_stub_module("app.utils.crypto", HashUtils=_Dummy)
+_stub_module("app.utils.http", RequestUtils=_Dummy, AsyncRequestUtils=_Dummy)
+_stub_module("version", APP_VERSION="test")
+
+from app.api.endpoints import system as system_endpoint
+
+
+class NettestSecurityTest(unittest.TestCase):
+    def test_nettest_targets_are_served_by_backend(self):
+        resp = asyncio.run(system_endpoint.nettest_targets(_="token"))
+
+        self.assertTrue(resp.success)
+        self.assertTrue(any(item["id"] == "pip_proxy" for item in resp.data))
+        self.assertTrue(any(item["id"] == "github_proxy_web" for item in resp.data))
+
+    def test_nettest_blocks_unknown_target(self):
+        class FailIfCalled:
+            def __init__(self, *args, **kwargs):
+                raise AssertionError("nettest should reject unknown targets before any outbound request")
+
+        with patch.object(system_endpoint, "AsyncRequestUtils", FailIfCalled):
+            resp = asyncio.run(
+                system_endpoint.nettest(
+                    target_id="unknown-target",
+                    _="token",
+                )
+            )
+
+        self.assertFalse(resp.success)
+        self.assertIn("不存在", resp.message)
+
+    def test_nettest_blocks_unapproved_redirect(self):
+        captured = {"calls": 0}
+
+        class FakeResponse:
+            def __init__(self, status_code, headers=None, text=""):
+                self.status_code = status_code
+                self.headers = headers or {}
+                self.text = text
+
+            async def aclose(self):
+                return None
+
+        class FakeAsyncRequestUtils:
+            def __init__(self, **kwargs):
+                captured["init_kwargs"] = kwargs
+
+            async def get_res(self, url, allow_redirects=True):
+                captured["calls"] += 1
+                return FakeResponse(
+                    302,
+                    headers={"location": "https://169.254.169.254/latest/meta-data/"},
+                )
+
+        with patch.object(system_endpoint, "AsyncRequestUtils", FakeAsyncRequestUtils), patch.object(
+            system_endpoint.settings,
+            "GITHUB_PROXY",
+            "https://ghproxy.example/",
+        ):
+            resp = asyncio.run(
+                system_endpoint.nettest(
+                    target_id="github_proxy_web",
+                    _="token",
+                )
+            )
+
+        self.assertFalse(resp.success)
+        self.assertIn("跳转", resp.message)
+        self.assertEqual(captured["calls"], 1)
+
+    def test_nettest_uses_safe_http_options_and_server_side_content_check(self):
+        captured = {}
+
+        class FakeAsyncRequestUtils:
+            def __init__(self, **kwargs):
+                captured["init_kwargs"] = kwargs
+
+            async def get_res(self, url, allow_redirects=True):
+                captured["url"] = url
+                captured["allow_redirects"] = allow_redirects
+                return SimpleNamespace(status_code=200, text="MoviePilot README")
+
+        with patch.object(system_endpoint, "AsyncRequestUtils", FakeAsyncRequestUtils), patch.object(
+            system_endpoint.settings,
+            "GITHUB_PROXY",
+            "https://ghproxy.example/",
+        ):
+            resp = asyncio.run(
+                system_endpoint.nettest(
+                    target_id="github_proxy_web",
+                    include="tag_name",
+                    _="token",
+                )
+            )
+
+        self.assertTrue(resp.success)
+        self.assertEqual(
+            captured["url"],
+            "https://ghproxy.example/https://github.com/jxxghp/MoviePilot/blob/v2/README.md",
+        )
+        self.assertFalse(captured["allow_redirects"])
+        self.assertTrue(captured["init_kwargs"]["verify"])
+        self.assertFalse(captured["init_kwargs"]["follow_redirects"])
+
+    def test_nettest_fails_when_expected_content_is_missing(self):
+        class FakeAsyncRequestUtils:
+            def __init__(self, **kwargs):
+                pass
+
+            async def get_res(self, url, allow_redirects=True):
+                return SimpleNamespace(status_code=200, text="proxy landing page")
+
+        with patch.object(system_endpoint, "AsyncRequestUtils", FakeAsyncRequestUtils), patch.object(
+            system_endpoint.settings,
+            "PIP_PROXY",
+            "https://pypi.tuna.tsinghua.edu.cn/simple/",
+        ):
+            resp = asyncio.run(
+                system_endpoint.nettest(
+                    target_id="pip_proxy",
+                    _="token",
+                )
+            )
+
+        self.assertFalse(resp.success)
+        self.assertIn("PIP加速代理", resp.message)
+
+
+if __name__ == "__main__":
+    unittest.main()