MaverickMCP

""" Comprehensive Security Configuration for Maverick MCP. This module provides centralized security configuration including CORS settings, security headers, rate limiting, and environment-specific security policies. All security settings are validated to prevent common misconfigurations. """ import os from pydantic import BaseModel, Field, model_validator class CORSConfig(BaseModel): """CORS (Cross-Origin Resource Sharing) configuration with validation.""" # Origins configuration allowed_origins: list[str] = Field( default_factory=lambda: _get_cors_origins(), description="List of allowed origins for CORS requests", ) # Credentials and methods allow_credentials: bool = Field( default=True, description="Whether to allow credentials in CORS requests" ) allowed_methods: list[str] = Field( default=["GET", "POST", "PUT", "DELETE", "OPTIONS", "PATCH"], description="Allowed HTTP methods for CORS requests", ) # Headers configuration allowed_headers: list[str] = Field( default=[ "Authorization", "Content-Type", "X-API-Key", "X-Request-ID", "X-Requested-With", "Accept", "Origin", "User-Agent", "Cache-Control", ], description="Allowed headers for CORS requests", ) exposed_headers: list[str] = Field( default=[ "X-Process-Time", "X-RateLimit-Limit", "X-RateLimit-Remaining", "X-RateLimit-Reset", "X-Request-ID", ], description="Headers exposed to the client", ) # Cache and validation max_age: int = Field( default=86400, # 24 hours description="Maximum age for CORS preflight cache in seconds", ) @model_validator(mode="after") def validate_cors_security(self): """Validate CORS configuration for security.""" # Critical: Wildcard origins with credentials is dangerous if self.allow_credentials and "*" in self.allowed_origins: raise ValueError( "CORS Security Error: Cannot use wildcard origin ('*') with " "allow_credentials=True. This is a serious security vulnerability. " "Specify explicit origins instead." ) # Warning for wildcard origins without credentials if "*" in self.allowed_origins and not self.allow_credentials: # This is allowed but should be logged import logging logger = logging.getLogger(__name__) logger.warning( "CORS Warning: Using wildcard origin ('*') without credentials. " "Consider using specific origins for better security." ) return self class SecurityHeadersConfig(BaseModel): """Security headers configuration.""" # Content security x_content_type_options: str = Field( default="nosniff", description="X-Content-Type-Options header value" ) x_frame_options: str = Field( default="DENY", description="X-Frame-Options header value (DENY, SAMEORIGIN, or ALLOW-FROM)", ) x_xss_protection: str = Field( default="1; mode=block", description="X-XSS-Protection header value" ) referrer_policy: str = Field( default="strict-origin-when-cross-origin", description="Referrer-Policy header value", ) permissions_policy: str = Field( default="geolocation=(), microphone=(), camera=(), usb=(), magnetometer=()", description="Permissions-Policy header value", ) # HSTS (HTTP Strict Transport Security) hsts_max_age: int = Field( default=31536000, # 1 year description="HSTS max-age in seconds", ) hsts_include_subdomains: bool = Field( default=True, description="Include subdomains in HSTS policy" ) hsts_preload: bool = Field( default=False, description="Enable HSTS preload (requires manual submission to browser vendors)", ) # Content Security Policy csp_default_src: list[str] = Field( default=["'self'"], description="CSP default-src directive" ) csp_script_src: list[str] = Field( default=["'self'", "'unsafe-inline'"], description="CSP script-src directive", ) csp_style_src: list[str] = Field( default=["'self'", "'unsafe-inline'"], description="CSP style-src directive" ) csp_img_src: list[str] = Field( default=["'self'", "data:", "https:"], description="CSP img-src directive" ) csp_connect_src: list[str] = Field( default=["'self'"], description="CSP connect-src directive", ) csp_frame_src: list[str] = Field( default=["'none'"], description="CSP frame-src directive" ) csp_object_src: list[str] = Field( default=["'none'"], description="CSP object-src directive" ) @property def hsts_header_value(self) -> str: """Generate HSTS header value.""" value = f"max-age={self.hsts_max_age}" if self.hsts_include_subdomains: value += "; includeSubDomains" if self.hsts_preload: value += "; preload" return value @property def csp_header_value(self) -> str: """Generate Content-Security-Policy header value.""" directives = [ f"default-src {' '.join(self.csp_default_src)}", f"script-src {' '.join(self.csp_script_src)}", f"style-src {' '.join(self.csp_style_src)}", f"img-src {' '.join(self.csp_img_src)}", f"connect-src {' '.join(self.csp_connect_src)}", f"frame-src {' '.join(self.csp_frame_src)}", f"object-src {' '.join(self.csp_object_src)}", "base-uri 'self'", "form-action 'self'", ] return "; ".join(directives) class RateLimitConfig(BaseModel): """Rate limiting configuration.""" # Basic rate limits default_rate_limit: str = Field( default="1000 per hour", description="Default rate limit for all endpoints" ) # User-specific limits authenticated_limit_per_minute: int = Field( default=60, description="Rate limit for authenticated users per minute" ) anonymous_limit_per_minute: int = Field( default=10, description="Rate limit for anonymous users per minute" ) # Endpoint-specific limits auth_endpoints_limit: str = Field( default="10 per hour", description="Rate limit for authentication endpoints (login, signup)", ) api_endpoints_limit: str = Field( default="60 per minute", description="Rate limit for API endpoints" ) sensitive_endpoints_limit: str = Field( default="5 per minute", description="Rate limit for sensitive operations" ) webhook_endpoints_limit: str = Field( default="100 per minute", description="Rate limit for webhook endpoints" ) # Redis configuration for rate limiting redis_url: str | None = Field( default_factory=lambda: os.getenv("AUTH_REDIS_URL", "redis://localhost:6379/1"), description="Redis URL for rate limiting storage", ) enabled: bool = Field( default_factory=lambda: os.getenv("RATE_LIMITING_ENABLED", "true").lower() == "true", description="Enable rate limiting", ) class TrustedHostsConfig(BaseModel): """Trusted hosts configuration.""" allowed_hosts: list[str] = Field( default_factory=lambda: _get_trusted_hosts(), description="List of trusted host patterns", ) enforce_in_development: bool = Field( default=False, description="Whether to enforce trusted hosts in development" ) class SecurityConfig(BaseModel): """Comprehensive security configuration for Maverick MCP.""" # Environment detection environment: str = Field( default_factory=lambda: os.getenv("ENVIRONMENT", "development").lower(), description="Environment (development, staging, production)", ) # Sub-configurations cors: CORSConfig = Field( default_factory=CORSConfig, description="CORS configuration" ) headers: SecurityHeadersConfig = Field( default_factory=SecurityHeadersConfig, description="Security headers configuration", ) rate_limiting: RateLimitConfig = Field( default_factory=RateLimitConfig, description="Rate limiting configuration" ) trusted_hosts: TrustedHostsConfig = Field( default_factory=TrustedHostsConfig, description="Trusted hosts configuration" ) # General security settings force_https: bool = Field( default_factory=lambda: os.getenv("FORCE_HTTPS", "false").lower() == "true", description="Force HTTPS in production", ) strict_security: bool = Field( default_factory=lambda: os.getenv("STRICT_SECURITY", "false").lower() == "true", description="Enable strict security mode", ) @model_validator(mode="after") def validate_environment_security(self): """Validate security configuration based on environment.""" if self.environment == "production": # Production security requirements if not self.force_https: import logging logger = logging.getLogger(__name__) logger.warning( "Production Warning: FORCE_HTTPS is disabled in production. " "Set FORCE_HTTPS=true for better security." ) # Validate CORS for production if "*" in self.cors.allowed_origins: import logging logger = logging.getLogger(__name__) logger.error( "Production Error: Wildcard CORS origins detected in production. " "This is a security risk and should be fixed." ) return self def get_cors_middleware_config(self) -> dict: """Get CORS middleware configuration dictionary.""" return { "allow_origins": self.cors.allowed_origins, "allow_credentials": self.cors.allow_credentials, "allow_methods": self.cors.allowed_methods, "allow_headers": self.cors.allowed_headers, "expose_headers": self.cors.exposed_headers, "max_age": self.cors.max_age, } def get_security_headers(self) -> dict[str, str]: """Get security headers dictionary.""" headers = { "X-Content-Type-Options": self.headers.x_content_type_options, "X-Frame-Options": self.headers.x_frame_options, "X-XSS-Protection": self.headers.x_xss_protection, "Referrer-Policy": self.headers.referrer_policy, "Permissions-Policy": self.headers.permissions_policy, "Content-Security-Policy": self.headers.csp_header_value, } # Add HSTS only in production or when HTTPS is forced if self.environment == "production" or self.force_https: headers["Strict-Transport-Security"] = self.headers.hsts_header_value return headers def is_production(self) -> bool: """Check if running in production environment.""" return self.environment == "production" def is_development(self) -> bool: """Check if running in development environment.""" return self.environment in ["development", "dev", "local"] def _get_cors_origins() -> list[str]: """Get CORS origins based on environment.""" environment = os.getenv("ENVIRONMENT", "development").lower() cors_origins_env = os.getenv("CORS_ORIGINS") if cors_origins_env: # Parse comma-separated origins from environment return [origin.strip() for origin in cors_origins_env.split(",")] if environment == "production": return [ "https://app.maverick-mcp.com", "https://maverick-mcp.com", "https://www.maverick-mcp.com", ] elif environment in ["staging", "test"]: return [ "https://staging.maverick-mcp.com", "https://test.maverick-mcp.com", "http://localhost:3000", "http://localhost:3001", ] else: # Development return [ "http://localhost:3000", "http://localhost:3001", "http://127.0.0.1:3000", "http://127.0.0.1:3001", "http://localhost:8080", "http://localhost:5173", # Vite default ] def _get_trusted_hosts() -> list[str]: """Get trusted hosts based on environment.""" environment = os.getenv("ENVIRONMENT", "development").lower() trusted_hosts_env = os.getenv("TRUSTED_HOSTS") if trusted_hosts_env: # Parse comma-separated hosts from environment return [host.strip() for host in trusted_hosts_env.split(",")] if environment == "production": return ["api.maverick-mcp.com", "*.maverick-mcp.com", "maverick-mcp.com"] elif environment in ["staging", "test"]: return [ "staging.maverick-mcp.com", "test.maverick-mcp.com", "*.maverick-mcp.com", "localhost", "127.0.0.1", ] else: # Development - allow any host return ["*"] # Create singleton instance security_config = SecurityConfig() def get_security_config() -> SecurityConfig: """Get the security configuration instance.""" return security_config def validate_security_config() -> dict[str, any]: """Validate the current security configuration.""" config = get_security_config() issues = [] warnings = [] # Check for dangerous CORS configuration if config.cors.allow_credentials and "*" in config.cors.allowed_origins: issues.append("CRITICAL: Wildcard CORS origins with credentials enabled") # Check production-specific requirements if config.is_production(): if "*" in config.cors.allowed_origins: issues.append("CRITICAL: Wildcard CORS origins in production") if not config.force_https: warnings.append("HTTPS not enforced in production") if "localhost" in str(config.cors.allowed_origins).lower(): warnings.append("Localhost origins found in production CORS config") # Check for insecure headers if config.headers.x_frame_options not in ["DENY", "SAMEORIGIN"]: warnings.append("X-Frame-Options not set to DENY or SAMEORIGIN") return { "valid": len(issues) == 0, "issues": issues, "warnings": warnings, "environment": config.environment, "cors_origins": config.cors.allowed_origins, "force_https": config.force_https, }