From f3256e515e1cf40f3f4b1e3c2c4de7eb44ec4ed4 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sat, 2 May 2026 16:16:19 +0200 Subject: [PATCH] refactor(config): consolidate NEXTCLOUD_PUBLIC_ISSUER_URL through Settings Lift NEXTCLOUD_PUBLIC_ISSUER_URL out of raw os.getenv reads into Settings.nextcloud_public_issuer_url across all 8 production call sites (app.py x2, oauth_routes.py x2, browser_oauth_routes.py, provision_routes.py, userinfo_routes.py, elicitation.py). cli.py remains the env-write source so the existing config-by-flag pipeline still works. Also addresses remaining PR #757 review nits: - elicitation.py: align URL-present/absent wording on "open in your browser" so users don't try clicking in the terminal - test_scope_authorization_stored.py: lock in the deliberately-shared fall-through branch with explicit declined/cancelled decorator tests - test_elicitation.py: switch from monkeypatch.setenv to patch(get_settings) since Settings is now the canonical surface Co-Authored-By: Claude Opus 4.7 (1M context) --- nextcloud_mcp_server/app.py | 4 +- .../auth/browser_oauth_routes.py | 3 +- nextcloud_mcp_server/auth/elicitation.py | 31 +++--- nextcloud_mcp_server/auth/oauth_routes.py | 4 +- nextcloud_mcp_server/auth/provision_routes.py | 3 +- nextcloud_mcp_server/auth/userinfo_routes.py | 2 +- nextcloud_mcp_server/config.py | 7 ++ tests/unit/test_elicitation.py | 98 +++++++++++-------- tests/unit/test_scope_authorization_stored.py | 83 ++++++++++++++++ 9 files changed, 174 insertions(+), 61 deletions(-) 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..0d7628d8 100644 --- a/nextcloud_mcp_server/auth/browser_oauth_routes.py +++ b/nextcloud_mcp_server/auth/browser_oauth_routes.py @@ -22,6 +22,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 @@ -167,7 +168,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 a5d1abfd..62b3d0bd 100644 --- a/nextcloud_mcp_server/auth/elicitation.py +++ b/nextcloud_mcp_server/auth/elicitation.py @@ -5,19 +5,20 @@ when the client supports it, or falling back to returning the URL in a message. """ import logging -import os 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 NEXTCLOUD_PUBLIC_ISSUER_URL / -# 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. +# 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" @@ -40,14 +41,15 @@ class ProvisioningRequiredConfirmation(BaseModel): def _astrolabe_settings_url() -> str | None: - """Construct the Astrolabe settings page URL from environment. + """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 + 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. """ + settings = get_settings() base = ( - os.getenv("NEXTCLOUD_PUBLIC_ISSUER_URL") or os.getenv("NEXTCLOUD_HOST") or "" + settings.nextcloud_public_issuer_url or settings.nextcloud_host or "" ).strip() if not base: return None @@ -132,9 +134,10 @@ async def present_provisioning_required(ctx: Context) -> str: has to translate. The Astrolabe settings URL is reconstructed from - ``NEXTCLOUD_PUBLIC_ISSUER_URL`` / ``NEXTCLOUD_HOST``; if Astrolabe is not - installed the link 404s and the user falls back to the tool path - suggested in the same message. + ``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`: @@ -148,7 +151,7 @@ async def present_provisioning_required(ctx: Context) -> str: 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 directly.\n\n" + "Login Flow v2 URL you can open in your browser.\n\n" "Then check the box below and retry the original request." ) else: 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..0ea380bf 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 @@ -251,7 +250,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/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..95652f80 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -31,6 +31,7 @@ _DEFAULTS: dict[str, Any] = { "nextcloud_ca_bundle": None, "nextcloud_mcp_server_url": None, "nextcloud_resource_uri": None, + "nextcloud_public_issuer_url": None, # OAuth/OIDC "oidc_discovery_url": None, "nextcloud_oidc_client_id": None, @@ -406,6 +407,11 @@ 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 + # Nextcloud SSL/TLS settings nextcloud_verify_ssl: bool = True nextcloud_ca_bundle: str | None = None @@ -777,6 +783,7 @@ 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", # 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 index 9785fca4..9dcac205 100644 --- a/tests/unit/test_elicitation.py +++ b/tests/unit/test_elicitation.py @@ -1,7 +1,7 @@ """Unit tests for the MCP elicitation helpers.""" from types import SimpleNamespace -from unittest.mock import AsyncMock, MagicMock +from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -14,55 +14,63 @@ from nextcloud_mcp_server.auth.elicitation import ( pytestmark = pytest.mark.unit -@pytest.fixture(autouse=True) -def clear_nc_env(monkeypatch): - """Strip the NC URL env vars by default; tests opt back in.""" - monkeypatch.delenv("NEXTCLOUD_PUBLIC_ISSUER_URL", raising=False) - monkeypatch.delenv("NEXTCLOUD_HOST", raising=False) +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(monkeypatch): +def test_astrolabe_settings_url_prefers_public_issuer(): """Public issuer wins over host so the link is browser-reachable in Docker.""" - monkeypatch.setenv("NEXTCLOUD_PUBLIC_ISSUER_URL", "https://nc.example.com") - monkeypatch.setenv("NEXTCLOUD_HOST", "http://internal:8080") - - assert ( - _astrolabe_settings_url() == f"https://nc.example.com{ASTROLABE_SETTINGS_PATH}" + 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(monkeypatch): - """Trailing slash on NEXTCLOUD_PUBLIC_ISSUER_URL is normalized.""" - monkeypatch.setenv("NEXTCLOUD_PUBLIC_ISSUER_URL", "https://nc.example.com/") - - 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(monkeypatch): - """When only NEXTCLOUD_HOST is set, use it (and strip a trailing slash).""" - monkeypatch.setenv("NEXTCLOUD_HOST", "https://only-host.example.com/") - - assert ( - _astrolabe_settings_url() - == f"https://only-host.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).""" - assert _astrolabe_settings_url() is None + fake = _fake_settings() + with patch("nextcloud_mcp_server.auth.elicitation.get_settings", return_value=fake): + assert _astrolabe_settings_url() is None -async def test_present_provisioning_required_elicits_with_url(monkeypatch): +async def test_present_provisioning_required_elicits_with_url(): """When NC URL is set and the client supports elicitation, send the URL.""" - monkeypatch.setenv("NEXTCLOUD_PUBLIC_ISSUER_URL", "https://nc.example.com") - + fake = _fake_settings(public_issuer_url="https://nc.example.com") ctx = MagicMock() ctx.elicit = AsyncMock(return_value=SimpleNamespace(action="accept", data=None)) - result = await present_provisioning_required(ctx) + 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() @@ -71,12 +79,14 @@ async def test_present_provisioning_required_elicits_with_url(monkeypatch): assert "nc_auth_provision_access" in sent_message -async def test_present_provisioning_required_without_url(monkeypatch): +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)) - result = await present_provisioning_required(ctx) + 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"] @@ -92,46 +102,56 @@ async def test_present_provisioning_required_no_elicit_method(): ctx = _NoElicit() - result = await present_provisioning_required(ctx) # type: ignore[arg-type] + 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")) - result = await present_provisioning_required(ctx) + 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")) - result = await present_provisioning_required(ctx) + 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)) - result = await present_provisioning_required(ctx) + 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)) - result = await present_provisioning_required(ctx) + 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 432290ef..e4969070 100644 --- a/tests/unit/test_scope_authorization_stored.py +++ b/tests/unit/test_scope_authorization_stored.py @@ -196,6 +196,89 @@ async def test_decorator_uses_legacy_message_when_elicitation_unsupported(): 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),