Cloud Network Security: VPCs, Security Groups, and Firewalls

requirements.txt:

boto3>=1.34.0
botocore>=1.34.0
ipaddress>=1.0.23

Complete Secure VPC Builder Implementation:

#!/usr/bin/env python3
"""
Cloud Network Security - Secure VPC Builder
Production-ready VPC configuration with comprehensive error handling
"""

import boto3
from botocore.exceptions import ClientError, BotoCoreError
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, asdict
import ipaddress
import logging
import os
import time

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


class VPCConfigError(Exception):
    """Base exception for VPC configuration errors."""
    pass


class InvalidCIDRError(VPCConfigError):
    """Raised when CIDR block is invalid."""
    pass


class VPCExistsError(VPCConfigError):
    """Raised when VPC already exists."""
    pass


@dataclass
class SubnetConfig:
    """Configuration for a subnet."""
    name: str
    cidr_block: str
    availability_zone: str
    public: bool = False
    tags: Optional[Dict[str, str]] = None
    
    def __post_init__(self):
        """Initialize default tags."""
        if self.tags is None:
            self.tags = {'Name': self.name}


@dataclass
class VPCCreationResult:
    """Result of VPC creation."""
    vpc_id: str
    vpc_cidr: str
    subnets: List[Dict]
    internet_gateway_id: Optional[str] = None
    nat_gateway_id: Optional[str] = None
    route_table_ids: List[str] = None
    
    def __post_init__(self):
        """Initialize default lists."""
        if self.route_table_ids is None:
            self.route_table_ids = []
    
    def to_dict(self) -> Dict:
        """Convert to dictionary for serialization."""
        return asdict(self)


class SecureVPCBuilder:
    """Builds secure VPC configurations with comprehensive error handling and best practices."""
    
    def __init__(
        self, 
        region_name: str = 'us-east-1',
        aws_access_key_id: Optional[str] = None,
        aws_secret_access_key: Optional[str] = None
    ):
        """Initialize VPC builder with AWS clients.
        
        Args:
            region_name: AWS region for VPC (default: us-east-1)
            aws_access_key_id: AWS access key (defaults to env/credentials)
            aws_secret_access_key: AWS secret key (defaults to env/credentials)
            
        Raises:
            VPCConfigError: If AWS clients cannot be initialized
        """
        self.region_name = region_name
        
        try:
            # Initialize AWS session
            session = boto3.Session(
                aws_access_key_id=aws_access_key_id or os.getenv('AWS_ACCESS_KEY_ID'),
                aws_secret_access_key=aws_secret_access_key or os.getenv('AWS_SECRET_ACCESS_KEY'),
                region_name=region_name
            )
            
            self.ec2 = session.client('ec2', region_name=region_name)
            
            logger.info(f"Initialized SecureVPCBuilder for region: {region_name}")
            
        except (ClientError, BotoCoreError) as e:
            error_msg = f"Failed to initialize AWS clients: {e}"
            logger.error(error_msg)
            raise VPCConfigError(error_msg) from e
    
    def validate_cidr(self, cidr: str, min_prefix: Optional[int] = None) -> Tuple[bool, Optional[str]]:
        """Validate CIDR block format with comprehensive checks.
        
        Args:
            cidr: CIDR block to validate (e.g., '10.0.0.0/16')
            min_prefix: Minimum prefix length required (e.g., 16 for /16)
            
        Returns:
            Tuple of (is_valid, error_message)
        """
        if not cidr or not isinstance(cidr, str):
            return False, "CIDR block must be a non-empty string"
        
        try:
            network = ipaddress.ip_network(cidr, strict=True)
            
            # Check if IPv4 (IPv6 requires additional handling)
            if not isinstance(network, ipaddress.IPv4Network):
                return False, "Only IPv4 networks are currently supported"
            
            # Check minimum prefix length
            if min_prefix is not None and network.prefixlen < min_prefix:
                return False, f"Prefix length /{network.prefixlen} is less than required /{min_prefix}"
            
            # Check for RFC 1918 private ranges (recommended for VPCs)
            if not network.is_private:
                return False, "VPC CIDR should use RFC 1918 private address space (10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16)"
            
            return True, None
        
        except ValueError as e:
            return False, f"Invalid CIDR format: {e}"
        except Exception as e:
            return False, f"Unexpected error validating CIDR: {e}"
    
    def create_secure_vpc(
        self, 
        cidr_block: str = '10.0.0.0/16',
        name: str = 'secure-vpc',
        subnet_configs: Optional[List[SubnetConfig]] = None,
        enable_dns: bool = True,
        enable_dns_hostnames: bool = True,
        create_internet_gateway: bool = True,
        create_nat_gateway: bool = False
    ) -> VPCCreationResult:
        """Create a secure VPC configuration with comprehensive error handling.
        
        Args:
            cidr_block: CIDR block for VPC (default: 10.0.0.0/16)
            name: Name tag for VPC (default: secure-vpc)
            subnet_configs: List of subnet configurations (default: creates public/private)
            enable_dns: Enable DNS resolution (default: True)
            enable_dns_hostnames: Enable DNS hostnames (default: True)
            create_internet_gateway: Create internet gateway for public subnets (default: True)
            create_nat_gateway: Create NAT gateway for private subnets (default: False, expensive)
            
        Returns:
            VPCCreationResult with all created resources
            
        Raises:
            InvalidCIDRError: If CIDR block is invalid
            VPCExistsError: If VPC with same CIDR exists
            VPCConfigError: If VPC creation fails
        """
        # Validate CIDR block
        is_valid, error_msg = self.validate_cidr(cidr_block, min_prefix=16)
        if not is_valid:
            raise InvalidCIDRError(error_msg or f"Invalid CIDR block: {cidr_block}")
        
        try:
            logger.info(f"Creating secure VPC with CIDR: {cidr_block}")
            
            # Create VPC
            vpc_response = self.ec2.create_vpc(
                CidrBlock=cidr_block,
                AmazonProvidedIpv6CidrBlock=False,  # Disable IPv6 for simplicity
                TagSpecifications=[{
                    'ResourceType': 'vpc',
                    'Tags': [
                        {'Key': 'Name', 'Value': name},
                        {'Key': 'Purpose', 'Value': 'secure-network'},
                        {'Key': 'ManagedBy', 'Value': 'SecureVPCBuilder'}
                    ]
                }]
            )
            vpc_id = vpc_response['Vpc']['VpcId']
            logger.info(f"Created VPC: {vpc_id}")
            
            # Wait for VPC to be available
            self._wait_for_vpc_available(vpc_id)
            
            # Enable DNS settings
            if enable_dns:
            try:
                self.ec2.modify_vpc_attribute(
                    VpcId=vpc_id,
                        EnableDnsSupport={'Value': True}
                    )
                    logger.info("Enabled DNS support")
                except ClientError as e:
                    logger.warning(f"Failed to enable DNS support: {e}")
            
            if enable_dns_hostnames:
                try:
                self.ec2.modify_vpc_attribute(
                    VpcId=vpc_id,
                        EnableDnsHostnames={'Value': True}
                )
                    logger.info("Enabled DNS hostnames")
            except ClientError as e:
                    logger.warning(f"Failed to enable DNS hostnames: {e}")
            
            # Create subnets
            if subnet_configs is None:
                # Default: Create public and private subnets
                vpc_network = ipaddress.ip_network(cidr_block)
                subnet_size = max(8, vpc_network.prefixlen + 8)  # /24 subnets by default
                
                # Get availability zones
                azs = self._get_availability_zones()
                if len(azs) < 2:
                    raise VPCConfigError("Need at least 2 availability zones for HA")
                
                subnet_configs = [
                    SubnetConfig(
                        name=f"{name}-public-1",
                        cidr_block=str(list(vpc_network.subnets(new_prefix=subnet_size))[0]),
                        availability_zone=azs[0],
                        public=True,
                        tags={'Name': f"{name}-public-1", 'Type': 'public'}
                    ),
                    SubnetConfig(
                        name=f"{name}-private-1",
                        cidr_block=str(list(vpc_network.subnets(new_prefix=subnet_size))[1]),
                        availability_zone=azs[1],
                        public=False,
                        tags={'Name': f"{name}-private-1", 'Type': 'private'}
                    )
                ]
            
            created_subnets = []
            for subnet_config in subnet_configs:
                subnet = self._create_subnet(vpc_id, subnet_config)
                created_subnets.append(subnet)
            
            # Create internet gateway for public subnets
            internet_gateway_id = None
            if create_internet_gateway:
                internet_gateway_id = self._create_internet_gateway(vpc_id, name)
                
                # Attach to public subnets
                for subnet_config, subnet in zip(subnet_configs, created_subnets):
                    if subnet_config.public:
                        self._attach_internet_gateway_to_subnet(
                            internet_gateway_id, 
                            subnet['SubnetId'],
                            subnet_config.name
                        )
            
            # Create NAT gateway if requested
            nat_gateway_id = None
            if create_nat_gateway:
                # Find first public subnet for NAT gateway
                public_subnets = [
                    (sc, s) for sc, s in zip(subnet_configs, created_subnets) if sc.public
                ]
                if public_subnets:
                    nat_subnet_config, nat_subnet = public_subnets[0]
                    nat_gateway_id = self._create_nat_gateway(
                        nat_subnet['SubnetId'],
                        f"{name}-nat-gateway"
                    )
            
            # Get route tables
            route_tables = self._get_route_tables(vpc_id)
            route_table_ids = [rt['RouteTableId'] for rt in route_tables]
            
            result = VPCCreationResult(
                vpc_id=vpc_id,
                vpc_cidr=cidr_block,
                subnets=created_subnets,
                internet_gateway_id=internet_gateway_id,
                nat_gateway_id=nat_gateway_id,
                route_table_ids=route_table_ids
            )
            
            logger.info(f"Successfully created secure VPC: {vpc_id}")
            return result
            
        except ClientError as e:
            error_code = e.response['Error']['Code']
            error_msg = e.response['Error']['Message']
            
            if error_code == 'InvalidCidrBlock':
                raise InvalidCIDRError(f"Invalid CIDR block: {cidr_block} - {error_msg}")
            elif error_code == 'VpcLimitExceeded':
                raise VPCConfigError("VPC limit exceeded for account. Delete unused VPCs or request limit increase.")
            elif error_code == 'InvalidVpc.Range':
                raise InvalidCIDRError(f"CIDR block overlaps with existing VPC: {cidr_block}")
            else:
                logger.error(f"AWS API error creating VPC: {e}")
                raise VPCConfigError(f"VPC creation failed: {error_msg}") from e
        except (InvalidCIDRError, VPCExistsError):
            raise
        except Exception as e:
            error_msg = f"Unexpected error creating VPC: {e}"
            logger.error(error_msg, exc_info=True)
            raise VPCConfigError(error_msg) from e
    
    def _wait_for_vpc_available(self, vpc_id: str, max_wait: int = 300) -> None:
        """Wait for VPC to be available.
        
        Args:
            vpc_id: VPC ID to wait for
            max_wait: Maximum wait time in seconds (default: 300)
        """
        waiter = self.ec2.get_waiter('vpc_available')
        try:
            waiter.wait(VpcIds=[vpc_id], WaiterConfig={'MaxAttempts': max_wait // 5})
            logger.debug(f"VPC {vpc_id} is now available")
        except Exception as e:
            logger.warning(f"Error waiting for VPC availability: {e}")
    
    def _get_availability_zones(self) -> List[str]:
        """Get list of availability zones in region.
        
        Returns:
            List of availability zone names
        """
        try:
            response = self.ec2.describe_availability_zones(
                Filters=[{'Name': 'state', 'Values': ['available']}]
            )
            return [az['ZoneName'] for az in response['AvailabilityZones']]
        except ClientError as e:
            logger.error(f"Error getting availability zones: {e}")
            raise VPCConfigError(f"Failed to get availability zones: {e}") from e
    
    def _create_subnet(self, vpc_id: str, config: SubnetConfig) -> Dict:
        """Create a subnet with error handling.
        
        Args:
            vpc_id: VPC ID
            config: Subnet configuration
            
        Returns:
            Subnet information dictionary
        """
        try:
            # Validate subnet CIDR
            is_valid, error_msg = self.validate_cidr(config.cidr_block)
            if not is_valid:
                raise InvalidCIDRError(error_msg or f"Invalid subnet CIDR: {config.cidr_block}")
            
            tags = [{'Key': k, 'Value': v} for k, v in config.tags.items()]
            
            response = self.ec2.create_subnet(
                    VpcId=vpc_id,
                CidrBlock=config.cidr_block,
                AvailabilityZone=config.availability_zone,
                    TagSpecifications=[{
                        'ResourceType': 'subnet',
                    'Tags': tags
                }]
            )
            
            subnet = response['Subnet']
            logger.info(f"Created subnet: {subnet['SubnetId']} ({config.name})")
            
            # Wait for subnet to be available
            time.sleep(2)  # Brief wait for subnet to become available
            
            return subnet
            
        except ClientError as e:
            error_code = e.response['Error']['Code']
            if error_code == 'InvalidSubnet.Conflict':
                raise VPCConfigError(f"Subnet CIDR conflicts with existing subnet: {config.cidr_block}")
            else:
                logger.error(f"Error creating subnet {config.name}: {e}")
                raise VPCConfigError(f"Failed to create subnet: {e}") from e
    
    def _create_internet_gateway(self, vpc_id: str, vpc_name: str) -> str:
        """Create and attach internet gateway.
        
        Args:
            vpc_id: VPC ID
            vpc_name: VPC name for tagging
            
        Returns:
            Internet gateway ID
        """
        try:
            # Create internet gateway
            igw_response = self.ec2.create_internet_gateway(
                    TagSpecifications=[{
                    'ResourceType': 'internet-gateway',
                    'Tags': [
                        {'Key': 'Name', 'Value': f"{vpc_name}-igw"},
                        {'Key': 'VpcId', 'Value': vpc_id}
                    ]
                }]
            )
            igw_id = igw_response['InternetGateway']['InternetGatewayId']
            logger.info(f"Created internet gateway: {igw_id}")
            
            # Attach to VPC
            self.ec2.attach_internet_gateway(
                InternetGatewayId=igw_id,
                VpcId=vpc_id
            )
            logger.info(f"Attached internet gateway {igw_id} to VPC {vpc_id}")
            
            return igw_id
            
        except ClientError as e:
            error_code = e.response['Error']['Code']
            if error_code == 'Resource.AlreadyAssociated':
                logger.warning(f"Internet gateway already attached to VPC")
                # Try to find existing IGW
                response = self.ec2.describe_internet_gateways(
                    Filters=[{'Name': 'attachment.vpc-id', 'Values': [vpc_id]}]
                )
                if response['InternetGateways']:
                    return response['InternetGateways'][0]['InternetGatewayId']
            logger.error(f"Error creating internet gateway: {e}")
            raise VPCConfigError(f"Failed to create internet gateway: {e}") from e
    
    def _attach_internet_gateway_to_subnet(
        self, 
        igw_id: str, 
        subnet_id: str,
        subnet_name: str
    ) -> None:
        """Configure route table to route traffic through internet gateway.
        
        Args:
            igw_id: Internet gateway ID
            subnet_id: Subnet ID
            subnet_name: Subnet name for logging
        """
        try:
            # Get route table for subnet
            route_tables = self.ec2.describe_route_tables(
                Filters=[{'Name': 'association.subnet-id', 'Values': [subnet_id]}]
            )
            
            if not route_tables['RouteTables']:
                # Create route table if doesn't exist
                route_table = self.ec2.create_route_table(
                    VpcId=subnet_id.split('-')[0],  # Extract VPC ID from subnet
                    TagSpecifications=[{
                        'ResourceType': 'route-table',
                        'Tags': [{'Key': 'Name', 'Value': f"{subnet_name}-rt"}]
                    }]
                )
                route_table_id = route_table['RouteTable']['RouteTableId']
            else:
                route_table_id = route_tables['RouteTables'][0]['RouteTableId']
            
            # Add route to internet gateway
            self.ec2.create_route(
                RouteTableId=route_table_id,
                DestinationCidrBlock='0.0.0.0/0',
                GatewayId=igw_id
            )
            
            # Associate route table with subnet
            self.ec2.associate_route_table(
                RouteTableId=route_table_id,
                SubnetId=subnet_id
            )
            
            logger.info(f"Configured routing for subnet {subnet_name} through IGW")
            
        except ClientError as e:
            logger.warning(f"Error configuring routing for subnet {subnet_name}: {e}")
    
    def _create_nat_gateway(self, subnet_id: str, name: str) -> str:
        """Create NAT gateway (requires Elastic IP).
        
        Args:
            subnet_id: Public subnet ID for NAT gateway
            name: Name for NAT gateway
            
        Returns:
            NAT gateway ID
        """
        try:
            # Allocate Elastic IP
            eip_response = self.ec2.allocate_address(Domain='vpc')
            eip_allocation_id = eip_response['AllocationId']
            logger.info(f"Allocated Elastic IP: {eip_response['PublicIp']}")
            
            # Create NAT gateway
            nat_response = self.ec2.create_nat_gateway(
                SubnetId=subnet_id,
                AllocationId=eip_allocation_id,
                TagSpecifications=[{
                    'ResourceType': 'nat-gateway',
                    'Tags': [{'Key': 'Name', 'Value': name}]
                }]
            )
            nat_gateway_id = nat_response['NatGateway']['NatGatewayId']
            logger.info(f"Created NAT gateway: {nat_gateway_id} (this may take a few minutes)")
            
            return nat_gateway_id
            
        except ClientError as e:
            logger.error(f"Error creating NAT gateway: {e}")
            raise VPCConfigError(f"Failed to create NAT gateway: {e}") from e
    
    def _get_route_tables(self, vpc_id: str) -> List[Dict]:
        """Get route tables for VPC.
        
        Args:
            vpc_id: VPC ID
            
        Returns:
            List of route tables
        """
        try:
            response = self.ec2.describe_route_tables(
                Filters=[{'Name': 'vpc-id', 'Values': [vpc_id]}]
            )
            return response['RouteTables']
        except ClientError as e:
            logger.warning(f"Error getting route tables: {e}")
            return []


# Example usage
if __name__ == "__main__":
    builder = SecureVPCBuilder(region_name='us-east-1')
    
    try:
        result = builder.create_secure_vpc(
            cidr_block='10.0.0.0/16',
            name='secure-production-vpc',
            create_internet_gateway=True,
            create_nat_gateway=False  # Set to True for private subnet internet access
        )
        
        print(f"Created VPC: {result.vpc_id}")
        print(f"CIDR: {result.vpc_cidr}")
        print(f"Subnets: {len(result.subnets)}")
        if result.internet_gateway_id:
            print(f"Internet Gateway: {result.internet_gateway_id}")
        if result.nat_gateway_id:
            print(f"NAT Gateway: {result.nat_gateway_id}")
    
    except VPCConfigError as e:
        print(f"Error: {e}")

Unit Tests:

# test_secure_vpc_builder.py
import pytest
from unittest.mock import Mock, patch, MagicMock
from secure_vpc_builder import (
    SecureVPCBuilder,
    VPCCreationResult,
    SubnetConfig,
    InvalidCIDRError,
    VPCConfigError
)
from botocore.exceptions import ClientError


class TestSecureVPCBuilder:
    """Unit tests for SecureVPCBuilder."""
    
    @pytest.fixture
    def mock_ec2_client(self):
        """Create mock EC2 client."""
        with patch('secure_vpc_builder.boto3.Session') as mock_session:
            mock_session_instance = Mock()
            mock_client = Mock()
            mock_session.return_value.client.return_value = mock_client
            mock_session.return_value = mock_session_instance
            mock_session_instance.client.return_value = mock_client
            yield mock_client
    
    @pytest.fixture
    def builder(self, mock_ec2_client):
        """Create SecureVPCBuilder instance."""
        return SecureVPCBuilder(region_name='us-east-1')
    
    def test_validate_cidr_valid(self, builder):
        """Test CIDR validation with valid CIDR."""
        is_valid, error = builder.validate_cidr('10.0.0.0/16')
        assert is_valid is True
        assert error is None
    
    def test_validate_cidr_invalid_format(self, builder):
        """Test CIDR validation with invalid format."""
        is_valid, error = builder.validate_cidr('invalid')
        assert is_valid is False
        assert error is not None
    
    def test_validate_cidr_public_ip(self, builder):
        """Test CIDR validation rejects public IP ranges."""
        is_valid, error = builder.validate_cidr('8.8.8.0/24')
        assert is_valid is False
        assert 'private' in error.lower()
    
    def test_create_vpc_success(self, builder, mock_ec2_client):
        """Test successful VPC creation."""
        # Mock responses
        mock_ec2_client.create_vpc.return_value = {
            'Vpc': {'VpcId': 'vpc-12345', 'CidrBlock': '10.0.0.0/16'}
        }
        mock_ec2_client.describe_availability_zones.return_value = {
            'AvailabilityZones': [
                {'ZoneName': 'us-east-1a', 'State': 'available'},
                {'ZoneName': 'us-east-1b', 'State': 'available'}
            ]
        }
        mock_ec2_client.create_subnet.return_value = {
            'Subnet': {'SubnetId': 'subnet-12345', 'VpcId': 'vpc-12345'}
        }
        mock_ec2_client.describe_route_tables.return_value = {
            'RouteTables': [{'RouteTableId': 'rtb-12345'}]
        }
        
        # Mock waiter
        mock_waiter = Mock()
        builder.ec2.get_waiter = Mock(return_value=mock_waiter)
        
        result = builder.create_secure_vpc(create_internet_gateway=False)
        
        assert isinstance(result, VPCCreationResult)
        assert result.vpc_id == 'vpc-12345'
    
    def test_create_vpc_invalid_cidr(self, builder):
        """Test VPC creation with invalid CIDR raises error."""
        with pytest.raises(InvalidCIDRError):
            builder.create_secure_vpc(cidr_block='invalid-cidr')


if __name__ == '__main__':
    pytest.main([__file__, '-v'])

Complete Security Group Builder Implementation:

#!/usr/bin/env python3
"""
Cloud Network Security - Security Group Builder
Production-ready security group configuration with comprehensive validation
"""

import boto3
from botocore.exceptions import ClientError, BotoCoreError
from typing import Dict, List, Optional, Set
from dataclasses import dataclass, asdict
from enum import Enum
import ipaddress
import logging
import os

logger = logging.getLogger(__name__)


class SecurityGroupError(Exception):
    """Base exception for security group errors."""
    pass


class InvalidRuleError(SecurityGroupError):
    """Raised when security group rule is invalid."""
    pass


class Protocol(Enum):
    """Network protocols."""
    TCP = "tcp"
    UDP = "udp"
    ICMP = "icmp"
    ALL = "-1"
    
    def __str__(self) -> str:
        return self.value


@dataclass
class SecurityGroupRule:
    """Represents a security group rule with comprehensive validation."""
    protocol: str
    from_port: int
    to_port: int
    cidr_ip: Optional[str] = None
    source_sg: Optional[str] = None
    description: str = ""
    ipv6_cidr: Optional[str] = None
    prefix_list_id: Optional[str] = None
    
    def to_dict(self) -> Dict:
        """Convert rule to AWS API format."""
        rule = {
            'IpProtocol': self.protocol,
            'FromPort': self.from_port,
            'ToPort': self.to_port,
        }
        
        if self.description:
            rule['Description'] = self.description
        
        # Add IP ranges or source security group
        ip_ranges = []
        if self.cidr_ip:
            ip_ranges.append({
                'CidrIp': self.cidr_ip,
                'Description': self.description or f"{self.protocol} from {self.cidr_ip}"
            })
        
        if self.ipv6_cidr:
            ip_ranges.append({
                'CidrIpv6': self.ipv6_cidr,
                'Description': self.description or f"{self.protocol} from {self.ipv6_cidr}"
            })
        
        if ip_ranges:
            rule['IpRanges'] = ip_ranges
        
        if self.source_sg:
            rule['UserIdGroupPairs'] = [{
                'GroupId': self.source_sg,
                'Description': self.description or f"{self.protocol} from security group"
            }]
        
        if self.prefix_list_id:
            rule['PrefixListIds'] = [{
                'PrefixListId': self.prefix_list_id,
                'Description': self.description
            }]
        
        return rule


class SecurityGroupBuilder:
    """Builds secure security groups with comprehensive error handling and best practices."""
    
    # Common port definitions
    COMMON_PORTS = {
        'HTTP': 80,
        'HTTPS': 443,
        'SSH': 22,
        'RDP': 3389,
        'MYSQL': 3306,
        'POSTGRES': 5432,
        'MONGODB': 27017,
        'REDIS': 6379
    }
    
    def __init__(
        self,
        region_name: str = 'us-east-1',
        aws_access_key_id: Optional[str] = None,
        aws_secret_access_key: Optional[str] = None
    ):
        """Initialize security group builder.
        
        Args:
            region_name: AWS region (default: us-east-1)
            aws_access_key_id: AWS access key (defaults to env/credentials)
            aws_secret_access_key: AWS secret key (defaults to env/credentials)
        """
        self.region_name = region_name
        
        try:
            session = boto3.Session(
                aws_access_key_id=aws_access_key_id or os.getenv('AWS_ACCESS_KEY_ID'),
                aws_secret_access_key=aws_secret_access_key or os.getenv('AWS_SECRET_ACCESS_KEY'),
                region_name=region_name
            )
            
            self.ec2 = session.client('ec2', region_name=region_name)
            logger.info(f"Initialized SecurityGroupBuilder for region: {region_name}")
            
        except (ClientError, BotoCoreError) as e:
            error_msg = f"Failed to initialize AWS clients: {e}"
            logger.error(error_msg)
            raise SecurityGroupError(error_msg) from e
    
    def validate_rule(self, rule: SecurityGroupRule) -> Tuple[bool, Optional[str]]:
        """Validate security group rule.
        
        Args:
            rule: Security group rule to validate
            
        Returns:
            Tuple of (is_valid, error_message)
        """
        # Validate protocol
        valid_protocols = ['tcp', 'udp', 'icmp', '-1', 'icmpv6', '58']
        if rule.protocol not in valid_protocols:
            return False, f"Invalid protocol: {rule.protocol}. Must be one of {valid_protocols}"
        
        # Validate ports
        if rule.protocol in ['tcp', 'udp']:
            if rule.from_port < 0 or rule.from_port > 65535:
                return False, f"Invalid from_port: {rule.from_port}. Must be 0-65535"
            if rule.to_port < 0 or rule.to_port > 65535:
                return False, f"Invalid to_port: {rule.to_port}. Must be 0-65535"
            if rule.from_port > rule.to_port:
                return False, f"from_port ({rule.from_port}) cannot be greater than to_port ({rule.to_port})"
        
        # Validate CIDR blocks
        if rule.cidr_ip:
            try:
                network = ipaddress.ip_network(rule.cidr_ip, strict=False)
                if not isinstance(network, ipaddress.IPv4Network):
                    return False, "IPv4 CIDR required for cidr_ip field"
            except ValueError as e:
                return False, f"Invalid CIDR block: {e}"
        
        if rule.ipv6_cidr:
            try:
                network = ipaddress.ip_network(rule.ipv6_cidr, strict=False)
                if not isinstance(network, ipaddress.IPv6Network):
                    return False, "IPv6 CIDR required for ipv6_cidr field"
            except ValueError as e:
                return False, f"Invalid IPv6 CIDR block: {e}"
        
        # Must have either CIDR, source SG, or prefix list
        if not any([rule.cidr_ip, rule.source_sg, rule.prefix_list_id, rule.ipv6_cidr]):
            return False, "Rule must specify cidr_ip, source_sg, prefix_list_id, or ipv6_cidr"
        
        # Check for overly permissive rules
        if rule.cidr_ip == '0.0.0.0/0' and rule.protocol in ['tcp', 'udp']:
            if rule.from_port in [22, 3389]:  # SSH, RDP
                return False, f"CRITICAL: Allowing {rule.from_port} from 0.0.0.0/0 is a security risk. Restrict to specific IPs."
        
        return True, None
    
    def create_security_group(
        self, 
        vpc_id: str, 
        name: str,
                             description: str = "Secure security group",
        rules: Optional[List[SecurityGroupRule]] = None,
        tags: Optional[Dict[str, str]] = None
    ) -> Dict:
        """Create a secure security group with comprehensive error handling.
        
        Args:
            vpc_id: VPC ID for security group
            name: Name for security group
            description: Description for security group
            rules: List of rules to add (default: HTTPS only from anywhere)
            tags: Additional tags for security group
            
        Returns:
            Security group information dictionary
            
        Raises:
            SecurityGroupError: If security group creation fails
            InvalidRuleError: If any rule is invalid
        """
        if not vpc_id:
            raise SecurityGroupError("VPC ID is required")
        
        if not name:
            raise SecurityGroupError("Security group name is required")
        
        # Validate and set default rules
        if rules is None:
            rules = [
                SecurityGroupRule(
                    protocol='tcp',
                    from_port=443,
                    to_port=443,
                    cidr_ip='0.0.0.0/0',
                    description='HTTPS access'
                )
            ]
        
        # Validate all rules before creation
        for i, rule in enumerate(rules):
            is_valid, error_msg = self.validate_rule(rule)
            if not is_valid:
                raise InvalidRuleError(f"Rule {i+1} is invalid: {error_msg}")
        
        try:
            logger.info(f"Creating security group '{name}' in VPC {vpc_id}")
            
            # Prepare tags
            tag_list = [{'Key': 'Name', 'Value': name}]
            if tags:
                tag_list.extend([{'Key': k, 'Value': v} for k, v in tags.items()])
            tag_list.append({'Key': 'ManagedBy', 'Value': 'SecurityGroupBuilder'})
            
            # Create security group
            sg_response = self.ec2.create_security_group(
                GroupName=name,
                Description=description,
                VpcId=vpc_id,
                TagSpecifications=[{
                    'ResourceType': 'security-group',
                    'Tags': tag_list
                }]
            )
            sg_id = sg_response['GroupId']
            logger.info(f"Created security group: {sg_id}")
            
            # Add rules
            if rules:
                self.add_rules_to_security_group(sg_id, rules)
            
            # Get full security group details
            sg_details = self.get_security_group(sg_id)
            
            return sg_details
            
        except ClientError as e:
            error_code = e.response['Error']['Code']
            error_msg = e.response['Error']['Message']
            
            if error_code == 'InvalidGroup.Duplicate':
                raise SecurityGroupError(f"Security group '{name}' already exists in VPC") from e
            elif error_code == 'InvalidVpcID.NotFound':
                raise SecurityGroupError(f"VPC {vpc_id} not found") from e
            else:
                logger.error(f"AWS API error creating security group: {e}")
                raise SecurityGroupError(f"Failed to create security group: {error_msg}") from e
        except (SecurityGroupError, InvalidRuleError):
            raise
        except Exception as e:
            error_msg = f"Unexpected error creating security group: {e}"
            logger.error(error_msg, exc_info=True)
            raise SecurityGroupError(error_msg) from e
    
    def add_rules_to_security_group(
        self,
        sg_id: str,
        rules: List[SecurityGroupRule],
        is_egress: bool = False
    ) -> None:
        """Add rules to existing security group.
        
        Args:
            sg_id: Security group ID
            rules: List of rules to add
            is_egress: If True, add egress rules; if False, add ingress rules
            
        Raises:
            InvalidRuleError: If any rule is invalid
            SecurityGroupError: If rule addition fails
        """
        if not rules:
            return
        
        # Validate all rules
        for i, rule in enumerate(rules):
            is_valid, error_msg = self.validate_rule(rule)
            if not is_valid:
                raise InvalidRuleError(f"Rule {i+1} is invalid: {error_msg}")
        
        try:
            # Convert rules to AWS API format
            aws_rules = [rule.to_dict() for rule in rules]
            
            if is_egress:
                self.ec2.authorize_security_group_egress(
                    GroupId=sg_id,
                    IpPermissions=aws_rules
                )
                logger.info(f"Added {len(aws_rules)} egress rules to security group {sg_id}")
            else:
                self.ec2.authorize_security_group_ingress(
                    GroupId=sg_id,
                    IpPermissions=aws_rules
                )
                logger.info(f"Added {len(aws_rules)} ingress rules to security group {sg_id}")
                
        except ClientError as e:
            error_code = e.response['Error']['Code']
            
            if error_code == 'InvalidPermission.Duplicate':
                logger.warning(f"Rule already exists in security group {sg_id}")
            elif error_code == 'InvalidGroup.NotFound':
                raise SecurityGroupError(f"Security group {sg_id} not found") from e
            else:
                error_msg = f"Failed to add rules to security group: {e}"
                logger.error(error_msg)
                raise SecurityGroupError(error_msg) from e
    
    def get_security_group(self, sg_id: str) -> Dict:
        """Get security group details.
        
        Args:
            sg_id: Security group ID
            
        Returns:
            Security group details dictionary
        """
        try:
            response = self.ec2.describe_security_groups(GroupIds=[sg_id])
            if not response['SecurityGroups']:
                raise SecurityGroupError(f"Security group {sg_id} not found")
            
            return response['SecurityGroups'][0]
            
        except ClientError as e:
            error_msg = f"Failed to get security group {sg_id}: {e}"
            logger.error(error_msg)
            raise SecurityGroupError(error_msg) from e
    
    def create_secure_web_server_sg(self, vpc_id: str, allowed_cidrs: Optional[List[str]] = None) -> Dict:
        """Create security group for secure web server (HTTPS only).
        
        Args:
            vpc_id: VPC ID
            allowed_cidrs: List of allowed CIDR blocks (default: anywhere)
            
        Returns:
            Security group information
        """
        if allowed_cidrs is None:
            allowed_cidrs = ['0.0.0.0/0']
        
        rules = []
        for cidr in allowed_cidrs:
            rules.append(SecurityGroupRule(
                protocol='tcp',
                from_port=443,
                to_port=443,
                cidr_ip=cidr,
                description=f'HTTPS access from {cidr}'
            ))
            # Allow HTTP for redirect to HTTPS (optional)
            rules.append(SecurityGroupRule(
                protocol='tcp',
                from_port=80,
                to_port=80,
                cidr_ip=cidr,
                description=f'HTTP access from {cidr} (redirect to HTTPS)'
            ))
        
        return self.create_security_group(
            vpc_id=vpc_id,
            name='secure-web-server-sg',
            description='Security group for web servers (HTTPS only)',
            rules=rules
        )
    
    def create_secure_ssh_sg(self, vpc_id: str, allowed_cidrs: List[str]) -> Dict:
        """Create security group for SSH access (restricted to specific IPs).
        
        Args:
            vpc_id: VPC ID
            allowed_cidrs: List of allowed CIDR blocks (REQUIRED for SSH)
            
        Returns:
            Security group information
            
        Raises:
            SecurityGroupError: If no allowed CIDRs provided
        """
        if not allowed_cidrs:
            raise SecurityGroupError("SSH access requires allowed CIDR blocks for security")
        
        rules = []
        for cidr in allowed_cidrs:
            rules.append(SecurityGroupRule(
                protocol='tcp',
                from_port=22,
                to_port=22,
                cidr_ip=cidr,
                description=f'SSH access from {cidr}'
            ))
        
        return self.create_security_group(
            vpc_id=vpc_id,
            name='secure-ssh-sg',
            description='Security group for SSH access (restricted IPs)',
            rules=rules
        )


# Example usage
if __name__ == "__main__":
    builder = SecurityGroupBuilder(region_name='us-east-1')
    
    # Create secure web server security group
    web_sg = builder.create_secure_web_server_sg(
        vpc_id='vpc-12345678',
        allowed_cidrs=['0.0.0.0/0']  # Or restrict to specific IPs
    )
    print(f"Created web server security group: {web_sg['GroupId']}")
    
    # Create SSH security group (must specify allowed IPs)
    ssh_sg = builder.create_secure_ssh_sg(
        vpc_id='vpc-12345678',
        allowed_cidrs=['203.0.113.0/24', '198.51.100.0/24']  # Your office IPs
    )
    print(f"Created SSH security group: {ssh_sg['GroupId']}")

Unit Tests:

# test_security_group_builder.py
import pytest
from security_group_builder import (
    SecurityGroupBuilder,
    SecurityGroupRule,
    SecurityGroupError,
    InvalidRuleError
)


class TestSecurityGroupBuilder:
    """Unit tests for SecurityGroupBuilder."""
    
    @pytest.fixture
    def mock_ec2_client(self):
        """Create mock EC2 client."""
        with patch('security_group_builder.boto3.Session') as mock_session:
            mock_client = Mock()
            mock_session.return_value.client.return_value = mock_client
            yield mock_client
    
    @pytest.fixture
    def builder(self, mock_ec2_client):
        """Create SecurityGroupBuilder instance."""
        return SecurityGroupBuilder(region_name='us-east-1')
    
    def test_validate_rule_valid(self, builder):
        """Test rule validation with valid rule."""
        rule = SecurityGroupRule(
            protocol='tcp',
            from_port=443,
            to_port=443,
            cidr_ip='10.0.0.0/8',
            description='HTTPS'
        )
        
        is_valid, error = builder.validate_rule(rule)
        assert is_valid is True
        assert error is None
    
    def test_validate_rule_invalid_cidr(self, builder):
        """Test rule validation rejects invalid CIDR."""
        rule = SecurityGroupRule(
            protocol='tcp',
            from_port=443,
            to_port=443,
            cidr_ip='invalid-cidr'
        )
        
        is_valid, error = builder.validate_rule(rule)
        assert is_valid is False
        assert error is not None
    
    def test_validate_rule_ssh_public_warning(self, builder):
        """Test rule validation warns about public SSH access."""
        rule = SecurityGroupRule(
            protocol='tcp',
            from_port=22,
            to_port=22,
            cidr_ip='0.0.0.0/0'
        )
        
        is_valid, error = builder.validate_rule(rule)
        assert is_valid is False
        assert 'security risk' in error.lower()


if __name__ == '__main__':
    pytest.main([__file__, '-v'])

# Verify VPC exists
aws ec2 describe-vpcs --vpc-ids vpc-12345678

# Verify security groups
aws ec2 describe-security-groups --group-ids sg-12345678

# Test connectivity (from instance)
nc -zv 10.0.1.10 443  # Test HTTPS to public subnet
nc -zv 10.0.2.10 5432  # Test database port to private subnet

# Verify no open ports
aws ec2 describe-security-groups --filters "Name=ip-permission.cidr,Values=0.0.0.0/0"
# Should show minimal open ports

# Delete security groups (must remove rules first)
aws ec2 delete-security-group --group-id sg-12345678

# Delete subnets
aws ec2 delete-subnet --subnet-id subnet-12345678

# Delete VPC (must delete all resources first)
aws ec2 delete-vpc --vpc-id vpc-12345678

# Verify cleanup
aws ec2 describe-vpcs --vpc-ids vpc-12345678
# Should return: "The vpc ID 'vpc-12345678' does not exist"