From 2da8b38aeb383744d5f3c5d4ecfcaf2e3bc36e0b Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sat, 2 May 2026 12:02:18 +0200 Subject: [PATCH] feat(auth): elicit Astrolabe URL on missing app password MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a tool requiring Nextcloud access is called without a stored app password (Login Flow v2 mode), the @require_scopes decorator now invokes MCP elicitation with a clickable Astrolabe settings URL — reconstructed from NEXTCLOUD_PUBLIC_ISSUER_URL / NEXTCLOUD_HOST — before raising ProvisioningRequiredError. Clients without elicitation support fall back to the existing text error. Surfaced by cbcoutinho/nextcloud-mcp-server#752, where users hit a 401 after OAuth and had no clickable URL to start Login Flow v2 from. Co-Authored-By: Claude Opus 4.7 (1M context) --- nextcloud_mcp_server/auth/elicitation.py | 109 +++++++++++++++ .../auth/scope_authorization.py | 11 +- tests/unit/test_elicitation.py | 128 ++++++++++++++++++ tests/unit/test_scope_authorization_stored.py | 106 ++++++++++++++- 4 files changed, 352 insertions(+), 2 deletions(-) create mode 100644 tests/unit/test_elicitation.py diff --git a/nextcloud_mcp_server/auth/elicitation.py b/nextcloud_mcp_server/auth/elicitation.py index 871e827b..f8c0e1a8 100644 --- a/nextcloud_mcp_server/auth/elicitation.py +++ b/nextcloud_mcp_server/auth/elicitation.py @@ -5,12 +5,21 @@ when the client supports it, or falling back to returning the URL in a message. """ import logging +import os from mcp.server.fastmcp import Context from pydantic import BaseModel, Field logger = logging.getLogger(__name__) +# Path of the Astrolabe Nextcloud app's settings UI. The full URL is +# reconstructed at elicitation time from NEXTCLOUD_PUBLIC_ISSUER_URL / +# NEXTCLOUD_HOST so the user gets a browser-reachable link without needing a +# separate config knob. If the Astrolabe app is not installed this path will +# 404, and the user falls back to the nc_auth_provision_access tool path +# mentioned in the same message. +ASTROLABE_SETTINGS_PATH = "/index.php/apps/astrolabe/settings" + class LoginFlowConfirmation(BaseModel): """Schema for Login Flow v2 confirmation elicitation.""" @@ -21,6 +30,30 @@ class LoginFlowConfirmation(BaseModel): ) +class ProvisioningRequiredConfirmation(BaseModel): + """Schema for the 'app password not provisioned' elicitation.""" + + acknowledged: bool = Field( + default=False, + description="Check this box after enabling Nextcloud access", + ) + + +def _astrolabe_settings_url() -> str | None: + """Construct the Astrolabe settings page URL from environment. + + Prefers ``NEXTCLOUD_PUBLIC_ISSUER_URL`` (the browser-reachable public URL) + over ``NEXTCLOUD_HOST`` (which may be an internal hostname in Docker + deployments). Returns None if neither is set. + """ + base = ( + os.getenv("NEXTCLOUD_PUBLIC_ISSUER_URL") or os.getenv("NEXTCLOUD_HOST") or "" + ).strip() + if not base: + return None + return f"{base.rstrip('/')}{ASTROLABE_SETTINGS_PATH}" + + async def present_login_url( ctx: Context, login_url: str, @@ -86,3 +119,79 @@ async def present_login_url( "falling back to message" ) return "message_only" + + +async def present_provisioning_required(ctx: Context) -> str: + """Elicit a provisioning prompt when a tool is called without an app password. + + Used by the ``@require_scopes`` decorator (Login Flow v2 path) to give + the user a clickable Astrolabe settings URL — or a fallback instruction + to call the ``nc_auth_provision_access`` MCP tool — instead of just + raising a plain ``ProvisioningRequiredError`` text message that an LLM + has to translate. + + The Astrolabe settings URL is reconstructed from + ``NEXTCLOUD_PUBLIC_ISSUER_URL`` / ``NEXTCLOUD_HOST``; if Astrolabe is not + installed the link 404s and the user falls back to the tool path + suggested in the same message. + + Returns: + Same string contract as :func:`present_login_url`: + ``"accepted"`` / ``"declined"`` / ``"cancelled"`` / ``"message_only"``. + """ + settings_url = _astrolabe_settings_url() + + if settings_url: + message = ( + "Nextcloud access is not yet provisioned for this user.\n\n" + f"Open this URL to enable it via the Astrolabe app:\n\n{settings_url}\n\n" + "If the Astrolabe app is not installed, ask your MCP client to call " + "the `nc_auth_provision_access` tool instead — it will return a " + "Login Flow v2 URL you can open directly.\n\n" + "Then check the box below and retry the original request." + ) + else: + message = ( + "Nextcloud access is not yet provisioned for this user.\n\n" + "Ask your MCP client to call the `nc_auth_provision_access` tool — " + "it will return a Login Flow v2 URL you can open in your browser to " + "grant access.\n\n" + "Then check the box below and retry the original request." + ) + + if not hasattr(ctx, "elicit"): + logger.debug( + "Elicitation not available on context — returning message_only " + "(plain ProvisioningRequiredError will surface to the caller)" + ) + return "message_only" + + try: + result = await ctx.elicit( + message=message, + schema=ProvisioningRequiredConfirmation, + ) + + if result.action == "accept": + logger.info("User acknowledged provisioning-required prompt") + return "accepted" + elif result.action == "decline": + logger.info("User declined provisioning-required prompt") + return "declined" + else: + logger.info("User cancelled provisioning-required prompt") + return "cancelled" + + except NotImplementedError: + logger.debug( + "Elicitation not supported by client — falling back to plain error" + ) + return "message_only" + except Exception as e: + logger.warning( + "Provisioning elicitation failed unexpectedly (%s: %s), " + "falling back to plain error", + type(e).__name__, + e, + ) + return "message_only" diff --git a/nextcloud_mcp_server/auth/scope_authorization.py b/nextcloud_mcp_server/auth/scope_authorization.py index 5317d271..cacdbd29 100644 --- a/nextcloud_mcp_server/auth/scope_authorization.py +++ b/nextcloud_mcp_server/auth/scope_authorization.py @@ -151,7 +151,16 @@ def require_scopes(*required_scopes: str): stored_scopes = await _get_stored_scopes(user_id) if stored_scopes is None: - # No stored app password → require provisioning + # No stored app password → require provisioning. Try to + # elicit a clickable Astrolabe / Login-Flow-v2 link so + # the user has somewhere to click; the elicit helper + # silently falls back when the client lacks support. + from nextcloud_mcp_server.auth.elicitation import ( # noqa: PLC0415 + present_provisioning_required, + ) + + await present_provisioning_required(ctx) + error_msg = ( f"Access denied to {func_name}: " f"Nextcloud access not provisioned. " diff --git a/tests/unit/test_elicitation.py b/tests/unit/test_elicitation.py new file mode 100644 index 00000000..b7487de3 --- /dev/null +++ b/tests/unit/test_elicitation.py @@ -0,0 +1,128 @@ +"""Unit tests for the MCP elicitation helpers.""" + +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from nextcloud_mcp_server.auth.elicitation import ( + ASTROLABE_SETTINGS_PATH, + _astrolabe_settings_url, + present_provisioning_required, +) + +pytestmark = pytest.mark.unit + + +@pytest.fixture(autouse=True) +def clear_nc_env(monkeypatch): + """Strip the NC URL env vars by default; tests opt back in.""" + monkeypatch.delenv("NEXTCLOUD_PUBLIC_ISSUER_URL", raising=False) + monkeypatch.delenv("NEXTCLOUD_HOST", raising=False) + + +def test_astrolabe_settings_url_prefers_public_issuer(monkeypatch): + """Public issuer wins over host so the link is browser-reachable in Docker.""" + monkeypatch.setenv("NEXTCLOUD_PUBLIC_ISSUER_URL", "https://nc.example.com") + monkeypatch.setenv("NEXTCLOUD_HOST", "http://internal:8080") + + assert ( + _astrolabe_settings_url() == f"https://nc.example.com{ASTROLABE_SETTINGS_PATH}" + ) + + +def test_astrolabe_settings_url_falls_back_to_host(monkeypatch): + """When only NEXTCLOUD_HOST is set, use it (and strip a trailing slash).""" + monkeypatch.setenv("NEXTCLOUD_HOST", "https://only-host.example.com/") + + assert ( + _astrolabe_settings_url() + == f"https://only-host.example.com{ASTROLABE_SETTINGS_PATH}" + ) + + +def test_astrolabe_settings_url_returns_none_when_unset(): + """No NC URL configured → None (caller renders the tool-only message).""" + assert _astrolabe_settings_url() is None + + +async def test_present_provisioning_required_elicits_with_url(monkeypatch): + """When NC URL is set and the client supports elicitation, send the URL.""" + monkeypatch.setenv("NEXTCLOUD_PUBLIC_ISSUER_URL", "https://nc.example.com") + + ctx = MagicMock() + ctx.elicit = AsyncMock(return_value=SimpleNamespace(action="accept", data=None)) + + result = await present_provisioning_required(ctx) + + assert result == "accepted" + ctx.elicit.assert_awaited_once() + sent_message = ctx.elicit.await_args.kwargs["message"] + assert "https://nc.example.com/index.php/apps/astrolabe/settings" in sent_message + assert "nc_auth_provision_access" in sent_message + + +async def test_present_provisioning_required_without_url(monkeypatch): + """When neither NC URL is set, fall back to the tool-only message.""" + ctx = MagicMock() + ctx.elicit = AsyncMock(return_value=SimpleNamespace(action="accept", data=None)) + + result = await present_provisioning_required(ctx) + + assert result == "accepted" + sent_message = ctx.elicit.await_args.kwargs["message"] + assert "astrolabe" not in sent_message.lower() + assert "nc_auth_provision_access" in sent_message + + +async def test_present_provisioning_required_no_elicit_method(): + """Contexts that don't expose ctx.elicit fall back to message_only.""" + + class _NoElicit: + pass + + ctx = _NoElicit() + + result = await present_provisioning_required(ctx) # type: ignore[arg-type] + + assert result == "message_only" + + +async def test_present_provisioning_required_handles_not_implemented(): + """SDK clients that don't support elicitation raise NotImplementedError.""" + ctx = MagicMock() + ctx.elicit = AsyncMock(side_effect=NotImplementedError("client lacks elicit")) + + result = await present_provisioning_required(ctx) + + assert result == "message_only" + + +async def test_present_provisioning_required_handles_unexpected_error(): + """Any other elicit failure (e.g. transport) is fail-open to message_only.""" + ctx = MagicMock() + ctx.elicit = AsyncMock(side_effect=RuntimeError("transport boom")) + + result = await present_provisioning_required(ctx) + + assert result == "message_only" + + +async def test_present_provisioning_required_decline_returns_declined(): + """User chose 'decline' on the prompt → propagate that to the caller.""" + ctx = MagicMock() + ctx.elicit = AsyncMock(return_value=SimpleNamespace(action="decline", data=None)) + + result = await present_provisioning_required(ctx) + + assert result == "declined" + + +async def test_present_provisioning_required_cancel_returns_cancelled(): + """User chose 'cancel' on the prompt → propagate that to the caller.""" + ctx = MagicMock() + ctx.elicit = AsyncMock(return_value=SimpleNamespace(action="cancel", data=None)) + + result = await present_provisioning_required(ctx) + + assert result == "cancelled" diff --git a/tests/unit/test_scope_authorization_stored.py b/tests/unit/test_scope_authorization_stored.py index a7315e70..1a071d84 100644 --- a/tests/unit/test_scope_authorization_stored.py +++ b/tests/unit/test_scope_authorization_stored.py @@ -4,13 +4,17 @@ Tests the third enforcement mode in scope_authorization.py that checks application-level scopes stored alongside app passwords. """ -from unittest.mock import AsyncMock, patch +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch import pytest +from mcp.server.fastmcp import Context from nextcloud_mcp_server.auth.scope_authorization import ( + ProvisioningRequiredError, _get_stored_scopes, _scope_cache, + require_scopes, ) pytestmark = pytest.mark.unit @@ -91,3 +95,103 @@ async def test_get_stored_scopes_storage_error(): pytest.raises(RuntimeError, match="DB error"), ): await _get_stored_scopes("alice") + + +def _make_login_flow_ctx() -> MagicMock: + """Build a minimal Context shaped like the Login-Flow-v2 / OAuth case. + + request_context.access_token must be non-None to pass the BasicAuth-mode + short-circuit in require_scopes; the token's actual scopes don't matter + because the Login-Flow-v2 branch checks stored scopes instead. + """ + ctx = MagicMock() + ctx.request_context = SimpleNamespace( + access_token=SimpleNamespace(scopes=[], token="opaque") + ) + ctx.elicit = AsyncMock(return_value=SimpleNamespace(action="accept", data=None)) + return ctx + + +async def test_decorator_elicits_before_raising_when_app_password_missing(): + """When no app password is stored, the decorator must elicit a clickable + Astrolabe / Login-Flow-v2 prompt to the client *before* raising + ProvisioningRequiredError. + + Why: an LLM-only error message ("call nc_auth_provision_access") is + unfriendly to humans whose MCP client supports elicitation. See + cbcoutinho/nextcloud-mcp-server#752. + """ + ctx = _make_login_flow_ctx() + + @require_scopes("notes.read") + async def fake_tool_missing_pwd(ctx: Context): # noqa: ARG001 + return "ok" + + fake_settings = SimpleNamespace(enable_login_flow=True) + elicit_mock = AsyncMock(return_value="accepted") + + with ( + patch( + "nextcloud_mcp_server.auth.scope_authorization.get_settings", + return_value=fake_settings, + ), + patch( + "nextcloud_mcp_server.auth.scope_authorization._get_stored_scopes", + return_value=None, + ), + patch( + "nextcloud_mcp_server.auth.token_utils.extract_user_id_from_token", + return_value="alice", + ), + patch( + "nextcloud_mcp_server.auth.elicitation.present_provisioning_required", + elicit_mock, + ), + pytest.raises(ProvisioningRequiredError), + ): + await fake_tool_missing_pwd(ctx=ctx) + + elicit_mock.assert_awaited_once_with(ctx) + + +async def test_decorator_does_not_elicit_when_scopes_only_partially_missing(): + """When the user *has* an app password but is missing some requested + scopes, the decorator raises InsufficientScopeError (step-up auth), + not ProvisioningRequiredError — and must not elicit the + provisioning-required prompt, because the user is already provisioned. + """ + from nextcloud_mcp_server.auth.scope_authorization import ( + InsufficientScopeError, + ) + + ctx = _make_login_flow_ctx() + + @require_scopes("notes.write") + async def fake_tool_missing_scope(ctx: Context): # noqa: ARG001 + return "ok" + + fake_settings = SimpleNamespace(enable_login_flow=True) + elicit_mock = AsyncMock() + + with ( + patch( + "nextcloud_mcp_server.auth.scope_authorization.get_settings", + return_value=fake_settings, + ), + patch( + "nextcloud_mcp_server.auth.scope_authorization._get_stored_scopes", + return_value=["notes.read"], # has read, lacks write + ), + patch( + "nextcloud_mcp_server.auth.token_utils.extract_user_id_from_token", + return_value="alice", + ), + patch( + "nextcloud_mcp_server.auth.elicitation.present_provisioning_required", + elicit_mock, + ), + pytest.raises(InsufficientScopeError), + ): + await fake_tool_missing_scope(ctx=ctx) + + elicit_mock.assert_not_awaited()