Merge pull request #6891 from MiCat-S/fix/webpush-send-superuser

app/api/endpoints/message.py

@@ -367,10 +367,10 @@ async def subscribe(
 )
 def send_notification(
     payload: _SchemaSubscriptionMessage,
-    _: _SchemaTokenPayload = Depends(verify_token),
+    _: ApiPrincipal = Depends(get_current_active_superuser),
 ):
     """
-    发送webpush通知
+    发送webpush通知；会推送到全部已订阅浏览器，因此仅限超级管理员调用
     """
     from app.adapters.network.webpush import WebPushDeliveryError, send_webpush
 

tests/test_api_authorization.py

@@ -8,6 +8,7 @@ from fastapi import HTTPException
 from starlette.requests import Request
 from starlette.responses import Response
 
+from app.adapters.web.security.access import verify_token
 from app.api.deps import (
     get_current_active_manage_user,
     get_current_active_manage_user_async,
@@ -20,6 +21,7 @@ from app.api.endpoints import dashboard as dashboard_endpoint
 from app.api.endpoints import github as github_endpoint
 from app.api.endpoints import history as history_endpoint
 from app.api.endpoints import login as login_endpoint
+from app.api.endpoints import message as message_endpoint
 from app.api.endpoints import plugin as plugin_endpoint
 from app.api.endpoints import rule as rule_endpoint
 from app.api.endpoints import site as site_endpoint
@@ -155,6 +157,12 @@ def test_transfer_history_clear_requires_superuser():
     assert _dependency_of(history_endpoint.empty_transfer_history, "_") is get_current_active_superuser
 
 
+def test_webpush_send_requires_superuser_and_subscribe_accepts_login_user():
+    """主动推送到全部浏览器订阅只允许管理员，订阅本身保持登录用户可用。"""
+    assert _dependency_of(message_endpoint.send_notification, "_") is get_current_active_superuser
+    assert _dependency_of(message_endpoint.subscribe, "_") is verify_token
+
+
 def test_manage_page_endpoints_accept_manage_permission():
     """管理页面接口允许具备 manage 权限的普通用户访问。"""
     sync_endpoints = [

tests/test_webpush_subscription.py

@@ -1,12 +1,18 @@
 from types import SimpleNamespace
 
 import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
 
+from app.adapters.web.security import access
+from app.api.context import get_host_runtime, get_sync_session
 from app.api.endpoints import message as message_endpoint
 from app.api.endpoints.message import is_webpush_subscription_gone
-from app.runtime.config import global_vars
+from app.application.security.token import create_access_token
+from app.runtime.config import global_vars, settings
 from app.runtime.webpush import webpush_registry
 from app.schemas.message import SubscriptionMessage
+from app.schemas.token import TokenPayload
 
 
 @pytest.fixture(autouse=True)
@@ -119,3 +125,99 @@ def test_send_notification_reports_partial_delivery(monkeypatch):
 
     assert response.success is True
     assert response.message == "消息已发送到 1 个设备，1 个设备发送失败"
+
+
+def _webpush_send_client(monkeypatch, user) -> tuple[TestClient, list[dict]]:
+    """构造挂载真实消息路由的客户端，并记录实际发出的浏览器通知。"""
+    deliveries = []
+    webpush_registry.upsert({"endpoint": "https://push.example/a", "keys": {}})
+    monkeypatch.setattr(access, "_token_identity_validator", lambda _payload: None)
+    monkeypatch.setattr(
+        message_endpoint,
+        "get_api_runtime_config_snapshot",
+        lambda: SimpleNamespace(vapid_private_key="private", vapid_subject="mailto:test@example.com"),
+    )
+    monkeypatch.setattr(
+        "app.adapters.network.webpush.send_webpush",
+        lambda **kwargs: deliveries.append(kwargs),
+    )
+    repository = SimpleNamespace(get_by_id=lambda user_id: user if user_id == user.id else None)
+    app = FastAPI()
+    app.include_router(message_endpoint.router, prefix="/message")
+    app.dependency_overrides[get_host_runtime] = lambda: SimpleNamespace(
+        authentication=SimpleNamespace(user_repository=lambda _db: repository)
+    )
+    app.dependency_overrides[get_sync_session] = lambda: object()
+    return TestClient(app), deliveries
+
+
+def _bearer(user) -> dict[str, str]:
+    """为测试用户生成登录 JWT 请求头。"""
+    token = create_access_token(
+        userid=user.id,
+        username=user.name,
+        super_user=user.is_superuser,
+    )
+    return {"Authorization": f"Bearer {token}"}
+
+
+def _user(*, superuser: bool) -> SimpleNamespace:
+    """构造当前用户依赖所需的最小用户投影。"""
+    return SimpleNamespace(id=7, name="tester", is_active=True, is_superuser=superuser)
+
+
+def test_send_notification_rejects_regular_user_without_delivery(monkeypatch):
+    """普通用户调用 webpush/send 应返回 403，且不向任何订阅发送通知。"""
+    user = _user(superuser=False)
+    client, deliveries = _webpush_send_client(monkeypatch, user)
+
+    with client:
+        response = client.post(
+            "/message/webpush/send",
+            json={"title": "测试", "url": "https://example.com"},
+            headers=_bearer(user),
+        )
+
+    assert response.status_code == 403
+    assert deliveries == []
+
+
+def test_send_notification_allows_superuser(monkeypatch):
+    """超级管理员登录后可以向已订阅浏览器发送通知。"""
+    user = _user(superuser=True)
+    client, deliveries = _webpush_send_client(monkeypatch, user)
+
+    with client:
+        response = client.post(
+            "/message/webpush/send",
+            json={"title": "测试"},
+            headers=_bearer(user),
+        )
+
+    assert response.status_code == 200
+    assert response.json()["success"] is True
+    assert len(deliveries) == 1
+    assert deliveries[0]["subscription"]["endpoint"] == "https://push.example/a"
+
+
+def test_send_notification_allows_api_token(monkeypatch):
+    """使用 API_TOKEN 的外部集成按超级管理员身份继续可以发送通知。"""
+    user = _user(superuser=True)
+    client, deliveries = _webpush_send_client(monkeypatch, user)
+    monkeypatch.setattr(settings, "API_TOKEN", "webpush-test-api-token")
+    monkeypatch.setattr(
+        access,
+        "_superuser_token_payload_provider",
+        lambda: TokenPayload(sub=user.id, username=user.name, super_user=True, purpose="authentication"),
+    )
+
+    with client:
+        response = client.post(
+            "/message/webpush/send",
+            json={"title": "测试"},
+            headers={"X-API-KEY": "webpush-test-api-token"},
+        )
+
+    assert response.status_code == 200
+    assert response.json()["success"] is True
+    assert len(deliveries) == 1