Kris
ff36b9f334
新增消息日志、发件箱、会话等领域模型，新增多类型业务异常定义，添加工厂协议端口与数据库工作单元实现，同时补充飞书、钩子、Web渠道的请求验证器，以及数据转换类和异常翻译工具
67 lines
2.3 KiB
Python
67 lines
2.3 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import hmac
|
|
import logging
|
|
import time
|
|
|
|
from yuxi.channel.domain.port.external.channel_request_verifier_port import (
|
|
VerifyResult,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class FeishuRequestVerifier:
|
|
def __init__(
|
|
self,
|
|
verification_token: str = "",
|
|
encrypt_key: str = "",
|
|
timestamp_tolerance: int = 300,
|
|
):
|
|
self._verification_token = verification_token
|
|
self._encrypt_key = encrypt_key
|
|
self._timestamp_tolerance = timestamp_tolerance
|
|
|
|
@property
|
|
def channel_type(self) -> str:
|
|
return "feishu"
|
|
|
|
@property
|
|
def enabled(self) -> bool:
|
|
return bool(self._encrypt_key)
|
|
|
|
@property
|
|
def has_verification_token(self) -> bool:
|
|
return bool(self._verification_token)
|
|
|
|
async def verify(self, body: bytes, headers: dict[str, str]) -> VerifyResult:
|
|
if not self._encrypt_key:
|
|
return VerifyResult(passed=True, method="feishu_none", reason="no encrypt_key configured")
|
|
|
|
signature = headers.get("x-lark-signature", "")
|
|
timestamp = headers.get("x-lark-request-timestamp", "")
|
|
nonce = headers.get("x-lark-request-nonce", "")
|
|
|
|
if not signature or not timestamp:
|
|
return VerifyResult(passed=False, method="feishu_signature", reason="missing signature headers")
|
|
|
|
try:
|
|
ts = float(timestamp)
|
|
if abs(time.time() - ts) > self._timestamp_tolerance:
|
|
return VerifyResult(passed=False, method="feishu_signature", reason="timestamp expired")
|
|
except (ValueError, TypeError):
|
|
return VerifyResult(passed=False, method="feishu_signature", reason="invalid timestamp")
|
|
|
|
sign_string = f"{timestamp}{nonce}{self._encrypt_key}{body.decode('utf-8', errors='replace')}"
|
|
expected = hashlib.sha256(sign_string.encode("utf-8")).hexdigest()
|
|
if not hmac.compare_digest(signature.lower(), expected.lower()):
|
|
return VerifyResult(passed=False, method="feishu_signature", reason="signature mismatch")
|
|
|
|
return VerifyResult(passed=True, method="feishu_signature")
|
|
|
|
def verify_token(self, token: str) -> bool:
|
|
if not self._verification_token:
|
|
return True
|
|
return hmac.compare_digest(token, self._verification_token)
|