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

Test multi-header authentication functionality.
"""

# Standard
import base64
import logging
from unittest.mock import AsyncMock, MagicMock, patch

# Third-Party
from fastapi import Request
from pydantic import ValidationError
import pytest
from starlette.datastructures import FormData

# First-Party
from mcpgateway.admin import admin_add_gateway
from mcpgateway.config import settings
from mcpgateway.schemas import GatewayCreate, GatewayRead, GatewayUpdate
from mcpgateway.services.gateway_service import GatewayService
from mcpgateway.utils.services_auth import decode_auth, encode_auth


class TestMultiAuthHeaders:
    """Test cases for multi-header authentication feature."""

    @pytest.mark.asyncio
    async def test_gateway_create_with_valid_multi_headers(self):
        """Test creating gateway with valid multi-auth headers."""
        auth_headers = [{"key": "X-API-Key", "value": "secret123"}, {"key": "X-Client-ID", "value": "client456"}, {"key": "X-Region", "value": "us-east-1"}]

        gateway = GatewayCreate(name="Test Gateway", url="http://example.com", auth_type="authheaders", auth_headers=auth_headers)

        assert gateway.auth_value is not None
        decoded = decode_auth(gateway.auth_value)
        assert decoded["X-API-Key"] == "secret123"
        assert decoded["X-Client-ID"] == "client456"
        assert decoded["X-Region"] == "us-east-1"

    @pytest.mark.asyncio
    async def test_gateway_create_with_empty_headers_list(self):
        """Test creating gateway with empty auth_headers list."""
        with pytest.raises(ValidationError) as exc_info:
            GatewayCreate(name="Test Gateway", url="http://example.com", auth_type="authheaders", auth_headers=[])

        assert "either 'auth_headers' list or both" in str(exc_info.value).lower()

    @pytest.mark.asyncio
    async def test_gateway_create_with_duplicate_header_keys(self):
        """Test handling of duplicate header keys (last value wins)."""
        auth_headers = [{"key": "X-API-Key", "value": "first_value"}, {"key": "X-API-Key", "value": "second_value"}, {"key": "X-Client-ID", "value": "client123"}]

        gateway = GatewayCreate(name="Test Gateway", url="http://example.com", auth_type="authheaders", auth_headers=auth_headers)

        decoded = decode_auth(gateway.auth_value)
        assert decoded["X-API-Key"] == "second_value"  # Last value should win
        assert decoded["X-Client-ID"] == "client123"

    @pytest.mark.asyncio
    async def test_gateway_create_with_empty_header_values(self):
        """Test creating gateway with empty header values."""
        auth_headers = [{"key": "X-API-Key", "value": ""}, {"key": "X-Client-ID", "value": "client123"}]

        gateway = GatewayCreate(name="Test Gateway", url="http://example.com", auth_type="authheaders", auth_headers=auth_headers)

        decoded = decode_auth(gateway.auth_value)
        assert decoded["X-API-Key"] == ""  # Empty values should be allowed
        assert decoded["X-Client-ID"] == "client123"

    @pytest.mark.asyncio
    async def test_gateway_create_with_missing_key_in_header(self):
        """Test creating gateway with missing key in header object."""
        auth_headers = [{"value": "secret123"}, {"key": "X-Client-ID", "value": "client123"}]  # Missing 'key' field

        gateway = GatewayCreate(name="Test Gateway", url="http://example.com", auth_type="authheaders", auth_headers=auth_headers)

        decoded = decode_auth(gateway.auth_value)
        assert "X-Client-ID" in decoded
        assert len(decoded) == 1  # Only valid header should be included

    @pytest.mark.asyncio
    async def test_backward_compatibility_single_headers(self):
        """Test backward compatibility with single header fields."""
        gateway = GatewayCreate(name="Test Gateway", url="http://example.com", auth_type="authheaders", auth_header_key="X-API-Key", auth_header_value="secret123")

        decoded = decode_auth(gateway.auth_value)
        assert decoded["X-API-Key"] == "secret123"

    @pytest.mark.asyncio
    async def test_multi_headers_priority_over_single(self):
        """Test that multi-headers take priority over single header fields."""
        auth_headers = [{"key": "X-Multi-Header", "value": "multi_value"}]

        gateway = GatewayCreate(
            name="Test Gateway",
            url="http://example.com",
            auth_type="authheaders",
            auth_headers=auth_headers,
            auth_header_key="X-Single-Header",  # Should be ignored
            auth_header_value="single_value",  # Should be ignored
        )

        decoded = decode_auth(gateway.auth_value)
        assert "X-Multi-Header" in decoded
        assert "X-Single-Header" not in decoded

    @pytest.mark.asyncio
    async def test_gateway_update_add_multi_headers(self):
        """Test updating gateway to add multi-headers."""
        auth_headers = [{"key": "X-New-Header", "value": "new_value"}]

        gateway = GatewayUpdate(auth_type="authheaders", auth_headers=auth_headers)

        assert gateway.auth_value is not None
        decoded = decode_auth(gateway.auth_value)
        assert decoded["X-New-Header"] == "new_value"

    @pytest.mark.asyncio
    async def test_special_characters_in_headers_rejected(self):
        """Test headers with invalid special characters are rejected."""
        auth_headers = [{"key": "X-Special-!@#", "value": "value-with-特殊字符"}, {"key": "Content-Type", "value": "application/json; charset=utf-8"}]

        with pytest.raises(ValidationError) as exc_info:
            GatewayCreate(name="Test Gateway", url="http://example.com", auth_type="authheaders", auth_headers=auth_headers)

        assert "Invalid header key format" in str(exc_info.value)
        assert "X-Special-!@#" in str(exc_info.value)

    @pytest.mark.asyncio
    async def test_valid_special_characters_in_values(self):
        """Test headers with special characters in values (allowed) but valid keys."""
        auth_headers = [{"key": "X-Special-Header", "value": "value-with-特殊字符"}, {"key": "Content-Type", "value": "application/json; charset=utf-8"}]

        gateway = GatewayCreate(name="Test Gateway", url="http://example.com", auth_type="authheaders", auth_headers=auth_headers)

        decoded = decode_auth(gateway.auth_value)
        assert decoded["X-Special-Header"] == "value-with-特殊字符"
        assert decoded["Content-Type"] == "application/json; charset=utf-8"

    @pytest.mark.asyncio
    async def test_case_sensitivity_preservation(self):
        """Test that header key case is preserved."""
        auth_headers = [{"key": "X-API-Key", "value": "value1"}, {"key": "x-api-key", "value": "value2"}, {"key": "X-Api-Key", "value": "value3"}]

        gateway = GatewayCreate(name="Test Gateway", url="http://example.com", auth_type="authheaders", auth_headers=auth_headers)

        decoded = decode_auth(gateway.auth_value)
        # All three variations should be preserved as separate keys
        assert len(decoded) == 3

    @pytest.mark.asyncio
    async def test_admin_endpoint_with_invalid_json(self):
        """Test admin endpoint handling of invalid JSON."""
        mock_db = MagicMock()
        mock_user = {"email": "test_user", "db": mock_db}

        form_data = FormData([("name", "Test Gateway"), ("url", "http://example.com"), ("auth_type", "authheaders"), ("auth_headers", "{invalid json}")])

        mock_request = MagicMock(spec=Request)
        mock_request.form = AsyncMock(return_value=form_data)

        with patch("mcpgateway.admin.gateway_service.register_gateway", AsyncMock()):
            response = await admin_add_gateway(mock_request, mock_db, user=mock_user)
            # Should handle invalid JSON gracefully
            assert response.status_code in [200, 422]

    @pytest.mark.asyncio
    async def test_large_number_of_headers(self):
        """Test handling of large number of headers."""
        auth_headers = [{"key": f"X-Header-{i}", "value": f"value-{i}"} for i in range(100)]

        gateway = GatewayCreate(name="Test Gateway", url="http://example.com", auth_type="authheaders", auth_headers=auth_headers)

        decoded = decode_auth(gateway.auth_value)
        assert len(decoded) == 100
        assert decoded["X-Header-50"] == "value-50"

    @pytest.mark.asyncio
    async def test_authorization_header_in_multi_headers(self):
        """Test including Authorization header in multi-headers."""
        auth_headers = [{"key": "Authorization", "value": "Bearer token123"}, {"key": "X-API-Key", "value": "secret"}]

        gateway = GatewayCreate(name="Test Gateway", url="http://example.com", auth_type="authheaders", auth_headers=auth_headers)

        decoded = decode_auth(gateway.auth_value)
        assert decoded["Authorization"] == "Bearer token123"
        assert decoded["X-API-Key"] == "secret"

    @pytest.mark.asyncio
    async def test_gateway_create_invalid_header_key_format(self):
        """Test creating gateway with invalid header key format."""
        auth_headers = [{"key": "Invalid@Key!", "value": "secret123"}]

        with pytest.raises(ValidationError) as exc_info:
            GatewayCreate(name="Test Gateway", url="http://example.com", auth_type="authheaders", auth_headers=auth_headers)

        assert "Invalid header key format" in str(exc_info.value)

    @pytest.mark.asyncio
    async def test_gateway_create_excessive_headers(self):
        """Test creating gateway with more than 100 headers."""
        auth_headers = [{"key": f"X-Header-{i}", "value": f"value-{i}"} for i in range(101)]

        with pytest.raises(ValidationError) as exc_info:
            GatewayCreate(name="Test Gateway", url="http://example.com", auth_type="authheaders", auth_headers=auth_headers)

        assert "Maximum of 100 headers allowed" in str(exc_info.value)

    @pytest.mark.asyncio
    async def test_gateway_create_duplicate_keys_with_warning(self, caplog):
        """Test creating gateway with duplicate header keys logs warning."""
        caplog.set_level(logging.WARNING, logger="mcpgateway.schemas")
        auth_headers = [
            {"key": "X-API-Key", "value": "first_value"},
            {"key": "X-API-Key", "value": "second_value"},  # Duplicate
            {"key": "X-Client-ID", "value": "client123"},
        ]

        gateway = GatewayCreate(name="Test Gateway", url="http://example.com", auth_type="authheaders", auth_headers=auth_headers)

        # Check that duplicate warning was logged
        assert "Duplicate header keys detected" in caplog.text
        assert "X-API-Key" in caplog.text

        # Check that last value wins
        decoded = decode_auth(gateway.auth_value)
        assert decoded["X-API-Key"] == "second_value"
        assert decoded["X-Client-ID"] == "client123"

    @pytest.mark.asyncio
    async def test_gateway_create_mixed_valid_invalid_keys(self):
        """Test creating gateway with mixed valid and invalid header keys."""
        auth_headers = [
            {"key": "Valid-Header", "value": "test123"},
            {"key": "Invalid@Key!", "value": "should_fail"},  # This should fail validation
        ]

        with pytest.raises(ValidationError) as exc_info:
            GatewayCreate(name="Test Gateway", url="http://example.com", auth_type="authheaders", auth_headers=auth_headers)

        assert "Invalid header key format" in str(exc_info.value)
        assert "Invalid@Key!" in str(exc_info.value)

    @pytest.mark.asyncio
    async def test_gateway_create_edge_case_header_keys(self):
        """Test creating gateway with edge case header keys."""
        # Test valid edge cases
        auth_headers = [
            {"key": "X-API-Key", "value": "test1"},  # Standard format
            {"key": "X_API_KEY", "value": "test2"},  # Underscores allowed
            {"key": "API-Key-123", "value": "test3"},  # Numbers and hyphens
            {"key": "UPPERCASE", "value": "test4"},  # Uppercase
            {"key": "lowercase", "value": "test5"},  # Lowercase
        ]

        gateway = GatewayCreate(name="Test Gateway", url="http://example.com", auth_type="authheaders", auth_headers=auth_headers)

        decoded = decode_auth(gateway.auth_value)
        assert len(decoded) == 5
        assert decoded["X-API-Key"] == "test1"
        assert decoded["X_API_KEY"] == "test2"
        assert decoded["API-Key-123"] == "test3"

    def test_gateway_read_includes_masked_auth_headers(self, monkeypatch):
        """Ensure GatewayRead surfaces auth_headers and masks values."""
        monkeypatch.setattr(settings, "auth_encryption_secret", "unit-test-secret")
        auth_map = {"X-API-Key": "secret123", "X-Trace": "trace-value"}
        gateway_read = GatewayRead(
            name="Masked Gateway",
            url="http://example.com",
            auth_type="authheaders",
            auth_value=encode_auth(auth_map),
        )

        assert gateway_read.auth_headers is not None
        assert {header["key"] for header in gateway_read.auth_headers} == set(auth_map.keys())
        assert gateway_read.auth_headers_unmasked == gateway_read.auth_headers

        masked = gateway_read.masked()
        assert masked.auth_headers is not None
        for header in masked.auth_headers:
            if header["value"]:
                assert header["value"] == settings.masked_auth_value
        # SECURITY: After masking, unmasked fields must be None to prevent credential leakage
        assert masked.auth_headers_unmasked is None

    @pytest.mark.asyncio
    async def test_gateway_update_preserves_masked_header_values(self, monkeypatch):
        """Confirm updating a gateway retains existing header secrets when masked."""
        monkeypatch.setattr(settings, "auth_encryption_secret", "unit-test-secret")

        service = GatewayService()
        existing_headers = {"X-API-Key": "secret123", "X-Trace": "trace-1"}

        gateway_db_obj = MagicMock()
        gateway_db_obj.id = "gateway-1"
        gateway_db_obj.name = "Gateway"
        gateway_db_obj.slug = "gateway"
        gateway_db_obj.enabled = True
        gateway_db_obj.visibility = "public"
        gateway_db_obj.transport = "SSE"
        gateway_db_obj.tags = []
        gateway_db_obj.auth_type = "authheaders"
        gateway_db_obj.auth_value = encode_auth(existing_headers)
        gateway_db_obj.url = "http://example.com"
        gateway_db_obj.tools = []
        gateway_db_obj.resources = []
        gateway_db_obj.prompts = []
        gateway_db_obj.capabilities = {}
        gateway_db_obj.last_seen = None
        gateway_db_obj.version = 1

        mock_db = MagicMock()
        # First execute call returns gateway (selectinload query), subsequent calls return None (conflict checks)
        mock_db.execute.return_value = MagicMock(scalar_one_or_none=MagicMock(return_value=gateway_db_obj))
        mock_db.add_all = MagicMock()
        mock_db.delete = MagicMock()
        mock_db.commit = MagicMock()
        mock_db.refresh = MagicMock()
        mock_db.query = MagicMock(return_value=MagicMock(filter=MagicMock(return_value=MagicMock(first=MagicMock(return_value=None)))))

        monkeypatch.setattr(service, "_initialize_gateway", AsyncMock(return_value=({}, [], [], [])))
        monkeypatch.setattr(service, "_update_or_create_tools", MagicMock(return_value=[]))
        monkeypatch.setattr(service, "_update_or_create_resources", MagicMock(return_value=[]))
        monkeypatch.setattr(service, "_update_or_create_prompts", MagicMock(return_value=[]))
        monkeypatch.setattr(service, "_notify_gateway_updated", AsyncMock())

        # Mock model_validate to return a mock that returns itself when masked() is called
        mock_gateway_read = MagicMock()
        mock_gateway_read.masked.return_value = mock_gateway_read
        monkeypatch.setattr(GatewayRead, "model_validate", staticmethod(lambda value: mock_gateway_read))

        gateway_update = GatewayUpdate(
            name="Gateway",
            url="http://example.com",
            auth_type="authheaders",
            auth_headers=[
                {"key": "X-API-Key", "value": settings.masked_auth_value},
                {"key": "X-Trace", "value": "updated-trace"},
            ],
        )

        result = await service.update_gateway(
            mock_db,
            "gateway-1",
            gateway_update,
            modified_by=None,
        )

        updated_auth = gateway_db_obj.auth_value
        if isinstance(updated_auth, str):
            updated_auth = decode_auth(updated_auth)

        assert updated_auth["X-API-Key"] == "secret123"
        assert updated_auth["X-Trace"] == "updated-trace"
        # Result is now the masked GatewayRead (via our mock)
        assert result is mock_gateway_read
        # SECURITY: Verify .masked() is called to prevent credential leakage
        mock_gateway_read.masked.assert_called_once()

    def test_gateway_read_unmasked_basic_and_bearer(self, monkeypatch):
        """Verify GatewayRead retains unmasked values for basic and bearer auth."""
        monkeypatch.setattr(settings, "auth_encryption_secret", "unit-test-secret")

        # Basic auth
        creds = base64.b64encode(b"user:secret-pass").decode("utf-8")
        basic_gateway = GatewayRead(
            name="Basic Gateway",
            url="http://example.com",
            auth_type="basic",
            auth_value=encode_auth({"Authorization": f"Basic {creds}"}),
        )
        assert basic_gateway.auth_password_unmasked == "secret-pass"
        masked_basic = basic_gateway.masked()
        assert masked_basic.auth_password == settings.masked_auth_value
        # SECURITY: After masking, unmasked fields must be None to prevent credential leakage
        assert masked_basic.auth_password_unmasked is None

        # Bearer auth
        bearer_gateway = GatewayRead(
            name="Bearer Gateway",
            url="http://example.com",
            auth_type="bearer",
            auth_value=encode_auth({"Authorization": "Bearer token-123"}),
        )
        assert bearer_gateway.auth_token_unmasked == "token-123"
        masked_bearer = bearer_gateway.masked()
        assert masked_bearer.auth_token == settings.masked_auth_value
        # SECURITY: After masking, unmasked fields must be None to prevent credential leakage
        assert masked_bearer.auth_token_unmasked is None
