nextcloud-mcp-server/nextcloud_mcp_server/context.py

"""Helper functions for accessing context in MCP tools."""

import logging

from httpx import BasicAuth
from mcp.server.fastmcp import Context

from nextcloud_mcp_server.auth.context_helper import (
    get_client_from_context,
    get_session_client_from_context,
)
from nextcloud_mcp_server.auth.scope_authorization import ProvisioningRequiredError
from nextcloud_mcp_server.auth.storage import get_shared_storage
from nextcloud_mcp_server.client import NextcloudClient
from nextcloud_mcp_server.config import (
    DeploymentMode,
    get_deployment_mode,
    get_settings,
)

logger = logging.getLogger(__name__)


async def get_client(ctx: Context) -> NextcloudClient:
    """
    Get the appropriate Nextcloud client based on authentication mode.

    ADR-016 compliant implementation supporting three deployment modes:

    1. Smithery stateless mode (SMITHERY_DEPLOYMENT=true):
       Create client from session configuration (nextcloud_url, username, app_password)
       No persistent state - client created per-request from Smithery session config.

    2. BasicAuth mode: Returns shared client from lifespan context

    3. OAuth mode:
       a. Multi-audience mode (ENABLE_TOKEN_EXCHANGE=false, default):
          Token already contains both MCP and Nextcloud audiences - use directly
       b. Token exchange mode (ENABLE_TOKEN_EXCHANGE=true):
          Exchange MCP token for Nextcloud token via RFC 8693

    SECURITY: Token passthrough has been REMOVED. All OAuth modes validate
    proper token audiences per MCP Security Best Practices specification.

    Note: Nextcloud doesn't support OAuth scopes natively. Scopes are enforced
    by the MCP server via @require_scopes decorator, not by the IdP.

    This function automatically detects the authentication mode by checking
    the deployment mode and type of the lifespan context.

    Args:
        ctx: MCP request context

    Returns:
        NextcloudClient configured for the current authentication mode

    Raises:
        AttributeError: If context doesn't contain expected data
        ValueError: If Smithery mode but session config is missing required fields

    Example:
        ```python
        @mcp.tool()
        async def my_tool(ctx: Context):
            client = await get_client(ctx)
            return await client.capabilities()
        ```
    """
    deployment_mode = get_deployment_mode()

    # ADR-016: Smithery stateless mode - create client from session config
    if deployment_mode == DeploymentMode.SMITHERY_STATELESS:
        return _get_client_from_session_config(ctx)

    settings = get_settings()

    # Multi-user BasicAuth pass-through mode - extract credentials from request
    if settings.enable_multi_user_basic_auth:
        return _get_client_from_basic_auth(ctx)

    lifespan_ctx = ctx.request_context.lifespan_context

    # Login Flow v2 multi-user mode: app password is REQUIRED for NC API access
    # OAuth token is only used for MCP session identity, not NC API calls
    if hasattr(lifespan_ctx, "nextcloud_host") and settings.enable_login_flow:
        return await _get_client_from_login_flow(ctx, lifespan_ctx.nextcloud_host)

    # BasicAuth mode - use shared client (no token exchange)
    if hasattr(lifespan_ctx, "client"):
        return lifespan_ctx.client

    # OAuth mode (has 'nextcloud_host' attribute)
    if hasattr(lifespan_ctx, "nextcloud_host"):
        if settings.enable_token_exchange:
            # Mode 2: Exchange MCP token for Nextcloud token
            # Token was validated to have MCP audience in UnifiedTokenVerifier
            # Now exchange it for Nextcloud audience
            return await get_session_client_from_context(
                ctx, lifespan_ctx.nextcloud_host
            )
        else:
            # Mode 1: Multi-audience token - use directly
            # Token was validated to have MCP audience in UnifiedTokenVerifier
            # Nextcloud will independently validate its own audience when receiving API calls
            return get_client_from_context(ctx, lifespan_ctx.nextcloud_host)

    # Unknown context type
    raise AttributeError(
        f"Lifespan context does not have 'client' or 'nextcloud_host' attribute. "
        f"Type: {type(lifespan_ctx)}"
    )


def _get_client_from_session_config(ctx: Context) -> NextcloudClient:
    """
    Create NextcloudClient from Smithery session configuration.

    ADR-016: In Smithery stateless mode, each request includes session config
    with the user's Nextcloud credentials. This function creates a fresh client
    for each request - no state is persisted between requests.

    For container runtime, config is extracted from URL query parameters by
    SmitheryConfigMiddleware and stored in a context variable.

    Expected session config fields (from Smithery configSchema):
    - nextcloud_url: str - Nextcloud instance URL (required)
    - username: str - Nextcloud username (required)
    - app_password: str - Nextcloud app password (required)

    Args:
        ctx: MCP request context (not used directly for Smithery config)

    Returns:
        NextcloudClient configured with session credentials

    Raises:
        ValueError: If required session config fields are missing
    """
    # ADR-016: Get session config from context variable (set by SmitheryConfigMiddleware)
    from nextcloud_mcp_server.app import get_smithery_session_config  # noqa: PLC0415

    session_config = get_smithery_session_config()

    if session_config is None:
        raise ValueError(
            "Session configuration required in Smithery mode. "
            "Ensure nextcloud_url, username, and app_password are provided as URL query parameters."
        )

    # Extract required fields - config is always a dict from SmitheryConfigMiddleware
    nextcloud_url = session_config.get("nextcloud_url")
    username = session_config.get("username")
    app_password = session_config.get("app_password")

    # Validate required fields
    missing_fields = []
    if not nextcloud_url:
        missing_fields.append("nextcloud_url")
    if not username:
        missing_fields.append("username")
    if not app_password:
        missing_fields.append("app_password")

    if missing_fields:
        raise ValueError(
            f"Missing required session config fields: {', '.join(missing_fields)}. "
            f"Configure these in the Smithery connection settings."
        )

    # Type assertions after validation (for type checker)
    # These are guaranteed to be str after the missing_fields check above
    assert nextcloud_url is not None
    assert username is not None
    assert app_password is not None

    # Validate URL format
    if not nextcloud_url.startswith(("http://", "https://")):
        raise ValueError(
            f"Invalid nextcloud_url: {nextcloud_url}. "
            f"Must start with http:// or https://"
        )

    logger.debug(f"Creating Smithery client for {nextcloud_url} as {username}")

    # Create client with session credentials using BasicAuth
    return NextcloudClient(
        base_url=nextcloud_url,
        username=username,
        auth=BasicAuth(username, app_password),
    )


def _get_client_from_basic_auth(ctx: Context) -> NextcloudClient:
    """
    Create NextcloudClient from BasicAuth credentials in request headers.

    For multi-user BasicAuth pass-through mode, this function extracts
    username/password from the Authorization: Basic header (stored by
    BasicAuthMiddleware) and creates a client that passes these credentials
    through to Nextcloud APIs.

    The credentials are NOT stored persistently - they exist only for the
    duration of this request (stateless).

    Args:
        ctx: MCP request context with basic_auth in request state

    Returns:
        NextcloudClient configured with BasicAuth credentials

    Raises:
        ValueError: If BasicAuth credentials not found in request or if
                   NEXTCLOUD_HOST is not configured
    """
    settings = get_settings()

    # Validate that NEXTCLOUD_HOST is configured
    if not settings.nextcloud_host:
        raise ValueError(
            "NEXTCLOUD_HOST environment variable must be set for multi-user BasicAuth mode"
        )

    # Extract BasicAuth credentials from request state (set by BasicAuthMiddleware)
    # Access scope through the request object
    scope = getattr(ctx.request_context.request, "scope", None)
    if scope is None:
        raise ValueError("Request scope not available in context")

    request_state = scope.get("state", {})
    basic_auth = request_state.get("basic_auth")

    if not basic_auth:
        raise ValueError(
            "BasicAuth credentials not found in request. "
            "Ensure Authorization: Basic header is provided with valid credentials."
        )

    username = basic_auth.get("username")
    password = basic_auth.get("password")

    if not username or not password:
        raise ValueError("Invalid BasicAuth credentials - missing username or password")

    logger.debug(
        f"Creating multi-user BasicAuth client for {settings.nextcloud_host} as {username}"
    )

    # Create client that passes BasicAuth credentials through to Nextcloud
    # settings.nextcloud_host is guaranteed to be str after the check above
    return NextcloudClient(
        base_url=settings.nextcloud_host,
        username=username,
        auth=BasicAuth(username, password),
    )


async def _get_client_from_login_flow(
    ctx: Context, nextcloud_host: str
) -> NextcloudClient:
    """Create NextcloudClient from stored Login Flow v2 app password.

    In Login Flow v2 mode, the OAuth token only provides MCP session identity.
    Nextcloud API calls always use the stored app password obtained via Login Flow v2.

    Args:
        ctx: MCP context (used to extract user identity)
        nextcloud_host: Nextcloud instance URL

    Returns:
        NextcloudClient with stored app password credentials

    Raises:
        ProvisioningRequiredError: If no stored app password exists
    """
    from nextcloud_mcp_server.server.oauth_tools import (  # noqa: PLC0415
        extract_user_id_from_token,
    )

    user_id = await extract_user_id_from_token(ctx)
    if not user_id or user_id == "default_user":
        raise ProvisioningRequiredError(
            "Cannot determine user identity from MCP token."
        )

    storage = await get_shared_storage()

    app_data = await storage.get_app_password_with_scopes(user_id)
    if not app_data:
        raise ProvisioningRequiredError(
            "Nextcloud access not provisioned. "
            "Call nc_auth_provision_access to complete Login Flow."
        )

    username = app_data.get("username") or user_id

    logger.debug(f"Creating Login Flow v2 client for {nextcloud_host} as {username}")

    return NextcloudClient(
        base_url=nextcloud_host,
        username=username,
        auth=BasicAuth(username, app_data["app_password"]),
    )