JiuContinent
d7db84571d
feat: 添加飞书用户模块和订阅功能支持 - 新增feishu_users模块用于处理飞书用户身份验证和权限管理 - 新增subscriptions模块用于处理订阅相关功能 - 新增personalization模块用于个性化服务 - 在alembic迁移配置中注册新的模型模块 - 在API路由器中添加feishu_users和subscriptions路由 - 实现事件调度服务的改进，包括错误处理和状态更新优化 - 添加飞书命令处理的权限检查机制 - 实现飞书应用票据事件处理 - 改进审计日志记录功能 ```
85 lines
2.9 KiB
Python
85 lines
2.9 KiB
Python
import base64
|
|
import json
|
|
import time
|
|
from hashlib import sha256
|
|
|
|
import pytest
|
|
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
|
from cryptography.hazmat.primitives.padding import PKCS7
|
|
from fastapi import HTTPException
|
|
|
|
from app.core.config import get_settings
|
|
from app.modules.feishu.event_verification import FeishuWebhookVerifier
|
|
|
|
|
|
def _encrypt_event(payload: dict, encrypt_key: str) -> str:
|
|
key = sha256(encrypt_key.encode("utf-8")).digest()
|
|
padder = PKCS7(algorithms.AES.block_size).padder()
|
|
cleartext = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
|
padded = padder.update(cleartext) + padder.finalize()
|
|
encryptor = Cipher(algorithms.AES(key), modes.CBC(key[:16])).encryptor()
|
|
return base64.b64encode(encryptor.update(padded) + encryptor.finalize()).decode()
|
|
|
|
|
|
def test_plain_webhook_requires_the_configured_token(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("FEISHU_VERIFICATION_TOKEN", "verification-token")
|
|
monkeypatch.delenv("FEISHU_ENCRYPT_KEY", raising=False)
|
|
get_settings.cache_clear()
|
|
try:
|
|
raw = json.dumps(
|
|
{"header": {"token": "verification-token"}, "event": {}}
|
|
).encode()
|
|
assert FeishuWebhookVerifier().verify(raw, {})["event"] == {}
|
|
|
|
invalid = json.dumps(
|
|
{"header": {"token": "wrong-token"}, "event": {}}
|
|
).encode()
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
FeishuWebhookVerifier().verify(invalid, {})
|
|
assert exc_info.value.status_code == 401
|
|
finally:
|
|
get_settings.cache_clear()
|
|
|
|
|
|
def test_encrypted_webhook_requires_valid_signature_and_decrypts(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
encrypt_key = "test-encrypt-key"
|
|
monkeypatch.setenv("FEISHU_VERIFICATION_TOKEN", "verification-token")
|
|
monkeypatch.setenv("FEISHU_ENCRYPT_KEY", encrypt_key)
|
|
get_settings.cache_clear()
|
|
try:
|
|
event = {
|
|
"header": {
|
|
"token": "verification-token",
|
|
"tenant_key": "tenant-a",
|
|
},
|
|
"event": {"sender": {"sender_id": {"open_id": "ou-a"}}},
|
|
}
|
|
raw = json.dumps(
|
|
{"encrypt": _encrypt_event(event, encrypt_key)},
|
|
separators=(",", ":"),
|
|
).encode()
|
|
timestamp = str(int(time.time()))
|
|
nonce = "nonce"
|
|
signature = sha256(
|
|
timestamp.encode()
|
|
+ nonce.encode()
|
|
+ encrypt_key.encode()
|
|
+ raw
|
|
).hexdigest()
|
|
headers = {
|
|
"X-Lark-Request-Timestamp": timestamp,
|
|
"X-Lark-Request-Nonce": nonce,
|
|
"X-Lark-Signature": signature,
|
|
}
|
|
|
|
assert FeishuWebhookVerifier().verify(raw, headers) == event
|
|
|
|
headers["X-Lark-Signature"] = "invalid"
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
FeishuWebhookVerifier().verify(raw, headers)
|
|
assert exc_info.value.status_code == 401
|
|
finally:
|
|
get_settings.cache_clear()
|