nas/MoviePilot
1
0
Fork 0
mirror of https://github.com/jxxghp/MoviePilot.git synced 2026-11-04 06:15:14 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(system): add protocol-aware network checks

Browse Source
This commit is contained in:
jxxghp
2026-09-24 12:25:45 +08:00
parent 586c1e7710
commit 1d84d17dc4
11 changed files with 1222 additions and 562 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
app/agent/policy/mcp.py
View File

@@ -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.",

4
app/agent/policy/resources/api_mcp_schema.json
View File

@@ -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": [

14
app/api/endpoints/system.py
View File

@@ -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
],
)

534
app/application/network.py
View File

@@ -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

29
app/application/network/__init__.py Normal file
View File

@@ -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",
]

650
app/application/network/catalogue.py Normal file
View File

@@ -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

142
app/application/network/domain.py Normal file
View File

@@ -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}"

298
app/application/network/service.py Normal file
View File

@@ -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

11
app/schemas/system.py
View File

@@ -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):

98
app/startup/composition/network.py
View File

@@ -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(

2
tests/test_system_nettest.py
View File

@@ -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):
"""未知目标在应用服务目录匹配阶段即返回，不得触发传输端口。"""