Infrastructure Compliance: Automated Auditing, Policy Enforcement

# AWS Config Rules for CIS Compliance
resources:
  - resource_type: AWS::EC2::Instance
    properties:
      imageId: ami-*  # Not CIS compliant - use hardened AMI
    mode: periodic
    type: AWSManagedRulesCommon
    parameters:
      - name: DaysToExpiration
        value: 90

  - resource_type: AWS::S3::Bucket
    properties:
      bucketEncryption:
        serverSideEncryptionConfiguration:
          - serverSideEncryptionByDefault:
              sseAlgorithm: AES256
    mode: periodic

# CloudGuard CSPM Rules
rules:
  - id: "CIS-2.1.1"
    description: "Ensure MFA is enabled for the root account"
    severity: critical
    remediation:
      - action: "Enable MFA on root account"
        link: "https://console.aws.amazon.com/iam"

  - id: "CIS-2.1.2"
    description: "Ensure MFA is enabled for all IAM users"
    severity: high
    remediation:
      - action: "Enable MFA for IAM users"
        link: "https://console.aws.amazon.com/iam"

  - id: "CIS-2.3.1"
    description: "Ensure credentials unused for 90 days are disabled"
    severity: medium

# Install Prowler
pip install prowler
prowler aws --profile default

# Run specific checks
prowler aws -f cis -r us-east-1

# Export to multiple formats
prowler aws -M csv,json,html -o ./reports/

# prowler-config.yaml
output:
  format: json
  filename: prowler-report
  
checks:
  exclude:
    - extra7123  # Exclude specific check
    
compliance:
  frameworks:
    - cis
    - soc2
    - hipaa
    - pci

# Terraform AWS Config
resource "aws_config_configuration_recorder" "main" {
  name     = "main-recorder"
  role_arn = aws_iam_role.config_role.arn
  
  recording_group {
    all_supported                 = true
    include_global_resource_types = true
    resource_types_to_include    = []
  }
}

resource "aws_iam_role" "config_role" {
  name = "aws-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_policy" {
  role       = aws_iam_role.config_role.name
  policy_arn = "arn:aws:iam::aws:policy/service_role/AWSConfigRole"
}

# Custom Config Rule - Require Tags
resource "aws_config_config_rule" "require_tags" {
  name = "require-tags"
  
  source {
    owner             = "CUSTOM_LAMBDA"
    source_identifier = "arn:aws:lambda:us-east-1:123456789:function:require-tags"
  }
  
  scope {
    resources = ["AWS::EC2::Instance", "AWS::S3::Bucket"]
  }
}

#!/usr/bin/env python3
"""AWS Config Rule - Require Tags."""

import boto3
import json

config = boto3.client('config')

def evaluate_configuration_change(configuration_item):
    """Evaluate if resource has required tags."""
    
    required_tags = ['Environment', 'Owner', 'CostCenter']
    resource_tags = configuration_item.get('tags', {})
    
    missing_tags = [tag for tag in required_tags if tag not in resource_tags]
    
    if missing_tags:
        return {
            'complianceType': 'NON_COMPLIANT',
            'annotation': f"Missing required tags: {', '.join(missing_tags)}"
        }
    
    return {'complianceType': 'COMPLIANT'}

def handler(event, context):
    """Handle AWS Config evaluation."""
    
    invoking_event = json.loads(event['invokingEvent'])
    configuration_item = invoking_event['configurationItem']
    
    result = evaluate_configuration_item(configuration_item)
    
    config.put_evaluations(
        Evaluations=[
            {
                'ComplianceResourceType': configuration_item['resourceType'],
                'ComplianceResourceId': configuration_item['resourceId'],
                'ComplianceType': result['complianceType'],
                'Annotation': result.get('annotation'),
                'OrderingTimestamp': configuration_item['configurationItemCaptureTime']
            }
        ],
        ResultToken=event['resultToken']
    )

{
  "properties": {
    "displayName": "Require tags on resource groups",
    "description": "Enforces required tags on resource groups",
    "mode": "All",
    "parameters": {
      "tagName": {
        "type": "String",
        "displayName": "Tag Name",
        "description": "The name of the tag"
      }
    },
    "policyRule": {
      "if": {
        "allOf": [
          {
            "field": "type",
            "equals": "Microsoft.Resources/subscriptions/resourceGroups"
          },
          {
            "not": {
              "field": "[concat('tags[', parameters('tagName'), ']')]",
              "exists": "true"
            }
          }
        ]
      },
      "then": {
        "effect": "deny"
      }
    }
  }
}

# Assign policy
az policy assignment create \
  --name "require-tags" \
  --policy /path/to/policy.json \
  --params '{"tagName":"Environment"}' \
  --scope "/subscriptions/xxx"

# List non-compliant resources
az policy state list \
  --filter "complianceState eq 'NonCompliant'" \
  --query "[].{resource:resourceId, policy:policyAssignmentName}"

#!/usr/bin/env python3
"""SOC2 evidence collection automation."""

import boto3
import json
from datetime import datetime, timedelta

class SOC2EvidenceCollector:
    """Collect evidence for SOC2 compliance."""
    
    def __init__(self):
        self.iam = boto3.client('iam')
        self.cloudtrail = boto3.client('cloudtrail')
        self.config = boto3.client('config')
        
    def collect_access_review(self):
        """CC6.1: Access Control - Access Review"""
        
        # Get all IAM users
        users = self.iam.list_users()['Users']
        
        evidence = {
            'timestamp': datetime.now().isoformat(),
            'total_users': len(users),
            'users': []
        }
        
        for user in users:
            # Check MFA status
            mfa = self.iam.list_mfa_devices(UserName=user['UserName'])
            
            # Get attached policies
            policies = self.iam.list_attached_user_policies(
                UserName=user['UserName']
            )['AttachedPolicies']
            
            evidence['users'].append({
                'username': user['UserName'],
                'arn': user['Arn'],
                'mfa_enabled': len(mfa['MFADevices']) > 0,
                'policies': [p['PolicyName'] for p in policies],
                'created': user['CreateDate'].isoformat()
            })
        
        return evidence
    
    def collect_logging_evidence(self):
        """CC7.2: Logging - CloudTrail"""
        
        trails = self.cloudtrail.lookup_events(
            LookupAttributes=[
                {'AttributeKey': 'EventSource', 'AttributeValue': 's3.amazonaws.com'}
            ],
            MaxResults=100
        )
        
        return {
            'timestamp': datetime.now().isoformat(),
            'total_events': len(trails['Events']),
            'trail_status': 'Enabled'
        }
    
    def collect_encryption_evidence(self):
        """CC6.7: Encryption at Rest"""
        
        s3 = boto3.client('s3')
        buckets = s3.list_buckets()['Buckets']
        
        evidence = {
            'timestamp': datetime.now().isoformat(),
            'buckets': []
        }
        
        for bucket in buckets:
            encryption = s3.get_bucket_encryption(Bucket=bucket['Name'])
            evidence['buckets'].append({
                'name': bucket['Name'],
                'encrypted': 'ServerSideEncryptionConfiguration' in encryption
            })
        
        return evidence
    
    def generate_report(self):
        """Generate comprehensive evidence report."""
        
        report = {
            'generated_at': datetime.now().isoformat(),
            'access_review': self.collect_access_review(),
            'logging_evidence': self.collect_logging_evidence(),
            'encryption_evidence': self.collect_encryption_evidence()
        }
        
        # Save to S3
        s3 = boto3.client('s3')
        s3.put_object(
            Bucket='soc2-evidence-bucket',
            Key=f"evidence/{datetime.now().strftime('%Y%m%d')}.json",
            Body=json.dumps(report, indent=2)
        )
        
        return report

if __name__ == '__main__':
    collector = SOC2EvidenceCollector()
    report = collector.generate_report()
    print(f"Evidence collected: {json.dumps(report, indent=2)}")

# CloudWatch Dashboard for SOC2
{
  "widgets": [
    {
      "type": "metric",
      "properties": {
        "title": "Failed Login Attempts",
        "metrics": [
          ["AWS/CloudTrail", "ConsoleLogin", "ErrorCode", "FailedConsoleLogin"],
          [".", "AwsServiceEvent"]
        ],
        "period": 300,
        "stat": "Sum"
      }
    },
    {
      "type": "metric",
      "properties": {
        "title": "IAM Policy Changes",
        "metrics": [
          ["AWS/Config", "ResourceCompliance", "ComplianceType", "NON_COMPLIANT"],
          [".", "ComplianceType", "COMPLIANT"]
        ]
      }
    }
  ]
}

# Require resource limits
apiVersion: templates.gatekeeper.sh/v1beta1
kind: ConstraintTemplate
metadata:
  name: k8srequiredresources
spec:
  crd:
    spec:
      names:
        kind: K8sRequiredResources
  targets:
    - target: admission.k8s.gatekeeper.sh
      rego: |
        package k8srequiredresources
        
        violation[{"msg": msg}] {
          container := input.request.object.spec.containers[_]
          not container.resources
          msg := sprintf("Container %s must have resources defined", [container.name])
        }
        
        violation[{"msg": msg}] {
          container := input.request.object.spec.containers[_]
          not container.resources.limits
          msg := sprintf("Container %s must have resource limits", [container.name])
        }

#!/usr/bin/env python3
"""Generate compliance audit reports."""

import boto3
from datetime import datetime
import json

def generate_audit_report():
    """Generate automated audit report."""
    
    config = boto3.client('config')
    
    # Get compliance summary
    compliance = config.get_compliance_summary_by_config_rule()
    
    # Generate report
    report = {
        'report_date': datetime.now().isoformat(),
        'period': 'Q1 2026',
        'summary': {
            'compliant_rules': compliance.get('CompliantRules', 0),
            'non_compliant_rules': compliance.get('NonCompliantRules', 0),
            'total_rules': compliance.get('CompliantRules', 0) + compliance.get('NonCompliantRules', 0)
        },
        'findings': [],
        'recommendations': []
    }
    
    return report

if __name__ == '__main__':
    report = generate_audit_report()
    print(json.dumps(report, indent=2))