"""QQ Bot Webhook 签名校验模块(v2 占位)。 v1 采用 WebSocket 长连接模式(决策 3),Webhook HTTP 回调模式 v2 迭代。 本模块提供 Ed25519 签名校验占位实现,供 v2 启用 Webhook 模式时完善。 提供以下能力: - ``verify_webhook_signature``: 校验 Webhook Ed25519 签名(v2 占位)。 - ``is_timestamp_valid``: 校验时间戳是否在允许的时钟偏移窗口内(防重放)。 - ``is_callback_verification``: 识别 QQ Bot 回调地址验证请求。 依赖方向:仅依赖标准库与 ``yuxi.channels.contract`` 错误类型,不污染框架层。 Ed25519 校验依赖 ``cryptography``,缺失时返回 ``False`` 显式暴露缺失依赖。 """ from __future__ import annotations import time from typing import Any try: from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey _HAS_CRYPTOGRAPHY = True except ImportError: _HAS_CRYPTOGRAPHY = False # 时间戳允许的最大时钟偏移(秒),防重放攻击 _MAX_TIMESTAMP_SKEW_SECONDS = 300 def verify_webhook_signature( timestamp: str, body: str, signature: str, public_key_pem: str, ) -> bool: """校验 QQ Bot Webhook Ed25519 签名(v2 占位)。 签名算法:Ed25519(``timestamp + body``),公钥从 QQ 开放平台获取。 使用恒定时间比对以防止时序攻击。 v1 场景下 WebSocket 事件无需 Webhook 签名校验,本方法供 v2 启用 Webhook 模式时调用。 Args: timestamp: ``X-Signature-Timestamp`` 头值(秒级时间戳字符串)。 body: 原始请求体字符串。 signature: ``X-Signature-Ed25519`` 头值(十六进制编码)。 public_key_pem: QQ 开放平台公钥 PEM 字符串。 Returns: 签名校验结果,``True`` 表示通过。**不抛异常**:输入类型异常、 依赖缺失或签名不匹配时返回 ``False``,由调用方决定是否记录 日志或抛出 ``ValidationError``。 """ if not _HAS_CRYPTOGRAPHY: return False try: public_key = serialization.load_pem_public_key( public_key_pem.encode("utf-8"), ) if not isinstance(public_key, Ed25519PublicKey): return False message = f"{timestamp}{body}".encode() sig_bytes = bytes.fromhex(signature) public_key.verify(sig_bytes, message) return True except (ValueError, TypeError, InvalidSignature, UnicodeEncodeError): return False def is_timestamp_valid(timestamp: str, max_skew_seconds: int = _MAX_TIMESTAMP_SKEW_SECONDS) -> bool: """校验时间戳是否在允许的时钟偏移窗口内(防重放)。 Args: timestamp: 时间戳字符串(秒级)。 max_skew_seconds: 允许的最大时间偏差(秒),默认 300(5 分钟)。 Returns: ``True`` 表示时间戳有效。解析失败或超出窗口返回 ``False``, **不抛异常**。 """ try: ts = int(timestamp) except (TypeError, ValueError): return False skew = abs(int(time.time()) - ts) return skew <= max_skew_seconds def is_callback_verification(payload: dict[str, Any]) -> bool: """识别 QQ Bot 回调地址验证请求。 QQ Bot 注册 Webhook 时发送 ``{"plain_token": "xxx", "event_ts": "xxx"}`` 请求用于验证回调地址,适配器需用 Ed25519 私钥对 ``plain_token + event_ts`` 签名并返回 ``{"plain_token": "xxx", "signature": "xxx"}`` 响应(v2 迭代)。 Args: payload: Webhook 请求体解析后的 dict。 Returns: ``True`` 表示该请求为回调地址验证请求;``False`` 表示正常事件。 """ return "plain_token" in payload and "event_ts" in payload def build_callback_verification_response( plain_token: str, event_ts: str, private_key_pem: str, ) -> dict[str, str] | None: """构造回调地址验证响应(v2 占位)。 使用 Ed25519 私钥对 ``plain_token + event_ts`` 签名,返回 ``{"plain_token": "...", "signature": "..."}`` 响应体。 Args: plain_token: QQ Bot 发送的 plain_token。 event_ts: QQ Bot 发送的 event_ts。 private_key_pem: QQ 开放平台私钥 PEM 字符串。 Returns: 响应体 dict;依赖缺失或签名失败返回 ``None``。 """ if not _HAS_CRYPTOGRAPHY: return None try: from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey private_key = serialization.load_pem_private_key( private_key_pem.encode("utf-8"), password=None, ) if not isinstance(private_key, Ed25519PrivateKey): return None message = f"{plain_token}{event_ts}".encode() sig_bytes = private_key.sign(message) return { "plain_token": plain_token, "signature": sig_bytes.hex(), } except (ValueError, TypeError, UnicodeEncodeError): return None __all__ = [ "verify_webhook_signature", "is_timestamp_valid", "is_callback_verification", "build_callback_verification_response", ]