From 7464340763434604c1d31d4b2ace210b55b1c207 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sat, 2 May 2026 16:33:43 +0200 Subject: [PATCH] fix(auth): address PR #757 round-2 review feedback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four review items from the second-round review on PR #757: - scope_authorization: broaden the post-elicit retry message to acknowledge the 5-minute scope-cache TTL — if the LFv2 poller is still in-flight at acknowledge-time, the immediate retry can still hit a stale cache. - elicitation: extract a shared `_run_elicit(ctx, message, schema, *, log_label)` helper so `present_login_url` and `present_provisioning_required` no longer duplicate the hasattr-guard / try-NotImplementedError / try-Exception fallback block. The data-acknowledged warning specific to login-flow stays in `present_login_url` so behaviour is preserved exactly. - elicitation: detect missing http:// / https:// scheme in `_astrolabe_settings_url`, log a warning, and return None — caller renders the safe tool-only fallback instead of producing a broken link. New unit test locks this in. - browser_oauth_routes: replace the stray `os.getenv(\"NEXTCLOUD_HOST\")` in `_should_use_secure_cookies` with `get_settings().nextcloud_host` for consistency with the rest of the file (PR #757 review nit). Co-Authored-By: Claude Opus 4.7 (1M context) --- .../auth/browser_oauth_routes.py | 5 +- nextcloud_mcp_server/auth/elicitation.py | 162 ++++++++++-------- .../auth/scope_authorization.py | 10 +- tests/unit/test_elicitation.py | 14 ++ 4 files changed, 116 insertions(+), 75 deletions(-) diff --git a/nextcloud_mcp_server/auth/browser_oauth_routes.py b/nextcloud_mcp_server/auth/browser_oauth_routes.py index 0d7628d8..f579d9f9 100644 --- a/nextcloud_mcp_server/auth/browser_oauth_routes.py +++ b/nextcloud_mcp_server/auth/browser_oauth_routes.py @@ -44,8 +44,9 @@ def _should_use_secure_cookies() -> bool: if explicit == "false": return False - # Auto-detect from NEXTCLOUD_HOST protocol - nextcloud_host = os.getenv("NEXTCLOUD_HOST", "") + # Auto-detect from NEXTCLOUD_HOST protocol (read via Settings for + # consistency with the rest of this file). + nextcloud_host = get_settings().nextcloud_host or "" return nextcloud_host.startswith("https://") diff --git a/nextcloud_mcp_server/auth/elicitation.py b/nextcloud_mcp_server/auth/elicitation.py index 62b3d0bd..014c9b34 100644 --- a/nextcloud_mcp_server/auth/elicitation.py +++ b/nextcloud_mcp_server/auth/elicitation.py @@ -5,6 +5,7 @@ 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 @@ -45,7 +46,9 @@ def _astrolabe_settings_url() -> str | None: 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. + deployments). Returns None if neither is set, 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 = ( @@ -53,9 +56,70 @@ def _astrolabe_settings_url() -> str | None: ).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, @@ -84,44 +148,27 @@ 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( - "Elicitation failed unexpectedly (%s: %s), falling back to message", - type(e).__name__, - e, + "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: @@ -163,39 +210,10 @@ async def present_provisioning_required(ctx: Context) -> str: "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" + outcome, _ = await _run_elicit( + ctx, + message, + ProvisioningRequiredConfirmation, + log_label="provisioning-required prompt", + ) + return outcome diff --git a/nextcloud_mcp_server/auth/scope_authorization.py b/nextcloud_mcp_server/auth/scope_authorization.py index 25d02ab2..a027e1b8 100644 --- a/nextcloud_mcp_server/auth/scope_authorization.py +++ b/nextcloud_mcp_server/auth/scope_authorization.py @@ -168,11 +168,19 @@ def require_scopes(*required_scopes: str): # 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). Both nc_auth_provision_access + # and the Astrolabe web route invalidate the cache when + # they finish, but if the LFv2 poller is still in-flight + # at acknowledge-time the next retry can still hit the + # stale cache — hence the "wait a moment" qualifier. 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"provisioning, please retry the request — " + f"if it still fails, provisioning may still be " + f"completing; wait a moment and try again." ) else: error_msg = ( diff --git a/tests/unit/test_elicitation.py b/tests/unit/test_elicitation.py index 9dcac205..4183ebb8 100644 --- a/tests/unit/test_elicitation.py +++ b/tests/unit/test_elicitation.py @@ -63,6 +63,20 @@ def test_astrolabe_settings_url_returns_none_when_unset(): 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")