ForcePilot/backend/test/unit/channels/plugins/feishu/test_signature.py

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