S3 Bucket Security Audit — Public Access, Versioning & Encryption

import boto3
import json
from botocore.exceptions import ClientError


def audit_s3_buckets(region: str = "us-east-1") -> list[dict]:
    """
    s3.list_buckets() returns ALL buckets in the account (globally scoped).
    Unlike most AWS services, S3 buckets don't require a region filter —
    list_buckets() always returns everything regardless of client region.

    We then make per-bucket API calls for each compliance check.
    """
    s3 = boto3.client("s3", region_name=region)
    report = []

    buckets = s3.list_buckets()["Buckets"]
    print(f"Auditing {len(buckets)} S3 buckets...\n")

    for bucket in buckets:
        name = bucket["Name"]
        issues: list[str] = []

        # ── Check 1: Public Access Block ─────────────────────────
        # get_public_access_block() returns four boolean settings:
        # - BlockPublicAcls:       Block new public ACLs
        # - IgnorePublicAcls:      Ignore existing public ACLs
        # - BlockPublicPolicy:     Block new public bucket policies
        # - RestrictPublicBuckets: Restrict access for public-policy buckets
        # ALL four must be True for the bucket to be fully protected.
        try:
            pab = s3.get_public_access_block(Bucket=name)[
                "PublicAccessBlockConfiguration"
            ]
            if not all([
                pab.get("BlockPublicAcls"),
                pab.get("IgnorePublicAcls"),
                pab.get("BlockPublicPolicy"),
                pab.get("RestrictPublicBuckets"),
            ]):
                issues.append("PUBLIC_ACCESS_NOT_FULLY_BLOCKED")
        except ClientError as e:
            if "NoSuchPublicAccessBlockConfiguration" in str(e):
                issues.append("NO_PUBLIC_ACCESS_BLOCK")
            else:
                issues.append(f"PAB_CHECK_ERROR")

        # ── Check 2: Bucket Policy — Public Principal ─────────────
        # get_bucket_policy() returns the policy as a JSON string.
        # We parse it and look for statements where:
        #   Effect = "Allow"  AND  Principal = "*" or {"AWS": "*"}
        # These grant public (anonymous) access to the bucket.
        try:
            raw_policy = s3.get_bucket_policy(Bucket=name)["Policy"]
            policy = json.loads(raw_policy)
            for stmt in policy.get("Statement", []):
                principal = stmt.get("Principal")
                is_public = principal in ("*", {"AWS": "*"})
                if stmt.get("Effect") == "Allow" and is_public:
                    issues.append("PUBLIC_BUCKET_POLICY")
                    break   # One public statement is enough to flag it
        except ClientError as e:
            if "NoSuchBucketPolicy" not in str(e):
                # Real error (e.g., AccessDenied) — note it
                issues.append("POLICY_CHECK_ERROR")
            # NoSuchBucketPolicy means no policy exists — not necessarily bad

        # ── Check 3: Versioning ───────────────────────────────────
        # get_bucket_versioning() can return:
        #   {"Status": "Enabled"}   — versioning on
        #   {"Status": "Suspended"} — was on, now paused
        #   {}                      — never enabled
        # Only "Enabled" is compliant.
        try:
            versioning = s3.get_bucket_versioning(Bucket=name)
            if versioning.get("Status") != "Enabled":
                issues.append("VERSIONING_DISABLED")
        except ClientError:
            issues.append("VERSIONING_CHECK_FAILED")

        # ── Check 4: Default Encryption ───────────────────────────
        # get_bucket_encryption() raises ServerSideEncryptionConfigurationNotFoundError
        # if no default encryption is configured.
        # Valid algorithms: "AES256" (SSE-S3) or "aws:kms" (SSE-KMS).
        try:
            enc = s3.get_bucket_encryption(Bucket=name)
            rules = enc["ServerSideEncryptionConfiguration"]["Rules"]
            algo = rules[0]["ApplyServerSideEncryptionByDefault"]["SSEAlgorithm"]
            if algo not in ("aws:kms", "AES256"):
                issues.append("WEAK_ENCRYPTION")
        except ClientError as e:
            if "ServerSideEncryptionConfigurationNotFoundError" in str(e):
                issues.append("ENCRYPTION_DISABLED")

        # ── Check 5: Access Logging ───────────────────────────────
        # get_bucket_logging() returns {"LoggingEnabled": {...}} if logging
        # is configured, or just {} if disabled.
        # Logging is essential for security investigations and audits.
        try:
            logging_cfg = s3.get_bucket_logging(Bucket=name)
            if "LoggingEnabled" not in logging_cfg:
                issues.append("ACCESS_LOGGING_DISABLED")
        except ClientError:
            issues.append("LOGGING_CHECK_FAILED")

        # ── Record result ─────────────────────────────────────────
        compliant = len(issues) == 0
        report.append({
            "bucket":    name,
            "issues":    issues,
            "compliant": compliant,
        })

        status = "✅" if compliant else "❌"
        issue_str = ", ".join(issues) if issues else "All checks passed"
        print(f"{status} {name}: {issue_str}")

    # ── Summary ───────────────────────────────────────────────────
    non_compliant = [r for r in report if not r["compliant"]]
    print(f"\n{'='*60}")
    print(f"Total Buckets:    {len(report)}")
    print(f"Compliant:        {len(report) - len(non_compliant)}")
    print(f"Non-Compliant:    {len(non_compliant)}")

    if non_compliant:
        print("\nTop Issues:")
        from collections import Counter
        all_issues = [i for r in non_compliant for i in r["issues"]]
        for issue, count in Counter(all_issues).most_common():
            print(f"  {issue}: {count} bucket(s)")

    return report


def save_report(report: list[dict], output: str = "s3_audit_report.json") -> None:
    with open(output, "w") as f:
        json.dump(report, f, indent=2)
    print(f"\nFull report saved to {output}")


def auto_fix_public_access(bucket_name: str, dry_run: bool = True) -> None:
    """
    put_public_access_block() with all four settings = True
    is the AWS-recommended way to make a bucket fully private.
    This is a non-destructive operation — it doesn't delete any objects.
    """
    s3 = boto3.client("s3")
    if dry_run:
        print(f"[DRY-RUN] Would enable Public Access Block on: {bucket_name}")
        return

    s3.put_public_access_block(
        Bucket=bucket_name,
        PublicAccessBlockConfiguration={
            "BlockPublicAcls":       True,
            "IgnorePublicAcls":      True,
            "BlockPublicPolicy":     True,
            "RestrictPublicBuckets": True,
        },
    )
    print(f"✅ Public Access Block enabled on: {bucket_name}")


if __name__ == "__main__":
    report = audit_s3_buckets()
    save_report(report)

    # Auto-fix any bucket with PUBLIC_ACCESS_NOT_FULLY_BLOCKED
    for r in report:
        if "PUBLIC_ACCESS_NOT_FULLY_BLOCKED" in r["issues"]:
            auto_fix_public_access(r["bucket"], dry_run=True)   # Set dry_run=False to apply

s3 = boto3.client("s3", region_name=region)
buckets = s3.list_buckets()["Buckets"]

pab = s3.get_public_access_block(Bucket=name)["PublicAccessBlockConfiguration"]
if not all([
    pab.get("BlockPublicAcls"),
    pab.get("IgnorePublicAcls"),
    pab.get("BlockPublicPolicy"),
    pab.get("RestrictPublicBuckets"),
]):
    issues.append("PUBLIC_ACCESS_NOT_FULLY_BLOCKED")

raw_policy = s3.get_bucket_policy(Bucket=name)["Policy"]
policy = json.loads(raw_policy)
for stmt in policy.get("Statement", []):
    principal = stmt.get("Principal")
    is_public = principal in ("*", {"AWS": "*"})
    if stmt.get("Effect") == "Allow" and is_public:
        issues.append("PUBLIC_BUCKET_POLICY")
        break

versioning = s3.get_bucket_versioning(Bucket=name)
if versioning.get("Status") != "Enabled":
    issues.append("VERSIONING_DISABLED")

enc = s3.get_bucket_encryption(Bucket=name)
rules = enc["ServerSideEncryptionConfiguration"]["Rules"]
algo = rules[0]["ApplyServerSideEncryptionByDefault"]["SSEAlgorithm"]
if algo not in ("aws:kms", "AES256"):
    issues.append("WEAK_ENCRYPTION")

logging_cfg = s3.get_bucket_logging(Bucket=name)
if "LoggingEnabled" not in logging_cfg:
    issues.append("ACCESS_LOGGING_DISABLED")

non_compliant = [r for r in report if not r["compliant"]]
from collections import Counter
all_issues = [i for r in non_compliant for i in r["issues"]]
for issue, count in Counter(all_issues).most_common():
    print(f"  {issue}: {count} bucket(s)")

def auto_fix_public_access(bucket_name: str, dry_run: bool = True) -> None:
    s3.put_public_access_block(
        Bucket=bucket_name,
        PublicAccessBlockConfiguration={
            "BlockPublicAcls": True, "IgnorePublicAcls": True,
            "BlockPublicPolicy": True, "RestrictPublicBuckets": True,
        },
    )