Cloud Computing — Why Your $300K Bill Became $2.4M

from dataclasses import dataclass
from enum import Enum
from typing import List, Dict


class ServiceModel(Enum):
    IAAS = 'IaaS'
    PAAS = 'PaaS'
    SAAS = 'SaaS'
    SERVERLESS = 'Serverless'


@dataclass
class WorkloadProfile:
    name: str
    requires_custom_os: bool
    requires_custom_runtime: bool
    requires_host_access: bool
    stateful: bool
    traffic_pattern: str  # 'predictable', 'spiky', 'event-driven'
    team_size: int
    latency_sla_ms: int
    max_execution_time_minutes: int


class ServiceModelSelector:
    """Recommend cloud service model based on workload characteristics."""

    def recommend(self, workload: WorkloadProfile) -> Dict:
        """Return recommended service model with reasoning."""
        scores = {
            ServiceModel.IAAS: 0,
            ServiceModel.PAAS: 0,
            ServiceModel.SERVERLESS: 0,
        }

        # IaaS signals
        if workload.requires_custom_os:
            scores[ServiceModel.IAAS] += 3
        if workload.requires_custom_runtime:
            scores[ServiceModel.IAAS] += 2
        if workload.requires_host_access:
            scores[ServiceModel.IAAS] += 3
        if workload.stateful and workload.traffic_pattern == 'predictable':
            scores[ServiceModel.IAAS] += 1

        # PaaS signals
        if not workload.requires_custom_os and not workload.requires_host_access:
            scores[ServiceModel.PAAS] += 2
        if workload.team_size < 10:
            scores[ServiceModel.PAAS] += 2
        if workload.traffic_pattern == 'predictable':
            scores[ServiceModel.PAAS] += 1
        if workload.latency_sla_ms < 100:
            scores[ServiceModel.PAAS] += 1

        # Serverless signals
        if workload.traffic_pattern == 'event-driven':
            scores[ServiceModel.SERVERLESS] += 3
        if workload.traffic_pattern == 'spiky':
            scores[ServiceModel.SERVERLESS] += 2
        if workload.max_execution_time_minutes <= 15:
            scores[ServiceModel.SERVERLESS] += 1
        if workload.team_size < 5:
            scores[ServiceModel.SERVERLESS] += 2
        if workload.latency_sla_ms > 500:
            scores[ServiceModel.SERVERLESS] += 1

        # Penalize serverless for latency-sensitive workloads
        if workload.latency_sla_ms < 50:
            scores[ServiceModel.SERVERLESS] -= 3

        # Penalize IaaS for small teams
        if workload.team_size < 5:
            scores[ServiceModel.IAAS] -= 2

        best = max(scores, key=scores.get)

        return {
            'workload': workload.name,
            'recommendation': best.value,
            'scores': {k.value: v for k, v in scores.items()},
            'reasoning': self._explain(best, workload),
        }

    def _explain(self, model: ServiceModel, workload: WorkloadProfile) -> str:
        if model == ServiceModel.IAAS:
            return (
                f"IaaS recommended: workload requires custom OS/runtime/host access. "
                f"Team of {workload.team_size} can manage infrastructure operations."
            )
        elif model == ServiceModel.PAAS:
            return (
                f"PaaS recommended: standard runtime, no host access needed. "
                f"Team of {workload.team_size} benefits from reduced operational burden."
            )
        else:
            return (
                f"Serverless recommended: {workload.traffic_pattern} traffic pattern, "
                f"max execution {workload.max_execution_time_minutes}min. "
                f"Team of {workload.team_size} should focus on code, not infrastructure."
            )

    def validate_choice(self, model: ServiceModel, workload: WorkloadProfile) -> List[str]:
        """Validate that the chosen model fits the workload constraints."""
        warnings = []

        if model == ServiceModel.SERVERLESS:
            if workload.latency_sla_ms < 100:
                warnings.append(
                    f"WARNING: Serverless cold starts typically add 200-3000ms latency. "
                    f"SLA of {workload.latency_sla_ms}ms may be violated. "
                    f"Consider provisioned concurrency or PaaS."
                )
            if workload.max_execution_time_minutes > 15:
                warnings.append(
                    f"WARNING: Lambda max execution is 15 minutes. "
                    f"Workload requires {workload.max_execution_time_minutes} minutes. "
                    f"Use Fargate or ECS instead."
                )
            if workload.stateful:
                warnings.append(
                    f"WARNING: Serverless functions are stateless. "
                    f"Stateful workload requires external state store (DynamoDB, ElastiCache)."
                )

        if model == ServiceModel.IAAS:
            if workload.team_size < 5:
                warnings.append(
                    f"WARNING: IaaS requires OS patching, security hardening, and monitoring. "
                    f"Team of {workload.team_size} may lack operational capacity. "
                    f"Consider PaaS or managed services."
                )

        return warnings

from dataclasses import dataclass
from enum import Enum
from typing import List, Dict


class DeploymentModel(Enum):
    PUBLIC = 'Public Cloud'
    PRIVATE = 'Private Cloud'
    HYBRID = 'Hybrid Cloud'
    MULTI_CLOUD = 'Multi-Cloud'


@dataclass
class ComplianceRequirement:
    name: str
    data_residency: str  # 'any', 'country', 'on-premises'
    encryption_at_rest: bool
    encryption_in_transit: bool
    audit_trail: bool
    data_isolation: bool  # requires single-tenant


@dataclass
class WorkloadRequirements:
    name: str
    peak_traffic_multiplier: float  # peak / average
    latency_sla_ms: int
    data_volume_tb: float
    compliance: List[ComplianceRequirement]
    budget_monthly_usd: float
    team_cloud_experience_years: float


class DeploymentModelAnalyzer:
    """Analyze workload requirements and recommend deployment model."""

    def analyze(self, workload: WorkloadRequirements) -> Dict:
        """Score each deployment model against workload requirements."""
        scores = {model: 0 for model in DeploymentModel}
        warnings = []

        # Compliance analysis
        for req in workload.compliance:
            if req.data_residency == 'on-premises':
                scores[DeploymentModel.PRIVATE] += 5
                scores[DeploymentModel.HYBRID] += 3
                scores[DeploymentModel.PUBLIC] -= 3
                warnings.append(f"{req.name}: requires on-premises data — private or hybrid cloud required")
            elif req.data_isolation:
                scores[DeploymentModel.PRIVATE] += 3
                scores[DeploymentModel.HYBRID] += 2
                warnings.append(f"{req.name}: requires data isolation — consider dedicated tenancy or private cloud")

        # Elasticity analysis
        if workload.peak_traffic_multiplier > 5:
            scores[DeploymentModel.PUBLIC] += 3
            scores[DeploymentModel.HYBRID] += 2
            scores[DeploymentModel.PRIVATE] -= 2
            warnings.append(f"Peak traffic is {workload.peak_traffic_multiplier}x average — public cloud elasticity is critical")

        # Budget analysis
        if workload.budget_monthly_usd < 10000:
            scores[DeploymentModel.PUBLIC] += 2
            scores[DeploymentModel.PRIVATE] -= 3
            warnings.append(f"Budget ${workload.budget_monthly_usd}/mo — private cloud capex is prohibitive")
        elif workload.budget_monthly_usd > 500000:
            scores[DeploymentModel.PRIVATE] += 1
            scores[DeploymentModel.MULTI_CLOUD] += 1

        # Team experience
        if workload.team_cloud_experience_years < 2:
            scores[DeploymentModel.PUBLIC] += 2
            scores[DeploymentModel.MULTI_CLOUD] -= 3
            warnings.append(f"Team has {workload.team_cloud_experience_years} years cloud experience — multi-cloud adds unacceptable complexity")

        # Latency analysis
        if workload.latency_sla_ms < 10:
            scores[DeploymentModel.PRIVATE] += 3
            scores[DeploymentModel.HYBRID] += 1
            warnings.append(f"Sub-10ms SLA requires edge/on-premises — public cloud round-trip adds 20-80ms")

        best = max(scores, key=scores.get)

        return {
            'workload': workload.name,
            'recommendation': best.value,
            'scores': {k.value: v for k, v in scores.items()},
            'warnings': warnings,
        }

    def estimate_multi_cloud_complexity(self, num_providers: int) -> Dict:
        """Estimate operational complexity increase for multi-cloud."""
        base_complexity = 1.0
        multiplier = 1.0 + (num_providers - 1) * 1.5

        return {
            'providers': num_providers,
            'complexity_multiplier': round(multiplier, 1),
            'additional_requirements': [
                f'{num_providers}x IAM systems to manage',
                f'{num_providers}x monitoring dashboards',
                f'{num_providers}x CI/CD pipelines',
                f'{num_providers}x security posture configurations',
                f'Cross-cloud networking (VPN/direct connect to each provider)',
                f'Data transfer costs between providers',
                f'Team expertise required across {num_providers} provider ecosystems',
            ],
            'recommendation': (
                'Avoid multi-cloud unless driven by regulatory requirement or specific service need. '
                'Single-cloud with multi-region redundancy is more reliable and 2-3x cheaper to operate.'
            ),
        }

from dataclasses import dataclass
from typing import List, Dict, Optional
from enum import Enum


class PurchaseOption(Enum):
    ON_DEMAND = 'On-Demand'
    RESERVED_1YR = 'Reserved 1-Year'
    RESERVED_3YR = 'Reserved 3-Year'
    SAVINGS_PLAN = 'Savings Plan'
    SPOT = 'Spot'


@dataclass
class ComputeWorkload:
    name: str
    instance_type: str
    vcpus: int
    memory_gb: float
    avg_cpu_percent: float
    peak_cpu_percent: float
    hours_per_month: int
    is_fault_tolerant: bool
    is_stateful: bool
    traffic_predictability: str  # 'stable', 'variable', 'unpredictable'


@dataclass
class CostEstimate:
    workload: str
    current_option: str
    current_monthly_cost: float
    recommended_option: str
    recommended_monthly_cost: float
    savings_monthly: float
    savings_percent: float
    action: str


class CloudCostOptimizer:
    """Analyze workloads and recommend cost optimization strategies."""

    # Simplified pricing (per hour, representative m5.xlarge)
    PRICING = {
        'm5.xlarge': {
            PurchaseOption.ON_DEMAND: 0.192,
            PurchaseOption.RESERVED_1YR: 0.121,
            PurchaseOption.RESERVED_3YR: 0.078,
            PurchaseOption.SPOT: 0.04,
        },
        'm5.2xlarge': {
            PurchaseOption.ON_DEMAND: 0.384,
            PurchaseOption.RESERVED_1YR: 0.242,
            PurchaseOption.RESERVED_3YR: 0.156,
            PurchaseOption.SPOT: 0.08,
        },
        'c5.xlarge': {
            PurchaseOption.ON_DEMAND: 0.170,
            PurchaseOption.RESERVED_1YR: 0.107,
            PurchaseOption.RESERVED_3YR: 0.069,
            PurchaseOption.SPOT: 0.035,
        },
        'r5.xlarge': {
            PurchaseOption.ON_DEMAND: 0.252,
            PurchaseOption.RESERVED_1YR: 0.159,
            PurchaseOption.RESERVED_3YR: 0.102,
            PurchaseOption.SPOT: 0.052,
        },
    }

    def analyze_workload(self, workload: ComputeWorkload) -> CostEstimate:
        """Analyze a single workload and recommend optimal purchase option."""
        pricing = self.PRICING.get(workload.instance_type, self.PRICING['m5.xlarge'])
        current_cost = pricing[PurchaseOption.ON_DEMAND] * workload.hours_per_month

        # Right-size recommendation
        recommended_instance = self._right_size(workload)
        recommended_pricing = self.PRICING.get(recommended_instance, self.PRICING['m5.xlarge'])

        # Purchase option recommendation
        recommended_option = self._recommend_purchase_option(workload)
        recommended_cost = recommended_pricing[recommended_option] * workload.hours_per_month

        savings = current_cost - recommended_cost
        savings_pct = (savings / current_cost * 100) if current_cost > 0 else 0

        actions = []
        if recommended_instance != workload.instance_type:
            actions.append(f'Right-size from {workload.instance_type} to {recommended_instance}')
        if recommended_option != PurchaseOption.ON_DEMAND:
            actions.append(f'Switch from On-Demand to {recommended_option.value}')
        if workload.avg_cpu_percent < 20:
            actions.append(f'Average CPU {workload.avg_cpu_percent}% — significant headroom for downsizing')

        return CostEstimate(
            workload=workload.name,
            current_option=PurchaseOption.ON_DEMAND.value,
            current_monthly_cost=round(current_cost, 2),
            recommended_option=recommended_option.value,
            recommended_monthly_cost=round(recommended_cost, 2),
            savings_monthly=round(savings, 2),
            savings_percent=round(savings_pct, 1),
            action=' | '.join(actions) if actions else 'No optimization needed',
        )

    def _right_size(self, workload: ComputeWorkload) -> str:
        """Recommend right-sized instance based on actual utilization."""
        if workload.avg_cpu_percent < 20 and workload.peak_cpu_percent < 50:
            # Downsize by one tier
            if '2xlarge' in workload.instance_type:
                return workload.instance_type.replace('2xlarge', 'xlarge')
            elif 'xlarge' in workload.instance_type:
                return workload.instance_type.replace('xlarge', 'large')
        return workload.instance_type

    def _recommend_purchase_option(self, workload: ComputeWorkload) -> PurchaseOption:
        """Recommend purchase option based on workload characteristics."""
        if workload.is_fault_tolerant and not workload.is_stateful:
            return PurchaseOption.SPOT
        elif workload.traffic_predictability == 'stable':
            return PurchaseOption.RESERVED_1YR
        elif workload.traffic_predictability == 'variable':
            return PurchaseOption.SAVINGS_PLAN
        else:
            return PurchaseOption.ON_DEMAND

    def analyze_fleet(self, workloads: List[ComputeWorkload]) -> Dict:
        """Analyze an entire fleet of workloads."""
        estimates = [self.analyze_workload(w) for w in workloads]

        total_current = sum(e.current_monthly_cost for e in estimates)
        total_recommended = sum(e.recommended_monthly_cost for e in estimates)
        total_savings = total_current - total_recommended

        return {
            'workloads_analyzed': len(estimates),
            'total_current_monthly': round(total_current, 2),
            'total_optimized_monthly': round(total_recommended, 2),
            'total_monthly_savings': round(total_savings, 2),
            'total_annual_savings': round(total_savings * 12, 2),
            'savings_percent': round((total_savings / total_current * 100), 1) if total_current > 0 else 0,
            'estimates': [
                {
                    'workload': e.workload,
                    'current_cost': e.current_monthly_cost,
                    'optimized_cost': e.recommended_monthly_cost,
                    'savings': e.savings_monthly,
                    'action': e.action,
                }
                for e in sorted(estimates, key=lambda x: x.savings_monthly, reverse=True)
            ],
        }

    def estimate_nat_gateway_savings(self, nat_gateways: List[Dict]) -> Dict:
        """Estimate savings from decommissioning idle NAT Gateways."""
        idle_count = 0
        monthly_savings = 0.0

        for gw in nat_gateways:
            if gw.get('monthly_gb', 0) < 0.1:  # Less than 100MB/month
                idle_count += 1
                # NAT Gateway: $0.045/hr * 730 hrs = $32.85/month base cost
                monthly_savings += 32.85

        return {
            'total_nat_gateways': len(nat_gateways),
            'idle_nat_gateways': idle_count,
            'monthly_savings': round(monthly_savings, 2),
            'annual_savings': round(monthly_savings * 12, 2),
            'recommendation': (
                f'Decommission {idle_count} idle NAT Gateways. '
                f'Replace low-traffic NAT Gateways with NAT Instances (t3.nano at ~$7.50/month). '
                f'Use VPC Gateway Endpoints for S3/DynamoDB traffic (free).'
            ),
        }

from dataclasses import dataclass
from typing import List, Dict
from enum import Enum


class FailureMode(Enum):
    REGION_OUTAGE = 'Regional Outage'
    AZ_FAILURE = 'Availability Zone Failure'
    SERVICE_OUTAGE = 'Service-Specific Outage'
    NOISY_NEIGHBOR = 'Noisy Neighbor'
    API_RATE_LIMIT = 'API Rate Limiting'
    CONTROL_PLANE_OUTAGE = 'Control Plane Outage'


@dataclass
class ArchitectureComponent:
    name: str
    service_type: str  # 'compute', 'database', 'storage', 'networking', 'managed'
    deployment_scope: str  # 'single-az', 'multi-az', 'multi-region'
    is_managed: bool
    has_autoscaling: bool
    depends_on_control_plane_at_runtime: bool
    stateful: bool


@dataclass
class ResilienceAssessment:
    component: str
    failure_mode: str
    risk_level: str  # 'LOW', 'MEDIUM', 'HIGH', 'CRITICAL'
    current_mitigation: str
    recommended_mitigation: str
    estimated_downtime_minutes: int


class ResilienceAnalyzer:
    """Analyze architecture resilience against common cloud failure modes."""

    def assess_component(self, component: ArchitectureComponent) -> List[ResilienceAssessment]:
        """Assess a single component against all failure modes."""
        assessments = []

        # Regional outage assessment
        if component.deployment_scope == 'single-region':
            assessments.append(ResilienceAssessment(
                component=component.name,
                failure_mode=FailureMode.REGION_OUTAGE.value,
                risk_level='HIGH' if component.stateful else 'MEDIUM',
                current_mitigation='None — single region deployment',
                recommended_mitigation='Deploy multi-region with automated failover. Use global databases (Aurora Global, Spanner) for stateful workloads.',
                estimated_downtime_minutes=120,
            ))
        elif component.deployment_scope == 'multi-region':
            assessments.append(ResilienceAssessment(
                component=component.name,
                failure_mode=FailureMode.REGION_OUTAGE.value,
                risk_level='LOW',
                current_mitigation='Multi-region deployment with failover',
                recommended_mitigation='Validate failover automation with regular drills. Test DNS TTL propagation.',
                estimated_downtime_minutes=5,
            ))

        # AZ failure assessment
        if component.deployment_scope == 'single-az':
            assessments.append(ResilienceAssessment(
                component=component.name,
                failure_mode=FailureMode.AZ_FAILURE.value,
                risk_level='HIGH',
                current_mitigation='None — single AZ deployment',
                recommended_mitigation='Deploy across 3+ AZs. Use managed services with multi-AZ built-in.',
                estimated_downtime_minutes=60,
            ))

        # Control plane dependency
        if component.depends_on_control_plane_at_runtime:
            assessments.append(ResilienceAssessment(
                component=component.name,
                failure_mode=FailureMode.CONTROL_PLANE_OUTAGE.value,
                risk_level='CRITICAL',
                current_mitigation='None — runtime dependency on control plane',
                recommended_mitigation='Cache credentials and configuration. Use static fallbacks. Never depend on control plane for data path.',
                estimated_downtime_minutes=180,
            ))

        # Noisy neighbor (non-managed, shared tenancy)
        if not component.is_managed and component.service_type == 'compute':
            assessments.append(ResilienceAssessment(
                component=component.name,
                failure_mode=FailureMode.NOISY_NEIGHBOR.value,
                risk_level='MEDIUM',
                current_mitigation='Unknown — shared tenancy assumed',
                recommended_mitigation='Monitor CPU steal time. Switch to dedicated tenancy or compute-optimized instances if steal > 5%.',
                estimated_downtime_minutes=0,  # Performance degradation, not downtime
            ))

        return assessments

    def assess_architecture(self, components: List[ArchitectureComponent]) -> Dict:
        """Assess entire architecture resilience."""
        all_assessments = []
        for component in components:
            all_assessments.extend(self.assess_component(component))

        critical = [a for a in all_assessments if a.risk_level == 'CRITICAL']
        high = [a for a in all_assessments if a.risk_level == 'HIGH']
        medium = [a for a in all_assessments if a.risk_level == 'MEDIUM']

        return {
            'total_components': len(components),
            'total_risks': len(all_assessments),
            'critical_risks': len(critical),
            'high_risks': len(high),
            'medium_risks': len(medium),
            'overall_risk': 'CRITICAL' if critical else 'HIGH' if high else 'MEDIUM' if medium else 'LOW',
            'assessments': [
                {
                    'component': a.component,
                    'failure_mode': a.failure_mode,
                    'risk': a.risk_level,
                    'recommendation': a.recommended_mitigation,
                }
                for a in sorted(all_assessments, key=lambda a: ['CRITICAL', 'HIGH', 'MEDIUM', 'LOW'].index(a.risk_level))
            ],
        }

import json
from dataclasses import dataclass
from typing import List, Dict, Set


@dataclass
class IAMStatement:
    effect: str  # 'Allow' or 'Deny'
    actions: List[str]
    resources: List[str]
    conditions: Dict


@dataclass
class SecurityFinding:
    severity: str  # 'CRITICAL', 'HIGH', 'MEDIUM', 'LOW'
    category: str
    description: str
    recommendation: str
    affected_resource: str


class IAMPolicyAnalyzer:
    """Analyze IAM policies for security misconfigurations."""

    DANGEROUS_ACTIONS = {
        'iam:CreateUser', 'iam:CreateRole', 'iam:AttachRolePolicy',
        'iam:PutRolePolicy', 'iam:CreatePolicyVersion', 'iam:SetDefaultPolicyVersion',
        'sts:AssumeRole', 'sts:AssumeRoleWithSAML',
        's3:DeleteBucket', 's3:DeleteBucketPolicy', 's3:PutBucketPolicy',
        'ec2:RunInstances', 'ec2:CreateKeyPair',
        'lambda:CreateFunction', 'lambda:UpdateFunctionCode',
        'kms:Decrypt', 'kms:CreateGrant',
    }

    PRIVILEGE_ESCALATION_PATTERNS = [
        {'actions': ['iam:PutRolePolicy', 'iam:AttachRolePolicy'], 'description': 'Can attach arbitrary policies to roles — full privilege escalation'},
        {'actions': ['iam:CreatePolicyVersion', 'iam:SetDefaultPolicyVersion'], 'description': 'Can modify policy versions — privilege escalation via policy versioning'},
        {'actions': ['lambda:CreateFunction', 'iam:PassRole'], 'description': 'Can create Lambda with privileged role — code execution with escalated privileges'},
        {'actions': ['ec2:RunInstances', 'iam:PassRole'], 'description': 'Can launch EC2 with privileged role — code execution with escalated privileges'},
    ]

    def analyze_policy(self, policy_document: Dict, policy_name: str = 'unknown') -> List[SecurityFinding]:
        """Analyze a single IAM policy document for security issues."""
        findings = []
        statements = policy_document.get('Statement', [])

        for stmt in statements:
            effect = stmt.get('Effect', '')
            actions = stmt.get('Action', [])
            if isinstance(actions, str):
                actions = [actions]
            resources = stmt.get('Resource', [])
            if isinstance(resources, str):
                resources = [resources]
            conditions = stmt.get('Condition', {})

            # Check for wildcard actions
            if '*' in actions and effect == 'Allow':
                findings.append(SecurityFinding(
                    severity='CRITICAL',
                    category='Wildcard Actions',
                    description=f'Policy grants wildcard (*) actions — full AWS access',
                    recommendation='Replace * with specific actions required. Use AWS managed policies as reference.',
                    affected_resource=policy_name,
                ))

            # Check for wildcard resources with dangerous actions
            if '*' in resources and effect == 'Allow':
                dangerous_in_policy = set(actions) & self.DANGEROUS_ACTIONS
                if dangerous_in_policy:
                    findings.append(SecurityFinding(
                        severity='HIGH',
                        category='Wildcard Resource with Dangerous Actions',
                        description=f'Dangerous actions on all resources: {dangerous_in_policy}',
                        recommendation='Scope resources to specific ARNs. Never grant dangerous actions on Resource: *.',
                        affected_resource=policy_name,
                    ))

            # Check for privilege escalation patterns
            action_set = set(actions)
            for pattern in self.PRIVILEGE_ESCALATION_PATTERNS:
                if set(pattern['actions']).issubset(action_set) and effect == 'Allow':
                    findings.append(SecurityFinding(
                        severity='CRITICAL',
                        category='Privilege Escalation',
                        description=pattern['description'],
                        recommendation=f'Remove or scope actions: {pattern["actions"]}. Use permission boundaries to limit escalation.',
                        affected_resource=policy_name,
                    ))

            # Check for missing conditions
            if effect == 'Allow' and not conditions and set(actions) & self.DANGEROUS_ACTIONS:
                findings.append(SecurityFinding(
                    severity='MEDIUM',
                    category='Missing Conditions',
                    description='Dangerous actions granted without condition constraints',
                    recommendation='Add conditions: aws:MultiFactorAuthPresent, aws:SourceIp, aws:PrincipalOrgID.',
                    affected_resource=policy_name,
                ))

        return findings

    def analyze_bucket_policy(self, bucket_policy: Dict, bucket_name: str) -> List[SecurityFinding]:
        """Analyze S3 bucket policy for public access and over-permissioning."""
        findings = []
        statements = bucket_policy.get('Statement', [])

        for stmt in statements:
            principal = stmt.get('Principal', '')
            effect = stmt.get('Effect', '')

            if principal == '*' and effect == 'Allow':
                findings.append(SecurityFinding(
                    severity='CRITICAL',
                    category='Public S3 Access',
                    description=f'Bucket {bucket_name} allows public access via Principal: *',
                    recommendation='Remove public access. Use S3 Block Public Access setting. Require authentication for all access.',
                    affected_resource=bucket_name,
                ))

        return findings

    def generate_least_privilege_policy(self, actions_used: List[str], resources: List[str]) -> Dict:
        """Generate a least-privilege IAM policy from observed actions."""
        return {
            'Version': '2012-10-17',
            'Statement': [
                {
                    'Effect': 'Allow',
                    'Action': sorted(set(actions_used)),
                    'Resource': resources,
                    'Condition': {
                        'Bool': {'aws:MultiFactorAuthPresent': 'true'}
                    }
                }
            ],
        }

// io.thecodeforge — devops tutorial

// This Kubernetes Deployment template assumes services are stateless.
// Stateless = every request can go to any healthy replica.
apiVersion: apps/v1
kind: Deployment
metadata:
  name: payment-processor
spec:
  replicas: 3
  selector:
    matchLabels:
      app: payment-processor
  template:
    metadata:
      labels:
        app: payment-processor
    spec:
      containers:
      - name: payment-app
        image: payments:v2.4.1
        ports:
        - containerPort: 8080
        env:
        - name: SESSION_STORE
          value: "redis://payment-sessions:6379"  // Sessions out of memory, into Redis
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: db-secret
              key: connection-string
        resources:
          requests:
            cpu: "500m"
            memory: "512Mi"
          limits:
            cpu: "1000m"
            memory: "1Gi"
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: payment-processor-autoscaler
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: payment-processor
  minReplicas: 3
  maxReplicas: 20
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70

// io.thecodeforge — devops tutorial

// AWS API Gateway with per-client rate limiting and regional failover
// This protects backend services by rate-limiting at the edge, before a request hits your services.
openapi: "3.0.1"
info:
  title: api-catalog
  version: "2024-03-15"
x-amazon-apigateway-api-key-source: "HEADER"
paths:
  /orders:
    post:
      x-amazon-apigateway-request-validator: "validate-body"
      x-amazon-apigateway-usage-plan:
        - api-key: "required"
          throttling:
            burstLimit: 100   // Allow short bursts of 100 requests
            rateLimit: 50     // Sustained rate of 50 requests per second per client
          quota:
            limit: 50000      // 50k requests per day per API key
            period: DAY
      x-amazon-apigateway-integration:
        type: HTTP_PROXY
        httpMethod: POST
        uri: "https://abc123.execute-api.us-east-1.amazonaws.com/v1/orders"
        # Regional failover: primary region us-east-1, fallback us-west-2
        connectionType: VPC_LINK
        connectionId: "east-coast-vpc-link"
---
// Health check config for global load balancer (AWS CloudFront + Route53)
// If health checks fail for us-east-1, traffic routes to us-west-2
global-secondary-region:
  failover: true
  health-check:
    type: HTTP
    path: /health
    interval: 30
    threshold: 3
    timeout: 5

// io.thecodeforge — devops tutorial
// Production Python service

FROM python:3.11-slim AS builder
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir --user -r requirements.txt

FROM python:3.11-slim
COPY --from=builder /root/.local /root/.local
COPY app/ /app/
USER 1000:1000
ENV PATH=/root/.local/bin:$PATH
EXPOSE 8080
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8080"]

// io.thecodeforge — devops tutorial
#!/bin/bash
set -euo pipefail

REGION="us-east-1"
DRY_RUN=${1:-"true"}

SNAPSHOTS=$(aws ec2 describe-snapshots \
  --region "$REGION" \
  --owner-ids self \
  --query "Snapshots[?StartTime<='$(date -d '-30 days' +%Y-%m-%d)'].SnapshotId" \
  --output text)

echo "Found snapshots to delete:"
echo "$SNAPSHOTS"

if [ "$DRY_RUN" = "true" ]; then
  echo "[DRY RUN] Would delete $SNAPSHOTS"
  exit 0
fi

for snap in $SNAPSHOTS; do
  aws ec2 delete-snapshot --snapshot-id "$snap" --region "$REGION"
done

// io.thecodeforge — devops tutorial

prerequisites:
  linux:
    - file_permissions
    - process_management
    - systemd_services
  scripting:
    - bash_loops_and_conditions
    - python_automation
  networking:
    - tcp_ip_basics
    - dns_lookup
    - http_status_codes
  git:
    - branching_strategies
    - merge_conflicts
  cloud:
    - iam_policies
    - vm_deployment

// io.thecodeforge — devops tutorial

devops_core_concepts:
  idempotency:
    description: "Same input always produces same output"
    anti_pattern: "shell scripts without state checks"
  immutable_infrastructure:
    description: "Replace servers, never patch them"
    anti_pattern: "ssh into prod to 'fix' config"
  observability:
    components:
      - structured_logging
      - high_cardinality_metrics
      - distributed_tracing
  infrastructure_as_code:
    tools:
      - terraform
      - cloudformation
    rule: "no manual console changes"
  blameless_culture:
    focus: "system failures, not individual mistakes"