diff --git a/gns3server/api/routes/mcp/__init__.py b/gns3server/api/routes/mcp/__init__.py index 39ab5af51..ce77a48c7 100644 --- a/gns3server/api/routes/mcp/__init__.py +++ b/gns3server/api/routes/mcp/__init__.py @@ -44,6 +44,7 @@ from mcp.server.transport_security import TransportSecuritySettings from gns3server.config import Config from gns3server.services import auth_service +from gns3server.utils.request_utils import extract_client_info from .projects import ( list_projects_handler, get_project_handler, create_project_handler, delete_project_handler, open_project_handler, close_project_handler, @@ -516,6 +517,11 @@ def _make_auth_wrapper(inner_app): server_ready = await wait_for_mcp_ready() if not server_ready: # Server initialization timed out - return 503 Service Unavailable + client_info = extract_client_info(scope, auth_service) + log.warning( + f"Rejecting MCP connection - GNS3 server initialization not complete. " + f"Client: {client_info['host']}:{client_info['port']} ({client_info['user_info']}, Path: {client_info['path']})" + ) response = Response( "GNS3 server initialization not complete - please retry later", status_code=503 diff --git a/gns3server/utils/request_utils.py b/gns3server/utils/request_utils.py new file mode 100644 index 000000000..aee3bdb2e --- /dev/null +++ b/gns3server/utils/request_utils.py @@ -0,0 +1,91 @@ +""" +Utilities for extracting request information from ASGI scope. + +This module provides reusable functions for extracting client information +from ASGI scope dictionaries for logging and debugging purposes. +""" + +import logging +from urllib.parse import parse_qs +from typing import Dict, Any, Optional + +log = logging.getLogger(__name__) + + +def extract_client_info(scope: Dict[str, Any], auth_service_instance: Optional[Any] = None) -> Dict[str, str]: + """ + Extract client information from ASGI scope for logging purposes. + + Args: + scope: ASGI scope dictionary containing request metadata + auth_service_instance: Optional auth service instance for token validation + + Returns: + Dictionary with client information: + - host: Client IP address + - port: Client port + - path: Request path + - method: Request method + - username: Authenticated username (if token provided and valid) + - user_info: Human-readable user info string + """ + # Extract client address and port + client = scope.get("client", (None, None)) + client_host = client[0] if client and client[0] else "unknown" + client_port = str(client[1]) if client and len(client) > 1 else "unknown" + + # Extract request info + path = scope.get("path", "unknown") + method = scope.get("method", "unknown") + + # Try to extract username from token + username = None + if auth_service_instance: + try: + headers = dict(scope.get("headers", [])) + auth = headers.get(b"authorization", b"").decode() + token = None + + # Try Authorization header + if auth.startswith("Bearer "): + token = auth[7:] + + # Try query parameter + if not token: + params = parse_qs(scope.get("query_string", b"").decode()) + tokens = params.get("token", []) + if tokens: + token = tokens[0] + + # Validate and extract username + if token: + username = auth_service_instance.get_username_from_token(token) + except Exception as e: + log.debug(f"Failed to extract username from token: {e}") + username = None + + # Create user-friendly info string + user_info = f"user '{username}'" if username else "unauthenticated user" + + return { + "host": client_host, + "port": client_port, + "path": path, + "method": method, + "username": username, + "user_info": user_info + } + + +def format_client_log(client_info: Dict[str, str], message: str) -> str: + """ + Format a log message with client information. + + Args: + client_info: Client information dict from extract_client_info() + message: Log message + + Returns: + Formatted log string with client prefix + """ + return f"{message} - Client: {client_info['host']}:{client_info['port']} ({client_info['user_info']}, Path: {client_info['path']})"