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

refactor(webhooks): bound queue waits, route URLs through dynaconf

Browse Source 
Addresses round-3 review feedback on PR #747:

- webhook_receiver: wrap send_stream.send() in anyio.fail_after(1.0)
  and return 503 with reason="queue full" if the queue is saturated.
  Avoids pinning the handler until NC's outbound timeout fires; the
  503 retry contract is the same as the existing "sync not running"
  branch.
- webhook_receiver: revise the compare_digest comment to match what
  the function actually guarantees — it avoids the per-character
  short-circuit of `==` but is not fully constant-time across length
  differences.
- _get_webhook_uri: read WEBHOOK_INTERNAL_URL and
  NEXTCLOUD_MCP_SERVER_URL via dynaconf so operators using
  settings.toml (rather than env vars) aren't silently routed into
  the docker/localhost fallback. Adds webhook_internal_url to
  Settings/_DEFAULTS/_field_map; nextcloud_mcp_server_url already
  existed. Docker-detection markers stay on os.getenv since they're
  container-runtime signals, not user-facing config.
- webhook_routes: sweep remaining f-string logger calls to lazy %s
  formatting per CLAUDE.md.
- client/webhooks: modernise full file's type hints to
  dict / list / | None per CLAUDE.md.

Tests:
- New test_returns_503_when_queue_is_full exercises the timeout
  branch with a saturated buffer and a shortened deadline.
- test_webhook_uri tests now patch get_settings (matching the
  auth-pair tests in the same file) instead of monkeypatching env
  vars directly.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-04-30 04:17:00 +02:00
co-authored by Claude Opus 4.7
parent 42675e7c20
commit c1368b9a7f
6 changed files with 122 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nextcloud_mcp_server/auth/webhook_routes.py
+25 -19
View File
@@ -72,7 +72,7 @@ async def _get_installed_apps(http_client: httpx.AsyncClient) -> list[str]:
app_keys = set(capabilities.keys()) - core_keys
return sorted(app_keys)
except Exception as e:
logger.warning(f"Failed to get installed apps from capabilities: {e}")
logger.warning("Failed to get installed apps from capabilities: %s", e)
return []
@@ -81,10 +81,10 @@ def _get_webhook_uri() -> str:
Priority (highest first):
1. ``WEBHOOK_INTERNAL_URL`` — explicit override, e.g. for split
internal/external URLs.
2. ``NEXTCLOUD_MCP_SERVER_URL`` — the configured public URL. This is
set on cloud deployments (ECS, k8s) and is the URL Nextcloud must
POST to.
internal/external URLs (read via dynaconf, so env vars and
settings.toml both work).
2. ``NEXTCLOUD_MCP_SERVER_URL`` — the configured public URL set on
cloud deployments (ECS, k8s); the URL NC must POST to.
3. ``/.dockerenv`` (or podman / ``DOCKER_CONTAINER=true``) → internal
docker-compose service name. Only relevant when no public URL is
configured — i.e. local dev where MCP and NC share a Docker
@@ -96,14 +96,16 @@ def _get_webhook_uri() -> str:
docker-compose hostname (e.g. ``http://mcp:8000``) that NC cannot
resolve, dropping every webhook delivery.
"""
webhook_url = os.getenv("WEBHOOK_INTERNAL_URL")
if webhook_url:
return f"{webhook_url}/webhooks/nextcloud"
settings = get_settings()
if settings.webhook_internal_url:
return f"{settings.webhook_internal_url}/webhooks/nextcloud"
server_url = os.getenv("NEXTCLOUD_MCP_SERVER_URL")
if server_url:
return f"{server_url}/webhooks/nextcloud"
if settings.nextcloud_mcp_server_url:
return f"{settings.nextcloud_mcp_server_url}/webhooks/nextcloud"
# Docker-environment markers stay on os.getenv: they're container-runtime
# signals (filesystem markers, optional service-name override) rather
# than user-facing config that would belong in settings.toml.
is_docker = (
os.path.exists("/.dockerenv")
or os.path.exists("/run/.containerenv")
@@ -284,7 +286,7 @@ async def _get_enabled_presets(
return enabled_presets
except Exception as e:
logger.error(f"Failed to list webhooks: {e}")
logger.error("Failed to list webhooks: %s", e)
return {}
@@ -329,7 +331,7 @@ async def webhook_management_pane(request: Request) -> HTMLResponse:
# Get installed apps to filter presets
installed_apps = await _get_installed_apps(http_client)
logger.debug(f"Installed apps: {installed_apps}")
logger.debug("Installed apps: %s", installed_apps)
# Get currently enabled presets (from database or API)
enabled_presets = await _get_enabled_presets(webhooks_client, storage)
@@ -404,7 +406,7 @@ async def webhook_management_pane(request: Request) -> HTMLResponse:
return HTMLResponse(content=html_content)
except Exception as e:
logger.error(f"Error loading webhook management pane: {e}", exc_info=True)
logger.error("Error loading webhook management pane: %s", e, exc_info=True)
return HTMLResponse(
content=f"""
<div class="warning">
@@ -462,7 +464,9 @@ async def enable_webhook_preset(request: Request) -> HTMLResponse:
for webhook_id in registered_ids:
await storage.store_webhook(webhook_id, preset_id)
logger.info(
f"Persisted {len(registered_ids)} webhook(s) for preset '{preset_id}' to database"
"Persisted %d webhook(s) for preset '%s' to database",
len(registered_ids),
preset_id,
)
# Return updated card
@@ -494,7 +498,7 @@ async def enable_webhook_preset(request: Request) -> HTMLResponse:
)
except Exception as e:
logger.error(f"Failed to enable preset {preset_id}: {e}", exc_info=True)
logger.error("Failed to enable preset %s: %s", preset_id, e, exc_info=True)
return HTMLResponse(
content=f'<div class="warning">Failed to enable preset: {str(e)}</div>',
status_code=500,
@@ -548,13 +552,15 @@ async def disable_webhook_preset(request: Request) -> HTMLResponse:
for webhook_id in webhook_ids:
await webhooks_client.delete_webhook(webhook_id)
logger.info(f"Deleted webhook {webhook_id} from preset {preset_id}")
logger.info("Deleted webhook %s from preset %s", webhook_id, preset_id)
# Remove from database
if storage:
deleted_count = await storage.clear_preset_webhooks(preset_id)
logger.info(
f"Removed {deleted_count} webhook(s) for preset '{preset_id}' from database"
"Removed %d webhook(s) for preset '%s' from database",
deleted_count,
preset_id,
)
# Return updated card
@@ -584,7 +590,7 @@ async def disable_webhook_preset(request: Request) -> HTMLResponse:
)
except Exception as e:
logger.error(f"Failed to disable preset {preset_id}: {e}", exc_info=True)
logger.error("Failed to disable preset %s: %s", preset_id, e, exc_info=True)
return HTMLResponse(
content=f'<div class="warning">Failed to disable preset: {str(e)}</div>',
status_code=500,
nextcloud_mcp_server/client/webhooks.py
+9 -9
View File
@@ -1,6 +1,6 @@
"""Client for Nextcloud Webhook Listeners API operations."""
from typing import Any, Dict, List, Optional
from typing import Any
from nextcloud_mcp_server.client.base import BaseNextcloudClient
@@ -11,15 +11,15 @@ class WebhooksClient(BaseNextcloudClient):
app_name = "webhooks"
def _get_webhook_headers(
self, additional_headers: Optional[Dict[str, str]] = None
) -> Dict[str, str]:
self, additional_headers: dict[str, str] | None = None
) -> dict[str, str]:
"""Get standard headers required for Webhook Listeners API calls."""
headers = {"OCS-APIRequest": "true", "Accept": "application/json"}
if additional_headers:
headers.update(additional_headers)
return headers
async def list_webhooks(self) -> List[Dict[str, Any]]:
async def list_webhooks(self) -> list[dict[str, Any]]:
"""List all registered webhooks for the current user.
Returns:
@@ -40,10 +40,10 @@ class WebhooksClient(BaseNextcloudClient):
uri: str,
http_method: str = "POST",
auth_method: str = "none",
headers: Optional[Dict[str, str]] = None,
headers: dict[str, str] | None = None,
auth_data: dict[str, str] | None = None,
event_filter: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
event_filter: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Register a new webhook for the specified event.
Args:
@@ -63,7 +63,7 @@ class WebhooksClient(BaseNextcloudClient):
Returns:
Webhook registration details including webhook ID
"""
data: Dict[str, Any] = {
data: dict[str, Any] = {
"httpMethod": http_method,
"uri": uri,
"event": event,
@@ -101,7 +101,7 @@ class WebhooksClient(BaseNextcloudClient):
headers=headers,
)
async def get_webhook(self, webhook_id: int) -> Dict[str, Any]:
async def get_webhook(self, webhook_id: int) -> dict[str, Any]:
"""Get details of a specific webhook registration.
Args:
nextcloud_mcp_server/config.py
+8
View File
@@ -57,6 +57,9 @@ _DEFAULTS: dict[str, Any] = {
# tell NC to add `Authorization: Bearer <secret>` to webhook deliveries
# and the receiver rejects unauthenticated requests.
"webhook_secret": None,
# Internal URL override for webhook registration; wins over
# NEXTCLOUD_MCP_SERVER_URL when set (e.g. split internal/external URLs).
"webhook_internal_url": None,
# Vector sync
"vector_sync_scan_interval": 300,
"vector_sync_processor_workers": 3,
@@ -440,6 +443,10 @@ class Settings:
# delivery. When unset, registration uses authMethod="none" and the
# receiver accepts unauthenticated POSTs (backward-compatible).
webhook_secret: str | None = None
# Internal URL override for webhook registration. Highest-priority
# source for the URL we register with NC (above
# nextcloud_mcp_server_url and the docker-detection fallback).
webhook_internal_url: str | None = None
# Vector sync settings (ADR-007)
vector_sync_enabled: bool = False
@@ -780,6 +787,7 @@ def get_settings() -> Settings:
"token_storage_db": "TOKEN_STORAGE_DB",
# Webhook auth (ADR-010)
"webhook_secret": "WEBHOOK_SECRET",
"webhook_internal_url": "WEBHOOK_INTERNAL_URL",
# Vector sync settings (ADR-007)
"vector_sync_scan_interval": "VECTOR_SYNC_SCAN_INTERVAL",
"vector_sync_processor_workers": "VECTOR_SYNC_PROCESSOR_WORKERS",
nextcloud_mcp_server/vector/webhook_receiver.py
+20 -4
View File
@@ -8,6 +8,7 @@ in :mod:`nextcloud_mcp_server.app`.
import hmac
import logging
import anyio
from starlette.requests import Request
from starlette.responses import JSONResponse
@@ -53,9 +54,10 @@ async def handle_nextcloud_webhook(request: Request) -> JSONResponse:
if secret:
provided = request.headers.get("authorization", "")
expected = f"Bearer {secret}"
# Always run compare_digest so the constant-time path is taken even
# when the header is missing — `compare_digest("", expected)` returns
# False without leaking length information.
# Use compare_digest to avoid the character-by-character short-circuit
# of `==`. compare_digest still returns False for differing lengths
# but isn't fully constant-time across them; that's fine here — a
# secret length leak is not a sensitive signal.
if not hmac.compare_digest(provided, expected):
logger.warning("Webhook rejected: missing or invalid Authorization header")
return JSONResponse(
@@ -94,7 +96,21 @@ async def handle_nextcloud_webhook(request: Request) -> JSONResponse:
)
try:
await send_stream.send(task)
with anyio.fail_after(1.0):
await send_stream.send(task)
except TimeoutError:
# Queue is saturated (default 10 000 tasks). Returning 503 lets NC
# retry rather than pinning this handler until its outbound timeout
# fires; the queue-pressure signal also surfaces in metrics.
logger.warning(
"Webhook task drop: queue full for %s_%s",
task.doc_type,
task.doc_id,
)
return JSONResponse(
{"status": "unavailable", "reason": "queue full"},
status_code=503,
)
except Exception as e:
logger.error(
"Failed to queue webhook task for %s_%s: %s",
tests/unit/test_webhook_endpoint.py
+27
View File
@@ -165,6 +165,33 @@ def test_returns_500_when_stream_is_closed():
assert response.json()["status"] == "error"
def test_returns_503_when_queue_is_full(monkeypatch):
"""When the processor queue is saturated, the handler must time out
quickly with 503 instead of pinning until NC's outbound timeout fires."""
# Speed up the test — a 1s deadline matches production but is overkill
# for a unit test that's specifically exercising the timeout branch.
# Capture the original BEFORE patching so the override doesn't recurse
# into itself (the receiver imports the same anyio module object).
real_fail_after = anyio.fail_after
monkeypatch.setattr(
"nextcloud_mcp_server.vector.webhook_receiver.anyio.fail_after",
lambda _seconds: real_fail_after(0.05),
)
# Buffer of 1, no consumer → first send fills it, second blocks.
send_stream, _receive_stream = anyio.create_memory_object_stream(max_buffer_size=1)
send_stream.send_nowait("sentinel") # type: ignore[arg-type]
app = _make_app(send_stream=send_stream)
with TestClient(app) as client:
response = client.post("/webhooks/nextcloud", json=_NOTE_CREATED)
assert response.status_code == 503
body = response.json()
assert body["status"] == "unavailable"
assert body["reason"] == "queue full"
# --- WEBHOOK_SECRET authentication ---------------------------------------
tests/unit/test_webhook_uri.py
+33 -20
View File
@@ -1,9 +1,13 @@
"""Unit tests for ``_get_webhook_uri`` priority order and the
``webhook_auth_kwargs`` registration helper.
``webhook_auth_pair`` registration helper.
Cloud deployments register the webhook URI returned by this function with
Nextcloud. ECS Fargate also exposes ``/.dockerenv``, so an explicit public
URL must win over the docker auto-detection branch.
URL must win over the docker auto-detection branch. The URL fields are
read via dynaconf (``Settings``), so tests patch ``get_settings`` directly.
The docker-detection markers (``/.dockerenv``, ``DOCKER_CONTAINER``,
``NEXTCLOUD_MCP_SERVICE_NAME``, ``NEXTCLOUD_MCP_PORT``) remain on
``os.getenv`` and are exercised via env-var monkeypatching.
"""
import pytest
@@ -15,9 +19,7 @@ from nextcloud_mcp_server.auth.webhook_routes import (
)
from nextcloud_mcp_server.config import Settings
ENV_VARS = (
"WEBHOOK_INTERNAL_URL",
"NEXTCLOUD_MCP_SERVER_URL",
DOCKER_ENV_VARS = (
"NEXTCLOUD_MCP_SERVICE_NAME",
"NEXTCLOUD_MCP_PORT",
"DOCKER_CONTAINER",
@@ -26,10 +28,21 @@ ENV_VARS = (
@pytest.fixture(autouse=True)
def _clean_env(monkeypatch):
for name in ENV_VARS:
for name in DOCKER_ENV_VARS:
monkeypatch.delenv(name, raising=False)
def _patch_settings(monkeypatch, **overrides) -> None:
"""Make ``get_settings()`` (as called inside webhook_routes) return a
Settings instance with the given URL/secret fields set; everything else
falls back to the dataclass defaults."""
monkeypatch.setattr(
webhook_routes,
"get_settings",
lambda: Settings(**overrides),
)
def _no_docker_markers(monkeypatch):
monkeypatch.setattr(
"nextcloud_mcp_server.auth.webhook_routes.os.path.exists",
@@ -46,8 +59,11 @@ def _docker_markers(monkeypatch):
@pytest.mark.unit
def test_webhook_internal_url_wins_over_everything(monkeypatch):
monkeypatch.setenv("WEBHOOK_INTERNAL_URL", "https://internal.example.com")
monkeypatch.setenv("NEXTCLOUD_MCP_SERVER_URL", "https://public.example.com")
_patch_settings(
monkeypatch,
webhook_internal_url="https://internal.example.com",
nextcloud_mcp_server_url="https://public.example.com",
)
_docker_markers(monkeypatch)
assert _get_webhook_uri() == "https://internal.example.com/webhooks/nextcloud"
@@ -57,8 +73,9 @@ def test_webhook_internal_url_wins_over_everything(monkeypatch):
def test_public_url_wins_over_docker_detection(monkeypatch):
"""The bug-fix case: ECS containers have /.dockerenv but a public URL is
set. Docker auto-detection must NOT clobber the explicit public URL."""
monkeypatch.setenv(
"NEXTCLOUD_MCP_SERVER_URL", "https://holy-bluegill.astrolabecloud.com"
_patch_settings(
monkeypatch,
nextcloud_mcp_server_url="https://holy-bluegill.astrolabecloud.com",
)
_docker_markers(monkeypatch)
@@ -72,6 +89,7 @@ def test_public_url_wins_over_docker_detection(monkeypatch):
def test_docker_detection_used_when_no_public_url(monkeypatch):
"""docker-compose dev: no public URL set, /.dockerenv exists → use the
docker-compose service name."""
_patch_settings(monkeypatch)
_docker_markers(monkeypatch)
assert _get_webhook_uri() == "http://mcp:8000/webhooks/nextcloud"
@@ -79,6 +97,7 @@ def test_docker_detection_used_when_no_public_url(monkeypatch):
@pytest.mark.unit
def test_docker_detection_honors_service_name_and_port_overrides(monkeypatch):
_patch_settings(monkeypatch)
monkeypatch.setenv("NEXTCLOUD_MCP_SERVICE_NAME", "mcp-login-flow")
monkeypatch.setenv("NEXTCLOUD_MCP_PORT", "8004")
_docker_markers(monkeypatch)
@@ -88,6 +107,7 @@ def test_docker_detection_honors_service_name_and_port_overrides(monkeypatch):
@pytest.mark.unit
def test_docker_container_env_var_triggers_docker_branch(monkeypatch):
_patch_settings(monkeypatch)
monkeypatch.setenv("DOCKER_CONTAINER", "true")
_no_docker_markers(monkeypatch)
@@ -96,6 +116,7 @@ def test_docker_container_env_var_triggers_docker_branch(monkeypatch):
@pytest.mark.unit
def test_localhost_fallback_when_nothing_set(monkeypatch):
_patch_settings(monkeypatch)
_no_docker_markers(monkeypatch)
assert _get_webhook_uri() == "http://localhost:8000/webhooks/nextcloud"
@@ -104,23 +125,15 @@ def test_localhost_fallback_when_nothing_set(monkeypatch):
# --- webhook_auth_pair() --------------------------------------------------
def _patch_secret(monkeypatch, secret: str | None) -> None:
monkeypatch.setattr(
webhook_routes,
"get_settings",
lambda: Settings(webhook_secret=secret),
)
@pytest.mark.unit
def test_auth_pair_returns_none_when_secret_unset(monkeypatch):
_patch_secret(monkeypatch, None)
_patch_settings(monkeypatch, webhook_secret=None)
assert webhook_auth_pair() == ("none", None)
@pytest.mark.unit
def test_auth_pair_emits_bearer_header_when_secret_set(monkeypatch):
_patch_secret(monkeypatch, "supersecret")
_patch_settings(monkeypatch, webhook_secret="supersecret")
assert webhook_auth_pair() == (
"header",
{"Authorization": "Bearer supersecret"},