AWS S3 Bucket Server-Side Encryption Audit: SOC2 Compliance

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