# -*- coding: utf-8 -*- # pylint: disable=wrong-import-position, import-outside-toplevel, no-name-in-module """Location: ./mcpgateway/main.py Copyright contributors to the MCP-CONTEXT-FORGE project SPDX-License-Identifier: Apache-2.0 ContextForge AI Gateway - Main FastAPI Application. This module defines the core FastAPI application for the Model Context Protocol (MCP) Gateway. It serves as the entry point for handling all HTTP and WebSocket traffic. Features and Responsibilities: - Initializes and orchestrates services for tools, resources, prompts, servers, gateways, and roots. - Supports full MCP protocol operations: initialize, ping, notify, complete, and sample. - Integrates authentication (JWT and basic), CORS, caching, and middleware. - Serves a rich Admin UI for managing gateway entities via HTMX-based frontend. - Exposes routes for JSON-RPC, SSE, and WebSocket transports. - Manages application lifecycle including startup and graceful shutdown of all services. Structure: - Declares routers for MCP protocol operations and administration. - Registers dependencies (e.g., DB sessions, auth handlers). - Applies middleware including custom documentation protection. - Configures resource caching and session registry using pluggable backends. - Provides OpenAPI metadata and redirect handling depending on UI feature flags. """ # Standard import asyncio import base64 from contextlib import asynccontextmanager, suppress from datetime import datetime, timezone from functools import lru_cache import html import json import logging import math import multiprocessing import os import re import signal import sys import threading from typing import Any, AsyncIterator, Dict, List, Optional, TypeAlias, Union from urllib.parse import urlparse, urlunparse import uuid import warnings # Third-Party from cpex.framework import HttpHookType, PluginError, PluginViolationError, PromptHookType, ResourceHookType from fastapi import APIRouter, Body, Depends, FastAPI, HTTPException, Query, Request, status, WebSocket, WebSocketDisconnect from fastapi.background import BackgroundTasks from fastapi.exception_handlers import request_validation_exception_handler as fastapi_default_validation_handler from fastapi.exceptions import RequestValidationError from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse, RedirectResponse, Response, StreamingResponse from fastapi.security import HTTPAuthorizationCredentials from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from jinja2 import Environment, FileSystemLoader from jsonpath_ng.ext import parse from jsonpath_ng.jsonpath import JSONPath import orjson from pydantic import ValidationError from sqlalchemy import text from sqlalchemy.exc import DataError, IntegrityError from sqlalchemy.orm import Session from starlette.middleware.base import BaseHTTPMiddleware from starlette.requests import Request as starletteRequest from starlette.responses import Response as starletteResponse from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware # First-Party # Import the admin routes from the new module from mcpgateway import __version__ from mcpgateway import version as version_module from mcpgateway.auth import get_current_user, get_user_team_roles, TokenValidationError, validate_token_user from mcpgateway.auth_context import ( configuration_export_includes_roots, decode_internal_mcp_auth_context, encode_internal_mcp_auth_context, get_internal_mcp_auth_context, get_request_identity, get_rpc_filter_context, get_scoped_resource_access_context, get_token_teams_from_request, get_user_email, import_envelope_includes_roots, INTERNAL_MCP_SESSION_VALIDATED_HEADER, is_unrestricted_platform_admin, is_trusted_internal_mcp_request, selective_selection_includes_roots, ) from mcpgateway.cache import ResourceCache, SessionRegistry from mcpgateway.common.models import InitializeResult from mcpgateway.common.models import JSONRPCError as PydanticJSONRPCError from mcpgateway.common.models import ListResourceTemplatesResult, LogLevel, Root from mcpgateway.common.query_params import QueryGatewayId, QueryPaginationCursor, QueryTeamId, QueryVisibility from mcpgateway.common.validators import SecurityValidator from mcpgateway.config import get_settings, SecurityConfigurationError, settings from mcpgateway.db import A2AAgent as DbA2AAgent from mcpgateway.db import A2APushNotificationConfig from mcpgateway.db import A2ATask as DbA2ATask from mcpgateway.db import refresh_slugs_on_startup, SessionLocal from mcpgateway.db import Tool as DbTool from mcpgateway.deprecations import RUST_MCP_RUNTIME_DEPRECATION_MESSAGE, VALIDATION_MIDDLEWARE_DEPRECATION_MESSAGE from mcpgateway.handlers.sampling import SamplingError, SamplingHandler from mcpgateway.middleware.auth_context_stack import register_auth_context_middleware from mcpgateway.middleware.client_disconnect import ClientDisconnectMiddleware from mcpgateway.middleware.compression import SSEAwareCompressMiddleware from mcpgateway.middleware.correlation_id import CorrelationIDMiddleware from mcpgateway.middleware.forwarded_host import ForwardedHostMiddleware from mcpgateway.middleware.header_size_middleware import HeaderSizeMiddleware from mcpgateway.middleware.http_auth_middleware import HttpAuthMiddleware, run_pre_request_hooks from mcpgateway.middleware.protocol_version import MCPProtocolVersionMiddleware from mcpgateway.middleware.rate_limit_middleware import RateLimitMiddleware from mcpgateway.middleware.rbac import _ACCESS_DENIED_MSG, get_current_user_with_permissions, PermissionChecker, require_permission from mcpgateway.middleware.request_logging_middleware import RequestLoggingMiddleware from mcpgateway.middleware.security_headers import SecurityHeadersMiddleware from mcpgateway.middleware.token_scoping import ResourceOwnershipResult, token_scoping_middleware from mcpgateway.middleware.validation_middleware import ValidationMiddleware from mcpgateway.observability import configure_baggage_span_attribute_policy, extract_baggage_span_attribute_policy, init_telemetry, OpenTelemetryRequestMiddleware, otel_tracing_enabled from mcpgateway.plugins import ( enable_plugins, get_plugin_manager, get_plugin_manager_factory, init_plugin_manager_factory, shutdown_plugin_manager_factory, start_plugin_invalidation_listener, stop_plugin_invalidation_listener, ) from mcpgateway.plugins.violation_codes import PLUGIN_VIOLATION_CODE_MAPPING, PluginViolationCode, VALID_HTTP_STATUS_CODES from mcpgateway.routers.openapi_schema_router import router as openapi_schema_router from mcpgateway.routers.server_well_known import router as server_well_known_router from mcpgateway.routers.well_known import router as well_known_router from mcpgateway.schemas import ( A2AAgentCreate, A2AAgentRead, A2AAgentUpdate, A2APushNotificationConfigCreate, CursorPaginatedA2AAgentsResponse, CursorPaginatedGatewaysResponse, CursorPaginatedPromptsResponse, CursorPaginatedResourcesResponse, CursorPaginatedServersResponse, CursorPaginatedToolsResponse, GatewayCreate, GatewayRead, GatewayRefreshResponse, GatewayUpdate, HealthCheckResponse, HealthStatusItem, JsonPathModifier, MetricsResponse, PromptCreate, PromptExecuteArgs, PromptRead, PromptUpdate, ResourceCreate, ResourceRead, ResourceSubscription, ResourceUpdate, RootCreate, RootUpdate, RPCRequest, ServerCreate, ServerRead, ServerUpdate, TaggedEntity, TagInfo, ToolCreate, ToolRead, ToolUpdate, ) from mcpgateway.services.a2a_server_service import A2AServerService from mcpgateway.services.a2a_service import A2AAgentError, A2AAgentNameConflictError, A2AAgentNotFoundError, A2AAgentService from mcpgateway.services.cancellation_service import cancellation_service from mcpgateway.services.completion_service import CompletionError, CompletionService from mcpgateway.services.content_security import ContentPatternError, ContentSizeError, ContentTypeError, TemplateValidationError from mcpgateway.services.dataplane_publisher import DataplanePublisherService from mcpgateway.services.email_auth_service import EmailAuthService from mcpgateway.services.export_service import ExportError, ExportService from mcpgateway.services.gateway_service import GatewayConnectionError, GatewayDuplicateConflictError, GatewayError, GatewayLookupConflictError, GatewayNameConflictError, GatewayNotFoundError from mcpgateway.services.import_service import ConflictStrategy, ImportConflictError from mcpgateway.services.import_service import ImportError as ImportServiceError from mcpgateway.services.import_service import ImportService, ImportValidationError from mcpgateway.services.log_aggregator import get_log_aggregator from mcpgateway.services.logging_service import LoggingService from mcpgateway.services.mcp_apps import ( apply_tool_meta, build_mcp_apps_capabilities, filter_model_visible_tools, get_mcp_app_session_cleanup_service, mcp_app_session_service, mcp_apps_enabled, MCPAppsValidationError, serialize_resource_content_for_mcp, ) from mcpgateway.services.mcp_method_registry import mcp_method_registry from mcpgateway.services.metrics import setup_metrics from mcpgateway.services.permission_service import PermissionService from mcpgateway.services.prompt_service import PromptError, PromptLockConflictError, PromptNameConflictError, PromptNotFoundError from mcpgateway.services.resource_service import ResourceError, ResourceLockConflictError, ResourceNotFoundError, ResourceURIConflictError, ResourceValidationError from mcpgateway.services.server_service import ServerError, ServerLockConflictError, ServerNameConflictError, ServerNotFoundError from mcpgateway.services.tag_service import TagService from mcpgateway.services.tool_service import ToolError, ToolLockConflictError, ToolNameConflictError, ToolNotFoundError from mcpgateway.transports.sse_transport import SSETransport from mcpgateway.transports.streamablehttp_transport import ( _validate_streamable_session_access, get_streamable_http_auth_context, SessionManagerWrapper, set_shared_session_registry, streamable_http_auth, user_context_var, ) from mcpgateway.utils import uaid as uaid_utils from mcpgateway.utils.admin_check import is_admin_bypass_granted from mcpgateway.utils.csp_nonce import get_csp_nonce_from_request from mcpgateway.utils.error_formatter import ErrorFormatter, sanitize_validation_error_for_log, should_expose_error_details from mcpgateway.utils.header_filtering import filter_sensitive_headers as _filter_sensitive_headers from mcpgateway.utils.internal_http import internal_loopback_base_url, internal_loopback_verify from mcpgateway.utils.jq_runner import shutdown_jq_pool, start_jq_pool from mcpgateway.utils.metadata_capture import MetadataCapture from mcpgateway.utils.orjson_response import ORJSONResponse from mcpgateway.utils.passthrough_headers import set_global_passthrough_headers from mcpgateway.utils.paths import resolve_root_path from mcpgateway.utils.redis_client import close_redis_client, get_redis_client, is_redis_available from mcpgateway.utils.redis_isready import wait_for_redis_ready from mcpgateway.utils.retry_manager import ResilientHttpClient from mcpgateway.utils.token_scoping import validate_server_access from mcpgateway.utils.trace_context import clear_trace_context, set_trace_context_from_teams, set_trace_session_id from mcpgateway.utils.trace_redaction import safe_log_user from mcpgateway.utils.verify_credentials import ( _resolve_auth_header_name, extract_websocket_bearer_token, get_auth_header_value, is_proxy_auth_trust_active, require_admin_auth, require_docs_auth_override, ) from mcpgateway.validation.jsonrpc import JSONRPCError # Initialize logging service first logging_service = LoggingService() logger = logging_service.get_logger("mcpgateway") # Note: Logging configuration is handled by LoggingService during startup # Don't use basicConfig here as it conflicts with our dual logging setup # Note: DB readiness probing and bootstrap_db() are deferred to the lifespan # startup hook so that `import mcpgateway.main` does no I/O. See lifespan(). # Enable plugin subsystem at module load time, mirroring the old singleton pattern. # get_plugin_manager() guards on this flag, so it must be set before lifespan runs. if settings.plugins.enabled: enable_plugins(True) logger.info("Plugin subsystem enabled (factory will be initialized in lifespan)") # First-Party # First-Party - import module-level service singletons from mcpgateway.services.gateway_service import gateway_service # noqa: E402 from mcpgateway.services.prompt_service import prompt_service # noqa: E402 from mcpgateway.services.resource_service import resource_service # noqa: E402 from mcpgateway.services.root_service import root_service, RootServiceError, RootServiceNotFoundError, RootServiceValidationError # noqa: E402 from mcpgateway.services.server_service import server_service # noqa: E402 from mcpgateway.services.tool_service import tool_service # noqa: E402 # Services that do not expose module-level singletons are instantiated here completion_service = CompletionService() sampling_handler = SamplingHandler() tag_service = TagService() export_service = ExportService() import_service = ImportService() # Initialize A2A service only if A2A features are enabled a2a_service = A2AAgentService() if settings.mcpgateway_a2a_enabled else None # Initialize session manager for Streamable HTTP transport streamable_http_session = SessionManagerWrapper() # Wait for redis to be ready if settings.cache_type == "redis" and settings.redis_url is not None: # First-Party from mcpgateway.utils.redis_client import _build_ssl_kwargs wait_for_redis_ready( redis_url=settings.redis_url, max_retries=int(settings.redis_max_retries), retry_interval_ms=int(settings.redis_retry_interval_ms), ssl_kwargs=_build_ssl_kwargs(settings), sync=True, ) # Initialize session registry session_registry = SessionRegistry( backend=settings.cache_type, redis_url=settings.redis_url if settings.cache_type == "redis" else None, database_url=settings.database_url if settings.cache_type == "database" else None, session_ttl=settings.session_ttl, message_ttl=settings.message_ttl, ) set_shared_session_registry(session_registry) _INTERNAL_MCP_AUTH_CONTEXT_HEADER = "x-contextforge-auth-context" def _is_trusted_internal_mcp_runtime_request(request: Request) -> bool: """Return whether the request came from a trusted local internal source. Two callers are trusted today: - ``"rust"`` — the local Rust runtime sidecar (over loopback). - ``"affinity"`` — the in-process dispatch used by session-affinity forwarding to reach the owner worker, carrying the identity the edge already validated. Both share the same gates: a shared-secret HMAC header AND a loopback client address. Only the ``x-contextforge-mcp-runtime`` marker value differs. Args: request: Incoming request to inspect. Returns: ``True`` when the request carries a trusted internal-runtime marker from loopback, otherwise ``False``. """ return is_trusted_internal_mcp_request(request) def _is_jwt_token(token: str) -> bool: """Check if a token looks like a JWT (has 2 dots, 3 base64url parts). Rejects local opaque tokens (cf_sess_*, cf_pat_*) that remote gateways cannot validate. """ if not token: return False if token.startswith(("cf_sess_", "cf_pat_")): return False parts = token.split(".") if len(parts) != 3: return False for part in parts: if not part: return False try: padded = part + "=" * (-len(part) % 4) base64.urlsafe_b64decode(padded) except Exception: # pylint: disable=broad-exception-caught return False return True def _validate_internal_mcp_auth_context(auth_context: Dict[str, Any]) -> None: """Validate a decoded trusted-internal auth context, failing closed on malformed input. The public-only RBAC skip in ``_ensure_rpc_permission`` trusts this context, so a public-only context (``is_authenticated is False``) must not carry authenticated-only or elevated attributes. Field types are checked first to avoid downstream confusion (for example a string ``scoped_permissions`` would be iterated per-character). Args: auth_context: Decoded auth-context dict from ``decode_internal_mcp_auth_context``. Raises: HTTPException: 400 when the context is malformed or a public-only context claims teams, admin, or an identity. """ teams = auth_context.get("teams") if teams is not None and not isinstance(teams, list): raise HTTPException(status_code=400, detail="Invalid trusted MCP auth context: teams must be a list") scoped_permissions = auth_context.get("scoped_permissions") if scoped_permissions is not None and not isinstance(scoped_permissions, list): raise HTTPException(status_code=400, detail="Invalid trusted MCP auth context: scoped_permissions must be a list") # is_authenticated must be a real bool so the ``is False`` identity checks below (and the # public-only RBAC skip in _ensure_rpc_permission) are reliable. A truthy non-bool like # the string "false" or 0 would slip past ``is False`` and defeat the public-only flooring. is_authenticated = auth_context.get("is_authenticated") if is_authenticated is not None and not isinstance(is_authenticated, bool): raise HTTPException(status_code=400, detail="Invalid trusted MCP auth context: is_authenticated must be a bool") # A public-only (unauthenticated) context must map to exactly public privileges. # The RBAC skip relies on this invariant, so reject any contradictory attributes # rather than letting them ride an unauthenticated dispatch. if is_authenticated is False: if teams: raise HTTPException(status_code=400, detail="Invalid public-only auth context: teams must be empty") if auth_context.get("is_admin") is True or auth_context.get("permission_is_admin") is True: raise HTTPException(status_code=400, detail="Invalid public-only auth context: admin not permitted") if auth_context.get("email"): raise HTTPException(status_code=400, detail="Invalid public-only auth context: email not permitted") def _build_internal_mcp_forwarded_user(request: Request) -> Dict[str, Any]: """Build the authenticated user payload for internal Rust -> Python MCP dispatch. Args: request: Trusted internal request forwarded from the Rust runtime. Returns: Synthetic authenticated user payload used by internal MCP handlers. Raises: HTTPException: If the request is not trusted or the forwarded auth context is missing or invalid. """ if not _is_trusted_internal_mcp_runtime_request(request): raise HTTPException(status_code=403, detail="Internal MCP dispatch is only available to the local Rust runtime") header_value = request.headers.get(_INTERNAL_MCP_AUTH_CONTEXT_HEADER) if not header_value: raise HTTPException(status_code=400, detail="Missing trusted MCP auth context") try: auth_context = decode_internal_mcp_auth_context(header_value) except Exception as exc: logger.debug("Invalid trusted MCP auth context: %s", exc) raise HTTPException(status_code=400, detail="Invalid trusted MCP auth context") from exc # Fail closed on a malformed or self-contradictory context before it is stored # and trusted by the public-only RBAC skip downstream. _validate_internal_mcp_auth_context(auth_context) setattr(request.state, "_mcp_internal_auth_context", auth_context) if "teams" in auth_context and (auth_context["teams"] is None or isinstance(auth_context["teams"], list)): request.state.token_teams = auth_context["teams"] if request.headers.get(INTERNAL_MCP_SESSION_VALIDATED_HEADER) == "rust": auth_context["_rust_session_validated"] = True forwarded_auth_method = auth_context.get("auth_method") or "mcp_internal_forward" set_trace_context_from_teams( auth_context.get("teams"), user_email=auth_context.get("email"), is_admin=bool(auth_context.get("permission_is_admin", auth_context.get("is_admin", False))), auth_method=forwarded_auth_method, team_name=auth_context.get("team_name"), ) return { "email": auth_context.get("email"), "full_name": auth_context.get("email") or "MCP Internal Forward", "is_admin": bool(auth_context.get("permission_is_admin", auth_context.get("is_admin", False))), "auth_method": forwarded_auth_method, "token_use": auth_context.get("token_use"), } def _build_internal_mcp_auth_context_for_rpc(request: Request, user: Any) -> Dict[str, Any]: """Build the trusted-internal auth context for an affinity-forwarded ``/rpc`` request. Affinity forwarding of a JSON-RPC ``/rpc`` request must carry the caller's already-validated identity to the owner worker's ``/_internal/mcp/rpc`` dispatch, so the owner does not re-authenticate at the public route boundary (which would 401 OAuth and ``MCP_REQUIRE_AUTH=false`` public-only callers). The identity is derived from the verified request state via ``get_rpc_filter_context`` (the canonical Layer-1 policy source) and the cached verified JWT payload, never from inbound headers, so token-team and admin semantics are preserved. The result has the same shape ``get_streamable_http_auth_context()`` emits, so the owner-side ``_build_internal_mcp_forwarded_user`` reconstructs both forward paths identically, and it satisfies ``_validate_internal_mcp_auth_context``. Args: request: The incoming ``/rpc`` request (already authenticated by the route). user: The user object produced by the auth dependency. Returns: Encodable auth-context dict for ``encode_internal_mcp_auth_context``. """ # Layer-1 exception: forwards an auth context, does not derive visibility scope. # Needs the raw is_admin flag. email, token_teams, is_admin = get_rpc_filter_context(request, user) # Genuine anonymous / MCP_REQUIRE_AUTH=false public-only callers have no email. is_authenticated = email is not None scoped = _extract_scoped_permissions(request) scoped_permissions = sorted(scoped) if scoped else None cached = getattr(request.state, "_jwt_verified_payload", None) payload = cached[1] if (isinstance(cached, tuple) and len(cached) == 2 and isinstance(cached[1], dict)) else {} scopes = payload.get("scopes") if isinstance(payload.get("scopes"), dict) else {} scoped_server_id = scopes.get("server_id") context: Dict[str, Any] = { "email": email, # Authenticated callers keep their token teams (None == admin bypass); public-only # callers are floored to no teams so _validate_internal_mcp_auth_context accepts them. "teams": token_teams if is_authenticated else [], "is_authenticated": is_authenticated, "is_admin": bool(is_admin) if is_authenticated else False, "permission_is_admin": bool(is_admin) if is_authenticated else False, "auth_method": payload.get("auth_method") or ("jwt" if is_authenticated else "anonymous"), "token_use": payload.get("token_use"), } if scoped_permissions is not None: context["scoped_permissions"] = scoped_permissions if scoped_server_id: context["scoped_server_id"] = scoped_server_id return context def _enforce_internal_mcp_server_scope(request: Request, server_id: str) -> None: """Validate trusted internal server scope against any forwarded token server scope. Args: request: Trusted internal MCP request. server_id: Effective virtual server identifier for the operation. Raises: HTTPException: If the forwarded token scope does not authorize the server. """ auth_context = get_internal_mcp_auth_context(request) if not isinstance(auth_context, dict): return scoped_server_id = auth_context.get("scoped_server_id") if isinstance(scoped_server_id, str) and scoped_server_id and not validate_server_access({"server_id": scoped_server_id}, server_id): raise HTTPException(status_code=403, detail=f"Token not authorized for server: {server_id}") async def _authorize_internal_mcp_request(request: Request, db: Session, *, permission: str, method: str, server_id: Optional[str] = None): """Authorize trusted Rust-side MCP dispatch while preserving permissive MCP semantics. For authenticated callers, this enforces the same token-scope and RBAC rules as the regular RPC dispatcher. For unauthenticated MCP callers in permissive mode, StreamableHTTP middleware already downgraded them to public-only scope and enforced per-server OAuth, so the internal Rust -> Python hop should not re-deny public-only requests merely because there is no authenticated RBAC identity. Args: request: Trusted internal MCP request. db: Active database session. permission: RBAC permission required for the method. method: MCP method name being authorized. server_id: Optional virtual server identifier used for additional scope checks. Returns: The forwarded user payload used for downstream authorization and scoping. """ user = _build_internal_mcp_forwarded_user(request) auth_context = get_internal_mcp_auth_context(request) or {} if server_id: _enforce_internal_mcp_server_scope(request, server_id) if auth_context.get("is_authenticated", True) is True: await _ensure_rpc_permission(user, db, permission, method, request=request) return user def _build_internal_mcp_auth_scope( *, method: str, path: str, query_string: str, headers: Dict[str, str], client_ip: Optional[str], ) -> Dict[str, Any]: """Construct a synthetic ASGI scope for internal Rust -> Python MCP auth. Args: method: HTTP method of the original public MCP request. path: Public MCP path, for example ``/mcp`` or ``/servers//mcp``. query_string: Raw query string without the leading ``?``. headers: Public request headers to replay through auth/token scoping. client_ip: Effective client IP derived by Rust from the public request. Returns: ASGI scope dictionary suitable for token scoping and ``streamable_http_auth``. """ raw_headers = [] for name, value in headers.items(): if not isinstance(name, str) or not isinstance(value, str): continue raw_headers.append((name.lower().encode("latin-1"), value.encode("latin-1"))) return { "type": "http", "method": method.upper(), "path": path, "raw_path": path.encode("latin-1"), "query_string": query_string.encode("latin-1"), "headers": raw_headers, "client": (client_ip or "unknown", 0), "state": {}, } async def _run_internal_mcp_authentication( *, method: str, path: str, query_string: str, headers: Dict[str, str], client_ip: Optional[str], ) -> tuple[Optional[Response], Dict[str, Any]]: """Run token scoping and MCP transport auth for a direct Rust ingress request. Runs HTTP_PRE_REQUEST plugin hooks (e.g. WXO auth token exchange) before authentication so the Rust MCP path gets identical plugin behavior to the Python middleware chain. Args: method: HTTP method of the public request. path: Public request path. query_string: Raw query string without the leading ``?``. headers: Public request headers replayed from Rust. client_ip: Effective client IP for token-scope IP restriction checks. Returns: Tuple of ``(error_response, auth_context)``. ``error_response`` is ``None`` on success; otherwise it contains the exact response generated by the existing token-scoping/auth layers. """ # Run pre-request plugin hooks (e.g. WXO JWT → team token exchange) # before building the auth scope, so plugins can transform headers. plugin_manager = await get_plugin_manager() if plugin_manager and plugin_manager.has_hooks_for(HttpHookType.HTTP_PRE_REQUEST): headers, _, _ = await run_pre_request_hooks( plugin_manager=plugin_manager, headers=headers, path=path, method=method, client_host=client_ip, ) scope = _build_internal_mcp_auth_scope( method=method, path=path, query_string=query_string, headers=headers, client_ip=client_ip, ) request = starletteRequest(scope) sent_messages: list[dict[str, Any]] = [] async def _receive() -> dict[str, Any]: """Return an empty request body for the synthetic auth probe. Returns: Minimal ASGI ``http.request`` message with no body content. """ return {"type": "http.request", "body": b"", "more_body": False} async def _send(message: dict[str, Any]) -> None: """Capture ASGI response messages emitted by auth middleware. Args: message: ASGI response message emitted by the auth stack. """ sent_messages.append(message) def _captured_response() -> Response: """Build a concrete response from the captured ASGI messages. Returns: Response reconstructed from the captured auth middleware output. """ status_code = 500 response_headers: Dict[str, str] = {} body = b"" for message in sent_messages: if message.get("type") == "http.response.start": status_code = int(message.get("status", 500)) response_headers = { key.decode("latin-1"): value.decode("latin-1") for key, value in message.get("headers", []) if isinstance(key, (bytes, bytearray)) and isinstance(value, (bytes, bytearray)) } elif message.get("type") == "http.response.body": body += message.get("body", b"") return Response(content=body, status_code=status_code, headers=response_headers) async def _call_next(_request: starletteRequest) -> Response: """Run the existing Streamable HTTP auth layer for the synthetic request. Returns: Success response when authentication passes, otherwise the captured failure response emitted by the existing middleware chain. """ auth_ok = await streamable_http_auth(scope, _receive, _send) if auth_ok: return ORJSONResponse(status_code=200, content={"authenticated": True}) return _captured_response() original_context = user_context_var.get() user_context_var.set({}) try: if settings.email_auth_enabled: response = await token_scoping_middleware(request, _call_next) else: response = await _call_next(request) if response is None: response = _captured_response() if response.status_code >= 400: return response, {} return None, get_streamable_http_auth_context() finally: user_context_var.set(original_context) def _normalize_token_teams(teams: Optional[List]) -> List[str]: """ Normalize token teams to list of team IDs. SSO tokens may contain team dicts like {"id": "...", "name": "..."}. This normalizes to just IDs for consistent filtering. Args: teams: Raw teams from token payload (may be None, list of IDs, or list of dicts) Returns: List of team ID strings (empty list if None) Examples: >>> from mcpgateway import main >>> main._normalize_token_teams(None) [] >>> main._normalize_token_teams([]) [] >>> main._normalize_token_teams(["team_a", "team_b"]) ['team_a', 'team_b'] >>> main._normalize_token_teams([{"id": "team_a", "name": "Team A"}]) ['team_a'] >>> main._normalize_token_teams([{"id": "t1"}, "t2", {"name": "no_id"}]) ['t1', 't2'] """ if not teams: return [] normalized = [] for team in teams: if isinstance(team, dict): team_id = team.get("id") if team_id: normalized.append(team_id) elif isinstance(team, str): normalized.append(team) return normalized def _build_rpc_permission_user(user, db: Session) -> dict[str, Any]: """Build PermissionChecker user payload for method-level RPC checks. Args: user: Authenticated user context. db: Active database session. Returns: Permission checker payload with email and ``db`` keys. """ permission_user = dict(user) if isinstance(user, dict) else {"email": get_user_email(user)} if not permission_user.get("email"): permission_user["email"] = get_user_email(user) permission_user["db"] = db return permission_user def _extract_scoped_permissions(request: Request) -> set[str] | None: """Extract token scopes.permissions from cached JWT payload. Args: request: Incoming request context. Returns: None: no explicit scope cap (empty permissions or no JWT — defer to RBAC) set: explicit permission set (may contain '*' for wildcard) """ internal_auth_context = get_internal_mcp_auth_context(request) if isinstance(internal_auth_context, dict): permissions = internal_auth_context.get("scoped_permissions") if not permissions: return None return set(permissions) cached = getattr(request.state, "_jwt_verified_payload", None) if not cached or not isinstance(cached, tuple) or len(cached) != 2: return None _, payload = cached if not payload or not isinstance(payload, dict): return None scopes = payload.get("scopes") if not scopes or not isinstance(scopes, dict): return None permissions = scopes.get("permissions") if not permissions: # Empty list or None = defer to RBAC return None return set(permissions) def _is_permission_admin_user(user) -> bool: """Return whether the caller already has permission-layer admin authority. This is stricter than token-scope admin semantics. It is used only to skip redundant RBAC DB lookups after token scope caps have already been enforced. Args: user: Authenticated user object or dict-like payload. Returns: ``True`` when the caller already has permission-layer admin authority. """ if hasattr(user, "is_admin"): return bool(getattr(user, "is_admin", False)) if isinstance(user, dict): if "permission_is_admin" in user: return bool(user.get("permission_is_admin", False)) return False return False async def _ensure_rpc_permission(user, db: Session, permission: str, method: str, request: Request | None = None) -> None: """Require a specific RPC permission for a method branch. Enforces both layers: 1. Token scopes.permissions cap (if explicit permissions present) 2. RBAC role-based permission check Args: user: Authenticated user context. db: Active database session. permission: Permission required for the method. method: JSON-RPC method name being authorized. request: Optional FastAPI request for extracting token scopes. Raises: JSONRPCError: If the requester lacks the required permission. """ # Trusted-internal public-only dispatch: the originating edge already applied public-only # visibility (and per-server OAuth), so an unauthenticated internal hop must not be re-denied # by RBAC. Mirrors _authorize_internal_mcp_request(). This only fires for HMAC-trusted internal # requests (the auth context is set on request.state only after the trust gate passes); the # public /rpc path and authenticated internal callers (is_authenticated True) fall through. if request is not None: _internal_ctx = get_internal_mcp_auth_context(request) if isinstance(_internal_ctx, dict) and _internal_ctx.get("is_authenticated", True) is False: return # Layer 1: Token scope cap if request is not None: scoped = _extract_scoped_permissions(request) if scoped is not None and "*" not in scoped and permission not in scoped: logger.warning("RPC permission denied (token scope): method=%s, required=%s", method, permission) raise JSONRPCError(-32003, _ACCESS_DENIED_MSG, {"method": method}) if permission == "admin.system_config" and _is_permission_admin_user(user): return # Layer 2: RBAC check # /rpc payloads never carry a resource with an owning team, so we skip # resource/payload derivation (unlike @require_permission). For single- # team API tokens we extract team_id from the token itself; otherwise # fall back to check_any_team so team-scoped roles are found. # Layer 1 (token scope cap above) already restricts visibility. team_id: str | None = None check_any_team = False if isinstance(user, dict): team_id = user.get("team_id") if not team_id: check_any_team = True checker = PermissionChecker(_build_rpc_permission_user(user, db)) if not await checker.has_permission(permission, check_any_team=check_any_team, team_id=team_id): logger.warning("RPC permission denied (RBAC): method=%s, required=%s", method, permission) raise JSONRPCError(-32003, _ACCESS_DENIED_MSG, {"method": method}) def _serialize_mcp_tool_definition(tool: Any) -> Dict[str, Any]: """Return an MCP-compliant tool definition without API-only metadata fields. Args: tool: Tool ORM object, pydantic model, or dict-like payload. Returns: MCP-compatible tool definition dictionary. """ if hasattr(tool, "model_dump"): data = tool.model_dump(by_alias=True, exclude_none=True) elif isinstance(tool, dict): data = dict(tool) else: data = {} name = data.get("name", getattr(tool, "name", None)) title = data.get("title", getattr(tool, "title", None)) description = data.get("description", getattr(tool, "description", None)) input_schema = data.get("inputSchema", getattr(tool, "input_schema", None)) payload: Dict[str, Any] = {} if name is not None: payload["name"] = name if title is not None: payload["title"] = title if description is not None or name is not None or input_schema is not None: payload["description"] = description or "" if input_schema is not None: payload["inputSchema"] = input_schema output_schema = data.get("outputSchema", getattr(tool, "output_schema", None)) if output_schema is not None: payload["outputSchema"] = output_schema annotations = data.get("annotations", getattr(tool, "annotations", None)) if annotations is not None: payload["annotations"] = annotations extension_metadata = data.get("extensionMetadata") or data.get("extension_metadata") or getattr(tool, "extension_metadata", None) apply_tool_meta(payload, extension_metadata) return {key: value for key, value in payload.items() if value is not None} def _serialize_mcp_tool_definitions(tools: List[Any]) -> List[Dict[str, Any]]: """Serialize tool records to MCP tool definitions. Args: tools: Iterable of tool-like records to serialize. Returns: List of MCP-compatible tool definitions. """ return [_serialize_mcp_tool_definition(tool) for tool in filter_model_visible_tools(tools)] def _serialize_legacy_tool_payloads(tools: List[Any]) -> List[Dict[str, Any]]: """Serialize tool records using the legacy JSON-RPC shape. Args: tools: Iterable of tool-like records to serialize. Returns: List of legacy tool payload dictionaries. """ payloads: List[Dict[str, Any]] = [] for tool in filter_model_visible_tools(tools): if hasattr(tool, "model_dump"): payload = tool.model_dump(by_alias=True, exclude_none=True) elif isinstance(tool, dict): payload = dict(tool) else: payload = {} payloads.append(payload) return payloads def _enforce_scoped_resource_access(request: Request, db: Session, user, resource_path: str) -> None: """Apply token-scope ownership checks for a concrete resource path. This provides defense-in-depth for ID-based handlers so they continue to enforce visibility even if middleware coverage regresses. Args: request: Incoming request context. db: Active database session. user: Authenticated user context. resource_path: Canonical resource path (e.g. ``/tools/{id}``). Raises: HTTPException: If access to the target resource is not allowed. """ scoped_user_email, scoped_token_teams = get_scoped_resource_access_context(request, user) # Admin bypass / unrestricted scope if scoped_token_teams is None: return if ( token_scoping_middleware._check_resource_team_ownership( # pylint: disable=protected-access resource_path, scoped_token_teams, db=db, _user_email=scoped_user_email, ) is not ResourceOwnershipResult.ALLOWED ): logger.warning("Scoped resource access denied: user=%s, resource=%s", scoped_user_email, resource_path) raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=_ACCESS_DENIED_MSG) async def _assert_session_owner_or_admin(request: Request, user, session_id: str) -> None: """Ensure session operations are limited to the owner unless requester is admin. Args: request: Incoming request context. user: Authenticated user context. session_id: Target session identifier. Raises: HTTPException: If session is missing or requester is not authorized. """ session_owner = await session_registry.get_session_owner(session_id) if not session_owner: session_exists = await session_registry.session_exists(session_id) if session_exists is False: raise HTTPException(status_code=404, detail="Session not found") raise HTTPException(status_code=403, detail="Session owner metadata unavailable") requester_email, requester_is_admin = get_request_identity(request, user) if requester_is_admin: return if requester_email and requester_email == session_owner: return raise HTTPException(status_code=403, detail="Session access denied") async def _authorize_run_cancellation(request: Request, user, request_id: str, *, as_jsonrpc_error: bool) -> None: """Authorize a notifications/cancelled request for a specific run id. Args: request: Incoming request context. user: Authenticated user context. request_id: Run/request identifier to cancel. as_jsonrpc_error: Raise ``JSONRPCError`` when True, otherwise ``HTTPException``. Raises: JSONRPCError: When ``as_jsonrpc_error`` is True and cancellation is not authorized. HTTPException: When ``as_jsonrpc_error`` is False and cancellation is not authorized. """ # Layer-1 exception: compares requester against run owner, so it needs the raw # token teams and is_admin flag rather than the normalized visibility scope. requester_email, requester_token_teams, requester_is_admin = get_rpc_filter_context(request, user) requester_teams = [] if requester_token_teams is None else list(requester_token_teams) run_status = await cancellation_service.get_status(request_id) if run_status is None: # Notifications are best-effort; unknown request ids should be accepted # as no-ops rather than rejected as authorization failures. return run_owner_email = run_status.get("owner_email") run_owner_team_ids = run_status.get("owner_team_ids") or [] requester_is_owner = bool(run_owner_email and requester_email and run_owner_email == requester_email) requester_shares_team = bool(run_owner_team_ids and requester_teams and any(team in run_owner_team_ids for team in requester_teams)) unauthorized = not requester_is_admin and not requester_is_owner and not requester_shares_team if unauthorized: if as_jsonrpc_error: raise JSONRPCError(-32003, "Not authorized to cancel this run", {"requestId": request_id}) raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Not authorized to cancel this run") # Initialize cache resource_cache = ResourceCache(max_size=settings.resource_cache_size, ttl=settings.resource_cache_ttl) def _rust_build_included() -> bool: """Return whether the current image includes Rust MCP artifacts. Returns: ``True`` when the current image contains the Rust MCP binaries/plugins. """ return version_module.rust_build_included() def _rust_runtime_managed() -> bool: """Return whether the gateway expects to manage the Rust MCP sidecar locally. Returns: ``True`` when the gateway should launch and supervise the Rust sidecar. """ return version_module.rust_runtime_managed() def _current_mcp_transport_mount() -> str: """Return which public /mcp transport is currently mounted. Returns: Runtime label identifying the currently mounted public MCP transport. """ return version_module.current_mcp_transport_mount() def _should_mount_public_rust_transport() -> bool: """Return whether the public ``/mcp`` path should be served directly by Rust. Returns: ``True`` only when the Rust runtime is enabled and the session-auth reuse path is enabled, allowing Rust to safely own steady-state public MCP session traffic. Otherwise returns ``False`` and leaves public MCP on the Python ingress path. """ return version_module.should_mount_public_rust_transport() def _should_use_rust_public_session_stack() -> bool: """Return whether Rust should own the effective public MCP session stack. Returns: ``True`` only when the Rust runtime is enabled and session-auth reuse is enabled, allowing the public transport, session metadata, replay/resume, live-stream, and affinity behavior to stay on a consistent Rust-backed path. Otherwise returns ``False`` so the public MCP session stack falls back to Python semantics. """ return version_module.should_use_rust_public_session_stack() def _current_mcp_runtime_mode() -> str: """Return a compact runtime-mode label for observability. Returns: Human-readable runtime mode label for health/readiness reporting. """ return version_module.current_mcp_runtime_mode() def _current_mcp_session_core_mode() -> str: """Return which session core currently owns MCP session metadata. Returns: ``"rust"`` when the Rust session core is enabled, otherwise ``"python"``. """ return version_module.current_mcp_session_core_mode() def _current_mcp_event_store_mode() -> str: """Return which runtime currently owns MCP resumable event-store semantics. Returns: ``"rust"`` when the Rust event store is enabled, otherwise ``"python"``. """ return version_module.current_mcp_event_store_mode() def _current_mcp_resume_core_mode() -> str: """Return which runtime currently owns public MCP replay/resume behavior. Returns: ``"rust"`` when Rust owns replay/resume, otherwise ``"python"``. """ return version_module.current_mcp_resume_core_mode() def _current_mcp_live_stream_core_mode() -> str: """Return which runtime currently owns non-resume public GET /mcp SSE behavior. Returns: ``"rust"`` when Rust owns live GET /mcp streaming, otherwise ``"python"``. """ return version_module.current_mcp_live_stream_core_mode() def _current_mcp_affinity_core_mode() -> str: """Return which runtime currently owns MCP multi-worker session-affinity forwarding. Returns: ``"rust"`` when Rust owns session-affinity forwarding, otherwise ``"python"``. """ return version_module.current_mcp_affinity_core_mode() def _current_mcp_session_auth_reuse_mode() -> str: """Return which runtime currently owns MCP session-bound auth-context reuse. Returns: ``"rust"`` when Rust session auth reuse is enabled, otherwise ``"python"``. """ return version_module.current_mcp_session_auth_reuse_mode() def _mcp_runtime_status_payload() -> Dict[str, Any]: """Return MCP runtime diagnostics for health/readiness endpoints. Returns: Diagnostic payload describing the active MCP runtime configuration. """ return version_module.mcp_runtime_status_payload() def _apply_runtime_mode_headers(response: Response) -> None: """Attach MCP runtime mode headers to a response. Args: response: Response object to annotate. """ response.headers["x-contextforge-mcp-runtime-mode"] = _current_mcp_runtime_mode() response.headers["x-contextforge-mcp-transport-mounted"] = _current_mcp_transport_mount() response.headers["x-contextforge-rust-build-included"] = "true" if _rust_build_included() else "false" response.headers["x-contextforge-mcp-session-core-mode"] = _current_mcp_session_core_mode() response.headers["x-contextforge-mcp-event-store-mode"] = _current_mcp_event_store_mode() response.headers["x-contextforge-mcp-resume-core-mode"] = _current_mcp_resume_core_mode() response.headers["x-contextforge-mcp-live-stream-core-mode"] = _current_mcp_live_stream_core_mode() response.headers["x-contextforge-mcp-affinity-core-mode"] = _current_mcp_affinity_core_mode() response.headers["x-contextforge-mcp-session-auth-reuse-mode"] = _current_mcp_session_auth_reuse_mode() # Type aliases for improved readability ToolsResponse: TypeAlias = Union[List[ToolRead], CursorPaginatedToolsResponse, List[Dict[Any, Any]], Dict[Any, Any], ORJSONResponse] ToolResponse: TypeAlias = Union[ToolRead, Dict[Any, Any], ORJSONResponse] @lru_cache(maxsize=512) def _parse_jsonpath(jsonpath: str) -> JSONPath: """Cache parsed JSONPath expression. Args: jsonpath: The JSONPath expression string. Returns: Parsed JSONPath object. Raises: Exception: If the JSONPath expression is invalid. """ return parse(jsonpath) def _parse_apijsonpath(raw: Optional[Union[str, JsonPathModifier]]) -> Optional[JsonPathModifier]: """ Parse apijsonpath parameter from either a JSON string or a JsonPathModifier model. Performs early validation of JSONPath syntax to fail fast and provide clear error messages. Args: raw: Either a JSON-encoded string or a JsonPathModifier instance Returns: Parsed JsonPathModifier or None if raw is None Raises: HTTPException: If the JSON string is invalid, unexpected type provided, jsonpath expression is empty, or JSONPath syntax is invalid (400 Bad Request) """ if raw is None: return None if isinstance(raw, str): try: parsed = JsonPathModifier.model_validate(json.loads(raw)) # Validate jsonpath is not empty if provided if parsed.jsonpath is not None: if not parsed.jsonpath.strip(): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="JSONPath expression cannot be empty") # Early validation: ensure JSONPath syntax is valid try: _parse_jsonpath(parsed.jsonpath) except Exception as parse_ex: detail = f"Invalid JSONPath syntax: {parse_ex}" if settings.log_level == "DEBUG" else "Invalid JSONPath expression" raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=detail) return parsed except HTTPException: # Re-raise HTTPException as-is (includes empty jsonpath and syntax validation) raise except json.JSONDecodeError as ex: # User error: malformed JSON (JSONDecodeError is subclass of ValueError, so catch it specifically) detail = f"Invalid apijsonpath JSON: {ex}" if settings.log_level == "DEBUG" else "Invalid apijsonpath format" raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=detail) except ValidationError as ex: # Pydantic validation error detail = f"Invalid apijsonpath structure: {ex}" if settings.log_level == "DEBUG" else "Invalid apijsonpath structure" raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=detail) except Exception as ex: # Unexpected error - log it and return generic message logger.error(f"Unexpected error parsing apijsonpath: {ex}", exc_info=True) raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to parse apijsonpath") elif isinstance(raw, JsonPathModifier): # Validate jsonpath is not empty if provided if raw.jsonpath is not None: if not raw.jsonpath.strip(): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="JSONPath expression cannot be empty") # Early validation: ensure JSONPath syntax is valid try: _parse_jsonpath(raw.jsonpath) except Exception as parse_ex: detail = f"Invalid JSONPath syntax: {parse_ex}" if settings.log_level == "DEBUG" else "Invalid JSONPath expression" raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=detail) return raw # Unexpected type - fail fast with clear error message # Only show type name in debug mode to avoid information disclosure type_info = f": got {type(raw).__name__}" if settings.log_level == "DEBUG" else "" raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"Invalid apijsonpath type{type_info}") def jsonpath_modifier(data: Any, jsonpath: str = "$[*]", mappings: Optional[Dict[str, str]] = None) -> Union[List, Dict]: """ Applies the given JSONPath expression and mappings to the data. Uses cached parsed expressions for performance. Args: data: The JSON data to query. jsonpath: The JSONPath expression to apply. mappings: Optional dictionary of mappings where keys are new field names and values are JSONPath expressions. Returns: Union[List, Dict]: A list (or mapped list) or a Dict of extracted data. Raises: HTTPException: If there's an error parsing or executing the JSONPath expressions. Examples: >>> jsonpath_modifier({'a': 1, 'b': 2}, '$.a') [1] >>> jsonpath_modifier([{'a': 1}, {'a': 2}], '$[*].a') [1, 2] >>> jsonpath_modifier({'a': {'b': 2}}, '$.a.b') [2] >>> jsonpath_modifier({'a': 1}, '$.b') [] """ if not jsonpath: jsonpath = "$[*]" # Log jsonpath_modifier invocation with structured data (only if debug enabled) if logger.isEnabledFor(logging.DEBUG): data_length = len(data) if isinstance(data, list) else None logger.debug(f"jsonpath_modifier: path='{SecurityValidator.sanitize_log_message(jsonpath)}', has_mappings={mappings is not None}, data_type={type(data).__name__}, data_length={data_length}") try: main_expr: JSONPath = _parse_jsonpath(jsonpath) except Exception as e: logger.debug("Invalid main JSONPath expression: %s", e) raise HTTPException(status_code=400, detail="Invalid JSONPath expression") try: main_matches = main_expr.find(data) except Exception as e: logger.debug("Error executing main JSONPath: %s", e) raise HTTPException(status_code=400, detail="Error executing JSONPath expression") results = [match.value for match in main_matches] if mappings: results = transform_data_with_mappings(results, mappings) if len(results) == 1 and isinstance(results[0], dict): return results[0] return results def transform_data_with_mappings(data: list[Any], mappings: dict[str, str]) -> list[Any]: """ Applies mappings to data using cached JSONPath expressions. Parses each mapping expression once per call, not per item. Args: data: The set of data to apply mappings to. mappings: dictionary of mappings where keys are new field names Returns: list[Any]: A list (or mapped list) of re-mapped data Raises: HTTPException: If there's an error parsing or executing the JSONPath expressions. Examples: >>> transform_data_with_mappings([{'first_name': "Bruce", 'second_name': "Wayne"},{'first_name': "Diana", 'second_name': "Prince"}], {"n": "$.first_name"}) [{'n': 'Bruce'}, {'n': 'Diana'}] """ # Pre-parse all mapping expressions once (not per item) parsed_mappings: Dict[str, JSONPath] = {} for new_key, mapping_expr_str in mappings.items(): try: parsed_mappings[new_key] = _parse_jsonpath(mapping_expr_str) except Exception as e: logger.debug("Invalid mapping JSONPath for key '%s': %s", new_key, e) raise HTTPException(status_code=400, detail=f"Invalid JSONPath expression for key '{new_key}'") mapped_results = [] for item in data: mapped_item = {} for new_key, mapping_expr in parsed_mappings.items(): try: mapping_matches = mapping_expr.find(item) except Exception as e: logger.debug("Error executing mapping JSONPath for key '%s': %s", new_key, e) raise HTTPException(status_code=400, detail=f"Error executing JSONPath expression for key '{new_key}'") if not mapping_matches: mapped_item[new_key] = None elif len(mapping_matches) == 1: mapped_item[new_key] = mapping_matches[0].value else: mapped_item[new_key] = [m.value for m in mapping_matches] mapped_results.append(mapped_item) return mapped_results async def attempt_to_bootstrap_sso_providers(): """ Try to bootstrap SSO provider services based on settings. """ try: # First-Party from mcpgateway.utils.sso_bootstrap import bootstrap_sso_providers # pylint: disable=import-outside-toplevel await bootstrap_sso_providers() logger.info("SSO providers bootstrapped successfully") except Exception as e: logger.warning(f"Failed to bootstrap SSO providers: {e}") #################### # Startup/Shutdown # #################### def _can_manage_sighup_handler() -> bool: """Return whether this runtime context can safely install process signal handlers. Returns: ``True`` when startup is running on the process main thread and SIGHUP is available. """ return hasattr(signal, "SIGHUP") and threading.current_thread() is threading.main_thread() def _install_sighup_handler() -> bool: """Install the SIGHUP handler when the current runtime context supports it. Returns: ``True`` when the handler was installed in the current runtime context. """ if not _can_manage_sighup_handler(): logger.debug("Skipping SIGHUP handler registration outside the main thread") return False # First-Party from mcpgateway.handlers.signal_handlers import sighup_handler # pylint: disable=import-outside-toplevel signal.signal(signal.SIGHUP, sighup_handler) return True def _restore_default_sighup_handler() -> None: """Restore the default SIGHUP handler when the current runtime context supports it. Returns: ``None``. """ if not _can_manage_sighup_handler(): return signal.signal(signal.SIGHUP, signal.SIG_DFL) @asynccontextmanager async def lifespan(_app: FastAPI) -> AsyncIterator[None]: """ Manage the application's startup and shutdown lifecycle. The function initialises every core service on entry and then shuts them down in reverse order on exit. Args: _app (FastAPI): FastAPI app Yields: None Raises: SystemExit: When a critical startup error occurs that prevents the application from starting successfully. Exception: Any unhandled error that occurs during service initialisation or shutdown is re-raised to the caller. """ aggregation_stop_event: Optional[asyncio.Event] = None aggregation_loop_task: Optional[asyncio.Task] = None aggregation_backfill_task: Optional[asyncio.Task] = None siem_export_service: Optional[Any] = None dataplane_publisher_service: Optional[Any] = None # Initialize logging service FIRST to ensure all logging goes to dual output await logging_service.initialize() logger.info("Starting ContextForge services") # Start the sandboxed jq worker pool before any service that might invoke # tool response filters is initialised. A failure here (BrokenProcessPool, # TimeoutError, or OSError from the sandbox warm-up on Linux/subprocess # mode) is a hard startup failure and must propagate rather than letting # the gateway boot with a broken or absent sandbox. start_jq_pool() # Wait for the database to be ready, then run bootstrap (alembic + seed). # This used to run at module-import time, which made every test that # imported mcpgateway.main pay for a real DB probe and migration check. # `wait_for_db_ready(sync=True)` is a blocking probe, so offload it to # a worker thread to avoid stalling the event loop during startup. # First-Party from mcpgateway.bootstrap_db import main as bootstrap_db # pylint: disable=import-outside-toplevel from mcpgateway.utils.db_isready import wait_for_db_ready # pylint: disable=import-outside-toplevel await asyncio.to_thread( wait_for_db_ready, max_tries=int(settings.db_max_retries), interval=int(settings.db_retry_interval_ms) / 1000, sync=True, ) await bootstrap_db() # Initialize Redis client early (shared pool for all services) await get_redis_client() # Register the Redis provider with the plugin framework so framework # modules can reach Redis without importing mcpgateway.utils directly # (isolation enforced by scripts/pre-commit/check_framework_imports.py). # First-Party from mcpgateway.plugins._redis import set_shared_redis_provider # pylint: disable=import-outside-toplevel set_shared_redis_provider(get_redis_client) # Initialize SIEM export service early so security/audit events can flow from startup. # First-Party from mcpgateway.services.siem_export_service import get_siem_export_service # pylint: disable=import-outside-toplevel siem_export_service = get_siem_export_service() await siem_export_service.initialize() # Initialize rate limiter Redis early for validation # First-Party from mcpgateway.auth import _get_ratelimiter_redis_client # pylint: disable=import-outside-toplevel if settings.ratelimiter_redis_url: _get_ratelimiter_redis_client() # Triggers lazy init + logging # Initialize shared HTTP client (connection pool for all outbound requests) # First-Party from mcpgateway.services.http_client_service import SharedHttpClient # pylint: disable=import-outside-toplevel await SharedHttpClient.get_instance() # Update HTTP pool metrics after SharedHttpClient is initialized if hasattr(app.state, "update_http_pool_metrics"): app.state.update_http_pool_metrics() # Initialize the session-affinity service (Redis-backed worker mapping, # heartbeat, session-owner forwarding). Cross-worker upstream # ``ClientSession`` state lives in ``UpstreamSessionRegistry`` — # SessionAffinity owns only the affinity layer. # Always initialize SessionAffinity, regardless of # ``mcpgateway_session_affinity_enabled``. The affinity flag controls # cross-worker Redis routing; the GET /mcp listener-claim dict # (ADR-052) is single-node bookkeeping that lives on the same instance # and needs to be a process-wide singleton even with affinity off. # Without the always-init, the GET handler would fall back to a fresh # ``SessionAffinity()`` per request → each request gets its own # ``_listener_claims`` dict → two concurrent GETs both win the claim # and the single-listener invariant is broken in the default # single-node deployment. # # The Redis-backed background tasks (heartbeat, RPC listener) are # internally gated on the affinity flag, so always-init is safe. # First-Party from mcpgateway.services.session_affinity import init_session_affinity # pylint: disable=import-outside-toplevel # Use enable_notifications=False here so we don't double-init the # notification service — main.lifespan does that explicitly below. init_session_affinity(enable_notifications=False) logger.info( "Session-affinity service initialized (affinity_enabled=%s)", settings.mcpgateway_session_affinity_enabled, ) # Initialize the upstream session registry (#4205). 1:1 binding between a # downstream MCP session and its upstream session per gateway, replacing # the old identity-keyed sharing semantics. Always on — no feature flag. # # Wire a notification handler factory so server-initiated messages can be # forwarded to GET /mcp listeners (ADR-052). The factory bakes # downstream_session_id into the per-session handler closure. # First-Party from mcpgateway.services.notification_service import init_notification_service # pylint: disable=import-outside-toplevel from mcpgateway.services.upstream_session_registry import init_upstream_session_registry # pylint: disable=import-outside-toplevel _notification_svc = init_notification_service() # Initialize the worker before any session is wired through it. # Without this, list_changed notifications enqueue into a service # whose `_process_refresh_queue` worker never runs and refreshes # silently never fire. The gateway_service is wired here # unconditionally — the affinity branch's # `start_affinity_notification_service` only re-runs under affinity, # and without this hand-off the single-node default would leave # `_gateway_service is None` and drop every refresh. await _notification_svc.initialize(gateway_service=gateway_service) def _notification_handler_factory(url: str, gateway_id, *, downstream_session_id: str): # type: ignore[no-untyped-def] """Per-session message handler that routes upstream notifications and forwards server-initiated messages to the GET /mcp listener (ADR-052).""" return _notification_svc.create_message_handler( gateway_id or url, url, downstream_session_id=downstream_session_id, ) init_upstream_session_registry(message_handler_factory=_notification_handler_factory) logger.info("Upstream session registry initialized (notification fanout enabled)") # Initialize LLM chat router Redis client (only if LLM chat is enabled — # importing the router pulls in the langchain stack which is several # seconds of cold-start cost). if settings.llmchat_enabled: # First-Party from mcpgateway.routers.llmchat_router import init_redis as init_llmchat_redis # pylint: disable=import-outside-toplevel await init_llmchat_redis() try: # Validate security configuration validate_security_configuration() # Validate UAID security configuration validate_uaid_security_config() # Initialize the plugin manager factory whenever the YAML config is # available. We used to gate this on ``settings.plugins.enabled`` but # that broke runtime enable-from-disabled: a node that boots with the # flag off would never create the factory, so a later shared-toggle # flip to "enabled" from a peer worker left this node unable to run # plugins until restart. Keep initialisation unconditional; gate # *execution* on the shared toggle in ``get_plugin_manager``. # First-Party from mcpgateway.plugins.policy import HOOK_PAYLOAD_POLICIES # pylint: disable=import-outside-toplevel # Start the primary-worker elector before plugins initialize, since a # non-hook plugin may call is_primary_worker() in initialize(). Only the # redis backend needs an elector; the filelock backend stays lazy. if settings.primary_worker_election_backend == "redis": # First-Party from mcpgateway.services.leader_election import start_primary_worker_elector # pylint: disable=import-outside-toplevel await start_primary_worker_elector() logger.info("Primary-worker elector started (backend=redis)") try: init_plugin_manager_factory( yaml_path=settings.plugins.config_file, timeout=settings.plugins.plugin_timeout, hook_policies=HOOK_PAYLOAD_POLICIES, observability=None, # Will be set later if needed db_factory=SessionLocal, ) logger.info("Plugin manager factory initialized") except Exception as init_exc: if settings.plugins.enabled: # Operator asked for plugins — a failed init (bad YAML, missing # plugin module, validation error) must be a hard boot failure # rather than a silent no-op. Preserve the original loud-crash # semantics; the outer lifespan handler logs and re-raises. logger.error("Plugin manager factory initialization failed: %s", init_exc, exc_info=True) raise # Plugins disabled locally; we init opportunistically so a later # shared-toggle flip from a peer worker can turn the subsystem on # without restarting this node. If that opportunistic init fails, # the gateway still boots — but mark the node degraded so # ``get_plugin_manager`` emits an ERROR the first time the shared # toggle asks us to serve plugins we can't actually run. logger.warning( "Plugin manager factory init failed (%s); runtime-enable from a peer worker will require this node to restart", init_exc, ) # First-Party from mcpgateway.plugins import mark_factory_init_degraded # pylint: disable=import-outside-toplevel mark_factory_init_degraded() # Load SpanAttributeCustomizer baggage emission policy before telemetry starts # creating spans. Baggage remains the propagation mechanism; this policy # controls the span attribute names exported from allowed baggage keys. configure_baggage_span_attribute_policy(extract_baggage_span_attribute_policy(get_plugin_manager_factory())) init_telemetry() logger.info("Observability initialized") try: plugin_manager = await get_plugin_manager() if plugin_manager: logger.info(f"Plugin manager initialized with {plugin_manager.plugin_count} plugins") # Wire plugin manager to plugin service for admin endpoints # First-Party from mcpgateway.services.plugin_service import get_plugin_service # pylint: disable=import-outside-toplevel plugin_service = get_plugin_service() plugin_service.set_plugin_manager(plugin_manager) # Expose on app.state so the admin UI can show the correct enabled status app.state.plugin_manager = plugin_manager except Exception as diag_exc: logger.error(f"Plugin manager initialization failed: {diag_exc}", exc_info=True) raise # Always start the invalidation listener when a factory is live, even # if the local ``plugins.enabled`` setting is false. The listener # early-exits when no Redis provider is registered, so single-node # deployments don't spin. await start_plugin_invalidation_listener() logger.info("Plugin invalidation listener started") # Wire observability adapter to plugin manager if observability is enabled if settings.observability_enabled and _service is not None: # pylint: disable=possibly-used-before-assignment # First-Party from mcpgateway.plugins import set_global_observability # pylint: disable=import-outside-toplevel from mcpgateway.plugins.observability_adapter import ObservabilityServiceAdapter # pylint: disable=import-outside-toplevel set_global_observability(ObservabilityServiceAdapter(service=_service)) logger.info("🔍 Plugin observability adapter wired to ObservabilityService") if settings.enable_header_passthrough: await setup_passthrough_headers() else: logger.info("🔒 Header Passthrough: DISABLED") await tool_service.initialize() await resource_service.initialize() await prompt_service.initialize() await gateway_service.initialize() # Start heartbeat, RPC listener, and notification service for # multi-worker session affinity. The upstream-session pool is # owned by ``UpstreamSessionRegistry`` and runs unconditionally; # only the cross-worker affinity machinery is gated here. if settings.mcpgateway_session_affinity_enabled: # First-Party from mcpgateway.services.session_affinity import get_session_affinity, start_affinity_notification_service # pylint: disable=import-outside-toplevel await start_affinity_notification_service(gateway_service) pool = get_session_affinity() pool.start_heartbeat() pool._rpc_listener_task = asyncio.create_task(pool.start_rpc_listener()) # pylint: disable=protected-access logger.info("Multi-worker session affinity heartbeat and RPC listener started") await root_service.initialize() await completion_service.initialize() await sampling_handler.initialize() await export_service.initialize() await import_service.initialize() if a2a_service: await a2a_service.initialize() await resource_cache.initialize() await streamable_http_session.initialize() await session_registry.initialize() # Initialize OrchestrationService for tool cancellation if enabled if settings.mcpgateway_tool_cancellation_enabled: await cancellation_service.initialize() logger.info("Tool cancellation feature enabled") else: logger.info("Tool cancellation feature disabled") # Initialize elicitation service if settings.mcpgateway_elicitation_enabled: # First-Party from mcpgateway.services.elicitation_service import get_elicitation_service # pylint: disable=import-outside-toplevel elicitation_service = get_elicitation_service() await elicitation_service.start() logger.info("Elicitation service initialized") # Initialize metrics buffer service for batching metric writes if settings.metrics_buffer_enabled: # First-Party from mcpgateway.services.metrics_buffer_service import get_metrics_buffer_service # pylint: disable=import-outside-toplevel metrics_buffer_service = get_metrics_buffer_service() await metrics_buffer_service.start() if settings.db_metrics_recording_enabled: logger.info("Metrics buffer service initialized") else: logger.info("Metrics buffer service initialized (recording disabled)") # Initialize metrics cleanup service for automatic deletion of old metrics if settings.metrics_cleanup_enabled: # First-Party from mcpgateway.services.metrics_cleanup_service import get_metrics_cleanup_service # pylint: disable=import-outside-toplevel metrics_cleanup_service = get_metrics_cleanup_service() await metrics_cleanup_service.start() logger.info("Metrics cleanup service initialized (retention: %d days)", settings.metrics_retention_days) # Initialize MCP Apps session cleanup service for automatic deletion of expired AppBridge sessions if settings.mcpgateway_mcp_apps_enabled and settings.mcpgateway_mcp_apps_session_cleanup_enabled: mcp_app_session_cleanup_service = get_mcp_app_session_cleanup_service() await mcp_app_session_cleanup_service.start() logger.info("MCP Apps session cleanup service initialized") # Initialize metrics rollup service for hourly aggregation if settings.metrics_rollup_enabled: # First-Party from mcpgateway.services.metrics_rollup_service import get_metrics_rollup_service # pylint: disable=import-outside-toplevel metrics_rollup_service = get_metrics_rollup_service() await metrics_rollup_service.start() logger.info("Metrics rollup service initialized (interval: %dh)", settings.metrics_rollup_interval_hours) refresh_slugs_on_startup() # Initialize experimental dataplane publisher to send config data to redis if settings.dataplane_publisher: dataplane_publisher_service = DataplanePublisherService() await dataplane_publisher_service.start() # Bootstrap SSO providers from environment configuration if settings.sso_enabled: await attempt_to_bootstrap_sso_providers() logger.info("All services initialized successfully") # Warn about per-worker database connection pool multiplication if os.environ.get("GUNICORN_CMD_ARGS") or os.environ.get("GUNICORN_WORKERS"): cpu_count = multiprocessing.cpu_count() default_workers = min(2 * cpu_count + 1, 16) workers = int(os.environ.get("GUNICORN_WORKERS", str(default_workers))) total_pool = settings.db_pool_size + settings.db_max_overflow total_connections = workers * total_pool logger.warning( "⚠️ DATABASE POOL: Running with %d gunicorn workers. Total max DB connections = workers(%d) * (pool_size + max_overflow) = %d * %d = %d. Ensure PostgreSQL max_connections >= %d. ", workers, workers, workers, total_pool, total_connections, total_connections, ) # Warn about unsafe UAID configuration if A2A is enabled if settings.mcpgateway_a2a_enabled: uaid_allowed_domains = getattr(settings, "uaid_allowed_domains", []) if not uaid_allowed_domains: logger.warning( "⚠️ SECURITY: UAID_ALLOWED_DOMAINS is empty - cross-gateway routing is unrestricted. " "This allows UAID-based routing to ANY domain, including internal networks. " "Production deployments MUST configure UAID_ALLOWED_DOMAINS to restrict routing to trusted domains only. " 'Example: UAID_ALLOWED_DOMAINS=["trusted.example.com","gateway.example.org"]' ) _install_sighup_handler() # Start cache invalidation subscriber for cross-worker cache synchronization # First-Party from mcpgateway.cache.registry_cache import get_cache_invalidation_subscriber # pylint: disable=import-outside-toplevel cache_invalidation_subscriber = get_cache_invalidation_subscriber() await cache_invalidation_subscriber.start() # Start runtime-mode coordinator for cluster-wide override propagation # First-Party from mcpgateway.runtime_state import get_runtime_state_coordinator # pylint: disable=import-outside-toplevel runtime_state_coordinator = get_runtime_state_coordinator() await runtime_state_coordinator.start() # Reconfigure uvicorn loggers after startup to capture access logs in dual output logging_service.configure_uvicorn_after_startup() if settings.metrics_aggregation_enabled and settings.metrics_aggregation_auto_start: aggregation_stop_event = asyncio.Event() log_aggregator = get_log_aggregator() async def run_log_backfill() -> None: """Backfill log aggregation metrics for configured hours.""" hours = getattr(settings, "metrics_aggregation_backfill_hours", 0) if hours <= 0: return try: await asyncio.to_thread(log_aggregator.backfill, hours) logger.info("Log aggregation backfill completed for last %s hour(s)", hours) except Exception as backfill_error: # pragma: no cover - defensive logging logger.warning("Log aggregation backfill failed: %s", backfill_error) async def run_log_aggregation_loop() -> None: """Run continuous log aggregation at configured intervals. Raises: asyncio.CancelledError: When aggregation is stopped """ interval_seconds = settings.metrics_aggregation_interval_seconds or max(1, int(settings.metrics_aggregation_window_minutes)) * 60 logger.info( "Starting log aggregation loop (window=%s min)", log_aggregator.aggregation_window_minutes, ) try: while not aggregation_stop_event.is_set(): try: await asyncio.to_thread(log_aggregator.aggregate_all_components) except Exception as agg_error: # pragma: no cover - defensive logging logger.warning("Log aggregation loop iteration failed: %s", agg_error) try: await asyncio.wait_for(aggregation_stop_event.wait(), timeout=interval_seconds) except asyncio.TimeoutError: continue except asyncio.CancelledError: logger.debug("Log aggregation loop cancelled") raise finally: logger.info("Log aggregation loop stopped") aggregation_backfill_task = asyncio.create_task(run_log_backfill()) aggregation_loop_task = asyncio.create_task(run_log_aggregation_loop()) elif settings.metrics_aggregation_enabled: logger.info("Metrics aggregation auto-start disabled; performance metrics will be generated on-demand when requested.") yield except Exception as e: logger.error(f"Error during startup: {str(e)}") # For plugin errors, exit cleanly without stack trace spam if "Plugin initialization failed" in str(e): # Suppress uvicorn error logging for clean exit logging.getLogger("uvicorn.error").setLevel(logging.CRITICAL) raise SystemExit(1) raise finally: # Restore default SIGHUP handling in case we reset signal handlers. try: _restore_default_sighup_handler() except Exception as exc: # pragma: no cover - defensive logger.debug(f"Failed to restore default SIGHUP handler: {exc}") if aggregation_stop_event is not None: aggregation_stop_event.set() for task in (aggregation_backfill_task, aggregation_loop_task): if task: task.cancel() with suppress(asyncio.CancelledError): await task # Stop the plugin invalidation listener before the factory so in-flight # messages don't race with a half-torn-down cache. try: await stop_plugin_invalidation_listener() except Exception as e: logger.debug(f"Error stopping plugin invalidation listener: {e}") # Shutdown global plugin manager factory (no-op when plugins were never initialised) try: await shutdown_plugin_manager_factory() logger.info("Plugin manager shutdown complete") except Exception as e: logger.error(f"Error shutting down plugin manager: {str(e)}") # Stop cache invalidation subscriber try: # First-Party from mcpgateway.cache.registry_cache import get_cache_invalidation_subscriber # pylint: disable=import-outside-toplevel cache_invalidation_subscriber = get_cache_invalidation_subscriber() await cache_invalidation_subscriber.stop() except Exception as e: logger.debug(f"Error stopping cache invalidation subscriber: {e}") # Stop runtime-mode coordinator try: # First-Party from mcpgateway.runtime_state import get_runtime_state_coordinator # pylint: disable=import-outside-toplevel await get_runtime_state_coordinator().stop() except Exception as e: logger.debug(f"Error stopping runtime-mode coordinator: {e}") logger.info("Shutting down ContextForge services") # await stop_streamablehttp() # Build service list conditionally services_to_shutdown: List[Any] = [ resource_cache, sampling_handler, import_service, export_service, logging_service, completion_service, root_service, gateway_service, prompt_service, resource_service, tool_service, streamable_http_session, session_registry, ] if siem_export_service is not None: services_to_shutdown.insert(0, siem_export_service) # Add cancellation service if enabled if settings.mcpgateway_tool_cancellation_enabled: services_to_shutdown.insert(0, cancellation_service) # Shutdown early to stop accepting new cancellations if a2a_service: services_to_shutdown.insert(4, a2a_service) # Insert after export_service # Add elicitation service if enabled if settings.mcpgateway_elicitation_enabled: # First-Party from mcpgateway.services.elicitation_service import get_elicitation_service # pylint: disable=import-outside-toplevel elicitation_service = get_elicitation_service() services_to_shutdown.insert(5, elicitation_service) # Add metrics buffer service if enabled (flush remaining metrics before shutdown) if settings.metrics_buffer_enabled: # First-Party from mcpgateway.services.metrics_buffer_service import get_metrics_buffer_service # pylint: disable=import-outside-toplevel metrics_buffer_service = get_metrics_buffer_service() services_to_shutdown.insert(0, metrics_buffer_service) # Shutdown first to flush metrics # Add metrics rollup service if enabled (shutdown before cleanup) if settings.metrics_rollup_enabled: # First-Party from mcpgateway.services.metrics_rollup_service import get_metrics_rollup_service # pylint: disable=import-outside-toplevel metrics_rollup_service = get_metrics_rollup_service() services_to_shutdown.insert(1, metrics_rollup_service) # Add metrics cleanup service if enabled if settings.metrics_cleanup_enabled: # First-Party from mcpgateway.services.metrics_cleanup_service import get_metrics_cleanup_service # pylint: disable=import-outside-toplevel metrics_cleanup_service = get_metrics_cleanup_service() services_to_shutdown.insert(2, metrics_cleanup_service) if settings.mcpgateway_mcp_apps_enabled and settings.mcpgateway_mcp_apps_session_cleanup_enabled: mcp_app_session_cleanup_service = get_mcp_app_session_cleanup_service() services_to_shutdown.insert(3, mcp_app_session_cleanup_service) if dataplane_publisher_service is not None: services_to_shutdown.insert(3, dataplane_publisher_service) await shutdown_services(services_to_shutdown) # Stop the primary-worker elector (releases the redis lease if held). if settings.primary_worker_election_backend == "redis": # First-Party from mcpgateway.services.leader_election import stop_primary_worker_elector # pylint: disable=import-outside-toplevel await stop_primary_worker_elector() # Shutdown session-affinity service (before shared HTTP client). if settings.mcpgateway_session_affinity_enabled: # First-Party from mcpgateway.services.session_affinity import close_session_affinity # pylint: disable=import-outside-toplevel await close_session_affinity() # Drain upstream session registry (#4205): every (downstream_session_id, # gateway_id) → upstream ClientSession owned by this worker is closed. # First-Party from mcpgateway.services.upstream_session_registry import shutdown_upstream_session_registry # pylint: disable=import-outside-toplevel await shutdown_upstream_session_registry() # Shutdown shared HTTP client (after services, before Redis) await SharedHttpClient.shutdown() # Close Redis client last (after all services that use it) await close_redis_client() # Shut down the sandboxed jq worker pool shutdown_jq_pool() logger.info("Shutdown complete") async def shutdown_services(services_to_shutdown: list[Any]): """ Awaits shutdown of services provided in a list Args: services_to_shutdown (list[Any]): list of services to shutdown """ for service in services_to_shutdown: try: await service.shutdown() except Exception as e: logger.error(f"Error shutting down {service.__class__.__name__}: {str(e)}") async def setup_passthrough_headers(): """ Enables configuration and logs active settings as needed for when passthrough headers are enabled. """ logger.info(f"🔄 Header Passthrough: ENABLED (default headers: {settings.default_passthrough_headers})") if settings.enable_overwrite_base_headers: logger.warning("⚠️ Base Header Override: ENABLED - Client headers can override gateway headers") else: logger.info("🔒 Base Header Override: DISABLED - Gateway headers take precedence") # SECURITY AUDIT: Startup warning for sensitive header forwarding (Issue #3621 Phase 1) if settings.enable_sensitive_header_passthrough: logger.warning( "🔐 SECURITY AUDIT: Sensitive Header Passthrough ENABLED - " "whitelisted sensitive headers (Authorization, X-API-Key, etc.) will be forwarded to downstream A2A agents. " "Monitor metric 'a2a.downstream_headers.forwarded' for visibility (requires OBSERVABILITY_ENABLED=true). " "Only enable when trusted A2A agents require upstream credentials." ) db_gen = get_db() db = next(db_gen) # pylint: disable=stop-iteration-return try: await set_global_passthrough_headers(db) finally: db.commit() # End transaction cleanly db.close() # Initialize FastAPI app with orjson for 2-3x faster JSON serialization app = FastAPI( title=settings.app_name, version=__version__, description="ContextForge AI Gateway — an AI gateway, registry, and proxy for MCP, A2A, and REST/gRPC APIs. Exposes a unified control plane with centralized governance, discovery, and observability. Optimizes agent and tool calling, and supports plugins.", root_path=settings.app_root_path, lifespan=lifespan, default_response_class=ORJSONResponse, # Use orjson for high-performance JSON serialization ) # Setup metrics instrumentation setup_metrics(app) def validate_security_configuration(): """ Validate security configuration on startup. This function encapsulates: - verifying the configuration, - logging the output for warnings, - critical issues - security recommendations Args: None Raises: Passthrough Errors/Exceptions but doesn't raise any of its own. """ logger.info("🔒 Validating security configuration...") try: current_settings = get_settings() for _field_name, _secret_field in ( ("jwt_secret_key", current_settings.jwt_secret_key), ("auth_encryption_secret", current_settings.auth_encryption_secret), ): _val = _secret_field.get_secret_value() if _val.lower().startswith("__replace_me__"): _msg = f"{_field_name}: Value is an unset placeholder (__REPLACE_ME__). Run 'python -m mcpgateway.scripts.init_secrets' to generate strong values." if str(current_settings.environment).lower() == "production": raise SecurityConfigurationError(_msg) logger.warning("🔓 SECURITY WARNING - %s", _msg) security_status: settings.SecurityStatus = current_settings.get_security_status() security_warnings = security_status["warnings"] log_security_warnings(security_warnings) # Warn about ephemeral storage without strict user-in-DB mode if not getattr(current_settings, "require_user_in_db", False): is_ephemeral = ":memory:" in current_settings.database_url or current_settings.database_url == "sqlite:///./mcp.db" if is_ephemeral: logger.warning("Using potentially ephemeral storage with platform admin bootstrap enabled. Consider using persistent storage or setting REQUIRE_USER_IN_DB=true for production.") # Warn about default JWT issuer/audience in non-development environments if current_settings.environment != "development": if current_settings.jwt_issuer == "mcpgateway": logger.warning("Using default JWT_ISSUER in %s environment. Set a unique JWT_ISSUER per environment to prevent cross-environment token acceptance.", current_settings.environment) if current_settings.jwt_audience == "mcpgateway-api": logger.warning("Using default JWT_AUDIENCE in %s environment. Set a unique JWT_AUDIENCE per environment to prevent cross-environment token acceptance.", current_settings.environment) # UAID Cross-Gateway Routing Security Check if not current_settings.uaid_allowed_domains: if not current_settings.auth_required: logger.error( "⚠️ INSECURE CONFIGURATION: UAID_ALLOWED_DOMAINS is empty AND AUTH_REQUIRED=false. " "Cross-gateway routing is enabled without domain restrictions or authentication. " "This allows UAID-based agents to route to ANY remote gateway without validation. " "STRONGLY RECOMMENDED: Set UAID_ALLOWED_DOMAINS to restrict routing to trusted domains only." ) else: logger.warning( "⚠️ UAID_ALLOWED_DOMAINS is empty - cross-gateway routing allows ALL domains. " + "Any UAID-based agent can route to any remote gateway endpoint. " + "RECOMMENDED: Configure UAID_ALLOWED_DOMAINS to restrict routing to trusted gateways only. " + 'Example: UAID_ALLOWED_DOMAINS=["trusted-gateway.example.com", "partner.org"]' ) # Audit logging for explicit security overrides in production if current_settings.environment == "production" and not current_settings.require_strong_secrets: logger.warning("SECURITY AUDIT: REQUIRE_STRONG_SECRETS is explicitly disabled in a production environment. This override is being logged for audit purposes as per US-1 requirements.") log_security_recommendations(security_status) except SecurityConfigurationError as e: logger.critical(f"FAIL-CLOSED: {e}") sys.exit(1) def log_security_warnings(security_warnings: list[str]): """Log warnings from list of security warnings provided. Args: security_warnings: List of security warning messages. """ if security_warnings: logger.warning("=" * 60) logger.warning("🚨 SECURITY WARNINGS DETECTED:") logger.warning("=" * 60) for warning in security_warnings: logger.warning(f" {warning}") logger.warning("=" * 60) def log_critical_issues(critical_issues: list[Any]): """ Log critical based on configuration settings If REQUIRE_STRONG_SECRETS set, this will output critical errors and exit the mcpgateway server. Args: critical_issues: List Returns: None """ # Handle critical issues based on REQUIRE_STRONG_SECRETS setting if critical_issues: if settings.require_strong_secrets: logger.error("=" * 60) logger.error("💀 CRITICAL SECURITY ISSUES DETECTED:") logger.error("=" * 60) for issue in critical_issues: logger.error(f" ❌ {issue}") logger.error("=" * 60) logger.error("Startup aborted due to REQUIRE_STRONG_SECRETS=true") logger.error("To proceed anyway, set REQUIRE_STRONG_SECRETS=false") logger.error("=" * 60) sys.exit(1) else: # Log as warnings if not enforcing logger.warning("=" * 60) logger.warning("⚠️ Critical security issues detected (REQUIRE_STRONG_SECRETS=false):") for issue in critical_issues: logger.warning(f" • {issue}") logger.warning("=" * 60) def log_security_recommendations(security_status: settings.SecurityStatus): """ Log security recommendations based on configuration settings Args: security_status (settings.SecurityStatus): The SecurityStatus object for checking and logging current security settings from MCPGateway. Returns: None """ if not security_status["secure_secrets"] or not security_status["auth_enabled"]: logger.info("=" * 60) logger.info("📋 SECURITY RECOMMENDATIONS:") logger.info("=" * 60) if settings.jwt_secret_key in ("my-test-key", "my-test-key-but-now-longer-than-32-bytes"): # nosec B105 - checking for default value logger.info(" • Generate a strong JWT secret:") logger.info(" python3 -c 'import secrets; print(secrets.token_urlsafe(32))'") if settings.basic_auth_password.get_secret_value() == "changeme": # nosec B105 - checking for default value logger.info(" • Set a strong admin password in BASIC_AUTH_PASSWORD") if not settings.auth_required: logger.info(" • Enable authentication: AUTH_REQUIRED=true") if settings.skip_ssl_verify: logger.info(" • Enable SSL verification: SKIP_SSL_VERIFY=false") logger.info("=" * 60) def validate_uaid_security_config() -> None: """Validate UAID security configuration at startup. Behavior: - Logs ERROR if A2A enabled but UAID allowlist not configured - Fails startup if UAID_REQUIRE_ALLOWLIST_ON_STARTUP=true (strict mode) Design Decision (Issue #4236, Task #5): Default behavior is ERROR logging (non-blocking) to maintain backward compatibility and avoid breaking existing deployments. Operators can opt into fail-fast behavior via UAID_REQUIRE_ALLOWLIST_ON_STARTUP=true for stricter security posture. Rationale: - ERROR logging: Visible in logs, doesn't break deployments - Fail-fast (opt-in): Best for production, catches misconfig early - Not implemented: Admin UI banner (requires UI work, not always enabled) Raises: RuntimeError: If allowlist misconfigured and strict mode enabled """ if settings.mcpgateway_a2a_enabled: if not settings.uaid_allowed_domains and not settings.uaid_allow_all_domains: error_msg = ( "🚨 SECURITY: UAID cross-gateway routing is DISABLED. " "Configure UAID_ALLOWED_DOMAINS with trusted domains or set UAID_ALLOW_ALL_DOMAINS=true (unsafe for production). " "Cross-gateway UAID calls will fail until allowlist is configured." ) logger.error(error_msg) # Check for strict mode (fail-fast on misconfiguration) if settings.uaid_require_allowlist_on_startup: raise RuntimeError( f"{error_msg}\n\n" "Gateway startup aborted due to UAID_REQUIRE_ALLOWLIST_ON_STARTUP=true. " "Fix configuration or set UAID_REQUIRE_ALLOWLIST_ON_STARTUP=false to allow startup with ERROR log only." ) logger.info("✅ Security validation completed") # Global exceptions handlers @app.exception_handler(ValidationError) async def validation_exception_handler(_request: Request, exc: ValidationError): """Handle Pydantic validation errors globally. Intercepts ValidationError exceptions raised anywhere in the application and returns a properly formatted JSON error response with detailed validation error information. Args: _request: The FastAPI request object that triggered the validation error. (Unused but required by FastAPI's exception handler interface) exc: The Pydantic ValidationError exception containing validation failure details. Returns: JSONResponse: A 422 Unprocessable Entity response with formatted validation error details. Examples: >>> from pydantic import ValidationError, BaseModel >>> from fastapi import Request >>> import asyncio >>> >>> class TestModel(BaseModel): ... name: str ... age: int >>> >>> # Create a validation error >>> try: ... TestModel(name="", age="invalid") ... except ValidationError as e: ... # Test our handler ... result = asyncio.run(validation_exception_handler(None, e)) ... result.status_code 422 """ return ORJSONResponse(status_code=422, content=ErrorFormatter.format_validation_error(exc)) @app.exception_handler(RequestValidationError) async def request_validation_exception_handler(_request: Request, exc: RequestValidationError): """Handle FastAPI request validation errors (automatic request parsing). This handles ValidationErrors that occur during FastAPI's automatic request parsing before the request reaches your endpoint. Args: _request: The FastAPI request object that triggered validation error. exc: The RequestValidationError exception containing failure details. Returns: JSONResponse: A 422 Unprocessable Entity response with error details. """ logger.warning("Request validation error on %s: %s", _request.url.path if _request else "unknown", sanitize_validation_error_for_log(exc)) if not should_expose_error_details(): return ORJSONResponse(status_code=422, content={"detail": "An error occurred, please try again."}) if _request.url.path.startswith("/tools"): error_details = [] for error in exc.errors(): loc = error.get("loc", []) msg = error.get("msg", "Unknown error") ctx = error.get("ctx", {"error": {}}) type_ = error.get("type", "value_error") # Ensure ctx is JSON serializable if isinstance(ctx, dict): ctx_serializable = {k: (str(v) if isinstance(v, Exception) else v) for k, v in ctx.items()} else: ctx_serializable = str(ctx) error_detail = {"type": type_, "loc": loc, "msg": msg, "ctx": ctx_serializable} error_details.append(error_detail) return ORJSONResponse(status_code=422, content={"detail": error_details}) return await fastapi_default_validation_handler(_request, exc) @app.exception_handler(IntegrityError) async def database_exception_handler(_request: Request, exc: IntegrityError): """Handle SQLAlchemy database integrity constraint violations globally. Intercepts IntegrityError exceptions (e.g., unique constraint violations, foreign key constraints) and returns a properly formatted JSON error response. This provides consistent error handling for database constraint violations across the entire application. Args: _request: The FastAPI request object that triggered the database error. (Unused but required by FastAPI's exception handler interface) exc: The SQLAlchemy IntegrityError exception containing constraint violation details. Returns: JSONResponse: A 409 Conflict response with formatted database error details. Examples: >>> from sqlalchemy.exc import IntegrityError >>> from fastapi import Request >>> import asyncio >>> >>> # Create a mock integrity error >>> mock_error = IntegrityError("statement", {}, Exception("duplicate key")) >>> result = asyncio.run(database_exception_handler(None, mock_error)) >>> result.status_code 409 >>> # Verify ErrorFormatter.format_database_error is called >>> hasattr(result, 'body') True """ return ORJSONResponse(status_code=409, content=ErrorFormatter.format_database_error(exc)) @app.exception_handler(ContentSizeError) async def content_size_exception_handler(_request: Request, exc: ContentSizeError): """Handle content size limit violations globally. Args: _request: The incoming request (unused, required by FastAPI handler interface). exc: The ContentSizeError with actual_size, max_size, and content_type. Returns: ORJSONResponse: A 413 Payload Too Large response with structured error details. """ return ORJSONResponse(status_code=413, content={"detail": {"error": f"{exc.content_type} size limit exceeded", "message": str(exc), "actual_size": exc.actual_size, "max_size": exc.max_size}}) @app.exception_handler(TemplateValidationError) async def template_validation_exception_handler(_request: Request, exc: TemplateValidationError): """Handle template validation errors globally. Args: _request: The incoming request (unused, required by FastAPI handler interface). exc: The TemplateValidationError with template_name, reason, and pattern. Returns: ORJSONResponse: A 400 Bad Request response with structured error details. """ error_detail = { "error": "Template validation failed", "message": str(exc), "template_name": exc.template_name, "reason": exc.reason, } # DO NOT include pattern - it leaks internal security policy (CWE-209 fix) return ORJSONResponse(status_code=400, content={"detail": error_detail}) @app.exception_handler(ContentPatternError) async def content_pattern_error_handler(_request: Request, exc: ContentPatternError): """Handle malicious pattern detection errors globally (US-3). Returns HTTP 400 with structured error response. Does NOT leak internal patterns or content snippets (CWE-209 fix). Args: _request: The incoming request (unused, required by FastAPI handler interface). exc: The ContentPatternError with violation details. Returns: ORJSONResponse: A 400 Bad Request response with structured error details. """ return ORJSONResponse( status_code=400, content={ "detail": { "error": "Malicious pattern detected", "message": f"Content validation failed: {exc.content_type} contains potentially malicious patterns", "violation_type": exc.violation_type or "unknown", "content_type": exc.content_type, # DO NOT include pattern_matched or content_snippet (security) } }, ) # RFC 9110 §5.6.2 'token' pattern for header field names: # token = 1*tchar # tchar = "!" / "#" / "$" / "%" / "&" / "'" / "*" # / "+" / "-" / "." / "^" / "_" / "`" / "|" / "~" # / DIGIT / ALPHA _RFC9110_TOKEN_RE = re.compile(r"^[!#$%&'*+\-.^_`|~0-9A-Za-z]+$") def _validate_http_headers(headers: dict[str, str]) -> Optional[dict[str, str]]: """Validate headers according to RFC 9110. Args: headers: dict of headers Returns: Optional[dict[str, str]]: dictionary of valid headers Rules enforced: - Header name must match RFC 9110 'token'. - No whitespace before colon (enforced by dictionary usage). - Header value must not contain CTL characters (0x00–0x1F, 0x7F), except SP (0x20) and HTAB (0x09) which are allowed. """ validated: dict[str, str] = {} for key, value in headers.items(): # Validate header name (RFC 9110 token) if not _RFC9110_TOKEN_RE.match(key): logger.warning(f"Invalid header name: {key}") continue # RFC 9110: Reject CTLs (0x00–0x1F, 0x7F). Allow SP (0x20) and HTAB (0x09). valid = True for ch in value: code = ord(ch) if (0 <= code <= 31 or code == 127) and code not in (9, 32): valid = False break if not valid: logger.warning(f"Header value contains invalid characters: {key}") continue validated[key] = value return validated if validated else None @app.exception_handler(PluginViolationError) async def plugin_violation_exception_handler(_request: Request, exc: PluginViolationError): """Handle plugins violations globally. Intercepts PluginViolationError exceptions (e.g., OPA policy violation) and returns a properly formatted JSON error response. This provides consistent error handling for plugin violation across the entire application. Args: _request: The FastAPI request object that triggered the database error. (Unused but required by FastAPI's exception handler interface) exc: The PluginViolationError exception containing constraint violation details. Returns: JSONResponse: A response with error details in JSON-RPC format. Uses HTTP status code from violation if present (e.g., 429 for rate limiting), otherwise defaults to 200 for JSON-RPC compliance. Examples: >>> from cpex.framework import PluginViolationError >>> from cpex.framework.models import PluginViolation >>> from fastapi import Request >>> import asyncio >>> import json >>> >>> # Create a plugin violation error >>> mock_error = PluginViolationError(message="plugin violation",violation = PluginViolation( ... reason="Invalid input", ... description="The input contains prohibited content", ... code="PROHIBITED_CONTENT", ... details={"field": "message", "value": "test"} ... )) >>> result = asyncio.run(plugin_violation_exception_handler(None, mock_error)) >>> result.status_code 422 >>> content = orjson.loads(result.body.decode()) >>> content["error"]["code"] -32602 >>> "Plugin Violation:" in content["error"]["message"] True >>> content["error"]["data"]["plugin_error_code"] 'PROHIBITED_CONTENT' """ policy_violation = exc.violation.model_dump() if exc.violation else {} message = exc.violation.description if exc.violation else "A plugin violation occurred." policy_violation["message"] = exc.message status_code = exc.violation.mcp_error_code if exc.violation and exc.violation.mcp_error_code else -32602 violation_details: dict[str, Any] = {} http_status = 200 if exc.violation: if exc.violation.description: violation_details["description"] = exc.violation.description if exc.violation.details: violation_details["details"] = exc.violation.details if exc.violation.code: violation_details["plugin_error_code"] = exc.violation.code if exc.violation.plugin_name: violation_details["plugin_name"] = exc.violation.plugin_name # Use HTTP status code from violation if present (e.g., 429 for rate limiting) http_status = exc.violation.http_status_code if exc.violation.http_status_code else None if http_status and not VALID_HTTP_STATUS_CODES.get(http_status): logger.warning(f"Invalid HTTP status code {http_status} from violation, defaulting to 200") http_status = None if not http_status: logger.debug("Using Plugin violation code mapping for lack of http_status_code") mapping: Optional[PluginViolationCode] = PLUGIN_VIOLATION_CODE_MAPPING.get(exc.violation.code) if exc.violation.code else None if not mapping: http_status = 200 else: http_status = mapping.code json_rpc_error = PydanticJSONRPCError(code=status_code, message="Plugin Violation: " + message, data=violation_details) # Collect HTTP headers from violation if present headers = exc.violation.http_headers if exc.violation and exc.violation.http_headers else None response = ORJSONResponse(status_code=http_status, content={"error": json_rpc_error.model_dump()}) if headers: validated_headers = _validate_http_headers(headers) if validated_headers: response.headers.update(validated_headers) return response @app.exception_handler(PluginError) async def plugin_exception_handler(_request: Request, exc: PluginError): """Handle plugins errors globally. Intercepts PluginError exceptions and returns a properly formatted JSON error response. This provides consistent error handling for plugin error across the entire application. Args: _request: The FastAPI request object that triggered the database error. (Unused but required by FastAPI's exception handler interface) exc: The PluginError exception containing constraint violation details. Returns: JSONResponse: A 200 response with error details in JSON-RPC format. Examples: >>> from cpex.framework import PluginError >>> from cpex.framework.models import PluginErrorModel >>> from fastapi import Request >>> import asyncio >>> import json >>> >>> # Create a plugin error >>> mock_error = PluginError(error = PluginErrorModel( ... message="plugin error", ... code="timeout", ... plugin_name="abc", ... details={"field": "message", "value": "test"} ... )) >>> result = asyncio.run(plugin_exception_handler(None, mock_error)) >>> result.status_code 200 >>> content = orjson.loads(result.body.decode()) >>> content["error"]["code"] -32603 >>> "Plugin Error:" in content["error"]["message"] True >>> content["error"]["data"]["plugin_error_code"] 'timeout' >>> content["error"]["data"]["plugin_name"] 'abc' """ message = exc.error.message if exc.error else "A plugin error occurred." status_code = exc.error.mcp_error_code if exc.error else -32603 error_details: dict[str, Any] = {} if exc.error: if exc.error.details: error_details["details"] = exc.error.details if exc.error.code: error_details["plugin_error_code"] = exc.error.code if exc.error.plugin_name: error_details["plugin_name"] = exc.error.plugin_name json_rpc_error = PydanticJSONRPCError(code=status_code, message="Plugin Error: " + message, data=error_details) return ORJSONResponse(status_code=200, content={"error": json_rpc_error.model_dump()}) @app.exception_handler(Exception) async def unhandled_exception_handler(request: Request, _exc: Exception) -> ORJSONResponse: """Catch-all handler for unhandled exceptions. Logs the full exception server-side and returns a generic message to the client so that stack traces and internal details are never exposed in production responses. Args: request: The incoming request. _exc: The unhandled exception (unused; logged via logger.exception context). Returns: ORJSONResponse: 500 response with a generic error message. """ logger.exception( "Unhandled exception on %s %s", request.method, request.url.path, ) return ORJSONResponse( status_code=500, content={"detail": "An internal error occurred. Please try again."}, ) @app.exception_handler(ContentTypeError) async def content_type_exception_handler(_request: Request, exc: ContentTypeError): """Handle MIME type validation failures globally. Args: _request: The incoming request (unused, required by FastAPI handler interface). exc: The ContentTypeError with mime_type and allowed_types. Returns: ORJSONResponse: A 415 Unsupported Media Type response with error details. """ return ORJSONResponse( status_code=415, content={ "detail": { "error": "Unsupported MIME type", "message": str(exc), "mime_type": exc.mime_type, "allowed_types": exc.allowed_types[:5], # Limit to first 5 } }, ) def _normalize_scope_path(scope_path: str, root_path: str) -> str: """Strip ``root_path`` prefix from *scope_path* when a reverse proxy forwards the full path. Returns the route-only path (e.g. ``"/qa/gateway/docs"`` -> ``"/docs"``). A ``root_path`` of ``"/"`` is ignored to avoid stripping the leading slash from every path. Trailing slashes on *root_path* are stripped before comparison so that ``"/qa/gateway/"`` is handled identically to ``"/qa/gateway"``. Args: scope_path: The full path from the request scope. root_path: The root path prefix to be stripped. Returns: The normalized path with the root_path prefix removed. """ if root_path and len(root_path) > 1: root_path = root_path.rstrip("/") if root_path and len(root_path) > 1 and scope_path.startswith(root_path): rest = scope_path[len(root_path) :] # Ensure we matched a full path segment, not a partial prefix # e.g. root_path="/app" must not strip from "/application/admin" if not rest or rest[0] == "/": return rest or "/" return scope_path class DocsAuthMiddleware(BaseHTTPMiddleware): """ Middleware to protect FastAPI's auto-generated documentation routes (/docs, /redoc, and /openapi.json) using Bearer token authentication. If a request to one of these paths is made without a valid token, the request is rejected with a 401 or 403 error. Note: OPTIONS requests are exempt from authentication to support CORS preflight as per RFC 7231 Section 4.3.7 (OPTIONS must not require authentication). Note: When DOCS_ALLOW_BASIC_AUTH is enabled, Basic Authentication is also accepted using BASIC_AUTH_USER and BASIC_AUTH_PASSWORD credentials. """ async def dispatch(self, request: Request, call_next): """ Intercepts incoming requests to check if they are accessing protected documentation routes. If so, it requires a valid Bearer token; otherwise, it allows the request to proceed. Args: request (Request): The incoming HTTP request. call_next (Callable): The function to call the next middleware or endpoint. Returns: Response: Either the standard route response or a 401/403 error response. Examples: >>> import asyncio >>> from unittest.mock import Mock, AsyncMock, patch >>> from fastapi import HTTPException >>> from fastapi.responses import JSONResponse >>> >>> # Test unprotected path - should pass through >>> middleware = DocsAuthMiddleware(None) >>> request = Mock() >>> request.url.path = "/api/tools" >>> request.scope = {"path": "/api/tools", "root_path": ""} >>> request.method = "GET" >>> request.headers.get.return_value = None >>> call_next = AsyncMock(return_value="response") >>> >>> result = asyncio.run(middleware.dispatch(request, call_next)) >>> result 'response' >>> >>> # Test that middleware checks protected paths >>> request.url.path = "/docs" >>> isinstance(middleware, DocsAuthMiddleware) True """ protected_paths = ["/docs", "/redoc", "/openapi.json"] # Allow OPTIONS requests to pass through for CORS preflight (RFC 7231) if request.method == "OPTIONS": return await call_next(request) # Get path from scope to handle root_path correctly scope_path = request.scope.get("path", request.url.path) root_path = resolve_root_path(request) scope_path = _normalize_scope_path(scope_path, root_path) is_protected = any(scope_path.startswith(p) for p in protected_paths) if is_protected: try: token = get_auth_header_value(request.headers) cookie_token = request.cookies.get("jwt_token") # Use dedicated docs authentication that bypasses global auth settings await require_docs_auth_override(token, cookie_token) except HTTPException as e: return ORJSONResponse(status_code=e.status_code, content={"detail": e.detail}, headers=e.headers if e.headers else None) # Proceed to next middleware or route return await call_next(request) class AdminAuthMiddleware(BaseHTTPMiddleware): """ Middleware to protect Admin UI routes (/admin/*) requiring admin privileges. Exempts login-related paths and static assets: - /v1/admin/login - login page - /v1/admin/logout - logout action - /v1/admin/forgot-password - self-service password reset request page - /v1/admin/reset-password/* - self-service password reset completion page - /admin/static/* - static assets All other /admin/* routes require the user to be authenticated AND be an admin. Non-admin authenticated users receive a 403 Forbidden response. Note: This middleware respects the auth_required setting. When auth_required=False (typically in test environments), the middleware allows requests to pass through and relies on endpoint-level authentication which can be mocked in tests. """ # Public paths under /admin that do not require prior authentication. EXEMPT_PATHS = [ "/v1/admin/login", "/v1/admin/logout", "/v1/admin/forgot-password", "/v1/admin/reset-password", "/admin/static", # Legacy path "/v1/admin/static", # Versioned path ] @staticmethod def _strip_v1(path: str) -> str: """Strip /v1 prefix from path for normalization. Args: path: Path to normalize. Returns: Path with /v1 prefix removed if present. Examples: >>> AdminAuthMiddleware._strip_v1("/v1/admin/login") '/admin/login' >>> AdminAuthMiddleware._strip_v1("/admin/login") '/admin/login' """ return path[len("/v1") :] if path.startswith("/v1/") else path @staticmethod def _error_response(request: Request, root_path: str, status_code: int, detail: str, error_param: str = None): """Return appropriate error response based on request Accept header. Args: request: The incoming HTTP request. root_path: The root path prefix for the application. status_code: HTTP status code for JSON responses. detail: Error message detail. error_param: Optional error parameter for login redirect URL. Returns: Response with HX-Redirect for HTMX requests, RedirectResponse for HTML requests, ORJSONResponse for API requests. """ accept_header = request.headers.get("accept", "") is_htmx = request.headers.get("hx-request") == "true" if "text/html" in accept_header or is_htmx: login_url = f"{root_path}/admin/login" if root_path else "/admin/login" if error_param: login_url = f"{login_url}?error={error_param}" if is_htmx: return Response(status_code=200, headers={"HX-Redirect": login_url}) return RedirectResponse(url=login_url, status_code=302) return ORJSONResponse(status_code=status_code, content={"detail": detail}) @staticmethod def _auth_error_param(detail: str) -> Optional[str]: """Map TokenValidationError detail to browser redirect error param.""" normalized = (detail or "").lower() if "revoked" in normalized: return "token_revoked" if "disabled" in normalized: return "account_disabled" if "expired" in normalized or "idle timeout" in normalized: return "session_expired" return None async def dispatch(self, request: Request, call_next): # pylint: disable=too-many-return-statements """ Check admin privileges for admin routes. Args: request (Request): The incoming HTTP request. call_next (Callable): The function to call the next middleware or endpoint. Returns: Response: Either the standard route response or a 401/403 error response. """ # Skip admin auth check if auth is not required (e.g., test environments) # This allows tests to mock authentication at the dependency level if not settings.auth_required: return await call_next(request) # Get path from scope to handle root_path correctly scope_path = request.scope.get("path", request.url.path) root_path = resolve_root_path(request) scope_path = _normalize_scope_path(scope_path, root_path) # Allow OPTIONS requests for CORS preflight (RFC 7231) if request.method == "OPTIONS": return await call_next(request) # Check if this is an admin route (versioned /v1/admin/* or legacy /admin/*) is_admin_route = scope_path.startswith("/admin") or scope_path.startswith("/v1/admin") if not is_admin_route: return await call_next(request) # Normalize to unversioned path for exempt/permission checks so that # both direct (/v1/admin/login) and proxy-prefixed (/qa/gateway/admin/login) # paths are handled uniformly. check_path = self._strip_v1(scope_path) # Check if path is exempt (login, logout, static) is_exempt = any(check_path.startswith(self._strip_v1(p)) for p in self.EXEMPT_PATHS) if is_exempt: return await call_next(request) # For protected admin routes, verify admin status try: raw_token = None auth_user_email = None auth_user_is_admin = False auth_header = get_auth_header_value(request.headers) cookie_token = request.cookies.get("jwt_token") or request.cookies.get("access_token") # Preserve existing precedence: cookie first, then Authorization bearer. if cookie_token: raw_token = cookie_token elif auth_header: scheme, _, credentials_value = auth_header.partition(" ") if scheme.lower() == "bearer" and credentials_value: raw_token = credentials_value.strip() or None if raw_token: try: auth_user = await validate_token_user(request, raw_token) except TokenValidationError as exc: logger.warning( "Admin auth token validation failed: %s", SecurityValidator.sanitize_log_message(str(exc.detail)), ) return self._error_response( request, root_path, exc.status_code, exc.detail, self._auth_error_param(exc.detail), ) auth_user_email = auth_user.email auth_user_is_admin = bool(auth_user.is_admin) elif is_proxy_auth_trust_active(settings): proxy_user = request.headers.get(settings.proxy_user_header) if proxy_user: request.state.auth_method = "proxy" auth_user_email = proxy_user # Preserve existing proxy behavior: DB active/admin check, # with platform-admin bootstrap when REQUIRE_USER_IN_DB=false. with SessionLocal() as db: auth_service = EmailAuthService(db) proxy_db_user = await auth_service.get_user_by_email(proxy_user) if not proxy_db_user: platform_admin_email = getattr(settings, "platform_admin_email", "admin@example.com") if not settings.require_user_in_db and proxy_user == platform_admin_email: logger.info( "Platform admin bootstrap authentication for %s", SecurityValidator.sanitize_log_message(str(proxy_user)), ) auth_user_is_admin = True else: return self._error_response(request, root_path, 401, "User not found") else: if not proxy_db_user.is_active: logger.warning( "Admin access denied for disabled user: %s", SecurityValidator.sanitize_log_message(str(proxy_user)), ) return self._error_response(request, root_path, 403, "Account is disabled", "account_disabled") auth_user_is_admin = bool(proxy_db_user.is_admin) if not auth_user_email: return self._error_response(request, root_path, 401, "Authentication required") token_teams = getattr(request.state, "token_teams", None) # Preserve public-only denial invariant. if token_teams is not None and len(token_teams) == 0: logger.warning( "Admin access denied for public-only token: %s", SecurityValidator.sanitize_log_message(str(auth_user_email)), ) return self._error_response( request, root_path, 403, "Admin privileges required", "admin_required", ) # Validate optional team_id against token-visible teams. request_team_id = request.query_params.get("team_id") if request_team_id: try: request_team_id = uuid.UUID(request_team_id).hex except (ValueError, AttributeError): pass validated_team_id = request_team_id if token_teams and request_team_id and request_team_id in token_teams else None # validate_token_user already returned DB-authoritative is_admin, # including platform-admin bootstrap. if not auth_user_is_admin: with SessionLocal() as db: permission_service = PermissionService(db) has_admin_access = await permission_service.has_admin_permission( auth_user_email, team_id=validated_team_id, token_teams=token_teams, ) if not has_admin_access: logger.warning( "Admin access denied for user without admin permissions: %s", SecurityValidator.sanitize_log_message(str(auth_user_email)), ) return self._error_response( request, root_path, 403, "Admin privileges required", "admin_required", ) except HTTPException as exc: return self._error_response(request, root_path, exc.status_code, exc.detail) except Exception as exc: logger.error("Admin auth middleware error: %s", exc) return ORJSONResponse(status_code=500, content={"detail": "Authentication error"}) return await call_next(request) class MCPPathRewriteMiddleware: """ Middleware that rewrites paths ending with '/mcp' to '/mcp/', after performing authentication. - Rewrites exact '/mcp' to '/mcp/' so Starlette's mount does not emit a 307 redirect. - Rewrites paths like '/servers//mcp' to '/mcp/'. - Keeps ASGI ``raw_path`` aligned with rewritten paths when present. - Only exact '/mcp' and server-scoped MCP transport paths are rewritten. - Authentication is performed before any path rewriting. - If authentication fails, the request is not processed further. - All other requests are passed through without change. - Routes through the middleware stack (including CORSMiddleware) for proper CORS preflight handling. Attributes: application (Callable): The next ASGI application to process the request. """ def __init__(self, application, dispatch=None): """ Initialize the middleware with the ASGI application. Args: application (Callable): The next ASGI application to handle the request. dispatch (Callable, optional): An optional dispatch function for additional middleware processing. Example: >>> import asyncio >>> from unittest.mock import AsyncMock, patch >>> app_mock = AsyncMock() >>> middleware = MCPPathRewriteMiddleware(app_mock) >>> isinstance(middleware.application, AsyncMock) True """ self.application = application self.dispatch = dispatch # this can be TokenScopingMiddleware async def __call__(self, scope, receive, send): """ Intercept and potentially rewrite the incoming HTTP request path. Args: scope (dict): The ASGI connection scope. receive (Callable): Awaitable that yields events from the client. send (Callable): Awaitable used to send events to the client. Examples: >>> import asyncio >>> from unittest.mock import AsyncMock, patch >>> app_mock = AsyncMock() >>> middleware = MCPPathRewriteMiddleware(app_mock) >>> # Test path rewriting for /servers/123/mcp >>> scope = { "type": "http", "path": "/servers/123/mcp", "headers": [(b"host", b"example.com")] } >>> receive = AsyncMock() >>> send = AsyncMock() >>> with patch('mcpgateway.main.streamable_http_auth', return_value=True): ... asyncio.run(middleware(scope, receive, send)) >>> scope["path"] '/mcp/' >>> app_mock.assert_called() >>> # Test regular path (no rewrite) >>> scope = { "type": "http","path": "/tools","headers": [(b"host", b"example.com")] } >>> with patch('mcpgateway.main.streamable_http_auth', return_value=True): ... asyncio.run(middleware(scope, receive, send)) ... scope["path"] '/tools' """ if scope["type"] != "http": await self.application(scope, receive, send) return # If a dispatch (request middleware) is provided, adapt it if self.dispatch is not None: request = starletteRequest(scope, receive=receive) async def call_next(_req: starletteRequest) -> starletteResponse: """ Handles the next request in the middleware chain by calling a streamable HTTP response. Args: _req (starletteRequest): The incoming request to be processed. Returns: starletteResponse: A response generated from the streamable HTTP call. """ return await self._call_streamable_http(scope, receive, send) response = await self.dispatch(request, call_next) if response is None: # Either the dispatch handled the response itself, # or it blocked the request. Just return. return await response(scope, receive, send) return # Otherwise, just continue as normal await self._call_streamable_http(scope, receive, send) async def _call_streamable_http(self, scope, receive, send): """ Handles the streamable HTTP request after authentication and path rewriting. If auth succeeds and path ends with /mcp, rewrites to /mcp/ and calls self.application (continuing through middleware stack including CORSMiddleware). Args: scope (dict): The ASGI connection scope containing request metadata. receive (Callable): The function to receive events from the client. send (Callable): The function to send events to the client. Example: >>> import asyncio >>> from unittest.mock import AsyncMock, patch >>> app_mock = AsyncMock() >>> middleware = MCPPathRewriteMiddleware(app_mock) >>> scope = {"type": "http", "path": "/servers/123/mcp"} >>> receive = AsyncMock() >>> send = AsyncMock() >>> with patch('mcpgateway.main.streamable_http_auth', return_value=True): ... asyncio.run(middleware._call_streamable_http(scope, receive, send)) >>> app_mock.assert_called_once_with(scope, receive, send) >>> # Exact /mcp is normalized to avoid Starlette's mount redirect. >>> scope = {"type": "http", "path": "/mcp"} >>> with patch('mcpgateway.main.streamable_http_auth', return_value=True): ... asyncio.run(middleware._call_streamable_http(scope, receive, send)) >>> scope["path"] '/mcp/' """ # Auth check first auth_ok = await streamable_http_auth(scope, receive, send) if not auth_ok: return original_path = scope.get("path", "") scope["modified_path"] = original_path # Strip root_path prefix before pattern matching. # In reverse proxy deployments, scope["path"] may contain the full path # including the proxy prefix (e.g., "/dev/mcp-gateway/service/gateway/servers/123/mcp"). # We need to strip this prefix to correctly match the /servers/ pattern. root_path = (scope.get("root_path") or settings.app_root_path or "").rstrip("/") app_path = _normalize_scope_path(original_path, root_path) # Update modified_path to the app-relative path (without root_path prefix). # This ensures streamablehttp_transport can extract server_id via regex (#4266). scope["modified_path"] = app_path # Skip rewriting for well-known URIs (RFC 9728 OAuth metadata, etc.) # These paths may end with /mcp but should not be rewritten to the MCP transport if not app_path.startswith("/.well-known/"): if app_path == "/mcp": self._apply_mcp_rewrite(scope, root_path) await self.application(scope, receive, send) return if app_path.endswith("/mcp") or (app_path.endswith("/mcp/") and app_path != "/mcp/"): # SECURITY: Only rewrite recognised MCP paths — /servers/{id}/mcp. # Arbitrary prefixes (e.g. /foo/mcp) must NOT be rewritten to # /mcp/ as that would expose the global MCP transport under # undocumented aliases, broadening the externally reachable # route surface. if app_path.startswith("/servers/"): # Validate that a non-empty server_id segment is present. # Without this check, paths like /servers//mcp (empty ID) # would be rewritten and silently fall through (#3891). _srv_match = re.match(r"/servers/([^/]+)/mcp", app_path) if not _srv_match: response = ORJSONResponse({"detail": "Invalid server identifier"}, status_code=404) await response(scope, receive, send) return else: # Not a /servers/ path — do not rewrite, pass through await self.application(scope, receive, send) return # Rewrite to /mcp/ and continue through middleware (lets CORSMiddleware handle preflight) # Preserve root_path prefix when rewriting self._apply_mcp_rewrite(scope, root_path) await self.application(scope, receive, send) return await self.application(scope, receive, send) @staticmethod def _apply_mcp_rewrite(scope, root_path: str) -> str: """Rewrite a validated MCP transport path to the mounted /mcp/ app path.""" original_path = scope.get("path", "") new_path = f"{root_path}/mcp/" if root_path else "/mcp/" scope["path"] = new_path if "raw_path" in scope: try: # ASGI raw_path stores raw octets; latin-1 preserves a 1:1 byte mapping for valid values. scope["raw_path"] = new_path.encode("latin-1") except (UnicodeEncodeError, ValueError): logger.warning("MCPPathRewriteMiddleware: non-latin-1 raw_path skipped for %s", new_path) logger.debug("MCPPathRewriteMiddleware: %s -> %s", original_path, new_path) return new_path # Configure CORS with environment-aware origins cors_origins = list(settings.allowed_origins) if settings.allowed_origins else [] # Ensure we never use wildcard in production if settings.environment == "production" and not cors_origins: logger.warning("No CORS origins configured for production environment. CORS will be disabled.") cors_origins = [] app.add_middleware( CORSMiddleware, allow_origins=cors_origins, allow_credentials=settings.cors_allow_credentials, allow_methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"], allow_headers=["*"], expose_headers=["Content-Length", "X-Request-ID", "X-Password-Change-Required"], max_age=600, # Cache preflight requests for 10 minutes ) # Add response compression middleware (Brotli, Zstd, GZip) # Automatically negotiates compression algorithm based on client Accept-Encoding header # Priority: Brotli (best compression) > Zstd (fast) > GZip (universal fallback) # Only compress responses larger than minimum_size to avoid overhead # NOTE: When json_response_enabled=False (SSE mode), /mcp paths are excluded from # compression to prevent buffering/breaking of streaming responses. See middleware/compression.py. if settings.compression_enabled: app.add_middleware( SSEAwareCompressMiddleware, minimum_size=settings.compression_minimum_size, gzip_level=settings.compression_gzip_level, brotli_quality=settings.compression_brotli_quality, zstd_level=settings.compression_zstd_level, ) logger.info( f"🗜️ Response compression enabled (SSE-aware): minimum_size={settings.compression_minimum_size}B, " f"gzip_level={settings.compression_gzip_level}, " f"brotli_quality={settings.compression_brotli_quality}, " f"zstd_level={settings.compression_zstd_level}" ) else: logger.info("🚫 Response compression disabled") # Add security headers middleware app.add_middleware(SecurityHeadersMiddleware) # Add RFC 6585 § 5 header size validation middleware (before rate limiting for early rejection) if settings.header_size_validation_enabled: app.add_middleware(HeaderSizeMiddleware) logger.info( f"📏 RFC 6585 header size validation enabled: max_total={settings.max_header_total_size_bytes}B, max_field={settings.max_header_field_size_bytes}B, max_count={settings.max_header_count}" ) # Add rate limiting middleware (after HttpAuthMiddleware for user-aware limiting) if settings.rate_limiting_enabled: app.add_middleware(RateLimitMiddleware) logger.info( f"🚦 RFC 6585 rate limiting enabled: Redis={settings.rate_limiting_redis_enabled}, " f"Tiers[CRITICAL={settings.rate_limit_critical_rpm}, " f"HIGH={settings.rate_limit_high_rpm}, " f"MEDIUM={settings.rate_limit_medium_rpm}, " f"LOW={settings.rate_limit_low_rpm}]" ) # Add validation middleware if explicitly enabled if settings.validation_middleware_enabled: app.add_middleware(ValidationMiddleware) logger.warning("🔒 Input validation and output sanitization middleware enabled. %s", VALIDATION_MIDDLEWARE_DEPRECATION_MESSAGE) else: logger.info("🔒 Input validation and output sanitization middleware disabled") # Add MCP Protocol Version validation middleware (validates MCP-Protocol-Version header) app.add_middleware(MCPProtocolVersionMiddleware) # Add token scoping middleware (only when email auth is enabled) if settings.email_auth_enabled: app.add_middleware(BaseHTTPMiddleware, dispatch=token_scoping_middleware) # Add streamable HTTP middleware for /mcp routes with token scoping app.add_middleware(MCPPathRewriteMiddleware, dispatch=token_scoping_middleware) else: # Add streamable HTTP middleware for /mcp routes app.add_middleware(MCPPathRewriteMiddleware) # Add HTTP authentication hook middleware for plugins (before auth dependencies) # Middleware will get the global plugin manager at request time if factory exists app.add_middleware(HttpAuthMiddleware) # Add request logging middleware FIRST (always enabled for gateway boundary logging) # IMPORTANT: Must be registered BEFORE CorrelationIDMiddleware so it executes AFTER correlation ID is set # Gateway boundary logging (request_started/completed) runs regardless of log_requests setting # Detailed payload logging only runs if log_detailed_requests=True app.add_middleware( RequestLoggingMiddleware, enable_gateway_logging=True, log_detailed_requests=settings.log_requests, log_level=settings.log_level, max_body_size=settings.log_detailed_max_body_size, log_resolve_user_identity=settings.log_resolve_user_identity, log_detailed_skip_endpoints=settings.log_detailed_skip_endpoints, log_detailed_sample_rate=settings.log_detailed_sample_rate, ) # Add custom DocsAuthMiddleware app.add_middleware(DocsAuthMiddleware) # Add AdminAuthMiddleware to protect admin routes (requires admin privileges) # This ensures all /admin/* routes (except login/logout) require admin status app.add_middleware(AdminAuthMiddleware) # Rewrite Host header from X-Forwarded-Host when behind a reverse proxy. # Uvicorn's ProxyHeadersMiddleware handles X-Forwarded-Proto and X-Forwarded-For # but not X-Forwarded-Host (upstream issue encode/uvicorn#965). # This ensures request.base_url reflects the proxy's public host, fixing the # OAuth redirect_uri hint and other URL construction throughout the admin UI. # Registered alongside ProxyHeadersMiddleware with the same trust model. # # Registered BEFORE ProxyHeadersMiddleware so that it is inner (executes after # ProxyHeadersMiddleware in the ASGI call chain) and can rely on the scheme # already being corrected when deriving the default port for scope["server"]. app.add_middleware(ForwardedHostMiddleware) # Trust all proxies (or lock down with a list of host patterns) app.add_middleware(ProxyHeadersMiddleware, trusted_hosts="*") # Add correlation ID middleware if enabled # Note: Registered AFTER RequestLoggingMiddleware so correlation ID is available when RequestLoggingMiddleware executes if settings.correlation_id_enabled: app.add_middleware(CorrelationIDMiddleware) logger.info(f"✅ Correlation ID tracking enabled (header: {settings.correlation_id_header})") register_auth_context_middleware(app) # Add token usage logging middleware # This tracks API token usage for analytics and security monitoring # Note: Runs after AuthContextMiddleware so request.state.auth_method is available if settings.token_usage_logging_enabled: # First-Party from mcpgateway.middleware.token_usage_middleware import TokenUsageMiddleware # noqa: E402 app.add_middleware(TokenUsageMiddleware) logger.info("📊 Token usage logging middleware enabled - tracking API token usage") else: logger.info("📊 Token usage logging middleware disabled") # Add observability middleware if enabled # Note: Middleware runs in REVERSE order (last added runs first) # If AuthContextMiddleware is already registered, ObservabilityMiddleware wraps it # Execution order will be: AuthContext -> Observability -> Request Handler # Wire observability adapter into the plugin manager when observability is enabled # _service is a module-level global read later in lifespan(); it must always be bound # (even when this branch doesn't run at import time) so tests that flip # observability_enabled to True after import and then invoke lifespan() don't hit a # NameError on the module global. _service = None # pylint: disable=invalid-name if settings.observability_enabled: # First-Party from mcpgateway.middleware.observability_middleware import ObservabilityMiddleware from mcpgateway.services.observability_service import ObservabilityService _service = ObservabilityService() app.add_middleware(ObservabilityMiddleware, enabled=True, service=_service) # Plugin observability adapter will be set in lifespan after plugin_manager is initialized logger.info("🔍 Observability middleware enabled - tracing include-listed requests") else: logger.info("🔍 Observability middleware disabled") if otel_tracing_enabled(): app.add_middleware(OpenTelemetryRequestMiddleware) logger.info("🧵 OTEL request tracing middleware enabled for transport request roots") else: logger.info("🧵 OTEL request tracing middleware disabled") # Add OTEL baggage middleware after request tracing middleware so it executes first # and attaches baggage before the request-root span is created. if settings.otel_baggage_enabled and otel_tracing_enabled(): # First-Party from mcpgateway.middleware.baggage_middleware import BaggageMiddleware app.add_middleware(BaggageMiddleware) logger.info("🧳 OTEL baggage middleware enabled for HTTP header extraction") elif settings.otel_baggage_enabled and not otel_tracing_enabled(): logger.warning("🧳 OTEL baggage enabled but tracing disabled - baggage will not be captured in spans") else: logger.debug("🧳 OTEL baggage middleware disabled") # Database query logging middleware (for N+1 detection) if settings.db_query_log_enabled: # First-Party from mcpgateway.db import engine from mcpgateway.middleware.db_query_logging import setup_query_logging setup_query_logging(app, engine) logger.info(f"📊 Database query logging enabled - logs: {settings.db_query_log_file}") else: logger.debug("📊 Database query logging disabled (enable with DB_QUERY_LOG_ENABLED=true)") # Client disconnect middleware — MUST be outermost (added last, runs first). # Cancels in-flight request handlers when the client (nginx) closes the connection, # preventing CLOSE_WAIT accumulation and associated memory leaks. if settings.client_disconnect_middleware_enabled: app.add_middleware(ClientDisconnectMiddleware) logger.info("Client disconnect middleware enabled - cancels handlers on nginx timeout") else: logger.debug("Client disconnect middleware disabled (enable with CLIENT_DISCONNECT_MIDDLEWARE_ENABLED=true)") # Set up Jinja2 templates and store in app state for later use # auto_reload=False in production prevents re-parsing templates on each request (performance) jinja_env = Environment( loader=FileSystemLoader(str(settings.templates_dir)), autoescape=True, auto_reload=settings.templates_auto_reload, ) # Add custom filter to decode HTML entities for backward compatibility with old database records # that were stored with HTML entities (e.g., ' instead of ') # NOTE: This filter can be removed after all deployments have run the c1c2c3c4c5c6 migration, # which decodes all existing HTML entities in the database. After that migration, this filter # becomes a no-op since new data is stored without HTML encoding. def decode_html_entities(value: str) -> str: """Decode HTML entities in strings for display. This filter handles legacy data that was stored with HTML entities. New data is stored without encoding, but this ensures old records display correctly. TEMPORARY: Can be removed after c1c2c3c4c5c6 migration has been applied to all deployments. Args: value: String that may contain HTML entities Returns: String with HTML entities decoded to their original characters """ if not value: return value return html.unescape(value) jinja_env.filters["decode_html"] = decode_html_entities def tojson_attr(value: object) -> str: """JSON-encode a value for safe use inside double-quoted HTML attributes. Unlike the built-in ``|tojson`` filter (which returns ``Markup``, bypassing autoescape), this filter returns a plain ``str``. Jinja2 autoescape then HTML-encodes the ``"`` characters to ``"``, keeping the enclosing ``"``-delimited HTML attribute intact. The browser decodes the entities back to ``"`` before passing the value to the JS engine. Use ``|tojson_attr`` for inline event handlers (``onclick``, ``onsubmit``). Use the built-in ``|tojson`` for ``