nextcloud-mcp-server/nextcloud_mcp_server/auth/login_flow.py

"""Nextcloud Login Flow v2 HTTP client.

Implements the Nextcloud Login Flow v2 protocol for obtaining app passwords.
See: https://docs.nextcloud.com/server/latest/developer_manual/client_apis/LoginFlow/index.html#login-flow-v2

The flow has two steps:
1. Initiate: POST /index.php/login/v2 → returns login URL + poll endpoint/token
2. Poll: POST to poll endpoint with token → returns server URL, loginName, appPassword
"""

import logging
import ssl

from pydantic import BaseModel, Field

from nextcloud_mcp_server.http import nextcloud_httpx_client

logger = logging.getLogger(__name__)


class LoginFlowInitResponse(BaseModel):
    """Response from initiating Login Flow v2."""

    login_url: str = Field(description="URL to present to the user for browser login")
    poll_endpoint: str = Field(description="URL to poll for flow completion")
    poll_token: str = Field(description="Token to use when polling")


class LoginFlowPollResult(BaseModel):
    """Result of polling Login Flow v2."""

    status: str = Field(description="Flow status: 'pending', 'completed', or 'expired'")
    server: str | None = Field(None, description="Nextcloud server URL (on completion)")
    login_name: str | None = Field(
        None, description="Nextcloud login name (on completion)"
    )
    app_password: str | None = Field(
        None, description="Generated app password (on completion)"
    )


class LoginFlowV2Client:
    """HTTP client for Nextcloud Login Flow v2.

    This client handles the two-step Login Flow v2 process:
    1. Initiate a flow to get a login URL for the user
    2. Poll for completion to receive the app password

    Args:
        nextcloud_host: Base URL of the Nextcloud instance
        verify_ssl: SSL verification setting (True, False, or SSLContext)
    """

    def __init__(
        self,
        nextcloud_host: str,
        verify_ssl: bool | ssl.SSLContext = True,
    ):
        self.nextcloud_host = nextcloud_host.rstrip("/")
        self.verify_ssl = verify_ssl

    async def initiate(
        self, user_agent: str = "nextcloud-mcp-server"
    ) -> LoginFlowInitResponse:
        """Initiate Login Flow v2.

        Posts to /index.php/login/v2 to start a new login flow.

        Args:
            user_agent: User-Agent string for the app password name

        Returns:
            LoginFlowInitResponse with login URL and poll credentials

        Raises:
            httpx.HTTPStatusError: If the Nextcloud server returns an error
        """
        url = f"{self.nextcloud_host}/index.php/login/v2"

        async with nextcloud_httpx_client(
            verify=self.verify_ssl, timeout=15.0
        ) as client:
            response = await client.post(
                url,
                headers={"User-Agent": user_agent},
            )
            response.raise_for_status()
            data = response.json()

        poll_data = data.get("poll", {})

        result = LoginFlowInitResponse(
            login_url=data["login"],
            poll_endpoint=poll_data["endpoint"],
            poll_token=poll_data["token"],
        )

        logger.info(f"Login Flow v2 initiated: login_url={result.login_url[:60]}...")
        return result

    async def poll(self, poll_endpoint: str, poll_token: str) -> LoginFlowPollResult:
        """Poll for Login Flow v2 completion.

        Posts to the poll endpoint with the token. Nextcloud returns:
        - 200 with credentials when the user completes login
        - 404 when still pending
        - Other errors for expired/invalid flows

        Args:
            poll_endpoint: URL to poll (from initiate response)
            poll_token: Token for polling (from initiate response)

        Returns:
            LoginFlowPollResult with status and optional credentials
        """
        async with nextcloud_httpx_client(
            verify=self.verify_ssl, timeout=10.0
        ) as client:
            response = await client.post(
                poll_endpoint,
                data={"token": poll_token},
            )

        if response.status_code == 200:
            data = response.json()
            logger.info(
                f"Login Flow v2 completed: server={data.get('server')}, "
                f"loginName={data.get('loginName')}"
            )
            return LoginFlowPollResult(
                status="completed",
                server=data["server"],
                login_name=data["loginName"],
                app_password=data["appPassword"],
            )

        if response.status_code == 404:
            logger.debug("Login Flow v2 still pending")
            return LoginFlowPollResult(status="pending")

        # Any other status indicates the flow has expired or is invalid
        logger.warning(
            f"Login Flow v2 poll returned unexpected status: {response.status_code}"
        )
        return LoginFlowPollResult(status="expired")