diff --git a/app/api/endpoints/message.py b/app/api/endpoints/message.py index fafc285e..120a4c0f 100644 --- a/app/api/endpoints/message.py +++ b/app/api/endpoints/message.py @@ -17,7 +17,7 @@ from app.db.message_oper import MessageOper from app.db.systemconfig_oper import SystemConfigOper from app.db.user_oper import get_current_active_superuser from app.helper.service import ServiceConfigHelper -from app.helper.webpush import is_webpush_subscription_gone +from app.helper.webpush import is_webpush_subscription_gone, webpush_options_for_endpoint from app.log import logger from app.modules.wechat.WXBizMsgCrypt3 import WXBizMsgCrypt from app.schemas.types import MessageChannel, SystemConfigKey @@ -316,6 +316,7 @@ def send_notification( data=json.dumps(payload.model_dump()), vapid_private_key=settings.VAPID.get("privateKey"), vapid_claims={"sub": settings.VAPID.get("subject")}, + **webpush_options_for_endpoint(sub.get("endpoint")), ) except WebPushException as err: logger.error(f"WebPush发送失败: {str(err)}") diff --git a/app/helper/webpush.py b/app/helper/webpush.py index c45df3bb..0bd5d85c 100644 --- a/app/helper/webpush.py +++ b/app/helper/webpush.py @@ -2,6 +2,9 @@ from typing import Any from pywebpush import WebPushException +# WNS 默认 TTL(秒);ttl>0 时需配 X-WNS-Cache-Policy: cache +_WNS_DEFAULT_TTL = 86400 + def is_webpush_subscription_gone(error: WebPushException) -> bool: """ @@ -10,3 +13,25 @@ def is_webpush_subscription_gone(error: WebPushException) -> bool: response: Any = getattr(error, "response", None) status_code = getattr(response, "status_code", None) or getattr(response, "status", None) return status_code in {404, 410} + + +def is_wns_endpoint(endpoint: str | None) -> bool: + """ + 判断是否为 Microsoft WNS(Edge/Windows)推送端点。 + """ + return bool(endpoint and "notify.windows.com" in endpoint) + + +def webpush_options_for_endpoint(endpoint: str | None) -> dict[str, Any]: + """ + 按推送服务返回 pywebpush 额外参数。 + + WNS 要求 TTL 与 X-WNS-Cache-Policy 一致,否则返回 400。 + 见 https://github.com/web-push-libs/pywebpush/issues/162 + """ + if not is_wns_endpoint(endpoint): + return {} + return { + "ttl": _WNS_DEFAULT_TTL, + "headers": {"X-WNS-Cache-Policy": "cache"}, + } diff --git a/app/modules/webpush/__init__.py b/app/modules/webpush/__init__.py index 3301d6e7..ce4bbc72 100644 --- a/app/modules/webpush/__init__.py +++ b/app/modules/webpush/__init__.py @@ -4,7 +4,7 @@ from typing import Union, Tuple from pywebpush import webpush, WebPushException from app.core.config import global_vars, settings -from app.helper.webpush import is_webpush_subscription_gone +from app.helper.webpush import is_webpush_subscription_gone, webpush_options_for_endpoint from app.log import logger from app.modules import _ModuleBase, _MessageBase from app.schemas import Notification @@ -95,6 +95,7 @@ class WebPushModule(_ModuleBase, _MessageBase): vapid_claims={ "sub": settings.VAPID.get("subject") }, + **webpush_options_for_endpoint(sub.get("endpoint")), ) except WebPushException as err: logger.error(f"WebPush发送失败: {str(err)}") diff --git a/tests/test_webpush_helper.py b/tests/test_webpush_helper.py new file mode 100644 index 00000000..742695bd --- /dev/null +++ b/tests/test_webpush_helper.py @@ -0,0 +1,43 @@ +from app.helper.webpush import ( + is_webpush_subscription_gone, + is_wns_endpoint, + webpush_options_for_endpoint, +) +from pywebpush import WebPushException + + +class _FakeResponse: + def __init__(self, status_code: int): + self.status_code = status_code + + +def test_is_webpush_subscription_gone_for_expired_status_codes() -> None: + for status_code in (404, 410): + err = WebPushException("gone") + err.response = _FakeResponse(status_code) + assert is_webpush_subscription_gone(err) + + +def test_is_webpush_subscription_gone_for_other_errors() -> None: + err = WebPushException("bad request") + err.response = _FakeResponse(400) + assert not is_webpush_subscription_gone(err) + + +def test_is_wns_endpoint_detects_windows_push_url() -> None: + assert is_wns_endpoint("https://wns2-sg2p.notify.windows.com/w/?token=abc") + assert not is_wns_endpoint("https://web.push.apple.com/abc") + assert not is_wns_endpoint(None) + assert not is_wns_endpoint("") + + +def test_webpush_options_for_wns_endpoint() -> None: + options = webpush_options_for_endpoint("https://wns2-pn1p.notify.windows.com/x") + assert options == { + "ttl": 86400, + "headers": {"X-WNS-Cache-Policy": "cache"}, + } + + +def test_webpush_options_for_non_wns_endpoint() -> None: + assert webpush_options_for_endpoint("https://fcm.googleapis.com/fcm/send/abc") == {}