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 == ""
|