tracer-cloud--opensre/tests/integrations/test_messaging_security.py

"""Tests for integrations.messaging_security — identity model, pairing, and authorization."""

from __future__ import annotations

import time

from integrations.messaging_security import (
    _MAX_PAIRING_ATTEMPTS,
    _PAIRING_CODE_TTL_SECONDS,
    AuthorizationResult,
    MessagingIdentityPolicy,
    MessagingPlatform,
    RejectionBehavior,
    authorize_inbound_message,
    complete_pairing,
    generate_pairing_code,
    hash_pairing_code,
    verify_pairing_code,
)

# ---------------------------------------------------------------------------
# MessagingIdentityPolicy model tests
# ---------------------------------------------------------------------------


class TestMessagingIdentityPolicy:
    def test_default_policy_has_empty_allowlists(self) -> None:
        policy = MessagingIdentityPolicy()
        assert policy.allowed_user_ids == []
        assert policy.allowed_chat_ids == []
        assert policy.require_dm_pairing is True
        assert policy.pairing_secret_hash is None
        assert policy.pairing_created_at is None
        assert policy.pairing_attempts == 0
        assert policy.rejection_behavior == RejectionBehavior.REPLY
        assert policy.inbound_enabled is False

    def test_policy_with_allowed_users(self) -> None:
        policy = MessagingIdentityPolicy(
            allowed_user_ids=["123", "456"],
            inbound_enabled=True,
        )
        assert policy.allowed_user_ids == ["123", "456"]
        assert policy.inbound_enabled is True

    def test_policy_serialization_roundtrip(self) -> None:
        policy = MessagingIdentityPolicy(
            allowed_user_ids=["user1"],
            allowed_chat_ids=["chat1"],
            require_dm_pairing=False,
            rejection_behavior=RejectionBehavior.DROP,
            inbound_enabled=True,
        )
        data = policy.model_dump(mode="json")
        restored = MessagingIdentityPolicy.model_validate(data)
        assert restored.allowed_user_ids == ["user1"]
        assert restored.allowed_chat_ids == ["chat1"]
        assert restored.require_dm_pairing is False
        assert restored.rejection_behavior == RejectionBehavior.DROP
        assert restored.inbound_enabled is True

    def test_policy_back_compat_empty_dict_loads(self) -> None:
        """Existing configs with no identity_policy should load with defaults."""
        policy = MessagingIdentityPolicy.model_validate({})
        assert policy.allowed_user_ids == []
        assert policy.inbound_enabled is False


# ---------------------------------------------------------------------------
# Pairing code tests
# ---------------------------------------------------------------------------


class TestPairingCode:
    def test_generate_pairing_code_length(self) -> None:
        code = generate_pairing_code()
        assert len(code) == 6

    def test_generate_pairing_code_is_alphanumeric_uppercase(self) -> None:
        code = generate_pairing_code()
        assert code.isalnum()
        assert code == code.upper()

    def test_generate_pairing_code_is_random(self) -> None:
        codes = {generate_pairing_code() for _ in range(100)}
        # With 36^6 possible codes, 100 should all be unique
        assert len(codes) == 100

    def test_hash_pairing_code_deterministic(self) -> None:
        code = "ABC123"
        h1 = hash_pairing_code(code)
        h2 = hash_pairing_code(code)
        assert h1 == h2

    def test_hash_pairing_code_case_insensitive(self) -> None:
        assert hash_pairing_code("ABC123") == hash_pairing_code("abc123")

    def test_hash_pairing_code_is_hex_string(self) -> None:
        h = hash_pairing_code("TEST01")
        assert len(h) == 64  # SHA-256 hex
        int(h, 16)  # Should not raise

    def test_verify_pairing_code_correct(self) -> None:
        code = "XYZ789"
        stored = hash_pairing_code(code)
        assert verify_pairing_code(code, stored) is True

    def test_verify_pairing_code_wrong(self) -> None:
        stored = hash_pairing_code("CORRECT")
        assert verify_pairing_code("WRONG1", stored) is False

    def test_verify_pairing_code_case_insensitive(self) -> None:
        code = "ABC123"
        stored = hash_pairing_code(code)
        assert verify_pairing_code("abc123", stored) is True


# ---------------------------------------------------------------------------
# Authorization tests
# ---------------------------------------------------------------------------


class TestAuthorizeInboundMessage:
    def test_inbound_disabled_rejects(self) -> None:
        policy = MessagingIdentityPolicy(inbound_enabled=False)
        result = authorize_inbound_message(policy=policy, user_id="123")
        assert not result.allowed
        assert "not enabled" in result.reason

    def test_pairing_attempt_allowed_when_pending(self) -> None:
        policy = MessagingIdentityPolicy(
            inbound_enabled=True,
            pairing_secret_hash=hash_pairing_code("ABC123"),
        )
        result = authorize_inbound_message(
            policy=policy, user_id="unknown", message_text="/pair ABC123"
        )
        assert result.allowed
        assert result.is_pairing_attempt

    def test_pairing_attempt_rejected_when_no_pending(self) -> None:
        """When no pairing is pending, /pair messages should be rejected."""
        policy = MessagingIdentityPolicy(
            inbound_enabled=True,
            pairing_secret_hash=None,
        )
        result = authorize_inbound_message(
            policy=policy, user_id="unknown", message_text="/pair ABC123"
        )
        assert not result.allowed
        assert "no pairing" in result.reason.lower()

    def test_pairing_attempt_case_insensitive(self) -> None:
        policy = MessagingIdentityPolicy(
            inbound_enabled=True,
            pairing_secret_hash=hash_pairing_code("ABC123"),
        )
        result = authorize_inbound_message(
            policy=policy, user_id="unknown", message_text="/Pair abc123"
        )
        assert result.allowed
        assert result.is_pairing_attempt

    def test_allowed_user_passes(self) -> None:
        policy = MessagingIdentityPolicy(
            inbound_enabled=True,
            allowed_user_ids=["user1", "user2"],
        )
        result = authorize_inbound_message(policy=policy, user_id="user1")
        assert result.allowed

    def test_disallowed_user_rejected(self) -> None:
        policy = MessagingIdentityPolicy(
            inbound_enabled=True,
            allowed_user_ids=["user1"],
        )
        result = authorize_inbound_message(policy=policy, user_id="intruder")
        assert not result.allowed
        assert "not in the allowed" in result.reason

    def test_empty_allowlist_with_pairing_required_rejects(self) -> None:
        policy = MessagingIdentityPolicy(
            inbound_enabled=True,
            require_dm_pairing=True,
            allowed_user_ids=[],
        )
        result = authorize_inbound_message(policy=policy, user_id="anyone")
        assert not result.allowed
        assert "No users have been paired" in result.reason

    def test_empty_allowlist_without_pairing_allows(self) -> None:
        policy = MessagingIdentityPolicy(
            inbound_enabled=True,
            require_dm_pairing=False,
            allowed_user_ids=[],
        )
        result = authorize_inbound_message(policy=policy, user_id="anyone")
        assert result.allowed

    def test_chat_id_restriction(self) -> None:
        policy = MessagingIdentityPolicy(
            inbound_enabled=True,
            allowed_user_ids=["user1"],
            allowed_chat_ids=["chat1"],
        )
        # Allowed chat
        result = authorize_inbound_message(policy=policy, user_id="user1", chat_id="chat1")
        assert result.allowed

        # Disallowed chat
        result = authorize_inbound_message(policy=policy, user_id="user1", chat_id="other_chat")
        assert not result.allowed
        assert "not in the allowed chat" in result.reason

    def test_chat_id_none_blocked_when_allowlist_configured(self) -> None:
        """When allowed_chat_ids is set, None chat_id should be blocked."""
        policy = MessagingIdentityPolicy(
            inbound_enabled=True,
            allowed_user_ids=["user1"],
            allowed_chat_ids=["chat1"],
        )
        result = authorize_inbound_message(policy=policy, user_id="user1", chat_id=None)
        assert not result.allowed
        assert "not in the allowed chat" in result.reason

    def test_pairing_blocked_from_restricted_chat(self) -> None:
        """Pairing attempts from outside allowed_chat_ids are blocked."""
        policy = MessagingIdentityPolicy(
            inbound_enabled=True,
            allowed_chat_ids=["chat1"],
            pairing_secret_hash=hash_pairing_code("CODE01"),
        )
        result = authorize_inbound_message(
            policy=policy, user_id="anyone", chat_id="other_chat", message_text="/pair CODE01"
        )
        assert not result.allowed
        assert "not in the allowed chat" in result.reason

    def test_authorization_result_bool(self) -> None:
        allowed = AuthorizationResult(allowed=True, reason="ok")
        denied = AuthorizationResult(allowed=False, reason="no")
        assert bool(allowed) is True
        assert bool(denied) is False


# ---------------------------------------------------------------------------
# Complete pairing tests
# ---------------------------------------------------------------------------


class TestCompletePairing:
    def test_successful_pairing(self) -> None:
        code = "TEST01"
        policy = MessagingIdentityPolicy(
            inbound_enabled=True,
            pairing_secret_hash=hash_pairing_code(code),
            pairing_created_at=time.time(),
        )
        success, message = complete_pairing(policy=policy, user_id="new_user", code=code)
        assert success is True
        assert "successful" in message.lower()
        assert "new_user" in policy.allowed_user_ids
        assert policy.pairing_secret_hash is None
        assert policy.pairing_attempts == 0

    def test_pairing_wrong_code_increments_attempts(self) -> None:
        policy = MessagingIdentityPolicy(
            inbound_enabled=True,
            pairing_secret_hash=hash_pairing_code("CORRECT"),
            pairing_created_at=time.time(),
        )
        success, message = complete_pairing(policy=policy, user_id="user1", code="WRONG1")
        assert success is False
        assert "invalid" in message.lower()
        assert "user1" not in policy.allowed_user_ids
        # Hash should NOT be cleared on single failure
        assert policy.pairing_secret_hash is not None
        assert policy.pairing_attempts == 1

    def test_pairing_no_pending(self) -> None:
        policy = MessagingIdentityPolicy(
            inbound_enabled=True,
            pairing_secret_hash=None,
        )
        success, message = complete_pairing(policy=policy, user_id="user1", code="ANY123")
        assert success is False
        assert "no pairing" in message.lower()

    def test_pairing_does_not_duplicate_user(self) -> None:
        code = "DUP001"
        policy = MessagingIdentityPolicy(
            inbound_enabled=True,
            allowed_user_ids=["existing_user"],
            pairing_secret_hash=hash_pairing_code(code),
            pairing_created_at=time.time(),
        )
        success, _ = complete_pairing(policy=policy, user_id="existing_user", code=code)
        assert success is True
        assert policy.allowed_user_ids.count("existing_user") == 1

    def test_brute_force_invalidates_code(self) -> None:
        """After MAX_PAIRING_ATTEMPTS failures, the code is invalidated."""
        code = "SECRET"
        policy = MessagingIdentityPolicy(
            inbound_enabled=True,
            pairing_secret_hash=hash_pairing_code(code),
            pairing_created_at=time.time(),
        )
        # Exhaust all attempts
        for i in range(_MAX_PAIRING_ATTEMPTS):
            success, _ = complete_pairing(policy=policy, user_id="attacker", code=f"WRONG{i}")
            assert success is False

        # Code should now be invalidated
        assert policy.pairing_secret_hash is None
        assert policy.pairing_attempts == 0

    def test_expired_code_rejected(self) -> None:
        """A pairing code that has exceeded its TTL is rejected."""
        code = "EXPIRE"
        policy = MessagingIdentityPolicy(
            inbound_enabled=True,
            pairing_secret_hash=hash_pairing_code(code),
            pairing_created_at=time.time() - _PAIRING_CODE_TTL_SECONDS - 1,
        )
        success, message = complete_pairing(policy=policy, user_id="user1", code=code)
        assert success is False
        assert "expired" in message.lower()
        assert policy.pairing_secret_hash is None

    def test_missing_pairing_created_at_treated_as_expired(self) -> None:
        """A hash with no timestamp (legacy or corrupted) is treated as expired."""
        code = "LEGACY"
        policy = MessagingIdentityPolicy(
            inbound_enabled=True,
            pairing_secret_hash=hash_pairing_code(code),
            pairing_created_at=None,
        )
        success, message = complete_pairing(policy=policy, user_id="user1", code=code)
        assert success is False
        assert "expired" in message.lower()
        assert policy.pairing_secret_hash is None


# ---------------------------------------------------------------------------
# Platform enum tests
# ---------------------------------------------------------------------------


class TestMessagingPlatform:
    def test_platform_values(self) -> None:
        assert MessagingPlatform.TELEGRAM.value == "telegram"
        assert MessagingPlatform.SLACK.value == "slack"
        assert MessagingPlatform.DISCORD.value == "discord"