Scan AWS EC2 Security Groups for Access Audit

import boto3 from botocore.exceptions import NoCredentialsError, ClientError creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] def get_regions(ec2_client): return [region['RegionName'] for region in ec2_client.describe_regions()['Regions']] def list_security_groups(ec2_client, region=None): paginator = ec2_client.get_paginator('describe_security_groups') security_groups = [] try: for page in paginator.paginate(): for sg in page['SecurityGroups']: # Fetching associated instances for each security group associated_instances = [] reservations = ec2_client.describe_instances(Filters=[{'Name': 'instance.group-id', 'Values': [sg['GroupId']]}]) for reservation in reservations['Reservations']: for instance in reservation['Instances']: associated_instances.append(instance['InstanceId']) # Fetching tags for each security group tags = {tag['Key']: tag['Value'] for tag in sg.get('Tags', [])} security_groups.append({ 'GroupId': sg['GroupId'], 'GroupName': sg.get('GroupName', 'N/A'), 'Description': sg.get('Description', 'N/A'), 'VpcId': sg.get('VpcId', 'N/A'), 'Region': region, 'AssociatedInstances': associated_instances, 'Tags': tags }) except ClientError as e: print(f"Error retrieving security groups in {region}: {e}") return security_groups def list_all_security_groups(region_name=None): """ List all security groups in a specified region or in all regions if no region is specified. Args: region_name (str, optional): AWS region name. Lists security groups in all regions if None. Defaults to None. """ # Initialize client for the default region to fetch regions if needed ec2_client = boto3.client('ec2', aws_access_key_id=access_key, aws_secret_access_key=secret_key, region_name='us-east-1') if region_name: #print(f"Listing security groups in region: {region_name}") regions = [region_name] else: #print("No specific region provided. Listing security groups in all regions.") regions = get_regions(ec2_client) all_security_groups = [] for region in regions: #print(f"Processing region: {region}") regional_client = boto3.client('ec2', aws_access_key_id=access_key, aws_secret_access_key=secret_key, region_name=region) all_security_groups.extend(list_security_groups(regional_client, region)) return all_security_groups def display_security_groups(security_groups): # Initialize table with the desired structure and headers table = context.newtable() table.title = "Security Group Details" table.num_cols = 7 # 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 = ["Region", "GroupId", "GroupName", "Description", "VpcId", "Associated Instances", "Tags"] # Set headers in the first row for col_num, header in enumerate(headers): table.setval(0, col_num, header) # Sort the security group data by Region for better organization security_groups.sort(key=lambda x: x['Region']) # Populate the table with security group data for row_num, sg in enumerate(security_groups, start=1): # Starting from the second row table.num_rows += 1 # Add a row for each security group values = [ sg['Region'], sg['GroupId'], sg['GroupName'], sg['Description'], sg['VpcId'], ', '.join(sg['AssociatedInstances']), ', '.join([f"{k}: {v}" for k, v in sg['Tags'].items()]) ] for col_num, value in enumerate(values): table.setval(row_num, col_num, value) # Example usage try: #region_name = None # Set to None to list security groups for all available regions or specify a region security_groups = list_all_security_groups(region_name) display_security_groups(security_groups) except NoCredentialsError: print("Error: AWS credentials not available. Please configure them.") except ClientError as e: print(f"AWS Client error: {e}") except Exception as e: print(f"An unexpected error occurred: {e}")

import boto3 from botocore.exceptions import ClientError, NoCredentialsError, BotoCoreError creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] 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=access_key,aws_secret_access_key=secret_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_open_ports(security_groups, aws_access_key_id, aws_secret_access_key): """ Scan the provided security groups to find open ports accessible from any IP (0.0.0.0/0). Args: security_groups (list): A list of dictionaries, each representing a security group. aws_access_key_id (str): AWS access key ID. aws_secret_access_key (str): AWS secret access key. Returns: list: A list of dictionaries detailing the open ports, including the region of each security group. """ open_ports = [] clients = {} # Cache to store EC2 clients for each region 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: # Fetch the detailed info of each security group using the regional client response = clients[region].describe_security_groups(GroupIds=[sg['GroupId']]) for group in response['SecurityGroups']: for perm in group['IpPermissions']: # Check if rule allows all IP addresses for ip_range in perm['IpRanges']: if ip_range['CidrIp'] == '0.0.0.0/0': # Collect information about the open port and protocol open_ports.append({ 'GroupId': sg['GroupId'], 'GroupName': sg['GroupName'], 'Region': region, # Including the region in the details 'IpProtocol': perm['IpProtocol'], 'FromPort': perm.get('FromPort', 'All'), 'ToPort': perm.get('ToPort', 'All'), 'Description': ip_range.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 open_ports def display_open_ports(data): """ Displays details about open ports in a table format. Args: data (list): A list of dictionaries containing open port details. """ # Initialize table with the desired structure and headers table = context.newtable() table.title = "Open Ports Details" table.num_cols = 7 # 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 = ["Group ID", "Group Name", "Region", "Protocol", "From Port", "To Port", "Description"] # Set headers in the first row for col_num, header in enumerate(headers): table.setval(0, col_num, header) # Sort the open port data by Group ID for better organization data.sort(key=lambda x: x["GroupId"]) # Populate the table with open port data for row_num, port in enumerate(data, start=1): # Starting from the second row table.num_rows += 1 # Add a row for each entry values = [ port["GroupId"], port["GroupName"], str(port["Region"]), port["IpProtocol"], str(port["FromPort"]), str(port["ToPort"]), port["Description"] ] for col_num, value in enumerate(values): table.setval(row_num, col_num, value) # Example usage try: open_ports = scan_open_ports(security_groups, access_key, secret_key) if open_ports: table = display_open_ports(open_ports) else: print("No ports open to all found.") except Exception as e: print(f"An error occurred during the scanning process: {e}")

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}")