feat: 自动登记中心推送实例并安全轮换凭据

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)

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,

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",)

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

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

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": "尚未取得系统推送令牌"}

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):
     """
     通知操作绑定当前登录身份；来源由启用的通知渠道配置选择。

app/schemas/types.py

@@ -426,6 +426,8 @@ class SystemConfigKey(Enum):
     MediaServers = "MediaServers"
     # 消息通知配置
     Notifications = "Notifications"
+    # App 中心推送自动接入凭据，业务内部维护且不进入通知渠道配置
+    AppPushCredentials = "AppPushCredentials"
     # 通知场景开关设置
     NotificationSwitchs = "NotificationSwitchs"
     # 目录配置

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

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-消息推送)。
 

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

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}]