Kris
b323922cdc
本次提交清理了多个单元测试文件中的多余空行，调整了部分导入的顺序，同时修复了一处feishu错误翻译的导入顺序问题，移除了application/conftest.py中未使用的fake_report_repo fixture，优化代码格式提升可读性。
363 lines
11 KiB
Python
363 lines
11 KiB
Python
"""signature 模块单元测试。
|
||
|
||
覆盖 Webhook 签名校验（``verify_webhook_signature``）、时间戳防重放校验
|
||
（``is_timestamp_valid``）、AES-256-CBC 事件解密（``decrypt_event``）与
|
||
URL 验证 challenge 提取（``extract_challenge``）。解密测试在
|
||
``cryptography`` 不可用时通过 monkeypatch 验证 DependencyError 路径，
|
||
可用时验证实际解密行为。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import hashlib
|
||
import json
|
||
import time
|
||
|
||
import pytest
|
||
from yuxi.channels.contract.errors import DependencyError, ValidationError
|
||
from yuxi.channels.plugins.feishu import signature
|
||
|
||
pytestmark = pytest.mark.unit
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# verify_webhook_signature
|
||
# ------------------------------------------------------------------
|
||
|
||
|
||
def _compute_signature(timestamp: str, nonce: str, encrypt_key: str, body: str) -> str:
|
||
"""计算飞书 Webhook 签名（SHA256）。"""
|
||
raw = f"{timestamp}{nonce}{encrypt_key}{body}".encode()
|
||
return hashlib.sha256(raw).hexdigest()
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestVerifyWebhookSignature:
|
||
"""verify_webhook_signature 签名校验测试。"""
|
||
|
||
def test_valid_signature_returns_true(self) -> None:
|
||
# Arrange
|
||
timestamp = "1700000000"
|
||
nonce = "abc123"
|
||
encrypt_key = "my_key"
|
||
body = '{"event":"message"}'
|
||
signature_str = _compute_signature(timestamp, nonce, encrypt_key, body)
|
||
|
||
# Act
|
||
result = signature.verify_webhook_signature(
|
||
timestamp,
|
||
nonce,
|
||
encrypt_key,
|
||
body,
|
||
signature_str,
|
||
)
|
||
|
||
# Assert
|
||
assert result is True
|
||
|
||
def test_invalid_signature_returns_false(self) -> None:
|
||
# Arrange
|
||
timestamp = "1700000000"
|
||
nonce = "abc123"
|
||
encrypt_key = "my_key"
|
||
body = '{"event":"message"}'
|
||
|
||
# Act
|
||
result = signature.verify_webhook_signature(
|
||
timestamp,
|
||
nonce,
|
||
encrypt_key,
|
||
body,
|
||
"wrong_signature",
|
||
)
|
||
|
||
# Assert
|
||
assert result is False
|
||
|
||
def test_tampered_body_returns_false(self) -> None:
|
||
# Arrange - 签名基于原始 body 计算，body 被篡改
|
||
timestamp = "1700000000"
|
||
nonce = "abc123"
|
||
encrypt_key = "my_key"
|
||
original_body = '{"event":"message"}'
|
||
tampered_body = '{"event":"edited"}'
|
||
signature_str = _compute_signature(timestamp, nonce, encrypt_key, original_body)
|
||
|
||
# Act
|
||
result = signature.verify_webhook_signature(
|
||
timestamp,
|
||
nonce,
|
||
encrypt_key,
|
||
tampered_body,
|
||
signature_str,
|
||
)
|
||
|
||
# Assert
|
||
assert result is False
|
||
|
||
def test_type_error_returns_false_without_raising(self) -> None:
|
||
# Arrange - 非字符串输入导致 TypeError，应返回 False 不抛异常
|
||
# Act
|
||
result = signature.verify_webhook_signature(
|
||
None,
|
||
"nonce",
|
||
"key",
|
||
"body",
|
||
"sig", # type: ignore[arg-type]
|
||
)
|
||
|
||
# Assert
|
||
assert result is False
|
||
|
||
def test_empty_strings_returns_false(self) -> None:
|
||
# Arrange - 空字符串 body 与不匹配的签名
|
||
# Act
|
||
result = signature.verify_webhook_signature(
|
||
"",
|
||
"",
|
||
"",
|
||
"",
|
||
"some_sig",
|
||
)
|
||
|
||
# Assert - 空字符串可编码，但签名不匹配
|
||
assert result is False
|
||
|
||
def test_empty_strings_with_correct_signature_returns_true(self) -> None:
|
||
# Arrange - 空字符串可正确计算签名
|
||
sig = _compute_signature("", "", "", "")
|
||
|
||
# Act
|
||
result = signature.verify_webhook_signature("", "", "", "", sig)
|
||
|
||
# Assert
|
||
assert result is True
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# is_timestamp_valid
|
||
# ------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestIsTimestampValid:
|
||
"""is_timestamp_valid 时间戳防重放测试。"""
|
||
|
||
def test_current_timestamp_valid(self) -> None:
|
||
# Arrange
|
||
ts = str(int(time.time()))
|
||
|
||
# Act
|
||
result = signature.is_timestamp_valid(ts)
|
||
|
||
# Assert
|
||
assert result is True
|
||
|
||
def test_timestamp_within_skew_window_valid(self) -> None:
|
||
# Arrange - 100 秒前，在 300 秒窗口内
|
||
ts = str(int(time.time()) - 100)
|
||
|
||
# Act
|
||
result = signature.is_timestamp_valid(ts)
|
||
|
||
# Assert
|
||
assert result is True
|
||
|
||
def test_timestamp_future_within_skew_valid(self) -> None:
|
||
# Arrange - 100 秒后，在 300 秒窗口内
|
||
ts = str(int(time.time()) + 100)
|
||
|
||
# Act
|
||
result = signature.is_timestamp_valid(ts)
|
||
|
||
# Assert
|
||
assert result is True
|
||
|
||
def test_expired_timestamp_invalid(self) -> None:
|
||
# Arrange - 600 秒前，超出 300 秒窗口
|
||
ts = str(int(time.time()) - 600)
|
||
|
||
# Act
|
||
result = signature.is_timestamp_valid(ts)
|
||
|
||
# Assert
|
||
assert result is False
|
||
|
||
def test_custom_max_skew_window(self) -> None:
|
||
# Arrange - 50 秒前，使用 30 秒窗口
|
||
ts = str(int(time.time()) - 50)
|
||
|
||
# Act
|
||
result = signature.is_timestamp_valid(ts, max_skew_seconds=30)
|
||
|
||
# Assert
|
||
assert result is False
|
||
|
||
def test_invalid_format_returns_false(self) -> None:
|
||
# Arrange - 非数字字符串
|
||
# Act
|
||
result = signature.is_timestamp_valid("not-a-timestamp")
|
||
|
||
# Assert
|
||
assert result is False
|
||
|
||
def test_none_input_returns_false(self) -> None:
|
||
# Arrange / Act
|
||
result = signature.is_timestamp_valid(None) # type: ignore[arg-type]
|
||
|
||
# Assert
|
||
assert result is False
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# decrypt_event
|
||
# ------------------------------------------------------------------
|
||
|
||
|
||
def _encrypt_payload(encrypt_key: str, plaintext_dict: dict) -> str:
|
||
"""使用 AES-256-CBC 加密构造测试用 payload。"""
|
||
from cryptography.hazmat.backends import default_backend
|
||
from cryptography.hazmat.primitives import padding as sympad
|
||
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
||
|
||
key = hashlib.sha256(encrypt_key.encode("utf-8")).digest()[:32]
|
||
iv = b"\x00" * 16
|
||
plaintext = json.dumps(plaintext_dict).encode("utf-8")
|
||
padder = sympad.PKCS7(128).padder()
|
||
padded = padder.update(plaintext) + padder.finalize()
|
||
cipher = Cipher(algorithms.AES(key), modes.CBC(iv), backend=default_backend())
|
||
encryptor = cipher.encryptor()
|
||
ciphertext = encryptor.update(padded) + encryptor.finalize()
|
||
return base64.b64encode(iv + ciphertext).decode("utf-8")
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestDecryptEvent:
|
||
"""decrypt_event AES-256-CBC 解密测试。"""
|
||
|
||
def test_cryptography_unavailable_raises_dependency_error(
|
||
self,
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
# Arrange - 模拟 cryptography 不可用
|
||
monkeypatch.setattr(signature, "_HAS_CRYPTOGRAPHY", False)
|
||
|
||
# Act / Assert
|
||
with pytest.raises(DependencyError) as exc_info:
|
||
signature.decrypt_event("payload", "key")
|
||
assert exc_info.value.dep == "cryptography"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_valid_decryption_returns_dict(self) -> None:
|
||
# Arrange - 需要 cryptography
|
||
pytest.importorskip("cryptography")
|
||
encrypt_key = "test_encrypt_key"
|
||
original = {"event": "message", "data": "hello"}
|
||
payload = _encrypt_payload(encrypt_key, original)
|
||
|
||
# Act
|
||
result = signature.decrypt_event(payload, encrypt_key)
|
||
|
||
# Assert
|
||
assert result == original
|
||
|
||
def test_invalid_base64_raises_validation_error(self) -> None:
|
||
# Arrange
|
||
pytest.importorskip("cryptography")
|
||
|
||
# Act / Assert
|
||
with pytest.raises(ValidationError) as exc_info:
|
||
signature.decrypt_event("!!!not-valid-base64!!!", "key")
|
||
assert exc_info.value.field == "encrypt_payload"
|
||
assert exc_info.value.message == "decrypt_failed"
|
||
|
||
def test_invalid_json_raises_validation_error(self) -> None:
|
||
# Arrange - 解密成功但 JSON 解析失败
|
||
pytest.importorskip("cryptography")
|
||
from cryptography.hazmat.backends import default_backend
|
||
from cryptography.hazmat.primitives import padding as sympad
|
||
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
|
||
|
||
encrypt_key = "key"
|
||
key = hashlib.sha256(encrypt_key.encode("utf-8")).digest()[:32]
|
||
iv = b"\x00" * 16
|
||
plaintext = b"not-json-at-all"
|
||
padder = sympad.PKCS7(128).padder()
|
||
padded = padder.update(plaintext) + padder.finalize()
|
||
cipher = Cipher(algorithms.AES(key), modes.CBC(iv), backend=default_backend())
|
||
encryptor = cipher.encryptor()
|
||
ciphertext = encryptor.update(padded) + encryptor.finalize()
|
||
payload = base64.b64encode(iv + ciphertext).decode("utf-8")
|
||
|
||
# Act / Assert
|
||
with pytest.raises(ValidationError):
|
||
signature.decrypt_event(payload, encrypt_key)
|
||
|
||
def test_validation_error_preserves_cause(self) -> None:
|
||
# Arrange
|
||
pytest.importorskip("cryptography")
|
||
|
||
# Act / Assert - ValidationError 通过 raise ... from exc 链式保留原始异常
|
||
with pytest.raises(ValidationError) as exc_info:
|
||
signature.decrypt_event("!!!invalid!!!", "key")
|
||
assert exc_info.value.__cause__ is not None
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# extract_challenge
|
||
# ------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestExtractChallenge:
|
||
"""extract_challenge URL 验证 challenge 提取测试。"""
|
||
|
||
def test_valid_challenge_string_returned(self) -> None:
|
||
# Arrange
|
||
payload = {"challenge": "ajksdfhkjdsfhkj", "token": "xxx"}
|
||
|
||
# Act
|
||
result = signature.extract_challenge(payload)
|
||
|
||
# Assert
|
||
assert result == "ajksdfhkjdsfhkj"
|
||
|
||
def test_missing_challenge_returns_none(self) -> None:
|
||
# Arrange
|
||
payload = {"event": "message"}
|
||
|
||
# Act
|
||
result = signature.extract_challenge(payload)
|
||
|
||
# Assert
|
||
assert result is None
|
||
|
||
def test_non_string_challenge_returns_none(self) -> None:
|
||
# Arrange - challenge 为数字
|
||
payload = {"challenge": 12345}
|
||
|
||
# Act
|
||
result = signature.extract_challenge(payload)
|
||
|
||
# Assert
|
||
assert result is None
|
||
|
||
def test_empty_payload_returns_none(self) -> None:
|
||
# Arrange
|
||
# Act
|
||
result = signature.extract_challenge({})
|
||
|
||
# Assert
|
||
assert result is None
|
||
|
||
def test_empty_string_challenge_returned(self) -> None:
|
||
# Arrange - 空字符串仍是有效字符串
|
||
payload = {"challenge": ""}
|
||
|
||
# Act
|
||
result = signature.extract_challenge(payload)
|
||
|
||
# Assert
|
||
assert result == ""
|