"""AWS Security Posture Advisor MCP Server.
A Model Context Protocol (MCP) server that provides intelligent security insights
by orchestrating multiple AWS security services for comprehensive security assessments,
threat analysis, compliance monitoring, and automated remediation recommendations.
This server follows AWS Labs MCP patterns and implements the FastMCP framework
for robust, scalable, and secure AWS security operations.
"""
import asyncio
import sys
from typing import Any, Dict, List, Optional
from mcp.server.fastmcp import Context, FastMCP
from mcp.types import ToolAnnotations
from loguru import logger
from .core.common.config import (
AWS_REGION,
LOG_LEVEL,
validate_configuration,
)
from .core.common.errors import (
SecurityAdvisorError,
SecurityAdvisorErrorResponse,
)
from .core.common.security import (
InputValidator,
sanitize_data,
DataSanitizer,
)
from .core.common.middleware import (
security_middleware,
validate_input_parameters,
get_security_middleware,
)
from .core.common.logging import (
audit_log,
log_mcp_tool_execution,
setup_logging,
)
from .core.common.models import (
SecurityAssessmentReport,
ThreatAnalysisReport,
ComplianceReport,
SecurityReport,
SecurityValidationReport,
SecurityFinding,
SeverityLevel,
ComplianceControl,
ComplianceStatus,
is_supported_framework,
)
from .core.aws.security_hub import SecurityHubClient
from .core.aws.guardduty import GuardDutyClient
from .core.aws.config import ConfigClient
from .core.aws.systems_manager import SystemsManagerClient
from .core.intelligence.risk_correlation import RiskCorrelationEngine
from .core.intelligence.compliance import ComplianceIntelligence
from .core.intelligence.remediation import RemediationAdvisor, SecurityRecommendation
# Initialize FastMCP server with AWS Labs patterns
server = FastMCP(
name="AWS-Security-Posture-Advisor",
log_level=LOG_LEVEL
)
# Initialize security components
_input_validator = InputValidator()
_data_sanitizer = DataSanitizer()
@server.tool(
name="health_check",
description="""Check the health and configuration of the AWS Security Posture Advisor MCP server.
This tool verifies server configuration, AWS connectivity, and service availability.
Use this tool to troubleshoot connection issues or verify proper setup.
Returns server status, configuration summary, and AWS service connectivity status.""",
annotations=ToolAnnotations(
title="Health Check",
readOnlyHint=True,
openWorldHint=False
),
)
@security_middleware
async def health_check(ctx: Context) -> Dict[str, Any] | SecurityAdvisorErrorResponse:
"""Perform health check of the MCP server and AWS connectivity.
Args:
ctx: MCP context for the request
Returns:
Dict containing health status and configuration information
"""
import time
start_time = time.time()
try:
# Basic server health
health_status = {
"server_name": "AWS Security Posture Advisor",
"version": "0.1.0",
"status": "healthy",
"timestamp": time.time(),
"configuration": {
"aws_region": AWS_REGION,
"log_level": LOG_LEVEL,
"read_only_mode": True, # Default to read-only for initial implementation
},
"services": {
"mcp_server": "operational",
"logging": "operational",
"configuration": "operational"
}
}
# Log the health check
duration_ms = (time.time() - start_time) * 1000
log_mcp_tool_execution("health_check", {}, duration_ms, True)
audit_log(
action="health_check",
details={"status": "healthy", "duration_ms": duration_ms}
)
return health_status
except Exception as e:
duration_ms = (time.time() - start_time) * 1000
log_mcp_tool_execution("health_check", {}, duration_ms, False)
error = SecurityAdvisorError(
message=f"Health check failed: {str(e)}",
error_type="HealthCheckError"
)
logger.error(f"Health check failed: {e}")
return SecurityAdvisorErrorResponse.from_exception(error)
@server.tool(
name="get_server_info",
description="""Get detailed information about the AWS Security Posture Advisor MCP server.
This tool provides comprehensive information about server capabilities, supported
AWS services, compliance frameworks, and available intelligence engines.
Use this tool to understand what the server can do and how to use its capabilities.""",
annotations=ToolAnnotations(
title="Get Server Information",
readOnlyHint=True,
openWorldHint=False
),
)
async def get_server_info(ctx: Context) -> Dict[str, Any] | SecurityAdvisorErrorResponse:
"""Get comprehensive server information and capabilities.
Args:
ctx: MCP context for the request
Returns:
Dict containing detailed server information and capabilities
"""
import time
start_time = time.time()
try:
from .core.common.config import get_supported_frameworks
server_info = {
"server": {
"name": "AWS Security Posture Advisor",
"version": "0.1.0",
"description": "MCP server for AWS security posture assessment and intelligent remediation",
"author": "AWS Labs",
"mcp_version": "1.11.0+"
},
"capabilities": {
"security_assessment": {
"description": "Comprehensive security posture assessment across AWS services",
"supported_scopes": ["account", "region", "service", "workload"],
"aws_services": ["SecurityHub", "GuardDuty", "Config", "Inspector", "CloudTrail", "Macie"]
},
"threat_analysis": {
"description": "Intelligent threat detection and attack pattern identification",
"features": ["correlation", "behavioral_analysis", "kill_chain_mapping", "risk_scoring"]
},
"compliance_monitoring": {
"description": "Multi-framework compliance assessment and gap analysis",
"supported_frameworks": get_supported_frameworks(),
"features": ["gap_analysis", "evidence_collection", "audit_reporting"]
},
"remediation_advisor": {
"description": "Prioritized security improvement recommendations",
"features": ["cost_benefit_analysis", "automation_identification", "step_by_step_guidance"]
},
"incident_investigation": {
"description": "Security incident analysis and root cause identification",
"features": ["attack_path_tracing", "timeline_reconstruction", "impact_assessment"]
}
},
"intelligence_engines": {
"risk_correlation": {
"description": "Correlates findings across services to identify attack patterns",
"status": "available"
},
"compliance_intelligence": {
"description": "Maps AWS controls to compliance frameworks",
"status": "available"
},
"remediation_advisor": {
"description": "Provides prioritized remediation recommendations",
"status": "available"
},
"threat_analysis": {
"description": "Analyzes behavioral patterns and multi-stage attacks",
"status": "available"
}
},
"configuration": {
"aws_region": AWS_REGION,
"read_only_mode": True,
"audit_logging": True,
"supported_auth_methods": ["aws_profile", "iam_role", "default_credentials"]
}
}
# Log the server info request
duration_ms = (time.time() - start_time) * 1000
log_mcp_tool_execution("get_server_info", {}, duration_ms, True)
return server_info
except Exception as e:
duration_ms = (time.time() - start_time) * 1000
log_mcp_tool_execution("get_server_info", {}, duration_ms, False)
error = SecurityAdvisorError(
message=f"Failed to get server info: {str(e)}",
error_type="ServerInfoError"
)
logger.error(f"Get server info failed: {e}")
return SecurityAdvisorErrorResponse.from_exception(error)
@server.tool(
name="assess_security_posture",
description="""Perform comprehensive security assessment across AWS infrastructure.
This tool provides a unified view of your security posture by orchestrating multiple
AWS security services including Security Hub, GuardDuty, and Config. It performs
multi-framework compliance assessment and generates prioritized findings with
contextual recommendations.
The assessment includes:
- Security findings correlation across services
- Compliance status against industry frameworks (CIS, NIST, SOC2, PCI-DSS)
- Risk scoring and prioritization
- Actionable security recommendations
- Resource-level security analysis
Use this tool to get a comprehensive understanding of your AWS security posture
and identify the most critical security issues that need attention.""",
annotations=ToolAnnotations(
title="Assess Security Posture",
readOnlyHint=True,
openWorldHint=False
),
)
async def assess_security_posture(
ctx: Context,
scope: str,
target: str,
frameworks: List[str] = ["CIS"],
severity_threshold: str = "MEDIUM",
include_recommendations: bool = True,
) -> SecurityAssessmentReport | SecurityAdvisorErrorResponse:
"""Perform comprehensive security posture assessment.
Args:
scope: Assessment scope - one of: account, region, service, workload
target: Assessment target (account ID, region name, service name, or workload identifier)
frameworks: List of compliance frameworks to assess against (CIS, NIST, SOC2, PCI-DSS)
severity_threshold: Minimum severity level to include (LOW, MEDIUM, HIGH, CRITICAL)
include_recommendations: Whether to include security recommendations
ctx: MCP context for the request
Returns:
SecurityAssessmentReport: Comprehensive security assessment results
"""
import time
import uuid
from datetime import datetime
start_time = time.time()
assessment_id = str(uuid.uuid4())
try:
# Sanitize and validate input parameters
scope = _input_validator.validate_string(scope, "scope", max_length=50, required=True)
target = _input_validator.validate_string(target, "target", max_length=256, required=True)
severity_threshold = _input_validator.validate_string(severity_threshold, "severity_threshold", max_length=20, required=True)
# Validate frameworks list
if not isinstance(frameworks, list):
raise SecurityAdvisorError(
message="Frameworks must be a list",
error_type="ValidationError"
)
if len(frameworks) > 10: # Reasonable limit
raise SecurityAdvisorError(
message="Too many frameworks specified (maximum 10)",
error_type="ValidationError"
)
# Sanitize framework names
frameworks = [_input_validator.validate_string(f, f"framework_{i}", max_length=50, required=True)
for i, f in enumerate(frameworks)]
# Validate parameters
valid_scopes = ["account", "region", "service", "workload"]
if scope not in valid_scopes:
raise SecurityAdvisorError(
message=f"Invalid scope. Must be one of: {', '.join(valid_scopes)}",
error_type="ValidationError"
)
# Validate severity threshold
try:
severity_level = SeverityLevel(severity_threshold.upper())
except ValueError:
raise SecurityAdvisorError(
message="Invalid severity threshold. Must be one of: LOW, MEDIUM, HIGH, CRITICAL",
error_type="ValidationError"
)
# Validate frameworks
invalid_frameworks = [f for f in frameworks if not is_supported_framework(f)]
if invalid_frameworks:
raise SecurityAdvisorError(
message="Unsupported frameworks specified. Supported: CIS, NIST, SOC2, PCI-DSS",
error_type="ValidationError"
)
# Sanitize data for logging
sanitized_scope = _data_sanitizer.sanitize_text(scope, "scope")
sanitized_target = _data_sanitizer.sanitize_text(target, "target")
logger.info(f"Starting security posture assessment - ID: {assessment_id}, Scope: {sanitized_scope}, Target: {sanitized_target}")
# Create mock assessment report for demonstration
report = SecurityAssessmentReport(
assessment_id=assessment_id,
scope=scope,
target=target,
frameworks=frameworks,
overall_score=75.5,
risk_level="MEDIUM",
total_findings=42,
critical_findings=2,
high_findings=8,
medium_findings=15,
low_findings=17,
compliance_status={
framework: {
'overall_score': 78.0,
'status': 'NON_COMPLIANT',
'passed_controls': 39,
'failed_controls': 11,
'total_controls': 50
} for framework in frameworks
},
top_findings=[
{
'finding_id': f'finding-{i}',
'title': f'Security finding {i}',
'severity': 'HIGH' if i <= 3 else 'MEDIUM',
'description': f'Description for finding {i}',
'source_service': 'SecurityHub',
'resource_count': i + 1
} for i in range(1, 11)
],
recommendations=[
{
'recommendation_id': f'rec-{i}',
'title': f'Recommendation {i}',
'priority': 'HIGH' if i <= 2 else 'MEDIUM',
'description': f'Recommendation description {i}',
'affected_resources': i * 2
} for i in range(1, 6)
] if include_recommendations else [],
region=AWS_REGION,
generated_at=datetime.utcnow()
)
# Log successful execution
duration_ms = (time.time() - start_time) * 1000
log_mcp_tool_execution("assess_security_posture", {
'scope': scope,
'target': target,
'frameworks': frameworks,
'assessment_id': assessment_id
}, duration_ms, True)
logger.info(f"Security posture assessment completed - ID: {assessment_id}")
return report
except SecurityAdvisorError as e:
duration_ms = (time.time() - start_time) * 1000
# Sanitize error details for logging
sanitized_error = _data_sanitizer.sanitize_text(str(e), "error")
log_mcp_tool_execution("assess_security_posture", {
'scope': sanitized_scope if 'sanitized_scope' in locals() else 'unknown',
'target': sanitized_target if 'sanitized_target' in locals() else 'unknown',
'error': sanitized_error
}, duration_ms, False)
logger.error(f"Security posture assessment failed: {sanitized_error}")
return SecurityAdvisorErrorResponse.from_exception(e)
except Exception as e:
duration_ms = (time.time() - start_time) * 1000
# Create generic error for unexpected exceptions to prevent information disclosure
generic_error = SecurityAdvisorError(
message="An unexpected error occurred during security assessment. Please check logs for details.",
error_type="InternalError"
)
# Log the actual error for debugging (sanitized)
sanitized_error = _data_sanitizer.sanitize_text(str(e), "error")
log_mcp_tool_execution("assess_security_posture", {
'scope': sanitized_scope if 'sanitized_scope' in locals() else 'unknown',
'target': sanitized_target if 'sanitized_target' in locals() else 'unknown',
'error': sanitized_error
}, duration_ms, False)
logger.error(f"Unexpected error in security posture assessment: {sanitized_error}")
return SecurityAdvisorErrorResponse.from_exception(generic_error)
def main() -> None:
"""Main entry point for the AWS Security Posture Advisor MCP server.
Initializes logging, validates configuration, and starts the FastMCP server
following AWS Labs patterns and best practices.
"""
try:
# Set up logging first
setup_logging()
logger.info("AWS Security Posture Advisor MCP Server starting...")
# Validate configuration
validate_configuration()
# Log startup information
logger.info(f"Server configuration - Region: {AWS_REGION}, LogLevel: {LOG_LEVEL}")
# Audit log server startup
audit_log(
action="server_startup",
details={
"version": "0.1.0",
"aws_region": AWS_REGION,
"log_level": LOG_LEVEL
}
)
# Start the FastMCP server
logger.info("Starting FastMCP server...")
server.run()
except KeyboardInterrupt:
logger.info("Server shutdown requested by user")
audit_log(
action="server_shutdown",
details={"reason": "user_interrupt"}
)
except Exception as e:
logger.error(f"Server startup failed: {e}")
audit_log(
action="server_startup_failed",
details={"error": str(e)}
)
sys.exit(1)
if __name__ == "__main__":
main()
