mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-10-29 03:18:14 +08:00
187 lines
7.0 KiB
Python
187 lines
7.0 KiB
Python
"""
|
||
Web Push 协议发送测试：用模拟浏览器订阅解密投递内容，并校验 VAPID 签名与请求头。
|
||
"""
|
||
import base64
|
||
import json
|
||
import time
|
||
from types import SimpleNamespace
|
||
from unittest.mock import patch
|
||
|
||
import http_ece
|
||
import jwt
|
||
import pytest
|
||
from cryptography.hazmat.primitives import serialization
|
||
from cryptography.hazmat.primitives.asymmetric import ec
|
||
from py_vapid import Vapid
|
||
|
||
from app.adapters.network.webpush import WebPushDeliveryError, send_webpush
|
||
|
||
|
||
def _b64url(data: bytes) -> str:
|
||
return base64.urlsafe_b64encode(data).rstrip(b"=").decode()
|
||
|
||
|
||
@pytest.fixture
|
||
def browser():
|
||
"""
|
||
模拟浏览器 PushSubscription：接收方 P-256 密钥对与 16 字节 auth 密钥。
|
||
"""
|
||
private_key = ec.generate_private_key(ec.SECP256R1())
|
||
public_raw = private_key.public_key().public_bytes(
|
||
serialization.Encoding.X962, serialization.PublicFormat.UncompressedPoint
|
||
)
|
||
auth = b"0123456789abcdef"
|
||
subscription = {
|
||
"endpoint": "https://push.example.com/send/abc?x=1",
|
||
"keys": {"p256dh": _b64url(public_raw), "auth": _b64url(auth)},
|
||
}
|
||
return SimpleNamespace(private_key=private_key, auth=auth, subscription=subscription)
|
||
|
||
|
||
@pytest.fixture
|
||
def vapid():
|
||
"""
|
||
生成与宿主 VAPID 配置同格式（base64url 原始私钥）的服务端密钥。
|
||
"""
|
||
key = Vapid()
|
||
key.generate_keys()
|
||
raw = key.private_key.private_numbers().private_value.to_bytes(32, "big")
|
||
return SimpleNamespace(private_key=_b64url(raw), public_key=key.public_key)
|
||
|
||
|
||
def _post(status_code=201):
|
||
"""
|
||
拦截 RequestUtils 构造与 POST，记录请求头和正文。
|
||
"""
|
||
captured = {}
|
||
|
||
class _Utils:
|
||
def __init__(self, headers=None, timeout=None, verify=None, **_):
|
||
captured.update(headers=headers, timeout=timeout, verify=verify)
|
||
|
||
def post_res(self, url, data=None, **_):
|
||
"""
|
||
记录推送地址与加密请求体，并返回指定 HTTP 状态的离线响应。
|
||
"""
|
||
captured.update(url=url, body=data)
|
||
return SimpleNamespace(status_code=status_code, reason="R", text="detail")
|
||
|
||
return captured, patch("app.adapters.network.webpush.RequestUtils", _Utils)
|
||
|
||
|
||
def test_payload_decrypts_with_browser_keys_and_vapid_verifies(browser, vapid):
|
||
"""
|
||
浏览器私钥能解出原文；VAPID JWT 由服务端公钥签名，aud 为端点源站。
|
||
"""
|
||
captured, patcher = _post()
|
||
payload = json.dumps({"title": "标题", "body": "内容"}, ensure_ascii=False)
|
||
with patcher:
|
||
send_webpush(browser.subscription, payload, vapid.private_key, "mailto:admin@example.com")
|
||
|
||
decrypted = http_ece.decrypt(
|
||
captured["body"],
|
||
private_key=browser.private_key,
|
||
auth_secret=browser.auth,
|
||
version="aes128gcm",
|
||
)
|
||
assert decrypted.decode("utf-8") == payload
|
||
|
||
headers = captured["headers"]
|
||
assert headers["Content-Encoding"] == "aes128gcm"
|
||
assert headers["TTL"] == "0"
|
||
assert captured["url"] == browser.subscription["endpoint"]
|
||
assert captured["verify"] is True
|
||
|
||
scheme, _, params = headers["Authorization"].partition(" ")
|
||
assert scheme == "vapid"
|
||
fields = dict(item.strip().split("=", 1) for item in params.split(","))
|
||
public_pem = vapid.public_key.public_bytes(
|
||
serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo
|
||
)
|
||
claims = jwt.decode(fields["t"], public_pem, algorithms=["ES256"], audience="https://push.example.com")
|
||
assert claims["sub"] == "mailto:admin@example.com"
|
||
assert time.time() < claims["exp"] <= time.time() + 24 * 3600
|
||
assert base64.urlsafe_b64decode(fields["k"] + "==") == vapid.public_key.public_bytes(
|
||
serialization.Encoding.X962, serialization.PublicFormat.UncompressedPoint
|
||
)
|
||
|
||
|
||
def test_each_send_uses_fresh_ephemeral_key(browser, vapid):
|
||
"""
|
||
同一内容两次加密结果不同（盐与临时 ECDH 公钥都随机），且都能解密。
|
||
"""
|
||
captured, patcher = _post()
|
||
bodies = []
|
||
with patcher:
|
||
for _ in range(2):
|
||
send_webpush(browser.subscription, "same", vapid.private_key, "mailto:a@b.c")
|
||
bodies.append(captured["body"])
|
||
assert bodies[0] != bodies[1]
|
||
for body in bodies:
|
||
assert http_ece.decrypt(body, private_key=browser.private_key,
|
||
auth_secret=browser.auth, version="aes128gcm") == b"same"
|
||
|
||
|
||
def test_endpoint_headers_and_ttl_are_forwarded(browser, vapid):
|
||
"""
|
||
WNS 等端点的附加请求头和 TTL 原样发送，VAPID 头不被调用方覆盖。
|
||
"""
|
||
captured, patcher = _post()
|
||
with patcher:
|
||
send_webpush(browser.subscription, "x", vapid.private_key, "mailto:a@b.c",
|
||
ttl=86400, headers={"X-WNS-Cache-Policy": "cache", "Authorization": "fake"})
|
||
assert captured["headers"]["X-WNS-Cache-Policy"] == "cache"
|
||
assert captured["headers"]["TTL"] == "86400"
|
||
assert captured["headers"]["Authorization"].startswith("vapid t=")
|
||
|
||
|
||
@pytest.mark.parametrize("status_code", [404, 410, 413])
|
||
def test_rejected_delivery_raises_with_response(browser, vapid, status_code):
|
||
"""
|
||
推送服务返回 202 以上状态码时抛错并带上响应，供调用方清理失效订阅。
|
||
"""
|
||
_, patcher = _post(status_code)
|
||
with patcher, pytest.raises(WebPushDeliveryError) as exc:
|
||
send_webpush(browser.subscription, "x", vapid.private_key, "mailto:a@b.c")
|
||
assert exc.value.response.status_code == status_code
|
||
|
||
|
||
@pytest.mark.parametrize("keys", [{}, {"p256dh": "AAAA", "auth": "AAAA"}])
|
||
def test_invalid_subscription_keys_are_rejected_before_request(vapid, keys):
|
||
"""
|
||
缺少或格式错误的订阅密钥在发请求前报错，且不带响应（不会被当成失效订阅删除）。
|
||
"""
|
||
captured, patcher = _post()
|
||
with patcher, pytest.raises(WebPushDeliveryError) as exc:
|
||
send_webpush({"endpoint": "https://push.example.com/x", "keys": keys}, "x",
|
||
vapid.private_key, "mailto:a@b.c")
|
||
assert exc.value.response is None
|
||
assert "url" not in captured
|
||
|
||
|
||
def test_no_response_is_reported(browser, vapid):
|
||
"""
|
||
网络失败（RequestUtils 返回 None）时抛错而不是静默成功。
|
||
"""
|
||
with patch("app.adapters.network.webpush.RequestUtils") as utils:
|
||
utils.return_value.post_res.return_value = None
|
||
with pytest.raises(WebPushDeliveryError):
|
||
send_webpush(browser.subscription, "x", vapid.private_key, "mailto:a@b.c")
|
||
|
||
|
||
def test_host_configured_vapid_key_signs(browser):
|
||
"""
|
||
宿主内置的 VAPID 私钥（base64url 原始格式）可以签出有效的 Authorization 头。
|
||
"""
|
||
from app.runtime.settings import get_runtime_setting
|
||
|
||
vapid_config = get_runtime_setting("VAPID")
|
||
captured, patcher = _post()
|
||
with patcher:
|
||
send_webpush(browser.subscription, "x", vapid_config["privateKey"], vapid_config["subject"])
|
||
fields = dict(
|
||
item.strip().split("=", 1)
|
||
for item in captured["headers"]["Authorization"].partition(" ")[2].split(",")
|
||
)
|
||
assert _b64url(base64.urlsafe_b64decode(fields["k"] + "==")) == vapid_config["publicKey"]
|