# -*- coding: utf-8 -*-
"""Location: ./tests/unit/mcpgateway/utils/test_passthrough_headers_security.py
Copyright 2026
SPDX-License-Identifier: Apache-2.0
Authors: Mihai Criveti

Security tests for HTTP header passthrough utilities.
"""

# Standard
from unittest.mock import Mock, patch

# First-Party
from mcpgateway.cache.global_config_cache import global_config_cache
from mcpgateway.utils.passthrough_headers import (
    get_passthrough_headers,
    MAX_HEADER_VALUE_LENGTH,
    sanitize_header_value,
    validate_header_name,
)


class TestHeaderSecurity:
    """Test security features of header passthrough."""

    def setup_method(self):
        """Clear the global config cache before each test to ensure isolation."""
        global_config_cache.invalidate()

    @patch("mcpgateway.utils.passthrough_headers.settings")
    def test_feature_flag_disabled_by_default(self, mock_settings):
        """Test that the feature is disabled by default for security."""
        mock_settings.enable_header_passthrough = False

        mock_db = Mock()
        request_headers = {"x-tenant-id": "test"}
        base_headers = {"Content-Type": "application/json"}

        result = get_passthrough_headers(request_headers, base_headers, mock_db)

        # Should return only base headers when disabled
        assert result == base_headers

        # Database should not be queried when feature is disabled
        mock_db.query.assert_not_called()

    def test_sanitize_header_value_injection_prevention(self):
        """Test that header value sanitization prevents injection attacks."""
        # Test newline injection
        malicious_value = "valid-value\r\nX-Injected: malicious"
        result = sanitize_header_value(malicious_value)
        assert "\r" not in result
        assert "\n" not in result
        assert result == "valid-valueX-Injected: malicious"

    def test_sanitize_header_value_control_chars(self):
        """Test that control characters are removed."""
        # Include various control characters (ASCII 0-31 except tab)
        control_chars = "".join(chr(i) for i in range(32) if i != 9)  # Exclude tab
        malicious_value = f"start{control_chars}end"

        result = sanitize_header_value(malicious_value)

        # Should only contain printable chars and tab
        for char in result:
            char_code = ord(char)
            assert char_code >= 32 or char_code == 9

    def test_sanitize_header_value_length_limit(self):
        """Test that header values are limited to prevent DoS attacks."""
        oversized_value = "A" * (MAX_HEADER_VALUE_LENGTH * 2)
        result = sanitize_header_value(oversized_value)

        assert len(result) == MAX_HEADER_VALUE_LENGTH
        assert result == "A" * MAX_HEADER_VALUE_LENGTH

    def test_validate_header_name_injection_prevention(self):
        """Test that header name validation prevents injection."""
        malicious_names = [
            "Valid-Name\r\nX-Injected: value",  # Newline injection
            "Name With Spaces",  # Spaces not allowed
            "Name_With_Underscores",  # Underscores not allowed
            "Name.With.Dots",  # Dots not allowed
            "Name:With:Colons",  # Colons not allowed
            "Name/With/Slashes",  # Slashes not allowed
            "Name@With@Symbols",  # Special chars not allowed
            "",  # Empty string
            "Name\x00WithNull",  # Null byte
        ]

        for malicious_name in malicious_names:
            assert not validate_header_name(malicious_name), f"Should reject: {malicious_name!r}"

    def test_validate_header_name_allows_safe_names(self):
        """Test that safe header names are allowed."""
        safe_names = [
            "Authorization",
            "X-Tenant-Id",
            "Content-Type",
            "User-Agent",
            "X-123-Test",
            "X",
            "A-B-C-123",
        ]

        for safe_name in safe_names:
            assert validate_header_name(safe_name), f"Should allow: {safe_name!r}"

    @patch("mcpgateway.utils.passthrough_headers.settings")
    def test_header_validation_applied_in_passthrough(self, mock_settings, caplog):
        """Test that header validation is applied during passthrough."""
        mock_settings.enable_header_passthrough = True

        mock_db = Mock()
        mock_db.query.return_value.first.return_value = None  # No global config
        mock_settings.default_passthrough_headers = ["Invalid Header Name", "Valid-Header"]

        request_headers = {"invalid header name": "should-be-rejected", "valid-header": "should-pass"}
        base_headers = {"Content-Type": "application/json"}

        # Standard
        import logging

        with caplog.at_level(logging.WARNING, logger="mcpgateway.utils.passthrough_headers"):
            result = get_passthrough_headers(request_headers, base_headers, mock_db)

        # Only valid header should pass through
        expected = {"Content-Type": "application/json", "Valid-Header": "should-pass"}
        assert result == expected

        # Should log warning about invalid header name
        assert any("Invalid header name" in record.message for record in caplog.records)

    @patch("mcpgateway.utils.passthrough_headers.settings")
    def test_header_sanitization_applied_in_passthrough(self, mock_settings, caplog):
        """Test that header sanitization is applied during passthrough."""
        mock_settings.enable_header_passthrough = True

        mock_db = Mock()
        mock_db.query.return_value.first.return_value = None
        mock_settings.default_passthrough_headers = ["X-Test"]

        # Request with dangerous header value
        request_headers = {"x-test": "value\r\nwith\x01dangerous\x02chars"}
        base_headers = {"Content-Type": "application/json"}

        result = get_passthrough_headers(request_headers, base_headers, mock_db)

        # Header value should be sanitized
        expected = {"Content-Type": "application/json", "X-Test": "valuewithdangerouschars"}  # Sanitized
        assert result == expected

    @patch("mcpgateway.utils.passthrough_headers.settings")
    def test_empty_sanitized_header_skipped(self, mock_settings, caplog):
        """Test that headers that become empty after sanitization are skipped."""
        mock_settings.enable_header_passthrough = True

        mock_db = Mock()
        mock_db.query.return_value.first.return_value = None
        mock_settings.default_passthrough_headers = ["X-Test"]

        # Header with only dangerous characters
        request_headers = {"x-test": "\r\n\x01\x02\x03   "}  # Only control chars and whitespace
        base_headers = {"Content-Type": "application/json"}

        # Standard
        import logging

        with caplog.at_level(logging.WARNING, logger="mcpgateway.utils.passthrough_headers"):
            result = get_passthrough_headers(request_headers, base_headers, mock_db)

        # Header should be skipped
        expected = {"Content-Type": "application/json"}
        assert result == expected

        # Should log warning about empty header
        assert any("became empty after sanitization" in record.message for record in caplog.records)

    def test_sanitization_exception_handling(self, caplog):
        """Test that sanitization handles edge cases gracefully."""
        # Test with non-string input (should not happen in practice)
        with patch("mcpgateway.utils.passthrough_headers.sanitize_header_value") as mock_sanitize:
            mock_sanitize.side_effect = Exception("Test error")

            mock_settings = Mock()
            mock_settings.enable_header_passthrough = True
            mock_settings.default_passthrough_headers = ["X-Test"]

            mock_db = Mock()
            mock_db.query.return_value.first.return_value = None

            with patch("mcpgateway.utils.passthrough_headers.settings", mock_settings):
                # Standard
                import logging

                with caplog.at_level(logging.WARNING):
                    result = get_passthrough_headers({"x-test": "value"}, {"Content-Type": "application/json"}, mock_db)

                # Should skip the header due to sanitization error
                expected = {"Content-Type": "application/json"}
                assert result == expected

    def test_regex_pattern_security(self):
        """Test that the regex pattern is secure against ReDoS attacks."""
        # Test with potentially problematic input
        problematic_inputs = [
            "A" * 1000,  # Very long string
            "A-" * 500,  # Repeated pattern
            "A" + "-" * 1000 + "B",  # Long middle section
        ]

        # Standard
        import time

        for test_input in problematic_inputs:
            start_time = time.time()
            validate_header_name(test_input)
            end_time = time.time()

            # Should complete quickly (less than 1 second)
            assert (end_time - start_time) < 1.0, f"Regex took too long for: {test_input[:50]}..."

            # Our regex doesn't restrict length, so long valid patterns will pass
            # This is by design - length limiting is handled by sanitization

    @patch("mcpgateway.utils.passthrough_headers.settings")
    def test_authorization_header_security(self, mock_settings, caplog):
        """Test security around Authorization header handling."""
        mock_settings.enable_header_passthrough = True

        # Test that Authorization header is properly blocked with gateway auth
        mock_db = Mock()
        mock_db.query.return_value.first.return_value = None
        mock_settings.default_passthrough_headers = ["Authorization"]

        mock_gateway = Mock()
        mock_gateway.passthrough_headers = None
        mock_gateway.auth_type = "basic"
        mock_gateway.name = "secure-gateway"

        request_headers = {"authorization": "Bearer potentially-leaked-token"}
        base_headers = {}

        # Standard
        import logging

        with caplog.at_level(logging.WARNING, logger="mcpgateway.utils.passthrough_headers"):
            result = get_passthrough_headers(request_headers, base_headers, mock_db, mock_gateway)

        # Authorization should be blocked
        assert "Authorization" not in result

        # Should log security warning with gateway name
        assert any("Skipping Authorization header passthrough due to basic auth configuration on gateway secure-gateway" in record.message for record in caplog.records)

    def test_case_sensitivity_security(self):
        """Test that case sensitivity doesn't create security bypasses."""
        # Ensure that different cases of the same header are handled consistently
        test_cases = [
            ("Authorization", "authorization"),
            ("X-Tenant-Id", "x-tenant-id"),
            ("Content-Type", "content-type"),
        ]

        for config_case, request_case in test_cases:
            # Test that request headers are properly matched regardless of case
            mock_settings = Mock()
            mock_settings.enable_header_passthrough = True
            mock_settings.default_passthrough_headers = [config_case]

            mock_db = Mock()
            mock_db.query.return_value.first.return_value = None

            with patch("mcpgateway.utils.passthrough_headers.settings", mock_settings):
                result = get_passthrough_headers({request_case: "test-value"}, {}, mock_db)

                # Should match and use config case
                assert config_case in result
                assert result[config_case] == "test-value"

    def test_memory_safety_with_large_inputs(self):
        """Test memory safety with large inputs."""
        # Test with many headers
        large_request_headers = {f"x-header-{i}": f"value-{i}" for i in range(1000)}
        large_allowed_headers = [f"X-Header-{i}" for i in range(1000)]

        mock_settings = Mock()
        mock_settings.enable_header_passthrough = True
        mock_settings.default_passthrough_headers = large_allowed_headers

        mock_db = Mock()
        mock_db.query.return_value.first.return_value = None

        with patch("mcpgateway.utils.passthrough_headers.settings", mock_settings):
            # Should handle large inputs without crashing
            result = get_passthrough_headers(large_request_headers, {}, mock_db)

            # All headers should be processed
            assert len(result) == 1000

    def test_unicode_security(self):
        """Test security with Unicode characters in headers."""
        # Test various Unicode scenarios that could be problematic
        unicode_tests = [
            ("X-Unicode", "value-with-emoji-🔒"),
            ("X-Unicode", "value-with-accents-café"),
            ("X-Unicode", "value-with-chinese-中文"),
            ("X-Unicode", "value-with-rtl-עברית"),
        ]

        for header_name, header_value in unicode_tests:
            result = sanitize_header_value(header_value)
            # Should preserve Unicode characters (they're >= ASCII 32)
            assert result == header_value

            # Header name validation should work with Unicode
            unicode_header_name = "X-Unicode-Test-🔒"
            assert not validate_header_name(unicode_header_name)  # Emoji not allowed in names


class TestConfigSecurity:
    """Test security of configuration parsing."""

    def test_environment_variable_json_parsing(self):
        """Test JSON parsing of environment variables."""
        # Standard
        import json

        # First-Party
        from mcpgateway.config import Settings

        # Test with valid JSON
        test_headers = ["X-Test-1", "X-Test-2"]
        # Test JSON parsing through direct instantiation
        with patch("os.environ.get", return_value=json.dumps(test_headers)):
            settings = Settings()
            assert settings.default_passthrough_headers == test_headers

    def test_environment_variable_csv_fallback(self):
        """Test CSV fallback for environment variables."""
        # First-Party
        from mcpgateway.config import Settings

        # Test CSV parsing by simulating the __init__ logic directly
        # since pydantic tries JSON first and fails
        Settings()

        # Manually call the parsing logic that would be triggered by env var
        with patch("os.environ.get", return_value="X-Test-1,X-Test-2,X-Test-3"):
            # Simulate what happens in Settings.__init__
            default_value = "X-Test-1,X-Test-2,X-Test-3"
            try:
                # Try JSON parsing first - this should fail
                # Standard
                import json

                parsed = json.loads(default_value)
            except json.JSONDecodeError:
                # Fallback to comma-separated parsing - this should work
                parsed = [h.strip() for h in default_value.split(",") if h.strip()]

            assert parsed == ["X-Test-1", "X-Test-2", "X-Test-3"]

    def test_environment_variable_security_validation(self):
        """Test that environment variable parsing validates header names."""
        # Standard
        import json

        # First-Party
        from mcpgateway.config import Settings

        # Test with malicious header names
        malicious_headers = ["Valid-Header", "Invalid Header Name", "Another@Bad#Name"]

        # Test with malicious headers through direct instantiation
        with patch.dict("os.environ", {"DEFAULT_PASSTHROUGH_HEADERS": json.dumps(malicious_headers)}, clear=False):
            settings = Settings()

            # Should include all headers (validation happens during use, not config)
            assert len(settings.default_passthrough_headers) == 3
