diff --git a/.claude/skills/pre-push-review/SKILL.md b/.claude/skills/pre-push-review/SKILL.md index f2214bf7..26883d49 100644 --- a/.claude/skills/pre-push-review/SKILL.md +++ b/.claude/skills/pre-push-review/SKILL.md @@ -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 (2–5min) ```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, diff --git a/docs/login-flow-v2.md b/docs/login-flow-v2.md index 73c91ab4..10558169 100644 --- a/docs/login-flow-v2.md +++ b/docs/login-flow-v2.md @@ -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:///.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: diff --git a/nextcloud_mcp_server/api/passwords.py b/nextcloud_mcp_server/api/passwords.py index 4dd5d046..df3f5ba4 100644 --- a/nextcloud_mcp_server/api/passwords.py +++ b/nextcloud_mcp_server/api/passwords.py @@ -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}") diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index 68fd08bd..b79e87dd 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -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 diff --git a/nextcloud_mcp_server/auth/browser_oauth_routes.py b/nextcloud_mcp_server/auth/browser_oauth_routes.py index c1050532..8b33b0fb 100644 --- a/nextcloud_mcp_server/auth/browser_oauth_routes.py +++ b/nextcloud_mcp_server/auth/browser_oauth_routes.py @@ -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) diff --git a/nextcloud_mcp_server/auth/elicitation.py b/nextcloud_mcp_server/auth/elicitation.py index 871e827b..57b59d4c 100644 --- a/nextcloud_mcp_server/auth/elicitation.py +++ b/nextcloud_mcp_server/auth/elicitation.py @@ -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 diff --git a/nextcloud_mcp_server/auth/oauth_routes.py b/nextcloud_mcp_server/auth/oauth_routes.py index 1eb39286..f19d9b50 100644 --- a/nextcloud_mcp_server/auth/oauth_routes.py +++ b/nextcloud_mcp_server/auth/oauth_routes.py @@ -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) diff --git a/nextcloud_mcp_server/auth/provision_routes.py b/nextcloud_mcp_server/auth/provision_routes.py index b0b3fab1..07962ba2 100644 --- a/nextcloud_mcp_server/auth/provision_routes.py +++ b/nextcloud_mcp_server/auth/provision_routes.py @@ -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("/")) diff --git a/nextcloud_mcp_server/auth/scope_authorization.py b/nextcloud_mcp_server/auth/scope_authorization.py index 5317d271..6e6efd66 100644 --- a/nextcloud_mcp_server/auth/scope_authorization.py +++ b/nextcloud_mcp_server/auth/scope_authorization.py @@ -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": diff --git a/nextcloud_mcp_server/auth/userinfo_routes.py b/nextcloud_mcp_server/auth/userinfo_routes.py index d3f20a57..a9277134 100644 --- a/nextcloud_mcp_server/auth/userinfo_routes.py +++ b/nextcloud_mcp_server/auth/userinfo_routes.py @@ -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) diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index eedc254c..eb2cb59f 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -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", diff --git a/tests/unit/test_elicitation.py b/tests/unit/test_elicitation.py new file mode 100644 index 00000000..4183ebb8 --- /dev/null +++ b/tests/unit/test_elicitation.py @@ -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" diff --git a/tests/unit/test_scope_authorization_stored.py b/tests/unit/test_scope_authorization_stored.py index a7315e70..85f76394 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,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()