import boto3 import argparse import json import time def enforce_vpc_flow_logs( log_destination_type, log_destination_arn, traffic_type='ALL', region_name='us-east-1', dry_run=True ): """ Enforces VPC Flow Logs configuration for all VPCs in a given region. It ensures Flow Logs are enabled and configured to send to a specified destination. Args: log_destination_type (str): Type of destination ('s3' or 'cloudwatch'). log_destination_arn (str): ARN of the S3 bucket or CloudWatch Log Group. traffic_type (str): The type of traffic to log ('ACCEPT', 'REJECT', or 'ALL'). region_name (str): The AWS region to enforce Flow Logs. dry_run (bool): If True, only reports changes without applying them. If False, applies changes. """ ec2_client = boto3.client('ec2', region_name=region_name) iam_client = boto3.client('iam', region_name=region_name) flow_log_role_name = "FlowLogsRole" flow_log_role_arn = "" print(f"Starting VPC Flow Log enforcement in region {region_name}...") if dry_run: print("*** DRY RUN MODE: No changes will be applied. ***") # 1. Ensure Flow Log IAM Role exists and has necessary permissions print("\n>>> Step 1: Ensuring Flow Log IAM Role exists...") try: # Check if role exists role_response = iam_client.get_role(RoleName=flow_log_role_name) flow_log_role_arn = role_response['Role']['Arn'] print(f" IAM Role '{flow_log_role_name}' already exists: {flow_log_role_arn}") except iam_client.exceptions.NoSuchEntityException: # Create role if it doesn't exist print(f" IAM Role '{flow_log_role_name}' not found. Creating...") assume_role_policy_document = { "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": {"Service": "ec2.amazonaws.com"}, "Action": "sts:AssumeRole" } ] } create_role_response = iam_client.create_role( RoleName=flow_log_role_name, AssumeRolePolicyDocument=json.dumps(assume_role_policy_document) ) flow_log_role_arn = create_role_response['Role']['Arn'] print(f" Created IAM Role: {flow_log_role_arn}") # Attach policy for S3 or CloudWatch Logs if log_destination_type == 's3': policy_arn = 'arn:aws:iam::aws:policy/AmazonS3FullAccess' # Refine this to specific bucket in production elif log_destination_type == 'cloudwatch': policy_arn = 'arn:aws:iam::aws:policy/CloudWatchLogsFullAccess' # Refine this to specific log group in production else: print("Error: Invalid log_destination_type. Must be 's3' or 'cloudwatch'. Exiting.") return iam_client.attach_role_policy( RoleName=flow_log_role_name, PolicyArn=policy_arn ) print(f" Attached policy {policy_arn} to role '{flow_log_role_name}'.") time.sleep(10) # Give IAM time to propagate except Exception as e: print(f"Error with IAM Role '{flow_log_role_name}': {e}") return # 2. List all VPCs print("\n>>> Step 2: Listing all VPCs...") vpcs = ec2_client.describe_vpcs()['Vpcs'] print(f" Found {len(vpcs)} VPCs.") # 3. Enforce Flow Logs for each VPC print("\n>>> Step 3: Enforcing Flow Logs for each VPC...") for vpc in vpcs: vpc_id = vpc['VpcId'] vpc_name = 'N/A' for tag in vpc.get('Tags', []): if tag['Key'] == 'Name': vpc_name = tag['Value'] break print(f" Checking VPC '{vpc_id}' ({vpc_name})...") # Check existing Flow Logs for this VPC existing_flow_logs = ec2_client.describe_flow_logs( Filters=[ {'Name': 'resource-id', 'Values': [vpc_id]} ] )['FlowLogs'] needs_update = True flow_log_id_to_delete = None if existing_flow_logs: for fl in existing_flow_logs: if fl['TrafficType'] == traffic_type and \ fl['LogDestinationType'] == log_destination_type.upper() and \ fl['LogDestination'] == log_destination_arn: print(f" Flow Log already configured correctly for VPC '{vpc_id}'. Flow Log ID: {fl['FlowLogId']}. Skipping.") needs_update = False break else: print(f" Existing Flow Log {fl['FlowLogId']} found but configuration differs. Will be replaced.") flow_log_id_to_delete = fl['FlowLogId'] if needs_update: if not dry_run: # Delete existing if misconfigured if flow_log_id_to_delete: print(f" Deleting existing Flow Log {flow_log_id_to_delete} for VPC '{vpc_id}'...") ec2_client.delete_flow_logs(FlowLogIds=[flow_log_id_to_delete]) print(f" Deleted Flow Log {flow_log_id_to_delete}.") print(f" Creating Flow Log for VPC '{vpc_id}'...") create_fl_response = ec2_client.create_flow_logs( ResourceIds=[vpc_id], ResourceType='VPC', TrafficType=traffic_type, LogDestinationType=log_destination_type.upper(), LogDestination=log_destination_arn, DeliverLogsPermissionArn=flow_log_role_arn, TagSpecifications=[ { 'ResourceType': 'vpc-flow-log', 'Tags': [ {'Key': 'Name', 'Value': f'{vpc_id}-flow-log'} ] }, ] ) flow_log_id = create_fl_response['FlowLogIds'][0] print(f" Flow Log '{flow_log_id}' created successfully for VPC '{vpc_id}'.") else: print(f" (Dry run) Would create/update Flow Log for VPC '{vpc_id}' to send to {log_destination_arn}.") except Exception as e: print(f"Error enforcing Flow Logs for VPC '{vpc_id}': {e}") print("\nVPC Flow Log enforcement completed.") if __name__ == "__main__": parser = argparse.ArgumentParser(description="Enforce VPC Flow Logs configuration for all VPCs.") parser.add_argument("--log-destination-type", required=True, choices=['s3', 'cloudwatch'], help="Type of destination ('s3' or 'cloudwatch').") parser.add_argument("--log-destination-arn", required=True, help="ARN of the S3 bucket or CloudWatch Log Group.") parser.add_argument("--traffic-type", default="ALL", choices=['ACCEPT', 'REJECT', 'ALL'], help="The type of traffic to log (default: ALL).") parser.add_argument("--region", default="us-east-1", help="AWS region to enforce Flow Logs (default: us-east-1).") parser.add_argument("--dry-run", action="store_true", help="If set, only reports changes without applying them.") args = parser.parse_args() enforce_vpc_flow_logs( log_destination_type=args.log_destination_type, log_destination_arn=args.log_destination_arn, traffic_type=args.traffic_type, region_name=args.region, dry_run=args.dry_run )