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"""

Error Loading Webhooks

-

{str(e)}

+

{html.escape(str(e))}

""", 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'
Unknown preset: {preset_id}
', + content=f'
Unknown preset: {html.escape(preset_id)}
', 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'
Failed to enable preset: {str(e)}
', + content=f'
Failed to enable preset: {html.escape(str(e))}
', 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'
Unknown preset: {preset_id}
', + content=f'
Unknown preset: {html.escape(preset_id)}
', 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'
Failed to disable preset: {str(e)}
', + content=f'
Failed to disable preset: {html.escape(str(e))}
', 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 `" + + with TestClient(app) as client: + response = client.post(f"/app/webhooks/enable/{payload}") + + assert response.status_code == 404 + assert "<script>alert(1)</script>" in response.text + assert "" 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 = "" + + with TestClient(app) as client: + response = client.delete(f"/app/webhooks/disable/{payload}") + + assert response.status_code == 404 + assert "<script>alert(2)</script>" in response.text + assert "" 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("

") + + 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 "</p><script>x</script>" in response.text + assert "" not in response.text + + +def test_disable_exception_message_is_html_escaped(monkeypatch): + async def _boom(_request): + raise RuntimeError("

") + + 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 "</p><script>y</script>" in response.text + assert "" not in response.text