From 668368de7b5b2d9827615dcfcd3188e91e90e79f Mon Sep 17 00:00:00 2001 From: Tobiasee16 Date: Thu, 14 May 2026 12:55:21 +0200 Subject: [PATCH 1/2] Block unauthenticated access to MCP endpoints Problem: Any client knowing a chat_id (a UUID) could directly call get_drawn_layers(session_id=) and retrieve another user's map geometry. Fix: - Added mcp_auth.py. This makes it impossible for other users to call tool endpoints directly from the outside by generating a new token when the server starts. - All six entries in server.py are now wrapped with MCPAuthMiddleware - session:manager.py includes the token as a header in every MCP server registration so the Copilot SDK includes it automatically on outgoing ccalls. --- backend/mcp_auth.py | 69 ++++++++++++++++++++++++++++++++++++++ backend/server.py | 15 +++++---- backend/session_manager.py | 9 +++++ 3 files changed, 87 insertions(+), 6 deletions(-) create mode 100644 backend/mcp_auth.py diff --git a/backend/mcp_auth.py b/backend/mcp_auth.py new file mode 100644 index 0000000..4cab440 --- /dev/null +++ b/backend/mcp_auth.py @@ -0,0 +1,69 @@ +""" +Internal authentication for MCP server endpoints. + +All MCP endpoints (/mcp/*) must only be called by the server-side session +manager, never directly by external clients. A per-process secret token is +generated at startup and required in the X-MCP-Internal-Token header on every +incoming HTTP request. Requests missing or carrying an incorrect token receive +a 401 response before the MCP handler ever runs. + +Usage: + # server.py + from mcp_auth import MCPAuthMiddleware, MCP_INTERNAL_SECRET + Mount("/mcp/map", app=MCPAuthMiddleware(map_app)) + + # session_manager.py + from mcp_auth import MCP_INTERNAL_SECRET + "headers": {"X-MCP-Internal-Token": MCP_INTERNAL_SECRET} +""" + +import secrets + +# Generated once per server process. Session manager embeds this value in +# the Authorization header when it registers MCP server URLs with the Copilot +# SDK, so only in-process calls succeed. +MCP_INTERNAL_SECRET: str = secrets.token_urlsafe(32) + +_HEADER_NAME: bytes = b"x-mcp-internal-token" +_EXPECTED: bytes = MCP_INTERNAL_SECRET.encode() + + +class MCPAuthMiddleware: + """ + Lightweight ASGI middleware that guards MCP HTTP endpoints. + + Non-HTTP scopes (lifespan, websocket) are forwarded unchanged so that + FastMCP startup/shutdown and any future transport upgrades still work. + """ + + __slots__ = ("_app",) + + def __init__(self, app) -> None: + self._app = app + + async def __call__(self, scope, receive, send) -> None: + if scope["type"] == "http": + headers: dict[bytes, bytes] = dict(scope.get("headers", [])) + token: bytes = headers.get(_HEADER_NAME, b"") + if not secrets.compare_digest(token, _EXPECTED): + await _reject_401(send) + return + + await self._app(scope, receive, send) + + +async def _reject_401(send) -> None: + await send( + { + "type": "http.response.start", + "status": 401, + "headers": [[b"content-type", b"application/json"]], + } + ) + await send( + { + "type": "http.response.body", + "body": b'{"error":"Unauthorized"}', + "more_body": False, + } + ) diff --git a/backend/server.py b/backend/server.py index 3ccaeea..af8fedf 100644 --- a/backend/server.py +++ b/backend/server.py @@ -51,6 +51,7 @@ from blob_storage import list_documents from config import ALLOWED_ORIGINS, DEMO_MODE, HOST, PORT from copilot import CopilotClient +from mcp_auth import MCPAuthMiddleware from sanitizer import ( sanitize_completed_thinking as _sanitize_completed_thinking, sanitize_thinking as _sanitize_thinking, @@ -580,12 +581,14 @@ async def test_search_chunk(request: Request): app = Starlette( routes=[ # MCP servers -- each accessible at /mcp//mcp - Mount("/mcp/db", app=db_app), - Mount("/mcp/geo", app=geo_app), - Mount("/mcp/docs", app=docs_app), - Mount("/mcp/vector", app=vector_app), - Mount("/mcp/map", app=map_app), - Mount("/mcp/search", app=search_app), + # Wrapped with MCPAuthMiddleware: only in-process requests carrying the + # per-startup X-MCP-Internal-Token header are accepted. + Mount("/mcp/db", app=MCPAuthMiddleware(db_app)), + Mount("/mcp/geo", app=MCPAuthMiddleware(geo_app)), + Mount("/mcp/docs", app=MCPAuthMiddleware(docs_app)), + Mount("/mcp/vector", app=MCPAuthMiddleware(vector_app)), + Mount("/mcp/map", app=MCPAuthMiddleware(map_app)), + Mount("/mcp/search", app=MCPAuthMiddleware(search_app)), # Auth endpoints Route("/api/auth/register", endpoint=register, methods=["POST"]), diff --git a/backend/session_manager.py b/backend/session_manager.py index de02eae..169e8f1 100644 --- a/backend/session_manager.py +++ b/backend/session_manager.py @@ -8,6 +8,7 @@ from copilot import CopilotClient from copilot.session import PermissionHandler, PermissionRequestResult from mcp_servers.map_server import get_and_clear_shapes, store_map_context, clear_map_context +from mcp_auth import MCP_INTERNAL_SECRET from copilot.generated.session_events import SessionEventType from usage_tracker import get_or_create_tracker, discard_tracker from config import ( @@ -200,36 +201,44 @@ async def get_or_create_for_chat( streaming=True, reasoning_effort="high", # MCP servers the orchestrator can invoke. + # The X-MCP-Internal-Token header is required by MCPAuthMiddleware + # so that only in-process calls are accepted. mcp_servers={ "database": { "type": "http", "url": f"{SERVER_BASE_URL}/mcp/db/mcp", "tools": ["*"], + "headers": {"X-MCP-Internal-Token": MCP_INTERNAL_SECRET}, }, "geo": { "type": "http", "url": f"{SERVER_BASE_URL}/mcp/geo/mcp", "tools": ["*"], + "headers": {"X-MCP-Internal-Token": MCP_INTERNAL_SECRET}, }, "docs": { "type": "http", "url": f"{SERVER_BASE_URL}/mcp/docs/mcp", "tools": ["*"], + "headers": {"X-MCP-Internal-Token": MCP_INTERNAL_SECRET}, }, "vector": { "type": "http", "url": f"{SERVER_BASE_URL}/mcp/vector/mcp", "tools": ["*"], + "headers": {"X-MCP-Internal-Token": MCP_INTERNAL_SECRET}, }, "map": { "type": "http", "url": f"{SERVER_BASE_URL}/mcp/map/mcp", "tools": ["*"], + "headers": {"X-MCP-Internal-Token": MCP_INTERNAL_SECRET}, }, "search": { "type": "http", "url": f"{SERVER_BASE_URL}/mcp/search/mcp", "tools": ["*"], + "headers": {"X-MCP-Internal-Token": MCP_INTERNAL_SECRET}, }, }, on_permission_request=permission_handler, From f49b6837709cf481b9e191b7446a2ca3e36a3545 Mon Sep 17 00:00:00 2001 From: Tobiasee16 Date: Thu, 14 May 2026 13:43:01 +0200 Subject: [PATCH 2/2] Fix geo_server: Allowlist - Add _ALLOWED_HOSTS frozenset; _fetch_json now rejects any URL whose scheme is not https or hostname is not ws.geonorge.no --- backend/mcp_servers/geo_server.py | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/backend/mcp_servers/geo_server.py b/backend/mcp_servers/geo_server.py index b3ced73..6715ee0 100644 --- a/backend/mcp_servers/geo_server.py +++ b/backend/mcp_servers/geo_server.py @@ -9,9 +9,11 @@ import json import logging +import re import urllib.request import urllib.parse import urllib.error +from urllib.parse import urlparse from fastmcp import FastMCP from db import query @@ -30,9 +32,21 @@ _KARTVERKET_STED_URL = "https://ws.geonorge.no/stedsnavn/v1/sted" _KARTVERKET_KOMMUNEINFO_URL = "https://ws.geonorge.no/kommuneinfo/v1/punkt" +# Allowlist of hosts _fetch_json is permitted to contact (SSRF guard). +_ALLOWED_HOSTS: frozenset[str] = frozenset({ + "ws.geonorge.no", +}) + +# Only allow printable place-name characters; forbids CRLF and other controls. +_SAFE_PLACENAME_RE = re.compile(r"^[\w\s\-.,/()æøåÆØÅ]{1,200}$", re.UNICODE) + def _fetch_json(url: str) -> dict | None: - """Fetch JSON from a URL, returning None on failure.""" + """Fetch JSON from an allowlisted URL, returning None on failure.""" + parsed = urlparse(url) + if parsed.scheme != "https" or parsed.hostname not in _ALLOWED_HOSTS: + logger.error("_fetch_json avvist URL utenfor allowlist: %s", url) + return None try: req = urllib.request.Request(url, headers={"Accept": "application/json"}) with urllib.request.urlopen(req, timeout=10) as resp: @@ -247,6 +261,8 @@ async def forward_geocode(name: str) -> str: return json.dumps({"error": "Tomt søkeord."}) search_term = name.strip() + if not _SAFE_PLACENAME_RE.match(search_term): + return json.dumps({"error": "Ugyldig tegn i søkeord."}) # 1) Try /navn with wildcard — precise match on skrivemåte navn_params = urllib.parse.urlencode({