Kris
361dbe2149
本次提交修复了多个测试文件中的问题： 1. 为各管道测试类添加tracer_port属性初始化 2. 修复QQBot客户端关闭异常的错误类型匹配 3. 修正配对接口查询的状态值小写格式 4. 移除ChannelType枚举值的显式.value调用 5. 新增QQ常量的导出项与Instagram测试桩函数 6. 修复API接口返回值解析，正确访问data字段 7. 调整企业微信富文本消息的断言结构 8. 修正微信iLink的消息类型测试用例 9. 替换废弃的datetime.utc相关导入为UTC常量 10. 修复QQBot白名单适配器的返回值结构 11. 调整outbox工具的channel_msg_id处理逻辑 12. 新增多个适配器与服务的测试用例，覆盖异常降级、缓存处理等场景 13. 重构部分QQBot适配器的辅助函数测试，清理冗余代码 14. 修复微信iLink客户端的上传接口调用参数与缓存清理逻辑 15. 修正配置导出接口的敏感字段过滤与返回值结构
307 lines
9.3 KiB
Python
307 lines
9.3 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 - 钉钉官方规则：key=sign_secret, msg=string_to_sign
|
||
timestamp = 1700000000000
|
||
sign_secret = "my_secret"
|
||
string_to_sign = f"{timestamp}\n{sign_secret}"
|
||
expected = base64.b64encode(
|
||
hmac.new(
|
||
key=sign_secret.encode("utf-8"),
|
||
msg=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 == ""
|