microwind/ai-skills/cloud-config-analyzer

问题:
云资源安全配置不当
 
错误示例:
- 存储桶公开访问
- 安全组规则过于宽松
- 密钥硬编码在配置中
- 未启用加密
 
解决方案:
1. 实施最小权限原则
2. 启用默认加密
3. 使用密钥管理服务
4. 定期安全审计

问题:
云资源配置过度导致成本浪费
 
错误示例:
- 开发环境使用生产级实例
- 未使用自动缩放
- 闲置资源未清理
- 存储数据冗余
 
解决方案:
1. 使用合适的实例类型
2. 配置自动缩放
3. 定期清理闲置资源
4. 实施生命周期管理

问题:
网络配置导致性能或安全问题
 
错误示例:
- 跨区域流量未优化
- 防火墙规则冲突
- DNS配置错误
- 负载均衡配置不当
 
解决方案:
1. 优化网络拓扑
2. 使用CDN加速
3. 配置健康检查
4. 实施网络分段

import json
import yaml
import boto3
from typing import Dict, List, Any, Optional, Tuple
from dataclasses import dataclass
from enum import Enum
import re
from datetime import datetime, timedelta
class CloudProvider(Enum):
    AWS = "aws"
    AZURE = "azure"
    GCP = "gcp"
    ALIBABA = "alibaba"
class ResourceType(Enum):
    EC2 = "ec2"
    S3 = "s3"
    RDS = "rds"
    LAMBDA = "lambda"
    VPC = "vpc"
    SECURITY_GROUP = "security_group"
    IAM = "iam"
class RiskLevel(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"
@dataclass
class CloudResource:
    """云资源"""
    id: str
    type: ResourceType
    provider: CloudProvider
    region: str
    config: Dict[str, Any]
    tags: Dict[str, str]
    created_at: datetime
@dataclass
class SecurityIssue:
    """安全问题"""
    resource_id: str
    issue_type: str
    description: str
    risk_level: RiskLevel
    recommendation: str
@dataclass
class CostIssue:
    """成本问题"""
    resource_id: str
    issue_type: str
    description: str
    potential_savings: float
    recommendation: str
@dataclass
class ComplianceIssue:
    """合规问题"""
    resource_id: str
    standard: str
    requirement: str
    status: str
    remediation: str
class CloudConfigAnalyzer:
    """云配置分析器"""
    
    def __init__(self, provider: CloudProvider = CloudProvider.AWS):
        self.provider = provider
        self.client = self._initialize_client()
        self.security_rules = self._load_security_rules()
        self.cost_rules = self._load_cost_rules()
        self.compliance_rules = self._load_compliance_rules()
    
    def _initialize_client(self):
        """初始化云客户端"""
        if self.provider == CloudProvider.AWS:
            return boto3.client('ec2')
        elif self.provider == CloudProvider.AZURE:
            # Azure client initialization
            pass
        elif self.provider == CloudProvider.GCP:
            # GCP client initialization
            pass
        return None
    
    def analyze_security(self, resources: List[CloudResource]) -> List[SecurityIssue]:
        """分析安全配置"""
        issues = []
        
        for resource in resources:
            # 检查特定资源类型的安全问题
            if resource.type == ResourceType.S3:
                issues.extend(self._analyze_s3_security(resource))
            elif resource.type == ResourceType.EC2:
                issues.extend(self._analyze_ec2_security(resource))
            elif resource.type == ResourceType.SECURITY_GROUP:
                issues.extend(self._analyze_security_group(resource))
            elif resource.type == ResourceType.IAM:
                issues.extend(self._analyze_iam_security(resource))
        
        return issues
    
    def _analyze_s3_security(self, resource: CloudResource) -> List[SecurityIssue]:
        """分析S3安全配置"""
        issues = []
        config = resource.config
        
        # 检查公开访问
        if config.get('public_read', False):
            issues.append(SecurityIssue(
                resource_id=resource.id,
                issue_type="public_access",
                description="S3存储桶配置了公开读取访问",
                risk_level=RiskLevel.HIGH,
                recommendation="禁用公开访问，使用IAM策略控制访问"
            ))
        
        # 检查加密
        if not config.get('encryption_enabled', False):
            issues.append(SecurityIssue(
                resource_id=resource.id,
                issue_type="encryption_disabled",
                description="S3存储桶未启用加密",
                risk_level=RiskLevel.MEDIUM,
                recommendation="启用S3默认加密"
            ))
        
        # 检查版本控制
        if not config.get('versioning_enabled', False):
            issues.append(SecurityIssue(
                resource_id=resource.id,
                issue_type="versioning_disabled",
                description="S3存储桶未启用版本控制",
                risk_level=RiskLevel.MEDIUM,
                recommendation="启用S3版本控制以防止意外删除"
            ))
        
        # 检查访问日志
        if not config.get('access_logging_enabled', False):
            issues.append(SecurityIssue(
                resource_id=resource.id,
                issue_type="logging_disabled",
                description="S3存储桶未启用访问日志",
                risk_level=RiskLevel.LOW,
                recommendation="启用S3访问日志以监控访问模式"
            ))
        
        return issues
    
    def _analyze_ec2_security(self, resource: CloudResource) -> List[SecurityIssue]:
        """分析EC2安全配置"""
        issues = []
        config = resource.config
        
        # 检查公开端口
        public_ports = config.get('public_ports', [])
        dangerous_ports = [22, 3389, 3306, 5432, 1433]
        
        for port in public_ports:
            if port in dangerous_ports:
                issues.append(SecurityIssue(
                    resource_id=resource.id,
                    issue_type="dangerous_public_port",
                    description=f"EC2实例公开了危险端口: {port}",
                    risk_level=RiskLevel.HIGH,
                    recommendation=f"限制端口{port}的访问，仅允许特定IP访问"
                ))
        
        # 检查IAM角色
        if not config.get('iam_role_attached', False):
            issues.append(SecurityIssue(
                resource_id=resource.id,
                issue_type="no_iam_role",
                description="EC2实例未附加IAM角色",
                risk_level=RiskLevel.MEDIUM,
                recommendation="为EC2实例附加适当的IAM角色"
            ))
        
        # 检查监控
        if not config.get('detailed_monitoring', False):
            issues.append(SecurityIssue(
                resource_id=resource.id,
                issue_type="no_detailed_monitoring",
                description="EC2实例未启用详细监控",
                risk_level=RiskLevel.LOW,
                recommendation="启用详细监控以获得更好的性能洞察"
            ))
        
        return issues
    
    def _analyze_security_group(self, resource: CloudResource) -> List[SecurityIssue]:
        """分析安全组配置"""
        issues = []
        config = resource.config
        
        # 检查0.0.0.0/0规则
        rules = config.get('rules', [])
        for rule in rules:
            if rule.get('cidr') == '0.0.0.0/0' and rule.get('port') in [22, 3389]:
                issues.append(SecurityIssue(
                    resource_id=resource.id,
                    issue_type="wide_open_admin_port",
                    description=f"管理端口{rule.get('port')}对所有IP开放",
                    risk_level=RiskLevel.CRITICAL,
                    recommendation=f"限制端口{rule.get('port')}仅访问特定IP"
                ))
        
        return issues
    
    def _analyze_iam_security(self, resource: CloudResource) -> List[SecurityIssue]:
        """分析IAM安全配置"""
        issues = []
        config = resource.config
        
        # 检查管理员权限
        if config.get('has_admin_privileges', False):
            issues.append(SecurityIssue(
                resource_id=resource.id,
                issue_type="excessive_privileges",
                description="IAM用户/角色拥有管理员权限",
                risk_level=RiskLevel.HIGH,
                recommendation="遵循最小权限原则，仅授予必要权限"
            ))
        
        # 检查MFA
        if not config.get('mfa_enabled', False):
            issues.append(SecurityIssue(
                resource_id=resource.id,
                issue_type="no_mfa",
                description="IAM用户未启用多因素认证",
                risk_level=RiskLevel.HIGH,
                recommendation="为所有IAM用户启用MFA"
            ))
        
        # 检查访问密钥
        if config.get('has_access_keys', False) and config.get('key_age_days', 0) > 90:
            issues.append(SecurityIssue(
                resource_id=resource.id,
                issue_type="old_access_keys",
                description="访问密钥使用时间超过90天",
                risk_level=RiskLevel.MEDIUM,
                recommendation="定期轮换访问密钥"
            ))
        
        return issues
    
    def analyze_costs(self, resources: List[CloudResource]) -> List[CostIssue]:
        """分析成本配置"""
        issues = []
        
        for resource in resources:
            if resource.type == ResourceType.EC2:
                issues.extend(self._analyze_ec2_costs(resource))
            elif resource.type == ResourceType.S3:
                issues.extend(self._analyze_s3_costs(resource))
            elif resource.type == ResourceType.RDS:
                issues.extend(self._analyze_rds_costs(resource))
        
        return issues
    
    def _analyze_ec2_costs(self, resource: CloudResource) -> List[CostIssue]:
        """分析EC2成本"""
        issues = []
        config = resource.config
        
        # 检查实例类型
        instance_type = config.get('instance_type', '')
        if instance_type.startswith('t2') and config.get('environment') == 'production':
            issues.append(CostIssue(
                resource_id=resource.id,
                issue_type="burstable_instance_in_prod",
                description="生产环境使用突发性能实例",
                potential_savings=50.0,
                recommendation="在生产环境使用固定性能实例"
            ))
        
        # 检查利用率
        cpu_utilization = config.get('cpu_utilization', 0)
        if cpu_utilization < 10:
            issues.append(CostIssue(
                resource_id=resource.id,
                issue_type="low_utilization",
                description=f"EC2实例CPU利用率过低: {cpu_utilization}%",
                potential_savings=80.0,
                recommendation="考虑缩小实例规模或使用自动缩放"
            ))
        
        # 检查未使用实例
        if config.get('status') == 'stopped' and config.get('stopped_days', 0) > 7:
            issues.append(CostIssue(
                resource_id=resource.id,
                issue_type="unused_instance",
                description=f"EC2实例已停止{config.get('stopped_days')}天",
                potential_savings=100.0,
                recommendation="终止未使用的实例或使用实例计划"
            ))
        
        return issues
    
    def _analyze_s3_costs(self, resource: CloudResource) -> List[CostIssue]:
        """分析S3成本"""
        issues = []
        config = resource.config
        
        # 检查存储类别
        storage_class = config.get('storage_class', 'STANDARD')
        last_access_days = config.get('last_access_days', 0)
        
        if storage_class == 'STANDARD' and last_access_days > 30:
            issues.append(CostIssue(
                resource_id=resource.id,
                issue_type="infrequent_access",
                description="数据超过30天未访问但仍使用标准存储",
                potential_savings=40.0,
                recommendation="将不常访问的数据迁移到低频访问存储"
            ))
        
        # 检查生命周期策略
        if not config.get('lifecycle_policy_enabled', False):
            issues.append(CostIssue(
                resource_id=resource.id,
                issue_type="no_lifecycle_policy",
                description="S3存储桶未配置生命周期策略",
                potential_savings=60.0,
                recommendation="配置生命周期策略自动转换存储类别"
            ))
        
        return issues
    
    def _analyze_rds_costs(self, resource: CloudResource) -> List[CostIssue]:
        """分析RDS成本"""
        issues = []
        config = resource.config
        
        # 检查实例大小
        insta