advanced_analysis.py

"""
Advanced AWS Security Groups Analysis with Enhanced Features
This module provides additional analysis capabilities beyond the basic security group analysis
"""

import pandas as pd
import numpy as np
from typing import Dict, List, Any, Tuple
from collections import defaultdict
import ipaddress
from security_group_analyzer import SecurityGroupAnalyzer, SecurityGroupRule

class AdvancedSecurityAnalyzer:
    """Enhanced security analysis with risk scoring and recommendations"""
    
    def __init__(self, analyzer: SecurityGroupAnalyzer):
        self.analyzer = analyzer
        self.risk_scores = {}
        
    def calculate_risk_score(self, rule: SecurityGroupRule) -> int:
        """Calculate risk score for a security group rule (0-100)"""
        risk_score = 0
        
        # Public access is high risk
        if rule.is_public:
            risk_score += 50
            
        # Common vulnerable ports
        vulnerable_ports = {22, 3389, 23, 21, 135, 139, 445, 1433, 3306, 5432, 6379}
        if rule.from_port in vulnerable_ports:
            risk_score += 30
            
        # Wide port ranges are risky
        if rule.from_port is not None and rule.to_port is not None:
            port_range = rule.to_port - rule.from_port
            if port_range > 100:
                risk_score += 20
            elif port_range > 10:
                risk_score += 10
                
        # Protocol considerations
        if rule.ip_protocol == 'tcp' and rule.from_port == 80:
            risk_score += 10  # HTTP is unencrypted
        elif rule.ip_protocol == 'tcp' and rule.from_port == 443:
            risk_score += 5   # HTTPS is better but still public
            
        # All traffic rules (-1 protocol)
        if rule.ip_protocol == '-1':
            risk_score += 25
            
        return min(risk_score, 100)  # Cap at 100
    
    def get_risk_analysis(self) -> pd.DataFrame:
        """Generate comprehensive risk analysis"""
        risk_data = []
        
        for rule in self.analyzer.security_groups_data:
            risk_score = self.calculate_risk_score(rule)
            
            risk_data.append({
                'account_id': rule.account_id,
                'region': rule.region,
                'security_group_id': rule.security_group_id,
                'security_group_name': rule.security_group_name,
                'rule_type': rule.rule_type,
                'protocol': rule.ip_protocol,
                'port_range': f"{rule.from_port}-{rule.to_port}" if rule.from_port else "All",
                'is_public': rule.is_public,
                'risk_score': risk_score,
                'risk_level': self.get_risk_level(risk_score),
                'cidr_blocks': ', '.join(rule.cidr_blocks),
                'description': rule.description
            })
        
        return pd.DataFrame(risk_data)
    
    def get_risk_level(self, score: int) -> str:
        """Convert numeric risk score to descriptive level"""
        if score >= 80:
            return "CRITICAL"
        elif score >= 60:
            return "HIGH"
        elif score >= 40:
            return "MEDIUM"
        elif score >= 20:
            return "LOW"
        else:
            return "MINIMAL"
    
    def get_remediation_recommendations(self) -> List[Dict[str, Any]]:
        """Generate specific remediation recommendations"""
        recommendations = []
        
        # Analyze public access rules
        public_rules = [rule for rule in self.analyzer.security_groups_data if rule.is_public]
        
        for rule in public_rules:
            rec = {
                'account_id': rule.account_id,
                'security_group_id': rule.security_group_id,
                'security_group_name': rule.security_group_name,
                'issue': 'Public Access (0.0.0.0/0)',
                'severity': 'HIGH',
                'recommendation': self.get_specific_recommendation(rule),
                'port': rule.from_port,
                'protocol': rule.ip_protocol
            }
            recommendations.append(rec)
        
        return recommendations
    
    def get_specific_recommendation(self, rule: SecurityGroupRule) -> str:
        """Get specific recommendation based on rule characteristics"""
        port = rule.from_port
        protocol = rule.ip_protocol
        
        recommendations_map = {
            22: "SSH access should be restricted to specific IP ranges or use bastion hosts",
            3389: "RDP access should be restricted to specific IP ranges or use VPN",
            80: "HTTP traffic should be behind a load balancer or use HTTPS (443)",
            443: "HTTPS traffic should be behind a load balancer with proper SSL termination",
            1433: "SQL Server access should never be public - use private subnets",
            3306: "MySQL access should never be public - use private subnets",
            5432: "PostgreSQL access should never be public - use private subnets",
            6379: "Redis access should never be public - use private subnets"
        }
        
        if port in recommendations_map:
            return recommendations_map[port]
        else:
            return f"Port {port} ({protocol}) should not be publicly accessible - restrict to specific IP ranges"
    
    def get_security_group_relationships(self) -> pd.DataFrame:
        """Analyze relationships and dependencies between security groups"""
        relationships = []
        
        # Group rules by security group
        sg_rules = defaultdict(list)
        for rule in self.analyzer.security_groups_data:
            sg_rules[rule.security_group_id].extend(rule.security_group_refs)
        
        # Analyze relationships
        for sg_id, referenced_sgs in sg_rules.items():
            if referenced_sgs:
                sg_info = next((rule for rule in self.analyzer.security_groups_data 
                               if rule.security_group_id == sg_id), None)
                
                if sg_info:
                    relationships.append({
                        'source_sg_id': sg_id,
                        'source_sg_name': sg_info.security_group_name,
                        'source_account': sg_info.account_id,
                        'source_region': sg_info.region,
                        'referenced_sgs': ', '.join(set(referenced_sgs)),
                        'reference_count': len(set(referenced_sgs)),
                        'vpc_id': sg_info.vpc_id
                    })
        
        return pd.DataFrame(relationships)
    
    def analyze_unused_security_groups(self) -> List[Dict[str, Any]]:
        """Identify potentially unused security groups"""
        # This is a simplified analysis - in practice, you'd need to check EC2 instances, 
        # load balancers, RDS instances, etc.
        
        sg_usage = defaultdict(int)
        
        # Count references to each security group
        for rule in self.analyzer.security_groups_data:
            for sg_ref in rule.security_group_refs:
                sg_usage[sg_ref] += 1
        
        # Find security groups with no references and minimal rules
        potentially_unused = []
        
        sg_rules_count = defaultdict(int)
        for rule in self.analyzer.security_groups_data:
            sg_rules_count[rule.security_group_id] += 1
        
        for rule in self.analyzer.security_groups_data:
            sg_id = rule.security_group_id
            if (sg_usage[sg_id] == 0 and sg_rules_count[sg_id] <= 2):  # Only default rules
                potentially_unused.append({
                    'account_id': rule.account_id,
                    'region': rule.region,
                    'security_group_id': sg_id,
                    'security_group_name': rule.security_group_name,
                    'rule_count': sg_rules_count[sg_id],
                    'referenced_by': sg_usage[sg_id],
                    'vpc_id': rule.vpc_id
                })
        
        # Remove duplicates
        seen = set()
        unique_unused = []
        for item in potentially_unused:
            key = (item['account_id'], item['security_group_id'])
            if key not in seen:
                seen.add(key)
                unique_unused.append(item)
        
        return unique_unused
    
    def get_compliance_report(self) -> Dict[str, Any]:
        """Generate compliance report for common security standards"""
        
        total_rules = len(self.analyzer.security_groups_data)
        public_rules = len([rule for rule in self.analyzer.security_groups_data if rule.is_public])
        
        # Common vulnerable ports analysis
        vulnerable_ports = {22, 3389, 23, 21, 135, 139, 445, 1433, 3306, 5432, 6379}
        vulnerable_public_rules = []
        
        for rule in self.analyzer.security_groups_data:
            if rule.is_public and rule.from_port in vulnerable_ports:
                vulnerable_public_rules.append(rule)
        
        # Wide-open rules (all ports, all protocols)
        wide_open_rules = [
            rule for rule in self.analyzer.security_groups_data 
            if rule.is_public and rule.ip_protocol == '-1'
        ]
        
        compliance_score = 100
        issues = []
        
        if public_rules > 0:
            compliance_score -= min(30, (public_rules / total_rules) * 100)
            issues.append(f"Found {public_rules} rules allowing public access")
        
        if vulnerable_public_rules:
            compliance_score -= min(40, len(vulnerable_public_rules) * 5)
            issues.append(f"Found {len(vulnerable_public_rules)} rules exposing vulnerable ports publicly")
        
        if wide_open_rules:
            compliance_score -= min(30, len(wide_open_rules) * 10)
            issues.append(f"Found {len(wide_open_rules)} rules allowing all traffic publicly")
        
        return {
            'compliance_score': max(0, compliance_score),
            'total_rules_analyzed': total_rules,
            'public_rules_count': public_rules,
            'vulnerable_public_rules': len(vulnerable_public_rules),
            'wide_open_rules': len(wide_open_rules),
            'issues': issues,
            'status': 'PASS' if compliance_score >= 80 else 'FAIL',
            'recommendation': 'Review and restrict public access rules' if compliance_score < 80 else 'Security posture is acceptable'
        }


def generate_executive_summary(analyzer: SecurityGroupAnalyzer) -> str:
    """Generate executive summary report"""
    
    advanced_analyzer = AdvancedSecurityAnalyzer(analyzer)
    compliance_report = advanced_analyzer.get_compliance_report()
    risk_analysis = advanced_analyzer.get_risk_analysis()
    
    # Calculate summary statistics
    total_sgs = len(set(rule.security_group_id for rule in analyzer.security_groups_data))
    accounts_analyzed = len(set(rule.account_id for rule in analyzer.security_groups_data))
    regions_analyzed = len(set(rule.region for rule in analyzer.security_groups_data))
    
    high_risk_rules = len(risk_analysis[risk_analysis['risk_level'].isin(['HIGH', 'CRITICAL'])])
    
    summary = f"""
    ========================================
    AWS SECURITY GROUPS ANALYSIS - EXECUTIVE SUMMARY
    ========================================
    
    SCOPE:
    • Accounts Analyzed: {accounts_analyzed}
    • Regions Analyzed: {regions_analyzed}
    • Security Groups: {total_sgs}
    • Total Rules: {len(analyzer.security_groups_data)}
    
    SECURITY POSTURE:
    • Compliance Score: {compliance_report['compliance_score']:.1f}/100
    • Status: {compliance_report['status']}
    • Public Access Rules: {compliance_report['public_rules_count']}
    • High-Risk Rules: {high_risk_rules}
    
    CRITICAL FINDINGS:
    """
    
    for issue in compliance_report['issues']:
        summary += f"    • {issue}\n"
    
    summary += f"""
    
    RECOMMENDATIONS:
    • {compliance_report['recommendation']}
    • Review all rules with risk level HIGH or CRITICAL
    • Implement least-privilege access controls
    • Use VPC endpoints for AWS services where possible
    • Regular security group audits should be performed
    
    NEXT STEPS:
    1. Review detailed risk analysis report
    2. Prioritize remediation of CRITICAL and HIGH risk rules
    3. Implement network segmentation best practices
    4. Set up automated security group monitoring
    """
    
    return summary