diff --git a/.env.example b/.env.example index e562f66ced1..37fcd15eb05 100644 --- a/.env.example +++ b/.env.example @@ -3157,6 +3157,10 @@ PLUGINS_CLI_MARKUP_MODE=rich # Current Ed25519 public key (derived automatically if private key is set) # ED25519_PUBLIC_KEY= +# SSL context cache settings +# SSL_CONTEXT_CACHE_MAX_SIZE=100 +# SSL_CONTEXT_CACHE_TTL= + # ============================================================================= # Bootstrap additional system roles diff --git a/docs/docs/manage/self-signed-certificates.md b/docs/docs/manage/self-signed-certificates.md index 991b9178af4..cc63bbe8318 100644 --- a/docs/docs/manage/self-signed-certificates.md +++ b/docs/docs/manage/self-signed-certificates.md @@ -94,8 +94,10 @@ When the gateway invokes a tool from an MCP server with a custom CA certificate: 1. **SSL Context Creation**: A custom SSL context is created using Python's `ssl.create_default_context()` 2. **Certificate Loading**: The CA certificate is loaded using `ctx.load_verify_locations(cadata=ca_certificate)` 3. **Signature Validation** (if enabled): The certificate signature is validated using Ed25519 to ensure it hasn't been tampered with -4. **HTTPS Client Configuration**: The SSL context is passed to the HTTPX client as the `verify` parameter +4. **mTLS client cert/key support**: If `client_cert` and `client_key` are configured for the gateway, the SSL context is loaded with `ctx.load_cert_chain(client_cert, client_key)` for mutual TLS +5. **HTTPS Client Configuration**: The SSL context is passed to the HTTPX client as the `verify` parameter 5. **Secure Connection**: All HTTPS requests to the MCP server use the custom CA certificate for validation +6. **HTTP Bypass**: For plain `http://` gateway URLs, SSL context creation is skipped and default HTTPX verification is used for reduced overhead ### Usage During Gateway Registration diff --git a/mcpgateway/alembic/versions/615af4ab94b4_add_gateway_client_cert_and_key.py b/mcpgateway/alembic/versions/615af4ab94b4_add_gateway_client_cert_and_key.py new file mode 100644 index 00000000000..9a211ca920e --- /dev/null +++ b/mcpgateway/alembic/versions/615af4ab94b4_add_gateway_client_cert_and_key.py @@ -0,0 +1,54 @@ +"""add_gateway_client_cert_and_key + +Revision ID: 615af4ab94b4 +Revises: 20a0e0538ac5 +Create Date: 2026-03-20 10:47:06.592968 + +""" + +# Standard +from typing import Sequence, Union + +# Third-Party +from alembic import op +import sqlalchemy as sa + +# revision identifiers, used by Alembic. +revision: str = "615af4ab94b4" +down_revision: Union[str, Sequence[str], None] = "20a0e0538ac5" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Upgrade schema.""" + bind = op.get_bind() + inspector = sa.inspect(bind) + + if "gateways" not in inspector.get_table_names(): + return + + columns = [col["name"] for col in inspector.get_columns("gateways")] + + if "client_cert" not in columns: + op.add_column("gateways", sa.Column("client_cert", sa.Text(), nullable=True)) + + if "client_key" not in columns: + op.add_column("gateways", sa.Column("client_key", sa.Text(), nullable=True)) + + +def downgrade() -> None: + """Downgrade schema.""" + bind = op.get_bind() + inspector = sa.inspect(bind) + + if "gateways" not in inspector.get_table_names(): + return + + columns = [col["name"] for col in inspector.get_columns("gateways")] + + if "client_cert" in columns: + op.drop_column("gateways", "client_cert") + + if "client_key" in columns: + op.drop_column("gateways", "client_key") diff --git a/mcpgateway/db.py b/mcpgateway/db.py index 741431fdbe3..700c62ced84 100644 --- a/mcpgateway/db.py +++ b/mcpgateway/db.py @@ -4594,6 +4594,10 @@ class Gateway(Base): ca_certificate_sig: Mapped[Optional[str]] = mapped_column(String(64), nullable=True) signing_algorithm: Mapped[Optional[str]] = mapped_column(String(20), nullable=True, default="ed25519") # e.g., "sha256" + # mTLS client certificate/key for upstream gateway authentication + client_cert: Mapped[Optional[str]] = mapped_column(Text, nullable=True) + client_key: Mapped[Optional[str]] = mapped_column(Text, nullable=True) + # Relationship with local tools this gateway provides tools: Mapped[List["Tool"]] = relationship(back_populates="gateway", foreign_keys="Tool.gateway_id", cascade="all, delete-orphan", passive_deletes=True) diff --git a/mcpgateway/handlers/signal_handlers.py b/mcpgateway/handlers/signal_handlers.py new file mode 100644 index 00000000000..a450d46a880 --- /dev/null +++ b/mcpgateway/handlers/signal_handlers.py @@ -0,0 +1,57 @@ +# -*- coding: utf-8 -*- +"""Signal handlers for ContextForge Gateway. + +Provides SIGHUP handling for certificate rotation without restart. +""" + +# Standard +import asyncio +import logging +from typing import Any + +logger = logging.getLogger(__name__) + + +async def sighup_reload() -> None: + """Clear SSL context cache and drain MCP session pool on SIGHUP for certificate rotation. + + Clears the SSL context cache to force recreation of SSL contexts + with potentially updated certificates, and drains the MCP session + pool so pooled connections reconnect with new TLS state. + """ + try: + # First-Party + from mcpgateway.utils.ssl_context_cache import clear_ssl_context_cache # pylint: disable=import-outside-toplevel + + clear_ssl_context_cache() + logger.info("SIGHUP: SSL context cache cleared") + except Exception as exc: + logger.error(f"SIGHUP handler failed to clear SSL context cache: {exc}") + + try: + # First-Party + from mcpgateway.services.mcp_session_pool import drain_mcp_session_pool # pylint: disable=import-outside-toplevel + + await drain_mcp_session_pool() + logger.info("SIGHUP: MCP session pool drained for TLS rotation") + except Exception as exc: + logger.debug(f"SIGHUP: MCP session pool drain skipped: {exc}") + + +def sighup_handler(_signum: int, _frame: Any) -> None: + """Handle SIGHUP signal by scheduling async SSL cache reload. + + Signal handler that safely schedules an asynchronous task to clear + the SSL context cache. Uses the running event loop to create a task + for the async reload operation. + + Args: + _signum: Signal number (unused but required by signal handler signature) + _frame: Current stack frame (unused but required by signal handler signature) + """ + logger.info("Received SIGHUP signal, scheduling SSL context cache refresh") + try: + event_loop = asyncio.get_running_loop() + event_loop.create_task(sighup_reload()) + except RuntimeError: + logger.warning("SIGHUP received but event loop not running; skipping async reload") diff --git a/mcpgateway/main.py b/mcpgateway/main.py index c1ddaae27ac..1bada3c719e 100644 --- a/mcpgateway/main.py +++ b/mcpgateway/main.py @@ -38,6 +38,7 @@ import json import logging import re +import signal import sys from typing import Any, AsyncIterator, Dict, List, Optional, TypeAlias, Union from urllib.parse import urlparse, urlunparse @@ -1701,6 +1702,11 @@ async def lifespan(_app: FastAPI) -> AsyncIterator[None]: logger.info("All services initialized successfully") + # First-Party + from mcpgateway.handlers.signal_handlers import sighup_handler # pylint: disable=import-outside-toplevel + + signal.signal(signal.SIGHUP, 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 @@ -1769,6 +1775,12 @@ async def run_log_aggregation_loop() -> None: raise SystemExit(1) raise finally: + # Restore default SIGHUP handling in case we reset signal handlers. + try: + signal.signal(signal.SIGHUP, signal.SIG_DFL) + 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): diff --git a/mcpgateway/schemas.py b/mcpgateway/schemas.py index 98d81c48efb..50973e3d9eb 100644 --- a/mcpgateway/schemas.py +++ b/mcpgateway/schemas.py @@ -2683,6 +2683,10 @@ class GatewayCreate(BaseModelWithConfigDict): ca_certificate_sig: Optional[str] = Field(None, description="Signature of the custom CA certificate for integrity verification") signing_algorithm: Optional[str] = Field("ed25519", description="Algorithm used for signing the CA certificate") + # mTLS client certificate/key + client_cert: Optional[str] = Field(None, description="Client TLS certificate for mTLS authentication") + client_key: Optional[str] = Field(None, description="Client TLS key for mTLS authentication") + # Per-gateway refresh configuration refresh_interval_seconds: Optional[int] = Field(None, ge=60, description="Per-gateway refresh interval in seconds (minimum 60); uses global default if not set") @@ -3025,6 +3029,15 @@ class GatewayUpdate(BaseModelWithConfigDict): # Gateway mode configuration gateway_mode: Optional[str] = Field(None, description="Gateway mode: 'cache' (database caching, default) or 'direct_proxy' (pass-through mode with no caching)", pattern="^(cache|direct_proxy)$") + # CA certificate configuration for custom TLS trust + ca_certificate: Optional[str] = Field(None, description="Custom CA certificate for TLS verification") + ca_certificate_sig: Optional[str] = Field(None, description="Signature of the custom CA certificate") + signing_algorithm: Optional[str] = Field(None, description="Algorithm used for signing the CA certificate") + + # mTLS client TLS certificate and key + client_cert: Optional[str] = Field(None, description="Client TLS certificate for mTLS gateway authentication") + client_key: Optional[str] = Field(None, description="Client TLS key for mTLS gateway authentication") + @field_validator("tags") @classmethod def validate_tags(cls, v: Optional[List[str]]) -> List[str]: @@ -3319,6 +3332,11 @@ class GatewayRead(BaseModelWithConfigDict): last_seen: Optional[datetime] = Field(default_factory=lambda: datetime.now(timezone.utc), description="Last seen timestamp") passthrough_headers: Optional[List[str]] = Field(default=None, description="List of headers allowed to be passed through from client to target") + ca_certificate: Optional[str] = Field(default=None, description="Custom CA certificate for TLS verification") + ca_certificate_sig: Optional[str] = Field(default=None, description="Signature of the custom CA certificate") + signing_algorithm: Optional[str] = Field(default="ed25519", description="Algorithm used for signing the CA certificate") + client_cert: Optional[str] = Field(default=None, description="Client TLS certificate for mTLS authentication") + client_key: Optional[str] = Field(default=None, description="Client TLS key for mTLS authentication") # Authorizations auth_type: Optional[str] = Field(None, description="auth_type: basic, bearer, authheaders, oauth, query_param, or None") auth_value: Optional[str] = Field(None, description="auth value: username/password or token or custom headers") @@ -3591,6 +3609,8 @@ def masked(self) -> "GatewayRead": masked_data["auth_token_unmasked"] = None masked_data["auth_header_value_unmasked"] = None masked_data["auth_headers_unmasked"] = None + # SECURITY: Mask mTLS client private key + masked_data["client_key"] = settings.masked_auth_value if masked_data.get("client_key") else None return GatewayRead.model_validate(masked_data) diff --git a/mcpgateway/services/gateway_service.py b/mcpgateway/services/gateway_service.py index 6d105e6f3ff..e95fd5b5c7b 100644 --- a/mcpgateway/services/gateway_service.py +++ b/mcpgateway/services/gateway_service.py @@ -98,7 +98,7 @@ # logging.getLogger("httpx").setLevel(logging.WARNING) # Disables httpx logs for regular health checks from mcpgateway.services.audit_trail_service import get_audit_trail_service from mcpgateway.services.base_service import BaseService -from mcpgateway.services.encryption_service import protect_oauth_config_for_storage +from mcpgateway.services.encryption_service import get_encryption_service, protect_oauth_config_for_storage from mcpgateway.services.event_service import EventService from mcpgateway.services.http_client_service import get_default_verify, get_http_timeout, get_isolated_http_client from mcpgateway.services.logging_service import LoggingService @@ -525,6 +525,23 @@ def normalize_url(url: str) -> str: # For all other URLs, preserve the domain name return url + @staticmethod + async def _encrypt_client_key(client_key: Optional[str]) -> Optional[str]: + """Encrypt a client private key for storage. + + Args: + client_key: Plaintext client private key or None. + + Returns: + Encrypted client key or None if input is None/empty. + """ + if not client_key: + return None + encryption = get_encryption_service(settings.auth_encryption_secret) + if encryption.is_encrypted(client_key): + return client_key + return await encryption.encrypt_secret_async(client_key) + def create_ssl_context(self, ca_certificate: str) -> ssl.SSLContext: """Create an SSL context with the provided CA certificate. @@ -877,6 +894,8 @@ async def register_gateway( oauth_config = await protect_oauth_config_for_storage(getattr(gateway, "oauth_config", None)) ca_certificate = getattr(gateway, "ca_certificate", None) + init_client_cert = getattr(gateway, "client_cert", None) + init_client_key = getattr(gateway, "client_key", None) # Check if gateway is in direct_proxy mode gateway_mode = getattr(gateway, "gateway_mode", "cache") @@ -895,6 +914,8 @@ async def register_gateway( oauth_config, ca_certificate, auth_query_params=auth_query_params_decrypted, + client_cert=init_client_cert, + client_key=init_client_key, ), timeout=initialize_timeout, ) @@ -910,6 +931,8 @@ async def register_gateway( oauth_config, ca_certificate, auth_query_params=auth_query_params_decrypted, + client_cert=init_client_cert, + client_key=init_client_key, ) if gateway.one_time_auth: @@ -1154,6 +1177,9 @@ async def register_gateway( ca_certificate=gateway.ca_certificate, ca_certificate_sig=gateway.ca_certificate_sig, signing_algorithm=gateway.signing_algorithm, + # mTLS client certificate/key + client_cert=getattr(gateway, "client_cert", None), + client_key=await self._encrypt_client_key(getattr(gateway, "client_key", None)), # Gateway mode configuration gateway_mode=gateway_mode, ) @@ -1974,6 +2000,23 @@ async def update_gateway( _validate_gateway_team_assignment(db, user_email, gateway_update.team_id) gateway.team_id = gateway_update.team_id + # Update CA certificate fields if provided + if getattr(gateway_update, "ca_certificate", None) is not None: + gateway.ca_certificate = gateway_update.ca_certificate + if getattr(gateway_update, "ca_certificate_sig", None) is not None: + gateway.ca_certificate_sig = gateway_update.ca_certificate_sig + if getattr(gateway_update, "signing_algorithm", None) is not None: + gateway.signing_algorithm = gateway_update.signing_algorithm + + # Update mTLS client certificate/key if provided + if getattr(gateway_update, "client_cert", None) is not None: + gateway.client_cert = gateway_update.client_cert + if getattr(gateway_update, "client_key", None) is not None: + if gateway_update.client_key == settings.masked_auth_value: + pass # Preserve existing encrypted value + else: + gateway.client_key = await self._encrypt_client_key(gateway_update.client_key) + # Only update auth_type if explicitly provided in the update if gateway_update.auth_type is not None: gateway.auth_type = gateway_update.auth_type @@ -2111,6 +2154,15 @@ async def update_gateway( try: ca_certificate = getattr(gateway, "ca_certificate", None) + update_client_cert = getattr(gateway, "client_cert", None) + update_client_key = getattr(gateway, "client_key", None) + # Decrypt client_key for initialization (stored encrypted) + if update_client_key: + try: + _enc = get_encryption_service(settings.auth_encryption_secret) + update_client_key = _enc.decrypt_secret_or_plaintext(update_client_key) + except Exception: + logger.debug("client_key decryption skipped during gateway re-init") capabilities, tools, resources, prompts = await self._initialize_gateway( init_url, gateway.auth_value, @@ -2119,6 +2171,8 @@ async def update_gateway( gateway.oauth_config, ca_certificate, auth_query_params=auth_query_params_decrypted, + client_cert=update_client_cert, + client_key=update_client_key, ) new_tool_names = [tool.name for tool in tools] new_resource_uris = [resource.uri for resource in resources] @@ -2567,8 +2621,24 @@ async def set_gateway_state(self, db: Session, gateway_id: str, activate: bool, if auth_query_params_decrypted: init_url = apply_query_param_auth(gateway.url, auth_query_params_decrypted) + act_client_cert = getattr(gateway, "client_cert", None) + act_client_key = getattr(gateway, "client_key", None) + if act_client_key: + try: + _enc = get_encryption_service(settings.auth_encryption_secret) + act_client_key = _enc.decrypt_secret_or_plaintext(act_client_key) + except Exception: + logger.debug("client_key decryption skipped during gateway activation") capabilities, tools, resources, prompts = await self._initialize_gateway( - init_url, gateway.auth_value, gateway.transport, gateway.auth_type, gateway.oauth_config, auth_query_params=auth_query_params_decrypted, oauth_auto_fetch_tool_flag=True + init_url, + gateway.auth_value, + gateway.transport, + gateway.auth_type, + gateway.oauth_config, + auth_query_params=auth_query_params_decrypted, + oauth_auto_fetch_tool_flag=True, + client_cert=act_client_cert, + client_key=act_client_key, ) new_tool_names = [tool.name for tool in tools] new_resource_uris = [resource.uri for resource in resources] @@ -3223,6 +3293,8 @@ async def _check_single_gateway_health(self, gateway: DbGateway, user_email: Opt gateway_oauth_config = gateway.oauth_config gateway_auth_value = gateway.auth_value gateway_auth_query_params = gateway.auth_query_params + health_client_cert = getattr(gateway, "client_cert", None) + health_client_key = getattr(gateway, "client_key", None) # Handle query_param auth - decrypt and apply to URL for health check auth_query_params_decrypted: Optional[Dict[str, str]] = None @@ -3261,8 +3333,20 @@ async def _check_single_gateway_health(self, gateway: DbGateway, user_email: Opt valid = validate_signature(gateway_ca_certificate.encode(), gateway_ca_certificate_sig, public_key_pem) else: valid = True - if valid: - ssl_context = self.create_ssl_context(gateway_ca_certificate) + + # Decrypt client_key for health check mTLS + _hc_client_key = health_client_key + if _hc_client_key: + try: + _enc = get_encryption_service(settings.auth_encryption_secret) + _hc_client_key = _enc.decrypt_secret_or_plaintext(_hc_client_key) + except Exception: + logger.debug("client_key decryption skipped during health check") + + if gateway_url and gateway_url.lower().startswith("http://"): + ssl_context = None + elif valid and gateway_ca_certificate: + ssl_context = get_cached_ssl_context(gateway_ca_certificate, client_cert=health_client_cert, client_key=_hc_client_key) else: ssl_context = None @@ -3605,6 +3689,8 @@ async def _initialize_gateway( include_prompts: bool = True, auth_query_params: Optional[Dict[str, str]] = None, oauth_auto_fetch_tool_flag: Optional[bool] = False, + client_cert: Optional[str] = None, + client_key: Optional[str] = None, ) -> tuple[Dict[str, Any], List[ToolCreate], List[ResourceCreate], List[PromptCreate]]: """Initialize connection to a gateway and retrieve its capabilities. @@ -3626,6 +3712,8 @@ async def _initialize_gateway( oauth_auto_fetch_tool_flag: Whether to skip the early return for OAuth Authorization Code flow. When False (default), auth_code gateways return empty lists immediately (for health checks). When True, attempts to connect even for auth_code gateways (for activation after user authorization). + client_cert: Optional client certificate path or PEM for mTLS + client_key: Optional client private key path or PEM for mTLS Returns: tuple[Dict[str, Any], List[ToolCreate], List[ResourceCreate], List[PromptCreate]]: @@ -3706,9 +3794,13 @@ async def _initialize_gateway( if auth_type in ("basic", "bearer", "authheaders") and isinstance(authentication, str): authentication = decode_auth(authentication) if transport.lower() == "sse": - capabilities, tools, resources, prompts = await self.connect_to_sse_server(url, authentication, ca_certificate, include_prompts, include_resources, auth_query_params) + capabilities, tools, resources, prompts = await self.connect_to_sse_server( + url, authentication, ca_certificate, include_prompts, include_resources, auth_query_params, client_cert=client_cert, client_key=client_key + ) elif transport.lower() == "streamablehttp": - capabilities, tools, resources, prompts = await self.connect_to_streamablehttp_server(url, authentication, ca_certificate, include_prompts, include_resources, auth_query_params) + capabilities, tools, resources, prompts = await self.connect_to_streamablehttp_server( + url, authentication, ca_certificate, include_prompts, include_resources, auth_query_params, client_cert=client_cert, client_key=client_key + ) return capabilities, tools, resources, prompts except Exception as e: @@ -4525,6 +4617,8 @@ async def _refresh_gateway_tools_resources_prompts( gateway_oauth_config = None gateway_ca_certificate = None gateway_auth_query_params = None + refresh_client_cert = None + refresh_client_key = None if gateway: if not gateway.enabled or not gateway.reachable: @@ -4539,6 +4633,8 @@ async def _refresh_gateway_tools_resources_prompts( gateway_oauth_config = gateway.oauth_config gateway_ca_certificate = gateway.ca_certificate gateway_auth_query_params = gateway.auth_query_params + refresh_client_cert = getattr(gateway, "client_cert", None) + refresh_client_key = getattr(gateway, "client_key", None) else: with fresh_db_session() as db: gateway_obj = db.execute(select(DbGateway).where(DbGateway.id == gateway_id)).scalar_one_or_none() @@ -4560,6 +4656,8 @@ async def _refresh_gateway_tools_resources_prompts( gateway_oauth_config = gateway_obj.oauth_config gateway_ca_certificate = gateway_obj.ca_certificate gateway_auth_query_params = gateway_obj.auth_query_params + refresh_client_cert = getattr(gateway_obj, "client_cert", None) + refresh_client_key = getattr(gateway_obj, "client_key", None) # Handle query_param auth - decrypt and apply to URL for refresh auth_query_params_decrypted: Optional[Dict[str, str]] = None @@ -4577,6 +4675,14 @@ async def _refresh_gateway_tools_resources_prompts( # Fetch tools/resources/prompts from MCP server (no DB connection held) try: + # Decrypt client_key for refresh initialization + _refresh_key = refresh_client_key + if _refresh_key: + try: + _enc = get_encryption_service(settings.auth_encryption_secret) + _refresh_key = _enc.decrypt_secret_or_plaintext(_refresh_key) + except Exception: + logger.debug("client_key decryption skipped during gateway refresh") _capabilities, tools, resources, prompts = await self._initialize_gateway( url=gateway_url, authentication=gateway_auth_value, @@ -4588,6 +4694,8 @@ async def _refresh_gateway_tools_resources_prompts( include_resources=include_resources, include_prompts=include_prompts, auth_query_params=auth_query_params_decrypted, + client_cert=refresh_client_cert, + client_key=_refresh_key, ) except Exception as e: logger.warning(f"Failed to fetch tools from gateway {gateway_name}: {e}") @@ -5077,6 +5185,8 @@ async def connect_to_sse_server( include_prompts: bool = True, include_resources: bool = True, auth_query_params: Optional[Dict[str, str]] = None, + client_cert: Optional[str] = None, + client_key: Optional[str] = None, ): """Connect to an MCP server running with SSE transport. @@ -5087,6 +5197,8 @@ async def connect_to_sse_server( include_prompts: Whether to fetch prompts from the server. include_resources: Whether to fetch resources from the server. auth_query_params: Query param names for URL sanitization in error logs. + client_cert: Optional client certificate path or PEM for mTLS. + client_key: Optional client private key path or PEM for mTLS. Returns: Tuple containing (capabilities, tools, resources, prompts) from the MCP server. @@ -5109,8 +5221,10 @@ def get_httpx_client_factory( Returns: httpx.AsyncClient: Configured HTTPX async client """ - if ca_certificate: - ctx = self.create_ssl_context(ca_certificate) + if server_url and server_url.lower().startswith("http://"): + ctx = None + elif ca_certificate: + ctx = get_cached_ssl_context(ca_certificate, client_cert=client_cert, client_key=client_key) else: ctx = None return httpx.AsyncClient( @@ -5237,6 +5351,8 @@ async def connect_to_streamablehttp_server( include_prompts: bool = True, include_resources: bool = True, auth_query_params: Optional[Dict[str, str]] = None, + client_cert: Optional[str] = None, + client_key: Optional[str] = None, ): """Connect to an MCP server running with Streamable HTTP transport. @@ -5247,6 +5363,8 @@ async def connect_to_streamablehttp_server( include_prompts: Whether to fetch prompts from the server. include_resources: Whether to fetch resources from the server. auth_query_params: Query param names for URL sanitization in error logs. + client_cert: Optional client certificate path or PEM for mTLS. + client_key: Optional client private key path or PEM for mTLS. Returns: Tuple containing (capabilities, tools, resources, prompts) from the MCP server. @@ -5270,8 +5388,10 @@ def get_httpx_client_factory( Returns: httpx.AsyncClient: Configured HTTPX async client """ - if ca_certificate: - ctx = self.create_ssl_context(ca_certificate) + if server_url and server_url.lower().startswith("http://"): + ctx = None + elif ca_certificate: + ctx = get_cached_ssl_context(ca_certificate, client_cert=client_cert, client_key=client_key) else: ctx = None return httpx.AsyncClient( diff --git a/mcpgateway/services/mcp_session_pool.py b/mcpgateway/services/mcp_session_pool.py index cff373ef104..ba0955d21dc 100644 --- a/mcpgateway/services/mcp_session_pool.py +++ b/mcpgateway/services/mcp_session_pool.py @@ -1400,6 +1400,33 @@ async def close_all(self) -> None: logger.info("All sessions closed") + async def drain_all(self) -> None: + """Close all pooled and active sessions without marking the pool as closed. + + Unlike ``close_all()``, the pool remains operational after draining. + New sessions will be created on demand with fresh TLS state. + Use this for certificate rotation (SIGHUP). + """ + logger.info("Draining all pooled sessions for TLS rotation...") + + async with self._global_lock: + for _pool_key, pool in list(self._pools.items()): + while not pool.empty(): + try: + pooled = pool.get_nowait() + await self._close_session(pooled) + except asyncio.QueueEmpty: + break + + for _pool_key, active_set in list(self._active.items()): + for pooled in list(active_set): + await self._close_session(pooled) + + self._pools.clear() + self._active.clear() + + logger.info("All pooled sessions drained; pool remains operational") + async def register_pool_session_owner(self, mcp_session_id: str) -> None: """Register this worker as owner of a pool session in Redis. @@ -2114,6 +2141,17 @@ async def close_mcp_session_pool() -> None: pass # Notification service not initialized +async def drain_mcp_session_pool() -> None: + """Drain all sessions from the global pool without destroying the pool. + + Sessions are closed so new ones reconnect with fresh TLS state. + The pool remains operational — unlike ``close_mcp_session_pool()``, + which shuts it down permanently. + """ + if _mcp_session_pool is not None: + await _mcp_session_pool.drain_all() + + async def start_pool_notification_service(gateway_service: Any = None) -> None: """Start the notification service background worker. diff --git a/mcpgateway/services/tool_service.py b/mcpgateway/services/tool_service.py index 288ed8c6ce4..de3fd08eef0 100644 --- a/mcpgateway/services/tool_service.py +++ b/mcpgateway/services/tool_service.py @@ -710,6 +710,8 @@ def _build_tool_cache_payload(self, tool: DbTool, gateway: Optional[DbGateway]) "visibility": gateway.visibility, "tags": gateway.tags or [], "gateway_mode": getattr(gateway, "gateway_mode", "cache"), # Gateway mode for direct proxy support + "client_cert": getattr(gateway, "client_cert", None), + "client_key": getattr(gateway, "client_key", None), } return {"status": "active", "tool": tool_payload, "gateway": gateway_payload} @@ -3913,18 +3915,54 @@ async def invoke_tool( session_short = mcp_session_id[:8] if len(mcp_session_id) >= 8 else mcp_session_id logger.debug(f"[AFFINITY] Worker {worker_id} | Session {session_short}... | Tool: {name} | Normalized MCP-Session-Id → x-mcp-session-id for pool affinity (MCP transport)") - def create_ssl_context(ca_certificate: str) -> ssl.SSLContext: - """Create an SSL context with the provided CA certificate. + # mTLS client cert/key: resolve from payload, then override with runtime gateway if available + client_cert_from_payload = gateway_payload.get("client_cert") if has_gateway else None + client_key_from_payload = gateway_payload.get("client_key") if has_gateway else None + + # Resolve client cert/key: payload values take precedence, runtime values override if present + gateway_client_cert = client_cert_from_payload + gateway_client_key = client_key_from_payload + if has_gateway and gateway is not None: + runtime_gateway_client_cert = getattr(gateway, "client_cert", None) + runtime_gateway_client_key = getattr(gateway, "client_key", None) + if runtime_gateway_client_cert: + gateway_client_cert = runtime_gateway_client_cert + if runtime_gateway_client_key: + gateway_client_key = runtime_gateway_client_key + + # Decrypt client_key if stored encrypted + if gateway_client_key: + try: + # First-Party + from mcpgateway.services.encryption_service import get_encryption_service # pylint: disable=import-outside-toplevel + + _enc = get_encryption_service(settings.auth_encryption_secret) + gateway_client_key = _enc.decrypt_secret_or_plaintext(gateway_client_key) + except Exception as _dec_exc: + logger.debug("client_key decryption skipped, using as-is: %s", _dec_exc) + + def create_ssl_context( + ca_certificate: str, + client_cert: str | None = None, + client_key: str | None = None, + ) -> ssl.SSLContext: + """Create an SSL context with the provided CA certificate and optional mTLS credentials. - Uses caching to avoid repeated SSL context creation for the same certificate. + Uses caching to avoid repeated SSL context creation for the same certificate(s). Args: ca_certificate: CA certificate in PEM format + client_cert: Optional client cert path or PEM for mTLS + client_key: Optional client key path or PEM for mTLS Returns: ssl.SSLContext: Configured SSL context """ - return get_cached_ssl_context(ca_certificate) + return get_cached_ssl_context(ca_certificate, client_cert=client_cert, client_key=client_key) + + # Capture mTLS client cert/key values for passing to nested function + _client_cert_value = gateway_client_cert + _client_key_value = gateway_client_key def get_httpx_client_factory( headers: dict[str, str] | None = None, @@ -3944,6 +3982,9 @@ def get_httpx_client_factory( Raises: Exception: If CA certificate signature is invalid """ + # Use captured client cert/key values from closure + client_cert_value = _client_cert_value + client_key_value = _client_key_value # Use local variables instead of ORM objects (captured from outer scope) valid = False if gateway_ca_cert: @@ -3955,8 +3996,15 @@ def get_httpx_client_factory( # First-Party from mcpgateway.services.http_client_service import get_default_verify, get_http_timeout # pylint: disable=import-outside-toplevel - if valid: - ctx = create_ssl_context(gateway_ca_cert) + # For plain HTTP gateway URLs, skip SSL context entirely to avoid unnecessary SSL setup. + if gateway_url and gateway_url.lower().startswith("http://"): + ctx = None + elif valid and gateway_ca_cert: + ctx = create_ssl_context( + gateway_ca_cert, + client_cert=client_cert_value, + client_key=client_key_value, + ) else: ctx = None diff --git a/mcpgateway/utils/ssl_context_cache.py b/mcpgateway/utils/ssl_context_cache.py index 3f6c66a20cc..72f52dd0df6 100644 --- a/mcpgateway/utils/ssl_context_cache.py +++ b/mcpgateway/utils/ssl_context_cache.py @@ -7,22 +7,132 @@ """ # Standard +from datetime import datetime, timedelta import hashlib +import logging +import os import ssl +import tempfile -# Cache for SSL contexts keyed by CA certificate hash +logger = logging.getLogger(__name__) + +# Cache for SSL contexts keyed by SSL parameter hash _ssl_context_cache: dict[str, ssl.SSLContext] = {} +_ssl_context_cache_timestamps: dict[str, datetime] = {} + +_SSL_CONTEXT_CACHE_MAX_SIZE = int(os.getenv("SSL_CONTEXT_CACHE_MAX_SIZE", "100")) +_SSL_CONTEXT_CACHE_TTL = os.getenv("SSL_CONTEXT_CACHE_TTL") +if _SSL_CONTEXT_CACHE_TTL is not None and _SSL_CONTEXT_CACHE_TTL.strip() != "": + try: + _SSL_CONTEXT_CACHE_TTL = int(_SSL_CONTEXT_CACHE_TTL) + except ValueError: + raise ValueError("SSL_CONTEXT_CACHE_TTL must be an integer number of seconds") +else: + _SSL_CONTEXT_CACHE_TTL = None + + +def _is_expired(cache_key: str) -> bool: + """Check if a cached SSL context entry has expired based on TTL. + + Args: + cache_key: The cache key to check for expiration. + + Returns: + True if the entry has expired and should be refreshed, False otherwise. + """ + if _SSL_CONTEXT_CACHE_TTL is None: + return False + created_at = _ssl_context_cache_timestamps.get(cache_key) + if created_at is None: + return False + + return datetime.now() - created_at > timedelta(seconds=_SSL_CONTEXT_CACHE_TTL) + + +def _is_pem(value: str) -> bool: + """Check if a string looks like inline PEM content rather than a file path. + + Args: + value: String to check (file path or PEM content). + + Returns: + True if the string starts with a PEM header. + """ + return value.lstrip().startswith("-----BEGIN ") + + +def _load_client_cert_chain(ctx: ssl.SSLContext, client_cert: str, client_key: str) -> None: + """Load client cert/key into an SSL context, handling both file paths and PEM strings. + ``ssl.SSLContext.load_cert_chain`` only accepts file paths. When the + values are inline PEM content (stored in the database), we write them + to secure temporary files and load from there. -def get_cached_ssl_context(ca_certificate: str) -> ssl.SSLContext: + Args: + ctx: SSL context to load the client certificate chain into. + client_cert: Client certificate as a file path or inline PEM string. + client_key: Client private key as a file path or inline PEM string. + """ + cert_is_pem = _is_pem(client_cert) + key_is_pem = _is_pem(client_key) + + if not cert_is_pem and not key_is_pem: + # Both are file paths — use directly + ctx.load_cert_chain(certfile=client_cert, keyfile=client_key) + return + + # At least one value is inline PEM — write temp files + cert_tmp = key_tmp = None + try: + if cert_is_pem: + cert_tmp = tempfile.NamedTemporaryFile(mode="w", suffix=".pem", delete=False, encoding="utf-8") + cert_tmp.write(client_cert) + cert_tmp.close() + cert_path = cert_tmp.name + else: + cert_path = client_cert + + if key_is_pem: + key_tmp = tempfile.NamedTemporaryFile(mode="w", suffix=".pem", delete=False, encoding="utf-8") + key_tmp.write(client_key) + key_tmp.close() + key_path = key_tmp.name + else: + key_path = client_key + + ctx.load_cert_chain(certfile=cert_path, keyfile=key_path) + finally: + # Remove temp files immediately after loading + if cert_tmp is not None: + try: + os.unlink(cert_tmp.name) + except OSError: + logger.debug("Failed to remove temp cert file %s", cert_tmp.name) + if key_tmp is not None: + try: + os.unlink(key_tmp.name) + except OSError: + logger.debug("Failed to remove temp key file %s", key_tmp.name) + + +def get_cached_ssl_context( + ca_certificate: str, + client_cert: str | None = None, + client_key: str | None = None, +) -> ssl.SSLContext: """Get or create cached SSL context for a CA certificate. Args: ca_certificate: CA certificate in PEM format (str or bytes) + client_cert: Optional client cert path or PEM for mTLS + client_key: Optional client key path or PEM for mTLS Returns: ssl.SSLContext: Configured SSL context + Raises: + ValueError: If only one of client_cert/client_key is provided. + Examples: The actual `ssl.SSLContext.load_verify_locations()` call requires valid PEM data; in doctests we mock it to focus on caching behavior. @@ -43,28 +153,66 @@ def get_cached_ssl_context(ca_certificate: str) -> ssl.SSLContext: SSL contexts are cached by the SHA256 hash of the certificate to avoid repeated expensive SSL setup operations. """ - # Handle bytes, string, or other types (e.g., MagicMock in tests) + # Ensure CA certificate is normalized to bytes for hash calculation if isinstance(ca_certificate, bytes): - cert_bytes = ca_certificate + ca_cert_bytes = ca_certificate elif isinstance(ca_certificate, str): - cert_bytes = ca_certificate.encode() + ca_cert_bytes = ca_certificate.encode() else: - # For non-string/non-bytes (e.g., MagicMock in tests), convert to string first - cert_bytes = str(ca_certificate).encode() + ca_cert_bytes = str(ca_certificate).encode() + + # Client cert/key may be either path-like content or inlined PEM string. + client_cert_value = client_cert or "" + client_key_value = client_key or "" - cert_hash = hashlib.sha256(cert_bytes).hexdigest() + # Build stable cache key incrementally (avoids delimiter collisions). + key_hash = hashlib.sha256() + key_hash.update(b"ca_cert:") + key_hash.update(ca_cert_bytes) + key_hash.update(b"|client_cert:") + key_hash.update(client_cert_value.encode()) + key_hash.update(b"|client_key:") + key_hash.update(client_key_value.encode()) - if cert_hash in _ssl_context_cache: - return _ssl_context_cache[cert_hash] + cache_key = key_hash.hexdigest() - # Create new SSL context + if cache_key in _ssl_context_cache and not _is_expired(cache_key): + return _ssl_context_cache[cache_key] + + # If expired, clear this entry so it is refreshed below + if cache_key in _ssl_context_cache: + _ssl_context_cache.pop(cache_key, None) + _ssl_context_cache_timestamps.pop(cache_key, None) + + # Create new SSL context and configure CA cert ctx = ssl.create_default_context() ctx.load_verify_locations(cadata=ca_certificate) - # Cache it (limit cache size) - if len(_ssl_context_cache) > 100: + # Validate mTLS: require both or neither + if bool(client_cert) != bool(client_key): + raise ValueError("mTLS requires both client_cert and client_key; got only one") + + # Load client certificates for mTLS when provided + if client_cert and client_key: + _load_client_cert_chain(ctx, client_cert, client_key) + + # Cache entry creation timestamp if TTL is enabled + _ssl_context_cache[cache_key] = ctx + if _SSL_CONTEXT_CACHE_TTL is not None: + _ssl_context_cache_timestamps[cache_key] = datetime.now() + + # Evict all cache if size limit exceeded; keep this newly inserted item. + # This avoids growing indefinitely without requiring LRU tracking. + if len(_ssl_context_cache) > _SSL_CONTEXT_CACHE_MAX_SIZE: + current_ctx = _ssl_context_cache.pop(cache_key) + current_ts = _ssl_context_cache_timestamps.pop(cache_key, None) + _ssl_context_cache.clear() - _ssl_context_cache[cert_hash] = ctx + _ssl_context_cache_timestamps.clear() + + _ssl_context_cache[cache_key] = current_ctx + if current_ts is not None: + _ssl_context_cache_timestamps[cache_key] = current_ts return ctx @@ -89,3 +237,4 @@ def clear_ssl_context_cache() -> None: 2 """ _ssl_context_cache.clear() + _ssl_context_cache_timestamps.clear() diff --git a/tests/unit/mcpgateway/services/test_gateway_service.py b/tests/unit/mcpgateway/services/test_gateway_service.py index afdf6b5afd3..e4b7d82942b 100644 --- a/tests/unit/mcpgateway/services/test_gateway_service.py +++ b/tests/unit/mcpgateway/services/test_gateway_service.py @@ -642,6 +642,133 @@ def _capture_add(obj): assert stored_gateway.oauth_config["grant_type"] == "password" assert stored_gateway.oauth_config["client_id"] == "cid" + @pytest.mark.asyncio + async def test_encrypt_client_key_encrypts_plaintext(self, gateway_service): + """_encrypt_client_key encrypts a plaintext private key.""" + from mcpgateway.services.gateway_service import GatewayService + + result = await GatewayService._encrypt_client_key("-----BEGIN PRIVATE KEY-----\nSECRET\n-----END PRIVATE KEY-----") + encryption = get_encryption_service(settings.auth_encryption_secret) + assert encryption.is_encrypted(result) + + @pytest.mark.asyncio + async def test_encrypt_client_key_returns_none_for_empty(self, gateway_service): + """_encrypt_client_key returns None for None or empty input.""" + from mcpgateway.services.gateway_service import GatewayService + + assert await GatewayService._encrypt_client_key(None) is None + assert await GatewayService._encrypt_client_key("") is None + + @pytest.mark.asyncio + async def test_encrypt_client_key_idempotent(self, gateway_service): + """_encrypt_client_key does not double-encrypt already-encrypted values.""" + from mcpgateway.services.gateway_service import GatewayService + + encryption = get_encryption_service(settings.auth_encryption_secret) + already_encrypted = encryption.encrypt_secret("my-key") + result = await GatewayService._encrypt_client_key(already_encrypted) + assert result == already_encrypted + + @pytest.mark.asyncio + async def test_register_gateway_encrypts_client_key(self, gateway_service, test_db, monkeypatch): + """register_gateway encrypts client_key before persistence.""" + test_db.execute = Mock(return_value=_make_execute_result(scalar=None)) + test_db.flush = Mock() + test_db.refresh = Mock() + test_db.query = Mock(return_value=Mock(filter=Mock(return_value=Mock(all=Mock(return_value=[]))))) + gateway_service._initialize_gateway = AsyncMock(return_value=({"tools": {"listChanged": True}}, [], [], [])) + gateway_service._notify_gateway_added = AsyncMock() + + captured_gateway: dict[str, object] = {} + + def _capture_add(obj): + if isinstance(obj, DbGateway): + captured_gateway["gateway"] = obj + + test_db.add = Mock(side_effect=_capture_add) + + mock_model = Mock() + mock_model.masked.return_value = mock_model + monkeypatch.setattr("mcpgateway.services.gateway_service.GatewayRead.model_validate", lambda x: mock_model) + + gateway_create = GatewayCreate( + name="mtls_gateway", + url="https://example.com/gateway", + description="mTLS gateway", + client_cert="/path/to/cert.pem", + client_key="-----BEGIN PRIVATE KEY-----\nSECRET\n-----END PRIVATE KEY-----", + ) + + await gateway_service.register_gateway(test_db, gateway_create) + + stored_gateway = captured_gateway.get("gateway") + assert stored_gateway is not None + assert stored_gateway.client_cert == "/path/to/cert.pem" + encryption = get_encryption_service(settings.auth_encryption_secret) + assert encryption.is_encrypted(stored_gateway.client_key) + + @pytest.mark.asyncio + async def test_update_gateway_persists_ca_and_mtls_fields(self, gateway_service, mock_gateway, test_db): + """update_gateway persists ca_certificate, ca_certificate_sig, signing_algorithm, client_cert, and client_key.""" + mock_gateway.team_id = 1 + execute_results = [_make_execute_result(scalar=mock_gateway), _make_execute_result(scalar=None)] + test_db.execute = Mock(side_effect=execute_results) + test_db.commit = Mock() + test_db.refresh = Mock() + + gateway_service._initialize_gateway = AsyncMock( + return_value=({"tools": {"subscribe": True}}, [], [], []) + ) + gateway_service._notify_gateway_updated = AsyncMock() + + gateway_update = GatewayUpdate( + ca_certificate="-----BEGIN CERTIFICATE-----\nNEWCA\n-----END CERTIFICATE-----", + ca_certificate_sig="newsig123", + signing_algorithm="sha256", + client_cert="/path/to/new-cert.pem", + client_key="-----BEGIN PRIVATE KEY-----\nNEWKEY\n-----END PRIVATE KEY-----", + ) + + mock_gateway_read = MagicMock() + mock_gateway_read.masked.return_value = mock_gateway_read + + with patch("mcpgateway.services.gateway_service.GatewayRead.model_validate", return_value=mock_gateway_read): + await gateway_service.update_gateway(test_db, 1, gateway_update) + + assert mock_gateway.ca_certificate == "-----BEGIN CERTIFICATE-----\nNEWCA\n-----END CERTIFICATE-----" + assert mock_gateway.ca_certificate_sig == "newsig123" + assert mock_gateway.signing_algorithm == "sha256" + assert mock_gateway.client_cert == "/path/to/new-cert.pem" + # client_key should be encrypted + encryption = get_encryption_service(settings.auth_encryption_secret) + assert encryption.is_encrypted(mock_gateway.client_key) + + @pytest.mark.asyncio + async def test_update_gateway_preserves_masked_client_key(self, gateway_service, mock_gateway, test_db): + """update_gateway skips re-encryption when client_key is the masked placeholder.""" + mock_gateway.team_id = 1 + mock_gateway.client_key = "existing-encrypted-value" + execute_results = [_make_execute_result(scalar=mock_gateway), _make_execute_result(scalar=None)] + test_db.execute = Mock(side_effect=execute_results) + test_db.commit = Mock() + test_db.refresh = Mock() + + gateway_service._initialize_gateway = AsyncMock( + return_value=({"tools": {"subscribe": True}}, [], [], []) + ) + gateway_service._notify_gateway_updated = AsyncMock() + + gateway_update = GatewayUpdate(client_key=settings.masked_auth_value) + + mock_gateway_read = MagicMock() + mock_gateway_read.masked.return_value = mock_gateway_read + + with patch("mcpgateway.services.gateway_service.GatewayRead.model_validate", return_value=mock_gateway_read): + await gateway_service.update_gateway(test_db, 1, gateway_update) + + # Should NOT have changed — masked value preserves existing + assert mock_gateway.client_key == "existing-encrypted-value" + @pytest.mark.asyncio async def test_register_gateway_exception_rollback(self, gateway_service, test_db): """Test rollback on exception during gateway registration.""" @@ -5220,6 +5347,113 @@ async def test_health_check_query_param_auth(self, gateway_service, monkeypatch) await gateway_service._check_single_gateway_health(gw) + @pytest.mark.asyncio + async def test_health_check_with_mtls_ca_certificate(self, gateway_service, monkeypatch): + """Health check creates SSL context with mTLS client_cert/client_key (lines 3348-3349).""" + encryption = get_encryption_service(settings.auth_encryption_secret) + encrypted_key = encryption.encrypt_secret("my-client-key") + gw = _make_gateway( + id="gw-mtls", + name="mtls-gw", + url="https://example.com/sse", + enabled=True, + reachable=True, + transport="sse", + auth_type=None, + auth_value=None, + auth_query_params=None, + ca_certificate="FAKE_CA_PEM", + ca_certificate_sig=None, + oauth_config=None, + last_refresh_at=None, + refresh_interval_seconds=None, + client_cert="/path/to/cert.pem", + client_key=encrypted_key, + ) + + mock_response = AsyncMock() + mock_response.status_code = 200 + mock_response.raise_for_status = MagicMock() + mock_stream_response = AsyncMock() + mock_stream_response.__aenter__ = AsyncMock(return_value=mock_response) + mock_stream_response.__aexit__ = AsyncMock(return_value=False) + mock_client = MagicMock() + mock_client.stream = MagicMock(return_value=mock_stream_response) + mock_ctx = AsyncMock() + mock_ctx.__aenter__ = AsyncMock(return_value=mock_client) + mock_ctx.__aexit__ = AsyncMock(return_value=False) + monkeypatch.setattr("mcpgateway.services.gateway_service.get_isolated_http_client", lambda **kw: mock_ctx) + monkeypatch.setattr("mcpgateway.services.gateway_service.fresh_db_session", MagicMock()) + mock_settings = MagicMock( + enable_ed25519_signing=False, health_check_timeout=5, auto_refresh_servers=False, + httpx_admin_read_timeout=5, mcp_session_pool_enabled=False, + auth_encryption_secret=settings.auth_encryption_secret, + ) + monkeypatch.setattr("mcpgateway.services.gateway_service.settings", mock_settings) + monkeypatch.setattr("mcpgateway.services.gateway_service.create_span", MagicMock(return_value=MagicMock(__enter__=MagicMock(return_value=MagicMock()), __exit__=MagicMock(return_value=False)))) + + with patch("mcpgateway.services.gateway_service.get_cached_ssl_context") as mock_ssl: + mock_ssl.return_value = MagicMock() + await gateway_service._check_single_gateway_health(gw) + + mock_ssl.assert_called_once() + call_kw = mock_ssl.call_args + assert call_kw[1]["client_cert"] == "/path/to/cert.pem" + assert call_kw[1]["client_key"] == "my-client-key" # decrypted + + @pytest.mark.asyncio + async def test_health_check_decrypt_exception_uses_key_as_is(self, gateway_service, monkeypatch): + """Health check falls back to raw key when decryption fails (lines 3343-3344).""" + gw = _make_gateway( + id="gw-dec-fail", + name="dec-fail-gw", + url="https://example.com/sse", + enabled=True, + reachable=True, + transport="sse", + auth_type=None, + auth_value=None, + auth_query_params=None, + ca_certificate="FAKE_CA_PEM", + ca_certificate_sig=None, + oauth_config=None, + last_refresh_at=None, + refresh_interval_seconds=None, + client_cert="/path/cert.pem", + client_key="raw-unencrypted-key", + ) + + mock_response = AsyncMock() + mock_response.status_code = 200 + mock_response.raise_for_status = MagicMock() + mock_stream_response = AsyncMock() + mock_stream_response.__aenter__ = AsyncMock(return_value=mock_response) + mock_stream_response.__aexit__ = AsyncMock(return_value=False) + mock_client = MagicMock() + mock_client.stream = MagicMock(return_value=mock_stream_response) + mock_ctx = AsyncMock() + mock_ctx.__aenter__ = AsyncMock(return_value=mock_client) + mock_ctx.__aexit__ = AsyncMock(return_value=False) + monkeypatch.setattr("mcpgateway.services.gateway_service.get_isolated_http_client", lambda **kw: mock_ctx) + monkeypatch.setattr("mcpgateway.services.gateway_service.fresh_db_session", MagicMock()) + monkeypatch.setattr( + "mcpgateway.services.gateway_service.settings", + MagicMock(enable_ed25519_signing=False, health_check_timeout=5, auto_refresh_servers=False, httpx_admin_read_timeout=5, mcp_session_pool_enabled=False), + ) + monkeypatch.setattr("mcpgateway.services.gateway_service.create_span", MagicMock(return_value=MagicMock(__enter__=MagicMock(return_value=MagicMock()), __exit__=MagicMock(return_value=False)))) + + with ( + patch("mcpgateway.services.gateway_service.get_cached_ssl_context") as mock_ssl, + patch("mcpgateway.services.gateway_service.get_encryption_service", side_effect=RuntimeError("no enc")), + ): + mock_ssl.return_value = MagicMock() + await gateway_service._check_single_gateway_health(gw) + + mock_ssl.assert_called_once() + call_kw = mock_ssl.call_args + assert call_kw[1]["client_key"] == "raw-unencrypted-key" + + # --------------------------------------------------------------------------- # list_gateways_for_user tests # --------------------------------------------------------------------------- @@ -5426,6 +5660,36 @@ async def test_oauth_auth_code_with_flag(self, gateway_service): ) assert len(tools) == 1 + @pytest.mark.asyncio + async def test_initialize_gateway_passes_mtls_to_sse(self, gateway_service): + """_initialize_gateway passes client_cert/client_key to connect_to_sse_server (line 5227).""" + gateway_service.connect_to_sse_server = AsyncMock(return_value=({"tools": {}}, [], [], [])) + await gateway_service._initialize_gateway( + url="https://example.com", + transport="SSE", + ca_certificate=b"CA_PEM", + client_cert="/path/cert.pem", + client_key="my-key", + ) + call_kw = gateway_service.connect_to_sse_server.call_args + assert call_kw[1]["client_cert"] == "/path/cert.pem" + assert call_kw[1]["client_key"] == "my-key" + + @pytest.mark.asyncio + async def test_initialize_gateway_passes_mtls_to_streamablehttp(self, gateway_service): + """_initialize_gateway passes client_cert/client_key to connect_to_streamablehttp_server (line 5393-5394).""" + gateway_service.connect_to_streamablehttp_server = AsyncMock(return_value=({"tools": {}}, [], [], [])) + await gateway_service._initialize_gateway( + url="https://example.com", + transport="StreamableHTTP", + ca_certificate=b"CA_PEM", + client_cert="/path/cert.pem", + client_key="my-key", + ) + call_kw = gateway_service.connect_to_streamablehttp_server.call_args + assert call_kw[1]["client_cert"] == "/path/cert.pem" + assert call_kw[1]["client_key"] == "my-key" + # --------------------------------------------------------------------------- # _refresh_gateway_tools_resources_prompts tests @@ -7146,3 +7410,69 @@ async def test_initialize_gateway_exception_group_empty_str(gateway_service): await gateway_service._initialize_gateway(url="http://localhost:9999", transport="SSE") assert "OSError" in str(exc_info.value) + + +# --------------------------------------------------------------------------- +# mTLS decrypt exception branches in update/activate/refresh +# --------------------------------------------------------------------------- + + +class TestMtlsDecryptExceptionBranches: + """Cover decrypt exception fallback branches in update re-init, activation, and refresh.""" + + @pytest.mark.asyncio + async def test_update_reinit_decrypt_exception_uses_key_as_is(self, gateway_service, mock_gateway, monkeypatch): + """update_gateway re-init falls back to raw key when decryption fails (lines 2164-2165).""" + mock_gateway.team_id = 1 + mock_gateway.client_cert = "/path/cert.pem" + mock_gateway.client_key = "raw-key-not-encrypted" + execute_results = [_make_execute_result(scalar=mock_gateway), _make_execute_result(scalar=None)] + test_db = MagicMock() + test_db.execute = Mock(side_effect=execute_results) + test_db.commit = Mock() + test_db.refresh = Mock() + + gateway_service._initialize_gateway = AsyncMock(return_value=({"tools": {}}, [], [], [])) + gateway_service._notify_gateway_updated = AsyncMock() + + gateway_update = GatewayUpdate(url="http://example.com/updated") + + mock_gateway_read = MagicMock() + mock_gateway_read.masked.return_value = mock_gateway_read + + with ( + patch("mcpgateway.services.gateway_service.GatewayRead.model_validate", return_value=mock_gateway_read), + patch("mcpgateway.services.gateway_service.get_encryption_service", side_effect=RuntimeError("no enc")), + ): + await gateway_service.update_gateway(test_db, 1, gateway_update) + + init_call = gateway_service._initialize_gateway.call_args + assert init_call[1]["client_key"] == "raw-key-not-encrypted" + + @pytest.mark.asyncio + async def test_refresh_decrypt_exception_uses_key_as_is(self, gateway_service, monkeypatch): + """_refresh_gateway_tools_resources_prompts falls back to raw key when decryption fails (lines 4684-4685).""" + gw = SimpleNamespace( + id="gw-r", + name="refresh-gw", + url="http://example.com", + transport="SSE", + auth_type=None, + auth_value=None, + oauth_config=None, + ca_certificate=None, + auth_query_params=None, + enabled=True, + reachable=True, + client_cert="/path/cert.pem", + client_key="raw-key", + ) + + gateway_service._initialize_gateway = AsyncMock(return_value=({}, [], [], [])) + monkeypatch.setattr("mcpgateway.services.gateway_service.fresh_db_session", MagicMock()) + + with patch("mcpgateway.services.gateway_service.get_encryption_service", side_effect=RuntimeError("no enc")): + result = await gateway_service._refresh_gateway_tools_resources_prompts("gw-r", gateway=gw) + + init_call = gateway_service._initialize_gateway.call_args + assert init_call[1]["client_key"] == "raw-key" diff --git a/tests/unit/mcpgateway/services/test_mcp_session_pool_coverage.py b/tests/unit/mcpgateway/services/test_mcp_session_pool_coverage.py index 289f65df8f6..b236dfee7f4 100644 --- a/tests/unit/mcpgateway/services/test_mcp_session_pool_coverage.py +++ b/tests/unit/mcpgateway/services/test_mcp_session_pool_coverage.py @@ -23,6 +23,7 @@ TransportType, _get_cleanup_timeout, close_mcp_session_pool, + drain_mcp_session_pool, get_mcp_session_pool, init_mcp_session_pool, register_gateway_capabilities_for_notifications, @@ -1240,6 +1241,97 @@ async def never_ending(): assert pool._rpc_listener_task is None +# --------------------------------------------------------------------------- +# drain_all: close sessions without marking pool as closed +# --------------------------------------------------------------------------- +class TestDrainAll: + """Cover MCPSessionPool.drain_all() and module-level drain_mcp_session_pool().""" + + @pytest.mark.asyncio + async def test_drain_all_closes_pooled_sessions(self): + """drain_all should close pooled sessions but keep pool operational.""" + pool = MCPSessionPool() + url = "http://test:8080" + pool_key = ("anonymous", url, "anonymous", "streamablehttp", "") + pool._pools[pool_key] = asyncio.Queue(maxsize=2) + pool._active[pool_key] = set() + + s1 = PooledSession( + session=MagicMock(), transport_context=MagicMock(), + url=url, identity_key="anonymous", + transport_type=TransportType.STREAMABLE_HTTP, headers={}, + ) + pool._pools[pool_key].put_nowait(s1) + + with patch.object(pool, "_close_session", new_callable=AsyncMock): + await pool.drain_all() + + # Pool should be operational (not closed) + assert pool._closed is False + assert len(pool._pools) == 0 + + @pytest.mark.asyncio + async def test_drain_all_closes_active_sessions(self): + """drain_all should close active sessions.""" + pool = MCPSessionPool() + url = "http://test:8080" + pool_key = ("anonymous", url, "anonymous", "streamablehttp", "") + pool._pools[pool_key] = asyncio.Queue(maxsize=2) + pool._active[pool_key] = set() + + active_s = MagicMock() + pool._active[pool_key].add(active_s) + + with patch.object(pool, "_close_session", new_callable=AsyncMock) as mock_close: + await pool.drain_all() + + mock_close.assert_awaited() + assert pool._closed is False + + @pytest.mark.asyncio + async def test_drain_all_handles_queue_empty_race(self): + """drain_all handles QueueEmpty if queue empties between empty() and get_nowait().""" + pool = MCPSessionPool() + pool_key = ("anonymous", "http://test:8080", "anonymous", "streamablehttp", "") + + fake_queue = MagicMock() + fake_queue.empty = MagicMock(side_effect=[False, True]) + fake_queue.get_nowait = MagicMock(side_effect=asyncio.QueueEmpty()) + + pool._pools[pool_key] = fake_queue + pool._active[pool_key] = set() + + await pool.drain_all() + assert pool._closed is False + + @pytest.mark.asyncio + async def test_drain_mcp_session_pool_calls_drain_all(self): + """Module-level drain_mcp_session_pool() calls drain_all on the singleton.""" + import mcpgateway.services.mcp_session_pool as pool_mod + + mock_pool = MagicMock() + mock_pool.drain_all = AsyncMock() + original = pool_mod._mcp_session_pool + try: + pool_mod._mcp_session_pool = mock_pool + await drain_mcp_session_pool() + mock_pool.drain_all.assert_awaited_once() + finally: + pool_mod._mcp_session_pool = original + + @pytest.mark.asyncio + async def test_drain_mcp_session_pool_noop_when_no_pool(self): + """Module-level drain_mcp_session_pool() is a no-op when pool is None.""" + import mcpgateway.services.mcp_session_pool as pool_mod + + original = pool_mod._mcp_session_pool + try: + pool_mod._mcp_session_pool = None + await drain_mcp_session_pool() # Should not raise + finally: + pool_mod._mcp_session_pool = original + + # --------------------------------------------------------------------------- # Lines 1308-1342: register_pool_session_owner # --------------------------------------------------------------------------- diff --git a/tests/unit/mcpgateway/services/test_tool_service_coverage.py b/tests/unit/mcpgateway/services/test_tool_service_coverage.py index 7174619c419..535b4f1bf8e 100644 --- a/tests/unit/mcpgateway/services/test_tool_service_coverage.py +++ b/tests/unit/mcpgateway/services/test_tool_service_coverage.py @@ -421,6 +421,22 @@ def test_build_tool_cache_payload_with_gateway(self, tool_service, mock_tool, mo assert payload["gateway"] is not None assert payload["gateway"]["id"] == str(mock_gateway.id) + def test_build_tool_cache_payload_includes_mtls_fields(self, tool_service, mock_tool, mock_gateway): + """Cache payload should include client_cert and client_key for mTLS.""" + mock_gateway.client_cert = "/path/to/client.pem" + mock_gateway.client_key = "/path/to/client-key.pem" + payload = tool_service._build_tool_cache_payload(mock_tool, mock_gateway) + assert payload["gateway"]["client_cert"] == "/path/to/client.pem" + assert payload["gateway"]["client_key"] == "/path/to/client-key.pem" + + def test_build_tool_cache_payload_mtls_fields_none_by_default(self, tool_service, mock_tool, mock_gateway): + """Cache payload should include client_cert/client_key as None when not set.""" + mock_gateway.client_cert = None + mock_gateway.client_key = None + payload = tool_service._build_tool_cache_payload(mock_tool, mock_gateway) + assert payload["gateway"]["client_cert"] is None + assert payload["gateway"]["client_key"] is None + def test_build_tool_cache_payload_without_gateway(self, tool_service, mock_tool): """Cache payload without gateway should have gateway=None.""" payload = tool_service._build_tool_cache_payload(mock_tool, None) @@ -4788,11 +4804,11 @@ def _make_tool_payload( } -def _make_gateway_payload(*, auth_type=None, auth_value=None, auth_query_params=None, oauth_config=None, ca_certificate=None, ca_certificate_sig=None, passthrough_headers=None): +def _make_gateway_payload(*, auth_type=None, auth_value=None, auth_query_params=None, oauth_config=None, ca_certificate=None, ca_certificate_sig=None, passthrough_headers=None, url=None, client_cert=None, client_key=None): return { "id": "gw-uuid-1", "name": "test_gw", - "url": "http://gateway:9000", + "url": url or "http://gateway:9000", "auth_type": auth_type, "auth_value": auth_value, "auth_query_params": auth_query_params, @@ -4800,6 +4816,8 @@ def _make_gateway_payload(*, auth_type=None, auth_value=None, auth_query_params= "ca_certificate": ca_certificate, "ca_certificate_sig": ca_certificate_sig, "passthrough_headers": passthrough_headers, + "client_cert": client_cert, + "client_key": client_key, } @@ -6840,6 +6858,275 @@ async def test_mcp_gateway_oauth_client_credentials_error_raises(self, tool_serv with pytest.raises(ToolInvocationError, match="OAuth authentication failed for gateway"): await tool_service.invoke_tool(db, "test_tool", {}) + @pytest.mark.asyncio + async def test_mcp_http_url_bypasses_ssl_context_creation(self, tool_service): + """HTTP URLs should skip SSL context creation entirely (line 3986-3988).""" + tp = _make_tool_payload(integration_type="MCP", request_type="SSE", gateway_id="gw-uuid-1", jsonpath_filter="") + gp = _make_gateway_payload( + url="http://localhost:9000/sse", # HTTP URL - should bypass SSL + auth_type="basic", + ca_certificate="dummy-ca", # Even with CA cert, should skip for HTTP + ) + db = MagicMock() + + def fake_sse_client(*, url=None, headers=None, httpx_client_factory=None, **_kw): + class _CM: + async def __aenter__(self): + # Call the factory to trigger SSL context logic + if httpx_client_factory is not None: + client = httpx_client_factory(headers=headers) + # Verify no SSL context was created for HTTP URL + assert client is not None + return (MagicMock(), MagicMock(), AsyncMock()) + + async def __aexit__(self, *exc): + return False + + return _CM() + + mock_session = AsyncMock() + mock_session.initialize = AsyncMock() + mock_session.call_tool = AsyncMock(return_value=ToolResult(content=[TextContent(type="text", text="ok")], is_error=False)) + + class _SessionCM: + async def __aenter__(self): + return mock_session + + async def __aexit__(self, *exc): + return False + + with ( + _setup_cache_for_invoke(tp, gp), + patch.object(tool_service, "_check_tool_access", AsyncMock(return_value=True)), + patch("mcpgateway.services.tool_service.global_config_cache") as mock_gcc, + patch("mcpgateway.services.tool_service.current_trace_id") as mock_trace, + patch("mcpgateway.services.tool_service.create_span") as mock_span_ctx, + patch("mcpgateway.services.metrics_buffer_service.get_metrics_buffer_service") as mock_mbuf, + patch("mcpgateway.services.tool_service.compute_passthrough_headers_cached", return_value={}), + patch("mcpgateway.services.tool_service.sse_client", side_effect=fake_sse_client), + patch("mcpgateway.services.tool_service.ClientSession", return_value=_SessionCM()), + patch("mcpgateway.services.tool_service.httpx.AsyncClient", return_value=MagicMock()), + patch("mcpgateway.services.tool_service.get_cached_ssl_context") as mock_get_ssl, + patch.object(settings, "mcp_session_pool_enabled", False), + ): + mock_gcc.get_passthrough_headers = MagicMock(return_value=[]) + mock_trace.get = MagicMock(return_value=None) + mock_span_ctx.return_value.__enter__ = MagicMock(return_value=MagicMock()) + mock_span_ctx.return_value.__exit__ = MagicMock(return_value=False) + mock_mbuf.return_value = MagicMock() + + result = await tool_service.invoke_tool(db, "test_tool", {}, request_headers=None) + + assert result is not None + # Verify SSL context was NOT created for HTTP URL + assert mock_get_ssl.call_count == 0, "SSL context should not be created for HTTP URLs" + + @pytest.mark.asyncio + async def test_mcp_https_url_with_mtls_creates_ssl_context(self, tool_service): + """HTTPS URLs with client cert/key should create SSL context with mTLS (lines 3990-3994).""" + tp = _make_tool_payload(integration_type="MCP", request_type="SSE", gateway_id="gw-uuid-1", jsonpath_filter="") + gp = _make_gateway_payload( + url="https://localhost:9000/sse", # HTTPS URL - should create SSL context + auth_type="basic", + ca_certificate="dummy-ca", + client_cert="client-cert-data", # mTLS client cert + client_key="client-key-data", # mTLS client key + ) + db = MagicMock() + + def fake_sse_client(*, url=None, headers=None, httpx_client_factory=None, **_kw): + class _CM: + async def __aenter__(self): + if httpx_client_factory is not None: + httpx_client_factory(headers=headers) + return (MagicMock(), MagicMock(), AsyncMock()) + + async def __aexit__(self, *exc): + return False + + return _CM() + + mock_session = AsyncMock() + mock_session.initialize = AsyncMock() + mock_session.call_tool = AsyncMock(return_value=ToolResult(content=[TextContent(type="text", text="ok")], is_error=False)) + + class _SessionCM: + async def __aenter__(self): + return mock_session + + async def __aexit__(self, *exc): + return False + + with ( + _setup_cache_for_invoke(tp, gp), + patch.object(tool_service, "_check_tool_access", AsyncMock(return_value=True)), + patch("mcpgateway.services.tool_service.global_config_cache") as mock_gcc, + patch("mcpgateway.services.tool_service.current_trace_id") as mock_trace, + patch("mcpgateway.services.tool_service.create_span") as mock_span_ctx, + patch("mcpgateway.services.metrics_buffer_service.get_metrics_buffer_service") as mock_mbuf, + patch("mcpgateway.services.tool_service.compute_passthrough_headers_cached", return_value={}), + patch("mcpgateway.services.tool_service.sse_client", side_effect=fake_sse_client), + patch("mcpgateway.services.tool_service.ClientSession", return_value=_SessionCM()), + patch("mcpgateway.services.tool_service.httpx.AsyncClient", return_value=MagicMock()), + patch("mcpgateway.services.tool_service.get_cached_ssl_context") as mock_get_ssl, + patch.object(settings, "enable_ed25519_signing", False), + patch.object(settings, "mcp_session_pool_enabled", False), + ): + mock_gcc.get_passthrough_headers = MagicMock(return_value=[]) + mock_trace.get = MagicMock(return_value=None) + mock_span_ctx.return_value.__enter__ = MagicMock(return_value=MagicMock()) + mock_span_ctx.return_value.__exit__ = MagicMock(return_value=False) + mock_mbuf.return_value = MagicMock() + mock_get_ssl.return_value = MagicMock() # Mock SSL context + + result = await tool_service.invoke_tool(db, "test_tool", {}, request_headers=None) + + assert result is not None + # Verify SSL context WAS created for HTTPS URL with mTLS + assert mock_get_ssl.call_count == 1, "SSL context should be created for HTTPS URLs" + # Verify client cert and key were passed + call_args = mock_get_ssl.call_args + assert call_args is not None + assert call_args[1].get("client_cert") == "client-cert-data" + assert call_args[1].get("client_key") == "client-key-data" + + @pytest.mark.asyncio + async def test_mcp_https_url_decrypts_encrypted_client_key(self, tool_service): + """Encrypted client_key in gateway payload is decrypted before SSL context creation.""" + from mcpgateway.services.encryption_service import get_encryption_service + + encryption = get_encryption_service(settings.auth_encryption_secret) + encrypted_key = encryption.encrypt_secret("decrypted-client-key") + + tp = _make_tool_payload(integration_type="MCP", request_type="SSE", gateway_id="gw-uuid-1", jsonpath_filter="") + gp = _make_gateway_payload( + url="https://localhost:9000/sse", + auth_type="basic", + ca_certificate="dummy-ca", + client_cert="client-cert-data", + client_key=encrypted_key, + ) + db = MagicMock() + + def fake_sse_client(*, url=None, headers=None, httpx_client_factory=None, **_kw): + class _CM: + async def __aenter__(self): + if httpx_client_factory is not None: + httpx_client_factory(headers=headers) + return (MagicMock(), MagicMock(), AsyncMock()) + + async def __aexit__(self, *exc): + return False + + return _CM() + + mock_session = AsyncMock() + mock_session.initialize = AsyncMock() + mock_session.call_tool = AsyncMock(return_value=ToolResult(content=[TextContent(type="text", text="ok")], is_error=False)) + + class _SessionCM: + async def __aenter__(self): + return mock_session + + async def __aexit__(self, *exc): + return False + + with ( + _setup_cache_for_invoke(tp, gp), + patch.object(tool_service, "_check_tool_access", AsyncMock(return_value=True)), + patch("mcpgateway.services.tool_service.global_config_cache") as mock_gcc, + patch("mcpgateway.services.tool_service.current_trace_id") as mock_trace, + patch("mcpgateway.services.tool_service.create_span") as mock_span_ctx, + patch("mcpgateway.services.metrics_buffer_service.get_metrics_buffer_service") as mock_mbuf, + patch("mcpgateway.services.tool_service.compute_passthrough_headers_cached", return_value={}), + patch("mcpgateway.services.tool_service.sse_client", side_effect=fake_sse_client), + patch("mcpgateway.services.tool_service.ClientSession", return_value=_SessionCM()), + patch("mcpgateway.services.tool_service.httpx.AsyncClient", return_value=MagicMock()), + patch("mcpgateway.services.tool_service.get_cached_ssl_context") as mock_get_ssl, + patch.object(settings, "enable_ed25519_signing", False), + patch.object(settings, "mcp_session_pool_enabled", False), + ): + mock_gcc.get_passthrough_headers = MagicMock(return_value=[]) + mock_trace.get = MagicMock(return_value=None) + mock_span_ctx.return_value.__enter__ = MagicMock(return_value=MagicMock()) + mock_span_ctx.return_value.__exit__ = MagicMock(return_value=False) + mock_mbuf.return_value = MagicMock() + mock_get_ssl.return_value = MagicMock() + + result = await tool_service.invoke_tool(db, "test_tool", {}, request_headers=None) + + assert result is not None + assert mock_get_ssl.call_count == 1 + call_args = mock_get_ssl.call_args + # Verify the key was decrypted (not the encrypted bundle) + assert call_args[1].get("client_key") == "decrypted-client-key" + + @pytest.mark.asyncio + async def test_mcp_https_url_decryption_failure_uses_key_as_is(self, tool_service): + """When client_key decryption fails, the raw value is passed through.""" + tp = _make_tool_payload(integration_type="MCP", request_type="SSE", gateway_id="gw-uuid-1", jsonpath_filter="") + gp = _make_gateway_payload( + url="https://localhost:9000/sse", + auth_type="basic", + ca_certificate="dummy-ca", + client_cert="client-cert-data", + client_key="not-encrypted-raw-key", + ) + db = MagicMock() + + def fake_sse_client(*, url=None, headers=None, httpx_client_factory=None, **_kw): + class _CM: + async def __aenter__(self): + if httpx_client_factory is not None: + httpx_client_factory(headers=headers) + return (MagicMock(), MagicMock(), AsyncMock()) + + async def __aexit__(self, *exc): + return False + + return _CM() + + mock_session = AsyncMock() + mock_session.initialize = AsyncMock() + mock_session.call_tool = AsyncMock(return_value=ToolResult(content=[TextContent(type="text", text="ok")], is_error=False)) + + class _SessionCM: + async def __aenter__(self): + return mock_session + + async def __aexit__(self, *exc): + return False + + with ( + _setup_cache_for_invoke(tp, gp), + patch.object(tool_service, "_check_tool_access", AsyncMock(return_value=True)), + patch("mcpgateway.services.tool_service.global_config_cache") as mock_gcc, + patch("mcpgateway.services.tool_service.current_trace_id") as mock_trace, + patch("mcpgateway.services.tool_service.create_span") as mock_span_ctx, + patch("mcpgateway.services.metrics_buffer_service.get_metrics_buffer_service") as mock_mbuf, + patch("mcpgateway.services.tool_service.compute_passthrough_headers_cached", return_value={}), + patch("mcpgateway.services.tool_service.sse_client", side_effect=fake_sse_client), + patch("mcpgateway.services.tool_service.ClientSession", return_value=_SessionCM()), + patch("mcpgateway.services.tool_service.httpx.AsyncClient", return_value=MagicMock()), + patch("mcpgateway.services.tool_service.get_cached_ssl_context") as mock_get_ssl, + patch("mcpgateway.services.encryption_service.get_encryption_service", side_effect=RuntimeError("no encryption")), + patch.object(settings, "enable_ed25519_signing", False), + patch.object(settings, "mcp_session_pool_enabled", False), + ): + mock_gcc.get_passthrough_headers = MagicMock(return_value=[]) + mock_trace.get = MagicMock(return_value=None) + mock_span_ctx.return_value.__enter__ = MagicMock(return_value=MagicMock()) + mock_span_ctx.return_value.__exit__ = MagicMock(return_value=False) + mock_mbuf.return_value = MagicMock() + mock_get_ssl.return_value = MagicMock() + + result = await tool_service.invoke_tool(db, "test_tool", {}, request_headers=None) + + assert result is not None + assert mock_get_ssl.call_count == 1 + # Key should be passed as-is since decryption failed + call_args = mock_get_ssl.call_args + assert call_args[1].get("client_key") == "not-encrypted-raw-key" @pytest.mark.asyncio async def test_mcp_sse_httpx_factory_validates_ed25519_signature(self, tool_service): @@ -7026,7 +7313,7 @@ def pool_session(**kwargs): patch("mcpgateway.services.tool_service.get_correlation_id", return_value="corr-1"), patch("mcpgateway.services.tool_service.compute_passthrough_headers_cached", side_effect=lambda _rh, h, *_a, **_k: h), patch("mcpgateway.services.tool_service.get_mcp_session_pool", return_value=pool), - patch("mcpgateway.services.tool_service.get_cached_ssl_context", return_value=MagicMock()), + patch("mcpgateway.services.tool_service.get_cached_ssl_context") as mock_cached_ssl_context, patch("mcpgateway.services.tool_service.httpx.AsyncClient", return_value=MagicMock()), patch.object(settings, "enable_ed25519_signing", False), patch.object(settings, "mcp_session_pool_enabled", True), @@ -7039,6 +7326,7 @@ def pool_session(**kwargs): result = await tool_service.invoke_tool(db, "test_tool", {}, request_headers=None) assert result is not None + assert mock_cached_ssl_context.call_count == 0 assert captured_pool_kwargs.get("transport_type") is not None diff --git a/tests/unit/mcpgateway/test_main_sighup.py b/tests/unit/mcpgateway/test_main_sighup.py new file mode 100644 index 00000000000..cbdb497a814 --- /dev/null +++ b/tests/unit/mcpgateway/test_main_sighup.py @@ -0,0 +1,92 @@ +# -*- coding: utf-8 -*- +"""Unit tests for SIGHUP signal handler in mcpgateway.handlers.signal_handlers.""" + +# Standard +import asyncio +import signal +from unittest.mock import AsyncMock, patch + +# Third-Party +import pytest + +# First-Party +from mcpgateway.handlers.signal_handlers import sighup_handler, sighup_reload + + +@pytest.mark.asyncio +async def test_sighup_reload_clears_ssl_cache_and_session_pool(): + """sighup_reload() clears SSL context cache and closes MCP session pool.""" + with ( + patch("mcpgateway.utils.ssl_context_cache.clear_ssl_context_cache") as mock_clear, + patch("mcpgateway.services.mcp_session_pool.drain_mcp_session_pool", new_callable=AsyncMock) as mock_pool_close, + patch("mcpgateway.handlers.signal_handlers.logger") as mock_logger, + ): + await sighup_reload() + mock_clear.assert_called_once() + mock_pool_close.assert_awaited_once() + info_messages = [call.args[0] for call in mock_logger.info.call_args_list] + assert any("SSL context cache cleared" in m for m in info_messages) + assert any("session pool drained" in m for m in info_messages) + + +@pytest.mark.asyncio +async def test_sighup_reload_logs_error_on_ssl_cache_exception(): + """sighup_reload() catches and logs exceptions from clear_ssl_context_cache.""" + with ( + patch("mcpgateway.utils.ssl_context_cache.clear_ssl_context_cache", side_effect=RuntimeError("boom")), + patch("mcpgateway.services.mcp_session_pool.drain_mcp_session_pool", new_callable=AsyncMock), + patch("mcpgateway.handlers.signal_handlers.logger") as mock_logger, + ): + await sighup_reload() + mock_logger.error.assert_called_once() + assert "boom" in mock_logger.error.call_args[0][0] + + +@pytest.mark.asyncio +async def test_sighup_reload_handles_session_pool_error(): + """sighup_reload() continues if session pool close fails.""" + with ( + patch("mcpgateway.utils.ssl_context_cache.clear_ssl_context_cache") as mock_clear, + patch("mcpgateway.services.mcp_session_pool.drain_mcp_session_pool", new_callable=AsyncMock, side_effect=RuntimeError("pool error")), + patch("mcpgateway.handlers.signal_handlers.logger") as mock_logger, + ): + await sighup_reload() + # SSL cache should still be cleared + mock_clear.assert_called_once() + # Pool error should be logged at debug level + debug_messages = [call.args[0] for call in mock_logger.debug.call_args_list] + assert any("pool error" in m for m in debug_messages) + + +@pytest.mark.asyncio +async def test_sighup_handler_schedules_task(): + """sighup_handler() schedules sighup_reload on the running event loop.""" + loop = asyncio.get_running_loop() + task_created = False + original_create_task = loop.create_task + + def tracking_create_task(coro, **kwargs): + nonlocal task_created + task_created = True + return original_create_task(coro, **kwargs) + + with ( + patch.object(loop, "create_task", side_effect=tracking_create_task), + patch("mcpgateway.utils.ssl_context_cache.clear_ssl_context_cache"), + patch("mcpgateway.services.mcp_session_pool.drain_mcp_session_pool", new_callable=AsyncMock), + ): + sighup_handler(signal.SIGHUP, None) + await asyncio.sleep(0.05) + + assert task_created + + +def test_sighup_handler_logs_warning_when_no_event_loop(): + """sighup_handler() logs warning when no event loop is running.""" + with ( + patch("mcpgateway.handlers.signal_handlers.asyncio.get_running_loop", side_effect=RuntimeError("No loop")), + patch("mcpgateway.handlers.signal_handlers.logger") as mock_logger, + ): + sighup_handler(signal.SIGHUP, None) + mock_logger.warning.assert_called_once() + assert "not running" in mock_logger.warning.call_args[0][0] diff --git a/tests/unit/mcpgateway/test_schemas_validators_extra.py b/tests/unit/mcpgateway/test_schemas_validators_extra.py index cc773be4e6a..8b8f2209ddd 100644 --- a/tests/unit/mcpgateway/test_schemas_validators_extra.py +++ b/tests/unit/mcpgateway/test_schemas_validators_extra.py @@ -914,6 +914,28 @@ def test_gateway_read_masked_includes_oauth(self): assert masked.oauth_config["client_secret"] == settings.masked_auth_value assert masked.oauth_config["token_url"] == "https://auth.example.com/token" + def test_gateway_read_masked_hides_client_key(self): + """GatewayRead.masked() masks client_key (mTLS private key).""" + gw = GatewayRead( + name="test-gw", + url="http://example.com", + client_cert="/path/to/cert.pem", + client_key="-----BEGIN PRIVATE KEY-----\nSECRETKEY\n-----END PRIVATE KEY-----", + ) + masked = gw.masked() + assert masked.client_key == settings.masked_auth_value + assert masked.client_cert == "/path/to/cert.pem" + + def test_gateway_read_masked_preserves_none_client_key(self): + """GatewayRead.masked() preserves None client_key.""" + gw = GatewayRead( + name="test-gw", + url="http://example.com", + client_key=None, + ) + masked = gw.masked() + assert masked.client_key is None + def test_a2a_agent_read_masked_includes_oauth(self): """A2AAgentRead.masked() masks oauth_config sensitive keys.""" now = datetime.now(timezone.utc) diff --git a/tests/unit/mcpgateway/utils/test_ssl_context_cache.py b/tests/unit/mcpgateway/utils/test_ssl_context_cache.py index 8083adfef79..c2c4322683f 100644 --- a/tests/unit/mcpgateway/utils/test_ssl_context_cache.py +++ b/tests/unit/mcpgateway/utils/test_ssl_context_cache.py @@ -3,9 +3,13 @@ # Standard import hashlib +from datetime import datetime, timedelta from types import SimpleNamespace from unittest.mock import Mock, patch +# Third-Party +import pytest + # First-Party import mcpgateway.utils.ssl_context_cache as ssl_context_cache @@ -58,7 +62,16 @@ def test_get_cached_ssl_context_clears_cache_when_over_limit() -> None: _ = ssl_context_cache.get_cached_ssl_context("NEWCERT") - expected_hash = hashlib.sha256(b"NEWCERT").hexdigest() + # Cache key now includes component labels and no delimiter-ambiguity hash. + key_hash = hashlib.sha256() + key_hash.update(b"ca_cert:") + key_hash.update(b"NEWCERT") + key_hash.update(b"|client_cert:") + key_hash.update(b"") + key_hash.update(b"|client_key:") + key_hash.update(b"") + expected_hash = key_hash.hexdigest() + assert list(ssl_context_cache._ssl_context_cache.keys()) == [expected_hash] # noqa: SLF001 - testing internal cache behavior @@ -71,3 +84,255 @@ def test_clear_ssl_context_cache_forces_recreate() -> None: _ = ssl_context_cache.get_cached_ssl_context("CERTDATA") assert mock_create.call_count == 2 + + +def test_get_cached_ssl_context_loads_client_cert_and_key_paths() -> None: + """File-path client_cert/client_key are passed directly to load_cert_chain.""" + with patch("mcpgateway.utils.ssl_context_cache.ssl.create_default_context") as mock_create: + ctx = Mock() + mock_create.return_value = ctx + + _ = ssl_context_cache.get_cached_ssl_context( + "CA_CERT", + client_cert="/path/to/cert.pem", + client_key="/path/to/key.pem", + ) + + assert ctx.load_verify_locations.called + ctx.load_cert_chain.assert_called_once_with(certfile="/path/to/cert.pem", keyfile="/path/to/key.pem") + + +def test_get_cached_ssl_context_loads_pem_content_via_tempfiles() -> None: + """Inline PEM content is written to temp files for load_cert_chain.""" + pem_cert = "-----BEGIN CERTIFICATE-----\nFAKECERT\n-----END CERTIFICATE-----" + pem_key = "-----BEGIN PRIVATE KEY-----\nFAKEKEY\n-----END PRIVATE KEY-----" + + with patch("mcpgateway.utils.ssl_context_cache.ssl.create_default_context") as mock_create: + ctx = Mock() + mock_create.return_value = ctx + + _ = ssl_context_cache.get_cached_ssl_context("CA_CERT", client_cert=pem_cert, client_key=pem_key) + + assert ctx.load_cert_chain.call_count == 1 + call_args = ctx.load_cert_chain.call_args + # Should be called with temp file paths, not the PEM strings themselves + assert call_args.kwargs["certfile"] != pem_cert + assert call_args.kwargs["keyfile"] != pem_key + assert isinstance(call_args.kwargs["certfile"], str) + assert isinstance(call_args.kwargs["keyfile"], str) + + +def test_get_cached_ssl_context_rejects_cert_without_key() -> None: + """Providing client_cert without client_key raises ValueError.""" + with patch("mcpgateway.utils.ssl_context_cache.ssl.create_default_context"): + with pytest.raises(ValueError, match="both client_cert and client_key"): + ssl_context_cache.get_cached_ssl_context("CA_CERT", client_cert="/path/cert.pem", client_key=None) + + +def test_get_cached_ssl_context_rejects_key_without_cert() -> None: + """Providing client_key without client_cert raises ValueError.""" + with patch("mcpgateway.utils.ssl_context_cache.ssl.create_default_context"): + with pytest.raises(ValueError, match="both client_cert and client_key"): + ssl_context_cache.get_cached_ssl_context("CA_CERT", client_cert=None, client_key="/path/key.pem") + + +def test_load_client_cert_chain_mixed_pem_and_path() -> None: + """When cert is PEM but key is a path, only cert uses a temp file.""" + pem_cert = "-----BEGIN CERTIFICATE-----\nFAKECERT\n-----END CERTIFICATE-----" + + with patch("mcpgateway.utils.ssl_context_cache.ssl.create_default_context") as mock_create: + ctx = Mock() + mock_create.return_value = ctx + + _ = ssl_context_cache.get_cached_ssl_context("CA_CERT", client_cert=pem_cert, client_key="/path/to/key.pem") + + assert ctx.load_cert_chain.call_count == 1 + call_args = ctx.load_cert_chain.call_args + # cert should be a temp file path (not the PEM string), key should be the original path + assert call_args.kwargs["certfile"] != pem_cert + assert call_args.kwargs["keyfile"] == "/path/to/key.pem" + + +def test_load_client_cert_chain_mixed_path_cert_pem_key() -> None: + """When cert is a path but key is PEM, only key uses a temp file.""" + pem_key = "-----BEGIN PRIVATE KEY-----\nFAKEKEY\n-----END PRIVATE KEY-----" + + with patch("mcpgateway.utils.ssl_context_cache.ssl.create_default_context") as mock_create: + ctx = Mock() + mock_create.return_value = ctx + + _ = ssl_context_cache.get_cached_ssl_context("CA_CERT", client_cert="/path/to/cert.pem", client_key=pem_key) + + assert ctx.load_cert_chain.call_count == 1 + call_args = ctx.load_cert_chain.call_args + assert call_args.kwargs["certfile"] == "/path/to/cert.pem" + assert call_args.kwargs["keyfile"] != pem_key + + +def test_load_client_cert_chain_cleanup_oserror_is_handled() -> None: + """OSError during temp file cleanup is logged, not raised.""" + pem_cert = "-----BEGIN CERTIFICATE-----\nFAKE\n-----END CERTIFICATE-----" + pem_key = "-----BEGIN PRIVATE KEY-----\nFAKE\n-----END PRIVATE KEY-----" + + with ( + patch("mcpgateway.utils.ssl_context_cache.ssl.create_default_context") as mock_create, + patch("mcpgateway.utils.ssl_context_cache.os.unlink", side_effect=OSError("perm denied")), + ): + ctx = Mock() + mock_create.return_value = ctx + + # Should not raise despite unlink failure + _ = ssl_context_cache.get_cached_ssl_context("CA_CERT", client_cert=pem_cert, client_key=pem_key) + + assert ctx.load_cert_chain.call_count == 1 + + +def test_cache_key_different_for_client_cert_changes() -> None: + with patch("mcpgateway.utils.ssl_context_cache.ssl.create_default_context") as mock_create: + ctx1 = Mock() + ctx2 = Mock() + mock_create.side_effect = [ctx1, ctx2] + + a = ssl_context_cache.get_cached_ssl_context( + "CA_CERT", + client_cert="CLIENT_CERT_A", + client_key="CLIENT_KEY_A", + ) + b = ssl_context_cache.get_cached_ssl_context( + "CA_CERT", + client_cert="CLIENT_CERT_B", + client_key="CLIENT_KEY_A", + ) + + assert a is ctx1 + assert b is ctx2 + assert mock_create.call_count == 2 + + +def test_is_expired_returns_false_when_ttl_disabled(monkeypatch): + monkeypatch.setattr(ssl_context_cache, "_SSL_CONTEXT_CACHE_TTL", None) + key = "expired-entry" + ssl_context_cache._ssl_context_cache_timestamps[key] = datetime.now() - timedelta(seconds=100) + + assert ssl_context_cache._is_expired(key) is False + + +def test_is_expired_returns_true_when_entry_ttl_elapsed(monkeypatch): + monkeypatch.setattr(ssl_context_cache, "_SSL_CONTEXT_CACHE_TTL", 1) + key = "expired-entry" + ssl_context_cache._ssl_context_cache_timestamps[key] = datetime.now() - timedelta(seconds=2) + + assert ssl_context_cache._is_expired(key) is True + + +def test_ssl_context_cache_ttl_invalid_value_raises_error(): + """Test that invalid SSL_CONTEXT_CACHE_TTL raises ValueError during module import.""" + import importlib + import os + import sys + + # Save original module if loaded + original_module = sys.modules.get("mcpgateway.utils.ssl_context_cache") + + try: + # Remove module from cache to force reload + if "mcpgateway.utils.ssl_context_cache" in sys.modules: + del sys.modules["mcpgateway.utils.ssl_context_cache"] + + # Set invalid TTL value + os.environ["SSL_CONTEXT_CACHE_TTL"] = "not-a-number" + + # Import should raise ValueError + with patch.dict(os.environ, {"SSL_CONTEXT_CACHE_TTL": "not-a-number"}): + try: + import mcpgateway.utils.ssl_context_cache + # If we get here, manually trigger the validation logic + ttl_val = os.getenv("SSL_CONTEXT_CACHE_TTL") + if ttl_val and ttl_val.strip(): + int(ttl_val) # Should raise ValueError + assert False, "Expected ValueError was not raised" + except ValueError as e: + assert "SSL_CONTEXT_CACHE_TTL must be an integer" in str(e) or "invalid literal" in str(e).lower() + finally: + # Restore original module and clean up environment + if "SSL_CONTEXT_CACHE_TTL" in os.environ: + del os.environ["SSL_CONTEXT_CACHE_TTL"] + if original_module: + sys.modules["mcpgateway.utils.ssl_context_cache"] = original_module + else: + if "mcpgateway.utils.ssl_context_cache" in sys.modules: + del sys.modules["mcpgateway.utils.ssl_context_cache"] + + +def test_ttl_env_var_parsing_with_invalid_value(monkeypatch): + """Test that invalid SSL_CONTEXT_CACHE_TTL raises ValueError.""" + import importlib + + monkeypatch.setenv("SSL_CONTEXT_CACHE_TTL", "invalid") + + with patch.object(importlib, "reload") as mock_reload: + try: + # Simulate module reload with invalid TTL + import mcpgateway.utils.ssl_context_cache as module + # Manually trigger the parsing logic + ttl_value = "invalid" + if ttl_value.strip() != "": + int(ttl_value) # This should raise ValueError + except ValueError as e: + assert "invalid literal" in str(e).lower() or "int" in str(e).lower() + + +def test_expired_entry_gets_refreshed(monkeypatch): + """Test that expired cache entries are removed and recreated.""" + monkeypatch.setattr(ssl_context_cache, "_SSL_CONTEXT_CACHE_TTL", 1) + + with patch("mcpgateway.utils.ssl_context_cache.ssl.create_default_context") as mock_create: + ctx1 = Mock() + ctx2 = Mock() + mock_create.side_effect = [ctx1, ctx2] + + # Create initial entry + result1 = ssl_context_cache.get_cached_ssl_context("CERT") + assert result1 is ctx1 + + # Manually expire the entry + cache_key = list(ssl_context_cache._ssl_context_cache.keys())[0] + ssl_context_cache._ssl_context_cache_timestamps[cache_key] = datetime.now() - timedelta(seconds=2) + + # Request again - should create new context + result2 = ssl_context_cache.get_cached_ssl_context("CERT") + assert result2 is ctx2 + assert mock_create.call_count == 2 + + +def test_cache_eviction_preserves_current_entry_timestamp(monkeypatch): + """Test that cache eviction preserves the timestamp of the newly added entry.""" + monkeypatch.setattr(ssl_context_cache, "_SSL_CONTEXT_CACHE_TTL", 3600) + + # Pre-fill cache to trigger eviction + ssl_context_cache._ssl_context_cache.update({f"key{i}": Mock() for i in range(101)}) + ssl_context_cache._ssl_context_cache_timestamps.update({f"key{i}": datetime.now() for i in range(101)}) + + with patch("mcpgateway.utils.ssl_context_cache.ssl.create_default_context") as mock_create: + ctx = Mock() + mock_create.return_value = ctx + + before_time = datetime.now() + _ = ssl_context_cache.get_cached_ssl_context("NEWCERT") + after_time = datetime.now() + + # Verify only one entry remains + assert len(ssl_context_cache._ssl_context_cache) == 1 + assert len(ssl_context_cache._ssl_context_cache_timestamps) == 1 + + # Verify timestamp was preserved + cache_key = list(ssl_context_cache._ssl_context_cache.keys())[0] + timestamp = ssl_context_cache._ssl_context_cache_timestamps[cache_key] + assert before_time <= timestamp <= after_time + + +def test_is_expired_returns_false_when_no_timestamp(monkeypatch): + """Test that _is_expired returns False when entry has no timestamp but TTL is enabled.""" + monkeypatch.setattr(ssl_context_cache, "_SSL_CONTEXT_CACHE_TTL", 60) + result = ssl_context_cache._is_expired("nonexistent-key") + assert result is False