Cloud Security Automation: Infrastructure as Code Security

Complete Cloud Security Automation with Terraform and Python:

requirements.txt:

boto3>=1.34.0
terraform>=1.0.0  # Note: Requires Terraform binary installed

Complete Terraform Configuration:

# terraform/main.tf
terraform {
  required_version = ">= 1.0"
  
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.0"
    }
    random = {
      source  = "hashicorp/random"
      version = "~> 3.0"
    }
  }
}

provider "aws" {
  region = var.aws_region
  
  default_tags {
    tags = {
      Environment = var.environment
      ManagedBy   = "terraform"
      Security    = "enabled"
    }
  }
}

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

# Secure S3 bucket with all security best practices
resource "aws_s3_bucket" "secure_bucket" {
  bucket = "${var.bucket_name_prefix}-${random_id.bucket_suffix.hex}"
  
  tags = {
    Name        = "${var.bucket_name_prefix}-secure"
    Description = "Secure S3 bucket with encryption and access controls"
  }
}

# Public access block - prevents accidental public exposure
resource "aws_s3_bucket_public_access_block" "secure_bucket" {
  bucket = aws_s3_bucket.secure_bucket.id

  block_public_acls       = true
  block_public_policy     = true
  ignore_public_acls      = true
  restrict_public_buckets = true
}

# Server-side encryption
resource "aws_s3_bucket_server_side_encryption_configuration" "secure_bucket" {
  bucket = aws_s3_bucket.secure_bucket.id

  rule {
    apply_server_side_encryption_by_default {
      sse_algorithm = var.encryption_algorithm
    }
    
    bucket_key_enabled = true
  }
}

# Versioning for data protection
resource "aws_s3_bucket_versioning" "secure_bucket" {
  bucket = aws_s3_bucket.secure_bucket.id

  versioning_configuration {
    status = "Enabled"
  }
}

# Lifecycle policy for cost optimization
resource "aws_s3_bucket_lifecycle_configuration" "secure_bucket" {
  bucket = aws_s3_bucket.secure_bucket.id

  rule {
    id     = "transition-to-ia"
    status = "Enabled"

    transition {
      days          = 30
      storage_class = "STANDARD_IA"
    }
  }

  rule {
    id     = "transition-to-glacier"
    status = "Enabled"

    transition {
      days          = 90
      storage_class = "GLACIER"
    }
  }

  rule {
    id     = "delete-old-versions"
    status = "Enabled"

    noncurrent_version_expiration {
      noncurrent_days = 90
    }
  }
}

# Bucket logging
resource "aws_s3_bucket_logging" "secure_bucket" {
  bucket = aws_s3_bucket.secure_bucket.id

  target_bucket = aws_s3_bucket.logging_bucket.id
  target_prefix = "s3-access-logs/"
}

# Logging bucket
resource "aws_s3_bucket" "logging_bucket" {
  bucket = "${var.bucket_name_prefix}-logs-${random_id.bucket_suffix.hex}"
}

resource "aws_s3_bucket_public_access_block" "logging_bucket" {
  bucket = aws_s3_bucket.logging_bucket.id

  block_public_acls       = true
  block_public_policy     = true
  ignore_public_acls      = true
  restrict_public_buckets = true
}

# AWS Config rule for S3 public read prohibition
resource "aws_config_config_rule" "s3_public_read_prohibited" {
  name        = "s3-public-read-prohibited"
  description = "Checks that S3 buckets prohibit public read access"

  source {
    owner             = "AWS"
    source_identifier = "S3_BUCKET_PUBLIC_READ_PROHIBITED"
  }

  depends_on = [aws_config_configuration_recorder.main]
}

# Config recorder
resource "aws_config_configuration_recorder" "main" {
  name     = "main-config-recorder"
  role_arn = aws_iam_role.config.arn

  recording_group {
    all_supported                 = true
    include_global_resource_types = true
  }
}

# IAM role for Config
resource "aws_iam_role" "config" {
  name = "config-role"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = "sts:AssumeRole"
        Effect = "Allow"
        Principal = {
          Service = "config.amazonaws.com"
        }
      }
    ]
  })
}

resource "aws_iam_role_policy_attachment" "config" {
  role       = aws_iam_role.config.name
  policy_arn = "arn:aws:iam::aws:policy/service-role/AWS_ConfigRole"
}

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

variable "environment" {
  description = "Environment name"
  type        = string
  default     = "production"
}

variable "bucket_name_prefix" {
  description = "Prefix for bucket names"
  type        = string
  default     = "secure-bucket"
}

variable "encryption_algorithm" {
  description = "S3 encryption algorithm"
  type        = string
  default     = "AES256"
}

Complete Python Automation Manager:

#!/usr/bin/env python3
"""
Cloud Security Automation - Infrastructure Automation Manager
Production-ready security automation with Terraform and AWS integration
"""

import boto3
from botocore.exceptions import ClientError, BotoCoreError
from typing import Dict, List, Optional
from dataclasses import dataclass, asdict
from enum import Enum
import logging
import os
import subprocess
import json
import tempfile

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


class AutomationError(Exception):
    """Base exception for automation errors."""
    pass


class TerraformError(AutomationError):
    """Raised when Terraform operations fail."""
    pass


@dataclass
class SecurityPolicy:
    """Security policy definition."""
    name: str
    description: str
    policy_type: str  # config_rule, s3_policy, iam_policy, etc.
    configuration: Dict
    
    def to_dict(self) -> Dict:
        """Convert to dictionary."""
        return asdict(self)


class SecurityAutomationManager:
    """Manages cloud security automation with Terraform and AWS."""
    
    def __init__(
        self,
        terraform_dir: str,
        region_name: str = 'us-east-1',
        aws_access_key_id: Optional[str] = None,
        aws_secret_access_key: Optional[str] = None
    ):
        """Initialize security automation manager.
        
        Args:
            terraform_dir: Directory containing Terraform files
            region_name: AWS region (default: us-east-1)
            aws_access_key_id: AWS access key (defaults to env/credentials)
            aws_secret_access_key: AWS secret key (defaults to env/credentials)
        """
        self.terraform_dir = terraform_dir
        self.region_name = region_name
        
        try:
            session = boto3.Session(
                aws_access_key_id=aws_access_key_id or os.getenv('AWS_ACCESS_KEY_ID'),
                aws_secret_access_key=aws_secret_access_key or os.getenv('AWS_SECRET_ACCESS_KEY'),
                region_name=region_name
            )
            
            self.ec2 = session.client('ec2', region_name=region_name)
            self.config = session.client('config', region_name=region_name)
            self.s3 = session.client('s3', region_name=region_name)
            
            logger.info(f"Initialized SecurityAutomationManager for region: {region_name}")
            
        except (ClientError, BotoCoreError) as e:
            error_msg = f"Failed to initialize AWS clients: {e}"
            logger.error(error_msg)
            raise AutomationError(error_msg) from e
    
    def terraform_init(self) -> bool:
        """Initialize Terraform.
        
        Returns:
            True if successful, False otherwise
            
        Raises:
            TerraformError: If initialization fails
        """
        try:
            logger.info("Initializing Terraform...")
            result = subprocess.run(
                ['terraform', 'init'],
                cwd=self.terraform_dir,
                capture_output=True,
                text=True,
                check=True
            )
            logger.info("Terraform initialized successfully")
            return True
        
        except subprocess.CalledProcessError as e:
            error_msg = f"Terraform init failed: {e.stderr}"
            logger.error(error_msg)
            raise TerraformError(error_msg) from e
    
    def terraform_validate(self) -> bool:
        """Validate Terraform configuration.
        
        Returns:
            True if valid, False otherwise
            
        Raises:
            TerraformError: If validation fails
        """
        try:
            logger.info("Validating Terraform configuration...")
            result = subprocess.run(
                ['terraform', 'validate'],
                cwd=self.terraform_dir,
                capture_output=True,
                text=True,
                check=True
            )
            logger.info("Terraform configuration is valid")
            return True
        
        except subprocess.CalledProcessError as e:
            error_msg = f"Terraform validation failed: {e.stderr}"
            logger.error(error_msg)
            raise TerraformError(error_msg) from e
    
    def terraform_plan(self, variables: Optional[Dict] = None) -> Dict:
        """Create Terraform execution plan.
        
        Args:
            variables: Optional Terraform variables
            
        Returns:
            Plan output as dictionary
            
        Raises:
            TerraformError: If plan fails
        """
        try:
            logger.info("Creating Terraform plan...")
            
            cmd = ['terraform', 'plan', '-json', '-out=tfplan']
            
            # Add variables if provided
            if variables:
                for key, value in variables.items():
                    cmd.extend(['-var', f'{key}={value}'])
            
            result = subprocess.run(
                cmd,
                cwd=self.terraform_dir,
                capture_output=True,
                text=True,
                check=True
            )
            
            # Parse JSON output
            plan_output = []
            for line in result.stdout.split('\n'):
                if line.strip():
                    try:
                        plan_output.append(json.loads(line))
                    except json.JSONDecodeError:
                        continue
            
            logger.info("Terraform plan created successfully")
            
            return {
                'success': True,
                'plan_output': plan_output
            }
        
        except subprocess.CalledProcessError as e:
            error_msg = f"Terraform plan failed: {e.stderr}"
            logger.error(error_msg)
            raise TerraformError(error_msg) from e
    
    def terraform_apply(self, auto_approve: bool = False) -> Dict:
        """Apply Terraform configuration.
        
        Args:
            auto_approve: If True, skip confirmation prompt
            
        Returns:
            Apply output as dictionary
            
        Raises:
            TerraformError: If apply fails
        """
        try:
            logger.info("Applying Terraform configuration...")
            
            cmd = ['terraform', 'apply', '-json']
            if auto_approve:
                cmd.append('-auto-approve')
            else:
                cmd.append('tfplan')
            
            result = subprocess.run(
                cmd,
                cwd=self.terraform_dir,
                capture_output=True,
                text=True,
                check=True
            )
            
            # Parse JSON output
            apply_output = []
            for line in result.stdout.split('\n'):
                if line.strip():
                    try:
                        apply_output.append(json.loads(line))
                    except json.JSONDecodeError:
                        continue
            
            logger.info("Terraform apply completed successfully")
            
            # Get outputs
            outputs = self.terraform_output()
            
            return {
                'success': True,
                'apply_output': apply_output,
                'outputs': outputs
            }
        
        except subprocess.CalledProcessError as e:
            error_msg = f"Terraform apply failed: {e.stderr}"
            logger.error(error_msg)
            raise TerraformError(error_msg) from e
    
    def terraform_output(self) -> Dict:
        """Get Terraform outputs.
        
        Returns:
            Dictionary of Terraform outputs
            
        Raises:
            TerraformError: If output retrieval fails
        """
        try:
            result = subprocess.run(
                ['terraform', 'output', '-json'],
                cwd=self.terraform_dir,
                capture_output=True,
                text=True,
                check=True
            )
            
            return json.loads(result.stdout)
        
        except subprocess.CalledProcessError as e:
            error_msg = f"Failed to get Terraform outputs: {e.stderr}"
            logger.error(error_msg)
            raise TerraformError(error_msg) from e
        except json.JSONDecodeError as e:
            error_msg = f"Failed to parse Terraform outputs: {e}"
            logger.error(error_msg)
            raise TerraformError(error_msg) from e
    
    def terraform_destroy(self, auto_approve: bool = False) -> bool:
        """Destroy Terraform-managed infrastructure.
        
        Args:
            auto_approve: If True, skip confirmation prompt
            
        Returns:
            True if successful, False otherwise
            
        Raises:
            TerraformError: If destroy fails
        """
        try:
            logger.warning("Destroying Terraform infrastructure...")
            
            cmd = ['terraform', 'destroy', '-json']
            if auto_approve:
                cmd.append('-auto-approve')
            
            result = subprocess.run(
                cmd,
                cwd=self.terraform_dir,
                capture_output=True,
                text=True,
                check=True
            )
            
            logger.info("Terraform destroy completed successfully")
            return True
        
        except subprocess.CalledProcessError as e:
            error_msg = f"Terraform destroy failed: {e.stderr}"
            logger.error(error_msg)
            raise TerraformError(error_msg) from e
    
    def validate_security_policies(self) -> Dict:
        """Validate security policies using AWS Config.
        
        Returns:
            Validation results dictionary
        """
        try:
            logger.info("Validating security policies...")
            
            # Get compliance summary
            response = self.config.get_compliance_summary_by_config_rule()
            
            compliance_results = {
                'total_rules': len(response.get('ComplianceSummariesByConfigRule', [])),
                'compliant': 0,
                'non_compliant': 0,
                'not_applicable': 0
            }
            
            for summary in response.get('ComplianceSummariesByConfigRule', []):
                compliance_type = summary.get('ComplianceSummary', {}).get('ComplianceType', '')
                if compliance_type == 'COMPLIANT':
                    compliance_results['compliant'] += 1
                elif compliance_type == 'NON_COMPLIANT':
                    compliance_results['non_compliant'] += 1
                else:
                    compliance_results['not_applicable'] += 1
            
            logger.info(f"Security policy validation complete: {compliance_results}")
            
            return compliance_results
        
        except ClientError as e:
            error_msg = f"Failed to validate security policies: {e}"
            logger.error(error_msg)
            raise AutomationError(error_msg) from e
    
    def deploy_secure_infrastructure(
        self,
        variables: Optional[Dict] = None,
        auto_approve: bool = False
    ) -> Dict:
        """Deploy secure infrastructure using Terraform.
        
        Args:
            variables: Optional Terraform variables
            auto_approve: If True, skip confirmation prompts
            
        Returns:
            Deployment result dictionary
        """
        try:
            # Initialize
            self.terraform_init()
            
            # Validate
            self.terraform_validate()
            
            # Plan
            plan_result = self.terraform_plan(variables=variables)
            
            # Apply
            apply_result = self.terraform_apply(auto_approve=auto_approve)
            
            # Validate security policies
            validation_result = self.validate_security_policies()
            
            return {
                'success': True,
                'plan': plan_result,
                'apply': apply_result,
                'security_validation': validation_result
            }
        
        except TerraformError:
            raise
        except Exception as e:
            error_msg = f"Failed to deploy secure infrastructure: {e}"
            logger.error(error_msg, exc_info=True)
            raise AutomationError(error_msg) from e


# Example usage
if __name__ == "__main__":
    manager = SecurityAutomationManager(
        terraform_dir='./terraform',
        region_name='us-east-1'
    )
    
    # Deploy secure infrastructure
    variables = {
        'aws_region': 'us-east-1',
        'environment': 'production',
        'bucket_name_prefix': 'secure-bucket'
    }
    
    try:
        result = manager.deploy_secure_infrastructure(
            variables=variables,
            auto_approve=False  # Set to True for automation
        )
        
        print("Deployment successful!")
        print(f"Outputs: {result['apply']['outputs']}")
        print(f"Security Validation: {result['security_validation']}")
    
    except AutomationError as e:
        print(f"Deployment failed: {e}")

See Step 1 for the complete Terraform configuration. The configuration includes:

1. AWS Config Rules for automated security policy validation
2. S3 Bucket Policies with encryption and access controls
3. IAM Policies for least-privilege access
4. Automated Validation through the Python manager

Example Security Policy Validation:

manager = SecurityAutomationManager(terraform_dir='./terraform')

# Validate security policies after deployment
validation = manager.validate_security_policies()
print(f"Compliant: {validation['compliant']}/{validation['total_rules']}")

Additional Config Rules Example:

# Additional security config rules
resource "aws_config_config_rule" "encrypted_volumes" {
  name        = "encrypted-volumes"
  description = "Checks that EBS volumes are encrypted"

  source {
    owner             = "AWS"
    source_identifier = "ENCRYPTED_VOLUMES"
  }
}

resource "aws_config_config_rule" "root_access_keys" {
  name        = "root-access-keys-check"
  description = "Checks that root user has no access keys"

  source {
    owner             = "AWS"
    source_identifier = "ROOT_ACCESS_KEY_CHECK"
  }
}