feat: Add ECS Security Analysis Tool - Core Infrastructure (PR #1)

Add foundational security analysis tool with cluster listing functionality.
This PR establishes the core infrastructure and module registration.

Features:
- Region validation and selection
- ECS cluster listing with metadata
- Interactive workflow for cluster selection
- Comprehensive error handling
- Module registration in main.py

Files Added:
- awslabs/ecs_mcp_server/api/security_analysis.py (223 lines)
- awslabs/ecs_mcp_server/modules/security_analysis.py (155 lines)
- tests/unit/test_security_analysis_api.py (168 lines)
- tests/unit/test_security_analysis_module.py (147 lines)

Files Modified:
- awslabs/ecs_mcp_server/main.py (module registration)

Testing:
- 14 unit tests with 91% coverage (with branch coverage)
- All tests pass
- Follows all coding patterns from existing codebase

Usage:
  analyze_ecs_security()  # List clusters in default region
  analyze_ecs_security(region='us-west-2')  # List in specific region

Total: 693 lines (378 production + 315 tests)

Author: Suresh Natarajan

src/ecs-mcp-server/awslabs/ecs_mcp_server/api/security_analysis.py

@@ -0,0 +1,223 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+API for ECS security analysis operations.
+
+This module provides functions for analyzing ECS cluster security configurations
+and generating security recommendations.
+"""
+
+import logging
+import os
+from typing import Any, Dict
+
+import boto3
+
+from awslabs.ecs_mcp_server.api.resource_management import ecs_api_operation
+
+logger = logging.getLogger(__name__)
+
+
+class RegionValidationError(Exception):
+    """Raised when an invalid AWS region is provided."""
+
+    pass
+
+
+def validate_region(region: str) -> None:
+    """
+    Validate that the provided region is a valid AWS region for ECS.
+
+    Args:
+        region: AWS region name to validate
+
+    Raises:
+        RegionValidationError: If the region is not valid for ECS
+    """
+    try:
+        # Get list of available regions for ECS service
+        available_regions = boto3.Session().get_available_regions("ecs")
+
+        if region not in available_regions:
+            regions_list = ", ".join(sorted(available_regions))
+            raise RegionValidationError(
+                f"Invalid AWS region '{region}'. Must be one of: {regions_list}"
+            )
+
+        logger.info(f"Region '{region}' validated successfully")
+    except Exception as e:
+        if isinstance(e, RegionValidationError):
+            raise
+        logger.error(f"Error validating region '{region}': {e}")
+        raise RegionValidationError(f"Failed to validate region '{region}': {str(e)}") from e
+
+
+def get_target_region(region: str | None = None) -> str:
+    """
+    Get the target AWS region for security analysis.
+
+    If region is provided, validates it and returns it.
+    If region is None, uses AWS_REGION environment variable (defaults to 'us-east-1').
+
+    Args:
+        region: Optional AWS region name
+
+    Returns:
+        Validated AWS region name
+
+    Raises:
+        RegionValidationError: If the region is invalid
+    """
+    if region is None:
+        # Get region from environment variable, default to us-east-1
+        region = os.environ.get("AWS_REGION", "us-east-1")
+        logger.info(f"No region specified, using region from environment: {region}")
+    else:
+        logger.info(f"Using specified region: {region}")
+
+    # Validate the region
+    validate_region(region)
+
+    return region
+
+
+async def list_clusters_in_region(region: str) -> list[Dict[str, Any]]:
+    """
+    List all ECS clusters in the specified region with their metadata.
+
+    Args:
+        region: AWS region to list clusters from
+
+    Returns:
+        List of cluster dictionaries with metadata
+
+    Raises:
+        Exception: If listing clusters fails
+    """
+    logger.info(f"Listing ECS clusters in region: {region}")
+
+    try:
+        # List cluster ARNs
+        list_response = await ecs_api_operation(api_operation="ListClusters", api_params={})
+
+        cluster_arns = list_response.get("clusterArns", [])
+
+        if not cluster_arns:
+            logger.info(f"No clusters found in region {region}")
+            return []
+
+        logger.info(f"Found {len(cluster_arns)} cluster(s) in region {region}")
+
+        # Describe clusters to get metadata
+        describe_response = await ecs_api_operation(
+            api_operation="DescribeClusters",
+            api_params={
+                "clusters": cluster_arns,
+                "include": ["ATTACHMENTS", "SETTINGS", "STATISTICS", "TAGS"],
+            },
+        )
+
+        clusters = describe_response.get("clusters", [])
+
+        # Format cluster information
+        all_clusters = []
+        for cluster in clusters:
+            cluster_info = {
+                "cluster_name": cluster.get("clusterName"),
+                "cluster_arn": cluster.get("clusterArn"),
+                "status": cluster.get("status"),
+                "running_tasks_count": cluster.get("runningTasksCount", 0),
+                "pending_tasks_count": cluster.get("pendingTasksCount", 0),
+                "active_services_count": cluster.get("activeServicesCount", 0),
+                "registered_container_instances_count": cluster.get(
+                    "registeredContainerInstancesCount", 0
+                ),
+                "tags": {tag["key"]: tag["value"] for tag in cluster.get("tags", [])},
+            }
+            all_clusters.append(cluster_info)
+
+        logger.info(f"Successfully retrieved metadata for {len(all_clusters)} cluster(s)")
+        return all_clusters
+
+    except Exception as e:
+        logger.error(f"Error listing clusters in region {region}: {e}")
+        raise Exception(f"Failed to list clusters in region '{region}': {str(e)}") from e
+
+
+def format_cluster_list(clusters: list[Dict[str, Any]], region: str) -> str:
+    """
+    Format a list of clusters for user selection.
+
+    Args:
+        clusters: List of cluster dictionaries
+        region: AWS region name
+
+    Returns:
+        Formatted string with cluster information
+    """
+    if not clusters:
+        return f"""
+No ECS clusters found in region: {region}
+
+To create a cluster, you can use the AWS CLI:
+```bash
+aws ecs create-cluster --cluster-name my-cluster --region {region}
+```
+
+Or use the AWS Console to create a cluster in the ECS service.
+"""
+
+    # Build the formatted output
+    lines = [
+        "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━",
+        f"📋 ECS CLUSTERS IN REGION: {region}",
+        "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━",
+        "",
+        f"Found {len(clusters)} cluster(s):",
+        "",
+    ]
+
+    for i, cluster in enumerate(clusters, 1):
+        cluster_name = cluster.get("cluster_name", "Unknown")
+        status = cluster.get("status", "Unknown")
+        running_tasks = cluster.get("running_tasks_count", 0)
+        active_services = cluster.get("active_services_count", 0)
+
+        lines.extend(
+            [
+                f"{i}. {cluster_name}",
+                f"   Status: {status}",
+                f"   Running Tasks: {running_tasks}",
+                f"   Active Services: {active_services}",
+                "",
+            ]
+        )
+
+    lines.extend(
+        [
+            "━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━",
+            "",
+            "To analyze a specific cluster, call this tool again with:",
+            "  cluster_names=['cluster-name']",
+            "",
+            "Example:",
+            f"  analyze_ecs_security("
+            f"cluster_names=['{clusters[0].get('cluster_name')}'], "
+            f"region='{region}')",
+            "",
+        ]
+    )
+
+    return "\n".join(lines)

src/ecs-mcp-server/awslabs/ecs_mcp_server/main.py

@@ -31,6 +31,7 @@
     deployment_status,
     infrastructure,
     resource_management,
+    security_analysis,
     troubleshooting,
 )
 from awslabs.ecs_mcp_server.utils.config import get_config
@@ -134,6 +135,7 @@ def _create_ecs_mcp_server() -> Tuple[FastMCP, Dict[str, Any]]:
     resource_management.register_module(mcp)
     troubleshooting.register_module(mcp)
     delete.register_module(mcp)
+    security_analysis.register_module(mcp)
 
     # Register all proxies
     aws_knowledge_proxy.register_proxy(mcp)

src/ecs-mcp-server/awslabs/ecs_mcp_server/modules/security_analysis.py

@@ -0,0 +1,155 @@
+# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+Security Analysis module for ECS MCP Server.
+This module provides tools and prompts for analyzing ECS security configurations.
+"""
+
+import logging
+
+from fastmcp import FastMCP
+from pydantic import Field
+
+from awslabs.ecs_mcp_server.api.security_analysis import (
+    format_cluster_list,
+    get_target_region,
+    list_clusters_in_region,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def register_module(mcp: FastMCP) -> None:
+    """Register security analysis module tools and prompts with the MCP server."""
+
+    # Define pydantic Field descriptions for all parameters
+    region_field = Field(
+        default=None,
+        description=(
+            "Optional AWS region to analyze. "
+            "If not provided, uses the AWS_REGION environment variable (defaults to 'us-east-1'). "
+            "Must be a valid AWS region where ECS is available. "
+            "Example: 'us-west-2', 'eu-west-1'"
+        ),
+    )
+
+    @mcp.tool(name="analyze_ecs_security", annotations=None)
+    async def mcp_analyze_ecs_security(
+        region: str | None = region_field,
+    ) -> str:
+        """
+        List ECS clusters available for security analysis.
+
+        This tool lists all ECS clusters in the specified AWS region, providing
+        an overview of available clusters that can be analyzed for security issues.
+
+        Interactive Workflow:
+
+        Step 1: List Available Clusters
+           - Call with NO parameters to list clusters in default region
+           - Call with region parameter to list clusters in specific region
+           - Returns formatted list of available clusters with metadata
+
+        Step 2: User Selects Clusters (Future)
+           - User reviews the list and selects which clusters to analyze
+           - Future versions will support cluster_names parameter for analysis
+
+        Usage Examples:
+
+        Example 1 - List clusters in default region:
+            analyze_ecs_security()
+            # Returns list of clusters in AWS_REGION for selection
+
+        Example 2 - List clusters in specific region:
+            analyze_ecs_security(region="us-west-2")
+            # Returns list of clusters in us-west-2 for selection
+
+        Cluster Information Provided:
+        - Cluster name and ARN
+        - Current status (ACTIVE, INACTIVE, etc.)
+        - Running tasks count
+        - Active services count
+        - Registered container instances count
+        - Resource tags
+
+        Parameters:
+            region: Optional AWS region. If None, uses AWS_REGION environment variable.
+
+        Returns:
+            Formatted list of available clusters for selection
+
+        Error Handling:
+            - Invalid region: Returns error message with list of valid regions
+            - No clusters found: Returns helpful message with cluster creation guidance
+        """
+        try:
+            # Step 1: Determine target region
+            logger.info("Step 1: Determining target region")
+            target_region = get_target_region(region)
+
+            # Step 2: List clusters for user selection
+            logger.info(f"Step 2: Listing clusters in region '{target_region}' for user selection")
+            clusters = await list_clusters_in_region(target_region)
+            return format_cluster_list(clusters, target_region)
+
+        except Exception as e:
+            import traceback
+
+            error_msg = f"Error during security analysis: {str(e)}"
+            logger.error(error_msg)
+            logger.error(f"Traceback: {traceback.format_exc()}")
+            return f"❌ {error_msg}\n\nDetailed error:\n{traceback.format_exc()}"
+
+    # Register prompt patterns for security analysis
+
+    @mcp.prompt("analyze ecs security")
+    def analyze_ecs_security_prompt():
+        """User wants to analyze ECS security"""
+        return ["analyze_ecs_security"]
+
+    @mcp.prompt("check ecs security")
+    def check_ecs_security_prompt():
+        """User wants to check ECS security"""
+        return ["analyze_ecs_security"]
+
+    @mcp.prompt("ecs security audit")
+    def ecs_security_audit_prompt():
+        """User wants to perform an ECS security audit"""
+        return ["analyze_ecs_security"]
+
+    @mcp.prompt("security best practices")
+    def security_best_practices_prompt():
+        """User wants to check security best practices"""
+        return ["analyze_ecs_security"]
+
+    @mcp.prompt("security recommendations")
+    def security_recommendations_prompt():
+        """User wants security recommendations"""
+        return ["analyze_ecs_security"]
+
+    @mcp.prompt("scan ecs clusters")
+    def scan_ecs_clusters_prompt():
+        """User wants to scan ECS clusters for security issues"""
+        return ["analyze_ecs_security"]
+
+    @mcp.prompt("ecs security scan")
+    def ecs_security_scan_prompt():
+        """User wants to perform an ECS security scan"""
+        return ["analyze_ecs_security"]
+
+    @mcp.prompt("list ecs clusters")
+    def list_ecs_clusters_prompt():
+        """User wants to list ECS clusters"""
+        return ["analyze_ecs_security"]