AWS Restricted Common Ports Audit

import boto3 from botocore.exceptions import BotoCoreError, ClientError creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] # Parameters for compliance check BLOCKED_PORTS = [20, 21, 3389, 3306, 4333] # Example ports that should be restricted def fetch_security_groups(ec2_client): """ Fetch all security groups from AWS EC2, with pagination support. """ print("Fetching security groups with pagination support...") security_groups = [] paginator = ec2_client.get_paginator('describe_security_groups') page_iterator = paginator.paginate() for page in page_iterator: security_groups.extend(page['SecurityGroups']) print(f"Fetched {len(page['SecurityGroups'])} security groups in this page.") return security_groups def check_compliance(security_groups, blocked_ports=[20, 21, 3389, 3306, 4333]): """ Check compliance of security groups against the AWS Config 'restricted-common-ports' rule. This includes handling for all traffic permissions and specific blocked TCP ports. """ compliant_groups, non_compliant_groups = [], [] for group in security_groups: # Flag to keep track of compliance status is_compliant = True for permission in group['IpPermissions']: # Check for all traffic permissions (-1 protocol) if permission['IpProtocol'] == '-1': for ip_range in permission.get('IpRanges', []): if ip_range.get('CidrIp') == '0.0.0.0/0': is_compliant = False break for ipv6_range in permission.get('Ipv6Ranges', []): if ipv6_range.get('CidrIpv6') == '::/0': is_compliant = False break # Check for specific blocked TCP ports elif permission['IpProtocol'] == 'tcp': from_port = permission.get('FromPort') to_port = permission.get('ToPort') if from_port is not None and to_port is not None: for blocked_port in blocked_ports: if from_port <= blocked_port <= to_port: # Check both IPv4 and IPv6 ranges for ip_range in permission.get('IpRanges', []): if ip_range.get('CidrIp') == '0.0.0.0/0': is_compliant = False break for ipv6_range in permission.get('Ipv6Ranges', []): if ipv6_range.get('CidrIpv6') == '::/0': is_compliant = False break if not is_compliant: # Mark as non-compliant and break the loop non_compliant_groups.append(group['GroupId']) break if is_compliant: compliant_groups.append(group['GroupId']) # Ensure each security group is only counted once for non-compliance non_compliant_groups = list(set(non_compliant_groups)) return compliant_groups, non_compliant_groups def get_all_regions(ec2_client): """ Fetch all AWS regions that support EC2 service. :param ec2_client: Initialized boto3 EC2 client :return: List of region names """ regions = [] try: regions_response = ec2_client.describe_regions() regions = [region['RegionName'] for region in regions_response['Regions']] except ClientError as e: print(f"An error occurred fetching regions: {e}") except Exception as e: print(f"Unexpected error: {e}") return regions def main1(region_name=None): """ Main function to perform compliance check on AWS security groups. :param region_name: AWS region name to check security groups. If None, checks all regions. """ print("Starting compliance check for restricted common ports...") regions_to_check = [region_name] if region_name else get_all_regions(boto3.client('ec2',aws_access_key_id=access_key,aws_secret_access_key=secret_key, region_name='us-east-1')) for region in regions_to_check: print(f"Checking region: {region}") ec2_client = boto3.client('ec2', aws_access_key_id=access_key,aws_secret_access_key=secret_key,region_name=region) security_groups = fetch_security_groups(ec2_client) if not security_groups: print("No security groups found or unable to fetch security groups.") continue compliant_groups, non_compliant_groups = check_compliance(security_groups) print(f"Compliance check complete for {region}.") print(f"Compliant groups: {len(compliant_groups)}") print(f"Non-compliant groups: {len(non_compliant_groups)}") if non_compliant_groups: print("Non-compliant security group IDs:", ", ".join(non_compliant_groups)) print("-" * 60) main1(region_name)