From 1d84d17dc4b74c3f50a27d85bd0f0900d97bc089 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Thu, 24 Sep 2026 12:25:45 +0800 Subject: [PATCH] feat(system): add protocol-aware network checks --- app/agent/policy/mcp.py | 2 +- .../policy/resources/api_mcp_schema.json | 4 +- app/api/endpoints/system.py | 14 +- app/application/network.py | 534 -------------- app/application/network/__init__.py | 29 + app/application/network/catalogue.py | 650 ++++++++++++++++++ app/application/network/domain.py | 142 ++++ app/application/network/service.py | 298 ++++++++ app/schemas/system.py | 11 +- app/startup/composition/network.py | 98 ++- tests/test_system_nettest.py | 2 +- 11 files changed, 1222 insertions(+), 562 deletions(-) delete mode 100644 app/application/network.py create mode 100644 app/application/network/__init__.py create mode 100644 app/application/network/catalogue.py create mode 100644 app/application/network/domain.py create mode 100644 app/application/network/service.py diff --git a/app/agent/policy/mcp.py b/app/agent/policy/mcp.py index 7d06e65ed..879800892 100644 --- a/app/agent/policy/mcp.py +++ b/app/agent/policy/mcp.py @@ -187,7 +187,7 @@ OPERATION_DESCRIPTIONS = { "system.module.catalog": "List MoviePilot host modules and service-type options for UI selectors.", "system.module.settings": "List built-in MoviePilot modules that can be enabled or disabled by an administrator.", "system.module.test": "Run the built-in availability test for one enabled MoviePilot module.", - "system.network.targets": "List approved built-in network-test targets without exposing their request URLs.", + "system.network.targets": "List approved network-test targets with sanitized destination addresses, without exposing request paths, query parameters, or credentials.", "system.network.test": "Test connectivity to one approved target or the legacy constrained URL input.", "torrent.cache.clear": "Delete every cached torrent context.", "torrent.cache.delete": "Delete one cached torrent context by site domain and cache hash.", diff --git a/app/agent/policy/resources/api_mcp_schema.json b/app/agent/policy/resources/api_mcp_schema.json index 3b65cc469..337755611 100644 --- a/app/agent/policy/resources/api_mcp_schema.json +++ b/app/agent/policy/resources/api_mcp_schema.json @@ -14764,7 +14764,7 @@ }, { "additionalProperties": false, - "description": "List approved built-in network-test targets without exposing their request URLs. Method: GET. Path: /api/v1/system/nettest/targets. Effect: safe_read. Collection response: data remains a list; omit both page and count to preserve the legacy complete result. Successful gateway output adds collection.result_count and the exact collection.total_count. For a count or summary, send page=1 and count=1, then read collection.total_count; do not query the database merely because item data is truncated.", + "description": "List approved network-test targets with sanitized destination addresses, without exposing request paths, query parameters, or credentials. Method: GET. Path: /api/v1/system/nettest/targets. Effect: safe_read. Collection response: data remains a list; omit both page and count to preserve the legacy complete result. Successful gateway output adds collection.result_count and the exact collection.total_count. For a count or summary, send page=1 and count=1, then read collection.total_count; do not query the database merely because item data is truncated.", "properties": { "operation_id": { "const": "system.network.targets", @@ -14773,7 +14773,7 @@ }, "query": { "additionalProperties": false, - "description": "Filters and control values for system.network.targets. List approved built-in network-test targets without exposing their request URLs. Use only the named fields below.", + "description": "Filters and control values for system.network.targets. List approved network-test targets with sanitized destination addresses, without exposing request paths, query parameters, or credentials. Use only the named fields below.", "properties": { "count": { "anyOf": [ diff --git a/app/api/endpoints/system.py b/app/api/endpoints/system.py index 23fd3ce82..e782eac23 100644 --- a/app/api/endpoints/system.py +++ b/app/api/endpoints/system.py @@ -880,13 +880,21 @@ async def nettest_targets( """ 获取网络测试目标。 - 这里只返回前端渲染所需的最小信息，避免把可请求 URL、内容校验规则和 - 跳转白名单暴露给客户端。 + 这里只返回前端渲染所需的最小信息；展示地址只包含来源主机，隐藏请求路径、 + 查询参数和凭据，内容校验规则与跳转白名单也不会暴露给客户端。 """ targets = get_configured_network_test_service().list_targets() return _SchemaResponse( success=True, - data=[{"id": item.id, "name": item.name, "icon": item.icon} for item in targets], + data=[ + { + "id": item.id, + "name": item.name, + "address": item.address, + "icon": item.icon, + } + for item in targets + ], ) diff --git a/app/application/network.py b/app/application/network.py deleted file mode 100644 index ca8d279dd..000000000 --- a/app/application/network.py +++ /dev/null @@ -1,534 +0,0 @@ -"""受控外部网络探测应用服务。""" - -from collections.abc import Callable, Mapping -from dataclasses import dataclass -from time import monotonic -from typing import Any, Optional, Protocol -from urllib.parse import urljoin, urlparse - -from app.foundation.url import UrlUtils - -_REDIRECT_STATUS_CODES = {301, 302, 303, 307, 308} -_MAX_REDIRECTS = 3 - - -class NetworkTestResponse(Protocol): - """网络探测所需的最小异步响应合同。""" - - status_code: int - headers: Mapping[str, str] - text: str - - async def aclose(self) -> None: - """释放响应占用的传输资源。""" - - -class NetworkTestTransport(Protocol): - """网络探测使用的受限 HTTPS GET 端口。""" - - async def get( - self, - url: str, - *, - proxy: Any = None, - headers: Optional[Mapping[str, str]] = None, - user_agent: Optional[str] = None, - ) -> Optional[NetworkTestResponse]: - """关闭自动重定向并校验证书后请求指定 HTTPS 地址。""" - - -class NetworkTestLogger(Protocol): - """网络探测所需的最小日志端口。""" - - def debug(self, msg: str, *args: Any, **kwargs: Any) -> None: - """记录不影响探测结果的诊断信息。""" - - def warning(self, msg: str, *args: Any, **kwargs: Any) -> None: - """记录被安全策略阻断的网络行为。""" - - -@dataclass(frozen=True, slots=True) -class NetworkTestTarget: - """可安全返回给客户端的网络测试目标投影。""" - - id: str - name: str - icon: str - - -@dataclass(frozen=True, slots=True) -class NetworkTestRule: - """仅由宿主构造、不会向客户端暴露的完整网络测试规则。""" - - id: str - name: str - icon: str - url: str - proxy: bool - allowed_redirect_prefixes: tuple[str, ...] - expected_text: Optional[str] = None - invalid_message: Optional[str] = None - proxy_name: Optional[str] = None - headers: tuple[tuple[str, str], ...] = () - - def public_target(self) -> NetworkTestTarget: - """投影客户端展示字段，避免泄露 URL、凭据和准入规则。""" - return NetworkTestTarget(id=self.id, name=self.name, icon=self.icon) - - def request_headers(self) -> Optional[dict[str, str]]: - """为单次请求创建与冻结规则隔离的请求头字典。""" - return dict(self.headers) or None - - -@dataclass(frozen=True, slots=True) -class NetworkTestResult: - """网络测试应用结果，不包含 HTTP 响应层对象。""" - - success: bool - message: Optional[str] = None - elapsed_ms: Optional[int] = None - - -SettingsReader = Callable[[str, Any], Any] - - -class NetworkTestService: - """维护目标目录、安全准入并执行服务端受控连通性测试。""" - - def __init__( - self, - *, - transport: NetworkTestTransport, - settings: SettingsReader, - logger: NetworkTestLogger, - ) -> None: - """注入通用传输、部署设置读取和运行日志端口。""" - self._transport = transport - self._settings = settings - self._logger = logger - - def list_targets(self) -> tuple[NetworkTestTarget, ...]: - """返回不含请求细节的当前网络测试目标目录。""" - return tuple(rule.public_target() for rule in self._build_rules()) - - async def execute( - self, - *, - target_id: Optional[str] = None, - url: Optional[str] = None, - include: Optional[str] = None, - ) -> NetworkTestResult: - """解析内置目标、执行安全校验并返回连通性测试结果。""" - target = self._find_rule(target_id=target_id, url=url) - if target is None: - return NetworkTestResult(success=False, message="测试目标不存在") - invalid_message = self._validate_rule(target) - if invalid_message: - self._logger.warning(f"拦截不安全的网络测试地址: {target.url}") - return NetworkTestResult(success=False, message=invalid_message) - if include: - self._logger.debug("nettest include 参数已忽略，改为服务端固定校验") - return await self._request(target) - - async def _request(self, target: NetworkTestRule) -> NetworkTestResult: - """手动处理受控重定向并校验最终响应。""" - started_at = monotonic() - current_url = target.url - redirect_count = 0 - response: Optional[NetworkTestResponse] = None - while redirect_count <= _MAX_REDIRECTS: - response = await self._transport.get( - current_url, - proxy=self._settings("PROXY", None) if target.proxy else None, - headers=target.request_headers(), - user_agent=self._settings("NORMAL_USER_AGENT", None), - ) - if response is None or response.status_code not in _REDIRECT_STATUS_CODES: - break - location = response.headers.get("location") - if not location: - break - next_url = urljoin(current_url, location) - if not self._is_allowed_redirect(next_url, target): - await self._close_response(response) - self._logger.warning( - f"拦截网络测试重定向: {current_url} -> {next_url}" - ) - return NetworkTestResult( - success=False, - message="测试目标发生了未授权跳转", - ) - await self._close_response(response) - response = None - current_url = next_url - redirect_count += 1 - - elapsed_ms = round((monotonic() - started_at) * 1000) - if redirect_count > _MAX_REDIRECTS: - await self._close_response(response) - return NetworkTestResult(success=False, message="测试目标重定向次数过多") - if response is None: - label = target.proxy_name or target.name - return NetworkTestResult( - success=False, - message=f"{label}无法连接", - elapsed_ms=elapsed_ms, - ) - - try: - return self._build_response_result(target, response, elapsed_ms) - finally: - await self._close_response(response) - - @staticmethod - def _build_response_result( - target: NetworkTestRule, - response: NetworkTestResponse, - elapsed_ms: int, - ) -> NetworkTestResult: - """把传输响应归一为稳定的应用结果。""" - if response.status_code == 200: - if target.expected_text and target.expected_text.lower() not in ( - response.text or "" - ).lower(): - return NetworkTestResult( - success=False, - message=target.invalid_message or "无效响应", - elapsed_ms=elapsed_ms, - ) - return NetworkTestResult(success=True, elapsed_ms=elapsed_ms) - if target.proxy_name: - message = f"{target.proxy_name}已失效，错误码：{response.status_code}" - else: - message = f"错误码：{response.status_code}" - if "github" in target.url: - if response.status_code == 401: - message = "Github Token已失效，请检查配置" - elif response.status_code in {403, 429}: - message = "触发限流，请配置Github Token" - return NetworkTestResult( - success=False, - message=message, - elapsed_ms=elapsed_ms, - ) - - def _find_rule( - self, - *, - target_id: Optional[str], - url: Optional[str], - ) -> Optional[NetworkTestRule]: - """优先按正式 target_id、否则按旧 URL 精确匹配内置规则。""" - rules = self._build_rules() - if target_id: - return next((rule for rule in rules if rule.id == target_id), None) - if url: - return next((rule for rule in rules if rule.url == url), None) - return None - - def _validate_rule(self, target: NetworkTestRule) -> Optional[str]: - """兜底校验服务端规则未生成非 HTTPS、带凭据或目录外地址。""" - parsed = urlparse(target.url) - if parsed.scheme.lower() != "https": - return "测试地址仅支持 HTTPS" - if not parsed.netloc: - return "测试地址无效" - if parsed.username or parsed.password: - return "测试地址不支持携带账号信息" - if not any(rule == target for rule in self._build_rules()): - return "测试地址不在允许的测试目标列表中" - return None - - @classmethod - def _is_allowed_redirect(cls, url: str, target: NetworkTestRule) -> bool: - """只允许重定向到当前测试项声明的协议、主机、端口和路径。""" - parsed = urlparse(url) - if parsed.scheme.lower() != "https" or not parsed.netloc: - return False - if parsed.username or parsed.password: - return False - return any( - cls._matches_prefix(url, prefix) - for prefix in target.allowed_redirect_prefixes - ) - - @staticmethod - def _matches_prefix(url: str, prefix: str) -> bool: - """按协议、主机、有效端口和路径前缀匹配允许的跳转范围。""" - parsed_url = urlparse(url) - parsed_prefix = urlparse(prefix) - if parsed_url.scheme.lower() != parsed_prefix.scheme.lower(): - return False - if (parsed_url.hostname or "").lower() != ( - parsed_prefix.hostname or "" - ).lower(): - return False - try: - url_port = parsed_url.port or ( - 443 if parsed_url.scheme.lower() == "https" else 80 - ) - prefix_port = parsed_prefix.port or ( - 443 if parsed_prefix.scheme.lower() == "https" else 80 - ) - except ValueError: - return False - if url_port != prefix_port: - return False - prefix_path = parsed_prefix.path or "/" - if prefix_path.endswith("/"): - return parsed_url.path.startswith(prefix_path) - return parsed_url.path == prefix_path or parsed_url.path.startswith( - f"{prefix_path}/" - ) - - async def _close_response( - self, - response: Optional[NetworkTestResponse], - ) -> None: - """安静释放已读取响应，关闭失败不覆盖真实探测结果。""" - if response is None or not hasattr(response, "aclose"): - return - try: - await response.aclose() - except Exception as err: # noqa: BLE001 - 资源回收失败仅记录诊断 - self._logger.debug(f"关闭网络测试响应失败: {err}") - - def _build_rules(self) -> tuple[NetworkTestRule, ...]: - """根据当前部署设置构建唯一网络测试目录和安全准入规则。""" - github_proxy = UrlUtils.standardize_base_url( - self._settings("GITHUB_PROXY", None) or "" - ) - pip_proxy = UrlUtils.standardize_base_url( - self._settings("PIP_PROXY", None) or "https://pypi.org/simple/" - ) - tmdb_key = self._settings("TMDB_API_KEY", None) - tmdb_domain = ( - self._settings("TMDB_API_DOMAIN", None) or "api.themoviedb.org" - ) - github_headers = self._freeze_headers( - self._settings("GITHUB_HEADERS", None) - ) - github_readme_url = "https://github.com/jxxghp/MoviePilot/blob/v2/README.md" - raw_readme_url = ( - "https://raw.githubusercontent.com/jxxghp/MoviePilot/v2/README.md" - ) - rules = [ - NetworkTestRule( - id="tmdb_api", - name="api.themoviedb.org", - icon="tmdb", - url=f"https://api.themoviedb.org/3/movie/550?api_key={tmdb_key}", - proxy=True, - allowed_redirect_prefixes=("https://api.themoviedb.org/3/",), - ), - NetworkTestRule( - id="tmdb_api_alt", - name="api.tmdb.org", - icon="tmdb", - url=f"https://api.tmdb.org/3/movie/550?api_key={tmdb_key}", - proxy=True, - allowed_redirect_prefixes=("https://api.tmdb.org/3/",), - ), - NetworkTestRule( - id="tmdb_web", - name="www.themoviedb.org", - icon="tmdb", - url="https://www.themoviedb.org", - proxy=True, - allowed_redirect_prefixes=("https://www.themoviedb.org/",), - ), - NetworkTestRule( - id="tvdb_api", - name="api.thetvdb.com", - icon="tvdb", - url="https://api.thetvdb.com/series/81189", - proxy=True, - allowed_redirect_prefixes=("https://api.thetvdb.com/",), - ), - NetworkTestRule( - id="fanart_api", - name="webservice.fanart.tv", - icon="fanart", - url="https://webservice.fanart.tv", - proxy=True, - allowed_redirect_prefixes=("https://webservice.fanart.tv/",), - ), - NetworkTestRule( - id="telegram_api", - name="api.telegram.org", - icon="telegram", - url="https://api.telegram.org", - proxy=True, - allowed_redirect_prefixes=( - "https://api.telegram.org/", - "https://core.telegram.org/", - ), - ), - NetworkTestRule( - id="wechat_api", - name="qyapi.weixin.qq.com", - icon="wechat", - url="https://qyapi.weixin.qq.com/cgi-bin/gettoken", - proxy=False, - allowed_redirect_prefixes=("https://qyapi.weixin.qq.com/",), - ), - NetworkTestRule( - id="douban_api", - name="frodo.douban.com", - icon="douban", - url="https://frodo.douban.com", - proxy=False, - allowed_redirect_prefixes=( - "https://frodo.douban.com/", - "https://www.douban.com/doubanapp/frodo", - ), - ), - NetworkTestRule( - id="slack_api", - name="slack.com", - icon="slack", - url="https://slack.com", - proxy=False, - allowed_redirect_prefixes=( - "https://slack.com/", - "https://www.slack.com/", - ), - ), - NetworkTestRule( - id="pip_proxy", - name="pypi.org", - icon="python", - url=f"{pip_proxy}rsa/", - proxy=True, - allowed_redirect_prefixes=( - pip_proxy, - "https://pypi.org/simple/", - ), - expected_text="pypi:repository-version", - invalid_message="PIP加速代理已失效，请检查配置", - proxy_name="PIP加速代理", - ), - NetworkTestRule( - id="github_proxy_web", - name="github.com", - icon="github", - url=( - f"{github_proxy}{github_readme_url}" - if github_proxy - else github_readme_url - ), - proxy=True, - allowed_redirect_prefixes=( - "https://github.com/", - *( - (f"{github_proxy}https://github.com/",) - if github_proxy - else () - ), - ), - expected_text="MoviePilot", - invalid_message=( - "Github加速代理已失效，请检查配置" - if github_proxy - else "无效响应" - ), - proxy_name="Github加速代理" if github_proxy else None, - headers=github_headers, - ), - NetworkTestRule( - id="github_api", - name="api.github.com", - icon="github", - url="https://api.github.com", - proxy=True, - allowed_redirect_prefixes=("https://api.github.com/",), - headers=github_headers, - ), - NetworkTestRule( - id="github_codeload", - name="codeload.github.com", - icon="github", - url="https://codeload.github.com", - proxy=True, - allowed_redirect_prefixes=( - "https://codeload.github.com/", - "https://github.com/", - ), - headers=github_headers, - ), - NetworkTestRule( - id="github_proxy_raw", - name="raw.githubusercontent.com", - icon="github", - url=( - f"{github_proxy}{raw_readme_url}" - if github_proxy - else raw_readme_url - ), - proxy=True, - allowed_redirect_prefixes=( - "https://raw.githubusercontent.com/", - *( - (f"{github_proxy}https://raw.githubusercontent.com/",) - if github_proxy - else () - ), - ), - expected_text="MoviePilot", - invalid_message=( - "Github加速代理已失效，请检查配置" - if github_proxy - else "无效响应" - ), - proxy_name="Github加速代理" if github_proxy else None, - headers=github_headers, - ), - ] - if tmdb_domain not in {"api.themoviedb.org", "api.tmdb.org"}: - rules.insert( - 2, - NetworkTestRule( - id="tmdb_api_configured", - name=str(tmdb_domain), - icon="tmdb", - url=f"https://{tmdb_domain}/3/movie/550?api_key={tmdb_key}", - proxy=True, - allowed_redirect_prefixes=( - f"https://{tmdb_domain}/3/", - ), - ), - ) - return tuple(rules) - - @staticmethod - def _freeze_headers(value: Any) -> tuple[tuple[str, str], ...]: - """把动态配置中的请求头复制为不可变、可比较的规则字段。""" - if not isinstance(value, Mapping): - return () - return tuple( - (str(key), str(item)) - for key, item in value.items() - if item is not None - ) - - -_configured_network_test_service: Optional[NetworkTestService] = None - - -def configure_network_test_service(service: NetworkTestService) -> None: - """由启动组合根登记唯一的网络测试应用服务。""" - global _configured_network_test_service - _configured_network_test_service = service - - -def reset_network_test_service() -> None: - """清除当前 lifespan 的网络测试应用服务。""" - global _configured_network_test_service - _configured_network_test_service = None - - -def get_configured_network_test_service() -> NetworkTestService: - """返回启动阶段登记的网络测试应用服务。""" - if _configured_network_test_service is None: - raise RuntimeError("网络测试服务尚未装配") - return _configured_network_test_service diff --git a/app/application/network/__init__.py b/app/application/network/__init__.py new file mode 100644 index 000000000..a741e9f9b --- /dev/null +++ b/app/application/network/__init__.py @@ -0,0 +1,29 @@ +"""网络探测应用服务及其公开类型。""" + +from .domain import ( + NetworkTestLogger, + NetworkTestResponse, + NetworkTestResult, + NetworkTestRule, + NetworkTestTarget, + NetworkTestTransport, +) +from .service import ( + NetworkTestService, + configure_network_test_service, + get_configured_network_test_service, + reset_network_test_service, +) + +__all__ = [ + "NetworkTestLogger", + "NetworkTestResponse", + "NetworkTestResult", + "NetworkTestRule", + "NetworkTestService", + "NetworkTestTarget", + "NetworkTestTransport", + "configure_network_test_service", + "get_configured_network_test_service", + "reset_network_test_service", +] diff --git a/app/application/network/catalogue.py b/app/application/network/catalogue.py new file mode 100644 index 000000000..d1653430d --- /dev/null +++ b/app/application/network/catalogue.py @@ -0,0 +1,650 @@ +"""按配置生成可控的网络连通性探测目标目录。""" + +from collections.abc import Callable, Collection, Mapping +from typing import Any, Optional +from urllib.parse import urlparse + +from app.foundation.url import UrlUtils + +from .domain import NetworkTestRule, SettingsReader + + +def _configured_service_rule( + *, + target_id: str, + url: str, + icon: str, + proxy: bool, + module_ids: tuple[str, ...] = (), + name_suffix: Optional[str] = None, + request_path: str = "", +) -> Optional[NetworkTestRule]: + """为 HTTPS 服务生成隐藏真实路径的目录项和同源跳转范围。""" + try: + parsed = urlparse(url) + hostname = parsed.hostname + except ValueError: + return None + if ( + parsed.scheme.lower() != "https" + or not parsed.netloc + or not hostname + or parsed.username + or parsed.password + or parsed.query + or parsed.fragment + ): + return None + try: + parsed.port + except ValueError: + return None + host_label = _address_label(url) or hostname + base_url = url if url.endswith("/") else f"{url}/" + target_url = f"{base_url}{request_path.lstrip('/')}" + redirect_prefix = f"{parsed.scheme}://{parsed.netloc}{parsed.path or '/'}" + if not redirect_prefix.endswith("/"): + redirect_prefix = f"{redirect_prefix}/" + return NetworkTestRule( + id=target_id, + name=name_suffix or host_label, + icon=icon, + url=target_url, + proxy=proxy, + allowed_redirect_prefixes=(redirect_prefix,), + module_ids=module_ids, + ) + + +def _address_label(url: str) -> Optional[str]: + """从 URL 中提取不含凭据的主机和显式端口作为目标名称。""" + try: + parsed = urlparse(url) + hostname = parsed.hostname + port = parsed.port + except ValueError: + return None + if not hostname: + return None + if port and port != 443: + return f"{hostname}:{port}" + return hostname + + +def _freeze_headers(value: Any) -> tuple[tuple[str, str], ...]: + """把动态配置中的请求头复制为不可变、可比较的规则字段。""" + if not isinstance(value, Mapping): + return () + return tuple((str(key), str(item)) for key, item in value.items() if item is not None) + + +def build_network_rules( + settings: SettingsReader, + enabled_module_ids_reader: Optional[Callable[[], Collection[str]]], +) -> tuple[NetworkTestRule, ...]: + """组合外部模块、代理和固定服务目标，并按模块开关裁剪目录。""" + enabled_module_ids = set(enabled_module_ids_reader()) if enabled_module_ids_reader is not None else None + github_proxy = UrlUtils.standardize_base_url(settings("GITHUB_PROXY", None) or "") + pip_proxy = UrlUtils.standardize_base_url(settings("PIP_PROXY", None) or "https://pypi.org/simple/") + rules = _build_network_proxy_rules(settings, github_proxy, pip_proxy) + rules.extend(_build_metadata_module_rules(settings)) + rules.extend(_build_notification_module_rules()) + rules.extend(_build_external_media_rules(settings)) + rules.extend(_build_music_service_rules()) + rules.extend(_build_configured_module_rules(settings)) + rules.extend(_build_fixed_service_rules(settings, github_proxy)) + if enabled_module_ids is None: + return tuple(rules) + return tuple(rule for rule in rules if not rule.module_ids or enabled_module_ids.intersection(rule.module_ids)) + + +def _build_metadata_module_rules(settings: SettingsReader) -> list[NetworkTestRule]: + """构建 TMDB、TVDB、Fanart 和豆瓣等启用时使用的媒体服务目标。""" + tmdb_key = settings("TMDB_API_KEY", None) + tmdb_domain = settings("TMDB_API_DOMAIN", None) or "api.themoviedb.org" + rules = [ + NetworkTestRule( + id="tmdb_api", + name="TMDB API", + icon="tmdb", + url=f"https://api.themoviedb.org/3/movie/550?api_key={tmdb_key}", + proxy=True, + allowed_redirect_prefixes=("https://api.themoviedb.org/3/",), + module_ids=("TheMovieDbModule",), + ), + NetworkTestRule( + id="tmdb_api_alt", + name="TMDB API 备用域名", + icon="tmdb", + url=f"https://api.tmdb.org/3/movie/550?api_key={tmdb_key}", + proxy=True, + allowed_redirect_prefixes=("https://api.tmdb.org/3/",), + module_ids=("TheMovieDbModule",), + ), + NetworkTestRule( + id="tmdb_web", + name="TMDB 网站", + icon="tmdb", + url="https://www.themoviedb.org", + proxy=True, + allowed_redirect_prefixes=("https://www.themoviedb.org/",), + module_ids=("TheMovieDbModule",), + ), + NetworkTestRule( + id="tvdb_api", + name="TVDB API", + icon="tvdb", + url="https://api4.thetvdb.com/v4/series/81189", + proxy=True, + allowed_redirect_prefixes=("https://api4.thetvdb.com/v4/",), + module_ids=("TheTvDbModule",), + success_status_codes=(200, 401), + ), + NetworkTestRule( + id="fanart_api", + name="Fanart 图库服务", + icon="fanart", + url="https://webservice.fanart.tv", + proxy=True, + allowed_redirect_prefixes=("https://webservice.fanart.tv/",), + module_ids=("FanartModule",), + ), + NetworkTestRule( + id="douban_api", + name="豆瓣媒体服务", + icon="douban", + url="https://frodo.douban.com", + proxy=False, + allowed_redirect_prefixes=( + "https://frodo.douban.com/", + "https://www.douban.com/doubanapp/frodo", + ), + module_ids=("DoubanModule",), + ), + NetworkTestRule( + id="douban_web", + name="豆瓣网页服务", + icon="douban", + url="https://movie.douban.com", + proxy=True, + allowed_redirect_prefixes=("https://movie.douban.com/",), + module_ids=("DoubanModule",), + ), + ] + if tmdb_domain not in {"api.themoviedb.org", "api.tmdb.org"}: + rules.insert( + 2, + NetworkTestRule( + id="tmdb_api_configured", + name="TMDB API 自定义域名", + icon="tmdb", + url=f"https://{tmdb_domain}/3/movie/550?api_key={tmdb_key}", + proxy=True, + allowed_redirect_prefixes=(f"https://{tmdb_domain}/3/",), + module_ids=("TheMovieDbModule",), + ), + ) + return rules + + +def _build_notification_module_rules() -> list[NetworkTestRule]: + """构建已启用通知渠道对应的固定服务目标。""" + return [ + NetworkTestRule( + id="telegram_api", + name="Telegram 通知", + icon="telegram", + url="https://api.telegram.org", + proxy=True, + allowed_redirect_prefixes=( + "https://api.telegram.org/", + "https://core.telegram.org/", + ), + module_ids=("TelegramModule",), + ), + NetworkTestRule( + id="wechat_api", + name="企业微信通知", + icon="wechat", + url="https://qyapi.weixin.qq.com/cgi-bin/gettoken", + proxy=False, + allowed_redirect_prefixes=("https://qyapi.weixin.qq.com/",), + module_ids=("WechatModule",), + ), + NetworkTestRule( + id="wechat_bot_websocket", + name="企业微信机器人长连接", + icon="wechat", + url="wss://openws.work.weixin.qq.com", + proxy=False, + allowed_redirect_prefixes=(), + probe_protocol="websocket", + module_ids=("WechatModule",), + ), + NetworkTestRule( + id="slack_api", + name="Slack 通知", + icon="slack", + url="https://slack.com", + proxy=False, + allowed_redirect_prefixes=( + "https://slack.com/", + "https://www.slack.com/", + ), + module_ids=("SlackModule",), + ), + NetworkTestRule( + id="dingtalk_api", + name="钉钉通知", + icon="dingtalk", + url="https://oapi.dingtalk.com", + proxy=False, + allowed_redirect_prefixes=("https://oapi.dingtalk.com/",), + module_ids=("DingTalkModule",), + ), + NetworkTestRule( + id="discord_api", + name="Discord 通知", + icon="discord", + url="https://discord.com", + proxy=True, + allowed_redirect_prefixes=("https://discord.com/",), + module_ids=("DiscordModule",), + ), + NetworkTestRule( + id="feishu_api", + name="飞书通知", + icon="feishu", + url="https://open.feishu.cn", + proxy=False, + allowed_redirect_prefixes=("https://open.feishu.cn/",), + module_ids=("FeishuModule",), + ), + NetworkTestRule( + id="qqbot_api", + name="QQ 机器人 API", + icon="qq", + url="https://api.sgroup.qq.com/gateway", + proxy=False, + allowed_redirect_prefixes=("https://api.sgroup.qq.com/",), + module_ids=("QQBotModule",), + success_status_codes=(200, 400, 401, 403, 404, 405), + ), + NetworkTestRule( + id="qqbot_auth", + name="QQ 机器人授权服务", + icon="qq", + url="https://bots.qq.com/app/getAppAccessToken", + proxy=False, + allowed_redirect_prefixes=("https://bots.qq.com/",), + http_method="POST", + request_json=(), + module_ids=("QQBotModule",), + success_status_codes=(200, 400, 401, 403, 404, 405), + ), + NetworkTestRule( + id="wechatclawbot_api", + name="微信 ClawBot 服务", + icon="wechat", + url="https://ilinkai.weixin.qq.com/ilink/bot/getconfig", + proxy=True, + allowed_redirect_prefixes=("https://ilinkai.weixin.qq.com/",), + http_method="POST", + request_json=(), + module_ids=("WechatClawBotModule",), + success_status_codes=(200, 400, 401, 404, 405), + display_address="https://ilinkai.weixin.qq.com", + ), + NetworkTestRule( + id="wechatclawbot_cdn", + name="微信 ClawBot 媒体 CDN", + icon="wechat", + url="https://novac2c.cdn.weixin.qq.com/c2c", + proxy=True, + allowed_redirect_prefixes=("https://novac2c.cdn.weixin.qq.com/",), + module_ids=("WechatClawBotModule",), + success_status_codes=(200, 403, 404), + ), + ] + + +def _build_external_media_rules(settings: SettingsReader) -> list[NetworkTestRule]: + """构建 AniList、IMDb 与 AcoustID 等外部媒体来源探测目标。""" + return [ + NetworkTestRule( + id="acoustid_api", + name="AcoustID 音频识别", + icon="site", + url=(f"https://api.acoustid.org/v2/lookup?client={settings('ACOUSTID_API_KEY', '')}&format=json"), + proxy=True, + allowed_redirect_prefixes=("https://api.acoustid.org/",), + module_ids=("AcoustIdModule",), + success_status_codes=(200, 400), + ), + NetworkTestRule( + id="anilist_api", + name="AniList 番剧元数据", + icon="bangumi", + url=("https://graphql.anilist.co/?query=%7B%20Media%28id%3A1%29%20%7Bid%7D%20%7D"), + proxy=True, + allowed_redirect_prefixes=("https://graphql.anilist.co/",), + module_ids=("AniListModule",), + ), + NetworkTestRule( + id="anilist_trace", + name="trace.moe 番剧识别", + icon="bangumi", + url="https://trace.moe", + proxy=True, + allowed_redirect_prefixes=("https://trace.moe/",), + module_ids=("AniListModule",), + ), + NetworkTestRule( + id="imdb_api", + name="IMDb 元数据", + icon="tmdb", + url="https://v2.sg.media-imdb.com/suggestion/x/inception.json", + proxy=True, + allowed_redirect_prefixes=("https://v2.sg.media-imdb.com/",), + module_ids=("ImdbModule",), + ), + NetworkTestRule( + id="imdb_graphql", + name="IMDb GraphQL 服务", + icon="tmdb", + url="https://caching.graphql.imdb.com", + proxy=True, + allowed_redirect_prefixes=("https://caching.graphql.imdb.com/",), + module_ids=("ImdbModule",), + ), + ] + + +def _build_music_service_rules() -> list[NetworkTestRule]: + """构建 MusicBrainz、ListenBrainz 和 TheAudioDB 服务目标。""" + return [ + NetworkTestRule( + id="musicbrainz_api", + name="MusicBrainz 音乐信息", + icon="site", + url=("https://musicbrainz.org/ws/2/recording?query=recording%3Aradiohead&fmt=json"), + proxy=True, + allowed_redirect_prefixes=("https://musicbrainz.org/",), + module_ids=("MusicBrainzModule", "ListenBrainzModule"), + ), + NetworkTestRule( + id="listenbrainz_api", + name="ListenBrainz 音乐统计", + icon="site", + url=("https://api.listenbrainz.org/1/stats/sitewide/artists?range=all_time&count=1"), + proxy=True, + allowed_redirect_prefixes=("https://api.listenbrainz.org/",), + module_ids=("ListenBrainzModule",), + ), + NetworkTestRule( + id="theaudiodb_api", + name="TheAudioDB 音乐信息", + icon="site", + url=("https://www.theaudiodb.com/api/v1/json/2/searchalbum.php?s=radiohead"), + proxy=True, + allowed_redirect_prefixes=("https://www.theaudiodb.com/",), + module_ids=("TheAudioDbModule",), + ), + ] + + +def _build_configured_module_rules(settings: SettingsReader) -> list[NetworkTestRule]: + """根据模块配置构建 Bangumi、封面、歌词和自定义代理目标。""" + rules: list[NetworkTestRule] = [] + bangumi_proxy = bool(settings("BANGUMI_PROXY_ENABLE", False)) + bangumi_domain = ( + (settings("BANGUMI_API_DOMAIN", None) or "https://api.bgm.tv") if bangumi_proxy else "https://api.bgm.tv" + ) + bangumi_rule = _configured_service_rule( + target_id="bangumi_api", + url=str(bangumi_domain or ""), + icon="bangumi", + proxy=True, + module_ids=("BangumiModule",), + name_suffix="Bangumi API", + ) + if bangumi_rule: + rules.append(bangumi_rule) + + configured_targets = ( + ( + "bangumi_image_proxy", + "BANGUMI_IMAGE_DOMAIN", + "bangumi", + ("BangumiModule",), + "", + "Bangumi 图片代理", + ), + ( + "music_cover_proxy", + "MUSIC_COVER_PROXY", + "site", + ("MusicBrainzModule", "ListenBrainzModule"), + "", + "音乐封面代理", + ), + ( + "lrclib_api", + "LRCLIB_BASE_URL", + "site", + ("LrclibModule",), + "/api/search?track_name=test", + "LRCLIB 歌词服务", + ), + ( + "amll_api", + "AMLL_BASE_URL", + "site", + ("AmllModule",), + "/v1/lyrics/search?musicName=test&page=1&pageSize=1", + "AMLL 歌词服务", + ), + ) + for target_id, setting_key, icon, module_ids, request_path, name in configured_targets: + default_url = "https://coverartarchive.org" if target_id == "music_cover_proxy" else "" + configured_url = str(settings(setting_key, default_url) or default_url).strip() + if not configured_url: + continue + rule = _configured_service_rule( + target_id=target_id, + url=configured_url, + icon=icon, + proxy=True, + module_ids=module_ids, + name_suffix=name, + request_path=request_path, + ) + if rule: + rules.append(rule) + return rules + + +def _build_network_proxy_rules( + settings: SettingsReader, + github_proxy: str, + pip_proxy: str, +) -> list[NetworkTestRule]: + """构建全局代理、GitHub 与 Python 包镜像的连通性目标。""" + rules: list[NetworkTestRule] = [] + network_proxy_rule = _build_network_proxy_rule(settings) + if network_proxy_rule: + rules.append(network_proxy_rule) + + configured_pip_proxy = settings("PIP_PROXY", None) + rules.append( + NetworkTestRule( + id="pip_proxy", + name="PIP 加速代理" if configured_pip_proxy else "PyPI 软件源", + icon="python", + url=f"{pip_proxy}rsa/", + proxy=True, + allowed_redirect_prefixes=(pip_proxy, "https://pypi.org/simple/"), + expected_text="pypi:repository-version", + invalid_message="PIP加速代理已失效，请检查配置", + proxy_name="PIP加速代理", + ) + ) + rules.extend(_build_github_proxy_rules(settings, github_proxy)) + return rules + + +def _build_network_proxy_rule(settings: SettingsReader) -> Optional[NetworkTestRule]: + """用 GitHub API 请求验证当前配置的通用出站代理。""" + proxy_config = settings("PROXY", None) + proxy_address = settings("PROXY_HOST", None) + if not proxy_address and isinstance(proxy_config, Mapping): + proxy_address = proxy_config.get("https") or proxy_config.get("http") + if not proxy_address: + return None + proxy_url = str(proxy_address).strip() + if not (_address_label(proxy_url) or _address_label(f"http://{proxy_url}")): + return None + return NetworkTestRule( + id="network_proxy", + name="网络代理", + icon="site", + url="https://api.github.com", + proxy=True, + allowed_redirect_prefixes=("https://api.github.com/",), + display_address=(proxy_url if "://" in proxy_url else f"http://{proxy_url}"), + ) + + +def _build_github_proxy_rules( + settings: SettingsReader, + github_proxy: str, +) -> list[NetworkTestRule]: + """构建 GitHub 页面、API、源码和归档下载连通性目标。""" + headers = _freeze_headers(settings("GITHUB_HEADERS", None)) + github_readme_url = "https://github.com/jxxghp/MoviePilot/blob/v2/README.md" + raw_readme_url = "https://raw.githubusercontent.com/jxxghp/MoviePilot/v2/README.md" + return [ + NetworkTestRule( + id="github_proxy_web", + name="GitHub 网页访问", + icon="github", + url=(f"{github_proxy}{github_readme_url}" if github_proxy else github_readme_url), + proxy=True, + allowed_redirect_prefixes=( + "https://github.com/", + *((f"{github_proxy}https://github.com/",) if github_proxy else ()), + ), + expected_text="MoviePilot", + invalid_message=("Github加速代理已失效，请检查配置" if github_proxy else "无效响应"), + proxy_name="Github加速代理" if github_proxy else None, + headers=headers, + display_address=github_proxy or github_readme_url, + ), + NetworkTestRule( + id="github_api", + name="GitHub API", + icon="github", + url="https://api.github.com", + proxy=True, + allowed_redirect_prefixes=("https://api.github.com/",), + headers=headers, + ), + NetworkTestRule( + id="github_codeload", + name="GitHub 归档下载", + icon="github", + url="https://codeload.github.com", + proxy=True, + allowed_redirect_prefixes=( + "https://codeload.github.com/", + "https://github.com/", + ), + headers=headers, + ), + NetworkTestRule( + id="github_proxy_raw", + name="GitHub 源码下载", + icon="github", + url=(f"{github_proxy}{raw_readme_url}" if github_proxy else raw_readme_url), + proxy=True, + allowed_redirect_prefixes=( + "https://raw.githubusercontent.com/", + *((f"{github_proxy}https://raw.githubusercontent.com/",) if github_proxy else ()), + ), + expected_text="MoviePilot", + invalid_message=("Github加速代理已失效，请检查配置" if github_proxy else "无效响应"), + proxy_name="Github加速代理" if github_proxy else None, + headers=headers, + display_address=github_proxy or raw_readme_url, + ), + ] + + +def _build_fixed_service_rules( + settings: SettingsReader, + github_proxy: str, +) -> list[NetworkTestRule]: + """保留插件市场和 MoviePilot 等非模块固定服务目标。""" + rules: list[NetworkTestRule] = [] + plugin_repo_url = "https://github.com/jxxghp/MoviePilot-Plugins" + plugin_repo_prefix = f"{plugin_repo_url}/" + plugin_proxy_prefix = f"{github_proxy}{plugin_repo_prefix}" if github_proxy else None + rules.append( + NetworkTestRule( + id="plugin_market", + name="插件升级与安装", + icon="github", + url=(f"{github_proxy}{plugin_repo_url}" if github_proxy else plugin_repo_url), + proxy=True, + allowed_redirect_prefixes=( + plugin_repo_prefix, + *((plugin_proxy_prefix,) if plugin_proxy_prefix else ()), + ), + display_address=github_proxy or plugin_repo_url, + ) + ) + + services = ( + ( + "moviepilot_server", + "MP_SERVER_HOST", + "MoviePilot 中心服务", + "https://movie-pilot.org", + ), + ( + "ocr_service", + "OCR_HOST", + "验证码识别服务", + "https://movie-pilot.org", + ), + ( + "cookiecloud_service", + "COOKIECLOUD_HOST", + "CookieCloud 服务", + "https://movie-pilot.org/cookiecloud", + ), + ( + "u115_auth_service", + "U115_AUTH_SERVER", + "115 授权服务", + "https://movie-pilot.org", + ), + ) + for target_id, setting_key, service_name, default_url in services: + if setting_key == "COOKIECLOUD_HOST" and ( + not settings("COOKIECLOUD_KEY", None) + or not settings("COOKIECLOUD_PASSWORD", None) + or settings("COOKIECLOUD_ENABLE_LOCAL", False) + ): + continue + url = str(settings(setting_key, default_url) or "").strip() + rule = _configured_service_rule( + target_id=target_id, + url=url, + icon="site", + proxy=True, + name_suffix=service_name, + ) + if rule: + rules.append(rule) + return rules diff --git a/app/application/network/domain.py b/app/application/network/domain.py new file mode 100644 index 000000000..2c75e994f --- /dev/null +++ b/app/application/network/domain.py @@ -0,0 +1,142 @@ +"""网络探测规则、结果和传输端口类型。""" + +from collections.abc import Callable, Mapping +from dataclasses import dataclass +from typing import Any, Optional, Protocol +from urllib.parse import urlparse + + +class NetworkTestResponse(Protocol): + """网络探测所需的最小异步响应合同。""" + + status_code: int + headers: Mapping[str, str] + text: str + + async def aclose(self) -> None: + """释放响应占用的传输资源。""" + + +class NetworkTestTransport(Protocol): + """网络探测使用的受限 HTTP 与 WebSocket 传输端口。""" + + async def get( + self, + url: str, + *, + proxy: Any = None, + headers: Optional[Mapping[str, str]] = None, + user_agent: Optional[str] = None, + ) -> Optional[NetworkTestResponse]: + """关闭自动重定向并校验证书后请求指定 HTTPS 地址。""" + + async def request( + self, + method: str, + url: str, + *, + proxy: Any = None, + headers: Optional[Mapping[str, str]] = None, + user_agent: Optional[str] = None, + json_body: Optional[Mapping[str, Any]] = None, + ) -> Optional[NetworkTestResponse]: + """使用指定 HTTP 方法请求 HTTPS 地址，并关闭自动重定向。""" + + async def websocket( + self, + url: str, + *, + headers: Optional[Mapping[str, str]] = None, + user_agent: Optional[str] = None, + timeout: float = 10, + ) -> bool: + """执行 WebSocket 握手并在确认连接后立即关闭。""" + + +class NetworkTestLogger(Protocol): + """网络探测所需的最小日志端口。""" + + def debug(self, msg: str, *args: Any, **kwargs: Any) -> None: + """记录不影响探测结果的诊断信息。""" + + def warning(self, msg: str, *args: Any, **kwargs: Any) -> None: + """记录被安全策略阻断的网络行为。""" + + +@dataclass(frozen=True, slots=True) +class NetworkTestTarget: + """可安全返回给客户端的网络测试目标投影。""" + + id: str + name: str + address: str + icon: str + + +@dataclass(frozen=True, slots=True) +class NetworkTestRule: + """仅由宿主构造、不会向客户端暴露的完整网络测试规则。""" + + id: str + name: str + icon: str + url: str + proxy: bool + allowed_redirect_prefixes: tuple[str, ...] + probe_protocol: str = "http" + http_method: str = "GET" + request_json: Optional[tuple[tuple[str, Any], ...]] = None + module_ids: tuple[str, ...] = () + success_status_codes: tuple[int, ...] = (200,) + expected_text: Optional[str] = None + invalid_message: Optional[str] = None + proxy_name: Optional[str] = None + headers: tuple[tuple[str, str], ...] = () + display_address: Optional[str] = None + + def public_target(self) -> NetworkTestTarget: + """投影展示字段，只公开目标主机而不暴露请求路径和凭据。""" + return NetworkTestTarget( + id=self.id, + name=self.name, + address=_safe_display_address(self.display_address or self.url), + icon=self.icon, + ) + + def request_headers(self) -> Optional[dict[str, str]]: + """为单次请求创建与冻结规则隔离的请求头字典。""" + return dict(self.headers) or None + + def request_body(self) -> Optional[dict[str, Any]]: + """为 HTTP 探测创建隔离的 JSON 请求体。""" + return dict(self.request_json) if self.request_json is not None else None + + +@dataclass(frozen=True, slots=True) +class NetworkTestResult: + """网络测试应用结果，不包含 HTTP 响应层对象。""" + + success: bool + message: Optional[str] = None + elapsed_ms: Optional[int] = None + + +SettingsReader = Callable[[str, Any], Any] + + +def _safe_display_address(url: str) -> str: + """只返回来源协议、主机和端口，隐藏路径、查询参数和凭据。""" + try: + parsed = urlparse(url) + hostname = parsed.hostname + port = parsed.port + except ValueError: + return "" + scheme = parsed.scheme.lower() + if scheme not in {"http", "https", "ws", "wss", "socks5", "socks5h"} or not hostname: + return "" + display_host = f"[{hostname}]" if ":" in hostname else hostname + default_ports = {"http": 80, "https": 443, "ws": 80, "wss": 443} + if port and port != default_ports.get(scheme): + display_host = f"{display_host}:{port}" + return f"{scheme}://{display_host}" diff --git a/app/application/network/service.py b/app/application/network/service.py new file mode 100644 index 000000000..d0dff579f --- /dev/null +++ b/app/application/network/service.py @@ -0,0 +1,298 @@ +"""执行受限网络连通性探测并管理其应用服务生命周期。""" + +from collections.abc import Collection, Mapping +from time import monotonic +from typing import Any, Callable, Optional +from urllib.parse import urljoin, urlparse + +from .catalogue import build_network_rules +from .domain import ( + NetworkTestLogger, + NetworkTestResponse, + NetworkTestResult, + NetworkTestRule, + NetworkTestTarget, + NetworkTestTransport, + SettingsReader, +) + +_REDIRECT_STATUS_CODES = {301, 302, 303, 307, 308} +_MAX_REDIRECTS = 3 + + +class NetworkTestService: + """维护目标目录、安全准入并执行服务端受控连通性测试。""" + + def __init__( + self, + *, + transport: NetworkTestTransport, + settings: SettingsReader, + logger: NetworkTestLogger, + enabled_module_ids: Optional[Callable[[], Collection[str]]] = None, + ) -> None: + """注入网络端口、部署设置和当前启用模块读取器。""" + self._transport = transport + self._settings = settings + self._logger = logger + self._enabled_module_ids = enabled_module_ids + + def list_targets(self) -> tuple[NetworkTestTarget, ...]: + """返回不含请求细节的当前网络测试目标目录。""" + return tuple(rule.public_target() for rule in self._build_rules()) + + async def execute( + self, + *, + target_id: Optional[str] = None, + url: Optional[str] = None, + include: Optional[str] = None, + ) -> NetworkTestResult: + """解析内置目标、执行安全校验并返回连通性测试结果。""" + target = self._find_rule(target_id=target_id, url=url) + if target is None: + return NetworkTestResult(success=False, message="测试目标不存在") + invalid_message = self._validate_rule(target) + if invalid_message: + self._logger.warning(f"拦截不安全的网络测试地址: {target.url}") + return NetworkTestResult(success=False, message=invalid_message) + if include: + self._logger.debug("nettest include 参数已忽略，改为服务端固定校验") + return await self._request(target) + + async def _request(self, target: NetworkTestRule) -> NetworkTestResult: + """手动处理受控重定向并校验最终响应。""" + started_at = monotonic() + if target.probe_protocol == "websocket": + return await self._request_websocket(target, started_at) + + current_url = target.url + current_method = target.http_method + current_json = target.request_body() + redirect_count = 0 + response: Optional[NetworkTestResponse] = None + while redirect_count <= _MAX_REDIRECTS: + response = await self._request_http( + target=target, + url=current_url, + method=current_method, + json_body=current_json, + ) + if response is None or response.status_code not in _REDIRECT_STATUS_CODES: + break + location = response.headers.get("location") + if not location: + break + next_url = urljoin(current_url, location) + if not self._is_allowed_redirect(next_url, target): + await self._close_response(response) + self._logger.warning(f"拦截网络测试重定向: {current_url} -> {next_url}") + return NetworkTestResult( + success=False, + message="测试目标发生了未授权跳转", + ) + response_status = response.status_code + await self._close_response(response) + response = None + current_url = next_url + if current_method == "POST" and response_status in {301, 302, 303}: + current_method = "GET" + current_json = None + redirect_count += 1 + + elapsed_ms = round((monotonic() - started_at) * 1000) + if redirect_count > _MAX_REDIRECTS: + await self._close_response(response) + return NetworkTestResult(success=False, message="测试目标重定向次数过多") + if response is None: + label = target.proxy_name or target.name + return NetworkTestResult( + success=False, + message=f"{label}无法连接", + elapsed_ms=elapsed_ms, + ) + + try: + return self._build_response_result(target, response, elapsed_ms) + finally: + await self._close_response(response) + + async def _request_http( + self, + *, + target: NetworkTestRule, + url: str, + method: str, + json_body: Optional[Mapping[str, Any]], + ) -> Optional[NetworkTestResponse]: + """按目标声明的方法发送 HTTP 探测，同时兼容旧 GET 传输端口。""" + options = { + "proxy": self._settings("PROXY", None) if target.proxy else None, + "headers": target.request_headers(), + "user_agent": self._settings("NORMAL_USER_AGENT", None), + } + request = getattr(self._transport, "request", None) + if callable(request): + return await request(method, url, json_body=json_body, **options) + if method == "GET": + return await self._transport.get(url, **options) + return None + + async def _request_websocket( + self, + target: NetworkTestRule, + started_at: float, + ) -> NetworkTestResult: + """探测 WebSocket 握手，不把 HTTP GET 当作长连接可用性。""" + probe = getattr(self._transport, "websocket", None) + connected = False + if callable(probe): + try: + connected = await probe( + target.url, + headers=target.request_headers(), + user_agent=self._settings("NORMAL_USER_AGENT", None), + timeout=10, + ) + except Exception as err: # noqa: BLE001 - 探测失败转成稳定的结果消息 + self._logger.debug(f"WebSocket 网络探测失败: {err}") + elapsed_ms = round((monotonic() - started_at) * 1000) + if connected: + return NetworkTestResult(success=True, elapsed_ms=elapsed_ms) + label = target.proxy_name or target.name + return NetworkTestResult( + success=False, + message=f"{label}无法连接", + elapsed_ms=elapsed_ms, + ) + + @staticmethod + def _build_response_result( + target: NetworkTestRule, + response: NetworkTestResponse, + elapsed_ms: int, + ) -> NetworkTestResult: + """把传输响应归一为稳定的应用结果。""" + if response.status_code in target.success_status_codes: + if target.expected_text and target.expected_text.lower() not in (response.text or "").lower(): + return NetworkTestResult( + success=False, + message=target.invalid_message or "无效响应", + elapsed_ms=elapsed_ms, + ) + return NetworkTestResult(success=True, elapsed_ms=elapsed_ms) + if target.proxy_name: + message = f"{target.proxy_name}已失效，错误码：{response.status_code}" + else: + message = f"错误码：{response.status_code}" + if "github" in target.url: + if response.status_code == 401: + message = "Github Token已失效，请检查配置" + elif response.status_code in {403, 429}: + message = "触发限流，请配置Github Token" + return NetworkTestResult( + success=False, + message=message, + elapsed_ms=elapsed_ms, + ) + + def _find_rule( + self, + *, + target_id: Optional[str], + url: Optional[str], + ) -> Optional[NetworkTestRule]: + """优先按正式 target_id、否则按旧 URL 精确匹配内置规则。""" + rules = self._build_rules() + if target_id: + return next((rule for rule in rules if rule.id == target_id), None) + if url: + return next((rule for rule in rules if rule.url == url), None) + return None + + def _validate_rule(self, target: NetworkTestRule) -> Optional[str]: + """兜底校验服务端规则协议、请求方法、凭据和目标目录。""" + parsed = urlparse(target.url) + expected_scheme = "wss" if target.probe_protocol == "websocket" else "https" + if target.probe_protocol not in {"http", "websocket"}: + return "测试协议无效" + if parsed.scheme.lower() != expected_scheme: + return f"测试地址仅支持 {expected_scheme.upper()}" + if target.probe_protocol == "http" and target.http_method not in {"GET", "POST"}: + return "测试方法无效" + if not parsed.netloc: + return "测试地址无效" + if parsed.username or parsed.password: + return "测试地址不支持携带账号信息" + if not any(rule == target for rule in self._build_rules()): + return "测试地址不在允许的测试目标列表中" + return None + + @classmethod + def _is_allowed_redirect(cls, url: str, target: NetworkTestRule) -> bool: + """只允许重定向到当前测试项声明的协议、主机、端口和路径。""" + parsed = urlparse(url) + if parsed.scheme.lower() != "https" or not parsed.netloc: + return False + if parsed.username or parsed.password: + return False + return any(cls._matches_prefix(url, prefix) for prefix in target.allowed_redirect_prefixes) + + @staticmethod + def _matches_prefix(url: str, prefix: str) -> bool: + """按协议、主机、有效端口和路径前缀匹配允许的跳转范围。""" + parsed_url = urlparse(url) + parsed_prefix = urlparse(prefix) + if parsed_url.scheme.lower() != parsed_prefix.scheme.lower(): + return False + if (parsed_url.hostname or "").lower() != (parsed_prefix.hostname or "").lower(): + return False + try: + url_port = parsed_url.port or (443 if parsed_url.scheme.lower() == "https" else 80) + prefix_port = parsed_prefix.port or (443 if parsed_prefix.scheme.lower() == "https" else 80) + except ValueError: + return False + if url_port != prefix_port: + return False + prefix_path = parsed_prefix.path or "/" + if prefix_path.endswith("/"): + return parsed_url.path.startswith(prefix_path) + return parsed_url.path == prefix_path or parsed_url.path.startswith(f"{prefix_path}/") + + async def _close_response( + self, + response: Optional[NetworkTestResponse], + ) -> None: + """安静释放已读取响应，关闭失败不覆盖真实探测结果。""" + if response is None or not hasattr(response, "aclose"): + return + try: + await response.aclose() + except Exception as err: # noqa: BLE001 - 资源回收失败仅记录诊断 + self._logger.debug(f"关闭网络测试响应失败: {err}") + + def _build_rules(self) -> tuple[NetworkTestRule, ...]: + """根据当前部署设置和已启用模块构建唯一网络测试目录。""" + return build_network_rules(self._settings, self._enabled_module_ids) + + +_configured_network_test_service: Optional[NetworkTestService] = None + + +def configure_network_test_service(service: NetworkTestService) -> None: + """由启动组合根登记唯一的网络测试应用服务。""" + global _configured_network_test_service + _configured_network_test_service = service + + +def reset_network_test_service() -> None: + """清除当前 lifespan 的网络测试应用服务。""" + global _configured_network_test_service + _configured_network_test_service = None + + +def get_configured_network_test_service() -> NetworkTestService: + """返回启动阶段登记的网络测试应用服务。""" + if _configured_network_test_service is None: + raise RuntimeError("网络测试服务尚未装配") + return _configured_network_test_service diff --git a/app/schemas/system.py b/app/schemas/system.py index c79cfdb7e..6afa981dc 100644 --- a/app/schemas/system.py +++ b/app/schemas/system.py @@ -62,7 +62,7 @@ class MediaServerConf(BaseModel): return None try: return int(value) - except (TypeError, ValueError): + except TypeError, ValueError: return None @@ -318,9 +318,12 @@ class RuleTestData(BaseModel): class NetTestTarget(BaseModel): """前端可选择的网络测试目标。""" - id: str - name: str - icon: str + id: str = Field(description="Stable identifier used to run the approved network test.") + name: str = Field(description="Short description of the tested function or service.") + address: str = Field( + description="Sanitized destination origin; paths, query parameters, and credentials are omitted." + ) + icon: str = Field(description="Frontend logo key for the tested service.") class SystemModuleInfo(BaseModel): diff --git a/app/startup/composition/network.py b/app/startup/composition/network.py index f0d57791f..c1d5fc504 100644 --- a/app/startup/composition/network.py +++ b/app/startup/composition/network.py @@ -1,5 +1,6 @@ """网络应用端口的宿主组合根。""" +import asyncio from collections.abc import Mapping from pathlib import Path from typing import Any, Callable, Optional, Union, cast @@ -59,7 +60,7 @@ from app.runtime.log import logger class _NetworkTestTransportAdapter: - """把通用异步 HTTP Adapter 收窄为网络探测 GET 端口。""" + """把通用异步 HTTP 与 WebSocket Adapter 收窄为网络探测端口。""" async def get( self, @@ -70,6 +71,28 @@ class _NetworkTestTransportAdapter: user_agent: Optional[str] = None, ) -> Optional[NetworkTestResponse]: """使用固定超时、证书校验和手动重定向策略请求目标。""" + return await self.request( + "GET", + url, + proxy=proxy, + headers=headers, + user_agent=user_agent, + ) + + async def request( + self, + method: str, + url: str, + *, + proxy: Any = None, + headers: Optional[Mapping[str, str]] = None, + user_agent: Optional[str] = None, + json_body: Optional[Mapping[str, Any]] = None, + ) -> Optional[NetworkTestResponse]: + """按请求规则发送 HTTP 请求并保持证书校验与重定向关闭。""" + request_kwargs: dict[str, Any] = {} + if json_body is not None: + request_kwargs["json"] = dict(json_body) response = await AsyncRequestUtils( proxies=proxy, headers=dict(headers) if headers else None, @@ -77,15 +100,65 @@ class _NetworkTestTransportAdapter: ua=user_agent or "", verify=True, follow_redirects=False, - ).get_res(url, allow_redirects=False) + ).request( + method=method, + url=url, + follow_redirects=False, + **request_kwargs, + ) return cast(Optional[NetworkTestResponse], response) + async def websocket( + self, + url: str, + *, + headers: Optional[Mapping[str, str]] = None, + user_agent: Optional[str] = None, + timeout: float = 10, + ) -> bool: + """执行一次不跟随重定向的 WSS 握手，并在线路建立后关闭连接。""" + + def connect() -> bool: + """同步建立 WebSocket 握手，供 asyncio.to_thread 调用。""" + import websocket + + connection = None + try: + request_headers = dict(headers) if headers else {} + if user_agent: + request_headers.setdefault("User-Agent", user_agent) + connection = websocket.create_connection( + url, + timeout=timeout, + header=request_headers or None, + redirect_limit=0, + ) + handshake = getattr(connection, "handshake_response", None) + return getattr(handshake, "status", None) == 101 + except Exception: + return False + finally: + if connection is not None: + try: + connection.close() + except Exception: + pass + + return await asyncio.to_thread(connect) + def _read_network_test_setting(key: str, default: Any = None) -> Any: """延迟读取已由组合根发布的部署设置，避免装配阶段提前取值。""" return get_runtime_settings().get(key, default) +def _list_enabled_network_module_ids() -> tuple[str, ...]: + """按模块运行时的当前配置投影网络相关的已启用宿主模块。""" + from app.runtime.extensions.module.manager import ModuleManager + + return tuple(spec.id for spec in ModuleManager().list_enabled_specs()) + + class _ImageTransportAdapter: """把通用 HTTP Adapter 收窄为图片应用服务的 GET 端口。""" @@ -211,9 +284,7 @@ class _DownloadHttpAdapter: raise_exception: bool = False, ) -> Optional[DownloadResponsePort]: """发送下载链 GET 请求并原样保留响应三态。""" - request = self._request( - cookies=cookies, ua=ua, headers=headers, proxies=proxies, timeout=timeout - ) + request = self._request(cookies=cookies, ua=ua, headers=headers, proxies=proxies, timeout=timeout) kwargs: dict[str, Any] = {"raise_exception": raise_exception} if params is not None: kwargs["params"] = params @@ -232,9 +303,7 @@ class _DownloadHttpAdapter: params: Optional[dict[str, Any]] = None, ) -> Optional[DownloadResponsePort]: """发送下载链 POST 请求并原样保留响应三态。""" - request = self._request( - cookies=cookies, ua=ua, headers=headers, proxies=proxies, timeout=timeout - ) + request = self._request(cookies=cookies, ua=ua, headers=headers, proxies=proxies, timeout=timeout) kwargs: dict[str, Any] = {} if params is not None: kwargs["params"] = params @@ -253,9 +322,7 @@ class _DownloadArchiveAdapter: archive_format: Optional[str], ) -> None: """使用宿主统一归档实现解压字幕文件。""" - SystemUtils.unpack_archive( - archive_file, extract_dir, archive_format=archive_format - ) + SystemUtils.unpack_archive(archive_file, extract_dir, archive_format=archive_format) def list_files(self, directory: Path, extensions: tuple[str, ...]) -> list[Path]: """使用宿主统一文件扫描实现列出字幕文件。""" @@ -287,9 +354,7 @@ class _ScrapingHttpAdapter: "proxies": proxies, "ua": ua, "timeout": timeout, - "referer": ( - "https://movie.douban.com/" if "doubanio.com" in url else None - ), + "referer": ("https://movie.douban.com/" if "doubanio.com" in url else None), } response = RequestUtils(**options).get_res(url) return cast(Optional[ScrapingResponsePort], response) @@ -305,9 +370,7 @@ class _ScrapingHttpAdapter: options: dict[str, Any] = { "proxies": proxies, "ua": ua, - "referer": ( - "https://movie.douban.com/" if "doubanio.com" in url else None - ), + "referer": ("https://movie.douban.com/" if "doubanio.com" in url else None), } response = RequestUtils(**options).get_stream(url=url) return cast(ScrapingStreamResponsePort, response) @@ -348,6 +411,7 @@ def configure_application_network_ports() -> None: transport=network_test_transport, settings=_read_network_test_setting, logger=logger, + enabled_module_ids=_list_enabled_network_module_ids, ) ) configure_image_ports( diff --git a/tests/test_system_nettest.py b/tests/test_system_nettest.py index be2b9b809..c02e640b8 100644 --- a/tests/test_system_nettest.py +++ b/tests/test_system_nettest.py @@ -263,7 +263,7 @@ class TestNettestSecurity: assert resp.success assert any(item["id"] == "pip_proxy" for item in resp.data) assert any(item["id"] == "github_proxy_web" for item in resp.data) - assert all(set(item) == {"id", "name", "icon"} for item in resp.data) + assert all(set(item) == {"id", "name", "address", "icon"} for item in resp.data) def test_nettest_blocks_unknown_target_before_transport(self): """未知目标在应用服务目录匹配阶段即返回，不得触发传输端口。"""