Coverage for mcpgateway / plugins / framework / settings.py: 100%
246 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-09 03:05 +0000
« prev ^ index » next coverage.py v7.13.4, created at 2026-03-09 03:05 +0000
1# -*- coding: utf-8 -*-
2"""Location: ./mcpgateway/plugins/framework/settings.py
4Copyright 2025
5SPDX-License-Identifier: Apache-2.0
6Authors: Fred Araujo
8Plugin framework configuration.
10Self-contained settings for the plugin framework, eliminating the
11dependency on mcpgateway.config.settings.
12"""
14# Standard
15from functools import lru_cache
16import logging
17import os
18from typing import Any, Literal
20# Third-Party
21from pydantic import AliasChoices, Field, field_validator, SecretStr
22from pydantic_settings import BaseSettings, SettingsConfigDict
24logger = logging.getLogger(__name__)
27def _empty_string_to_none(value: Any) -> Any:
28 """Treat empty optional env vars as unset (None).
30 Shared validator for optional fields that may arrive as empty strings
31 from the environment. Used by ``@field_validator(..., mode="before")``
32 across multiple lightweight settings classes.
34 Args:
35 value: The raw value from the environment variable.
37 Returns:
38 None if the value is an empty string, otherwise the original value.
39 """
40 if isinstance(value, str) and value.strip() == "":
41 return None
42 return value
45class PluginsSettings(BaseSettings):
46 """Plugin framework configuration.
48 All settings can be overridden via environment variables with the PLUGINS_ prefix.
49 For example: PLUGINS_ENABLED=true, PLUGINS_PLUGIN_TIMEOUT=60, PLUGINS_SKIP_SSL_VERIFY=true
50 """
52 enabled: bool = Field(default=False, description="Enable the plugin framework")
53 default_hook_policy: Literal["allow", "deny"] = Field(
54 default="allow",
55 description=(
56 "Default behavior for hooks without an explicit policy: 'allow' accepts all modifications"
57 " (backwards compatible), 'deny' rejects all. Standard hooks always have explicit policies;"
58 " this only affects custom hook types. Set to 'deny' for stricter production environments."
59 ),
60 )
61 config_file: str = Field(default="plugins/config.yaml", description="Path to main plugins configuration file")
62 plugin_timeout: int = Field(default=30, description="Plugin execution timeout in seconds")
63 log_level: str = Field(default="INFO", description="Logging level for plugin framework components")
64 skip_ssl_verify: bool = Field(
65 default=False,
66 description="Skip SSL certificate verification for plugin HTTP requests. WARNING: Only enable in dev environments with self-signed certificates.",
67 )
68 ssrf_protection_enabled: bool = Field(
69 default=True,
70 description=(
71 "Enable SSRF protection for plugin endpoint URLs. Blocks private/reserved IP ranges"
72 " (10.x, 172.16.x, 192.168.x, 127.x, 169.254.x). Disable for development or sidecar"
73 " plugin configurations that use private IPs."
74 ),
75 )
77 # HTTP client settings
78 httpx_max_connections: int = Field(default=200, description="Maximum total concurrent HTTP connections for plugin requests")
79 httpx_max_keepalive_connections: int = Field(default=100, description="Maximum idle keepalive connections to retain (typically 50%% of max_connections)")
80 httpx_keepalive_expiry: float = Field(default=30.0, description="Seconds before idle keepalive connections are closed")
81 httpx_connect_timeout: float = Field(default=5.0, description="Timeout in seconds for establishing new connections (5s for LAN, increase for WAN)")
82 httpx_read_timeout: float = Field(default=120.0, description="Timeout in seconds for reading response data (set high for slow MCP tool calls)")
83 httpx_write_timeout: float = Field(default=30.0, description="Timeout in seconds for writing request data")
84 httpx_pool_timeout: float = Field(default=10.0, description="Timeout in seconds waiting for a connection from the pool (fail fast on exhaustion)")
86 # CLI settings
87 cli_completion: bool = Field(default=False, description="Enable shell auto-completion for the mcpplugins CLI")
88 cli_markup_mode: Literal["markdown", "rich", "disabled"] | None = Field(default=None, description="Markup renderer for CLI output (rich, markdown, or disabled)")
90 # MCP client mTLS settings
91 client_mtls_certfile: str | None = Field(default=None, description="Path to PEM client certificate for mTLS")
92 client_mtls_keyfile: str | None = Field(default=None, description="Path to PEM client private key for mTLS")
93 client_mtls_ca_bundle: str | None = Field(default=None, description="Path to CA bundle for client certificate verification")
94 client_mtls_keyfile_password: SecretStr | None = Field(default=None, description="Password for encrypted client private key")
95 client_mtls_verify: bool | None = Field(default=None, description="Verify the upstream server certificate")
96 client_mtls_check_hostname: bool | None = Field(default=None, description="Enable hostname verification")
98 # MCP server SSL settings
99 server_ssl_keyfile: str | None = Field(default=None, description="Path to PEM server private key")
100 server_ssl_certfile: str | None = Field(default=None, description="Path to PEM server certificate")
101 server_ssl_ca_certs: str | None = Field(default=None, description="Path to CA certificates for client verification")
102 server_ssl_keyfile_password: SecretStr | None = Field(default=None, description="Password for encrypted server private key")
103 server_ssl_cert_reqs: int | None = Field(default=None, description="Client certificate requirement (0=NONE, 1=OPTIONAL, 2=REQUIRED)")
105 # MCP server settings
106 server_host: str | None = Field(default=None, description="MCP server host to bind to")
107 server_port: int | None = Field(default=None, description="MCP server port to bind to")
108 server_uds: str | None = Field(default=None, description="Unix domain socket path for MCP streamable HTTP")
109 server_ssl_enabled: bool | None = Field(default=None, description="Enable SSL/TLS for the MCP server")
111 # MCP runtime settings
112 config_path: str | None = Field(default=None, description="Path to plugin configuration file for external servers")
113 transport: str | None = Field(default=None, description="Transport type for external MCP server (http, stdio)")
115 # gRPC client mTLS settings
116 grpc_client_mtls_certfile: str | None = Field(default=None, description="Path to PEM client certificate for gRPC mTLS")
117 grpc_client_mtls_keyfile: str | None = Field(default=None, description="Path to PEM client private key for gRPC mTLS")
118 grpc_client_mtls_ca_bundle: str | None = Field(default=None, description="Path to CA bundle for gRPC client verification")
119 grpc_client_mtls_keyfile_password: SecretStr | None = Field(default=None, description="Password for encrypted gRPC client private key")
120 grpc_client_mtls_verify: bool | None = Field(default=None, description="Verify the gRPC upstream server certificate")
122 # gRPC server SSL settings
123 grpc_server_ssl_keyfile: str | None = Field(default=None, description="Path to PEM gRPC server private key")
124 grpc_server_ssl_certfile: str | None = Field(default=None, description="Path to PEM gRPC server certificate")
125 grpc_server_ssl_ca_certs: str | None = Field(default=None, description="Path to CA certificates for gRPC client verification")
126 grpc_server_ssl_keyfile_password: SecretStr | None = Field(default=None, description="Password for encrypted gRPC server private key")
127 grpc_server_ssl_client_auth: str | None = Field(default=None, description="gRPC client certificate requirement (none, optional, require)")
129 # gRPC server settings
130 grpc_server_host: str | None = Field(default=None, description="gRPC server host to bind to")
131 grpc_server_port: int | None = Field(default=None, description="gRPC server port to bind to")
132 grpc_server_uds: str | None = Field(default=None, description="Unix domain socket path for gRPC server")
133 grpc_server_ssl_enabled: bool | None = Field(default=None, description="Enable SSL/TLS for the gRPC server")
135 # Unix socket settings
136 unix_socket_path: str | None = Field(default=None, description="Path to the Unix domain socket", validation_alias=AliasChoices("PLUGINS_UNIX_SOCKET_PATH", "UNIX_SOCKET_PATH"))
138 @field_validator(
139 "client_mtls_verify",
140 "client_mtls_check_hostname",
141 "server_ssl_cert_reqs",
142 "server_port",
143 "server_ssl_enabled",
144 "grpc_client_mtls_verify",
145 "grpc_server_port",
146 "grpc_server_ssl_enabled",
147 mode="before",
148 )
149 @classmethod
150 def empty_string_to_none(cls, value: Any) -> Any:
151 """Delegate to shared validator.
153 Args:
154 value: The raw field value from environment or input.
156 Returns:
157 The original value, or None if the value was an empty string.
158 """
159 return _empty_string_to_none(value)
161 model_config = SettingsConfigDict(env_prefix="PLUGINS_", env_file=".env", env_file_encoding="utf-8", extra="ignore")
164class PluginsEnabledSettings(BaseSettings):
165 """Lightweight settings model for reading PLUGINS_ENABLED only."""
167 enabled: bool = False
168 model_config = SettingsConfigDict(env_prefix="PLUGINS_", env_file=".env", env_file_encoding="utf-8", extra="ignore")
171class PluginsConfigPathSettings(BaseSettings):
172 """Lightweight settings model for reading PLUGINS_CONFIG_PATH only."""
174 config_path: str | None = None
175 model_config = SettingsConfigDict(env_prefix="PLUGINS_", env_file=".env", env_file_encoding="utf-8", extra="ignore")
178class PluginsStartupSettings(BaseSettings):
179 """Lightweight settings for fields read during gateway startup.
181 Reads only ``config_file`` and ``plugin_timeout`` so that malformed
182 unrelated plugin env vars (e.g. ``PLUGINS_SERVER_PORT=abc``) do not
183 prevent the gateway from booting.
184 """
186 config_file: str = Field(default="plugins/config.yaml")
187 plugin_timeout: int = 30
188 model_config = SettingsConfigDict(env_prefix="PLUGINS_", env_file=".env", env_file_encoding="utf-8", extra="ignore")
191class PluginsPolicySettings(BaseSettings):
192 """Lightweight settings model for reading default hook policy only."""
194 default_hook_policy: Literal["allow", "deny"] = "allow"
195 model_config = SettingsConfigDict(env_prefix="PLUGINS_", env_file=".env", env_file_encoding="utf-8", extra="ignore")
198class PluginsSsrfSettings(BaseSettings):
199 """Lightweight settings model for reading SSRF protection flag only."""
201 ssrf_protection_enabled: bool = True
202 model_config = SettingsConfigDict(env_prefix="PLUGINS_", env_file=".env", env_file_encoding="utf-8", extra="ignore")
205class PluginsTransportSettings(BaseSettings):
206 """Lightweight settings for transport type and Unix socket path."""
208 transport: str | None = None
209 unix_socket_path: str | None = Field(default=None, validation_alias=AliasChoices("UNIX_SOCKET_PATH", "PLUGINS_UNIX_SOCKET_PATH"))
210 model_config = SettingsConfigDict(env_prefix="PLUGINS_", env_file=".env", env_file_encoding="utf-8", extra="ignore")
213class PluginsClientMtlsSettings(BaseSettings):
214 """Lightweight settings for MCP client mTLS configuration."""
216 client_mtls_certfile: str | None = None
217 client_mtls_keyfile: str | None = None
218 client_mtls_ca_bundle: str | None = None
219 client_mtls_keyfile_password: SecretStr | None = None
220 client_mtls_verify: bool | None = None
221 client_mtls_check_hostname: bool | None = None
223 @field_validator("client_mtls_verify", "client_mtls_check_hostname", mode="before")
224 @classmethod
225 def empty_string_to_none(cls, value: Any) -> Any:
226 """Delegate to shared validator.
228 Args:
229 value: The raw field value from environment or input.
231 Returns:
232 The original value, or None if the value was an empty string.
233 """
234 return _empty_string_to_none(value)
236 model_config = SettingsConfigDict(env_prefix="PLUGINS_", env_file=".env", env_file_encoding="utf-8", extra="ignore")
239class PluginsMcpServerSettings(BaseSettings):
240 """Lightweight settings for MCP server configuration."""
242 server_ssl_keyfile: str | None = None
243 server_ssl_certfile: str | None = None
244 server_ssl_ca_certs: str | None = None
245 server_ssl_keyfile_password: SecretStr | None = None
246 server_ssl_cert_reqs: int | None = None
247 server_host: str | None = None
248 server_port: int | None = None
249 server_uds: str | None = None
250 server_ssl_enabled: bool | None = None
252 @field_validator("server_ssl_cert_reqs", "server_port", "server_ssl_enabled", mode="before")
253 @classmethod
254 def empty_string_to_none(cls, value: Any) -> Any:
255 """Delegate to shared validator.
257 Args:
258 value: The raw field value from environment or input.
260 Returns:
261 The original value, or None if the value was an empty string.
262 """
263 return _empty_string_to_none(value)
265 model_config = SettingsConfigDict(env_prefix="PLUGINS_", env_file=".env", env_file_encoding="utf-8", extra="ignore")
268class PluginsGrpcClientMtlsSettings(BaseSettings):
269 """Lightweight settings for gRPC client mTLS configuration."""
271 grpc_client_mtls_certfile: str | None = None
272 grpc_client_mtls_keyfile: str | None = None
273 grpc_client_mtls_ca_bundle: str | None = None
274 grpc_client_mtls_keyfile_password: SecretStr | None = None
275 grpc_client_mtls_verify: bool | None = None
277 @field_validator("grpc_client_mtls_verify", mode="before")
278 @classmethod
279 def empty_string_to_none(cls, value: Any) -> Any:
280 """Delegate to shared validator.
282 Args:
283 value: The raw field value from environment or input.
285 Returns:
286 The original value, or None if the value was an empty string.
287 """
288 return _empty_string_to_none(value)
290 model_config = SettingsConfigDict(env_prefix="PLUGINS_", env_file=".env", env_file_encoding="utf-8", extra="ignore")
293class PluginsHttpClientSettings(BaseSettings):
294 """Lightweight settings for HTTP client (httpx) configuration."""
296 skip_ssl_verify: bool = False
297 httpx_max_connections: int = 200
298 httpx_max_keepalive_connections: int = 100
299 httpx_keepalive_expiry: float = 30.0
300 httpx_connect_timeout: float = 5.0
301 httpx_read_timeout: float = 120.0
302 httpx_write_timeout: float = 30.0
303 httpx_pool_timeout: float = 10.0
304 model_config = SettingsConfigDict(env_prefix="PLUGINS_", env_file=".env", env_file_encoding="utf-8", extra="ignore")
307class PluginsCliSettings(BaseSettings):
308 """Lightweight settings for mcpplugins CLI configuration."""
310 cli_completion: bool = False
311 cli_markup_mode: Literal["markdown", "rich", "disabled"] | None = None
312 model_config = SettingsConfigDict(env_prefix="PLUGINS_", env_file=".env", env_file_encoding="utf-8", extra="ignore")
315class PluginsGrpcServerSettings(BaseSettings):
316 """Lightweight settings for gRPC server configuration."""
318 grpc_server_ssl_keyfile: str | None = None
319 grpc_server_ssl_certfile: str | None = None
320 grpc_server_ssl_ca_certs: str | None = None
321 grpc_server_ssl_keyfile_password: SecretStr | None = None
322 grpc_server_ssl_client_auth: str | None = None
323 grpc_server_host: str | None = None
324 grpc_server_port: int | None = None
325 grpc_server_uds: str | None = None
326 grpc_server_ssl_enabled: bool | None = None
328 @field_validator("grpc_server_port", "grpc_server_ssl_enabled", mode="before")
329 @classmethod
330 def empty_string_to_none(cls, value: Any) -> Any:
331 """Delegate to shared validator.
333 Args:
334 value: The raw field value from environment or input.
336 Returns:
337 The original value, or None if the value was an empty string.
338 """
339 return _empty_string_to_none(value)
341 model_config = SettingsConfigDict(env_prefix="PLUGINS_", env_file=".env", env_file_encoding="utf-8", extra="ignore")
344@lru_cache(maxsize=1)
345def get_settings() -> PluginsSettings:
346 """Get cached plugins settings instance.
348 Returns:
349 PluginsSettings: A cached instance of the PluginsSettings class.
351 Examples:
352 >>> settings = get_settings()
353 >>> isinstance(settings, PluginsSettings)
354 True
355 >>> # Second call returns the same cached instance
356 >>> settings2 = get_settings()
357 >>> settings is settings2
358 True
359 """
360 # Instantiate a fresh Pydantic PluginsSettings object,
361 # loading from env vars or .env exactly once.
362 return PluginsSettings()
365@lru_cache()
366def get_enabled_settings() -> PluginsEnabledSettings:
367 """Get cached lightweight enabled flag settings instance.
369 Returns:
370 PluginsEnabledSettings: A cached instance.
371 """
372 return PluginsEnabledSettings()
375@lru_cache()
376def get_startup_settings() -> PluginsStartupSettings:
377 """Get cached lightweight startup settings (config_file, plugin_timeout).
379 Returns:
380 PluginsStartupSettings: A cached instance.
381 """
382 return PluginsStartupSettings()
385@lru_cache()
386def get_config_path_settings() -> PluginsConfigPathSettings:
387 """Get cached lightweight config-path settings instance.
389 Returns:
390 PluginsConfigPathSettings: A cached instance.
391 """
392 return PluginsConfigPathSettings()
395@lru_cache()
396def get_policy_settings() -> PluginsPolicySettings:
397 """Get cached lightweight policy settings instance.
399 Returns:
400 PluginsPolicySettings: A cached instance.
401 """
402 return PluginsPolicySettings()
405@lru_cache()
406def get_ssrf_settings() -> PluginsSsrfSettings:
407 """Get cached lightweight SSRF protection settings instance.
409 Returns:
410 PluginsSsrfSettings: A cached instance.
411 """
412 return PluginsSsrfSettings()
415@lru_cache()
416def get_transport_settings() -> PluginsTransportSettings:
417 """Get cached lightweight transport settings instance.
419 Returns:
420 PluginsTransportSettings: A cached instance.
421 """
422 return PluginsTransportSettings()
425@lru_cache()
426def get_client_mtls_settings() -> PluginsClientMtlsSettings:
427 """Get cached lightweight MCP client mTLS settings instance.
429 Returns:
430 PluginsClientMtlsSettings: A cached instance.
431 """
432 return PluginsClientMtlsSettings()
435@lru_cache()
436def get_mcp_server_settings() -> PluginsMcpServerSettings:
437 """Get cached lightweight MCP server settings instance.
439 Returns:
440 PluginsMcpServerSettings: A cached instance.
441 """
442 return PluginsMcpServerSettings()
445@lru_cache()
446def get_grpc_client_mtls_settings() -> PluginsGrpcClientMtlsSettings:
447 """Get cached lightweight gRPC client mTLS settings instance.
449 Returns:
450 PluginsGrpcClientMtlsSettings: A cached instance.
451 """
452 return PluginsGrpcClientMtlsSettings()
455@lru_cache()
456def get_http_client_settings() -> PluginsHttpClientSettings:
457 """Get cached lightweight HTTP client settings instance.
459 Returns:
460 PluginsHttpClientSettings: A cached instance.
461 """
462 return PluginsHttpClientSettings()
465@lru_cache()
466def get_cli_settings() -> PluginsCliSettings:
467 """Get cached lightweight CLI settings instance.
469 Returns:
470 PluginsCliSettings: A cached instance.
471 """
472 return PluginsCliSettings()
475@lru_cache()
476def get_grpc_server_settings() -> PluginsGrpcServerSettings:
477 """Get cached lightweight gRPC server settings instance.
479 Returns:
480 PluginsGrpcServerSettings: A cached instance.
481 """
482 return PluginsGrpcServerSettings()
485class LazySettingsWrapper:
486 """Lazily initialize plugins settings singleton on getattr."""
488 @staticmethod
489 def _parse_bool(value: str) -> bool:
490 """Parse common truthy string values.
492 Args:
493 value: The string value to parse.
495 Returns:
496 True if the value represents a truthy string.
497 """
498 return value.strip().lower() in {"1", "true", "yes", "on"}
500 @property
501 def enabled(self) -> bool:
502 """Access plugin enabled flag with env override support.
504 Returns:
505 True if plugin framework is enabled.
506 """
507 env_flag = os.getenv("PLUGINS_ENABLED")
508 if env_flag is not None:
509 return self._parse_bool(env_flag)
510 return get_enabled_settings().enabled
512 @property
513 def config_file(self) -> str:
514 """Access config_file without validating full plugin settings.
516 Returns:
517 The plugin configuration file path.
518 """
519 return get_startup_settings().config_file
521 @property
522 def plugin_timeout(self) -> int:
523 """Access plugin_timeout without validating full plugin settings.
525 Returns:
526 The plugin execution timeout in seconds.
527 """
528 return get_startup_settings().plugin_timeout
530 @property
531 def config_path(self) -> str | None:
532 """Access PLUGINS_CONFIG_PATH without validating full plugin settings.
534 Returns:
535 The config path or None if unset.
536 """
537 return get_config_path_settings().config_path
539 @property
540 def default_hook_policy(self) -> Literal["allow", "deny"]:
541 """Access default hook policy without validating full plugin settings.
543 Returns:
544 The default hook policy string.
545 """
546 return get_policy_settings().default_hook_policy
548 @property
549 def ssrf_protection_enabled(self) -> bool:
550 """Access SSRF protection flag without validating full plugin settings.
552 Returns:
553 True if SSRF protection is enabled.
554 """
555 return get_ssrf_settings().ssrf_protection_enabled
557 @property
558 def transport(self) -> str | None:
559 """Access transport type without validating full plugin settings.
561 Returns:
562 The transport type or None if unset.
563 """
564 return get_transport_settings().transport
566 @property
567 def unix_socket_path(self) -> str | None:
568 """Access Unix socket path without validating full plugin settings.
570 Returns:
571 The Unix socket path or None if unset.
572 """
573 return get_transport_settings().unix_socket_path
575 @property
576 def cli_completion(self) -> bool:
577 """Access CLI completion flag without validating full plugin settings.
579 Returns:
580 True if CLI completion is enabled.
581 """
582 return get_cli_settings().cli_completion
584 @property
585 def cli_markup_mode(self) -> Literal["markdown", "rich", "disabled"] | None:
586 """Access CLI markup mode without validating full plugin settings.
588 Returns:
589 The CLI markup mode or None if unset.
590 """
591 return get_cli_settings().cli_markup_mode
593 @staticmethod
594 def cache_clear() -> None:
595 """Clear the cached settings instance so the next access re-reads from env."""
596 get_settings.cache_clear()
597 get_enabled_settings.cache_clear()
598 get_startup_settings.cache_clear()
599 get_config_path_settings.cache_clear()
600 get_policy_settings.cache_clear()
601 get_ssrf_settings.cache_clear()
602 get_transport_settings.cache_clear()
603 get_client_mtls_settings.cache_clear()
604 get_mcp_server_settings.cache_clear()
605 get_http_client_settings.cache_clear()
606 get_cli_settings.cache_clear()
607 get_grpc_client_mtls_settings.cache_clear()
608 get_grpc_server_settings.cache_clear()
610 def __getattr__(self, key: str) -> Any:
611 """Get the real settings object and forward to it
613 Args:
614 key: The key to fetch from settings
616 Returns:
617 Any: The value of the attribute on the settings
618 """
620 return getattr(get_settings(), key)
623settings = LazySettingsWrapper()