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>
This commit is contained in:
Chris Coutinho
2026-04-30 14:17:45 +02:00
co-authored by Claude Opus 4.7
parent c1368b9a7f
commit 4a3857aabb
4 changed files with 170 additions and 10 deletions
+6 -5
View File
@@ -4,6 +4,7 @@ Provides browser-based endpoints for admin users to manage webhook configuration
using preset templates. Only accessible to Nextcloud administrators. using preset templates. Only accessible to Nextcloud administrators.
""" """
import html
import logging import logging
import os import os
@@ -411,7 +412,7 @@ async def webhook_management_pane(request: Request) -> HTMLResponse:
content=f""" content=f"""
<div class="warning"> <div class="warning">
<p><strong>Error Loading Webhooks</strong></p> <p><strong>Error Loading Webhooks</strong></p>
<p>{str(e)}</p> <p>{html.escape(str(e))}</p>
</div> </div>
""", """,
status_code=500, status_code=500,
@@ -447,7 +448,7 @@ async def enable_webhook_preset(request: Request) -> HTMLResponse:
preset = get_preset(preset_id) preset = get_preset(preset_id)
if not preset: if not preset:
return HTMLResponse( 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, status_code=404,
) )
@@ -500,7 +501,7 @@ async def enable_webhook_preset(request: Request) -> HTMLResponse:
except Exception as e: except Exception as e:
logger.error("Failed to enable preset %s: %s", preset_id, e, exc_info=True) logger.error("Failed to enable preset %s: %s", preset_id, e, exc_info=True)
return HTMLResponse( 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, status_code=500,
) )
@@ -534,7 +535,7 @@ async def disable_webhook_preset(request: Request) -> HTMLResponse:
preset = get_preset(preset_id) preset = get_preset(preset_id)
if not preset: if not preset:
return HTMLResponse( 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, status_code=404,
) )
@@ -592,6 +593,6 @@ async def disable_webhook_preset(request: Request) -> HTMLResponse:
except Exception as e: except Exception as e:
logger.error("Failed to disable preset %s: %s", preset_id, e, exc_info=True) logger.error("Failed to disable preset %s: %s", preset_id, e, exc_info=True)
return HTMLResponse( 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, status_code=500,
) )
@@ -52,12 +52,13 @@ async def handle_nextcloud_webhook(request: Request) -> JSONResponse:
""" """
secret = get_settings().webhook_secret secret = get_settings().webhook_secret
if secret: if secret:
provided = request.headers.get("authorization", "") provided = request.headers.get("authorization", "").encode("utf-8")
expected = f"Bearer {secret}" expected = f"Bearer {secret}".encode("utf-8")
# Use compare_digest to avoid the character-by-character short-circuit # Use compare_digest to avoid the character-by-character short-circuit
# of `==`. compare_digest still returns False for differing lengths # of `==`. Comparing as bytes is the conventional form and avoids any
# but isn't fully constant-time across them; that's fine here — a # surprise with non-ASCII input. compare_digest still returns False
# secret length leak is not a sensitive signal. # 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): if not hmac.compare_digest(provided, expected):
logger.warning("Webhook rejected: missing or invalid Authorization header") logger.warning("Webhook rejected: missing or invalid Authorization header")
return JSONResponse( return JSONResponse(
+25
View File
@@ -272,3 +272,28 @@ def test_secret_unset_accepts_unauthenticated(monkeypatch):
assert response.status_code == 200 assert response.status_code == 200
assert receive_stream.receive_nowait().doc_id == "437" 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"
+133
View File
@@ -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