feat: add client information logging to MCP connection rejection

Add reusable utility for extracting client information from ASGI scope:
- Create gns3server/utils/request_utils.py with extract_client_info()
- Extract client IP, port, path, method, and authenticated username
- Format comprehensive log messages with client context

Benefits:
- Better observability for connection rejection events
- Reusable utility for other modules
- Consistent client logging format across codebase
- Helps diagnose timing issues during server startup

Usage:
  from gns3server.utils.request_utils import extract_client_info
  client_info = extract_client_info(scope, auth_service)
  log.warning(f"Connection rejected - Client: {client_info['host']}:{client_info['port']} ({client_info['user_info']})")
This commit is contained in:
YueGuobin 2026-06-06 22:35:43 +08:00
parent 9e5b575d1c
commit 3958279b8a
No known key found for this signature in database
2 changed files with 97 additions and 0 deletions

View File

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

View File

@ -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']})"