fix(security): escape reflected input in SSO, roots, SIEM, LLM chat responses (CWE-79)

mcpgateway/main.py

@@ -7902,15 +7902,15 @@ async def remove_root(
     Returns:
         Status message indicating result.
     """
-    logger.debug(f"User '{safe_log_user(user)}' requested to remove root with URI: {uri}")
+    logger.debug(f"User '{safe_log_user(user)}' requested to remove root with URI: {html.escape(str(uri))}")
     try:
         await root_service.remove_root(uri)
     except RootServiceNotFoundError as e:
-        raise HTTPException(status_code=404, detail=str(e)) from e
+        raise HTTPException(status_code=404, detail=html.escape(str(e))) from e
     except Exception as e:
-        logger.error(f"Failed to remove root {uri}: {e}")
+        logger.error(f"Failed to remove root {html.escape(str(uri))}: {e}")
         raise HTTPException(status_code=500, detail="Internal error removing root") from e
-    return {"status": "success", "message": f"Root {uri} removed"}
+    return {"status": "success", "message": f"Root {html.escape(str(uri))} removed"}
 
 
 ##################

mcpgateway/routers/llmchat_router.py

@@ -791,6 +791,8 @@ async def connect(input_data: ConnectInput, request: Request, user=Depends(get_c
         All configuration values support environment variable fallbacks.
     """
     user_id = _resolve_user_id(input_data.user_id, user)
+    # Sanitize user_id for safe output (defense against potential XSS)
+    safe_user_id = SecurityValidator.sanitize_display_text(user_id, "user_id")
 
     try:
         # Validate user_id
@@ -875,7 +877,7 @@ async def connect(input_data: ConnectInput, request: Request, user=Depends(get_c
             logger.warning(f"Failed to extract tool names: {tool_error}")
             # Continue without tools list
 
-        return {"status": "connected", "user_id": user_id, "provider": config.llm.provider, "tool_count": len(tool_names), "tools": tool_names}
+        return {"status": "connected", "user_id": safe_user_id, "provider": config.llm.provider, "tool_count": len(tool_names), "tools": tool_names}
 
     except HTTPException:
         # Re-raise HTTP exceptions as-is
@@ -1053,6 +1055,8 @@ async def chat(input_data: ChatInput, user=Depends(get_current_user_with_permiss
         Client must maintain persistent connection for streaming.
     """
     user_id = _resolve_user_id(input_data.user_id, user)
+    # Sanitize user_id for safe output (defense against potential XSS)
+    safe_user_id = SecurityValidator.sanitize_display_text(user_id, "user_id")
 
     # Validate input
     if not user_id:
@@ -1082,7 +1086,7 @@ async def chat(input_data: ChatInput, user=Depends(get_current_user_with_permiss
                 result = await chat_service.chat_with_metadata(input_data.message)
 
                 return {
-                    "user_id": user_id,
+                    "user_id": safe_user_id,
                     "response": result["text"],
                     "tool_used": result["tool_used"],
                     "tools": result["tools"],
@@ -1163,6 +1167,9 @@ async def disconnect(input_data: DisconnectInput, user=Depends(get_current_user_
     if not user_id:
         raise HTTPException(status_code=400, detail="User ID is required")
 
+    # Sanitize user_id for safe output (defense against potential XSS)
+    safe_user_id = SecurityValidator.sanitize_display_text(user_id, "user_id")
+
     # Remove and shut down chat service
     chat_service = await get_active_session(user_id)
     await delete_active_session(user_id)
@@ -1171,19 +1178,21 @@ async def disconnect(input_data: DisconnectInput, user=Depends(get_current_user_
     await delete_user_config(user_id)
 
     if not chat_service:
-        return {"status": "no_active_session", "user_id": user_id, "message": "No active session to disconnect"}
+        return {"status": "no_active_session", "user_id": safe_user_id, "message": "No active session to disconnect"}
 
     try:
         # Clear chat history on disconnect
         await chat_service.clear_history()
-        logger.info(f"Chat session disconnected for {SecurityValidator.sanitize_log_message(user_id)}")
+        logger.info(f"Chat session disconnected for {safe_user_id}")
 
         await chat_service.shutdown()
-        return {"status": "disconnected", "user_id": user_id, "message": "Successfully disconnected"}
+        return {"status": "disconnected", "user_id": safe_user_id, "message": "Successfully disconnected"}
     except Exception as e:
-        logger.error(f"Error during disconnect for user {SecurityValidator.sanitize_log_message(user_id)}: {e}", exc_info=True)
+        logger.error(f"Error during disconnect for user {safe_user_id}: {e}", exc_info=True)
         # Session already removed, so return success with warning
-        return {"status": "disconnected_with_errors", "user_id": user_id, "message": "Disconnected but cleanup encountered errors", "warning": str(e)}
+        # Sanitize exception message to prevent XSS
+        safe_warning = SecurityValidator.sanitize_display_text(str(e), "warning")
+        return {"status": "disconnected_with_errors", "user_id": safe_user_id, "message": "Disconnected but cleanup encountered errors", "warning": safe_warning}
 
 
 @llmchat_router.get("/status/{user_id}")
@@ -1228,8 +1237,10 @@ async def status(user_id: str, user=Depends(get_current_user_with_permissions)):
         only that it exists in the active_sessions dictionary.
     """
     resolved_user_id = _resolve_user_id(user_id, user)
+    # Sanitize user_id for safe output (defense against potential XSS)
+    safe_user_id = SecurityValidator.sanitize_display_text(resolved_user_id, "user_id")
     connected = bool(await get_active_session(resolved_user_id))
-    return {"user_id": resolved_user_id, "connected": connected}
+    return {"user_id": safe_user_id, "connected": connected}
 
 
 @llmchat_router.get("/config/{user_id}")

mcpgateway/routers/siem.py

@@ -7,6 +7,7 @@
 """
 
 # Standard
+import html
 import logging
 from typing import Any, Dict, List, Optional
 
@@ -98,7 +99,7 @@ async def add_siem_destination(payload: DestinationUpsertRequest, _user=Depends(
     try:
         created = await service.add_destination(payload.model_dump(exclude_none=True))
     except ValueError as exc:
-        raise HTTPException(status_code=400, detail=str(exc)) from exc
+        raise HTTPException(status_code=400, detail=html.escape(str(exc))) from exc
     except Exception as exc:
         logger.error("Failed to add SIEM destination: %s", exc)
         raise HTTPException(status_code=500, detail="Failed to add SIEM destination") from exc
@@ -128,7 +129,7 @@ async def replace_siem_destinations(payload: DestinationBulkReplaceRequest, _use
     try:
         destinations = await service.replace_destinations([item.model_dump(exclude_none=True) for item in payload.destinations])
     except ValueError as exc:
-        raise HTTPException(status_code=400, detail=str(exc)) from exc
+        raise HTTPException(status_code=400, detail=html.escape(str(exc))) from exc
     except Exception as exc:
         logger.error("Failed to replace SIEM destinations: %s", exc)
         raise HTTPException(status_code=500, detail="Failed to replace SIEM destinations") from exc
@@ -153,12 +154,14 @@ async def test_siem_destination(destination_name: str, _user=Depends(get_current
     Raises:
         HTTPException: If destination is missing or test fails unexpectedly.
     """
+    # Sanitize destination_name to prevent XSS (CWE-79)
+    sanitized_name = html.escape(destination_name)
     service = get_siem_export_service()
 
     try:
-        return await service.test_destination(destination_name)
+        return await service.test_destination(sanitized_name)
     except KeyError as exc:
-        raise HTTPException(status_code=404, detail=str(exc)) from exc
+        raise HTTPException(status_code=404, detail=html.escape(str(exc))) from exc
     except Exception as exc:
-        logger.error("Failed SIEM destination test for %s: %s", destination_name, exc)
+        logger.error("Failed SIEM destination test for %s: %s", sanitized_name, exc)
         raise HTTPException(status_code=500, detail="Failed to test SIEM destination") from exc

mcpgateway/routers/sso.py

@@ -8,6 +8,7 @@
 """
 
 # Standard
+import html
 import secrets
 from typing import Dict, List, Optional
 from urllib.parse import urlparse
@@ -731,13 +732,13 @@ async def delete_sso_provider(
     sso_service = SSOService(db)
 
     if not sso_service.delete_provider(provider_id):
-        raise HTTPException(status_code=404, detail=f"SSO provider '{provider_id}' not found")
+        raise HTTPException(status_code=404, detail=f"SSO provider '{html.escape(provider_id)}' not found")
 
     db.commit()
     db.close()
     invalidate_trusted_provider_cache()
     await invalidate_external_identity_cache()
-    return {"message": f"SSO provider '{provider_id}' deleted successfully"}
+    return {"message": f"SSO provider '{html.escape(provider_id)}' deleted successfully"}
 
 
 # ---------------------------------------------------------------------------

mcpgateway/services/siem_export_service.py

@@ -17,6 +17,7 @@
 from dataclasses import dataclass, field
 from datetime import datetime, timedelta, timezone
 import hashlib
+import html
 import hmac
 import logging
 import os
@@ -274,15 +275,15 @@ async def test_destination(self, destination_name: str) -> Dict[str, Any]:
         except Exception as exc:
             latency_ms = (time.perf_counter() - start) * 1000.0
             return {
-                "name": destination_name,
+                "name": html.escape(destination_name),
                 "status": "failed",
                 "latency_ms": round(latency_ms, 2),
                 "error": str(exc),
             }
 
         latency_ms = (time.perf_counter() - start) * 1000.0
         return {
-            "name": destination_name,
+            "name": html.escape(destination_name),
             "status": "ok",
             "latency_ms": round(latency_ms, 2),
         }

tests/unit/mcpgateway/routers/test_llmchat_router.py

@@ -597,6 +597,27 @@ async def test_status_connected(monkeypatch: pytest.MonkeyPatch):
     assert result["connected"] is True
 
 
+@pytest.mark.asyncio
+async def test_status_sanitizes_user_id_xss():
+    """Attacker-controlled user_id must be sanitized before being reflected (CWE-79)."""
+    payload = "<b>evil-user</b>"
+    result = await llmchat_router.status(payload, user={"id": payload, "email": "evil@test.com"})
+
+    assert "<b>" not in result["user_id"]
+    assert "evil-user" in result["user_id"]
+
+
+@pytest.mark.asyncio
+async def test_disconnect_sanitizes_user_id_xss():
+    """Attacker-controlled user_id must be sanitized in disconnect responses (CWE-79)."""
+    payload = "<b>evil-user</b>"
+    result = await llmchat_router.disconnect(DisconnectInput(user_id=payload), user={"id": payload, "email": "evil@test.com"})
+
+    assert result["status"] == "no_active_session"
+    assert "<b>" not in result["user_id"]
+    assert "evil-user" in result["user_id"]
+
+
 @pytest.mark.asyncio
 async def test_get_config_sanitizes(monkeypatch: pytest.MonkeyPatch):
     config = llmchat_router.build_config(

tests/unit/mcpgateway/routers/test_siem.py

@@ -145,6 +145,60 @@ async def test_test_siem_destination_internal_error(monkeypatch):
     assert exc_info.value.status_code == 500
 
 
+@pytest.mark.asyncio
+async def test_add_siem_destination_escapes_validation_error_xss(monkeypatch):
+    """Attacker-controlled input embedded in a ValueError must be HTML-escaped in the 400 detail (CWE-79)."""
+    payload_str = "<script>alert(1)</script>"
+    mock_service = MagicMock()
+    mock_service.add_destination = AsyncMock(side_effect=ValueError(f"invalid destination {payload_str}"))
+    monkeypatch.setattr(siem, "get_siem_export_service", lambda: mock_service)
+
+    payload = siem.DestinationUpsertRequest(name="dest-1", type="webhook", url="https://example.com/hook")
+
+    with pytest.raises(HTTPException) as exc_info:
+        await siem.add_siem_destination(payload=payload, _user={"email": "admin@example.com"})
+
+    assert exc_info.value.status_code == 400
+    assert payload_str not in str(exc_info.value.detail)
+    assert "&lt;script&gt;" in str(exc_info.value.detail)
+
+
+@pytest.mark.asyncio
+async def test_replace_siem_destinations_escapes_validation_error_xss(monkeypatch):
+    """Attacker-controlled input embedded in a ValueError must be HTML-escaped in the 400 detail (CWE-79)."""
+    payload_str = "<script>alert(1)</script>"
+    mock_service = MagicMock()
+    mock_service.replace_destinations = AsyncMock(side_effect=ValueError(f"invalid destination {payload_str}"))
+    monkeypatch.setattr(siem, "get_siem_export_service", lambda: mock_service)
+
+    payload = siem.DestinationBulkReplaceRequest(destinations=[siem.DestinationUpsertRequest(name="dest-1", type="webhook", url="https://example.com/hook")])
+
+    with pytest.raises(HTTPException) as exc_info:
+        await siem.replace_siem_destinations(payload=payload, _user={"email": "admin@example.com"})
+
+    assert exc_info.value.status_code == 400
+    assert payload_str not in str(exc_info.value.detail)
+    assert "&lt;script&gt;" in str(exc_info.value.detail)
+
+
+@pytest.mark.asyncio
+async def test_test_siem_destination_escapes_name_xss(monkeypatch):
+    """Attacker-controlled destination_name must be HTML-escaped before use (CWE-79)."""
+    payload_str = "<script>alert(1)</script>"
+    mock_service = MagicMock()
+    mock_service.test_destination = AsyncMock(side_effect=KeyError(f"Unknown destination: {payload_str}"))
+    monkeypatch.setattr(siem, "get_siem_export_service", lambda: mock_service)
+
+    with pytest.raises(HTTPException) as exc_info:
+        await siem.test_siem_destination(destination_name=payload_str, _user={"email": "admin@example.com"})
+
+    assert exc_info.value.status_code == 404
+    assert payload_str not in str(exc_info.value.detail)
+    assert "&lt;script&gt;" in str(exc_info.value.detail)
+    # The escaped name (not the raw payload) is what reaches the service layer.
+    mock_service.test_destination.assert_awaited_once_with("&lt;script&gt;alert(1)&lt;/script&gt;")
+
+
 # ---------------------------------------------------------------------------
 # Deny-path regression tests (unauthenticated, insufficient permissions, feature disabled)
 # ---------------------------------------------------------------------------

tests/unit/mcpgateway/routers/test_sso_router.py

@@ -952,6 +952,47 @@ def delete_provider(self, _provider_id):
     assert "deleted successfully" in result["message"]
 
 
+@pytest.mark.asyncio
+async def test_delete_sso_provider_escapes_provider_id_xss(monkeypatch: pytest.MonkeyPatch):
+    """Attacker-controlled provider_id must be HTML-escaped in reflected responses (CWE-79)."""
+
+    class DummyService:
+        def __init__(self, _db):
+            pass
+
+        def delete_provider(self, _provider_id):
+            return False
+
+    monkeypatch.setattr(sso_router, "SSOService", DummyService)
+
+    payload = "<script>alert(1)</script>"
+    with pytest.raises(HTTPException) as excinfo:
+        await sso_router.delete_sso_provider(payload, db=MagicMock(), user={"email": "admin@example.com"})
+
+    assert excinfo.value.status_code == 404
+    assert payload not in str(excinfo.value.detail)
+    assert "&lt;script&gt;" in str(excinfo.value.detail)
+
+
+@pytest.mark.asyncio
+async def test_delete_sso_provider_success_escapes_provider_id_xss(monkeypatch: pytest.MonkeyPatch):
+    """Attacker-controlled provider_id must be HTML-escaped in the success message (CWE-79)."""
+
+    class DummyService:
+        def __init__(self, _db):
+            pass
+
+        def delete_provider(self, _provider_id):
+            return True
+
+    monkeypatch.setattr(sso_router, "SSOService", DummyService)
+
+    payload = "<script>alert(1)</script>"
+    result = await sso_router.delete_sso_provider(payload, db=MagicMock(), user={"email": "admin@example.com"})
+    assert payload not in result["message"]
+    assert "&lt;script&gt;" in result["message"]
+
+
 @pytest.mark.asyncio
 async def test_list_pending_approvals(monkeypatch: pytest.MonkeyPatch):
     now = datetime.now(timezone.utc)

tests/unit/mcpgateway/services/test_siem_export_service.py

@@ -133,3 +133,25 @@ async def test_webhook_send_uses_template(monkeypatch):
     await service._send_webhook(destination=destination, event=event)  # pylint: disable=protected-access
 
     mock_client.request.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_test_destination_escapes_destination_name_xss(monkeypatch):
+    """Attacker-controlled destination_name must be HTML-escaped in test results (CWE-79)."""
+    payload_str = "<script>alert(1)</script>"
+    service = svc.SIEMExportService()
+    service._destinations = {payload_str: {"name": payload_str, "type": "webhook", "url": "https://example.com/hook"}}  # pylint: disable=protected-access
+    service._send_to_destination = AsyncMock(side_effect=RuntimeError("connection refused"))  # pylint: disable=protected-access
+
+    result = await service.test_destination(payload_str)
+
+    assert result["status"] == "failed"
+    assert payload_str not in result["name"]
+    assert "&lt;script&gt;" in result["name"]
+
+    service._send_to_destination = AsyncMock(return_value=None)  # pylint: disable=protected-access
+    ok_result = await service.test_destination(payload_str)
+
+    assert ok_result["status"] == "ok"
+    assert payload_str not in ok_result["name"]
+    assert "&lt;script&gt;" in ok_result["name"]

tests/unit/mcpgateway/test_main.py

@@ -2584,6 +2584,28 @@ def test_remove_root_generic_exception(self, mock_remove, test_client, auth_head
         assert response.status_code == 500
         assert "Internal error" in response.json()["detail"]
 
+    @patch("mcpgateway.main.root_service.remove_root")
+    def test_remove_root_escapes_uri_in_success_message(self, mock_remove, test_client, auth_headers):
+        """Attacker-controlled uri must be HTML-escaped in the success message (CWE-79)."""
+        mock_remove.return_value = None
+        response = test_client.delete("/roots/%3Cscript%3Ealert(1)%3C%2Fscript%3E", headers=auth_headers)
+        assert response.status_code == 200
+        message = response.json()["message"]
+        assert "<script>" not in message
+        assert "&lt;script&gt;" in message
+
+    @patch("mcpgateway.main.root_service.remove_root")
+    def test_remove_root_escapes_uri_in_not_found_detail(self, mock_remove, test_client, auth_headers):
+        """Attacker-controlled uri must be HTML-escaped in the 404 detail (CWE-79)."""
+        from mcpgateway.services.root_service import RootServiceNotFoundError
+
+        mock_remove.side_effect = RootServiceNotFoundError("Root not found: <script>alert(1)</script>")
+        response = test_client.delete("/roots/%3Cscript%3Ealert(1)%3C%2Fscript%3E", headers=auth_headers)
+        assert response.status_code == 404
+        detail = response.json()["detail"]
+        assert "<script>" not in detail
+        assert "&lt;script&gt;" in detail
+
 
     @patch("mcpgateway.main.root_service.subscribe_changes")
     def test_subscribe_root_changes(self, mock_subscribe, test_client, auth_headers):