From 6680e090e6349951829bfcfbeafca0dec8acc9a0 Mon Sep 17 00:00:00 2001 From: jxxghp Date: Sat, 24 Oct 2026 09:17:24 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E8=87=AA=E5=8A=A8=E7=99=BB=E8=AE=B0?= =?UTF-8?q?=E4=B8=AD=E5=BF=83=E6=8E=A8=E9=80=81=E5=AE=9E=E4=BE=8B=E5=B9=B6?= =?UTF-8?q?=E5=AE=89=E5=85=A8=E8=BD=AE=E6=8D=A2=E5=87=AD=E6=8D=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/adapters/external/server.py | 36 +++- app/api/endpoints/notification.py | 13 +- app/application/security/secrets.py | 2 + app/application/settings/contract.py | 3 + .../resources/system_settings_catalog.json | 6 + app/modules/apppush/__init__.py | 107 +++++++++- app/schemas/push.py | 10 + app/schemas/types.py | 2 + docs/architecture-overview.md | 2 +- docs/mcp-api.md | 4 +- .../architecture/dependency-baseline.json | 7 +- tests/test_apppush_enrollment.py | 198 ++++++++++++++++++ 12 files changed, 372 insertions(+), 18 deletions(-) create mode 100644 tests/test_apppush_enrollment.py diff --git a/app/adapters/external/server.py b/app/adapters/external/server.py index 6d192fa7c..02a9dc5d1 100644 --- a/app/adapters/external/server.py +++ b/app/adapters/external/server.py @@ -21,7 +21,7 @@ from typing import ( Union, cast, ) -from urllib.parse import parse_qs, quote, urlparse, urlsplit +from urllib.parse import parse_qs, quote, urlparse, urlsplit, urlunsplit from httpx2 import Response as AsyncResponse from requests import Response @@ -195,6 +195,21 @@ class MoviePilotServerHelper: _user_uid: Optional[str] = None _github_user: Optional[str] = None + @staticmethod + def normalize_relay_url(server: str) -> str: + """ + 校验中转地址并统一持久凭据的作用域，不允许凭据、查询参数或片段。 + + :param server: 管理员配置的 HTTPS 服务根地址，回环调试允许 HTTP + :return: 规范化且去除尾部斜线的地址 + :raises ValueError: 地址不能安全接收实例凭据 + """ + target = urlsplit(server) + local = target.hostname in {"localhost", "127.0.0.1", "::1"} + if (target.scheme != "https" and not (local and target.scheme == "http")) or not target.hostname or target.username or target.password or target.query or target.fragment: + raise ValueError("中转地址必须使用 HTTPS，本机回环调试除外") + return urlunsplit((target.scheme, target.netloc.lower(), target.path.rstrip("/"), "", "")) + @staticmethod def relay_push(server: str, key: str, method: str, path: str, data: dict[str, Any]) -> dict[str, Any]: """ @@ -207,21 +222,28 @@ class MoviePilotServerHelper: :param data: 已绑定当前用户的注册或消息数据 :return: Server 返回的对象；失败抛出不含凭据或原始响应的 ValueError """ - target = urlsplit(server) - local = target.hostname in {"localhost", "127.0.0.1", "::1"} - if (target.scheme != "https" and not (local and target.scheme == "http")) or not target.hostname or target.username or target.password or target.query or target.fragment: - raise ValueError("中转地址必须使用 HTTPS，本机回环调试除外") - if method not in {"GET", "POST", "DELETE"} or path not in {"devices", "pairings", "messages"}: + server = MoviePilotServerHelper.normalize_relay_url(server) + allowed = {("GET", "devices"), ("POST", "devices"), ("DELETE", "devices"), + ("POST", "pairings"), ("POST", "messages"), ("POST", "instances"), + ("PUT", "instances/credentials"), ("DELETE", "instances")} + if (method, path) not in allowed: raise ValueError("不支持的通知接口") if len(key) < 32: raise ValueError("请配置至少 32 位的中转发送密钥") - response = RequestUtils(timeout=60, headers={"Authorization": f"Bearer {key}"}).request( + # 首次登记或恢复轮换之后还要执行原操作，两段请求均需落在 App 的65秒截止时间内。 + response = RequestUtils(timeout=20, headers={"Authorization": f"Bearer {key}"}).request( method, f"{server.rstrip('/')}/push/{path}", allow_redirects=False, verify=True, params=data if method == "GET" else None, json=data if method != "GET" else None, ) if response is None or response.status_code != 200: code = response.status_code if response is not None else 0 + if code == 403 and path == "instances": + raise ValueError("消息中心尚未开放自动接入，请联系中心运营者") + if code == 401: + raise ValueError("通知接入凭据无效或已停用，请联系中心运营者") + if code == 429: + raise ValueError("消息中心请求过于频繁，请稍后重试") raise ValueError(f"私有通知服务请求失败（{code}）") try: return _response_mapping(response) diff --git a/app/api/endpoints/notification.py b/app/api/endpoints/notification.py index 0fe85b11a..d6965cc4a 100644 --- a/app/api/endpoints/notification.py +++ b/app/api/endpoints/notification.py @@ -12,7 +12,7 @@ from app.api.response import ResponseAPIRouter from app.application.configuration import get_configured_system_config from app.chain.notification import NotificationChain from app.schemas.common import ManageRequest as _SchemaManageRequest -from app.schemas.push import AppPushRequest +from app.schemas.push import AppPushInstanceRequest, AppPushRequest from app.schemas.response import Response as _SchemaResponse from app.schemas.system import NotificationConf from app.schemas.types import SystemConfigKey @@ -20,6 +20,17 @@ from app.schemas.types import SystemConfigKey router = ResponseAPIRouter() +@router.post("/app/instance", summary="轮换 App 通知实例凭据", response_model=_SchemaResponse[Dict[str, Any]]) +def manage_app_instance(request: AppPushInstanceRequest, + current_user: ApiPrincipal = Depends(get_current_active_superuser)) -> _SchemaResponse[object]: + """ + 仅容器管理员能轮换整个实例的发送凭据，响应只包含结果，不下发密钥。 + """ + result = NotificationChain().manage_channel(channel="AppPush", action="rotate_instance", source=request.source, + owner=current_user.notification_identity, administrative=True) + return _SchemaResponse(success=bool(result.get("success")), message=result.get("message") or "", data=result.get("data")) + + @router.post("/app", summary="管理当前用户的 App 通知", response_model=_SchemaResponse[Dict[str, Any]]) def manage_app_notification( request: AppPushRequest, diff --git a/app/application/security/secrets.py b/app/application/security/secrets.py index ac875c210..c7172ec15 100644 --- a/app/application/security/secrets.py +++ b/app/application/security/secrets.py @@ -35,6 +35,8 @@ _SECRET_SETTING_NAMES = frozenset( { # CookieCloud 的用户 key 没有类型后缀，但与密码共同构成端到端加密凭据。 "cookiecloud_key", + # 自动推送接入是内部凭据状态，默认查询应整体脱敏，不能只根据嵌套字段名判断。 + "app_push_credentials", } ) _SECRET_SETTING_ENDINGS = ("_encrypt_key",) diff --git a/app/application/settings/contract.py b/app/application/settings/contract.py index 4ab24d9bd..c68d3d249 100644 --- a/app/application/settings/contract.py +++ b/app/application/settings/contract.py @@ -299,6 +299,9 @@ SYSTEMCONFIG_CONTRACTS: dict[SystemConfigKey, _SystemConfigContract] = { match_field="name", sensitive=True, ), + SystemConfigKey.AppPushCredentials: _internal_contract( + "notifications", Optional[dict[str, JsonData]], sensitive=True + ), SystemConfigKey.NotificationSwitchs: _contract( "notification_switches", Optional[list[NotificationSwitchConf]], diff --git a/app/application/settings/resources/system_settings_catalog.json b/app/application/settings/resources/system_settings_catalog.json index 8fe44d685..6121e1a84 100644 --- a/app/application/settings/resources/system_settings_catalog.json +++ b/app/application/settings/resources/system_settings_catalog.json @@ -1605,6 +1605,12 @@ } }, "systemconfig": { + "AppPushCredentials": { + "description": "App 中心推送自动接入的内部凭据状态；按服务地址隔离，不通过通知渠道配置下发", + "enum_name": "AppPushCredentials", + "source_file": "app/schemas/types.py", + "source_line": 430 + }, "AIAgentConfig": { "description": "AI智能体配置", "enum_name": "AIAgentConfig", diff --git a/app/modules/apppush/__init__.py b/app/modules/apppush/__init__.py index 5750435f6..d806356dd 100644 --- a/app/modules/apppush/__init__.py +++ b/app/modules/apppush/__init__.py @@ -2,22 +2,31 @@ 将现有 MoviePilot 通知经私有 Server 投递到 App 或主屏幕通知网页。 """ +import re +import secrets +import threading +from collections.abc import Callable from typing import Any, Optional, Union from app.adapters.external.server import MoviePilotServerHelper +from app.application.configuration import get_configured_system_config from app.modules import _MessageBase, _ModuleBase from app.runtime.log import logger +from app.runtime.settings import get_runtime_setting from app.schemas.message import Message from app.schemas.system import NotificationConf -from app.schemas.types import ModuleType, NotificationChannel +from app.schemas.types import ModuleType, NotificationChannel, SystemConfigKey class AppPushModule(_ModuleBase, _MessageBase[object]): """ 复用现有渠道配置、消息类型开关和用户路由，不持有 Apple 或 VAPID 私钥。 + + :cvar _relay_lock: 同一进程内串行创建、恢复与轮换凭据，数据库写入使用配置服务原子操作 """ _notification_user_target_keys = ("apppush",) + _relay_lock = threading.RLock() def init_module(self) -> None: """ @@ -73,11 +82,92 @@ class AppPushModule(_ModuleBase, _MessageBase[object]): def _request(self, conf: NotificationConf, method: str, path: str, data: dict[str, Any]) -> dict[str, Any]: """ - 从渠道配置读取唯一中转地址和实例密钥，不将密钥返回客户端或写入日志。 + 使用手工或自动接入凭据，不将内部凭据返回客户端或写入日志。 """ config = conf.config or {} - return MoviePilotServerHelper.relay_push(str(config.get("PUSH_SERVER_URL") or ""), - str(config.get("PUSH_RELAY_KEY") or ""), method, path, data) + server = self._server_url(conf) + key = str(config.get("PUSH_RELAY_KEY") or "") or self._automatic_key(server) + return MoviePilotServerHelper.relay_push(server, key, method, path, data) + + @staticmethod + def _server_url(conf: NotificationConf) -> str: + """ + 空地址沿用既有中心服务配置，地址规范化同时决定本机凭据作用域。 + """ + return MoviePilotServerHelper.normalize_relay_url(str((conf.config or {}).get("PUSH_SERVER_URL") + or get_runtime_setting("MP_SERVER_HOST") or "")) + + @staticmethod + def _save_registration(server: str, change: Callable[[dict[str, str]], dict[str, str]]) -> dict[str, str]: + """ + 在短事务内更新一个中心的凭据，不跨网络持有数据库事务或覆盖其他中心状态。 + + :param server: 已校验且规范化的服务根地址 + :param change: 纯状态变更；输入、输出只含内部字符串字段 + :return: 已提交的该中心状态，不得交给客户端 + """ + def update(current: Any) -> tuple[dict[str, str], dict[str, Any]]: + """ + 合并最新持久化值，拒绝损坏状态而非静默生成另一份实例身份。 + """ + if current is not None and not isinstance(current, dict): + raise ValueError("AppPush 接入状态损坏，请从备份恢复") + values = dict(current or {}) + prior = values.get(server) or {} + if not isinstance(prior, dict) or any(not isinstance(value, str) for value in prior.values()): + raise ValueError("AppPush 接入状态损坏，请从备份恢复") + registration = change(dict(prior)) + values[server] = registration + return registration, values + + return get_configured_system_config().update_atomically(SystemConfigKey.AppPushCredentials, update) + + def _automatic_key(self, server: str, *, rotate: bool = False) -> str: + """ + 先持久化随机凭据，再幂等登记；候选轮换凭据先写入，重启后接续相同请求。 + + :param server: 凭据所属的精确服务根地址 + :param rotate: 仅由容器管理员专用入口请求轮换；正常操作只恢复已有轮换 + :return: 当前已登记的内部发送凭据 + :raises ValueError: 身份或持久化失败、中心拒绝登记/轮换，不自动冒用 UID 恢复 + """ + with self._relay_lock: + registration = self._save_registration(server, lambda old: old or { + "key": "mpa_" + secrets.token_urlsafe(32), "instance": "", "pending": "", + }) + if not re.fullmatch(r"mpa_[a-zA-Z0-9_-]{43}", registration.get("key", "")): + raise ValueError("AppPush 接入凭据损坏，请从备份恢复") + if not registration.get("instance"): + uid = MoviePilotServerHelper.get_user_uid() + if not uid: + raise ValueError("无法取得 MoviePilot 用户唯一 ID") + result = MoviePilotServerHelper.relay_push(server, registration["key"], "POST", "instances", {"user_uid": uid}) + instance = result.get("instance") + if not isinstance(instance, str) or not re.fullmatch(r"[a-f0-9]{32}", instance): + raise ValueError("中心服务接入响应无效") + registration = self._save_registration(server, lambda old: {**old, "instance": instance}) + if rotate: + registration = self._save_registration(server, lambda old: { + **old, "pending": old.get("pending") or "mpa_" + secrets.token_urlsafe(32), + }) + if registration.get("pending"): + pending = registration["pending"] + result = MoviePilotServerHelper.relay_push(server, registration["key"], "PUT", "instances/credentials", {"key": pending}) + if result.get("instance") != registration["instance"] or result.get("rotated") is not True: + raise ValueError("中心服务轮换响应无效") + + def promote(old: dict[str, str]) -> dict[str, str]: + """ + 比较当前世代再提交，避免跨进程迟到的响应覆盖下一次轮换。 + """ + if old.get("key") == pending: + return old + if old.get("key") != registration["key"] or old.get("pending") != pending: + raise ValueError("AppPush 凭据已被其他操作更新，请重试") + return {**old, "key": pending, "pending": ""} + + registration = self._save_registration(server, promote) + return registration["key"] def channel_manage(self, channel: str, action: str, **params: Any) -> Optional[dict[str, Any]]: """ @@ -96,10 +186,15 @@ class AppPushModule(_ModuleBase, _MessageBase[object]): try: if action == "status": data = self._request(conf, "GET", "devices", {"owner": owner}) - data.update(owner=owner, relay_url=str((conf.config or {}).get("PUSH_SERVER_URL") or "").rstrip("/") + "/push/app/") + data.update(owner=owner, relay_url=self._server_url(conf) + "/push/app/") elif action == "pair": data = self._request(conf, "POST", "pairings", {"owner": owner, "origin": origin, "installation": installation}) - data["relay_url"] = str((conf.config or {}).get("PUSH_SERVER_URL") or "").rstrip("/") + "/push/app/" + data["relay_url"] = self._server_url(conf) + "/push/app/" + elif action == "rotate_instance" and params.get("administrative") is True: + if (conf.config or {}).get("PUSH_RELAY_KEY"): + return {"success": False, "message": "手工凭据请由中心运营者轮换"} + self._automatic_key(self._server_url(conf), rotate=True) + data = {"rotated": True} elif action == "register": if not params.get("token") or not params.get("topic"): return {"success": False, "message": "尚未取得系统推送令牌"} diff --git a/app/schemas/push.py b/app/schemas/push.py index 0c96eb6b3..05481780f 100644 --- a/app/schemas/push.py +++ b/app/schemas/push.py @@ -8,6 +8,16 @@ from urllib.parse import urlsplit from pydantic import BaseModel, ConfigDict, Field, field_validator +class AppPushInstanceRequest(BaseModel): + """ + 仅容器管理员可轮换实例凭据，普通用户的设备管理请求不能触发该动作。 + """ + + model_config = ConfigDict(extra="forbid") + action: Literal["rotate"] = Field(description="轮换当前已配置中心的自动接入凭据") + source: str = Field(min_length=1, max_length=255, description="已保存并启用的 AppPush 渠道名称") + + class AppPushRequest(BaseModel): """ 通知操作绑定当前登录身份；来源由启用的通知渠道配置选择。 diff --git a/app/schemas/types.py b/app/schemas/types.py index 31d5aab92..c950e5dc8 100644 --- a/app/schemas/types.py +++ b/app/schemas/types.py @@ -426,6 +426,8 @@ class SystemConfigKey(Enum): MediaServers = "MediaServers" # 消息通知配置 Notifications = "Notifications" + # App 中心推送自动接入凭据，业务内部维护且不进入通知渠道配置 + AppPushCredentials = "AppPushCredentials" # 通知场景开关设置 NotificationSwitchs = "NotificationSwitchs" # 目录配置 diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index ab91eb9c3..63d8cbaa8 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -817,7 +817,7 @@ flowchart LR | 指标 | 当前值 | |---|---:| | Python 模块 | 1131 | -| 内部导入边 | 9,727 | +| 内部导入边 | 9,730 | | 非平凡 SCC | 1（29 个模块的 TMDB vendor 精确 containment） | | Application / Chain 具体 Adapter 直连 | 0 / 0 | | Direct egress | 51（精确 policy 准入，Application / Chain 均为 0） | diff --git a/docs/mcp-api.md b/docs/mcp-api.md index cb84df629..a48974a04 100644 --- a/docs/mcp-api.md +++ b/docs/mcp-api.md @@ -36,7 +36,9 @@ 请求包含 `action`（`status/register/unregister/pair/test`）、随机 `installation` 和实例 `origin`。 `register` 另需 APNs `token/topic/environment`；`source` 可选，用于选择启用的 AppPush 渠道。 账号通知身份由后台注入，客户端不能指定其他用户。注销同时撤销该安装配对的主屏幕通知网页。 -AppPush 配置只保存 `PUSH_SERVER_URL` 和实例专用 `PUSH_RELAY_KEY`，Apple/VAPID 私钥放在私有 Server。 +AppPush 地址 `PUSH_SERVER_URL` 留空时使用 `MP_SERVER_HOST`；首次操作自动生成并先行持久保存256位实例凭据，再向中心幂等登记现有 `user_uid`。UID 只关联来源，不能据此获取、重置或认领已有凭据。内部 `AppPushCredentials` 按中心地址隔离，不进入通知渠道配置；中心只保存凭据摘要。已有 `PUSH_RELAY_KEY` 手工配置继续兼容。 +`POST /api/v1/notification/app/instance` 仅容器超级管理员可调用，正文为 `{"action":"rotate","source":"已保存的渠道名称"}`。轮换保留设备，候选凭据先落库，响应丢失可在重启后接续；返回值不包含密钥。普通用户的 `/notification/app` 不接受该动作。 +Apple/VAPID 私钥只放在中心 Server，不放入 App 或用户容器。 消息沿用现有类型开关和管理员/用户路由；无明确用户目标时不广播。部署与免费自签说明见 [MoviePilot-Server](https://github.com/jxxghp/MoviePilot-Server#私有-app-消息推送)。 diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 2adc3946f..a44aea04d 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -959,8 +959,8 @@ "runtime_only": true } }, - "edge_count": 9727, - "edge_sha256": "4c4361d6a435793000158fa6bf8896ac4a7dd3c5a15046a46556a46fba094245", + "edge_count": 9730, + "edge_sha256": "b5187460baedd0f182db4c52f2df0e3fb0f63af4912a969037b6cb3690e4b8b9", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -7161,9 +7161,12 @@ "app.modules.apppush -> app.adapters", "app.modules.apppush -> app.adapters.external", "app.modules.apppush -> app.adapters.external.server", + "app.modules.apppush -> app.application", + "app.modules.apppush -> app.application.configuration", "app.modules.apppush -> app.modules", "app.modules.apppush -> app.runtime", "app.modules.apppush -> app.runtime.log", + "app.modules.apppush -> app.runtime.settings", "app.modules.apppush -> app.schemas", "app.modules.apppush -> app.schemas.message", "app.modules.apppush -> app.schemas.system", diff --git a/tests/test_apppush_enrollment.py b/tests/test_apppush_enrollment.py new file mode 100644 index 000000000..f5829e6cd --- /dev/null +++ b/tests/test_apppush_enrollment.py @@ -0,0 +1,198 @@ +""" +中心自动接入的真实配置持久化、故障接续及管理员权限边界回归。 +""" + +from types import SimpleNamespace +from unittest.mock import MagicMock + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from app.adapters.external.server import MoviePilotServerHelper +from app.api.dependencies.auth import get_current_active_user +from app.api.endpoints import notification +from app.application.configuration import SystemConfigService +from app.application.settings.service import SystemSettingsService +from app.db.models.systemconfig import SystemConfig +from app.db.oper.systemconfig import SystemConfigOper +from app.db.session import SessionFactory +from app.modules.apppush import AppPushModule +from app.schemas.system import NotificationConf +from app.schemas.types import SystemConfigKey + +SERVER = "https://relay.test" +UID = "existing-deployment-user-unique-id" + + +def fresh_service() -> SystemConfigService: + """ + 创建独立配置快照并从测试数据库重载，模拟宿主重启后的实际持久化读取。 + """ + oper = object.__new__(SystemConfigOper) + SystemConfigOper.__init__(oper) + oper.load_snapshot() + return SystemConfigService(repository=oper) + + +@pytest.fixture +def configuration(monkeypatch): + """ + 只替换本模块的配置入口，复用测试引导的临时 SQL，结束时还原本键。 + """ + service = fresh_service() + original = service.get(SystemConfigKey.AppPushCredentials) + service.set(SystemConfigKey.AppPushCredentials, {}) + monkeypatch.setattr("app.modules.apppush.get_configured_system_config", lambda: service) + monkeypatch.setattr("app.modules.apppush.get_runtime_setting", lambda _key: SERVER) + monkeypatch.setattr(MoviePilotServerHelper, "get_user_uid", lambda: UID) + try: + yield service + finally: + service.set(SystemConfigKey.AppPushCredentials, original) + + +def test_registration_persists_before_network_and_reuses_key_after_restart(monkeypatch, configuration): + """ + 先落库再登记，渠道和普通状态响应均不包含密钥，重启不产生第二个实例。 + """ + calls = [] + + class Transport: + """ + 只替换最后 HTTP 边界，在发送时用独立 SQL 会话核对凭据已提交。 + """ + + def __init__(self, **options): + """ + 保存出站请求头里的私有凭据，不改变请求构造行为。 + """ + self.key = options["headers"]["Authorization"].removeprefix("Bearer ") + + def request(self, method, url, **options): + """ + 验证已有 UID、密钥存放位置和先落库约束，再响应登记或设备查询。 + """ + with SessionFactory() as db: + stored = SystemConfig.get_by_key(db, SystemConfigKey.AppPushCredentials.value).value + assert stored[SERVER]["key"] == self.key + calls.append((method, url, self.key)) + assert options["allow_redirects"] is False + if url.endswith('/instances'): + assert options["json"] == {"user_uid": UID} + payload = {"instance": "a" * 32, "registered": True} + else: + payload = {"devices": []} + return SimpleNamespace(status_code=200, json=lambda: payload) + + monkeypatch.setattr("app.adapters.external.server.RequestUtils", Transport) + conf = NotificationConf(name="automatic", type="apppush", enabled=True, config={}) + module = AppPushModule.__new__(AppPushModule) + monkeypatch.setattr(module, "get_configs", lambda: {"automatic": conf}) + result = module.channel_manage("AppPush", "status", owner="a" * 32) + key = configuration.get(SystemConfigKey.AppPushCredentials)[SERVER]["key"] + assert result["success"] is True and result["data"]["relay_url"] == SERVER + "/push/app/" + assert key.startswith('mpa_') and len(key) == 47 + assert key not in str(result) and key not in str(conf.model_dump()) + restarted = fresh_service() + monkeypatch.setattr("app.modules.apppush.get_configured_system_config", lambda: restarted) + second_module = AppPushModule.__new__(AppPushModule) + monkeypatch.setattr(second_module, "get_configs", lambda: {"automatic": conf}) + assert second_module.channel_manage("AppPush", "status", owner="a" * 32)["success"] is True + assert sum(url.endswith('/instances') for _, url, _ in calls) == 1 + assert {request_key for _, _, request_key in calls} == {key} + + +def test_rotation_response_loss_keeps_candidate_for_restart_recovery(monkeypatch, configuration): + """ + 已生效轮换的响应丢失时保留两份本地状态，重启只重试相同候选，不新建空间。 + """ + remote = {"key": "", "previous": "", "lost": False} + calls = [] + + class Transport: + """ + 模拟中心接受轮换后丢失响应，出站请求仍经过真实适配器。 + """ + + def __init__(self, **options): + """ + 提取真实构造的 Bearer 头用于服务端鉴权替身。 + """ + self.key = options["headers"]["Authorization"].removeprefix("Bearer ") + + def request(self, method, url, **options): + """ + 按中心幂等规则模拟登记和轮换，丢响应不会回滚远端已接受的状态。 + """ + calls.append((method, url, self.key)) + if url.endswith('/instances/credentials'): + candidate = options['json']['key'] + stored = fresh_service().get(SystemConfigKey.AppPushCredentials)[SERVER] + assert stored['pending'] == candidate + assert (self.key == remote['key'] or + (self.key == remote['previous'] and candidate == remote['key'])) + remote['previous'], remote['key'] = self.key, candidate + if not remote['lost']: + remote['lost'] = True + return None + payload = {"instance": "a" * 32, "rotated": True} + elif url.endswith('/instances'): + remote['key'] = self.key + payload = {"instance": "a" * 32, "registered": True} + else: + assert self.key == remote['key'] + payload = {"devices": []} + return SimpleNamespace(status_code=200, json=lambda: payload) + + monkeypatch.setattr("app.adapters.external.server.RequestUtils", Transport) + module = AppPushModule.__new__(AppPushModule) + original = module._automatic_key(SERVER) + with pytest.raises(ValueError, match="请求失败"): + module._automatic_key(SERVER, rotate=True) + stored = fresh_service().get(SystemConfigKey.AppPushCredentials)[SERVER] + assert stored['key'] == original and stored['pending'] == remote['key'] != original + restarted = fresh_service() + monkeypatch.setattr("app.modules.apppush.get_configured_system_config", lambda: restarted) + recovered = AppPushModule.__new__(AppPushModule)._automatic_key(SERVER) + assert recovered == remote['key'] != original + assert restarted.get(SystemConfigKey.AppPushCredentials)[SERVER]['pending'] == '' + assert sum(url.endswith('/instances') for _, url, _ in calls) == 1 + assert sum(url.endswith('/instances/credentials') for _, url, _ in calls) == 2 + spec = SystemSettingsService(MagicMock(), restarted, MagicMock()).describe(setting_key=SystemConfigKey.AppPushCredentials.value) + assert recovered not in str(spec) and spec['definition']['generic_write_allowed'] is False + + +def test_instance_rotation_requires_container_superuser_and_does_not_accept_keys(monkeypatch): + """ + 普通用户不能轮换整个容器的凭据，管理员请求不接受自选用户、密钥或 URL。 + """ + calls = [] + + class Chain: + """ + 记录已通过真实角色检查的专用管理调用。 + """ + + def manage_channel(self, **params): + """ + 只返回轮换结果，不向客户端返回凭据。 + """ + calls.append(params) + return {"success": True, "data": {"rotated": True}} + + monkeypatch.setattr(notification, "NotificationChain", Chain) + user = SimpleNamespace(notification_identity="a" * 32, is_superuser=False) + app = FastAPI() + app.include_router(notification.router, prefix="/notification") + app.dependency_overrides[get_current_active_user] = lambda: user + body = {"action": "rotate", "source": "automatic"} + with TestClient(app) as client: + assert client.post('/notification/app/instance', json=body).status_code == 403 + assert calls == [] + user.is_superuser = True + assert client.post('/notification/app/instance', json={**body, "key": "private"}).status_code == 422 + response = client.post('/notification/app/instance', json=body) + assert response.json()['data'] == {"rotated": True} + assert calls == [{"channel": "AppPush", "action": "rotate_instance", "source": "automatic", + "owner": "a" * 32, "administrative": True}]