fix(auth): address PR #757 round-2 review feedback
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) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
co-authored by
Claude Opus 4.7
Claude Opus 4.7
parent
f3256e515e
commit
7464340763
@@ -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://")
|
||||
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 = (
|
||||
|
||||
@@ -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")
|
||||
|
||||
Reference in New Issue
Block a user