Cloud Security Guide — Shared Responsibility, IAM, KMS, and Zero Trust

# iam_policy_simulator.py
# Simulate IAM policy evaluation
import json

class IAMPolicyEngine:
    def __init__(self):
        self.policies = []

    def add_policy(self, name, effect, actions, resources, conditions=None):
        self.policies.append({
            "name": name,
            "effect": effect,
            "actions": actions,
            "resources": resources,
            "conditions": conditions or {},
        })

    def evaluate(self, principal, action, resource, context=None):
        """Evaluate if an action is allowed."""
        matching = []
        for policy in self.policies:
            if action in policy["actions"] or "*" in policy["actions"]:
                if any(resource.startswith(r) or r == "*" for r in policy["resources"]):
                    matching.append(policy)

        # Deny takes precedence over Allow
        for policy in matching:
            if policy["effect"] == "Deny":
                return "DENY"
        for policy in matching:
            if policy["effect"] == "Allow":
                return "ALLOW"
        return "DENY (implicit)"

print("=== IAM Policy Evaluation ===\n")

engine = IAMPolicyEngine()
engine.add_policy("AdminAccess", "Allow", ["*"], ["*"])
engine.add_policy("S3ReadOnly", "Allow", ["s3:Get*", "s3:List*"], ["arn:aws:s3:::dodatech-data"])
engine.add_policy("ExplicitDeny", "Deny", ["s3:Delete*"], ["arn:aws:s3:::*"])

scenarios = [
    ("admin@dodatech", "s3:GetObject", "arn:aws:s3:::dodatech-data/report.pdf"),
    ("analyst@dodatech", "s3:GetObject", "arn:aws:s3:::dodatech-data/report.pdf"),
    ("analyst@dodatech", "s3:DeleteObject", "arn:aws:s3:::dodatech-data/report.pdf"),
    ("intern@dodatech", "s3:PutObject", "arn:aws:s3:::other-bucket/file.txt"),
]

print(f"{'Principal':<20} {'Action':<20} {'Resource':<45} {'Result'}")
print("-" * 100)
for principal, action, resource in scenarios:
    result = engine.evaluate(principal, action, resource)
    print(f"{principal:<20} {action:<20} {resource:<45} {result}")

# kms_demo.py
# Simulate KMS envelope encryption
import os
import json
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2

class KMS:
    """Simulated KMS for learning (simplified)."""
    def __init__(self):
        self.master_keys = {}
        self.data_keys = {}

    def create_key(self, key_id, description=""):
        self.master_keys[key_id] = {
            "id": key_id,
            "description": description,
            "created": "2026-06-15",
            "enabled": True,
        }
        print(f"[KMS] Created master key: {key_id}")
        return key_id

    def generate_data_key(self, key_id):
        """Generate a data key encrypted under the master key."""
        data_key = Fernet.generate_key()
        self.data_keys[key_id] = data_key
        print(f"[KMS] Generated data key for {key_id}")
        return data_key

    def encrypt(self, key_id, plaintext):
        if key_id not in self.data_keys:
            raise ValueError(f"Data key for {key_id} not found")
        f = Fernet(self.data_keys[key_id])
        ciphertext = f.encrypt(plaintext.encode())
        print(f"[KMS] Encrypted {len(plaintext)} bytes → {len(ciphertext)} bytes")
        return ciphertext

    def decrypt(self, key_id, ciphertext):
        if key_id not in self.data_keys:
            raise ValueError(f"Data key for {key_id} not found")
        f = Fernet(self.data_keys[key_id])
        plaintext = f.decrypt(ciphertext).decode()
        print(f"[KMS] Decrypted {len(ciphertext)} bytes → {len(plaintext)} bytes")
        return plaintext

kms = KMS()
key = kms.create_key("dodatech-app-key", "App secrets encryption")
kms.generate_data_key(key)
secret = "Password: SuperSecret123!"
encrypted = kms.encrypt(key, secret)
decrypted = kms.decrypt(key, encrypted)
print(f"\nOriginal: {secret}")
print(f"Decrypted: {decrypted}")
print(f"Match: {secret == decrypted}")

# security_groups.py
# Simulate security group evaluation
class SecurityGroup:
    def __init__(self, name):
        self.name = name
        self.inbound_rules = []
        self.outbound_rules = []

    def allow_inbound(self, protocol, port, source, description=""):
        self.inbound_rules.append({"protocol": protocol, "port": port, "source": source, "desc": description})

    def allow_outbound(self, protocol, port, destination, description=""):
        self.outbound_rules.append({"protocol": protocol, "port": port, "destination": destination, "desc": description})

    def check_inbound(self, protocol, port, source_ip):
        for rule in self.inbound_rules:
            if rule["protocol"] in (protocol, "-1") and \
               (rule["port"] == port or rule["port"] == "ALL") and \
               (rule["source"] == "0.0.0.0/0" or source_ip.startswith(rule["source"].replace("0.0.0.0/0", "").rstrip(".0"))):
                return "ALLOW", rule["desc"]
        return "DENY", "No matching rule"

web_sg = SecurityGroup("web-server-sg")
web_sg.allow_inbound("tcp", 80, "0.0.0.0/0", "HTTP traffic")
web_sg.allow_inbound("tcp", 443, "0.0.0.0/0", "HTTPS traffic")
web_sg.allow_inbound("tcp", 22, "10.0.0.0/8", "SSH from internal")
web_sg.allow_outbound("-1", "ALL", "0.0.0.0/0", "All outbound")

print("=== Security Group Rules: web-server-sg ===")
for rule in web_sg.inbound_rules:
    print(f"  {rule['protocol']:<5} {rule['port']:<8} {rule['source']:<20} {rule['desc']}")

print("\n=== Inbound Traffic Checks ===")
checks = [("tcp", 443, "203.0.113.5"), ("tcp", 22, "203.0.113.5"), ("tcp", 22, "10.0.1.50"), ("udp", 53, "8.8.8.8")]
for protocol, port, ip in checks:
    result, reason = web_sg.check_inbound(protocol, port, ip)
    print(f"  {protocol.upper()} {port} from {ip:<20} → {result:<6} ({reason})")

# cloud_security_audit.py
# Perform a basic security audit on a simulated cloud environment
import json

def audit_security_group(name, rules, checks):
    findings = []
    for protocol, port, source, description in rules:
        if source == "0.0.0.0/0" and port in (22, 3389, 3306, 5432, 6379, 27017):
            findings.append(f"HIGH: {name} allows public {description} ({protocol}/{port} from {source})")
        elif source == "0.0.0.0/0" and port in checks.get("restricted_ports", []):
            findings.append(f"MEDIUM: {name} allows public {description} ({protocol}/{port} from {source})")
    return findings

def check_iam_policy(policy):
    findings = []
    if policy.get("Action") == "*" and policy.get("Resource") == "*":
        findings.append("CRITICAL: IAM policy grants full admin access (*:* on *)")
    if isinstance(policy.get("Action"), list) and "*" in policy["Action"]:
        findings.append("HIGH: IAM policy uses wildcard Action")
    return findings

audit_results = []

# Security group checks
sg_rules = [
    ("web-sg", 22, "0.0.0.0/0", "SSH"),
    ("web-sg", 443, "0.0.0.0/0", "HTTPS"),
    ("db-sg", 3306, "10.0.0.0/16", "MySQL"),
    ("admin-sg", 22, "203.0.113.0/24", "SSH Admin"),
    ("redis-sg", 6379, "0.0.0.0/0", "Redis"),
]
checks = {"restricted_ports": [22, 3389, 3306, 5432, 6379, 27017, 9200]}
audit_results.extend(audit_security_group("sg-rules", sg_rules, checks))

# IAM policy checks
iam_policies = [
    {"name": "admin-policy", "Action": "*", "Resource": "*"},
    {"name": "s3-readonly", "Action": ["s3:GetObject", "s3:ListBucket"], "Resource": "arn:aws:s3:::my-bucket/*"},
    {"name": "dev-policy", "Action": ["ec2:*", "s3:*", "lambda:*"], "Resource": "*"},
]
for policy in iam_policies:
    audit_results.extend(check_iam_policy(policy))

print("=== Cloud Security Audit Report ===\n")
for finding in audit_results:
    print(f"  • {finding}")

print(f"\nTotal findings: {len(audit_results)}")