AWS S3 Bucket Security Audits

import json cmd = "aws s3api list-buckets" output = _exe(None, cmd,cred_label=cred_label) #Parse the JSON response response_data = json.loads(output) #Extract bucket names bucket_names = [bucket["Name"] for bucket in response_data["Buckets"]] #Print the extracted bucket names: for bucket_name in bucket_names: print(bucket_name)

import boto3 from botocore.exceptions import ClientError, NoCredentialsError, BotoCoreError import json creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] def is_write_public(bucket_policy): """ Determines if the bucket policy allows public write access. """ try: policy_document = json.loads(bucket_policy['Policy']) except json.JSONDecodeError: print("Error parsing the bucket policy JSON.") return False for statement in policy_document.get('Statement', []): actions = statement.get('Action', []) actions = [actions] if isinstance(actions, str) else actions principals = statement.get('Principal', {}) # Checking if the principal is set to '*' (public access) is_public_principal = principals == '*' or principals.get('AWS') == '*' # Checking for 's3:Put*' or 's3:*' actions public_write_actions = any(action in ['s3:Put*', 's3:*'] or action.startswith('s3:Put') for action in actions) if is_public_principal and public_write_actions: return True return False def is_acl_public_write(bucket_acl): """ Determines if the bucket ACL allows public write access. """ for grant in bucket_acl['Grants']: if grant['Grantee'].get('Type') == 'Group' and grant['Grantee'].get('URI') == 'http://acs.amazonaws.com/groups/global/AllUsers': if 'WRITE' in grant['Permission']: return True return False def check_s3_buckets_public_write(): """ Checks all S3 buckets in the account to ensure they do not allow public write access. """ try: s3 = boto3.client('s3',aws_access_key_id=access_key,aws_secret_access_key=secret_key) buckets = s3.list_buckets().get('Buckets', []) if not buckets: print("No S3 buckets found in the account.") return for bucket in buckets: bucket_name = bucket['Name'] is_compliant = True # Check block public access settings try: public_access_block = s3.get_public_access_block(Bucket=bucket_name) if public_access_block['PublicAccessBlockConfiguration'].get('BlockPublicAcls', False) is False: print(f"Bucket '{bucket_name}' is non-compliant: Public Access Block allows public write.") is_compliant = False except ClientError as e: if e.response['Error']['Code'] != 'NoSuchPublicAccessBlockConfiguration': raise # Check the bucket policy try: bucket_policy = s3.get_bucket_policy(Bucket=bucket_name) if is_write_public(bucket_policy): print(f"Bucket '{bucket_name}' is non-compliant: Policy allows public write access.") is_compliant = False except ClientError as e: if e.response['Error']['Code'] != 'NoSuchBucketPolicy': raise # Check bucket ACL try: bucket_acl = s3.get_bucket_acl(Bucket=bucket_name) if is_acl_public_write(bucket_acl): print(f"Bucket '{bucket_name}' is non-compliant: ACL allows public write access.") is_compliant = False except ClientError: raise if is_compliant: print(f"Bucket '{bucket_name}' is compliant: No public write access detected.") print("Public write access check complete for all S3 buckets.") except NoCredentialsError: print("No AWS credentials found. Please configure your credentials.") except BotoCoreError as e: print(f"An error occurred accessing AWS S3 service: {e}") except Exception as e: print(f"An unexpected error occurred: {e}") # Example usage check_s3_buckets_public_write() context.skip_sub_tasks=True

import json cmd = "aws s3api list-buckets" output = _exe(None, cmd,cred_label=cred_label) #Parse the JSON response response_data = json.loads(output) #Extract bucket names bucket_names = [bucket["Name"] for bucket in response_data["Buckets"]] #Print the extracted bucket names: for bucket_name in bucket_names: print(bucket_name)

import boto3 from botocore.exceptions import ClientError, NoCredentialsError, BotoCoreError import json creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] def is_read_public(bucket_policy): """ Determines if the bucket policy allows public read access. """ try: policy_document = json.loads(bucket_policy['Policy']) except json.JSONDecodeError: print("Error parsing the bucket policy JSON.") return False for statement in policy_document.get('Statement', []): actions = statement.get('Action', []) actions = [actions] if isinstance(actions, str) else actions principals = statement.get('Principal', {}) # Checking if the principal is set to '*' (public access) is_public_principal = principals == '*' or principals.get('AWS') == '*' # Checking for 's3:Get*' or 's3:*' actions public_read_actions = any(action in ['s3:Get*', 's3:*'] or action.startswith('s3:Get') for action in actions) if is_public_principal and public_read_actions: return True return False def is_acl_public_read(bucket_acl): """ Determines if the bucket ACL allows public read access. """ for grant in bucket_acl['Grants']: if grant['Grantee'].get('Type') == 'Group' and grant['Grantee'].get('URI') == 'http://acs.amazonaws.com/groups/global/AllUsers': if 'READ' in grant['Permission']: return True return False def check_s3_buckets_public_read(): """ Checks all S3 buckets in the account to ensure they do not allow public read access. """ try: s3 = boto3.client('s3',aws_access_key_id=access_key,aws_secret_access_key=secret_key) buckets = s3.list_buckets().get('Buckets', []) if not buckets: print("No S3 buckets found in the account.") return for bucket in buckets: bucket_name = bucket['Name'] is_compliant = True # Check block public access settings try: public_access_block = s3.get_public_access_block(Bucket=bucket_name) if not public_access_block['PublicAccessBlockConfiguration'].get('BlockPublicAcls', False): print(f"Bucket '{bucket_name}' is non-compliant: Public Access Block allows public read.") is_compliant = False except ClientError as e: if e.response['Error']['Code'] != 'NoSuchPublicAccessBlockConfiguration': raise # Check the bucket policy try: bucket_policy = s3.get_bucket_policy(Bucket=bucket_name) if is_read_public(bucket_policy): print(f"Bucket '{bucket_name}' is non-compliant: Policy allows public read access.") is_compliant = False except ClientError as e: if e.response['Error']['Code'] != 'NoSuchBucketPolicy': raise # Check bucket ACL try: bucket_acl = s3.get_bucket_acl(Bucket=bucket_name) if is_acl_public_read(bucket_acl): print(f"Bucket '{bucket_name}' is non-compliant: ACL allows public read access.") is_compliant = False except ClientError: raise if is_compliant: print(f"Bucket '{bucket_name}' is compliant: No public read access detected.") print("Public read access check complete for all S3 buckets.") except NoCredentialsError: print("No AWS credentials found. Please configure your credentials.") except BotoCoreError as e: print(f"An error occurred accessing AWS S3 service: {e}") except Exception as e: print(f"An unexpected error occurred: {e}") # Example usage check_s3_buckets_public_read() context.skip_sub_tasks=True

import json cmd = "aws s3api list-buckets" output = _exe(None, cmd,cred_label=cred_label) #Parse the JSON response response_data = json.loads(output) #Extract bucket names bucket_names = [bucket["Name"] for bucket in response_data["Buckets"]] #Print the extracted bucket names: for bucket_name in bucket_names: print(bucket_name)

# Compliance Rule: s3-bucket-server-side-encryption-enabled # This rule checks each S3 bucket for two key criteria: # 1. Default Encryption: The bucket must have server-side encryption enabled by default, # using either AES-256 or AWS KMS. # 2. Policy Compliance: The bucket policy must explicitly deny put-object requests that # are not accompanied by server-side encryption using AES-256 or AWS KMS. # A bucket is considered NON_COMPLIANT if it does not have default encryption enabled. import boto3 from botocore.exceptions import ClientError, NoCredentialsError, BotoCoreError import json creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] def is_encryption_enabled(bucket_name): """ Check if the specified S3 bucket has server-side encryption enabled. """ s3 = boto3.client('s3',aws_access_key_id=access_key,aws_secret_access_key=secret_key) try: # Attempt to retrieve the bucket encryption configuration encryption = s3.get_bucket_encryption(Bucket=bucket_name) #print(encryption) # for debugging # If this call is successful, encryption is enabled return True except ClientError as e: if e.response['Error']['Code'] == 'ServerSideEncryptionConfigurationNotFoundError': # Encryption is not enabled return False else: # Other errors raise def is_policy_compliant(bucket_name): """ Check if the bucket policy explicitly denies put-object requests without server-side encryption. """ s3 = boto3.client('s3',aws_access_key_id=access_key,aws_secret_access_key=secret_key) try: policy = s3.get_bucket_policy(Bucket=bucket_name) policy_document = json.loads(policy['Policy']) #print(f"Policy for bucket '{bucket_name}': {policy_document}") # Debug statement for statement in policy_document.get('Statement', []): if statement.get('Effect') == 'Deny': actions = statement.get('Action', []) if isinstance(actions, str): actions = [actions] if any(action.startswith('s3:Put') for action in actions): conditions = statement.get('Condition', {}).get('StringEquals', {}) encryption_condition = conditions.get('s3:x-amz-server-side-encryption', None) if encryption_condition in ['AES256', 'aws:kms']: #print(f"Bucket '{bucket_name}' has a compliant policy with encryption requirement.") # Debug statement return True #print(f"Bucket '{bucket_name}' does not have a compliant policy with encryption requirement.") # Debug statement return False except ClientError as e: if e.response['Error']['Code'] == 'NoSuchBucketPolicy': #print(f"No policy for bucket '{bucket_name}'.") # Debug statement return False else: raise def check_all_buckets_for_encryption(): """ Check all S3 buckets in the account for server-side encryption and compliance with bucket policy. """ try: s3 = boto3.client('s3',aws_access_key_id=access_key,aws_secret_access_key=secret_key) buckets = s3.list_buckets().get('Buckets', []) if not buckets: print("No S3 buckets found in the account.") return for bucket in buckets: bucket_name = bucket['Name'] encrypted = is_encryption_enabled(bucket_name) policy_compliant = is_policy_compliant(bucket_name) # A bucket is considered NON_COMPLIANT if it does not have default encryption enabled. if encrypted: if policy_compliant: print(f"Bucket '{bucket_name}' is COMPLIANT with server-side encryption and has a compliant policy.") else: print(f"Bucket '{bucket_name}' is COMPLIANT with server-side encryption but does not have a compliant policy.") else: print(f"Bucket '{bucket_name}' is NON_COMPLIANT with server-side encryption.") except NoCredentialsError: print("No AWS credentials found. Please configure your credentials.") except BotoCoreError as e: print(f"An error occurred accessing AWS S3 service: {e}") except Exception as e: print(f"An unexpected error occurred: {e}") check_all_buckets_for_encryption() context.skip_sub_tasks=True

import json cmd = "aws s3api list-buckets" output = _exe(None, cmd,cred_label=cred_label) #Parse the JSON response response_data = json.loads(output) #Extract bucket names bucket_names = [bucket["Name"] for bucket in response_data["Buckets"]] #Print the extracted bucket names: for bucket_name in bucket_names: print(bucket_name)

# SOC2 Compliance Guideline: S3 Bucket Logging import boto3 from botocore.exceptions import ClientError, NoCredentialsError, BotoCoreError creds = _get_creds(cred_label)['creds'] access_key = creds['username'] secret_key = creds['password'] def is_logging_enabled(bucket_name): """ Check if logging is enabled for the specified S3 bucket. """ s3 = boto3.client('s3', aws_access_key_id=access_key,aws_secret_access_key=secret_key) try: # Attempt to retrieve the bucket logging configuration logging_config = s3.get_bucket_logging(Bucket=bucket_name) # Logging is enabled if 'LoggingEnabled' key is present in the response return 'LoggingEnabled' in logging_config except ClientError as e: print(f"Error checking logging for bucket '{bucket_name}': {e}") raise def check_all_buckets_for_logging(): """ Check all S3 buckets in the account to ensure logging is enabled. """ try: s3 = boto3.client('s3',aws_access_key_id=access_key,aws_secret_access_key=secret_key) buckets = s3.list_buckets().get('Buckets', []) if not buckets: print("No S3 buckets found in the account.") return for bucket in buckets: bucket_name = bucket['Name'] if is_logging_enabled(bucket_name): print(f"Bucket '{bucket_name}' is COMPLIANT with logging enabled.") else: print(f"Bucket '{bucket_name}' is NON_COMPLIANT with logging disabled.") except NoCredentialsError: print("No AWS credentials found. Please configure your credentials.") except BotoCoreError as e: print(f"An error occurred accessing AWS S3 service: {e}") except Exception as e: print(f"An unexpected error occurred: {e}") check_all_buckets_for_logging() context.skip_sub_tasks=True