Basic PoC script to scan for public S3 buckets using an authenticated profile

s3_fuzz.py

import subprocess
import itertools
import argparse
from concurrent.futures import ThreadPoolExecutor

# Function to generate bucket name variations
def generate_bucket_names(keyword, fuzz_words):
    variations = []
    for fuzz_word in fuzz_words:
        variations.append(f"{keyword}{fuzz_word}")
        variations.append(f"{keyword}-{fuzz_word}")
        variations.append(f"{fuzz_word}{keyword}")
        variations.append(f"{fuzz_word}-{keyword}")
    return variations

# Function to scan a single bucket
def scan_bucket(bucket_name, profile):
    command = ["aws", "s3", "ls", f"s3://{bucket_name}", "--profile", profile]
    try:
        print(f"Scanning bucket: {bucket_name}")
        result = subprocess.run(command, capture_output=True, text=True)

        if result.returncode == 0:
            print(f"[SUCCESS] Bucket found: {bucket_name}")
            print(result.stdout)
        else:
            print(f"[FAIL] {bucket_name} does not exist or is not accessible.")
    except Exception as e:
        print(f"Error running command for bucket {bucket_name}: {e}")

# Function to scan buckets
def scan_buckets(keyword, profile, fuzz_file):
    try:
        # Read fuzzing words from the file
        with open(fuzz_file, "r") as file:
            fuzz_words = [line.strip() for line in file.readlines()]

        # Try the keyword alone first
        print("Starting scan with the keyword alone...")
        scan_bucket(keyword, profile)

        # Generate bucket name variations
        bucket_names = generate_bucket_names(keyword, fuzz_words)

        # Use ThreadPoolExecutor for multi-threading
        print("Starting scan with bucket name variations...")
        with ThreadPoolExecutor(max_workers=10) as executor:
            for bucket_name in bucket_names:
                executor.submit(scan_bucket, bucket_name, profile)

    except FileNotFoundError:
        print(f"Fuzzing file {fuzz_file} not found.")
    except Exception as e:
        print(f"Error: {e}")

# Main function to run the tool
def main():
    parser = argparse.ArgumentParser(description="Basic AWS S3 Bucket Scanning Tool")
    parser.add_argument("--keyword", required=True, help="Keyword for bucket scanning")
    parser.add_argument("--profile", required=True, help="AWS CLI profile to use")
    parser.add_argument("--fuzz_file", default="enum_tools/fuzz.txt", help="Path to the fuzzing wordlist file")

    args = parser.parse_args()

    # Start scanning
    scan_buckets(args.keyword, args.profile, args.fuzz_file)

if __name__ == "__main__":
    main()