Cloud Security Explained — A Beginner's Guide to Securing Cloud Infrastructure

# iam_audit.py — Audit IAM policies for security issues
import json

class IAMAuditor:
    """Audit IAM policies for security best practices."""

    def __init__(self):
        self.findings = []

    def check_wildcard_action(self, policy: dict, resource_name: str):
        """Flag policies that allow all actions ('*')."""
        statements = policy.get("Statement", [])
        for stmt in statements:
            effect = stmt.get("Effect", "Deny")
            actions = stmt.get("Action", [])
            if isinstance(actions, str):
                actions = [actions]

            if effect == "Allow" and "*" in actions:
                self.findings.append({
                    "resource": resource_name,
                    "issue": "Wildcard action (*) in Allow policy",
                    "severity": "HIGH",
                    "recommendation": "Replace * with specific actions needed"
                })

    def check_wildcard_principal(self, policy: dict, resource_name: str):
        """Flag policies that allow access to anyone (* Principal)."""
        statements = policy.get("Statement", [])
        for stmt in statements:
            principal = stmt.get("Principal", {})
            if isinstance(principal, str) and principal == "*":
                self.findings.append({
                    "resource": resource_name,
                    "issue": "Wildcard principal (*) — publicly accessible",
                    "severity": "CRITICAL",
                    "recommendation": "Restrict to specific ARNs or users"
                })

    def audit_s3_bucket_policy(self, bucket_name: str, policy: dict):
        """Audit S3 bucket policy for common misconfigurations."""
        self.check_wildcard_principal(policy, f"s3://{bucket_name}")
        self.check_wildcard_action(policy, f"s3://{bucket_name}")

    def report(self) -> str:
        """Generate audit report."""
        if not self.findings:
            return "No IAM issues found."
        report = "=== IAM Security Audit Report ===\n"
        for f in sorted(self.findings, key=lambda x: x["severity"]):
            report += f"[{f['severity']}] {f['resource']}: {f['issue']}\n"
            report += f"  Fix: {f['recommendation']}\n\n"
        return report

# Example: Auditing an overly permissive S3 policy
auditor = IAMAuditor()
bad_policy = {
    "Version": "2012-10-17",
    "Statement": [{
        "Effect": "Allow",
        "Principal": "*",
        "Action": "s3:*",
        "Resource": "arn:aws:s3:::my-company-data/*"
    }]
}
auditor.audit_s3_bucket_policy("my-company-data", bad_policy)
print(auditor.report())

# encryption_demo.py — Cloud encryption patterns
from cryptography.fernet import Fernet
import base64
import os

class CloudEncryption:
    """Demonstrates encryption patterns used in cloud services."""

    @staticmethod
    def generate_key() -> bytes:
        """Generate an encryption key (simulates KMS key generation)."""
        return Fernet.generate_key()

    @staticmethod
    def encrypt_file(data: bytes, key: bytes) -> bytes:
        """Encrypt data using symmetric encryption (AES-256)."""
        f = Fernet(key)
        return f.encrypt(data)

    @staticmethod
    def decrypt_file(encrypted: bytes, key: bytes) -> bytes:
        """Decrypt data."""
        f = Fernet(key)
        return f.decrypt(encrypted)

# Example: S3 Server-Side Encryption (simulated)
print("=== Cloud Encryption Patterns ===")

# 1. SSE-S3 (Amazon S3 managed keys)
sse_s3 = "AES-256 encryption, keys managed by AWS"
print(f"SSE-S3:  {sse_s3}")

# 2. SSE-KMS (AWS KMS managed keys)
sse_kms = "AES-256 encryption, keys managed via AWS KMS with rotation"
print(f"SSE-KMS: {sse_kms}")

# 3. SSE-C (Customer provided keys)
sse_c = "AES-256 encryption, customer manages keys outside AWS"
print(f"SSE-C:   {sse_c}")

# 4. Client-Side Encryption
key = CloudEncryption.generate_key()
data = b"Sensitive customer data"
encrypted = CloudEncryption.encrypt_file(data, key)
decrypted = CloudEncryption.decrypt_file(encrypted, key)
print(f"\nClient-Side Encryption:")
print(f"  Original:  {data.decode()}")
print(f"  Encrypted: {base64.b64encode(encrypted).decode()[:32]}...")
print(f"  Decrypted: {decrypted.decode()}")

# Example: Using ScoutSuite (open-source CSPM)
# Scans AWS/Azure/GCP for security misconfigurations

# Install ScoutSuite
pip install scoutsuite

# Run against AWS
scout aws --report-dir ./reports

# Key things it checks:
# - S3 buckets publicly accessible?
# - Security groups too permissive (0.0.0.0/0 on SSH)?
# - CloudTrail enabled in all regions?
# - IAM roles with excessive permissions?
# - Encryption enabled on RDS/EBS/S3?

# AWS WAF rules to enable
# - SQL injection prevention
# - XSS prevention
# - Rate limiting (100 req/s per IP)
# - IP reputation lists
# - Geographic blocking (if needed)
# - Block known bad user agents

# Terraform check — S3 bucket must have encryption
check "s3_encryption_required" {
  assert {
    condition = aws_s3_bucket.my_bucket.server_side_encryption_configuration != null
    error_message = "S3 bucket must have server-side encryption enabled"
  }
}

# cloud_security_scanner.py
# Check basic cloud security configurations
import json

class CloudSecurityScanner:
    """Simple cloud security configuration scanner."""

    def __init__(self):
        self.checks = []

    def check_s3_public_access(self, bucket_configs: list[dict]) -> list[dict]:
        """Check if any S3 buckets allow public access."""
        findings = []
        for bucket in bucket_configs:
            if bucket.get("public", False):
                findings.append({
                    "resource": f"s3://{bucket['name']}",
                    "check": "S3 Block Public Access",
                    "status": "FAIL",
                    "severity": "CRITICAL",
                    "fix": "Enable S3 Block Public Access at account level"
                })
        return findings

    def check_security_group_open(self, sg_configs: list[dict]) -> list[dict]:
        """Check for security groups open to 0.0.0.0/0 on sensitive ports."""
        findings = []
        sensitive_ports = [22, 3389, 3306, 5432, 6379, 27017]

        for sg in sg_configs:
            for rule in sg.get("inbound_rules", []):
                if rule.get("cidr") == "0.0.0.0/0":
                    port = rule.get("port")
                    if port in sensitive_ports:
                        findings.append({
                            "resource": f"sg-{sg.get('group_id', 'unknown')}",
                            "check": f"Port {port} open to 0.0.0.0/0",
                            "status": "FAIL",
                            "severity": "HIGH",
                            "fix": f"Restrict port {port} to specific IP ranges"
                        })
        return findings

    def run_scan(self, config: dict) -> dict:
        """Run all security checks."""
        all_findings = []
        all_findings.extend(self.check_s3_public_access(config.get("buckets", [])))
        all_findings.extend(self.check_security_group_open(config.get("security_groups", [])))

        return {
            "total_checks": len(all_findings),
            "failed": len([f for f in all_findings if f["status"] == "FAIL"]),
            "findings": sorted(all_findings, key=lambda x: x["severity"])
        }

# Example scan
scanner = CloudSecurityScanner()
config = {
    "buckets": [
        {"name": "my-public-data", "public": True},
        {"name": "my-private-data", "public": False}
    ],
    "security_groups": [
        {"group_id": "sg-123", "inbound_rules": [
            {"port": 22, "cidr": "0.0.0.0/0"},
            {"port": 80, "cidr": "0.0.0.0/0"}
        ]}
    ]
}
result = scanner.run_scan(config)
print(json.dumps(result, indent=2))