Multi-Region AWS CloudTrail Compliance Verification: SOC2 Compliance

import boto3 creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] # Set region to None for all regions, or specify a valid AWS region string for a specific region #target_region = None target_region = target_region if target_region else None try: # List all available AWS regions ec2_client = boto3.client('ec2',aws_access_key_id=access_key,aws_secret_access_key=secret_key,region_name='us-east-1') all_regions = [region['RegionName'] for region in ec2_client.describe_regions()['Regions']] except Exception as e: print(f"ERROR: An error occurred while listing AWS regions: {e}") all_regions = [] # Get trails for all regions or a specific region regions_to_check = all_regions if target_region is None else [target_region] all_trails = [] for region in regions_to_check: try: # List all trails in AWS CloudTrail for each region cloudtrail_client = boto3.client('cloudtrail', aws_access_key_id=access_key,aws_secret_access_key=secret_key,region_name=region) response = cloudtrail_client.describe_trails(includeShadowTrails=False) trails_in_region = response['trailList'] if not trails_in_region: print(f"INFO: No trails found in region {region}.") else: all_trails.extend(trails_in_region) except Exception as e: # Handle exceptions thrown while listing trails for a region print(f"ERROR: An error occurred while listing trails in region {region}: {e}") # Print all trails if not all_trails: print("INFO: No trails found in all specified regions.") else: try: #print(all_trails) # for downstream task for trail in all_trails: print(f"Trail Name: {trail['Name']}, Trail ARN: {trail['TrailARN']}, Home Region: {trail['HomeRegion']}") except KeyError as ke: print(f"ERROR: Missing key {ke} in trail information: {trail}") except Exception as e: print(f"ERROR: An error occurred while printing trail information: {e}") print(f"SUMMARY: Processed {len(regions_to_check)} regions and found a total of {len(all_trails)} trails.")

# Multi-Region CloudTrail Compliance Verification: SOC2 Guideline import boto3 from botocore.exceptions import ClientError, NoCredentialsError, PartialCredentialsError creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] def list_available_regions(service_name): """List all available regions for a given AWS service.""" ec2 = boto3.client('ec2',aws_access_key_id=access_key,aws_secret_access_key=secret_key,region_name='us-east-1') regions = [region['RegionName'] for region in ec2.describe_regions()['Regions']] return regions def check_trails_in_region(region_name, s3_bucket_name, sns_topic_arn, cloudwatch_log_group_arn, include_management_events, read_write_type): """Check CloudTrail trails in a specific region and return details of compliant and non-compliant trails.""" non_compliant_trails = [] compliant_trail_details = None try: cloudtrail_client = boto3.client('cloudtrail',aws_access_key_id=access_key,aws_secret_access_key=secret_key, region_name=region_name) trails = cloudtrail_client.describe_trails(includeShadowTrails=True)['trailList'] for trail in trails: if trail['IsMultiRegionTrail']: try: trail_config_response = cloudtrail_client.get_trail(Name=trail['TrailARN']) trail_config = trail_config_response.get('Trail', {}) trail_status_response = cloudtrail_client.get_trail_status(Name=trail['TrailARN']) is_logging = trail_status_response.get('IsLogging', False) except ClientError as e: print(f"Error in {region_name}: {e}") continue settings_match = ( is_logging and trail_config.get('S3BucketName') == s3_bucket_name and trail_config.get('SnsTopicARN') == sns_topic_arn and ('CloudWatchLogsLogGroupArn' not in trail_config or trail_config.get('CloudWatchLogsLogGroupArn') == cloudwatch_log_group_arn) and trail_config.get('IncludeGlobalServiceEvents') == include_management_events and trail_config.get('IsMultiRegionTrail', False) is True ) if settings_match: compliant_trail_details = { 'Region': region_name, 'Name': trail_config.get('Name'), 'S3BucketName': trail_config.get('S3BucketName'), 'SnsTopicARN': trail_config.get('SnsTopicARN'), 'CloudWatchLogsLogGroupArn': trail_config.get('CloudWatchLogsLogGroupArn'), 'IncludeManagementEvents': trail_config.get('IncludeGlobalServiceEvents'), 'IsMultiRegionTrail': trail_config.get('IsMultiRegionTrail') } return True, compliant_trail_details, non_compliant_trails else: non_compliant_trails.append(trail['Name']) return False, compliant_trail_details, non_compliant_trails except ClientError as e: print(f"AWS client error in region {region_name}: {e}") return False, compliant_trail_details, non_compliant_trails except Exception as e: print(f"An unexpected error occurred in region {region_name}: {e}") return False, compliant_trail_details, non_compliant_trails def check_cloudtrail_compliance(s3_bucket_name, sns_topic_arn, cloudwatch_log_group_arn, include_management_events, read_write_type): try: regions = list_available_regions('cloudtrail') compliant_in_any_region = False all_non_compliant_trails = {} compliant_trail_details = None for region in regions: compliant, details, non_compliant_trails = check_trails_in_region(region, s3_bucket_name, sns_topic_arn, cloudwatch_log_group_arn, include_management_events, read_write_type) all_non_compliant_trails[region] = non_compliant_trails if compliant: compliant_trail_details = details compliant_in_any_region = True break if compliant_in_any_region: print("Compliant Trail Found:") for key, value in compliant_trail_details.items(): print(f" {key}: {value}") else: print("Summary of Non-Compliant Trails by Region:") for region, trails in all_non_compliant_trails.items(): if trails: print(f" Region: {region}, Non-Compliant Trails: {', '.join(trails)}") else: print(f" Region: {region} has no non-compliant multi-region trails.") return compliant_in_any_region except NoCredentialsError: print("No AWS credentials found. Please configure your credentials.") return False except PartialCredentialsError: print("Incomplete AWS credentials. Please check your configuration.") return False except Exception as e: print(f"An unexpected error occurred: {e}") return False #s3_bucket_name='aws-cloudtrail-logs-355237452254-d5db7269' #sns_topic_arn='arn:aws:sns:ap-south-1:355237452254:aws-cloudtrail-logs-355237452254-0ac1f096' #cloudwatch_log_group_arn='arn:aws:logs:ap-south-1:355237452254:log-group:aws-cloudtrail-logs-355237452254-fc0d6f36:*' #include_management_events=True #read_write_type='ALL' #Type of events to record. Valid values are ReadOnly, WriteOnly and ALL. compliant = check_cloudtrail_compliance( s3_bucket_name, sns_topic_arn, cloudwatch_log_group_arn, include_management_events, read_write_type ) if compliant: print("\nAt least one compliant multi-region CloudTrail exists.") else: print("\nNo compliant multi-region CloudTrails found matching the specified criteria.") context.proceed = False