MCP server機能をセキュア化

1 files changed, 31 insertions, 7 deletions

diff --git a/src/ollama_web/routes/mcp.py b/src/ollama_web/routes/mcp.py
index 85d4bb3..ee8ba65 100644
--- a/src/ollama_web/routes/mcp.py
+++ b/src/ollama_web/routes/mcp.py
@@ -8,7 +8,13 @@ from typing import Any
 from starlette.requests import Request
 from starlette.responses import JSONResponse
 
-from ..mcp import load_mcp_config, save_mcp_config
+from ..mcp import (
+    load_mcp_config,
+    redact_secrets,
+    save_mcp_config,
+    validate_mcp_server_config,
+    validate_mcp_server_name,
+)
 
 logger = logging.getLogger("ollama_web.mcp_route")
 
@@ -21,7 +27,7 @@ async def get_servers(request: Request) -> JSONResponse:
     """GET /api/mcp/servers: return the current MCP server configuration."""
     data_dir = request.app.state.settings.data_dir
     config = load_mcp_config(data_dir)
-    return JSONResponse(config)
+    return JSONResponse(redact_secrets(config))
 
 
 async def put_servers(request: Request) -> JSONResponse:
@@ -51,7 +57,14 @@ async def put_servers(request: Request) -> JSONResponse:
 
     cleaned: dict[str, Any] = {"mcpServers": {}}
     errors: list[str] = []
-    for name, server in raw_servers.items():
+    data_dir = request.app.state.settings.data_dir
+    for raw_name, server in raw_servers.items():
+        try:
+            name = validate_mcp_server_name(raw_name)
+        except ValueError as exc:
+            errors.append(f"Server '{raw_name}' invalid name: {exc}")
+            continue
+
         if not isinstance(server, dict):
             errors.append(f"Server '{name}' must be an object")
             continue
@@ -59,11 +72,23 @@ async def put_servers(request: Request) -> JSONResponse:
         if "command" in server:
             valid = _allowed_fields(server, _ALLOWED_STDIO_FIELDS, name, errors)
             if valid:
-                cleaned["mcpServers"][name] = _clean_stdio_server(server)
+                cleaned_server = _clean_stdio_server(server)
+                try:
+                    validate_mcp_server_config(name, cleaned_server, data_dir)
+                except ValueError as exc:
+                    errors.append(f"Server '{name}' invalid stdio config: {exc}")
+                    continue
+                cleaned["mcpServers"][name] = cleaned_server
         elif "url" in server:
             valid = _allowed_fields(server, _ALLOWED_HTTP_FIELDS, name, errors)
             if valid:
-                cleaned["mcpServers"][name] = _clean_http_server(server)
+                cleaned_server = _clean_http_server(server)
+                try:
+                    validate_mcp_server_config(name, cleaned_server, data_dir)
+                except ValueError as exc:
+                    errors.append(f"Server '{name}' invalid http config: {exc}")
+                    continue
+                cleaned["mcpServers"][name] = cleaned_server
         else:
             errors.append(
                 f"Server '{name}' must define either 'command' (stdio) or 'url' (http)"
@@ -72,7 +97,6 @@ async def put_servers(request: Request) -> JSONResponse:
     if errors:
         return JSONResponse({"error": errors}, status_code=400)
 
-    data_dir = request.app.state.settings.data_dir
     save_mcp_config(data_dir, cleaned)
     return JSONResponse(cleaned)
 
@@ -116,4 +140,4 @@ def _clean_http_server(server: dict[str, Any]) -> dict[str, Any]:
                 pass
     if "terminate_on_close" in server:
         out["terminate_on_close"] = bool(server["terminate_on_close"])
-    return out
\ No newline at end of file
+    return out