Merge remote-tracking branch 'origin/master' into security/oauth-session-hardening-626
This commit is contained in:
Chris Coutinho
@@ -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"
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user