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

fix(webhooks): use app-password basic auth for NC API calls

Browse Source 
The webhook API endpoints in api/webhooks.py forwarded the inbound MCP
OAuth bearer token directly to Nextcloud as the Authorization header.
Per ADR-022 / docs/login-flow-v2.md the data leg from MCP server to
Nextcloud must use HTTP Basic Auth with the user's stored Login Flow v2
app password — bearer-forwarding requires upstream user_oidc patches that
were never merged and is incompatible with admin endpoints gated by
@PasswordConfirmationRequired (e.g. webhook_listeners/api/v1/webhooks,
which 401s).

PR #760 papered over the symptom for /api/v1/apps by switching to the
permissive /cloud/capabilities endpoint, but the same architectural
mistake remained on list_webhooks / create_webhook / delete_webhook,
which still 500'd on the astrolabe admin UI's preset page.

Changes:
- New helper api/_auth.py:get_basic_auth_for_user(user_id) reads the
  user's app password from encrypted storage and returns
  (username, app_password). Mirrors context.py:_get_client_from_login_flow
  but is callable from Starlette routes (no MCP Context required).
- All four endpoints in api/webhooks.py now use httpx.BasicAuth instead
  of forwarding the OAuth bearer; ProvisioningRequiredError is mapped to
  HTTP 412 so callers can render a "complete provisioning" CTA rather
  than receiving an opaque 500.
- Outbound NC requests now identify the user by the username recorded at
  Login Flow v2 provisioning time (which may differ from the IdP-issued
  user_id) — flowed into WebhooksClient and used for logging.

Tests:
- tests/unit/test_management_apps_endpoint.py: assertions updated to
  verify outbound NC request uses BasicAuth and carries no Authorization
  header. Replaced "missing-Authorization → 500" test with a
  ProvisioningRequiredError → 412 case.
- tests/unit/test_webhooks_api_auth.py (new): cross-endpoint coverage
  for list_webhooks, create_webhook, delete_webhook and the new helper —
  including 412 symmetry for all four endpoints.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-05-03 23:50:59 +02:00
co-authored by Claude Opus 4.7
parent beef77d785
commit a0e484d95b
4 changed files with 457 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=412,
)
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=412,
)
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=412,
)
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=412,
)
except Exception as e:
logger.error("Error deleting webhook for user %s: %s", user_id, e)
return JSONResponse(