From 4a3857aabb74d82490160aec77a9ebe49f4f4651 Mon Sep 17 00:00:00 2001
From: Chris Coutinho <chris@coutinho.io>
Date: Thu, 30 Apr 2026 14:17:45 +0200
Subject: [PATCH] fix(webhooks): escape HTML in error responses, compare bearer
 as bytes

Address the two Security findings from PR review:

- webhook_receiver: encode Authorization header and expected bearer to
  utf-8 bytes before hmac.compare_digest. Conventional form; doesn't
  rely on Python's implicit ASCII encoding.
- webhook_routes: html.escape user-influenced and exception-derived
  strings before interpolating into HTMLResponse content. Covers the
  preset_id path param echoed in the "Unknown preset" branch and the
  str(e) text rendered on handler exceptions.

Adds regression tests verifying compare_digest is invoked on bytes and
that <script> payloads (in preset_id and exception messages) are
emitted as escaped entities, not active markup.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 nextcloud_mcp_server/auth/webhook_routes.py   |  11 +-
 .../vector/webhook_receiver.py                |  11 +-
 tests/unit/test_webhook_endpoint.py           |  25 ++++
 tests/unit/test_webhook_routes_xss.py         | 133 ++++++++++++++++++
 4 files changed, 170 insertions(+), 10 deletions(-)
 create mode 100644 tests/unit/test_webhook_routes_xss.py

diff --git a/nextcloud_mcp_server/auth/webhook_routes.py b/nextcloud_mcp_server/auth/webhook_routes.py
index 5fcfb7cd..360a101d 100644
--- a/nextcloud_mcp_server/auth/webhook_routes.py
+++ b/nextcloud_mcp_server/auth/webhook_routes.py
@@ -4,6 +4,7 @@ Provides browser-based endpoints for admin users to manage webhook configuration
 using preset templates. Only accessible to Nextcloud administrators.
 """
 
+import html
 import logging
 import os
 
@@ -411,7 +412,7 @@ async def webhook_management_pane(request: Request) -> HTMLResponse:
             content=f"""
             <div class="warning">
                 <p><strong>Error Loading Webhooks</strong></p>
-                <p>{str(e)}</p>
+                <p>{html.escape(str(e))}</p>
             </div>
             """,
             status_code=500,
@@ -447,7 +448,7 @@ async def enable_webhook_preset(request: Request) -> HTMLResponse:
         preset = get_preset(preset_id)
         if not preset:
             return HTMLResponse(
-                content=f'<div class="warning">Unknown preset: {preset_id}</div>',
+                content=f'<div class="warning">Unknown preset: {html.escape(preset_id)}</div>',
                 status_code=404,
             )
 
@@ -500,7 +501,7 @@ async def enable_webhook_preset(request: Request) -> HTMLResponse:
     except Exception as e:
         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>',
+            content=f'<div class="warning">Failed to enable preset: {html.escape(str(e))}</div>',
             status_code=500,
         )
 
@@ -534,7 +535,7 @@ async def disable_webhook_preset(request: Request) -> HTMLResponse:
         preset = get_preset(preset_id)
         if not preset:
             return HTMLResponse(
-                content=f'<div class="warning">Unknown preset: {preset_id}</div>',
+                content=f'<div class="warning">Unknown preset: {html.escape(preset_id)}</div>',
                 status_code=404,
             )
 
@@ -592,6 +593,6 @@ async def disable_webhook_preset(request: Request) -> HTMLResponse:
     except Exception as e:
         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>',
+            content=f'<div class="warning">Failed to disable preset: {html.escape(str(e))}</div>',
             status_code=500,
         )
diff --git a/nextcloud_mcp_server/vector/webhook_receiver.py b/nextcloud_mcp_server/vector/webhook_receiver.py
index fc630cd5..1c25c19b 100644
--- a/nextcloud_mcp_server/vector/webhook_receiver.py
+++ b/nextcloud_mcp_server/vector/webhook_receiver.py
@@ -52,12 +52,13 @@ async def handle_nextcloud_webhook(request: Request) -> JSONResponse:
     """
     secret = get_settings().webhook_secret
     if secret:
-        provided = request.headers.get("authorization", "")
-        expected = f"Bearer {secret}"
+        provided = request.headers.get("authorization", "").encode("utf-8")
+        expected = f"Bearer {secret}".encode("utf-8")
         # 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.
+        # of `==`. Comparing as bytes is the conventional form and avoids any
+        # surprise with non-ASCII input. 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(
diff --git a/tests/unit/test_webhook_endpoint.py b/tests/unit/test_webhook_endpoint.py
index 4704e9e9..cd1b05fb 100644
--- a/tests/unit/test_webhook_endpoint.py
+++ b/tests/unit/test_webhook_endpoint.py
@@ -272,3 +272,28 @@ def test_secret_unset_accepts_unauthenticated(monkeypatch):
 
     assert response.status_code == 200
     assert receive_stream.receive_nowait().doc_id == "437"
+
+
+def test_compare_digest_is_called_with_bytes(monkeypatch, mocker):
+    """Regression: secret comparison must run on bytes, not strings, so
+    that future non-ASCII secret support doesn't depend on Python's
+    implicit ASCII encoding."""
+    _patch_secret(monkeypatch, "supersecret")
+    spy = mocker.spy(webhook_receiver.hmac, "compare_digest")
+
+    send_stream, _receive = anyio.create_memory_object_stream(max_buffer_size=4)
+    app = _make_app(send_stream=send_stream)
+
+    with TestClient(app) as client:
+        response = client.post(
+            "/webhooks/nextcloud",
+            json=_NOTE_CREATED,
+            headers={"Authorization": "Bearer supersecret"},
+        )
+
+    assert response.status_code == 200
+    assert spy.call_count == 1
+    provided_arg, expected_arg = spy.call_args.args
+    assert isinstance(provided_arg, bytes)
+    assert isinstance(expected_arg, bytes)
+    assert expected_arg == b"Bearer supersecret"
diff --git a/tests/unit/test_webhook_routes_xss.py b/tests/unit/test_webhook_routes_xss.py
new file mode 100644
index 00000000..5bb1d7a1
--- /dev/null
+++ b/tests/unit/test_webhook_routes_xss.py
@@ -0,0 +1,133 @@
+"""Unit tests verifying that user-influenced and exception-derived strings
+are HTML-escaped before they are rendered into ``HTMLResponse`` content.
+
+The handlers under test are decorated with ``@requires("authenticated")``,
+so we install an ``AuthenticationMiddleware`` backed by a trivial backend
+that always reports the request as authenticated. We then monkeypatch the
+internal helpers (``_get_authenticated_client``, ``is_nextcloud_admin``,
+``get_preset``) to drive the handler down the specific code path we want
+to exercise.
+"""
+
+import pytest
+from starlette.applications import Starlette
+from starlette.authentication import (
+    AuthCredentials,
+    AuthenticationBackend,
+    SimpleUser,
+)
+from starlette.middleware import Middleware
+from starlette.middleware.authentication import AuthenticationMiddleware
+from starlette.routing import Route
+from starlette.testclient import TestClient
+
+from nextcloud_mcp_server.auth import webhook_routes
+from nextcloud_mcp_server.auth.webhook_routes import (
+    disable_webhook_preset,
+    enable_webhook_preset,
+)
+
+pytestmark = pytest.mark.unit
+
+
+class _AlwaysAuthBackend(AuthenticationBackend):
+    async def authenticate(self, conn):
+        return AuthCredentials(["authenticated"]), SimpleUser("testuser")
+
+
+def _make_app() -> Starlette:
+    return Starlette(
+        routes=[
+            Route(
+                "/app/webhooks/enable/{preset_id:path}",
+                enable_webhook_preset,
+                methods=["POST"],
+            ),
+            Route(
+                "/app/webhooks/disable/{preset_id:path}",
+                disable_webhook_preset,
+                methods=["DELETE"],
+            ),
+        ],
+        middleware=[Middleware(AuthenticationMiddleware, backend=_AlwaysAuthBackend())],
+    )
+
+
+def _stub_admin_path(monkeypatch):
+    """Make the handler progress past auth/admin checks without real I/O."""
+
+    async def _fake_client(_request):
+        return object()  # never actually used because get_preset returns None
+
+    async def _fake_is_admin(_request, _client):
+        return True
+
+    monkeypatch.setattr(webhook_routes, "_get_authenticated_client", _fake_client)
+    monkeypatch.setattr(webhook_routes, "is_nextcloud_admin", _fake_is_admin)
+
+
+def test_enable_unknown_preset_id_is_html_escaped(monkeypatch):
+    """A `<script>` tag in the preset_id path param must be rendered as
+    escaped text, not active markup."""
+    _stub_admin_path(monkeypatch)
+    monkeypatch.setattr(webhook_routes, "get_preset", lambda _id: None)
+
+    app = _make_app()
+    payload = "<script>alert(1)</script>"
+
+    with TestClient(app) as client:
+        response = client.post(f"/app/webhooks/enable/{payload}")
+
+    assert response.status_code == 404
+    assert "&lt;script&gt;alert(1)&lt;/script&gt;" in response.text
+    assert "<script>alert(1)</script>" not in response.text
+
+
+def test_disable_unknown_preset_id_is_html_escaped(monkeypatch):
+    _stub_admin_path(monkeypatch)
+    monkeypatch.setattr(webhook_routes, "get_preset", lambda _id: None)
+
+    app = _make_app()
+    payload = "<script>alert(2)</script>"
+
+    with TestClient(app) as client:
+        response = client.delete(f"/app/webhooks/disable/{payload}")
+
+    assert response.status_code == 404
+    assert "&lt;script&gt;alert(2)&lt;/script&gt;" in response.text
+    assert "<script>alert(2)</script>" not in response.text
+
+
+def test_enable_exception_message_is_html_escaped(monkeypatch):
+    """If the handler raises, the exception text must be escaped before
+    it lands in the 500 response body."""
+
+    async def _boom(_request):
+        raise RuntimeError("</p><script>x</script>")
+
+    monkeypatch.setattr(webhook_routes, "_get_authenticated_client", _boom)
+
+    app = _make_app()
+
+    with TestClient(app) as client:
+        response = client.post("/app/webhooks/enable/notes_sync")
+
+    assert response.status_code == 500
+    assert "&lt;/p&gt;&lt;script&gt;x&lt;/script&gt;" in response.text
+    assert "<script>x</script>" not in response.text
+
+
+def test_disable_exception_message_is_html_escaped(monkeypatch):
+    async def _boom(_request):
+        raise RuntimeError("</p><script>y</script>")
+
+    monkeypatch.setattr(webhook_routes, "_get_authenticated_client", _boom)
+
+    app = _make_app()
+
+    with TestClient(app) as client:
+        response = client.delete("/app/webhooks/disable/notes_sync")
+
+    assert response.status_code == 500
+    assert "&lt;/p&gt;&lt;script&gt;y&lt;/script&gt;" in response.text
+    assert "<script>y</script>" not in response.text