refactor(webhooks): address PR review on auth-pass
- webhook_receiver: always run hmac.compare_digest (drop the `not provided or` short-circuit) so the constant-time path is taken regardless of whether the Authorization header is present. - client/webhooks: modernise the new `auth_data` type hint to `dict[str, str] | None` per CLAUDE.md. - tests/client: rename `test_create_webhook_with_auth_headers` → `test_create_webhook_with_static_headers` and use `auth_method="header"` (NC's webhook_listeners only supports "none" and "header"; the previous "bearer" value was invalid). - auth/webhook_routes: extract `_register_preset_webhooks` from `enable_webhook_preset` so the auth-threading behaviour is testable without standing up a Starlette app + auth middleware. - tests/unit: new test_webhook_routes_register covering the helper with secret set / unset, and verifying ids round-trip in order. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.7
parent
224428fca5
commit
f5f05b7c84
@@ -17,6 +17,7 @@ from nextcloud_mcp_server.client.webhooks import WebhooksClient
|
||||
from nextcloud_mcp_server.config import get_settings
|
||||
from nextcloud_mcp_server.server.webhook_presets import (
|
||||
WEBHOOK_PRESETS,
|
||||
WebhookPreset,
|
||||
filter_presets_by_installed_apps,
|
||||
get_preset,
|
||||
)
|
||||
@@ -139,6 +140,38 @@ def webhook_auth_pair() -> tuple[str, dict[str, str] | None]:
|
||||
return ("header", {"Authorization": f"Bearer {secret}"})
|
||||
|
||||
|
||||
async def _register_preset_webhooks(
|
||||
webhooks_client: WebhooksClient,
|
||||
preset: WebhookPreset,
|
||||
webhook_uri: str,
|
||||
) -> list[int]:
|
||||
"""Register every event in a preset against a single MCP webhook URI.
|
||||
|
||||
Threads the resolved ``(auth_method, auth_data)`` from
|
||||
:func:`webhook_auth_pair` onto each registration call so deliveries
|
||||
carry the configured ``Authorization`` header (when ``WEBHOOK_SECRET``
|
||||
is set) or fall through to ``authMethod="none"`` (backward-compatible
|
||||
when it's not).
|
||||
|
||||
Extracted from :func:`enable_webhook_preset` so the auth-threading
|
||||
behaviour is testable without standing up a Starlette app.
|
||||
"""
|
||||
auth_method, auth_data = webhook_auth_pair()
|
||||
registered_ids: list[int] = []
|
||||
for event_config in preset["events"]:
|
||||
webhook_data = await webhooks_client.create_webhook(
|
||||
event=event_config["event"],
|
||||
uri=webhook_uri,
|
||||
event_filter=event_config["filter"] if event_config["filter"] else None,
|
||||
auth_method=auth_method,
|
||||
auth_data=auth_data,
|
||||
)
|
||||
webhook_id = webhook_data["id"]
|
||||
registered_ids.append(webhook_id)
|
||||
logger.info("Registered webhook %s for %s", webhook_id, event_config["event"])
|
||||
return registered_ids
|
||||
|
||||
|
||||
async def _get_authenticated_client(request: Request) -> httpx.AsyncClient:
|
||||
"""Get an authenticated HTTP client for Nextcloud API calls.
|
||||
|
||||
@@ -419,20 +452,9 @@ async def enable_webhook_preset(request: Request) -> HTMLResponse:
|
||||
# Register webhooks
|
||||
webhooks_client = WebhooksClient(http_client, username)
|
||||
webhook_uri = _get_webhook_uri()
|
||||
registered_ids = []
|
||||
|
||||
auth_method, auth_data = webhook_auth_pair()
|
||||
for event_config in preset["events"]:
|
||||
webhook_data = await webhooks_client.create_webhook(
|
||||
event=event_config["event"],
|
||||
uri=webhook_uri,
|
||||
event_filter=event_config["filter"] if event_config["filter"] else None,
|
||||
auth_method=auth_method,
|
||||
auth_data=auth_data,
|
||||
)
|
||||
webhook_id = webhook_data["id"]
|
||||
registered_ids.append(webhook_id)
|
||||
logger.info(f"Registered webhook {webhook_id} for {event_config['event']}")
|
||||
registered_ids = await _register_preset_webhooks(
|
||||
webhooks_client, preset, webhook_uri
|
||||
)
|
||||
|
||||
# Persist webhook IDs to database
|
||||
storage = _get_storage(request)
|
||||
|
||||
@@ -41,7 +41,7 @@ class WebhooksClient(BaseNextcloudClient):
|
||||
http_method: str = "POST",
|
||||
auth_method: str = "none",
|
||||
headers: Optional[Dict[str, str]] = None,
|
||||
auth_data: Optional[Dict[str, str]] = None,
|
||||
auth_data: dict[str, str] | None = None,
|
||||
event_filter: Optional[Dict[str, Any]] = None,
|
||||
) -> Dict[str, Any]:
|
||||
"""Register a new webhook for the specified event.
|
||||
|
||||
@@ -53,7 +53,10 @@ async def handle_nextcloud_webhook(request: Request) -> JSONResponse:
|
||||
if secret:
|
||||
provided = request.headers.get("authorization", "")
|
||||
expected = f"Bearer {secret}"
|
||||
if not provided or not hmac.compare_digest(provided, expected):
|
||||
# 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.
|
||||
if not hmac.compare_digest(provided, expected):
|
||||
logger.warning("Webhook rejected: missing or invalid Authorization header")
|
||||
return JSONResponse(
|
||||
{"status": "unauthorized"},
|
||||
|
||||
Reference in New Issue
Block a user