diff --git a/nextcloud_mcp_server/api/passwords.py b/nextcloud_mcp_server/api/passwords.py index bf2651e6..212355b9 100644 --- a/nextcloud_mcp_server/api/passwords.py +++ b/nextcloud_mcp_server/api/passwords.py @@ -30,9 +30,11 @@ from ..http import nextcloud_httpx_client logger = logging.getLogger(__name__) # App password format regex (Nextcloud format: xxxxx-xxxxx-xxxxx-xxxxx-xxxxx) -APP_PASSWORD_PATTERN = re.compile( - r"^[a-zA-Z0-9]{5}-[a-zA-Z0-9]{5}-[a-zA-Z0-9]{5}-[a-zA-Z0-9]{5}-[a-zA-Z0-9]{5}$" -) +# Shape guard only — the authoritative check is the BasicAuth validation +# against Nextcloud below. Accepts both the dashed format a user copies from +# Security settings (xxxxx-xxxxx-xxxxx-xxxxx-xxxxx) and the raw token returned +# by the one-click ``core/getapppassword`` flow (a long alphanumeric string). +APP_PASSWORD_PATTERN = re.compile(r"^[a-zA-Z0-9-]{20,256}$") # Timeout for Nextcloud API validation requests (seconds) NEXTCLOUD_VALIDATION_TIMEOUT = 10.0 diff --git a/nextcloud_mcp_server/auth/login_flow.py b/nextcloud_mcp_server/auth/login_flow.py index 95dfd3e1..fbe422f0 100644 --- a/nextcloud_mcp_server/auth/login_flow.py +++ b/nextcloud_mcp_server/auth/login_flow.py @@ -68,17 +68,27 @@ class LoginFlowV2Client: 2. Poll for completion to receive the app password Args: - nextcloud_host: Base URL of the Nextcloud instance + nextcloud_host: Base URL of the Nextcloud instance, reachable by this + server (may be an internal/Docker hostname, e.g. http://app:80). verify_ssl: SSL verification setting (True, False, or SSLContext) + public_host: Externally-reachable Nextcloud base URL for the + browser-facing login URL (e.g. https://cloud.example.com). When the + server talks to Nextcloud over an internal hostname, Nextcloud + builds the login URL with that internal host — unusable in the + user's browser. If set, the login URL's origin is rewritten to this + public host. When None, the login URL is returned unchanged + (correct when nextcloud_host is already the public URL). """ def __init__( self, nextcloud_host: str, verify_ssl: bool | ssl.SSLContext = True, + public_host: str | None = None, ): self.nextcloud_host = nextcloud_host.rstrip("/") self.verify_ssl = verify_ssl + self.public_host = public_host.rstrip("/") if public_host else None async def initiate( self, user_agent: str = "nextcloud-mcp-server" @@ -119,8 +129,23 @@ class LoginFlowV2Client: # so server-side polling works across Docker networks. poll_endpoint = self._rewrite_to_nextcloud_host(raw_poll_endpoint) + # The login URL is opened in the *user's browser*, so it must use + # the externally-reachable host. Nextcloud builds it from the + # request host (our internal nextcloud_host), so rewrite it to the + # public host when one is configured (internal != external). + login_url = data["login"] + if self.public_host: + rewritten = rewrite_url_origin(login_url, self.public_host) + if rewritten != login_url: + logger.debug( + "Rewrote Login Flow v2 login_url to public host: %s → %s", + login_url, + rewritten, + ) + login_url = rewritten + result = LoginFlowInitResponse( - login_url=data["login"], + login_url=login_url, poll_endpoint=poll_endpoint, poll_token=poll_data["token"], ) diff --git a/nextcloud_mcp_server/auth/provision_routes.py b/nextcloud_mcp_server/auth/provision_routes.py index 651c196a..b83525f1 100644 --- a/nextcloud_mcp_server/auth/provision_routes.py +++ b/nextcloud_mcp_server/auth/provision_routes.py @@ -74,6 +74,7 @@ async def _poll_and_store(provision_id: str) -> None: flow_client = LoginFlowV2Client( nextcloud_host=nextcloud_host, verify_ssl=get_nextcloud_ssl_verify(), + public_host=settings.nextcloud_public_issuer_url, ) poll_endpoint = session["poll_endpoint"] @@ -205,6 +206,7 @@ async def provision_page( flow_client = LoginFlowV2Client( nextcloud_host=nextcloud_host, verify_ssl=get_nextcloud_ssl_verify(), + public_host=settings.nextcloud_public_issuer_url, ) init_response = await flow_client.initiate() except Exception as e: diff --git a/nextcloud_mcp_server/server/auth_tools.py b/nextcloud_mcp_server/server/auth_tools.py index 262e254f..8b5b22aa 100644 --- a/nextcloud_mcp_server/server/auth_tools.py +++ b/nextcloud_mcp_server/server/auth_tools.py @@ -113,6 +113,7 @@ def register_auth_tools(mcp: FastMCP) -> None: flow_client = LoginFlowV2Client( nextcloud_host=nextcloud_host, verify_ssl=get_nextcloud_ssl_verify(), + public_host=settings.nextcloud_public_issuer_url, ) init_response = await flow_client.initiate() except Exception as e: @@ -258,6 +259,7 @@ def register_auth_tools(mcp: FastMCP) -> None: flow_client = LoginFlowV2Client( nextcloud_host=nextcloud_host, verify_ssl=get_nextcloud_ssl_verify(), + public_host=settings.nextcloud_public_issuer_url, ) poll_result = await flow_client.poll( poll_endpoint=session["poll_endpoint"], @@ -431,6 +433,7 @@ def register_auth_tools(mcp: FastMCP) -> None: flow_client = LoginFlowV2Client( nextcloud_host=nextcloud_host, verify_ssl=get_nextcloud_ssl_verify(), + public_host=settings.nextcloud_public_issuer_url, ) init_response = await flow_client.initiate() except Exception as e: diff --git a/tests/unit/test_login_flow.py b/tests/unit/test_login_flow.py index a5dfc9bc..94524102 100644 --- a/tests/unit/test_login_flow.py +++ b/tests/unit/test_login_flow.py @@ -151,6 +151,43 @@ async def test_poll_expired(flow_client): assert result.app_password is None +async def test_initiate_rewrites_login_url_to_public_host(): + """When server↔Nextcloud uses an internal host, the browser-facing login + URL must be rewritten to the configured public host; the poll endpoint + stays on the internal host for server-side polling.""" + client = LoginFlowV2Client( + nextcloud_host="http://app:80", # internal Docker host + verify_ssl=False, + public_host="http://localhost:8080", # browser-reachable + ) + mock_response = _mock_response( + 200, + { + # Nextcloud builds these from the request (internal) host. + "login": "http://app/login/v2/flow/tok123", + "poll": { + "endpoint": "http://app/login/v2/poll", + "token": "secret-poll-token", + }, + }, + ) + mock_client = AsyncMock() + mock_client.post.return_value = mock_response + mock_client.__aenter__ = AsyncMock(return_value=mock_client) + mock_client.__aexit__ = AsyncMock(return_value=False) + + with patch( + "nextcloud_mcp_server.auth.login_flow.nextcloud_httpx_client", + return_value=mock_client, + ): + result = await client.initiate() + + # Browser-facing URL uses the public host... + assert result.login_url == "http://localhost:8080/login/v2/flow/tok123" + # ...while the poll endpoint stays on the internal host (server polls it). + assert result.poll_endpoint == "http://app:80/login/v2/poll" + + async def test_initiate_with_custom_user_agent(flow_client): """Test that custom user agent is passed in the request.""" mock_response = _mock_response( diff --git a/tests/unit/test_management_app_password_endpoints.py b/tests/unit/test_management_app_password_endpoints.py index d80c9671..2f0fd0f3 100644 --- a/tests/unit/test_management_app_password_endpoints.py +++ b/tests/unit/test_management_app_password_endpoints.py @@ -180,6 +180,25 @@ async def test_provision_app_password_invalid_format(): assert "Invalid app password format" in response.json()["error"] +def test_app_password_pattern_accepts_dashed_and_raw_tokens(): + """The format guard accepts both the dashed Security-settings format and + the raw token from the one-click ``core/getapppassword`` flow, and still + rejects short / illegal-character input.""" + from nextcloud_mcp_server.api.passwords import APP_PASSWORD_PATTERN + + # Dashed format a user copies from Security settings. + assert APP_PASSWORD_PATTERN.match("abcde-ABCDE-12345-fghij-67890") + # Raw 72-char token returned by core/getapppassword (one-click opt-in). + assert APP_PASSWORD_PATTERN.match( + "kZmgLDQnqQHUAxhRq4d2VssBfjsI0PaHbL4JySWtwJkzVgAf34c0sZshEjZjuj1PLbwrf83q" + ) + # Still rejects obviously-bad input. + assert not APP_PASSWORD_PATTERN.match("short") + assert not APP_PASSWORD_PATTERN.match("invalid-password") # < 20 chars + assert not APP_PASSWORD_PATTERN.match("has spaces not allowed in this token") + assert not APP_PASSWORD_PATTERN.match("contains/slash/" + "a" * 20) + + async def test_provision_app_password_success(temp_storage, mocker): """Test successful app password provisioning.""" # Mock settings (imported locally in the function)