admin/mcp-nextcloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #761 from cbcoutinho/fix/webhooks-api-basic-auth

Browse Source 
fix(webhooks): use app-password basic auth for NC API calls
This commit is contained in:
Chris Coutinho
2026-05-04 00:27:02 +02:00
committed by GitHub
parent beef77d785 b3a7587f1a
commit c21ec423d1
4 changed files with 459 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nextcloud_mcp_server/api/_auth.py
+48
View File
@@ -0,0 +1,48 @@
"""Credential resolution helpers for non-MCP-Context API endpoints.
Starlette routes (e.g. ``api/webhooks.py``) authenticate the inbound request
via OAuth bearer validation, then need to make Nextcloud API calls on behalf
of that user. Per ADR-022 / ``docs/login-flow-v2.md`` the data leg to
Nextcloud always uses **HTTP Basic Auth with the user's app password**, never
the OAuth token. This module resolves that credential pair.
The MCP tool path uses ``context.get_client(ctx)``; this is the equivalent
helper for callers that have a validated ``user_id`` but no MCP ``Context``.
"""
from nextcloud_mcp_server.auth.scope_authorization import ProvisioningRequiredError
from nextcloud_mcp_server.auth.storage import get_shared_storage
async def get_basic_auth_for_user(user_id: str) -> tuple[str, str]:
"""Resolve ``(username, app_password)`` for an OAuth-validated user.
Reads the per-user app password provisioned via Login Flow v2 from
encrypted SQLite storage. The username returned is the actual Nextcloud
username recorded at provisioning time (which may differ from the IdP
user-id when an external IdP is configured).
Args:
user_id: MCP user identifier extracted from a validated OAuth token.
Returns:
Tuple ``(username, app_password)`` ready to pass to
``httpx.BasicAuth``.
Raises:
ProvisioningRequiredError: No app password is stored for ``user_id``.
The caller should surface this to the client so the user can
complete Login Flow v2 (typically via ``nc_auth_provision_access``
or the Astrolabe settings page).
"""
storage = await get_shared_storage()
app_data = await storage.get_app_password_with_scopes(user_id)
if not app_data:
raise ProvisioningRequiredError(
f"No Nextcloud app password provisioned for user {user_id!r}. "
"Complete Login Flow v2 (nc_auth_provision_access) before "
"calling this endpoint."
)
username = app_data.get("username") or user_id
return username, app_data["app_password"]
nextcloud_mcp_server/api/webhooks.py
+51 -49
View File
@@ -6,18 +6,28 @@ These endpoints are used by the Nextcloud PHP app (Astrolabe) to:
- Create, list, and delete webhook registrations
All endpoints require OAuth bearer token authentication via UnifiedTokenVerifier.
Auth model: the OAuth bearer is validated at the perimeter to identify the
user (``validate_token_and_get_user``); calls to Nextcloud are then made with
the user's stored app password via HTTP Basic Auth (see
``docs/login-flow-v2.md`` and ADR-022). The OAuth bearer is NEVER forwarded
to Nextcloud — that pattern depended on upstream user_oidc patches that were
never merged and is incompatible with admin endpoints gated by
``@PasswordConfirmationRequired``.
"""
import logging
import httpx
from starlette.requests import Request
from starlette.responses import JSONResponse
from nextcloud_mcp_server.api._auth import get_basic_auth_for_user
from nextcloud_mcp_server.api.management import (
_sanitize_error_for_client,
extract_bearer_token,
validate_token_and_get_user,
)
from nextcloud_mcp_server.auth.scope_authorization import ProvisioningRequiredError
from nextcloud_mcp_server.auth.webhook_routes import webhook_auth_pair
from nextcloud_mcp_server.client.webhooks import WebhooksClient
@@ -47,29 +57,19 @@ async def get_installed_apps(request: Request) -> JSONResponse:
)
try:
# Get Bearer token from request — forwarded to Nextcloud so the
# capabilities response includes per-user / per-app entries (anonymous
# capabilities omits notes, tables, forms etc.).
token = extract_bearer_token(request)
if not token:
raise ValueError("Missing Authorization header")
username, app_password = await get_basic_auth_for_user(user_id)
# Get Nextcloud host from OAuth context
oauth_ctx = request.app.state.oauth_context
nextcloud_host = oauth_ctx.get("config", {}).get("nextcloud_host", "")
if not nextcloud_host:
raise ValueError("Nextcloud host not configured")
# Use OCS v2 capabilities. The legacy /ocs/v1.php/cloud/apps endpoint is
# admin-only AND @PasswordConfirmationRequired — neither is satisfiable
# via an OAuth bearer token, so it always 401s. Capabilities has no such
# gates and returns a map keyed by app id for every enabled app that
# implements OCSCapabilities, which is sufficient to populate the
# webhook presets UI.
# OCS v2 capabilities is keyed by app-id for every enabled app that
# implements OCSCapabilities — sufficient for the webhook presets UI
# without needing the admin-only /cloud/apps endpoint.
async with nextcloud_httpx_client(
base_url=nextcloud_host,
headers={"Authorization": f"Bearer {token}"},
auth=httpx.BasicAuth(username, app_password),
timeout=30.0,
) as client:
response = await client.get(
@@ -87,6 +87,12 @@ async def get_installed_apps(request: Request) -> JSONResponse:
return JSONResponse({"apps": apps})
except ProvisioningRequiredError as e:
logger.info("Provisioning required for user %s: %s", user_id, e)
return JSONResponse(
{"error": "Provisioning required", "message": str(e)},
status_code=428,
)
except Exception as e:
logger.error("Error getting installed apps for user %s: %s", user_id, e)
return JSONResponse(
@@ -119,30 +125,28 @@ async def list_webhooks(request: Request) -> JSONResponse:
)
try:
# Get Bearer token from request
token = extract_bearer_token(request)
if not token:
raise ValueError("Missing Authorization header")
username, app_password = await get_basic_auth_for_user(user_id)
# Get Nextcloud host from OAuth context
oauth_ctx = request.app.state.oauth_context
nextcloud_host = oauth_ctx.get("config", {}).get("nextcloud_host", "")
if not nextcloud_host:
raise ValueError("Nextcloud host not configured")
# Create authenticated HTTP client
async with nextcloud_httpx_client(
base_url=nextcloud_host,
headers={"Authorization": f"Bearer {token}"},
auth=httpx.BasicAuth(username, app_password),
timeout=30.0,
) as client:
# Use WebhooksClient to list webhooks
webhooks_client = WebhooksClient(client, user_id)
webhooks_client = WebhooksClient(client, username)
webhooks = await webhooks_client.list_webhooks()
return JSONResponse({"webhooks": webhooks})
except ProvisioningRequiredError as e:
logger.info("Provisioning required for user %s: %s", user_id, e)
return JSONResponse(
{"error": "Provisioning required", "message": str(e)},
status_code=428,
)
except Exception as e:
logger.error("Error listing webhooks for user %s: %s", user_id, e)
return JSONResponse(
@@ -198,27 +202,21 @@ async def create_webhook(request: Request) -> JSONResponse:
status_code=400,
)
# Get Bearer token from request
token = extract_bearer_token(request)
if not token:
raise ValueError("Missing Authorization header")
username, app_password = await get_basic_auth_for_user(user_id)
# Get Nextcloud host from OAuth context
oauth_ctx = request.app.state.oauth_context
nextcloud_host = oauth_ctx.get("config", {}).get("nextcloud_host", "")
if not nextcloud_host:
raise ValueError("Nextcloud host not configured")
# Create authenticated HTTP client
async with nextcloud_httpx_client(
base_url=nextcloud_host,
headers={"Authorization": f"Bearer {token}"},
auth=httpx.BasicAuth(username, app_password),
timeout=30.0,
) as client:
# Use WebhooksClient to create webhook. Inject auth headers when
# WEBHOOK_SECRET is configured so deliveries are authenticated.
webhooks_client = WebhooksClient(client, user_id)
# Inject delivery auth headers when WEBHOOK_SECRET is configured so
# that webhook deliveries from Nextcloud back to us are authenticated.
webhooks_client = WebhooksClient(client, username)
auth_method, auth_data = webhook_auth_pair()
webhook_data = await webhooks_client.create_webhook(
event=event,
@@ -230,6 +228,12 @@ async def create_webhook(request: Request) -> JSONResponse:
return JSONResponse({"webhook": webhook_data})
except ProvisioningRequiredError as e:
logger.info("Provisioning required for user %s: %s", user_id, e)
return JSONResponse(
{"error": "Provisioning required", "message": str(e)},
status_code=428,
)
except Exception as e:
logger.error("Error creating webhook for user %s: %s", user_id, e)
return JSONResponse(
@@ -278,30 +282,28 @@ async def delete_webhook(request: Request) -> JSONResponse:
status_code=400,
)
# Get Bearer token from request
token = extract_bearer_token(request)
if not token:
raise ValueError("Missing Authorization header")
username, app_password = await get_basic_auth_for_user(user_id)
# Get Nextcloud host from OAuth context
oauth_ctx = request.app.state.oauth_context
nextcloud_host = oauth_ctx.get("config", {}).get("nextcloud_host", "")
if not nextcloud_host:
raise ValueError("Nextcloud host not configured")
# Create authenticated HTTP client
async with nextcloud_httpx_client(
base_url=nextcloud_host,
headers={"Authorization": f"Bearer {token}"},
auth=httpx.BasicAuth(username, app_password),
timeout=30.0,
) as client:
# Use WebhooksClient to delete webhook
webhooks_client = WebhooksClient(client, user_id)
webhooks_client = WebhooksClient(client, username)
await webhooks_client.delete_webhook(webhook_id=webhook_id)
return JSONResponse({"success": True, "message": "Webhook deleted"})
except ProvisioningRequiredError as e:
logger.info("Provisioning required for user %s: %s", user_id, e)
return JSONResponse(
{"error": "Provisioning required", "message": str(e)},
status_code=428,
)
except Exception as e:
logger.error("Error deleting webhook for user %s: %s", user_id, e)
return JSONResponse(
tests/unit/test_management_apps_endpoint.py
+46 -24
View File
@@ -1,21 +1,23 @@
"""
Unit tests for the Management API /api/v1/apps endpoint.
These tests cover the regression where /api/v1/apps proxied to
/ocs/v1.php/cloud/apps — an admin-only and @PasswordConfirmationRequired
endpoint that always 401s for OAuth bearer tokens. The handler now uses
/ocs/v2.php/cloud/capabilities, which accepts the bearer and returns an
authenticated capability map keyed by app id.
The handler hits ``/ocs/v2.php/cloud/capabilities`` (not the legacy admin-only
``/ocs/v1.php/cloud/apps`` which was ``@PasswordConfirmationRequired``) and
authenticates via the user's stored Login Flow v2 app password using HTTP
Basic Auth. The OAuth bearer is **never** forwarded to Nextcloud (see
``docs/login-flow-v2.md`` and ADR-022).
"""
from unittest.mock import AsyncMock, MagicMock
import httpx
import pytest
from starlette.applications import Starlette
from starlette.routing import Route
from starlette.testclient import TestClient
from nextcloud_mcp_server.api.webhooks import get_installed_apps
from nextcloud_mcp_server.auth.scope_authorization import ProvisioningRequiredError
pytestmark = pytest.mark.unit
@@ -33,8 +35,19 @@ def _patch_token_validation(mocker, user_id: str = "admin") -> None:
)
def _patch_outbound_client(mocker, response: MagicMock) -> AsyncMock:
"""Patch the outbound httpx client; return the mocked .get() AsyncMock."""
def _patch_basic_auth(
mocker, username: str = "admin", app_password: str = "stored-app-pwd"
) -> AsyncMock:
"""Patch get_basic_auth_for_user to return canned credentials."""
return mocker.patch(
"nextcloud_mcp_server.api.webhooks.get_basic_auth_for_user",
new=AsyncMock(return_value=(username, app_password)),
)
def _patch_outbound_client(mocker, response: MagicMock) -> MagicMock:
"""Patch the outbound httpx client; return the factory MagicMock so tests
can introspect the kwargs (esp. ``auth=``) it was called with."""
mock_get = AsyncMock(return_value=response)
mock_client = AsyncMock()
mock_client.get = mock_get
@@ -48,7 +61,6 @@ def _patch_outbound_client(mocker, response: MagicMock) -> AsyncMock:
mocker.patch(
"nextcloud_mcp_server.api.webhooks.nextcloud_httpx_client", mock_factory
)
# Return both so tests can assert on factory kwargs and call args
mock_factory.attach_mock(mock_get, "get")
return mock_factory
@@ -63,6 +75,7 @@ def _capabilities_response(capabilities: dict[str, dict]) -> MagicMock:
async def test_returns_sorted_capability_keys(mocker):
"""Happy path: handler hits OCS v2 capabilities and returns sorted app keys."""
_patch_token_validation(mocker)
_patch_basic_auth(mocker)
response = _capabilities_response(
{
"notes": {"api_version": "1.4"},
@@ -82,22 +95,24 @@ async def test_returns_sorted_capability_keys(mocker):
assert http_response.status_code == 200
assert http_response.json() == {"apps": ["core", "files", "notes", "tables"]}
# Regression check: outbound URL is OCS v2 capabilities, NOT v1 cloud/apps.
# /ocs/v1.php/cloud/apps was admin-only + @PasswordConfirmationRequired
# which always 401s for OAuth bearer tokens — that was the bug.
# Outbound URL is OCS v2 capabilities, NOT v1 cloud/apps.
factory.get.assert_awaited_once()
called_path = factory.get.call_args.args[0]
assert called_path == "/ocs/v2.php/cloud/capabilities"
assert "/cloud/apps" not in called_path
# Bearer token forwarded so authenticated capabilities are returned
factory_call_kwargs = factory.call_args.kwargs
assert factory_call_kwargs["headers"] == {"Authorization": "Bearer test-token"}
# Outbound auth is BasicAuth (NOT Bearer) — the OAuth token must not be
# forwarded to Nextcloud per ADR-022 / docs/login-flow-v2.md.
factory_kwargs = factory.call_args.kwargs
assert "headers" not in factory_kwargs or "Authorization" not in (
factory_kwargs.get("headers") or {}
)
assert isinstance(factory_kwargs["auth"], httpx.BasicAuth)
async def test_empty_capabilities_returns_empty_list(mocker):
"""Empty capabilities map → empty apps list, not an error."""
_patch_token_validation(mocker)
_patch_basic_auth(mocker)
response = _capabilities_response({})
_patch_outbound_client(mocker, response)
@@ -114,6 +129,7 @@ async def test_empty_capabilities_returns_empty_list(mocker):
async def test_ocs_error_returns_500_with_sanitized_message(mocker):
"""Non-200 from Nextcloud OCS surfaces as a generic 500 to the caller."""
_patch_token_validation(mocker)
_patch_basic_auth(mocker)
error_response = MagicMock()
error_response.status_code = 503
_patch_outbound_client(mocker, error_response)
@@ -135,6 +151,7 @@ async def test_ocs_error_returns_500_with_sanitized_message(mocker):
async def test_missing_nextcloud_host_returns_500(mocker):
"""Misconfigured oauth_context (no nextcloud_host) surfaces as 500."""
_patch_token_validation(mocker)
_patch_basic_auth(mocker)
response = _capabilities_response({})
_patch_outbound_client(mocker, response)
@@ -149,17 +166,22 @@ async def test_missing_nextcloud_host_returns_500(mocker):
assert http_response.status_code == 500
async def test_missing_authorization_returns_500(mocker):
"""When token validation passes but Authorization header is absent, the
handler can't construct the outbound bearer header — returns 500 with a
sanitized message."""
async def test_unprovisioned_user_returns_428(mocker):
"""Users without a stored app password get HTTP 428 (Precondition Required,
RFC 6585) so the client can surface a 'complete Login Flow v2' UX rather
than a generic 500. 428 is the right semantic — the request requires a
prerequisite step (provisioning) before it can succeed."""
_patch_token_validation(mocker)
response = _capabilities_response({})
_patch_outbound_client(mocker, response)
mocker.patch(
"nextcloud_mcp_server.api.webhooks.get_basic_auth_for_user",
new=AsyncMock(side_effect=ProvisioningRequiredError("not provisioned")),
)
app = _build_test_app()
client = TestClient(app)
# No Authorization header
http_response = client.get("/api/v1/apps")
http_response = client.get(
"/api/v1/apps", headers={"Authorization": "Bearer test-token"}
)
assert http_response.status_code == 500
assert http_response.status_code == 428
assert http_response.json()["error"] == "Provisioning required"
tests/unit/test_webhooks_api_auth.py
+314
View File
@@ -0,0 +1,314 @@
"""Unit tests asserting the webhook API endpoints authenticate to Nextcloud
with an app-password BasicAuth credential — never by forwarding the inbound
OAuth bearer token.
Per ADR-022 / ``docs/login-flow-v2.md`` the data leg from MCP server to
Nextcloud always uses HTTP Basic Auth with a per-user app password obtained
via Login Flow v2. Forwarding OAuth bearers to Nextcloud was the obsolete
pre-ADR-022 pattern; it relies on upstream user_oidc patches that were never
merged and is incompatible with admin endpoints gated by
``@PasswordConfirmationRequired`` (e.g. ``webhook_listeners``).
"""
from unittest.mock import AsyncMock, MagicMock
import httpx
import pytest
from starlette.applications import Starlette
from starlette.routing import Route
from starlette.testclient import TestClient
from nextcloud_mcp_server.api.webhooks import (
create_webhook,
delete_webhook,
list_webhooks,
)
from nextcloud_mcp_server.auth.scope_authorization import ProvisioningRequiredError
pytestmark = pytest.mark.unit
def _build_test_app() -> Starlette:
app = Starlette(
routes=[
Route("/api/v1/webhooks", list_webhooks, methods=["GET"]),
Route("/api/v1/webhooks", create_webhook, methods=["POST"]),
Route("/api/v1/webhooks/{webhook_id}", delete_webhook, methods=["DELETE"]),
]
)
app.state.oauth_context = {"config": {"nextcloud_host": "http://nc.test"}}
return app
def _patch_token_validation(mocker, user_id: str = "admin") -> None:
mocker.patch(
"nextcloud_mcp_server.api.webhooks.validate_token_and_get_user",
new=AsyncMock(return_value=(user_id, {"sub": user_id})),
)
def _patch_basic_auth(
mocker, username: str = "admin", app_password: str = "stored-app-pwd"
) -> AsyncMock:
return mocker.patch(
"nextcloud_mcp_server.api.webhooks.get_basic_auth_for_user",
new=AsyncMock(return_value=(username, app_password)),
)
def _patch_webhooks_client(mocker, **methods) -> MagicMock:
"""Replace WebhooksClient with a stub. ``methods`` maps method name →
return value for AsyncMock(side_effect/return_value)."""
instance = MagicMock()
for name, value in methods.items():
setattr(instance, name, AsyncMock(return_value=value))
cls = MagicMock(return_value=instance)
mocker.patch("nextcloud_mcp_server.api.webhooks.WebhooksClient", cls)
return cls
def _patch_outbound_client_factory(mocker) -> MagicMock:
"""Patch nextcloud_httpx_client so we can assert on the kwargs (esp.
``auth=``) the handler called it with."""
mock_client = AsyncMock()
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
factory = MagicMock(return_value=mock_client)
mocker.patch("nextcloud_mcp_server.api.webhooks.nextcloud_httpx_client", factory)
return factory
def _assert_basic_auth_not_bearer(factory: MagicMock) -> None:
"""Outbound httpx kwargs must use ``auth=BasicAuth(...)`` and NOT a
Bearer ``Authorization`` header."""
factory.assert_called_once()
kwargs = factory.call_args.kwargs
assert isinstance(kwargs["auth"], httpx.BasicAuth), (
"Outbound NC request must use BasicAuth, not bearer-header forwarding"
)
headers = kwargs.get("headers") or {}
assert "Authorization" not in headers, (
"Outbound NC request must NOT carry an Authorization header — "
"the OAuth bearer must never be forwarded to Nextcloud"
)
# ---------------------------------------------------------------------------
# list_webhooks
# ---------------------------------------------------------------------------
async def test_list_webhooks_uses_basic_auth(mocker):
_patch_token_validation(mocker)
_patch_basic_auth(mocker, username="alice", app_password="alice-pwd")
factory = _patch_outbound_client_factory(mocker)
_patch_webhooks_client(
mocker, list_webhooks=[{"id": 1, "event": "test", "uri": "http://x"}]
)
client = TestClient(_build_test_app())
resp = client.get("/api/v1/webhooks", headers={"Authorization": "Bearer mcp-token"})
assert resp.status_code == 200
assert resp.json() == {"webhooks": [{"id": 1, "event": "test", "uri": "http://x"}]}
_assert_basic_auth_not_bearer(factory)
async def test_list_webhooks_returns_428_when_unprovisioned(mocker):
_patch_token_validation(mocker)
mocker.patch(
"nextcloud_mcp_server.api.webhooks.get_basic_auth_for_user",
new=AsyncMock(side_effect=ProvisioningRequiredError("not provisioned")),
)
client = TestClient(_build_test_app())
resp = client.get("/api/v1/webhooks", headers={"Authorization": "Bearer mcp-token"})
assert resp.status_code == 428
assert resp.json()["error"] == "Provisioning required"
# ---------------------------------------------------------------------------
# create_webhook
# ---------------------------------------------------------------------------
async def test_create_webhook_uses_basic_auth(mocker):
_patch_token_validation(mocker)
_patch_basic_auth(mocker, username="bob", app_password="bob-pwd")
factory = _patch_outbound_client_factory(mocker)
_patch_webhooks_client(
mocker,
create_webhook={"id": 42, "event": "OCP\\Events\\NodeCreated"},
)
mocker.patch(
"nextcloud_mcp_server.api.webhooks.webhook_auth_pair",
return_value=("none", None),
)
client = TestClient(_build_test_app())
resp = client.post(
"/api/v1/webhooks",
headers={"Authorization": "Bearer mcp-token"},
json={
"event": "OCP\\Events\\NodeCreated",
"uri": "http://mcp:8000/webhooks/nextcloud",
},
)
assert resp.status_code == 200
assert resp.json() == {"webhook": {"id": 42, "event": "OCP\\Events\\NodeCreated"}}
_assert_basic_auth_not_bearer(factory)
async def test_create_webhook_validates_required_fields(mocker):
_patch_token_validation(mocker)
_patch_basic_auth(mocker)
client = TestClient(_build_test_app())
resp = client.post(
"/api/v1/webhooks",
headers={"Authorization": "Bearer mcp-token"},
json={"event": "X"}, # missing uri
)
assert resp.status_code == 400
async def test_create_webhook_returns_428_when_unprovisioned(mocker):
_patch_token_validation(mocker)
mocker.patch(
"nextcloud_mcp_server.api.webhooks.get_basic_auth_for_user",
new=AsyncMock(side_effect=ProvisioningRequiredError("not provisioned")),
)
client = TestClient(_build_test_app())
resp = client.post(
"/api/v1/webhooks",
headers={"Authorization": "Bearer mcp-token"},
json={"event": "X", "uri": "http://x"},
)
assert resp.status_code == 428
assert resp.json()["error"] == "Provisioning required"
# ---------------------------------------------------------------------------
# delete_webhook
# ---------------------------------------------------------------------------
async def test_delete_webhook_uses_basic_auth(mocker):
_patch_token_validation(mocker)
_patch_basic_auth(mocker, username="carol", app_password="carol-pwd")
factory = _patch_outbound_client_factory(mocker)
_patch_webhooks_client(mocker, delete_webhook=None)
client = TestClient(_build_test_app())
resp = client.delete(
"/api/v1/webhooks/99", headers={"Authorization": "Bearer mcp-token"}
)
assert resp.status_code == 200
assert resp.json() == {"success": True, "message": "Webhook deleted"}
_assert_basic_auth_not_bearer(factory)
async def test_delete_webhook_rejects_non_integer_id(mocker):
_patch_token_validation(mocker)
_patch_basic_auth(mocker)
client = TestClient(_build_test_app())
resp = client.delete(
"/api/v1/webhooks/notanumber",
headers={"Authorization": "Bearer mcp-token"},
)
assert resp.status_code == 400
async def test_delete_webhook_returns_428_when_unprovisioned(mocker):
_patch_token_validation(mocker)
mocker.patch(
"nextcloud_mcp_server.api.webhooks.get_basic_auth_for_user",
new=AsyncMock(side_effect=ProvisioningRequiredError("not provisioned")),
)
client = TestClient(_build_test_app())
resp = client.delete(
"/api/v1/webhooks/99",
headers={"Authorization": "Bearer mcp-token"},
)
assert resp.status_code == 428
assert resp.json()["error"] == "Provisioning required"
# ---------------------------------------------------------------------------
# get_basic_auth_for_user helper
# ---------------------------------------------------------------------------
async def test_get_basic_auth_for_user_resolves_username_and_password(mocker):
"""Helper returns the stored Nextcloud username (not the OAuth user_id)
when one was recorded at provisioning time."""
from nextcloud_mcp_server.api._auth import get_basic_auth_for_user
storage = MagicMock()
storage.get_app_password_with_scopes = AsyncMock(
return_value={
"app_password": "encrypted-then-decrypted-pwd",
"scopes": ["notes.read"],
"username": "nc-username",
"created_at": "2026-01-01",
"updated_at": "2026-01-02",
}
)
mocker.patch(
"nextcloud_mcp_server.api._auth.get_shared_storage",
new=AsyncMock(return_value=storage),
)
username, password = await get_basic_auth_for_user("idp-user-id")
assert username == "nc-username"
assert password == "encrypted-then-decrypted-pwd"
async def test_get_basic_auth_for_user_falls_back_to_user_id(mocker):
"""When ``username`` is null in storage, the helper falls back to the
OAuth-issued user_id."""
from nextcloud_mcp_server.api._auth import get_basic_auth_for_user
storage = MagicMock()
storage.get_app_password_with_scopes = AsyncMock(
return_value={
"app_password": "pwd",
"scopes": None,
"username": None,
"created_at": "2026-01-01",
"updated_at": "2026-01-01",
}
)
mocker.patch(
"nextcloud_mcp_server.api._auth.get_shared_storage",
new=AsyncMock(return_value=storage),
)
username, _ = await get_basic_auth_for_user("admin")
assert username == "admin"
async def test_get_basic_auth_for_user_raises_when_unprovisioned(mocker):
from nextcloud_mcp_server.api._auth import get_basic_auth_for_user
storage = MagicMock()
storage.get_app_password_with_scopes = AsyncMock(return_value=None)
mocker.patch(
"nextcloud_mcp_server.api._auth.get_shared_storage",
new=AsyncMock(return_value=storage),
)
with pytest.raises(ProvisioningRequiredError):
await get_basic_auth_for_user("admin")