How Cloud Data Breaches Happen in 2026 (Beginner Breakdown)

# main.tf - Vulnerable Cloud Infrastructure
terraform {
  required_version = ">= 1.0"
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.0"
    }
  }
}

provider "aws" {
  region = var.aws_region
}

# Variables
variable "aws_region" {
  description = "AWS region"
  default     = "us-east-1"
}

variable "project_name" {
  description = "Project name for resource tagging"
  default     = "cloud-security-lab"
}

# VPC for lab environment
resource "aws_vpc" "lab_vpc" {
  cidr_block           = "10.0.0.0/16"
  enable_dns_hostnames = true
  enable_dns_support   = true

  tags = {
    Name        = "${var.project_name}-vpc"
    Environment = "lab"
    Purpose     = "security-training"
  }
}

# Public subnet (vulnerable configuration)
resource "aws_subnet" "public_subnet" {
  vpc_id                  = aws_vpc.lab_vpc.id
  cidr_block              = "10.0.1.0/24"
  availability_zone       = "${var.aws_region}a"
  map_public_ip_on_launch = true

  tags = {
    Name = "${var.project_name}-public-subnet"
  }
}

# Internet Gateway
resource "aws_internet_gateway" "lab_igw" {
  vpc_id = aws_vpc.lab_vpc.id

  tags = {
    Name = "${var.project_name}-igw"
  }
}

# Route table for public subnet
resource "aws_route_table" "public_rt" {
  vpc_id = aws_vpc.lab_vpc.id

  route {
    cidr_block = "0.0.0.0/0"
    gateway_id = aws_internet_gateway.lab_igw.id
  }

  tags = {
    Name = "${var.project_name}-public-rt"
  }
}

resource "aws_route_table_association" "public_rta" {
  subnet_id      = aws_subnet.public_subnet.id
  route_table_id = aws_route_table.public_rt.id
}

# VULNERABLE: Public S3 bucket
resource "aws_s3_bucket" "vulnerable_bucket" {
  bucket = "${var.project_name}-vulnerable-${random_id.bucket_suffix.hex}"

  tags = {
    Name        = "Vulnerable Public Bucket"
    Environment = "lab"
    Purpose     = "demonstration"
  }
}

# VULNERABLE: Public access allowed
resource "aws_s3_bucket_public_access_block" "vulnerable_bucket_pab" {
  bucket = aws_s3_bucket.vulnerable_bucket.id

  block_public_acls       = false  # VULNERABLE
  block_public_policy     = false  # VULNERABLE
  ignore_public_acls      = false  # VULNERABLE
  restrict_public_buckets = false  # VULNERABLE
}

# VULNERABLE: Public read policy
resource "aws_s3_bucket_policy" "vulnerable_bucket_policy" {
  bucket = aws_s3_bucket.vulnerable_bucket.id

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Sid       = "PublicReadGetObject"
        Effect    = "Allow"
        Principal = "*"
        Action    = "s3:GetObject"
        Resource  = "${aws_s3_bucket.vulnerable_bucket.arn}/*"
      }
    ]
  })
}

# VULNERABLE: Over-permissioned IAM user
resource "aws_iam_user" "vulnerable_user" {
  name = "${var.project_name}-vulnerable-user"

  tags = {
    Name = "Vulnerable IAM User"
  }
}

# VULNERABLE: Admin access
resource "aws_iam_user_policy_attachment" "vulnerable_user_admin" {
  user       = aws_iam_user.vulnerable_user.name
  policy_arn = "arn:aws:iam::aws:policy/AdministratorAccess"
}

# VULNERABLE: Long-lived access key
resource "aws_iam_access_key" "vulnerable_user_key" {
  user = aws_iam_user.vulnerable_user.name
}

# Random suffix for bucket name
resource "random_id" "bucket_suffix" {
  byte_length = 4
}

# Outputs
output "vulnerable_bucket_name" {
  description = "Name of vulnerable S3 bucket"
  value       = aws_s3_bucket.vulnerable_bucket.id
}

output "vulnerable_user_access_key" {
  description = "Access key for vulnerable user (for attack simulation)"
  value       = aws_iam_access_key.vulnerable_user_key.id
  sensitive   = true
}

output "vulnerable_user_secret_key" {
  description = "Secret key for vulnerable user (for attack simulation)"
  value       = aws_iam_access_key.vulnerable_user_key.secret
  sensitive   = true
}

output "vpc_id" {
  description = "VPC ID"
  value       = aws_vpc.lab_vpc.id
}

# Initialize Terraform
terraform init

# Review plan
terraform plan

# Deploy infrastructure
terraform apply -auto-approve

# Save outputs
terraform output -json > outputs.json

# Verify vulnerable bucket is public
aws s3api get-bucket-acl --bucket $(terraform output -raw vulnerable_bucket_name)

#!/usr/bin/env python3
"""
Cloud Attack Simulation Script
Simulates common cloud breach scenarios
"""

import boto3
import json
import sys
from datetime import datetime

class CloudAttackSimulator:
    def __init__(self, access_key_id, secret_access_key, region='us-east-1'):
        self.session = boto3.Session(
            aws_access_key_id=access_key_id,
            aws_secret_access_key=secret_access_key,
            region_name=region
        )
        self.s3 = self.session.client('s3')
        self.iam = self.session.client('iam')
        self.ec2 = self.session.client('ec2')
        
    def simulate_s3_breach(self, bucket_name):
        """Simulate S3 bucket breach"""
        print(f"\n🔓 Simulating S3 Bucket Breach...")
        print(f"   Target: {bucket_name}")
        
        try:
            # List objects
            response = self.s3.list_objects_v2(Bucket=bucket_name)
            print(f"   ✓ Can list objects: {response.get('KeyCount', 0)} objects found")
            
            # Try to read objects
            if 'Contents' in response:
                for obj in response['Contents'][:5]:  # First 5 objects
                    try:
                        content = self.s3.get_object(Bucket=bucket_name, Key=obj['Key'])
                        print(f"   ✓ Can read: {obj['Key']} ({obj['Size']} bytes)")
                    except Exception as e:
                        print(f"   ✗ Cannot read: {obj['Key']} - {e}")
            
            # Check bucket policy
            try:
                policy = self.s3.get_bucket_policy(Bucket=bucket_name)
                print(f"   ✓ Bucket policy accessible")
                policy_json = json.loads(policy['Policy'])
                print(f"   Policy: {json.dumps(policy_json, indent=2)}")
            except Exception as e:
                print(f"   ℹ No bucket policy or cannot access")
            
            return True
        except Exception as e:
            print(f"   ✗ Attack failed: {e}")
            return False
    
    def simulate_iam_abuse(self, user_name):
        """Simulate IAM privilege abuse"""
        print(f"\n🔓 Simulating IAM Privilege Abuse...")
        print(f"   Target user: {user_name}")
        
        try:
            # List attached policies
            policies = self.iam.list_attached_user_policies(UserName=user_name)
            print(f"   ✓ User policies: {len(policies.get('AttachedPolicies', []))}")
            for policy in policies.get('AttachedPolicies', []):
                print(f"     - {policy['PolicyName']} ({policy['PolicyArn']})")
            
            # Try to create resources (test admin access)
            try:
                # Try to create an S3 bucket
                test_bucket = f"test-bucket-{datetime.now().strftime('%Y%m%d%H%M%S')}"
                self.s3.create_bucket(Bucket=test_bucket)
                print(f"   ✓ Can create resources (admin access confirmed)")
                # Cleanup
                self.s3.delete_bucket(Bucket=test_bucket)
            except Exception as e:
                print(f"   ✗ Cannot create resources: {e}")
            
            # List all users
            users = self.iam.list_users()
            print(f"   ✓ Can enumerate users: {len(users.get('Users', []))} users")
            
            return True
        except Exception as e:
            print(f"   ✗ Attack failed: {e}")
            return False
    
    def simulate_token_theft(self):
        """Simulate token/credential theft scenario"""
        print(f"\n🔓 Simulating Token Theft...")
        
        try:
            # Get current identity
            sts = self.session.client('sts')
            identity = sts.get_caller_identity()
            print(f"   ✓ Current identity: {identity.get('Arn')}")
            print(f"   ✓ Account ID: {identity.get('Account')}")
            print(f"   ✓ User ID: {identity.get('UserId')}")
            
            # Try to assume roles (if possible)
            print(f"   ℹ Attempting to enumerate assumable roles...")
            # This would require additional permissions
            
            return True
        except Exception as e:
            print(f"   ✗ Attack failed: {e}")
            return False
    
    def generate_attack_report(self, results):
        """Generate attack simulation report"""
        report = {
            "timestamp": datetime.now().isoformat(),
            "simulation_type": "cloud_breach",
            "results": results,
            "summary": {
                "total_attacks": len(results),
                "successful": sum(1 for r in results if r.get("success")),
                "failed": sum(1 for r in results if not r.get("success"))
            }
        }
        
        filename = f"attack_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
        with open(filename, 'w') as f:
            json.dump(report, f, indent=2)
        
        print(f"\n📊 Attack report saved to: {filename}")
        return report

# Example usage
if __name__ == "__main__":
    # Load credentials from Terraform output
    try:
        with open('outputs.json', 'r') as f:
            outputs = json.load(f)
        
        access_key = outputs['vulnerable_user_access_key']['value']
        secret_key = outputs['vulnerable_user_secret_key']['value']
        bucket_name = outputs['vulnerable_bucket_name']['value']
        user_name = "cloud-security-lab-vulnerable-user"
        
        simulator = CloudAttackSimulator(access_key, secret_key)
        
        results = []
        
        # Run attack simulations
        s3_result = simulator.simulate_s3_breach(bucket_name)
        results.append({"attack": "s3_breach", "success": s3_result})
        
        iam_result = simulator.simulate_iam_abuse(user_name)
        results.append({"attack": "iam_abuse", "success": iam_result})
        
        token_result = simulator.simulate_token_theft()
        results.append({"attack": "token_theft", "success": token_result})
        
        # Generate report
        simulator.generate_attack_report(results)
        
    except FileNotFoundError:
        print("Error: outputs.json not found. Run 'terraform apply' first.")
        sys.exit(1)
    except Exception as e:
        print(f"Error: {e}")
        sys.exit(1)

#!/usr/bin/env python3
"""
Cloud Security Scanner
Scans infrastructure for misconfigurations
"""

import boto3
import json
from datetime import datetime

class CloudSecurityScanner:
    def __init__(self, region='us-east-1'):
        self.session = boto3.Session(region_name=region)
        self.s3 = self.session.client('s3')
        self.iam = self.session.client('iam')
        self.config = self.session.client('config')
        
    def scan_s3_buckets(self):
        """Scan S3 buckets for misconfigurations"""
        print("\n🔍 Scanning S3 Buckets...")
        findings = []
        
        try:
            buckets = self.s3.list_buckets()
            
            for bucket in buckets.get('Buckets', []):
                bucket_name = bucket['Name']
                bucket_findings = []
                
                # Check public access block
                try:
                    pab = self.s3.get_public_access_block(Bucket=bucket_name)
                    pab_config = pab.get('PublicAccessBlockConfiguration', {})
                    
                    if not pab_config.get('BlockPublicAcls', False):
                        bucket_findings.append({
                            "severity": "HIGH",
                            "issue": "Public ACLs not blocked",
                            "recommendation": "Enable BlockPublicAcls"
                        })
                    
                    if not pab_config.get('BlockPublicPolicy', False):
                        bucket_findings.append({
                            "severity": "HIGH",
                            "issue": "Public policies not blocked",
                            "recommendation": "Enable BlockPublicPolicy"
                        })
                except:
                    bucket_findings.append({
                        "severity": "CRITICAL",
                        "issue": "No public access block configured",
                        "recommendation": "Configure public access block"
                    })
                
                # Check bucket policy
                try:
                    policy = self.s3.get_bucket_policy(Bucket=bucket_name)
                    policy_json = json.loads(policy['Policy'])
                    
                    # Check for public access in policy
                    for statement in policy_json.get('Statement', []):
                        if statement.get('Principal') == '*' or statement.get('Principal') == {'AWS': '*'}:
                            bucket_findings.append({
                                "severity": "CRITICAL",
                                "issue": "Public access allowed in bucket policy",
                                "recommendation": "Remove public access from policy"
                            })
                except:
                    pass  # No policy or cannot access
                
                if bucket_findings:
                    findings.append({
                        "resource": f"s3://{bucket_name}",
                        "findings": bucket_findings
                    })
                    print(f"   ⚠️  {bucket_name}: {len(bucket_findings)} issues found")
                else:
                    print(f"   ✓ {bucket_name}: No issues found")
        
        except Exception as e:
            print(f"   ✗ Error scanning S3: {e}")
        
        return findings
    
    def scan_iam_users(self):
        """Scan IAM users for misconfigurations"""
        print("\n🔍 Scanning IAM Users...")
        findings = []
        
        try:
            users = self.iam.list_users()
            
            for user in users.get('Users', []):
                user_name = user['UserName']
                user_findings = []
                
                # Check for admin policies
                attached_policies = self.iam.list_attached_user_policies(UserName=user_name)
                for policy in attached_policies.get('AttachedPolicies', []):
                    if 'AdministratorAccess' in policy['PolicyArn']:
                        user_findings.append({
                            "severity": "CRITICAL",
                            "issue": "User has AdministratorAccess",
                            "recommendation": "Apply principle of least privilege"
                        })
                
                # Check for access keys
                access_keys = self.iam.list_access_keys(UserName=user_name)
                if len(access_keys.get('AccessKeyMetadata', [])) > 0:
                    user_findings.append({
                        "severity": "MEDIUM",
                        "issue": "User has access keys (long-lived credentials)",
                        "recommendation": "Use IAM roles instead of access keys"
                    })
                
                if user_findings:
                    findings.append({
                        "resource": f"iam:user/{user_name}",
                        "findings": user_findings
                    })
                    print(f"   ⚠️  {user_name}: {len(user_findings)} issues found")
                else:
                    print(f"   ✓ {user_name}: No issues found")
        
        except Exception as e:
            print(f"   ✗ Error scanning IAM: {e}")
        
        return findings
    
    def generate_scan_report(self, all_findings):
        """Generate security scan report"""
        total_findings = sum(len(f['findings']) for f in all_findings)
        critical = sum(1 for f in all_findings for issue in f['findings'] if issue['severity'] == 'CRITICAL')
        high = sum(1 for f in all_findings for issue in f['findings'] if issue['severity'] == 'HIGH')
        
        report = {
            "timestamp": datetime.now().isoformat(),
            "scan_type": "cloud_security",
            "summary": {
                "total_resources_scanned": len(all_findings),
                "total_findings": total_findings,
                "critical": critical,
                "high": high,
                "medium": total_findings - critical - high
            },
            "findings": all_findings
        }
        
        filename = f"security_scan_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
        with open(filename, 'w') as f:
            json.dump(report, f, indent=2)
        
        print(f"\n📊 Security scan report saved to: {filename}")
        print(f"   Total findings: {total_findings} (Critical: {critical}, High: {high})")
        
        return report

if __name__ == "__main__":
    scanner = CloudSecurityScanner()
    
    all_findings = []
    all_findings.extend(scanner.scan_s3_buckets())
    all_findings.extend(scanner.scan_iam_users())
    
    scanner.generate_scan_report(all_findings)

#!/usr/bin/env python3
"""
Cloud Security Remediation Script
Fixes common misconfigurations
"""

import boto3
import json
import sys

class CloudRemediator:
    def __init__(self, region='us-east-1'):
        self.session = boto3.Session(region_name=region)
        self.s3 = self.session.client('s3')
        self.iam = self.session.client('iam')
    
    def remediate_s3_public_access(self, bucket_name):
        """Fix S3 bucket public access"""
        print(f"\n🔧 Remediating S3 bucket: {bucket_name}")
        
        try:
            # Enable public access block
            self.s3.put_public_access_block(
                Bucket=bucket_name,
                PublicAccessBlockConfiguration={
                    'BlockPublicAcls': True,
                    'IgnorePublicAcls': True,
                    'BlockPublicPolicy': True,
                    'RestrictPublicBuckets': True
                }
            )
            print(f"   ✓ Public access block enabled")
            
            # Remove public policy
            try:
                self.s3.delete_bucket_policy(Bucket=bucket_name)
                print(f"   ✓ Public bucket policy removed")
            except:
                print(f"   ℹ No bucket policy to remove")
            
            return True
        except Exception as e:
            print(f"   ✗ Remediation failed: {e}")
            return False
    
    def remediate_iam_least_privilege(self, user_name):
        """Apply least privilege to IAM user"""
        print(f"\n🔧 Remediating IAM user: {user_name}")
        
        try:
            # Detach admin policy
            self.iam.detach_user_policy(
                UserName=user_name,
                PolicyArn='arn:aws:iam::aws:policy/AdministratorAccess'
            )
            print(f"   ✓ AdministratorAccess policy detached")
            
            # Create limited policy (read-only S3)
            limited_policy = {
                "Version": "2012-10-17",
                "Statement": [
                    {
                        "Effect": "Allow",
                        "Action": [
                            "s3:GetObject",
                            "s3:ListBucket"
                        ],
                        "Resource": "*"
                    }
                ]
            }
            
            policy_name = f"{user_name}-limited-policy"
            self.iam.put_user_policy(
                UserName=user_name,
                PolicyName=policy_name,
                PolicyDocument=json.dumps(limited_policy)
            )
            print(f"   ✓ Limited policy attached")
            
            return True
        except Exception as e:
            print(f"   ✗ Remediation failed: {e}")
            return False
    
    def rotate_access_keys(self, user_name):
        """Rotate IAM access keys"""
        print(f"\n🔧 Rotating access keys for: {user_name}")
        
        try:
            # List existing keys
            keys = self.iam.list_access_keys(UserName=user_name)
            
            # Delete old keys
            for key in keys.get('AccessKeyMetadata', []):
                self.iam.delete_access_key(
                    UserName=user_name,
                    AccessKeyId=key['AccessKeyId']
                )
                print(f"   ✓ Deleted old key: {key['AccessKeyId']}")
            
            # Create new key
            new_key = self.iam.create_access_key(UserName=user_name)
            print(f"   ✓ Created new access key: {new_key['AccessKey']['AccessKeyId']}")
            print(f"   ⚠️  Save new secret key securely: {new_key['AccessKey']['SecretAccessKey']}")
            
            return True
        except Exception as e:
            print(f"   ✗ Key rotation failed: {e}")
            return False

if __name__ == "__main__":
    remediator = CloudRemediator()
    
    # Load from scan report or specify manually
    bucket_name = input("Enter S3 bucket name to remediate: ").strip()
    user_name = input("Enter IAM user name to remediate: ").strip()
    
    print("\n⚠️  This will modify your AWS resources. Continue? (yes/no)")
    confirm = input().strip().lower()
    
    if confirm == 'yes':
        remediator.remediate_s3_public_access(bucket_name)
        remediator.remediate_iam_least_privilege(user_name)
        remediator.rotate_access_keys(user_name)
        print("\n✅ Remediation complete!")
    else:
        print("Remediation cancelled.")

#!/bin/bash
# cleanup.sh - Destroy lab infrastructure

echo "🧹 Cleaning up cloud security lab..."

# Destroy Terraform infrastructure
terraform destroy -auto-approve

# Remove local files
rm -f outputs.json
rm -f attack_report_*.json
rm -f security_scan_*.json
rm -f terraform.tfstate*
rm -rf .terraform

echo "✅ Cleanup complete!"

aws s3api put-public-access-block --bucket my-bucket --public-access-block-configuration BlockPublicAcls=true,IgnorePublicAcls=true,BlockPublicPolicy=true,RestrictPublicBuckets=true

aws s3api delete-public-access-block --bucket my-bucket