Check which buckets allow AWS S3 Bucket Public Write Access

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