Kris
532e7c59ec
1. 新增QQBot、Wecom、Dingtalk、Instagram、WechatMp等渠道的单元测试夹具与测试用例 2. 修复WechatILink的缓存键顺序与测试用例 3. 补全Feishu适配器的日志断言与测试逻辑 4. 清理WechatILink冗余的clear_context_token测试代码
306 lines
9.2 KiB
Python
306 lines
9.2 KiB
Python
"""dingtalk signature 模块单元测试。
|
||
|
||
覆盖钉钉 HTTP 模式 Webhook 回调的签名计算与验签（设计方案 §3.3）：
|
||
- ``compute_signature``：HMAC-SHA256 + Base64 编码。
|
||
- ``verify_signature``：正确签名通过、错误签名失败、时间戳超时拒绝、
|
||
空值拒绝（恒定时间比较防时序攻击）。
|
||
- ``extract_timestamp_sign``：从 payload 提取 timestamp 与 sign，
|
||
字段缺失返回 ``(0, "")``，timestamp 非整数抛 ``ValidationError``。
|
||
|
||
钉钉签名规则：拼接 ``timestamp + "\\n" + sign_secret``，计算 HMAC-SHA256
|
||
并 Base64 编码；timestamp 与当前时间差超过 60 分钟（3600 秒）拒绝。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import base64
|
||
import hashlib
|
||
import hmac
|
||
import time
|
||
|
||
import pytest
|
||
from yuxi.channels.contract.errors import ValidationError
|
||
from yuxi.channels.plugins.dingtalk import signature
|
||
|
||
pytestmark = pytest.mark.unit
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# compute_signature
|
||
# ------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestComputeSignature:
|
||
"""compute_signature 签名计算测试。"""
|
||
|
||
def test_returns_base64_string(self) -> None:
|
||
# Arrange
|
||
timestamp = 1700000000000
|
||
sign_secret = "my_secret"
|
||
|
||
# Act
|
||
result = signature.compute_signature(timestamp, sign_secret)
|
||
|
||
# Assert - 应为可解码的 Base64 字符串
|
||
assert isinstance(result, str)
|
||
decoded = base64.b64decode(result)
|
||
assert len(decoded) == 32 # SHA256 摘要长度
|
||
|
||
def test_matches_expected_hmac_sha256(self) -> None:
|
||
# Arrange
|
||
timestamp = 1700000000000
|
||
sign_secret = "my_secret"
|
||
string_to_sign = f"{timestamp}\n{sign_secret}"
|
||
expected = base64.b64encode(
|
||
hmac.new(
|
||
string_to_sign.encode("utf-8"),
|
||
digestmod=hashlib.sha256,
|
||
).digest()
|
||
).decode("utf-8")
|
||
|
||
# Act
|
||
result = signature.compute_signature(timestamp, sign_secret)
|
||
|
||
# Assert
|
||
assert result == expected
|
||
|
||
def test_different_secret_produces_different_signature(self) -> None:
|
||
# Arrange
|
||
timestamp = 1700000000000
|
||
|
||
# Act
|
||
sig1 = signature.compute_signature(timestamp, "secret1")
|
||
sig2 = signature.compute_signature(timestamp, "secret2")
|
||
|
||
# Assert
|
||
assert sig1 != sig2
|
||
|
||
def test_different_timestamp_produces_different_signature(self) -> None:
|
||
# Arrange
|
||
sign_secret = "my_secret"
|
||
|
||
# Act
|
||
sig1 = signature.compute_signature(1700000000000, sign_secret)
|
||
sig2 = signature.compute_signature(1700000000001, sign_secret)
|
||
|
||
# Assert
|
||
assert sig1 != sig2
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# verify_signature
|
||
# ------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestVerifySignature:
|
||
"""verify_signature 签名校验测试。"""
|
||
|
||
def test_valid_signature_returns_true(self) -> None:
|
||
# Arrange
|
||
timestamp = int(time.time() * 1000)
|
||
sign_secret = "my_secret"
|
||
sign = signature.compute_signature(timestamp, sign_secret)
|
||
|
||
# Act
|
||
result = signature.verify_signature(timestamp, sign, sign_secret)
|
||
|
||
# Assert
|
||
assert result is True
|
||
|
||
def test_invalid_signature_returns_false(self) -> None:
|
||
# Arrange
|
||
timestamp = int(time.time() * 1000)
|
||
sign_secret = "my_secret"
|
||
|
||
# Act
|
||
result = signature.verify_signature(timestamp, "wrong-signature", sign_secret)
|
||
|
||
# Assert
|
||
assert result is False
|
||
|
||
def test_tampered_timestamp_returns_false(self) -> None:
|
||
# Arrange - 签名基于原 timestamp 计算，timestamp 被篡改
|
||
timestamp = int(time.time() * 1000)
|
||
sign_secret = "my_secret"
|
||
sign = signature.compute_signature(timestamp, sign_secret)
|
||
|
||
# Act
|
||
result = signature.verify_signature(timestamp + 1, sign, sign_secret)
|
||
|
||
# Assert
|
||
assert result is False
|
||
|
||
def test_expired_timestamp_returns_false(self) -> None:
|
||
# Arrange - timestamp 超出 60 分钟（3600 秒）窗口
|
||
now = int(time.time() * 1000)
|
||
expired_ts = now - 3601 * 1000 # 3601 秒前
|
||
sign_secret = "my_secret"
|
||
sign = signature.compute_signature(expired_ts, sign_secret)
|
||
|
||
# Act
|
||
result = signature.verify_signature(expired_ts, sign, sign_secret, now_ts=now)
|
||
|
||
# Assert
|
||
assert result is False
|
||
|
||
def test_future_timestamp_beyond_tolerance_returns_false(self) -> None:
|
||
# Arrange - timestamp 在未来超出 60 分钟窗口
|
||
now = int(time.time() * 1000)
|
||
future_ts = now + 3601 * 1000 # 3601 秒后
|
||
sign_secret = "my_secret"
|
||
sign = signature.compute_signature(future_ts, sign_secret)
|
||
|
||
# Act
|
||
result = signature.verify_signature(future_ts, sign, sign_secret, now_ts=now)
|
||
|
||
# Assert
|
||
assert result is False
|
||
|
||
def test_timestamp_within_tolerance_returns_true(self) -> None:
|
||
# Arrange - 100 秒前，在 3600 秒窗口内
|
||
now = int(time.time() * 1000)
|
||
timestamp = now - 100 * 1000
|
||
sign_secret = "my_secret"
|
||
sign = signature.compute_signature(timestamp, sign_secret)
|
||
|
||
# Act
|
||
result = signature.verify_signature(timestamp, sign, sign_secret, now_ts=now)
|
||
|
||
# Assert
|
||
assert result is True
|
||
|
||
def test_empty_sign_returns_false(self) -> None:
|
||
# Arrange
|
||
timestamp = int(time.time() * 1000)
|
||
|
||
# Act
|
||
result = signature.verify_signature(timestamp, "", "secret")
|
||
|
||
# Assert
|
||
assert result is False
|
||
|
||
def test_empty_sign_secret_returns_false(self) -> None:
|
||
# Arrange
|
||
timestamp = int(time.time() * 1000)
|
||
|
||
# Act
|
||
result = signature.verify_signature(timestamp, "sig", "")
|
||
|
||
# Assert
|
||
assert result is False
|
||
|
||
def test_zero_timestamp_returns_false(self) -> None:
|
||
# Arrange
|
||
# Act
|
||
result = signature.verify_signature(0, "sig", "secret")
|
||
|
||
# Assert
|
||
assert result is False
|
||
|
||
def test_now_ts_none_uses_real_time(self) -> None:
|
||
# Arrange - now_ts 为 None 时取真实时间，近期 timestamp 应通过
|
||
timestamp = int(time.time() * 1000)
|
||
sign_secret = "my_secret"
|
||
sign = signature.compute_signature(timestamp, sign_secret)
|
||
|
||
# Act
|
||
result = signature.verify_signature(timestamp, sign, sign_secret)
|
||
|
||
# Assert
|
||
assert result is True
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# extract_timestamp_sign
|
||
# ------------------------------------------------------------------
|
||
|
||
|
||
@pytest.mark.unit
|
||
class TestExtractTimestampSign:
|
||
"""extract_timestamp_sign 提取测试。"""
|
||
|
||
def test_extracts_timestamp_and_sign(self) -> None:
|
||
# Arrange
|
||
payload = {"timestamp": 1700000000000, "sign": "abc123", "event": "msg"}
|
||
|
||
# Act
|
||
timestamp, sign = signature.extract_timestamp_sign(payload)
|
||
|
||
# Assert
|
||
assert timestamp == 1700000000000
|
||
assert sign == "abc123"
|
||
|
||
def test_missing_timestamp_returns_zero(self) -> None:
|
||
# Arrange
|
||
payload = {"sign": "abc123"}
|
||
|
||
# Act
|
||
timestamp, sign = signature.extract_timestamp_sign(payload)
|
||
|
||
# Assert
|
||
assert timestamp == 0
|
||
assert sign == "abc123"
|
||
|
||
def test_missing_sign_returns_empty_string(self) -> None:
|
||
# Arrange
|
||
payload = {"timestamp": 1700000000000}
|
||
|
||
# Act
|
||
timestamp, sign = signature.extract_timestamp_sign(payload)
|
||
|
||
# Assert
|
||
assert timestamp == 1700000000000
|
||
assert sign == ""
|
||
|
||
def test_missing_both_returns_defaults(self) -> None:
|
||
# Arrange
|
||
payload = {"event": "msg"}
|
||
|
||
# Act
|
||
timestamp, sign = signature.extract_timestamp_sign(payload)
|
||
|
||
# Assert
|
||
assert timestamp == 0
|
||
assert sign == ""
|
||
|
||
def test_empty_payload_returns_defaults(self) -> None:
|
||
# Arrange / Act
|
||
timestamp, sign = signature.extract_timestamp_sign({})
|
||
|
||
# Assert
|
||
assert timestamp == 0
|
||
assert sign == ""
|
||
|
||
def test_string_timestamp_converted_to_int(self) -> None:
|
||
# Arrange - timestamp 为字符串形式
|
||
payload = {"timestamp": "1700000000000", "sign": "sig"}
|
||
|
||
# Act
|
||
timestamp, sign = signature.extract_timestamp_sign(payload)
|
||
|
||
# Assert
|
||
assert timestamp == 1700000000000
|
||
assert sign == "sig"
|
||
|
||
def test_non_numeric_string_timestamp_raises_validation_error(self) -> None:
|
||
# Arrange - timestamp 为非数字字符串
|
||
payload = {"timestamp": "not-a-number", "sign": "sig"}
|
||
|
||
# Act / Assert
|
||
with pytest.raises(ValidationError) as exc_info:
|
||
signature.extract_timestamp_sign(payload)
|
||
assert exc_info.value.field == "timestamp"
|
||
|
||
def test_null_sign_normalized_to_empty_string(self) -> None:
|
||
# Arrange - sign 为 None
|
||
payload = {"timestamp": 1700000000000, "sign": None}
|
||
|
||
# Act
|
||
timestamp, sign = signature.extract_timestamp_sign(payload)
|
||
|
||
# Assert
|
||
assert timestamp == 1700000000000
|
||
assert sign == ""
|