From e02f1a8cd01992911a5c4ccb3abe2a22f44bac37 Mon Sep 17 00:00:00 2001 From: YueGuobin Date: Sun, 14 Jun 2026 14:00:41 +0800 Subject: [PATCH] fix: Store username in gns3_ctx during auth, use for short-lived download JWTs _ jw t_username_var set in _resolve_token for both JWT and API key auth. Passed to handlers via gns3_ctx['jwt_username']. No raw key exposure, no fake-user fallback. --- gns3server/api/routes/mcp/__init__.py | 23 +++++++++++++++++------ gns3server/api/routes/mcp/links.py | 15 +++++++++------ gns3server/api/routes/mcp/nodes.py | 9 ++++++++- gns3server/api/routes/mcp/symbols.py | 15 +++++++++++---- 4 files changed, 45 insertions(+), 17 deletions(-) diff --git a/gns3server/api/routes/mcp/__init__.py b/gns3server/api/routes/mcp/__init__.py index bd1271e69..b6c33aa7e 100644 --- a/gns3server/api/routes/mcp/__init__.py +++ b/gns3server/api/routes/mcp/__init__.py @@ -53,6 +53,7 @@ import gns3server.db.models as models from gns3server.services import auth_service from gns3server.utils.request_utils import extract_client_info from gns3server.db.repositories.api_keys import ApiKeysRepository +from gns3server.db.repositories.users import UsersRepository from .projects import ( list_projects_handler, get_project_handler, create_project_handler, delete_project_handler, open_project_handler, close_project_handler, @@ -187,6 +188,11 @@ async def wait_for_mcp_ready() -> bool: _jwt_token_var: contextvars.ContextVar[str | None] = contextvars.ContextVar( "mcp_jwt_token", default=None ) +# Username extracted during token validation — used by handlers to generate +# short-lived JWTs for download/console URLs without exposing the raw key. +_jwt_username_var: contextvars.ContextVar[str | None] = contextvars.ContextVar( + "mcp_jwt_username", default=None +) # ── Token validation ────────────────────────────────────────────────── @@ -201,12 +207,13 @@ async def _resolve_token(token: str) -> str | None: """ # Try JWT first try: - auth_service.get_username_from_token(token) + username = auth_service.get_username_from_token(token) + _jwt_username_var.set(username) return token except Exception: pass - # Try API key — pass through directly; REST API auth already supports gns3_ keys + # Try API key if token.startswith("gns3_") and _app is not None: db_engine = getattr(_app.state, "_db_engine", None) if db_engine is not None: @@ -218,8 +225,10 @@ async def _resolve_token(token: str) -> str | None: for db_key in result.scalars().all(): if bcrypt.checkpw(token.encode(), db_key.key_hash.encode()): await repo.update_last_used(db_key.api_key_id) - # Return the raw API key — it will be passed as Bearer token - # and validated by the REST API auth layer + user_repo = UsersRepository(db_session) + user = await user_repo.get_user(db_key.user_id) + if user: + _jwt_username_var.set(user.username) return token except Exception: pass @@ -277,6 +286,7 @@ def _run_handler_sync(handler, params: dict[str, Any]) -> list[dict[str, Any]]: ctx = { "server_url": _server_url(), "jwt_token": _jwt_token_var.get(), + "jwt_username": _jwt_username_var.get(), } result = handler(params, ctx) return [{"type": "text", "text": json.dumps(result, ensure_ascii=False, default=str)}] @@ -528,6 +538,7 @@ async def node_console( Returns the WebSocket URL, console type (telnet/ssh/vnc), and other connection details needed to interact with a node's console via WebSocket. + The URL includes a short-lived JWT (10 min) — reconnect if it expires. Complete workflow: 1. Call this tool with project_id and node_id to get the WebSocket URL @@ -937,7 +948,7 @@ async def link_capture_download( project_id: Annotated[str, Field(description="UUID of the project")], link_id: Annotated[str, Field(description="UUID of the link")], ) -> list[dict[str, Any]]: - """Get the download URL and instructions for a PCAP capture file. Use curl to download.""" + """Get the download URL and instructions for a PCAP capture file. The URL includes a short-lived JWT (10 min). Use curl to download.""" return await asyncio.to_thread(_run_handler_sync, download_capture_file_handler, { "project_id": project_id, "link_id": link_id, }) @@ -1151,7 +1162,7 @@ async def symbol_list() -> list[dict[str, Any]]: async def symbol_get( symbol_id: Annotated[str, Field(description="Symbol ID (e.g. ':/symbols/router.svg')")], ) -> list[dict[str, Any]]: - """Get a download URL for a symbol file (SVG). Use curl to download.""" + """Get a download URL for a symbol file (SVG). The URL includes a short-lived JWT (10 min). Use curl to download.""" return await asyncio.to_thread(_run_handler_sync, get_symbol_handler, { "symbol_id": symbol_id, }) diff --git a/gns3server/api/routes/mcp/links.py b/gns3server/api/routes/mcp/links.py index 1a685f651..3ec519836 100644 --- a/gns3server/api/routes/mcp/links.py +++ b/gns3server/api/routes/mcp/links.py @@ -191,15 +191,18 @@ def download_capture_file_handler(params: dict[str, Any], gns3_ctx: dict[str, An if not project_id or not link_id: return {"error": "project_id and link_id are required"} download_url = f"{gns3_ctx['server_url']}/v3/projects/{project_id}/links/{link_id}/capture/file" - # Generate a short-lived download token (10 min) so the user can curl without exposing their API key - download_token = auth_service.create_access_token("mcp-download", expires_in=10) - return { + # Short-lived JWT (10 min) — username stored during auth, never exposes raw key + username = gns3_ctx.get("jwt_username") + download_token = auth_service.create_access_token(username, expires_in=10) if username else None + result = { "link_id": link_id, "download_url": download_url, - "curl_command": f"curl -L -o capture.pcap -H 'Authorization: Bearer {download_token}' '{download_url}'", - "note": "This download link expires in 10 minutes. " - "The file is in pcap format and can be analyzed with Wireshark or tcpdump.", + "note": "The file is in pcap format and can be analyzed with Wireshark or tcpdump.", } + if download_token: + result["curl_command"] = f"curl -L -o capture.pcap -H 'Authorization: Bearer {download_token}' '{download_url}'" + result["note"] += " The download link includes a 10-minute token." + return result # ── Tool definitions ─────────────────────────────────────────────────────── diff --git a/gns3server/api/routes/mcp/nodes.py b/gns3server/api/routes/mcp/nodes.py index e60ec1a76..507dc8d58 100644 --- a/gns3server/api/routes/mcp/nodes.py +++ b/gns3server/api/routes/mcp/nodes.py @@ -27,6 +27,8 @@ from concurrent.futures import ThreadPoolExecutor, as_completed import logging +from gns3server.services import auth_service + log = logging.getLogger(__name__) BATCH_MAX_WORKERS = 10 @@ -276,7 +278,12 @@ def get_node_console_info_handler(params: dict[str, Any], gns3_ctx: dict[str, An node = conn.http_call("get", f"{conn.base_url}/projects/{project_id}/nodes/{node_id}").json() console_type = node.get("console_type", "unknown") - ws_url = f"{gns3_ctx['server_url']}/v3/projects/{project_id}/nodes/{node_id}/console/ws?token={gns3_ctx['jwt_token']}" + # Short-lived JWT for the WebSocket URL (10 min) + username = gns3_ctx.get("jwt_username") + ws_token = auth_service.create_access_token(username, expires_in=10) if username else None + ws_url = f"{gns3_ctx['server_url']}/v3/projects/{project_id}/nodes/{node_id}/console/ws" + if ws_token: + ws_url += f"?token={ws_token}" result = { "node_id": node_id, diff --git a/gns3server/api/routes/mcp/symbols.py b/gns3server/api/routes/mcp/symbols.py index a49e417d5..d404e651b 100644 --- a/gns3server/api/routes/mcp/symbols.py +++ b/gns3server/api/routes/mcp/symbols.py @@ -23,6 +23,8 @@ from typing import Any import logging +from gns3server.services import auth_service + log = logging.getLogger(__name__) @@ -51,13 +53,18 @@ def get_symbol_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict if not symbol_id: return {"error": "symbol_id is required"} download_url = f"{gns3_ctx['server_url']}/v3/symbols/{symbol_id}/raw" - auth_token = gns3_ctx['jwt_token'] - return { + username = gns3_ctx.get("jwt_username") + download_token = auth_service.create_access_token(username, expires_in=10) if username else None + result = { "symbol_id": symbol_id, "download_url": download_url, - "curl_command": f"curl -L -o '{symbol_id.replace(':', '').replace('/', '_')}.svg' -H 'Authorization: Bearer {auth_token}' '{download_url}'", - "note": "Symbol files are SVG images. Use curl to download.", + "note": "Symbol files are SVG images.", } + if download_token: + safe_name = symbol_id.replace(':', '').replace('/', '_') + result["curl_command"] = f"curl -L -o '{safe_name}.svg' -H 'Authorization: Bearer {download_token}' '{download_url}'" + result["note"] += " Download link includes a 10-minute token." + return result def get_symbol_dimensions_handler(params: dict[str, Any], gns3_ctx: dict[str, Any]) -> dict[str, Any]: