Filter out AWS EC2 Security Groups for Access Audit for Non-Standard IP Ranges(Excluding IPs for NAT Gateways and VPNs)

import boto3 from botocore.exceptions import ClientError, NoCredentialsError, BotoCoreError # Retrieve AWS credentials creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] # List of excluded IPs (IPs used for NAT Gateways and VPNs) excluded_ips = ['10.0.0.0/8', '172.16.0.0/12', '192.168.0.0/16'] def get_ec2_client(region_name, aws_access_key_id, aws_secret_access_key): """ Initialize and return an EC2 client for the specified AWS region. """ try: return boto3.client('ec2', aws_access_key_id=aws_access_key_id, aws_secret_access_key=aws_secret_access_key, region_name=region_name) except NoCredentialsError: print(f"Credentials not available for region: {region_name}") raise except Exception as e: print(f"Unexpected error during client initialization for region {region_name}: {e}") raise def scan_specific_access(security_groups, aws_access_key_id, aws_secret_access_key, excluded_ips): """ Scan the provided security groups to identify ingress rules granting access from specific source IPs, excluding common NAT and VPN ranges. Args: security_groups (list): Each dictionary represents a security group. aws_access_key_id (str): AWS access key ID. aws_secret_access_key (str): AWS secret access key. excluded_ips (list): Source IP ranges to exclude from the results. Returns: list: Details of the specific access rules identified. """ specific_access_rules = [] clients = {} # Cache EC2 clients to avoid reinitialization for sg in security_groups: region = sg['Region'] if region not in clients: clients[region] = get_ec2_client(region, aws_access_key_id, aws_secret_access_key) try: rules_response = clients[region].describe_security_group_rules(Filters=[{'Name': 'group-id', 'Values': [sg['GroupId']]}]) for rule in rules_response['SecurityGroupRules']: # Check for ingress rules with individual IPs, excluding common NAT and VPN ranges if 'CidrIpv4' in rule and not rule['IsEgress'] and not any(ip in rule['CidrIpv4'] for ip in excluded_ips): specific_access_rules.append({ 'RuleId': rule['SecurityGroupRuleId'], 'GroupId': sg['GroupId'], 'GroupName': sg['GroupName'], 'Region': region, 'IpProtocol': rule['IpProtocol'], 'FromPort': rule.get('FromPort', 'All'), 'ToPort': rule.get('ToPort', 'All'), 'SourceIP': rule['CidrIpv4'], 'Description': rule.get('Description', 'No description provided') }) except ClientError as e: print(f"ClientError for group {sg['GroupId']} in region {region}: {e}") except BotoCoreError as e: print(f"BotoCoreError for group {sg['GroupId']} in region {region}: {e}") except Exception as e: print(f"An unexpected error occurred for group {sg['GroupId']} in region {region}: {e}") return specific_access_rules def display_specific_access(data): """ Displays details about specific access rules in a table format. """ # Initialize table with the desired structure and headers table = context.newtable() table.title = "Access Audit for Non Standard IPs Rule Details" table.num_cols = 9 # Number of columns according to headers table.num_rows = 1 # Starts with one row for headers table.has_header_row = True # Define header names based on the new structure headers = ["Rule ID", "Group ID", "Group Name", "Region", "Protocol", "From Port", "To Port", "Source IP", "Description"] # Set headers in the first row for col_num, header in enumerate(headers): table.setval(0, col_num, header) # Sort the specific access data by Group ID for better organization data.sort(key=lambda x: x["GroupId"]) # Populate the table with specific access data for row_num, rule in enumerate(data, start=1): # Starting from the second row table.num_rows += 1 # Add a row for each entry values = [ rule["RuleId"], rule["GroupId"], rule["GroupName"], str(rule["Region"]), rule["IpProtocol"], str(rule["FromPort"]), str(rule["ToPort"]), rule["SourceIP"], rule["Description"] ] for col_num, value in enumerate(values): table.setval(row_num, col_num, value) # Usage Example try: specific_access_rules = scan_specific_access(security_groups, access_key, secret_key, excluded_ips) if specific_access_rules: display_specific_access(specific_access_rules) else: print("No specific access rules found or all are filtered by exclusions.") except Exception as e: print(f"An error occurred during the scanning process: {e}")