MCP Atlassian

"""Configuration module for the Confluence client."""

import logging
import os
from dataclasses import dataclass
from typing import Literal

from ..utils.env import (
    get_custom_headers,
    get_env_csv,
    get_env_float,
    get_env_int,
    get_env_int_list,
    is_env_ssl_verify,
)
from ..utils.oauth import (
    BYOAccessTokenOAuthConfig,
    OAuthConfig,
    get_oauth_config_from_env,
)
from ..utils.urls import is_atlassian_cloud_url


@dataclass
class ConfluenceConfig:
    """Confluence API configuration.

    Handles authentication for Confluence Cloud and Server/Data Center:
    - Cloud: username/API token (basic auth) or OAuth 2.0 (3LO)
    - Server/DC: personal access token or basic auth
    """

    url: str  # Base URL for Confluence
    auth_type: Literal["basic", "pat", "oauth"]  # Authentication type
    username: str | None = None  # Email or username
    api_token: str | None = None  # API token used as password
    personal_token: str | None = None  # Personal access token (Server/DC)
    oauth_config: OAuthConfig | BYOAccessTokenOAuthConfig | None = None
    ssl_verify: bool = True  # Whether to verify SSL certificates
    spaces_filter: str | None = None  # List of space keys to filter searches
    http_proxy: str | None = None  # HTTP proxy URL
    https_proxy: str | None = None  # HTTPS proxy URL
    no_proxy: str | None = None  # Comma-separated list of hosts to bypass proxy
    socks_proxy: str | None = None  # SOCKS proxy URL (optional)
    custom_headers: dict[str, str] | None = None  # Custom HTTP headers
    client_cert: str | None = None  # Client certificate file path (.pem)
    client_key: str | None = None  # Client private key file path (.pem)
    client_key_password: str | None = None  # Password for encrypted private key
    http_timeout: float | None = None  # Total HTTP timeout (seconds)
    http_connect_timeout: float | None = None  # HTTP connect timeout (seconds)
    http_read_timeout: float | None = None  # HTTP read timeout (seconds)
    retry_max: int = 3  # Max retries for transient failures
    retry_backoff: float = 0.5  # Retry backoff factor
    retry_statuses: tuple[int, ...] = (429, 500, 502, 503, 504)
    retry_methods: tuple[str, ...] = ("HEAD", "GET", "PUT", "DELETE", "OPTIONS")
    http_pool_connections: int | None = None  # HTTP pool connections
    http_pool_maxsize: int | None = None  # HTTP pool maxsize

    @property
    def is_cloud(self) -> bool:
        """Check if this is a cloud instance.

        Returns:
            True if this is a cloud instance (atlassian.net), False otherwise.
            Localhost URLs are always considered non-cloud (Server/Data Center).
        """
        # Multi-Cloud OAuth mode: URL might be None, but we use api.atlassian.com
        if (
            self.auth_type == "oauth"
            and self.oauth_config
            and self.oauth_config.cloud_id
        ):
            # OAuth with cloud_id uses api.atlassian.com which is always Cloud
            return True

        # For other auth types, check the URL
        return is_atlassian_cloud_url(self.url) if self.url else False

    @property
    def verify_ssl(self) -> bool:
        """Compatibility property for old code.

        Returns:
            The ssl_verify value
        """
        return self.ssl_verify

    @classmethod
    def from_env(cls) -> "ConfluenceConfig":
        """Create configuration from environment variables.

        Returns:
            ConfluenceConfig with values from environment variables

        Raises:
            ValueError: If any required environment variable is missing
        """
        url = os.getenv("CONFLUENCE_URL")
        if not url and not os.getenv("ATLASSIAN_OAUTH_ENABLE"):
            error_msg = "Missing required CONFLUENCE_URL environment variable"
            raise ValueError(error_msg)

        # Determine authentication type based on available environment variables
        username = os.getenv("CONFLUENCE_USERNAME")
        api_token = os.getenv("CONFLUENCE_API_TOKEN")
        personal_token = os.getenv("CONFLUENCE_PERSONAL_TOKEN")

        # Check for OAuth configuration
        oauth_config = get_oauth_config_from_env()
        auth_type = None

        # Use the shared utility function directly
        is_cloud = is_atlassian_cloud_url(url)

        if is_cloud:
            # Cloud: OAuth takes priority, then basic auth
            if oauth_config:
                auth_type = "oauth"
            elif username and api_token:
                auth_type = "basic"
            else:
                error_msg = "Cloud authentication requires CONFLUENCE_USERNAME and CONFLUENCE_API_TOKEN, or OAuth configuration (set ATLASSIAN_OAUTH_ENABLE=true for user-provided tokens)"
                raise ValueError(error_msg)
        else:  # Server/Data Center
            # Server/DC: PAT takes priority over OAuth (fixes #824)
            if personal_token:
                if oauth_config:
                    logger = logging.getLogger("mcp-atlassian.confluence.config")
                    logger.warning(
                        "Both PAT and OAuth configured for Server/DC. Using PAT."
                    )
                auth_type = "pat"
            elif oauth_config:
                auth_type = "oauth"
            elif username and api_token:
                auth_type = "basic"
            else:
                error_msg = "Server/Data Center authentication requires CONFLUENCE_PERSONAL_TOKEN or CONFLUENCE_USERNAME and CONFLUENCE_API_TOKEN"
                raise ValueError(error_msg)

        # SSL verification (for Server/DC)
        ssl_verify = is_env_ssl_verify("CONFLUENCE_SSL_VERIFY")

        # Get the spaces filter if provided
        spaces_filter = os.getenv("CONFLUENCE_SPACES_FILTER")

        # Proxy settings
        http_proxy = os.getenv("CONFLUENCE_HTTP_PROXY", os.getenv("HTTP_PROXY"))
        https_proxy = os.getenv("CONFLUENCE_HTTPS_PROXY", os.getenv("HTTPS_PROXY"))
        no_proxy = os.getenv("CONFLUENCE_NO_PROXY", os.getenv("NO_PROXY"))
        socks_proxy = os.getenv("CONFLUENCE_SOCKS_PROXY", os.getenv("SOCKS_PROXY"))

        # Custom headers - service-specific only
        custom_headers = get_custom_headers("CONFLUENCE_CUSTOM_HEADERS")

        # Client certificate settings
        client_cert = os.getenv("CONFLUENCE_CLIENT_CERT")
        client_key = os.getenv("CONFLUENCE_CLIENT_KEY")
        client_key_password = os.getenv("CONFLUENCE_CLIENT_KEY_PASSWORD")

        http_timeout = get_env_float(
            "CONFLUENCE_HTTP_TIMEOUT", get_env_float("MCP_HTTP_TIMEOUT")
        )
        http_connect_timeout = get_env_float(
            "CONFLUENCE_HTTP_CONNECT_TIMEOUT",
            get_env_float("MCP_HTTP_CONNECT_TIMEOUT"),
        )
        http_read_timeout = get_env_float(
            "CONFLUENCE_HTTP_READ_TIMEOUT", get_env_float("MCP_HTTP_READ_TIMEOUT")
        )
        retry_max = get_env_int(
            "CONFLUENCE_RETRY_MAX", get_env_int("MCP_RETRY_MAX", 3), 0
        )
        retry_backoff = get_env_float(
            "CONFLUENCE_RETRY_BACKOFF", get_env_float("MCP_RETRY_BACKOFF", 0.5), 0.0
        )
        retry_statuses = tuple(
            get_env_int_list(
                "CONFLUENCE_RETRY_STATUS_CODES",
                get_env_int_list("MCP_RETRY_STATUS_CODES", [429, 500, 502, 503, 504]),
            )
        )
        retry_methods = tuple(
            get_env_csv(
                "CONFLUENCE_RETRY_METHODS",
                get_env_csv(
                    "MCP_RETRY_METHODS",
                    ["HEAD", "GET", "PUT", "DELETE", "OPTIONS"],
                ),
            )
        )
        http_pool_connections = get_env_int(
            "CONFLUENCE_HTTP_POOL_CONNECTIONS",
            get_env_int("MCP_HTTP_POOL_CONNECTIONS"),
            1,
        )
        http_pool_maxsize = get_env_int(
            "CONFLUENCE_HTTP_POOL_MAXSIZE",
            get_env_int("MCP_HTTP_POOL_MAXSIZE"),
            1,
        )

        return cls(
            url=url,
            auth_type=auth_type,
            username=username,
            api_token=api_token,
            personal_token=personal_token,
            oauth_config=oauth_config,
            ssl_verify=ssl_verify,
            spaces_filter=spaces_filter,
            http_proxy=http_proxy,
            https_proxy=https_proxy,
            no_proxy=no_proxy,
            socks_proxy=socks_proxy,
            custom_headers=custom_headers,
            client_cert=client_cert,
            client_key=client_key,
            client_key_password=client_key_password,
            http_timeout=http_timeout,
            http_connect_timeout=http_connect_timeout,
            http_read_timeout=http_read_timeout,
            retry_max=retry_max if retry_max is not None else 0,
            retry_backoff=retry_backoff if retry_backoff is not None else 0.0,
            retry_statuses=retry_statuses,
            retry_methods=retry_methods,
            http_pool_connections=http_pool_connections,
            http_pool_maxsize=http_pool_maxsize,
        )

    def is_auth_configured(self) -> bool:
        """Check if the current authentication configuration is complete and valid for making API calls.

        Returns:
            bool: True if authentication is fully configured, False otherwise.
        """
        logger = logging.getLogger("mcp-atlassian.confluence.config")
        if self.auth_type == "oauth":
            # Handle different OAuth configuration types
            if self.oauth_config:
                # Full OAuth configuration (traditional mode)
                if isinstance(self.oauth_config, OAuthConfig):
                    if (
                        self.oauth_config.client_id
                        and self.oauth_config.client_secret
                        and self.oauth_config.redirect_uri
                        and self.oauth_config.scope
                        and self.oauth_config.cloud_id
                    ):
                        return True
                    # Minimal OAuth configuration (user-provided tokens mode)
                    # This is valid if we have oauth_config but missing client credentials
                    # In this case, we expect authentication to come from user-provided headers
                    elif (
                        not self.oauth_config.client_id
                        and not self.oauth_config.client_secret
                    ):
                        logger.debug(
                            "Minimal OAuth config detected - expecting user-provided tokens via headers"
                        )
                        return True
                # Bring Your Own Access Token mode
                elif isinstance(self.oauth_config, BYOAccessTokenOAuthConfig):
                    if self.oauth_config.cloud_id and self.oauth_config.access_token:
                        return True

            # Partial configuration is invalid
            logger.warning("Incomplete OAuth configuration detected")
            return False
        elif self.auth_type == "pat":
            return bool(self.personal_token)
        elif self.auth_type == "basic":
            return bool(self.username and self.api_token)
        logger.warning(
            f"Unknown or unsupported auth_type: {self.auth_type} in ConfluenceConfig"
        )
        return False