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:
co-authored by
Claude Opus 4.7
parent
f3256e515e
commit
7464340763
@@ -44,8 +44,9 @@ def _should_use_secure_cookies() -> bool:
|
|||||||
if explicit == "false":
|
if explicit == "false":
|
||||||
return False
|
return False
|
||||||
|
|
||||||
# Auto-detect from NEXTCLOUD_HOST protocol
|
# Auto-detect from NEXTCLOUD_HOST protocol (read via Settings for
|
||||||
nextcloud_host = os.getenv("NEXTCLOUD_HOST", "")
|
# consistency with the rest of this file).
|
||||||
|
nextcloud_host = get_settings().nextcloud_host or ""
|
||||||
return nextcloud_host.startswith("https://")
|
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
|
import logging
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
from mcp.server.fastmcp import Context
|
from mcp.server.fastmcp import Context
|
||||||
from pydantic import BaseModel, Field
|
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)
|
Prefers ``nextcloud_public_issuer_url`` (the browser-reachable public URL)
|
||||||
over ``nextcloud_host`` (which may be an internal hostname in Docker
|
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()
|
settings = get_settings()
|
||||||
base = (
|
base = (
|
||||||
@@ -53,9 +56,70 @@ def _astrolabe_settings_url() -> str | None:
|
|||||||
).strip()
|
).strip()
|
||||||
if not base:
|
if not base:
|
||||||
return None
|
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}"
|
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(
|
async def present_login_url(
|
||||||
ctx: Context,
|
ctx: Context,
|
||||||
login_url: str,
|
login_url: str,
|
||||||
@@ -84,44 +148,27 @@ async def present_login_url(
|
|||||||
f"Then check the box below and click OK."
|
f"Then check the box below and click OK."
|
||||||
)
|
)
|
||||||
|
|
||||||
if not hasattr(ctx, "elicit"):
|
outcome, result = await _run_elicit(
|
||||||
logger.debug(
|
ctx,
|
||||||
"Elicitation not available (no elicit method), returning URL in message"
|
message,
|
||||||
)
|
LoginFlowConfirmation,
|
||||||
return "message_only"
|
log_label="login flow completion",
|
||||||
|
)
|
||||||
|
|
||||||
try:
|
if (
|
||||||
result = await ctx.elicit(
|
outcome == "accepted"
|
||||||
message=message,
|
and result is not None
|
||||||
schema=LoginFlowConfirmation,
|
and hasattr(result, "data")
|
||||||
)
|
and not result.data.acknowledged
|
||||||
|
):
|
||||||
if result.action == "accept":
|
# User clicked OK without ticking the box — login completion is still
|
||||||
if hasattr(result, "data") and not result.data.acknowledged: # type: ignore[union-attr]
|
# verified via the LFv2 poller, so we proceed but flag it.
|
||||||
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:
|
|
||||||
logger.warning(
|
logger.warning(
|
||||||
"Elicitation failed unexpectedly (%s: %s), falling back to message",
|
"User accepted login flow without checking the acknowledged box — "
|
||||||
type(e).__name__,
|
"login completion will be verified via polling"
|
||||||
e,
|
|
||||||
)
|
)
|
||||||
return "message_only"
|
|
||||||
|
return outcome
|
||||||
|
|
||||||
|
|
||||||
async def present_provisioning_required(ctx: Context) -> str:
|
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."
|
"Then check the box below and retry the original request."
|
||||||
)
|
)
|
||||||
|
|
||||||
if not hasattr(ctx, "elicit"):
|
outcome, _ = await _run_elicit(
|
||||||
logger.debug(
|
ctx,
|
||||||
"Elicitation not available on context — returning message_only "
|
message,
|
||||||
"(plain ProvisioningRequiredError will surface to the caller)"
|
ProvisioningRequiredConfirmation,
|
||||||
)
|
log_label="provisioning-required prompt",
|
||||||
return "message_only"
|
)
|
||||||
|
return outcome
|
||||||
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"
|
|
||||||
|
|||||||
@@ -168,11 +168,19 @@ def require_scopes(*required_scopes: str):
|
|||||||
# elicitation isn't told to call the auth tool
|
# elicitation isn't told to call the auth tool
|
||||||
# again (which would loop).
|
# again (which would loop).
|
||||||
if elicit_result == "accepted":
|
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 = (
|
error_msg = (
|
||||||
f"Access denied to {func_name}: Nextcloud "
|
f"Access denied to {func_name}: Nextcloud "
|
||||||
f"access was not provisioned at the time of "
|
f"access was not provisioned at the time of "
|
||||||
f"this call. If you just completed "
|
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:
|
else:
|
||||||
error_msg = (
|
error_msg = (
|
||||||
|
|||||||
@@ -63,6 +63,20 @@ def test_astrolabe_settings_url_returns_none_when_unset():
|
|||||||
assert _astrolabe_settings_url() is None
|
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():
|
async def test_present_provisioning_required_elicits_with_url():
|
||||||
"""When NC URL is set and the client supports elicitation, send the URL."""
|
"""When NC URL is set and the client supports elicitation, send the URL."""
|
||||||
fake = _fake_settings(public_issuer_url="https://nc.example.com")
|
fake = _fake_settings(public_issuer_url="https://nc.example.com")
|
||||||
|
|||||||
Reference in New Issue
Block a user