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

@@ -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,
     })

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

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,

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]: