"""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, 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
timeout: int = 75 # Connection timeout in seconds
@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).
"""
# OAuth with cloud_id uses api.atlassian.com which is always Cloud
if (
self.auth_type == "oauth"
and self.oauth_config
and self.oauth_config.cloud_id
):
return True
# DC OAuth has base_url but no cloud_id — not Cloud
if (
self.auth_type == "oauth"
and self.oauth_config
and hasattr(self.oauth_config, "base_url")
and self.oauth_config.base_url
and not self.oauth_config.cloud_id
):
return False
# 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. "
"Set CONFLUENCE_URL to your Confluence base URL, for example "
"https://your-company.atlassian.net/wiki"
)
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 (pass service info for DC detection)
oauth_config = get_oauth_config_from_env(
service_url=url, service_type="confluence"
)
auth_type = None
# Use the shared utility function directly
is_cloud = is_atlassian_cloud_url(url) if url else False
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:
missing_fields: list[str] = []
if not username:
missing_fields.append("CONFLUENCE_USERNAME")
if not api_token:
missing_fields.append("CONFLUENCE_API_TOKEN")
missing_fields_text = ", ".join(missing_fields)
error_msg = (
"Cloud authentication requires CONFLUENCE_USERNAME and "
"CONFLUENCE_API_TOKEN, or OAuth configuration "
"(set ATLASSIAN_OAUTH_ENABLE=true for user-provided tokens). "
"Confluence Cloud authentication is incomplete. Missing: "
f"{missing_fields_text}. "
"Set CONFLUENCE_USERNAME and CONFLUENCE_API_TOKEN, or "
"enable OAuth with ATLASSIAN_OAUTH_ENABLE=true."
)
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. "
"Confluence Server/Data Center authentication is incomplete. "
"Set CONFLUENCE_PERSONAL_TOKEN, or set both "
"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")
# Timeout setting
timeout = 75 # Default timeout
if (
os.getenv("CONFLUENCE_TIMEOUT")
and os.getenv("CONFLUENCE_TIMEOUT", "").isdigit()
):
timeout = int(os.getenv("CONFLUENCE_TIMEOUT", "75"))
return cls(
url=url or "",
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,
timeout=timeout,
)
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":
if self.oauth_config:
# Minimal OAuth (user-provided tokens mode)
if isinstance(self.oauth_config, OAuthConfig):
if (
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
# DC OAuth: needs client_id + client_secret (no cloud_id needed)
if hasattr(self.oauth_config, "is_data_center"):
if self.oauth_config.is_data_center:
return bool(
self.oauth_config.client_id
and self.oauth_config.client_secret
)
# Cloud OAuth: full set required
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
# BYO Access Token mode
elif isinstance(self.oauth_config, BYOAccessTokenOAuthConfig):
if self.oauth_config.access_token:
# DC BYO: access_token is enough
if hasattr(self.oauth_config, "is_data_center"):
if self.oauth_config.is_data_center:
return True
# Cloud BYO: needs cloud_id + access_token
if self.oauth_config.cloud_id:
return True
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
