Merge pull request #757 from cbcoutinho/feat/elicit-on-missing-app-password

feat(auth): elicit Astrolabe URL on missing app password + external-IdP docs
This commit is contained in:
Chris Coutinho
2026-05-02 17:52:33 +02:00
committed by GitHub
13 changed files with 738 additions and 65 deletions
+10 -4
View File
@@ -38,7 +38,7 @@ judgment.
**Skip when:**
- Tiny diffs (typo fix, README tweak, dependency bump only).
- User has explicitly said "just push it" / "skip the review".
- Branch is `master` or has zero commits ahead of base.
- Branch is `master`, or has zero commits ahead of base **and** a clean working tree (`git status --short` empty).
## Workflow
@@ -50,10 +50,16 @@ Determine the base branch and diff range. Default base is `master`.
git fetch origin master --quiet
BASE=$(git merge-base HEAD origin/master)
git rev-list --count $BASE..HEAD # commits ahead
git diff --stat $BASE..HEAD # files touched
git diff --stat $BASE # files touched (incl. staged + unstaged)
git log --format="%h %s" $BASE..HEAD # commit list
git status --short # surface uncommitted state
```
**Diff scope:** the review uses `git diff $BASE` (base → working tree), which includes
committed + staged + unstaged changes. This means in-progress work is reviewed too —
flag any findings against half-written code as such, and don't penalize obvious WIP
(missing tests, TODO stubs) the user clearly hasn't finished yet.
If the user names a different base (e.g. `main`, a stacked branch), use that instead.
**Identify the change shape** — these classifications drive scope-aware checks (Phase 3):
@@ -95,8 +101,8 @@ broken build — fixing the failures may eliminate findings or change the diff.
### Phase 3 — Read the diff and run the project checklist (25min)
```bash
git diff $BASE..HEAD # full diff
git diff $BASE..HEAD -- '*.py' | head -2000 # python only, capped
git diff $BASE # full diff (incl. uncommitted)
git diff $BASE -- '*.py' | head -2000 # python only, capped
```
Read the **whole diff** before composing findings. Cross-file patterns (test symmetry,
+69
View File
@@ -72,6 +72,75 @@ NEXTCLOUD_PUBLIC_ISSUER_URL=https://your.nextcloud.example.com # Public URL of
When using an external IdP (Keycloak, Cognito, etc.), see [Keycloak Multi-Client Token Validation](keycloak-multi-client-validation.md) for how Nextcloud's `user_oidc` app handles realm-level token validation if you also federate Nextcloud's own login through the same IdP.
### External IdP setup (Authentik / Keycloak / Cognito)
When `OIDC_DISCOVERY_URL` points at a third-party IdP rather than Nextcloud's own
OIDC, three things have to line up — and most setup confusion (e.g.
[#752](https://github.com/cbcoutinho/nextcloud-mcp-server/issues/752)) comes from
mismatches across them.
#### Nextcloud apps to install
| App | When to install | Notes |
|---|---|---|
| `user_oidc` | **Required** if your external IdP also issues identities used by Nextcloud (i.e. you want SSO into Nextcloud through the same IdP). | Validates incoming Bearer tokens at the **realm** level — see [keycloak-multi-client-validation.md](keycloak-multi-client-validation.md). |
| `oidc` (Nextcloud-as-IdP) | **Skip.** | Only relevant when Nextcloud itself is the IdP. With an external IdP, `OIDC_DISCOVERY_URL` already points elsewhere. |
| `astrolabe` | **Optional.** | Provides a per-user "Enable Semantic Search" settings page that triggers Login Flow v2 from the Nextcloud UI. Without it, users provision via the `nc_auth_provision_access` MCP tool (which uses MCP elicitation for clients that support it). |
#### OIDC clients to register in your IdP
| Client | Required? | What it represents |
|---|---|---|
| **MCP server** | Yes | The MCP server's RP relationship with the IdP. Configured via `NEXTCLOUD_OIDC_CLIENT_ID` / `NEXTCLOUD_OIDC_CLIENT_SECRET`. Used for OIDC discovery, JWKS retrieval, and token validation. |
| **Astrolabe** | Only if Astrolabe is installed | Used by the Astrolabe Nextcloud app for its own per-user OAuth flow against the MCP server. |
| **MCP client** (e.g. Claude.ai, Claude Code) | Optional | The MCP server supports RFC 7591 Dynamic Client Registration, so MCP clients are auto-registered on first connect. Only register a static client if your IdP rejects DCR-issued clients or your MCP client cannot do DCR (see [#752 thread](https://github.com/cbcoutinho/nextcloud-mcp-server/issues/752#issuecomment-4362197279) for the Claude Code workarounds). When pre-allowlisting static MCP clients, set `ALLOWED_MCP_CLIENTS` — see [`auth/client_registry.py`](../nextcloud_mcp_server/auth/client_registry.py) for the format. |
#### Scopes the IdP must advertise on the MCP-server client
Standard OIDC scopes (`openid`, `profile`, `email`) are not enough on their own —
the MCP server gates every Nextcloud-touching tool on a per-app scope (e.g.
`notes.read`, `calendar.write`). Those scopes must be issuable by the IdP on the
MCP-server client, otherwise the client's tokens won't carry them and tool calls
will be filtered out at `list_tools` time.
The authoritative list is served at:
```
GET https://<your-mcp-server>/.well-known/oauth-protected-resource/mcp
```
…in the `scopes_supported` field. In Authentik and Keycloak, register each scope
as a custom scope and expose it as a claim/scope mapping on the MCP-server
client. Add `offline_access` if you want refresh tokens for background sync.
If you also want resource-prefixed scopes (e.g. AWS Cognito's
`https://mcp.example.com/notes.read`), set `OIDC_RESOURCE_SERVER_ID` so the MCP
server strips the prefix before matching against `@require_scopes` decorators.
See `_strip_resource_prefix` in [`scope_authorization.py`](../nextcloud_mcp_server/auth/scope_authorization.py).
#### Diagnosing "OAuth succeeded but Nextcloud returns 401"
This is the most common failure mode after wiring up an external IdP, and it
trips up first-time setups. The two legs are independent:
```
MCP client ──── OAuth/OIDC ────> MCP server ──── Basic Auth ────> Nextcloud
(auth leg, validated) (data leg, app password)
```
If a tool call returns `401 Unauthorized` from a Nextcloud URL after OAuth
succeeded, the **data leg** has no credentials yet — the user hasn't completed
Login Flow v2 to provision an app password for that account. Fix it by calling
`nc_auth_provision_access` from the MCP client (or visiting the Astrolabe
settings page if installed). Subsequent tool calls reuse the stored app
password.
When the MCP client supports MCP elicitation (spec 2025-11-25), the server now
elicits a clickable Astrolabe settings URL automatically on the first failing
tool call, so the user has somewhere to click instead of just an error string.
Clients without elicitation support fall back to the existing
`ProvisioningRequiredError` text message.
### Generating an Encryption Key
App passwords are stored encrypted with Fernet. Generate a key once and reuse it:
+2
View File
@@ -21,6 +21,7 @@ from starlette.requests import Request
from starlette.responses import JSONResponse
from nextcloud_mcp_server.api.management import _sanitize_error_for_client
from nextcloud_mcp_server.auth.scope_authorization import invalidate_scope_cache
from nextcloud_mcp_server.auth.storage import RefreshTokenStorage
from nextcloud_mcp_server.config import get_settings
@@ -305,6 +306,7 @@ async def provision_app_password(request: Request) -> JSONResponse:
await storage.store_app_password_with_scopes(
username, app_password, scopes=scopes, username=nc_username
)
invalidate_scope_cache(username)
_record_rate_limit_attempt(path_user_id, success=True)
logger.info(f"Provisioned app password for user: {username}")
+2 -2
View File
@@ -743,7 +743,7 @@ async def setup_oauth_config():
# ADR-005: Unified Token Verifier with proper audience validation
# Use public issuer URL for JWT validation if set (handles Docker internal/external URL mismatch)
# Tokens are issued with the public URL, but OIDC discovery returns internal URL
public_issuer_url = os.getenv("NEXTCLOUD_PUBLIC_ISSUER_URL")
public_issuer_url = settings.nextcloud_public_issuer_url
client_issuer = public_issuer_url if public_issuer_url else issuer
# Get MCP server URL for audience validation
mcp_server_url = os.getenv("NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000")
@@ -933,7 +933,7 @@ async def setup_oauth_config_for_multi_user_basic(
# Use public issuer URL for JWT validation if set (handles Docker internal/external URL mismatch)
# Tokens are issued with the public URL, but OIDC discovery returns internal URL
public_issuer_url = os.getenv("NEXTCLOUD_PUBLIC_ISSUER_URL")
public_issuer_url = settings.nextcloud_public_issuer_url
client_issuer = public_issuer_url if public_issuer_url else issuer
# Update settings with discovered values for UnifiedTokenVerifier
@@ -6,7 +6,6 @@ for accessing admin UI endpoints like /app.
import hashlib
import logging
import os
import secrets
import time
from base64 import urlsafe_b64encode
@@ -22,6 +21,7 @@ from nextcloud_mcp_server.auth.userinfo_routes import (
_get_userinfo_endpoint,
_query_idp_userinfo,
)
from nextcloud_mcp_server.config import get_settings
from ..http import nextcloud_httpx_client
@@ -29,22 +29,21 @@ logger = logging.getLogger(__name__)
def _should_use_secure_cookies() -> bool:
"""Determine if cookies should have secure flag.
"""Determine if cookies should have the Secure flag.
Checks COOKIE_SECURE env var first, then auto-detects from NEXTCLOUD_HOST.
Reads ``settings.cookie_secure`` first (set via the ``COOKIE_SECURE``
env var). Falls back to auto-detect from the ``nextcloud_host`` scheme
when unset.
Returns:
True if cookies should be secure (HTTPS), False otherwise
"""
# Explicit configuration takes precedence
explicit = os.getenv("COOKIE_SECURE", "").lower()
if explicit == "true":
return True
if explicit == "false":
return False
# Auto-detect from NEXTCLOUD_HOST protocol
nextcloud_host = os.getenv("NEXTCLOUD_HOST", "")
settings = get_settings()
if settings.cookie_secure is not None:
# Dynaconf auto-coerces "true"/"false" → bool but "1"/"0" → int;
# bool() normalises both.
return bool(settings.cookie_secure)
nextcloud_host = settings.nextcloud_host or ""
return nextcloud_host.startswith("https://")
@@ -167,7 +166,7 @@ async def oauth_login(request: Request) -> RedirectResponse | JSONResponse:
scopes += " offline_access"
# Replace internal Docker hostname with public URL
public_issuer = os.getenv("NEXTCLOUD_PUBLIC_ISSUER_URL")
public_issuer = get_settings().nextcloud_public_issuer_url
if public_issuer:
internal_parsed = parse_url(oauth_config["nextcloud_host"])
auth_parsed = parse_url(authorization_endpoint)
+166 -34
View File
@@ -5,12 +5,23 @@ when the client supports it, or falling back to returning the URL in a message.
"""
import logging
from typing import Any
from mcp.server.fastmcp import Context
from pydantic import BaseModel, Field
from nextcloud_mcp_server.config import get_settings
logger = logging.getLogger(__name__)
# Path of the Astrolabe Nextcloud app's settings UI. The full URL is
# reconstructed at elicitation time from settings.nextcloud_public_issuer_url
# / settings.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 +32,95 @@ 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 settings.
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 (or set to the empty
string), or if the configured base URL is missing an http:// or
https:// scheme — in the latter case the caller renders the tool-only
fallback message instead of a broken link.
"""
settings = get_settings()
base = (
settings.nextcloud_public_issuer_url or settings.nextcloud_host or ""
).strip()
if not base:
return None
if not base.startswith(("http://", "https://")):
# Bare hostname (e.g. "internal:8080") would silently produce a
# non-clickable URL. Surface the misconfiguration instead.
logger.warning(
"Cannot build Astrolabe settings URL: configured Nextcloud base URL "
"%r is missing an http:// or https:// scheme. Falling back to the "
"tool-only provisioning message.",
base,
)
return None
return f"{base.rstrip('/')}{ASTROLABE_SETTINGS_PATH}"
async def _run_elicit(
ctx: Context,
message: str,
schema: type[BaseModel],
*,
log_label: str,
) -> tuple[str, Any]:
"""Shared elicit-or-fallback flow used by all elicitation prompts.
Returns ``(outcome, result)`` where ``outcome`` is one of
``"accepted"`` / ``"declined"`` / ``"cancelled"`` / ``"message_only"``.
``result`` is the underlying ``ctx.elicit()`` return value when the
elicitation actually ran (any of the first three outcomes), else None.
Callers needing post-accept inspection (e.g. the data-acknowledged
warning in :func:`present_login_url`) read it from ``result``.
"""
if not hasattr(ctx, "elicit"):
logger.debug(
"Elicitation not available on context — message_only fallback (%s)",
log_label,
)
return "message_only", None
try:
result = await ctx.elicit(message=message, schema=schema)
except NotImplementedError:
logger.debug(
"Elicitation not supported by client — message_only fallback (%s)",
log_label,
)
return "message_only", None
except Exception as e:
logger.warning(
"Elicitation failed unexpectedly for %s (%s: %s), "
"falling back to message_only",
log_label,
type(e).__name__,
e,
)
return "message_only", None
if result.action == "accept":
logger.info("User acknowledged %s", log_label)
return "accepted", result
if result.action == "decline":
logger.info("User declined %s", log_label)
return "declined", result
logger.info("User cancelled %s", log_label)
return "cancelled", result
async def present_login_url(
ctx: Context,
login_url: str,
@@ -49,40 +149,72 @@ async def present_login_url(
f"Then check the box below and click OK."
)
if not hasattr(ctx, "elicit"):
logger.debug(
"Elicitation not available (no elicit method), returning URL in message"
)
return "message_only"
outcome, result = await _run_elicit(
ctx,
message,
LoginFlowConfirmation,
log_label="login flow completion",
)
try:
result = await ctx.elicit(
message=message,
schema=LoginFlowConfirmation,
)
if result.action == "accept":
if hasattr(result, "data") and not result.data.acknowledged: # type: ignore[union-attr]
logger.warning(
"User accepted login flow without checking the acknowledged box — "
"login completion will be verified via polling"
)
logger.info("User acknowledged login flow completion")
return "accepted"
elif result.action == "decline":
logger.info("User declined login flow")
return "declined"
else:
logger.info("User cancelled login flow")
return "cancelled"
except NotImplementedError:
# Elicitation not supported by this client/SDK - fall back to message
logger.debug("Elicitation not available, returning URL in message")
return "message_only"
except Exception as e:
if (
outcome == "accepted"
and result is not None
and hasattr(result, "data")
and not result.data.acknowledged
):
# User clicked OK without ticking the box — login completion is still
# verified via the LFv2 poller, so we proceed but flag it.
logger.warning(
f"Elicitation failed unexpectedly ({type(e).__name__}: {e}), "
"falling back to message"
"User accepted login flow without checking the acknowledged box — "
"login completion will be verified via polling"
)
return "message_only"
return outcome
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
``settings.nextcloud_public_issuer_url`` /
``settings.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 in your browser.\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."
)
outcome, _ = await _run_elicit(
ctx,
message,
ProvisioningRequiredConfirmation,
log_label="provisioning-required prompt",
)
return outcome
+2 -2
View File
@@ -360,7 +360,7 @@ async def oauth_authorize(request: Request) -> RedirectResponse | JSONResponse:
authorization_endpoint = discovery["authorization_endpoint"]
# Replace internal Docker hostname with public URL for browser access
public_issuer = os.getenv("NEXTCLOUD_PUBLIC_ISSUER_URL")
public_issuer = get_settings().nextcloud_public_issuer_url
if public_issuer:
internal_parsed = parse_url(oauth_config["nextcloud_host"])
auth_parsed = parse_url(authorization_endpoint)
@@ -507,7 +507,7 @@ async def oauth_authorize_nextcloud(
authorization_endpoint = discovery["authorization_endpoint"]
# Fix internal hostname for browser access
public_issuer = os.getenv("NEXTCLOUD_PUBLIC_ISSUER_URL")
public_issuer = get_settings().nextcloud_public_issuer_url
if public_issuer:
internal_parsed = parse_url(oauth_config["nextcloud_host"])
auth_parsed = parse_url(authorization_endpoint)
@@ -15,7 +15,6 @@ Flow:
import html
import logging
import os
import secrets
import time
from urllib.parse import urlparse
@@ -26,6 +25,7 @@ from starlette.responses import HTMLResponse, JSONResponse, RedirectResponse
from nextcloud_mcp_server.api.management import validate_token_and_get_user
from nextcloud_mcp_server.auth.login_flow import LoginFlowV2Client, rewrite_url_origin
from nextcloud_mcp_server.auth.scope_authorization import invalidate_scope_cache
from nextcloud_mcp_server.auth.storage import get_shared_storage
from nextcloud_mcp_server.config import get_nextcloud_ssl_verify, get_settings
@@ -115,6 +115,7 @@ async def _poll_and_store(provision_id: str) -> None:
scopes=None, # All scopes
username=result.login_name,
)
invalidate_scope_cache(effective_user_id)
session = _provision_sessions.get(provision_id)
if session:
session["status"] = "completed"
@@ -251,7 +252,7 @@ async def provision_page(
# LoginFlowV2Client) while login_url is rewritten to the public issuer
# URL here because the browser needs a publicly-reachable address.
login_url = init_response.login_url
public_issuer = os.getenv("NEXTCLOUD_PUBLIC_ISSUER_URL", "")
public_issuer = settings.nextcloud_public_issuer_url or ""
if public_issuer and nextcloud_host:
login_url = rewrite_url_origin(login_url, public_issuer.rstrip("/"))
@@ -151,13 +151,60 @@ 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
error_msg = (
f"Access denied to {func_name}: "
f"Nextcloud access not provisioned. "
f"Please call 'nc_auth_provision_access' first."
# 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,
)
logger.warning(error_msg)
elicit_result = await present_provisioning_required(ctx)
# Always raise — the decorator can't safely re-check
# stored scopes mid-call (TTL cache, plus the LFv2
# poller may still be running). Only the message
# changes so an LLM that just acknowledged the
# elicitation isn't told to call the auth tool
# again (which would loop).
if elicit_result == "accepted":
# Note: stored-scope lookups are cached for
# _SCOPE_CACHE_TTL (5 min). All three provisioning
# paths invalidate the cache on completion: the
# in-tool poller in nc_auth_check_status
# (auth_tools.py), the Astrolabe web route
# (provision_routes.py), and the BasicAuth REST
# endpoint (api/passwords.py). However, if the
# LFv2 poller is still in-flight at acknowledge-
# time the next retry can still hit a not-yet-
# populated entry — hence the "wait a moment"
# qualifier below.
logger.warning(
"Access denied to %s: app password missing "
"after user accepted elicitation; advising retry",
func_name,
)
error_msg = (
f"Access denied to {func_name}: Nextcloud "
f"access was not provisioned at the time of "
f"this call. If you just completed "
f"provisioning, please retry the request — "
f"if it still fails, provisioning may still be "
f"completing; wait a moment and try again."
)
else:
logger.warning(
"Access denied to %s: app password missing; "
"advising nc_auth_provision_access "
"(elicit_result=%s)",
func_name,
elicit_result,
)
error_msg = (
f"Access denied to {func_name}: "
f"Nextcloud access not provisioned. "
f"Please call 'nc_auth_provision_access' first."
)
raise ProvisioningRequiredError(error_msg)
if stored_scopes == "all":
+1 -1
View File
@@ -474,7 +474,7 @@ async def user_info_html(request: Request) -> HTMLResponse:
# otherwise fall back to NEXTCLOUD_HOST from settings
settings = get_settings()
nextcloud_host_for_links = (
os.getenv("NEXTCLOUD_PUBLIC_ISSUER_URL") or settings.nextcloud_host
settings.nextcloud_public_issuer_url or settings.nextcloud_host
)
# Build host info HTML (BasicAuth only)
+14
View File
@@ -31,6 +31,8 @@ _DEFAULTS: dict[str, Any] = {
"nextcloud_ca_bundle": None,
"nextcloud_mcp_server_url": None,
"nextcloud_resource_uri": None,
"nextcloud_public_issuer_url": None,
"cookie_secure": None,
# OAuth/OIDC
"oidc_discovery_url": None,
"nextcloud_oidc_client_id": None,
@@ -406,6 +408,16 @@ class Settings:
nextcloud_password: str | None = None
nextcloud_app_password: str | None = None # Preferred over nextcloud_password
# Browser-reachable public URL for OAuth/Login-Flow-v2 redirects when
# NEXTCLOUD_HOST is an internal Docker hostname. Falls back to
# nextcloud_host when unset.
nextcloud_public_issuer_url: str | None = None
# Browser cookie Secure flag. None = auto-detect from nextcloud_host
# scheme (https → True, else False). Set COOKIE_SECURE=true/false to
# override.
cookie_secure: bool | None = None
# Nextcloud SSL/TLS settings
nextcloud_verify_ssl: bool = True
nextcloud_ca_bundle: str | None = None
@@ -777,6 +789,8 @@ def get_settings() -> Settings:
"nextcloud_username": "NEXTCLOUD_USERNAME",
"nextcloud_password": "NEXTCLOUD_PASSWORD",
"nextcloud_app_password": "NEXTCLOUD_APP_PASSWORD",
"nextcloud_public_issuer_url": "NEXTCLOUD_PUBLIC_ISSUER_URL",
"cookie_secure": "COOKIE_SECURE",
# Nextcloud SSL/TLS settings
"nextcloud_verify_ssl": "NEXTCLOUD_VERIFY_SSL",
"nextcloud_ca_bundle": "NEXTCLOUD_CA_BUNDLE",
+171
View File
@@ -0,0 +1,171 @@
"""Unit tests for the MCP elicitation helpers."""
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from nextcloud_mcp_server.auth.elicitation import (
ASTROLABE_SETTINGS_PATH,
_astrolabe_settings_url,
present_provisioning_required,
)
pytestmark = pytest.mark.unit
def _fake_settings(
public_issuer_url: str | None = None, host: str | None = None
) -> SimpleNamespace:
"""Build a Settings-shaped object exposing only the fields elicitation reads."""
return SimpleNamespace(
nextcloud_public_issuer_url=public_issuer_url,
nextcloud_host=host,
)
def test_astrolabe_settings_url_prefers_public_issuer():
"""Public issuer wins over host so the link is browser-reachable in Docker."""
fake = _fake_settings(
public_issuer_url="https://nc.example.com", host="http://internal:8080"
)
with patch("nextcloud_mcp_server.auth.elicitation.get_settings", return_value=fake):
assert (
_astrolabe_settings_url()
== f"https://nc.example.com{ASTROLABE_SETTINGS_PATH}"
)
def test_astrolabe_settings_url_strips_trailing_slash_from_public_issuer():
"""Trailing slash on nextcloud_public_issuer_url is normalized."""
fake = _fake_settings(public_issuer_url="https://nc.example.com/")
with patch("nextcloud_mcp_server.auth.elicitation.get_settings", return_value=fake):
assert (
_astrolabe_settings_url()
== f"https://nc.example.com{ASTROLABE_SETTINGS_PATH}"
)
def test_astrolabe_settings_url_falls_back_to_host():
"""When only nextcloud_host is set, use it (and strip a trailing slash)."""
fake = _fake_settings(host="https://only-host.example.com/")
with patch("nextcloud_mcp_server.auth.elicitation.get_settings", return_value=fake):
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)."""
fake = _fake_settings()
with patch("nextcloud_mcp_server.auth.elicitation.get_settings", return_value=fake):
assert _astrolabe_settings_url() is None
def test_astrolabe_settings_url_returns_none_when_scheme_missing(caplog):
"""Bare hostname (no http:// or https://) → None + a warning so the operator
sees the misconfiguration instead of getting a silently-broken URL."""
fake = _fake_settings(host="internal-host:8080")
with patch("nextcloud_mcp_server.auth.elicitation.get_settings", return_value=fake):
with caplog.at_level("WARNING", logger="nextcloud_mcp_server.auth.elicitation"):
assert _astrolabe_settings_url() is None
assert any(
"missing an http:// or https://" in rec.message for rec in caplog.records
), (
f"expected scheme-missing warning, got records={[r.message for r in caplog.records]}"
)
async def test_present_provisioning_required_elicits_with_url():
"""When NC URL is set and the client supports elicitation, send the URL."""
fake = _fake_settings(public_issuer_url="https://nc.example.com")
ctx = MagicMock()
ctx.elicit = AsyncMock(return_value=SimpleNamespace(action="accept", data=None))
with patch("nextcloud_mcp_server.auth.elicitation.get_settings", return_value=fake):
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():
"""When neither NC URL is set, fall back to the tool-only message."""
fake = _fake_settings()
ctx = MagicMock()
ctx.elicit = AsyncMock(return_value=SimpleNamespace(action="accept", data=None))
with patch("nextcloud_mcp_server.auth.elicitation.get_settings", return_value=fake):
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()
fake = _fake_settings()
with patch("nextcloud_mcp_server.auth.elicitation.get_settings", return_value=fake):
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."""
fake = _fake_settings()
ctx = MagicMock()
ctx.elicit = AsyncMock(side_effect=NotImplementedError("client lacks elicit"))
with patch("nextcloud_mcp_server.auth.elicitation.get_settings", return_value=fake):
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."""
fake = _fake_settings()
ctx = MagicMock()
ctx.elicit = AsyncMock(side_effect=RuntimeError("transport boom"))
with patch("nextcloud_mcp_server.auth.elicitation.get_settings", return_value=fake):
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."""
fake = _fake_settings()
ctx = MagicMock()
ctx.elicit = AsyncMock(return_value=SimpleNamespace(action="decline", data=None))
with patch("nextcloud_mcp_server.auth.elicitation.get_settings", return_value=fake):
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."""
fake = _fake_settings()
ctx = MagicMock()
ctx.elicit = AsyncMock(return_value=SimpleNamespace(action="cancel", data=None))
with patch("nextcloud_mcp_server.auth.elicitation.get_settings", return_value=fake):
result = await present_provisioning_required(ctx)
assert result == "cancelled"
+233 -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.
"""
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,231 @@ 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_and_uses_retry_message_when_user_accepts():
"""When the elicit returns "accepted" the raised error must tell the user
to retry — *not* "call nc_auth_provision_access". The latter would loop
an LLM that just acknowledged the elicitation prompt.
See PR #757 review feedback (cbcoutinho/nextcloud-mcp-server#757).
"""
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 the elicitation module (not scope_authorization) because the
# decorator does a local import of present_provisioning_required to
# avoid a circular import, so the name is re-fetched at call-time.
patch(
"nextcloud_mcp_server.auth.elicitation.present_provisioning_required",
elicit_mock,
),
pytest.raises(ProvisioningRequiredError) as exc_info,
):
await fake_tool_missing_pwd(ctx=ctx)
elicit_mock.assert_awaited_once_with(ctx)
msg = str(exc_info.value)
assert "retry the request" in msg
assert "nc_auth_provision_access" not in msg
async def test_decorator_uses_legacy_message_when_elicitation_unsupported():
"""When the elicit helper returns "message_only" (client lacks elicit
support), the raised error must keep the existing
"call nc_auth_provision_access" instruction so an agent has something
actionable. Mirrors the "accepted" case but for the fallback branch."""
ctx = _make_login_flow_ctx()
@require_scopes("notes.read")
async def fake_tool_missing_pwd_no_elicit(ctx: Context): # noqa: ARG001
return "ok"
fake_settings = SimpleNamespace(enable_login_flow=True)
elicit_mock = AsyncMock(return_value="message_only")
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) as exc_info,
):
await fake_tool_missing_pwd_no_elicit(ctx=ctx)
elicit_mock.assert_awaited_once_with(ctx)
msg = str(exc_info.value)
assert "nc_auth_provision_access" in msg
assert "retry the request" not in msg
async def test_decorator_uses_legacy_message_when_user_declines():
"""When the elicit returns "declined" the user has explicitly declined the
provisioning prompt. They still need to provision before the tool can run,
so the raised error keeps the "call nc_auth_provision_access" instruction
(same fall-through branch as message_only). Lock in this behaviour so a
future refactor that splits the else-branch can't silently change it."""
ctx = _make_login_flow_ctx()
@require_scopes("notes.read")
async def fake_tool_user_declined(ctx: Context): # noqa: ARG001
return "ok"
fake_settings = SimpleNamespace(enable_login_flow=True)
elicit_mock = AsyncMock(return_value="declined")
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) as exc_info,
):
await fake_tool_user_declined(ctx=ctx)
elicit_mock.assert_awaited_once_with(ctx)
msg = str(exc_info.value)
assert "nc_auth_provision_access" in msg
assert "retry the request" not in msg
async def test_decorator_uses_legacy_message_when_user_cancels():
"""When the elicit returns "cancelled" (user dismissed the prompt without
answering), the user is still unprovisioned and needs to call the auth
tool. Same fall-through as declined and message_only — locked in by an
explicit test so the three callers don't drift apart in a future refactor."""
ctx = _make_login_flow_ctx()
@require_scopes("notes.read")
async def fake_tool_user_cancelled(ctx: Context): # noqa: ARG001
return "ok"
fake_settings = SimpleNamespace(enable_login_flow=True)
elicit_mock = AsyncMock(return_value="cancelled")
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) as exc_info,
):
await fake_tool_user_cancelled(ctx=ctx)
elicit_mock.assert_awaited_once_with(ctx)
msg = str(exc_info.value)
assert "nc_auth_provision_access" in msg
assert "retry the request" not in msg
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()