feat(auth): elicit Astrolabe URL on missing app password

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) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-05-02 12:02:18 +02:00
co-authored by Claude Opus 4.7
parent cb2b2e82d6
commit 2da8b38aeb
4 changed files with 352 additions and 2 deletions
+109
View File
@@ -5,12 +5,21 @@ when the client supports it, or falling back to returning the URL in a message.
""" """
import logging import logging
import os
from mcp.server.fastmcp import Context from mcp.server.fastmcp import Context
from pydantic import BaseModel, Field from pydantic import BaseModel, Field
logger = logging.getLogger(__name__) 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): class LoginFlowConfirmation(BaseModel):
"""Schema for Login Flow v2 confirmation elicitation.""" """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( async def present_login_url(
ctx: Context, ctx: Context,
login_url: str, login_url: str,
@@ -86,3 +119,79 @@ async def present_login_url(
"falling back to message" "falling back to message"
) )
return "message_only" 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"
@@ -151,7 +151,16 @@ def require_scopes(*required_scopes: str):
stored_scopes = await _get_stored_scopes(user_id) stored_scopes = await _get_stored_scopes(user_id)
if stored_scopes is None: 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 = ( error_msg = (
f"Access denied to {func_name}: " f"Access denied to {func_name}: "
f"Nextcloud access not provisioned. " f"Nextcloud access not provisioned. "
+128
View File
@@ -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"
+105 -1
View File
@@ -4,13 +4,17 @@ Tests the third enforcement mode in scope_authorization.py that checks
application-level scopes stored alongside app passwords. 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 import pytest
from mcp.server.fastmcp import Context
from nextcloud_mcp_server.auth.scope_authorization import ( from nextcloud_mcp_server.auth.scope_authorization import (
ProvisioningRequiredError,
_get_stored_scopes, _get_stored_scopes,
_scope_cache, _scope_cache,
require_scopes,
) )
pytestmark = pytest.mark.unit pytestmark = pytest.mark.unit
@@ -91,3 +95,103 @@ async def test_get_stored_scopes_storage_error():
pytest.raises(RuntimeError, match="DB error"), pytest.raises(RuntimeError, match="DB error"),
): ):
await _get_stored_scopes("alice") 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()