AWS DevSecOps Pipeline Security: Complete Automation Implementation Guide for 2025

• Introduction
  • Current Landscape Statistics
• DevSecOps Pipeline Security Architecture
  • Core Security Integration Points
  • AWS Services Integration
• Implementing Security-First CI/CD Pipeline
  • Complete CodePipeline with Integrated Security
• Container Security Integration
  • EKS Security with Policy Enforcement
• CI/CD Security Automation CloudFormation
• Best Practices and Recommendations
  • Implementation Guidelines
  • Security Considerations
  • Performance Optimization
• Implementation Roadmap
  • Phase 1: Foundation (Weeks 1-2)
  • Phase 2: Enhanced Security Scanning (Weeks 3-4)
  • Phase 3: Advanced Testing and Compliance (Weeks 5-6)
  • Phase 4: Container and Kubernetes Security (Weeks 7-8)
  • Phase 5: Optimization and Advanced Features (Weeks 9-10)
• Related Articles
• Additional Resources
  • Official Documentation
  • Tools and Frameworks
  • Industry Reports and Research
  • Community Resources
• Conclusion

Introduction

As organizations accelerate software delivery in 2025, the integration of security into DevOps pipelines has become mission-critical. With 88% of cloud security breaches attributed to human error and the average detection time remaining at 277 days, automated security validation within CI/CD processes represents the most effective defense against modern threats.

This comprehensive guide demonstrates how to implement end-to-end security automation in AWS DevSecOps pipelines, covering everything from infrastructure as code scanning to runtime security monitoring.

Current Landscape Statistics

• DevSecOps Adoption: 73% of organizations are actively implementing DevSecOps practices, with security automation being the top priority
• Pipeline Security Gaps: 67% of organizations lack comprehensive security testing in their CI/CD pipelines
• Vulnerability Detection: Automated security scanning reduces vulnerability discovery time from weeks to minutes
• Compliance Pressure: 89% of enterprises require automated compliance validation for regulatory requirements
• Cost of Delayed Security: Security issues found in production cost 100x more to fix than those caught during development

DevSecOps Pipeline Security Architecture

Core Security Integration Points

Modern DevSecOps pipelines must incorporate security at every stage to ensure continuous protection:

graph TB
    A[Developer Commit] --> B[Source Analysis]
    B --> C[Dependency Scanning]
    C --> D[SAST Analysis]
    D --> E[Container Build]
    E --> F[Image Scanning]
    F --> G[DAST Testing]
    G --> H[Infrastructure Scan]
    H --> I[Deployment]
    I --> J[Runtime Monitoring]
    
    K[Security Gates] --> B
    K --> C
    K --> D
    K --> F
    K --> G
    K --> H
    
    L[Policy Engine] --> K
    M[Compliance Framework] --> L
    N[Threat Intelligence] --> L

AWS Services Integration

Core DevSecOps Services:

• AWS CodeCommit for secure source control
• AWS CodeBuild for build automation with security scanning
• AWS CodePipeline for orchestrated security workflows
• AWS CodeDeploy for secure deployment automation

Security Integration Services:

• Amazon Inspector for vulnerability assessment
• AWS Security Hub for centralized security findings
• Amazon GuardDuty for runtime threat detection
• AWS Config for compliance validation

Container Security:

• Amazon ECR for secure container registry with scanning
• AWS Fargate for secure container runtime
• Amazon EKS with security policy enforcement

Implementing Security-First CI/CD Pipeline

Complete CodePipeline with Integrated Security

import boto3
import json
from datetime import datetime
from typing import Dict, List, Optional

class SecureDevSecOpsPipeline:
    def __init__(self):
        self.codepipeline = boto3.client('codepipeline')
        self.codebuild = boto3.client('codebuild')
        self.codecommit = boto3.client('codecommit')
        self.ecr = boto3.client('ecr')
        self.security_hub = boto3.client('securityhub')
        self.inspector = boto3.client('inspector2')
        
    def create_secure_pipeline(self, pipeline_name: str, repo_name: str) -> Dict:
        """Create a comprehensive DevSecOps pipeline with security gates"""
        
        try:
            # Create the pipeline structure
            pipeline_definition = {
                'name': pipeline_name,
                'roleArn': f'arn:aws:iam::{self.get_account_id()}:role/CodePipelineRole',
                'artifactStore': {
                    'type': 'S3',
                    'location': f'{pipeline_name}-artifacts-{self.get_account_id()}'
                },
                'stages': [
                    self.create_source_stage(repo_name),
                    self.create_security_analysis_stage(),
                    self.create_build_stage(),
                    self.create_security_testing_stage(),
                    self.create_deployment_stage()
                ]
            }
            
            # Create the pipeline
            response = self.codepipeline.create_pipeline(pipeline=pipeline_definition)
            
            # Setup security scanning projects
            self.setup_security_scanning_projects(pipeline_name)
            
            # Configure security notifications
            self.setup_security_notifications(pipeline_name)
            
            return {
                'pipeline_name': pipeline_name,
                'pipeline_arn': response['pipeline']['metadata']['pipelineArn'],
                'status': 'created',
                'security_features': [
                    'SAST scanning',
                    'DAST testing',
                    'Dependency scanning',
                    'Container scanning',
                    'Infrastructure scanning',
                    'Compliance validation'
                ]
            }
            
        except Exception as e:
            return {'status': 'failed', 'error': str(e)}
    
    def create_source_stage(self, repo_name: str) -> Dict:
        """Create source stage with security validation"""
        
        return {
            'name': 'Source',
            'actions': [
                {
                    'name': 'SourceAction',
                    'actionTypeId': {
                        'category': 'Source',
                        'owner': 'AWS',
                        'provider': 'CodeCommit',
                        'version': '1'
                    },
                    'configuration': {
                        'RepositoryName': repo_name,
                        'BranchName': 'main',
                        'PollForSourceChanges': 'false'
                    },
                    'outputArtifacts': [
                        {'name': 'SourceOutput'}
                    ]
                }
            ]
        }
    
    def create_security_analysis_stage(self) -> Dict:
        """Create comprehensive security analysis stage"""
        
        return {
            'name': 'SecurityAnalysis',
            'actions': [
                {
                    'name': 'SecretScanning',
                    'actionTypeId': {
                        'category': 'Build',
                        'owner': 'AWS',
                        'provider': 'CodeBuild',
                        'version': '1'
                    },
                    'configuration': {
                        'ProjectName': 'secret-scanning-project'
                    },
                    'inputArtifacts': [
                        {'name': 'SourceOutput'}
                    ],
                    'outputArtifacts': [
                        {'name': 'SecretScanOutput'}
                    ],
                    'runOrder': 1
                },
                {
                    'name': 'DependencyScanning',
                    'actionTypeId': {
                        'category': 'Build',
                        'owner': 'AWS',
                        'provider': 'CodeBuild',
                        'version': '1'
                    },
                    'configuration': {
                        'ProjectName': 'dependency-scanning-project'
                    },
                    'inputArtifacts': [
                        {'name': 'SourceOutput'}
                    ],
                    'outputArtifacts': [
                        {'name': 'DependencyScanOutput'}
                    ],
                    'runOrder': 1
                },
                {
                    'name': 'SASTAnalysis',
                    'actionTypeId': {
                        'category': 'Build',
                        'owner': 'AWS',
                        'provider': 'CodeBuild',
                        'version': '1'
                    },
                    'configuration': {
                        'ProjectName': 'sast-analysis-project'
                    },
                    'inputArtifacts': [
                        {'name': 'SourceOutput'}
                    ],
                    'outputArtifacts': [
                        {'name': 'SASTOutput'}
                    ],
                    'runOrder': 2
                }
            ]
        }
    
    def create_build_stage(self) -> Dict:
        """Create build stage with container security scanning"""
        
        return {
            'name': 'Build',
            'actions': [
                {
                    'name': 'BuildApplication',
                    'actionTypeId': {
                        'category': 'Build',
                        'owner': 'AWS',
                        'provider': 'CodeBuild',
                        'version': '1'
                    },
                    'configuration': {
                        'ProjectName': 'secure-build-project'
                    },
                    'inputArtifacts': [
                        {'name': 'SourceOutput'}
                    ],
                    'outputArtifacts': [
                        {'name': 'BuildOutput'}
                    ],
                    'runOrder': 1
                },
                {
                    'name': 'ContainerScanning',
                    'actionTypeId': {
                        'category': 'Build',
                        'owner': 'AWS',
                        'provider': 'CodeBuild',
                        'version': '1'
                    },
                    'configuration': {
                        'ProjectName': 'container-scanning-project'
                    },
                    'inputArtifacts': [
                        {'name': 'BuildOutput'}
                    ],
                    'outputArtifacts': [
                        {'name': 'ContainerScanOutput'}
                    ],
                    'runOrder': 2
                }
            ]
        }
    
    def create_security_testing_stage(self) -> Dict:
        """Create security testing stage with DAST and infrastructure scanning"""
        
        return {
            'name': 'SecurityTesting',
            'actions': [
                {
                    'name': 'DASTTesting',
                    'actionTypeId': {
                        'category': 'Build',
                        'owner': 'AWS',
                        'provider': 'CodeBuild',
                        'version': '1'
                    },
                    'configuration': {
                        'ProjectName': 'dast-testing-project'
                    },
                    'inputArtifacts': [
                        {'name': 'BuildOutput'}
                    ],
                    'outputArtifacts': [
                        {'name': 'DASTOutput'}
                    ],
                    'runOrder': 1
                },
                {
                    'name': 'InfrastructureScanning',
                    'actionTypeId': {
                        'category': 'Build',
                        'owner': 'AWS',
                        'provider': 'CodeBuild',
                        'version': '1'
                    },
                    'configuration': {
                        'ProjectName': 'infrastructure-scanning-project'
                    },
                    'inputArtifacts': [
                        {'name': 'BuildOutput'}
                    ],
                    'outputArtifacts': [
                        {'name': 'InfraScanOutput'}
                    ],
                    'runOrder': 1
                },
                {
                    'name': 'ComplianceValidation',
                    'actionTypeId': {
                        'category': 'Build',
                        'owner': 'AWS',
                        'provider': 'CodeBuild',
                        'version': '1'
                    },
                    'configuration': {
                        'ProjectName': 'compliance-validation-project'
                    },
                    'inputArtifacts': [
                        {'name': 'BuildOutput'}
                    ],
                    'outputArtifacts': [
                        {'name': 'ComplianceOutput'}
                    ],
                    'runOrder': 2
                }
            ]
        }
    
    def create_deployment_stage(self) -> Dict:
        """Create secure deployment stage with runtime monitoring"""
        
        return {
            'name': 'Deploy',
            'actions': [
                {
                    'name': 'DeployToStaging',
                    'actionTypeId': {
                        'category': 'Deploy',
                        'owner': 'AWS',
                        'provider': 'CloudFormation',
                        'version': '1'
                    },
                    'configuration': {
                        'ActionMode': 'CREATE_UPDATE',
                        'StackName': 'staging-stack',
                        'TemplatePath': 'BuildOutput::template.yaml',
                        'Capabilities': 'CAPABILITY_IAM',
                        'RoleArn': f'arn:aws:iam::{self.get_account_id()}:role/CloudFormationRole'
                    },
                    'inputArtifacts': [
                        {'name': 'BuildOutput'}
                    ],
                    'runOrder': 1
                },
                {
                    'name': 'SecurityValidation',
                    'actionTypeId': {
                        'category': 'Invoke',
                        'owner': 'AWS',
                        'provider': 'Lambda',
                        'version': '1'
                    },
                    'configuration': {
                        'FunctionName': 'security-validation-function'
                    },
                    'runOrder': 2
                }
            ]
        }
    
    def setup_security_scanning_projects(self, pipeline_name: str):
        """Setup CodeBuild projects for various security scans"""
        
        # Secret Scanning Project
        self.create_security_build_project(
            project_name='secret-scanning-project',
            description='Scan code for hardcoded secrets and credentials',
            buildspec=self.get_secret_scanning_buildspec()
        )
        
        # Dependency Scanning Project
        self.create_security_build_project(
            project_name='dependency-scanning-project',
            description='Scan dependencies for known vulnerabilities',
            buildspec=self.get_dependency_scanning_buildspec()
        )
        
        # SAST Analysis Project
        self.create_security_build_project(
            project_name='sast-analysis-project',
            description='Static Application Security Testing',
            buildspec=self.get_sast_analysis_buildspec()
        )
        
        # Container Scanning Project
        self.create_security_build_project(
            project_name='container-scanning-project',
            description='Container image vulnerability scanning',
            buildspec=self.get_container_scanning_buildspec()
        )
        
        # DAST Testing Project
        self.create_security_build_project(
            project_name='dast-testing-project',
            description='Dynamic Application Security Testing',
            buildspec=self.get_dast_testing_buildspec()
        )
        
        # Infrastructure Scanning Project
        self.create_security_build_project(
            project_name='infrastructure-scanning-project',
            description='Infrastructure as Code security scanning',
            buildspec=self.get_infrastructure_scanning_buildspec()
        )
        
        # Compliance Validation Project
        self.create_security_build_project(
            project_name='compliance-validation-project',
            description='Automated compliance framework validation',
            buildspec=self.get_compliance_validation_buildspec()
        )
    
    def create_security_build_project(self, project_name: str, description: str, buildspec: str):
        """Create a CodeBuild project for security scanning"""
        
        try:
            self.codebuild.create_project(
                name=project_name,
                description=description,
                source={
                    'type': 'CODEPIPELINE',
                    'buildspec': buildspec
                },
                artifacts={
                    'type': 'CODEPIPELINE'
                },
                environment={
                    'type': 'LINUX_CONTAINER',
                    'image': 'aws/codebuild/amazonlinux2-x86_64-standard:3.0',
                    'computeType': 'BUILD_GENERAL1_MEDIUM',
                    'privilegedMode': True
                },
                serviceRole=f'arn:aws:iam::{self.get_account_id()}:role/CodeBuildRole'
            )
        except Exception as e:
            print(f"Failed to create {project_name}: {str(e)}")
    
    def get_secret_scanning_buildspec(self) -> str:
        """Get buildspec for secret scanning"""
        
        return """
version: 0.2

phases:
  install:
    runtime-versions:
      python: 3.9
    commands:
      - pip install truffleHog3 detect-secrets
      - npm install -g secretlint
  
  pre_build:
    commands:
      - echo Starting secret scanning...
      - export SCAN_RESULTS_FILE="secret_scan_results.json"
  
  build:
    commands:
      # TruffleHog3 scanning
      - trufflehog3 --format json --output $SCAN_RESULTS_FILE .
      
      # detect-secrets scanning
      - detect-secrets scan --all-files . > detect_secrets_results.json
      
      # secretlint scanning
      - secretlint "**/*" --format json > secretlint_results.json
      
      # Aggregate results
      - python -c "
import json
import sys

# Load scan results
results = {
    'trufflehog': [],
    'detect_secrets': [],
    'secretlint': [],
    'summary': {'total_secrets': 0, 'high_risk': 0}
}

# Process TruffleHog results
try:
    with open('$SCAN_RESULTS_FILE', 'r') as f:
        trufflehog_results = json.load(f)
        results['trufflehog'] = trufflehog_results
        results['summary']['total_secrets'] += len(trufflehog_results)
except:
    pass

# Process detect-secrets results
try:
    with open('detect_secrets_results.json', 'r') as f:
        detect_secrets_results = json.load(f)
        results['detect_secrets'] = detect_secrets_results
        if 'results' in detect_secrets_results:
            for file_results in detect_secrets_results['results'].values():
                results['summary']['total_secrets'] += len(file_results)
except:
    pass

# Process secretlint results
try:
    with open('secretlint_results.json', 'r') as f:
        secretlint_results = json.load(f)
        results['secretlint'] = secretlint_results
        if isinstance(secretlint_results, list):
            results['summary']['total_secrets'] += len(secretlint_results)
except:
    pass

# Determine high risk secrets
results['summary']['high_risk'] = sum(1 for result in results['trufflehog'] 
                                    if result.get('reason', '').lower() in ['high_entropy', 'regex'])

# Save aggregated results
with open('aggregated_secret_scan.json', 'w') as f:
    json.dump(results, f, indent=2)

# Fail build if secrets found
if results['summary']['total_secrets'] > 0:
    print(f\"SECURITY ALERT: {results['summary']['total_secrets']} potential secrets found!\")
    print(f\"High risk secrets: {results['summary']['high_risk']}\")
    sys.exit(1)
else:
    print('No secrets detected in source code.')
"
  
  post_build:
    commands:
      - echo Secret scanning completed
      - |
        if [ $CODEBUILD_BUILD_SUCCEEDING -eq 0 ]; then
          echo "Secret scanning failed - stopping pipeline"
          aws sns publish --topic-arn arn:aws:sns:us-east-1:$AWS_ACCOUNT_ID:devsecops-alerts \
            --message "Secret scanning failed in pipeline. Review results immediately."
        fi

artifacts:
  files:
    - aggregated_secret_scan.json
    - secret_scan_results.json
    - detect_secrets_results.json
    - secretlint_results.json
"""
    
    def get_dependency_scanning_buildspec(self) -> str:
        """Get buildspec for dependency scanning"""
        
        return """
version: 0.2

phases:
  install:
    runtime-versions:
      python: 3.9
      nodejs: 16
    commands:
      - pip install safety pipenv
      - npm install -g audit-ci retire
      - curl -sfL https://raw.githubusercontent.com/aquasecurity/trivy/main/contrib/install.sh | sh -s -- -b /usr/local/bin
  
  pre_build:
    commands:
      - echo Starting dependency vulnerability scanning...
  
  build:
    commands:
      # Python dependency scanning with Safety
      - |
        if [ -f "requirements.txt" ]; then
          echo "Scanning Python dependencies..."
          safety check --json --output safety_results.json || true
        fi
      
      # Node.js dependency scanning
      - |
        if [ -f "package.json" ]; then
          echo "Scanning Node.js dependencies..."
          npm audit --json > npm_audit_results.json || true
          retire --js --json > retire_results.json || true
        fi
      
      # Container dependency scanning with Trivy
      - |
        if [ -f "Dockerfile" ]; then
          echo "Scanning container dependencies..."
          trivy fs --format json --output trivy_results.json .
        fi
      
      # Aggregate and analyze results
      - python -c "
import json
import sys

vulnerability_summary = {
    'total_vulnerabilities': 0,
    'critical': 0,
    'high': 0,
    'medium': 0,
    'low': 0,
    'sources': []
}

# Process Safety results (Python)
try:
    with open('safety_results.json', 'r') as f:
        safety_data = json.load(f)
        for vuln in safety_data:
            vulnerability_summary['total_vulnerabilities'] += 1
            # Safety doesn't provide severity, assume medium
            vulnerability_summary['medium'] += 1
        vulnerability_summary['sources'].append('safety')
except:
    pass

# Process NPM Audit results
try:
    with open('npm_audit_results.json', 'r') as f:
        npm_data = json.load(f)
        if 'vulnerabilities' in npm_data:
            for vuln_name, vuln_data in npm_data['vulnerabilities'].items():
                severity = vuln_data.get('severity', 'unknown').lower()
                vulnerability_summary['total_vulnerabilities'] += 1
                if severity in vulnerability_summary:
                    vulnerability_summary[severity] += 1
        vulnerability_summary['sources'].append('npm_audit')
except:
    pass

# Process Trivy results
try:
    with open('trivy_results.json', 'r') as f:
        trivy_data = json.load(f)
        if 'Results' in trivy_data:
            for result in trivy_data['Results']:
                if 'Vulnerabilities' in result:
                    for vuln in result['Vulnerabilities']:
                        severity = vuln.get('Severity', 'unknown').lower()
                        vulnerability_summary['total_vulnerabilities'] += 1
                        if severity in vulnerability_summary:
                            vulnerability_summary[severity] += 1
        vulnerability_summary['sources'].append('trivy')
except:
    pass

# Save summary
with open('dependency_scan_summary.json', 'w') as f:
    json.dump(vulnerability_summary, f, indent=2)

print(f'Dependency scan complete:')
print(f'Total vulnerabilities: {vulnerability_summary[\"total_vulnerabilities\"]}')
print(f'Critical: {vulnerability_summary[\"critical\"]}')
print(f'High: {vulnerability_summary[\"high\"]}')
print(f'Medium: {vulnerability_summary[\"medium\"]}')
print(f'Low: {vulnerability_summary[\"low\"]}')

# Fail build if critical or high vulnerabilities found
if vulnerability_summary['critical'] > 0 or vulnerability_summary['high'] > 5:
    print('SECURITY GATE FAILURE: Critical or excessive high vulnerabilities found')
    sys.exit(1)
"
  
  post_build:
    commands:
      - echo Dependency scanning completed
      - |
        if [ $CODEBUILD_BUILD_SUCCEEDING -eq 0 ]; then
          echo "Dependency scanning failed - stopping pipeline"
          aws sns publish --topic-arn arn:aws:sns:us-east-1:$AWS_ACCOUNT_ID:devsecops-alerts \
            --message "Dependency vulnerability scan failed. Critical vulnerabilities detected."
        fi

artifacts:
  files:
    - dependency_scan_summary.json
    - safety_results.json
    - npm_audit_results.json
    - retire_results.json
    - trivy_results.json
"""
    
    def get_sast_analysis_buildspec(self) -> str:
        """Get buildspec for SAST analysis"""
        
        return """
version: 0.2

phases:
  install:
    runtime-versions:
      python: 3.9
      java: corretto11
    commands:
      # Install SonarQube Scanner
      - wget https://binaries.sonarsource.com/Distribution/sonar-scanner-cli/sonar-scanner-cli-4.7.0.2747-linux.zip
      - unzip sonar-scanner-cli-4.7.0.2747-linux.zip
      - export PATH=$PATH:$(pwd)/sonar-scanner-4.7.0.2747-linux/bin
      
      # Install Bandit for Python
      - pip install bandit[toml]
      
      # Install ESLint security plugin for JavaScript
      - npm install -g eslint eslint-plugin-security
      
      # Install Semgrep
      - pip install semgrep
  
  pre_build:
    commands:
      - echo Starting Static Application Security Testing...
  
  build:
    commands:
      # Python SAST with Bandit
      - |
        if find . -name "*.py" -type f | head -1 | grep -q "\.py$"; then
          echo "Running Bandit for Python..."
          bandit -r . -f json -o bandit_results.json || true
        fi
      
      # JavaScript SAST with ESLint Security
      - |
        if [ -f "package.json" ]; then
          echo "Running ESLint Security for JavaScript..."
          echo '{
            "extends": ["plugin:security/recommended"],
            "plugins": ["security"],
            "rules": {
              "security/detect-buffer-noassert": "error",
              "security/detect-child-process": "error",
              "security/detect-disable-mustache-escape": "error",
              "security/detect-eval-with-expression": "error",
              "security/detect-no-csrf-before-method-override": "error",
              "security/detect-non-literal-fs-filename": "error",
              "security/detect-non-literal-regexp": "error",
              "security/detect-non-literal-require": "error",
              "security/detect-object-injection": "error",
              "security/detect-possible-timing-attacks": "error",
              "security/detect-pseudoRandomBytes": "error",
              "security/detect-unsafe-regex": "error"
            }
          }' > .eslintrc.json
          
          npx eslint . --format json > eslint_security_results.json || true
        fi
      
      # Multi-language SAST with Semgrep
      - |
        echo "Running Semgrep for multi-language analysis..."
        semgrep --config=auto --json --output=semgrep_results.json . || true
      
      # Aggregate SAST results
      - python -c "
import json
import sys

sast_summary = {
    'total_issues': 0,
    'critical': 0,
    'high': 0,
    'medium': 0,
    'low': 0,
    'tools_used': []
}

# Process Bandit results
try:
    with open('bandit_results.json', 'r') as f:
        bandit_data = json.load(f)
        if 'results' in bandit_data:
            for result in bandit_data['results']:
                issue_severity = result.get('issue_severity', 'MEDIUM').upper()
                sast_summary['total_issues'] += 1
                if issue_severity == 'HIGH':
                    sast_summary['high'] += 1
                elif issue_severity == 'MEDIUM':
                    sast_summary['medium'] += 1
                else:
                    sast_summary['low'] += 1
        sast_summary['tools_used'].append('bandit')
except:
    pass

# Process ESLint Security results
try:
    with open('eslint_security_results.json', 'r') as f:
        eslint_data = json.load(f)
        for file_result in eslint_data:
            if 'messages' in file_result:
                for message in file_result['messages']:
                    severity = message.get('severity', 1)
                    sast_summary['total_issues'] += 1
                    if severity == 2:  # Error
                        sast_summary['high'] += 1
                    else:  # Warning
                        sast_summary['medium'] += 1
        sast_summary['tools_used'].append('eslint_security')
except:
    pass

# Process Semgrep results
try:
    with open('semgrep_results.json', 'r') as f:
        semgrep_data = json.load(f)
        if 'results' in semgrep_data:
            for result in semgrep_data['results']:
                severity = result.get('extra', {}).get('severity', 'INFO').upper()
                sast_summary['total_issues'] += 1
                if severity == 'ERROR':
                    sast_summary['high'] += 1
                elif severity == 'WARNING':
                    sast_summary['medium'] += 1
                else:
                    sast_summary['low'] += 1
        sast_summary['tools_used'].append('semgrep')
except:
    pass

# Save summary
with open('sast_summary.json', 'w') as f:
    json.dump(sast_summary, f, indent=2)

print(f'SAST analysis complete:')
print(f'Total issues: {sast_summary[\"total_issues\"]}')
print(f'Critical: {sast_summary[\"critical\"]}')
print(f'High: {sast_summary[\"high\"]}')
print(f'Medium: {sast_summary[\"medium\"]}')
print(f'Low: {sast_summary[\"low\"]}')

# Security gate: fail if critical issues or too many high issues
if sast_summary['critical'] > 0 or sast_summary['high'] > 10:
    print('SECURITY GATE FAILURE: Critical issues or excessive high-severity issues found')
    sys.exit(1)
"
  
  post_build:
    commands:
      - echo SAST analysis completed
      - |
        if [ $CODEBUILD_BUILD_SUCCEEDING -eq 0 ]; then
          echo "SAST analysis failed - stopping pipeline"
          aws sns publish --topic-arn arn:aws:sns:us-east-1:$AWS_ACCOUNT_ID:devsecops-alerts \
            --message "SAST analysis failed. Critical security issues detected in code."
        fi

artifacts:
  files:
    - sast_summary.json
    - bandit_results.json
    - eslint_security_results.json
    - semgrep_results.json
"""
    
    def get_container_scanning_buildspec(self) -> str:
        """Get buildspec for container scanning"""
        
        return """
version: 0.2

phases:
  install:
    runtime-versions:
      docker: 20
    commands:
      # Install Trivy for container scanning
      - curl -sfL https://raw.githubusercontent.com/aquasecurity/trivy/main/contrib/install.sh | sh -s -- -b /usr/local/bin
      
      # Install Docker Bench Security
      - git clone https://github.com/docker/docker-bench-security.git
      
      # Install Hadolint for Dockerfile linting
      - wget https://github.com/hadolint/hadolint/releases/latest/download/hadolint-Linux-x86_64
      - chmod +x hadolint-Linux-x86_64
      - mv hadolint-Linux-x86_64 /usr/local/bin/hadolint
  
  pre_build:
    commands:
      - echo Starting container security scanning...
      - echo Logging in to Amazon ECR...
      - aws ecr get-login-password --region $AWS_DEFAULT_REGION | docker login --username AWS --password-stdin $AWS_ACCOUNT_ID.dkr.ecr.$AWS_DEFAULT_REGION.amazonaws.com
  
  build:
    commands:
      # Dockerfile security linting with Hadolint
      - |
        if [ -f "Dockerfile" ]; then
          echo "Linting Dockerfile..."
          hadolint --format json Dockerfile > hadolint_results.json || true
        fi
      
      # Build container image
      - |
        if [ -f "Dockerfile" ]; then
          echo "Building container image..."
          IMAGE_TAG=$IMAGE_REPO_NAME:$CODEBUILD_RESOLVED_SOURCE_VERSION
          docker build -t $IMAGE_TAG .
          
          # Scan the built image with Trivy
          echo "Scanning container image for vulnerabilities..."
          trivy image --format json --output trivy_image_results.json $IMAGE_TAG
          
          # Scan for secrets in image
          trivy image --format json --scanners secret --output trivy_secret_results.json $IMAGE_TAG
          
          # Scan for misconfigurations
          trivy image --format json --scanners config --output trivy_config_results.json $IMAGE_TAG
        fi
      
      # Run Docker Bench Security if Docker daemon is available
      - |
        if docker info > /dev/null 2>&1; then
          echo "Running Docker Bench Security..."
          cd docker-bench-security
          ./docker-bench-security.sh -c container_images > ../docker_bench_results.log 2>&1 || true
          cd ..
        fi
      
      # Aggregate container security results
      - python3 -c "
import json
import sys
import re

container_summary = {
    'dockerfile_issues': 0,
    'vulnerabilities': {
        'critical': 0,
        'high': 0,
        'medium': 0,
        'low': 0,
        'total': 0
    },
    'secrets_found': 0,
    'misconfigurations': 0,
    'compliance_issues': 0
}

# Process Hadolint results
try:
    with open('hadolint_results.json', 'r') as f:
        hadolint_data = json.load(f)
        container_summary['dockerfile_issues'] = len(hadolint_data)
except:
    pass

# Process Trivy image vulnerability results
try:
    with open('trivy_image_results.json', 'r') as f:
        trivy_data = json.load(f)
        if 'Results' in trivy_data:
            for result in trivy_data['Results']:
                if 'Vulnerabilities' in result:
                    for vuln in result['Vulnerabilities']:
                        severity = vuln.get('Severity', 'UNKNOWN').lower()
                        container_summary['vulnerabilities']['total'] += 1
                        if severity in container_summary['vulnerabilities']:
                            container_summary['vulnerabilities'][severity] += 1
except:
    pass

# Process Trivy secret scan results
try:
    with open('trivy_secret_results.json', 'r') as f:
        trivy_secret_data = json.load(f)
        if 'Results' in trivy_secret_data:
            for result in trivy_secret_data['Results']:
                if 'Secrets' in result:
                    container_summary['secrets_found'] += len(result['Secrets'])
except:
    pass

# Process Trivy config scan results
try:
    with open('trivy_config_results.json', 'r') as f:
        trivy_config_data = json.load(f)
        if 'Results' in trivy_config_data:
            for result in trivy_config_data['Results']:
                if 'Misconfigurations' in result:
                    container_summary['misconfigurations'] += len(result['Misconfigurations'])
except:
    pass

# Process Docker Bench results
try:
    with open('docker_bench_results.log', 'r') as f:
        bench_content = f.read()
        # Count WARN and INFO items as compliance issues
        warn_count = len(re.findall(r'\[WARN\]', bench_content))
        info_count = len(re.findall(r'\[INFO\]', bench_content))
        container_summary['compliance_issues'] = warn_count + info_count
except:
    pass

# Save summary
with open('container_scan_summary.json', 'w') as f:
    json.dump(container_summary, f, indent=2)

print(f'Container security scan complete:')
print(f'Dockerfile issues: {container_summary[\"dockerfile_issues\"]}')
print(f'Total vulnerabilities: {container_summary[\"vulnerabilities\"][\"total\"]}')
print(f'Critical: {container_summary[\"vulnerabilities\"][\"critical\"]}')
print(f'High: {container_summary[\"vulnerabilities\"][\"high\"]}')
print(f'Secrets found: {container_summary[\"secrets_found\"]}')
print(f'Misconfigurations: {container_summary[\"misconfigurations\"]}')
print(f'Compliance issues: {container_summary[\"compliance_issues\"]}')

# Security gate: fail if critical vulnerabilities or secrets found
if (container_summary['vulnerabilities']['critical'] > 0 or 
    container_summary['secrets_found'] > 0 or
    container_summary['vulnerabilities']['high'] > 20):
    print('SECURITY GATE FAILURE: Critical vulnerabilities or secrets found in container')
    sys.exit(1)
"
      
      # Push image to ECR if security scan passes
      - |
        if [ $CODEBUILD_BUILD_SUCCEEDING -eq 1 ] && [ -f "Dockerfile" ]; then
          echo "Security scan passed, pushing image to ECR..."
          IMAGE_TAG=$IMAGE_REPO_NAME:$CODEBUILD_RESOLVED_SOURCE_VERSION
          docker tag $IMAGE_TAG $AWS_ACCOUNT_ID.dkr.ecr.$AWS_DEFAULT_REGION.amazonaws.com/$IMAGE_TAG
          docker push $AWS_ACCOUNT_ID.dkr.ecr.$AWS_DEFAULT_REGION.amazonaws.com/$IMAGE_TAG
        fi
  
  post_build:
    commands:
      - echo Container scanning completed
      - |
        if [ $CODEBUILD_BUILD_SUCCEEDING -eq 0 ]; then
          echo "Container scanning failed - stopping pipeline"
          aws sns publish --topic-arn arn:aws:sns:us-east-1:$AWS_ACCOUNT_ID:devsecops-alerts \
            --message "Container security scan failed. Critical vulnerabilities or secrets detected."
        fi

artifacts:
  files:
    - container_scan_summary.json
    - hadolint_results.json
    - trivy_image_results.json
    - trivy_secret_results.json
    - trivy_config_results.json
    - docker_bench_results.log

environment:
  variables:
    IMAGE_REPO_NAME: my-secure-app
"""
    
    def get_dast_testing_buildspec(self) -> str:
        """Get buildspec for DAST testing"""
        
        return """
version: 0.2

phases:
  install:
    runtime-versions:
      python: 3.9
    commands:
      # Install OWASP ZAP
      - wget https://github.com/zaproxy/zaproxy/releases/download/v2.12.0/ZAP_2_12_0_Linux.tar.gz
      - tar -xzf ZAP_2_12_0_Linux.tar.gz
      - export PATH=$PATH:$(pwd)/ZAP_2.12.0
      
      # Install Nikto
      - git clone https://github.com/sullo/nikto.git
      - chmod +x nikto/program/nikto.pl
      
      # Install additional security tools
      - pip install requests beautifulsoup4
  
  pre_build:
    commands:
      - echo Starting Dynamic Application Security Testing...
      - echo Waiting for application deployment to be ready...
      - sleep 30  # Wait for staging deployment
  
  build:
    commands:
      # Basic connectivity test
      - |
        APP_URL=${APPLICATION_URL:-"http://localhost:8080"}
        echo "Testing application at: $APP_URL"
        
        # Test if application is responding
        curl -f $APP_URL/health || {
          echo "Application health check failed"
          exit 1
        }
      
      # OWASP ZAP baseline scan
      - |
        echo "Running OWASP ZAP baseline scan..."
        zap.sh -cmd -quickurl $APP_URL -quickout zap_baseline_results.xml -quickprogress || true
        
        # Convert XML to JSON for easier processing
        python3 -c "
import xml.etree.ElementTree as ET
import json

try:
    tree = ET.parse('zap_baseline_results.xml')
    root = tree.getroot()
    
    alerts = []
    for alert in root.findall('.//alertitem'):
        alert_data = {
            'name': alert.find('name').text if alert.find('name') is not None else 'Unknown',
            'riskdesc': alert.find('riskdesc').text if alert.find('riskdesc') is not None else 'Unknown',
            'confidence': alert.find('confidence').text if alert.find('confidence') is not None else 'Unknown',
            'desc': alert.find('desc').text if alert.find('desc') is not None else 'No description',
            'uri': alert.find('uri').text if alert.find('uri') is not None else 'Unknown'
        }
        alerts.append(alert_data)
    
    with open('zap_results.json', 'w') as f:
        json.dump(alerts, f, indent=2)
        
except Exception as e:
    print(f'Error processing ZAP results: {e}')
    with open('zap_results.json', 'w') as f:
        json.dump([], f)
"
      
      # Nikto web server scanner
      - |
        echo "Running Nikto web server scan..."
        perl nikto/program/nikto.pl -h $APP_URL -Format json -output nikto_results.json || true
      
      # Custom security tests
      - |
        echo "Running custom security tests..."
        python3 -c "
import requests
import json
import sys
from urllib.parse import urljoin

APP_URL = '$APP_URL'
security_tests = {
    'sql_injection': [],
    'xss_tests': [],
    'header_security': {},
    'ssl_security': {},
    'information_disclosure': []
}

# Test for security headers
try:
    response = requests.get(APP_URL, timeout=10)
    headers = response.headers
    
    security_tests['header_security'] = {
        'x_frame_options': headers.get('X-Frame-Options', 'MISSING'),
        'x_content_type_options': headers.get('X-Content-Type-Options', 'MISSING'),
        'strict_transport_security': headers.get('Strict-Transport-Security', 'MISSING'),
        'content_security_policy': headers.get('Content-Security-Policy', 'MISSING'),
        'x_xss_protection': headers.get('X-XSS-Protection', 'MISSING')
    }
    
    # Check for information disclosure in headers
    disclosure_headers = ['Server', 'X-Powered-By', 'X-AspNet-Version']
    for header in disclosure_headers:
        if header in headers:
            security_tests['information_disclosure'].append({
                'type': 'header_disclosure',
                'header': header,
                'value': headers[header]
            })
    
    # Basic XSS tests (safe payloads for testing)
    xss_payloads = ['<script>alert(1)</script>', '<img src=x onerror=alert(1)>']
    for payload in xss_payloads:
        try:
            test_url = f'{APP_URL}?test={payload}'
            test_response = requests.get(test_url, timeout=5)
            if payload in test_response.text:
                security_tests['xss_tests'].append({
                    'payload': payload,
                    'reflected': True,
                    'url': test_url
                })
        except:
            pass

except Exception as e:
    print(f'Error in custom security tests: {e}')

# Save custom test results
with open('custom_security_tests.json', 'w') as f:
    json.dump(security_tests, f, indent=2)
"
      
      # Aggregate DAST results
      - python3 -c "
import json
import sys

dast_summary = {
    'total_vulnerabilities': 0,
    'high_risk': 0,
    'medium_risk': 0,
    'low_risk': 0,
    'security_headers': {
        'missing': 0,
        'total_checked': 5
    },
    'information_disclosure': 0,
    'potential_xss': 0
}

# Process ZAP results
try:
    with open('zap_results.json', 'r') as f:
        zap_data = json.load(f)
        for alert in zap_data:
            risk = alert.get('riskdesc', '').lower()
            dast_summary['total_vulnerabilities'] += 1
            if 'high' in risk:
                dast_summary['high_risk'] += 1
            elif 'medium' in risk:
                dast_summary['medium_risk'] += 1
            else:
                dast_summary['low_risk'] += 1
except:
    pass

# Process custom security tests
try:
    with open('custom_security_tests.json', 'r') as f:
        custom_data = json.load(f)
        
        # Count missing security headers
        headers = custom_data.get('header_security', {})
        for header, value in headers.items():
            if value == 'MISSING':
                dast_summary['security_headers']['missing'] += 1
        
        # Count information disclosure issues
        dast_summary['information_disclosure'] = len(custom_data.get('information_disclosure', []))
        
        # Count potential XSS issues
        dast_summary['potential_xss'] = len(custom_data.get('xss_tests', []))
        
except:
    pass

# Save summary
with open('dast_summary.json', 'w') as f:
    json.dump(dast_summary, f, indent=2)

print(f'DAST testing complete:')
print(f'Total vulnerabilities: {dast_summary[\"total_vulnerabilities\"]}')
print(f'High risk: {dast_summary[\"high_risk\"]}')
print(f'Medium risk: {dast_summary[\"medium_risk\"]}')
print(f'Low risk: {dast_summary[\"low_risk\"]}')
print(f'Missing security headers: {dast_summary[\"security_headers\"][\"missing\"]}/{dast_summary[\"security_headers\"][\"total_checked\"]}')
print(f'Information disclosure issues: {dast_summary[\"information_disclosure\"]}')
print(f'Potential XSS issues: {dast_summary[\"potential_xss\"]}')

# Security gate: fail if high risk vulnerabilities or missing critical headers
if (dast_summary['high_risk'] > 0 or 
    dast_summary['security_headers']['missing'] > 2 or
    dast_summary['potential_xss'] > 0):
    print('SECURITY GATE FAILURE: High risk vulnerabilities or critical security issues found')
    sys.exit(1)
"
  
  post_build:
    commands:
      - echo DAST testing completed
      - |
        if [ $CODEBUILD_BUILD_SUCCEEDING -eq 0 ]; then
          echo "DAST testing failed - stopping pipeline"
          aws sns publish --topic-arn arn:aws:sns:us-east-1:$AWS_ACCOUNT_ID:devsecops-alerts \
            --message "DAST testing failed. High risk vulnerabilities detected in running application."
        fi

artifacts:
  files:
    - dast_summary.json
    - zap_results.json
    - zap_baseline_results.xml
    - nikto_results.json
    - custom_security_tests.json

environment:
  variables:
    APPLICATION_URL: http://staging-lb-12345.us-east-1.elb.amazonaws.com
"""
    
    def get_infrastructure_scanning_buildspec(self) -> str:
        """Get buildspec for infrastructure scanning"""
        
        return """
version: 0.2

phases:
  install:
    runtime-versions:
      python: 3.9
    commands:
      # Install Checkov for infrastructure scanning
      - pip install checkov
      
      # Install Terraform security scanner (tfsec)
      - curl -s https://raw.githubusercontent.com/aquasecurity/tfsec/master/scripts/install_linux.sh | bash
      
      # Install cfn-lint for CloudFormation
      - pip install cfn-lint
      
      # Install Terrascan
      - curl -L "$(curl -s https://api.github.com/repos/tenable/terrascan/releases/latest | grep -o -E "https://.+?_Linux_x86_64.tar.gz")" > terrascan.tar.gz
      - tar -xf terrascan.tar.gz terrascan && rm terrascan.tar.gz
      - install terrascan /usr/local/bin
  
  pre_build:
    commands:
      - echo Starting Infrastructure as Code security scanning...
  
  build:
    commands:
      # CloudFormation template scanning
      - |
        echo "Scanning CloudFormation templates..."
        find . -name "*.yaml" -o -name "*.yml" -o -name "*.json" | grep -E "(template|cloudformation|cfn)" | while read template; do
          echo "Scanning CloudFormation template: $template"
          cfn-lint "$template" --format json > "cfn_lint_$(basename $template).json" 2>/dev/null || true
          checkov -f "$template" --framework cloudformation --output json > "checkov_cfn_$(basename $template).json" 2>/dev/null || true
        done
      
      # Terraform configuration scanning
      - |
        if find . -name "*.tf" | head -1 | grep -q "\.tf$"; then
          echo "Scanning Terraform configurations..."
          
          # Run tfsec
          tfsec . --format json > tfsec_results.json 2>/dev/null || true
          
          # Run Checkov on Terraform
          checkov -d . --framework terraform --output json > checkov_terraform_results.json 2>/dev/null || true
          
          # Run Terrascan
          terrascan scan -t terraform -f json > terrascan_results.json 2>/dev/null || true
        fi
      
      # Kubernetes manifest scanning
      - |
        if find . -name "*.yaml" -o -name "*.yml" | xargs grep -l "apiVersion\|kind:" | head -1; then
          echo "Scanning Kubernetes manifests..."
          find . -name "*.yaml" -o -name "*.yml" | xargs grep -l "apiVersion\|kind:" | while read manifest; do
            checkov -f "$manifest" --framework kubernetes --output json > "checkov_k8s_$(basename $manifest).json" 2>/dev/null || true
          done
        fi
      
      # Docker configuration scanning
      - |
        if [ -f "Dockerfile" ]; then
          echo "Scanning Dockerfile..."
          checkov -f Dockerfile --framework dockerfile --output json > checkov_docker_results.json 2>/dev/null || true
        fi
      
      # Aggregate infrastructure scanning results
      - python3 -c "
import json
import sys
import glob
import os

infra_summary = {
    'total_issues': 0,
    'critical': 0,
    'high': 0,
    'medium': 0,
    'low': 0,
    'tools_used': [],
    'scanned_files': []
}

# Process CFN-Lint results
cfn_lint_files = glob.glob('cfn_lint_*.json')
if cfn_lint_files:
    infra_summary['tools_used'].append('cfn-lint')
    for file in cfn_lint_files:
        try:
            with open(file, 'r') as f:
                cfn_data = json.load(f)
                for issue in cfn_data:
                    level = issue.get('Level', 'Warning').lower()
                    infra_summary['total_issues'] += 1
                    if level == 'error':
                        infra_summary['high'] += 1
                    else:
                        infra_summary['medium'] += 1
                infra_summary['scanned_files'].append(file)
        except:
            pass

# Process Checkov results
checkov_files = glob.glob('checkov_*.json')
if checkov_files:
    infra_summary['tools_used'].append('checkov')
    for file in checkov_files:
        try:
            with open(file, 'r') as f:
                checkov_data = json.load(f)
                if 'results' in checkov_data and 'failed_checks' in checkov_data['results']:
                    for check in checkov_data['results']['failed_checks']:
                        severity = check.get('severity', 'MEDIUM').upper()
                        infra_summary['total_issues'] += 1
                        if severity == 'CRITICAL':
                            infra_summary['critical'] += 1
                        elif severity == 'HIGH':
                            infra_summary['high'] += 1
                        elif severity == 'MEDIUM':
                            infra_summary['medium'] += 1
                        else:
                            infra_summary['low'] += 1
                infra_summary['scanned_files'].append(file)
        except:
            pass

# Process tfsec results
try:
    with open('tfsec_results.json', 'r') as f:
        tfsec_data = json.load(f)
        if 'results' in tfsec_data:
            for result in tfsec_data['results']:
                severity = result.get('severity', 'MEDIUM').upper()
                infra_summary['total_issues'] += 1
                if severity == 'CRITICAL':
                    infra_summary['critical'] += 1
                elif severity == 'HIGH':
                    infra_summary['high'] += 1
                elif severity == 'MEDIUM':
                    infra_summary['medium'] += 1
                else:
                    infra_summary['low'] += 1
    infra_summary['tools_used'].append('tfsec')
    infra_summary['scanned_files'].append('tfsec_results.json')
except:
    pass

# Process Terrascan results
try:
    with open('terrascan_results.json', 'r') as f:
        terrascan_data = json.load(f)
        if 'results' in terrascan_data and 'violations' in terrascan_data['results']:
            for violation in terrascan_data['results']['violations']:
                severity = violation.get('severity', 'MEDIUM').upper()
                infra_summary['total_issues'] += 1
                if severity == 'HIGH':
                    infra_summary['high'] += 1
                elif severity == 'MEDIUM':
                    infra_summary['medium'] += 1
                else:
                    infra_summary['low'] += 1
    infra_summary['tools_used'].append('terrascan')
    infra_summary['scanned_files'].append('terrascan_results.json')
except:
    pass

# Save summary
with open('infrastructure_scan_summary.json', 'w') as f:
    json.dump(infra_summary, f, indent=2)

print(f'Infrastructure scanning complete:')
print(f'Total issues: {infra_summary[\"total_issues\"]}')
print(f'Critical: {infra_summary[\"critical\"]}')
print(f'High: {infra_summary[\"high\"]}')
print(f'Medium: {infra_summary[\"medium\"]}')
print(f'Low: {infra_summary[\"low\"]}')
print(f'Tools used: {infra_summary[\"tools_used\"]}')
print(f'Files scanned: {len(infra_summary[\"scanned_files\"])}')

# Security gate: fail if critical issues or too many high issues
if infra_summary['critical'] > 0 or infra_summary['high'] > 15:
    print('SECURITY GATE FAILURE: Critical infrastructure security issues found')
    sys.exit(1)
"
  
  post_build:
    commands:
      - echo Infrastructure scanning completed
      - |
        if [ $CODEBUILD_BUILD_SUCCEEDING -eq 0 ]; then
          echo "Infrastructure scanning failed - stopping pipeline"
          aws sns publish --topic-arn arn:aws:sns:us-east-1:$AWS_ACCOUNT_ID:devsecops-alerts \
            --message "Infrastructure security scan failed. Critical misconfigurations detected."
        fi

artifacts:
  files:
    - infrastructure_scan_summary.json
    - cfn_lint_*.json
    - checkov_*.json
    - tfsec_results.json
    - terrascan_results.json
"""
    
    def get_compliance_validation_buildspec(self) -> str:
        """Get buildspec for compliance validation"""
        
        return """
version: 0.2

phases:
  install:
    runtime-versions:
      python: 3.9
    commands:
      # Install AWS Config Rules evaluation tools
      - pip install boto3 botocore
      
      # Install compliance scanning tools
      - pip install checkov prowler
      
      # Install Cloud Custodian for policy validation
      - pip install c7n c7n-org
  
  pre_build:
    commands:
      - echo Starting compliance validation...
  
  build:
    commands:
      # Run AWS Config compliance checks
      - |
        echo "Evaluating AWS Config compliance rules..."
        python3 -c "
import boto3
import json

config_client = boto3.client('config')
compliance_summary = {
    'total_rules': 0,
    'compliant': 0,
    'non_compliant': 0,
    'insufficient_data': 0,
    'not_applicable': 0,
    'compliance_score': 0.0
}

try:
    # Get compliance summary
    response = config_client.get_compliance_summary_by_config_rule()
    
    summary = response['ComplianceSummary']
    compliance_summary['compliant'] = summary.get('ComplianceByConfigRule', {}).get('COMPLIANT', 0)
    compliance_summary['non_compliant'] = summary.get('ComplianceByConfigRule', {}).get('NON_COMPLIANT', 0)
    compliance_summary['insufficient_data'] = summary.get('ComplianceByConfigRule', {}).get('INSUFFICIENT_DATA', 0)
    compliance_summary['not_applicable'] = summary.get('ComplianceByConfigRule', {}).get('NOT_APPLICABLE', 0)
    
    compliance_summary['total_rules'] = (compliance_summary['compliant'] + 
                                       compliance_summary['non_compliant'] + 
                                       compliance_summary['insufficient_data'] + 
                                       compliance_summary['not_applicable'])
    
    if compliance_summary['total_rules'] > 0:
        compliance_summary['compliance_score'] = (compliance_summary['compliant'] / 
                                                 compliance_summary['total_rules']) * 100
    
    # Get detailed compliance information
    rules_response = config_client.describe_config_rules()
    detailed_compliance = []
    
    for rule in rules_response['ConfigRules']:
        rule_name = rule['ConfigRuleName']
        try:
            compliance_response = config_client.get_compliance_details_by_config_rule(
                ConfigRuleName=rule_name
            )
            
            for result in compliance_response['EvaluationResults']:
                detailed_compliance.append({
                    'rule_name': rule_name,
                    'resource_type': result['EvaluationResultIdentifier']['EvaluationResultQualifier']['ResourceType'],
                    'resource_id': result['EvaluationResultIdentifier']['EvaluationResultQualifier']['ResourceId'],
                    'compliance_type': result['ComplianceType'],
                    'result_recorded_time': result['ResultRecordedTime'].isoformat() if 'ResultRecordedTime' in result else None
                })
        except Exception as e:
            print(f'Error getting compliance details for rule {rule_name}: {e}')
    
    # Save results
    with open('aws_config_compliance.json', 'w') as f:
        json.dump({
            'summary': compliance_summary,
            'detailed_results': detailed_compliance
        }, f, indent=2, default=str)
    
    print(f'AWS Config compliance check complete:')
    print(f'Total rules: {compliance_summary[\"total_rules\"]}')
    print(f'Compliant: {compliance_summary[\"compliant\"]}')
    print(f'Non-compliant: {compliance_summary[\"non_compliant\"]}')
    print(f'Compliance score: {compliance_summary[\"compliance_score\"]:.1f}%')

except Exception as e:
    print(f'Error running AWS Config compliance check: {e}')
    with open('aws_config_compliance.json', 'w') as f:
        json.dump({'error': str(e)}, f)
"
      
      # Run Prowler for CIS benchmark compliance
      - |
        echo "Running Prowler CIS benchmark checks..."
        prowler aws --output-modes json --output-file prowler_results.json || true
      
      # Run SOC 2 Type II compliance checks
      - |
        echo "Running SOC 2 compliance validation..."
        python3 -c "
import boto3
import json

soc2_compliance = {
    'security': {
        'score': 0,
        'checks': [],
        'weight': 0.3
    },
    'availability': {
        'score': 0,
        'checks': [],
        'weight': 0.2
    },
    'processing_integrity': {
        'score': 0,
        'checks': [],
        'weight': 0.2
    },
    'confidentiality': {
        'score': 0,
        'checks': [],
        'weight': 0.15
    },
    'privacy': {
        'score': 0,
        'checks': [],
        'weight': 0.15
    },
    'overall_score': 0
}

# Mock SOC 2 compliance checks (in real implementation, these would be actual checks)
security_checks = [
    {'name': 'Multi-factor authentication enabled', 'status': 'pass', 'weight': 0.3},
    {'name': 'Encryption at rest enabled', 'status': 'pass', 'weight': 0.25},
    {'name': 'Network segmentation implemented', 'status': 'pass', 'weight': 0.2},
    {'name': 'Access controls implemented', 'status': 'pass', 'weight': 0.15},
    {'name': 'Security monitoring enabled', 'status': 'pass', 'weight': 0.1}
]

availability_checks = [
    {'name': 'Multi-AZ deployment', 'status': 'pass', 'weight': 0.4},
    {'name': 'Backup and recovery procedures', 'status': 'pass', 'weight': 0.3},
    {'name': 'Monitoring and alerting', 'status': 'pass', 'weight': 0.3}
]

# Calculate scores for each category
for check in security_checks:
    soc2_compliance['security']['checks'].append(check)
    if check['status'] == 'pass':
        soc2_compliance['security']['score'] += check['weight']

for check in availability_checks:
    soc2_compliance['availability']['checks'].append(check)
    if check['status'] == 'pass':
        soc2_compliance['availability']['score'] += check['weight']

# Calculate overall score
overall_score = 0
for category, data in soc2_compliance.items():
    if category != 'overall_score' and 'score' in data:
        overall_score += data['score'] * data['weight']

soc2_compliance['overall_score'] = overall_score

with open('soc2_compliance.json', 'w') as f:
    json.dump(soc2_compliance, f, indent=2)

print(f'SOC 2 compliance validation complete:')
print(f'Overall score: {overall_score:.2f}')
"
      
      # Run HIPAA compliance checks if applicable
      - |
        if [ '$HIPAA_REQUIRED' = 'true' ]; then
          echo "Running HIPAA compliance validation..."
          python3 -c "
import boto3
import json

hipaa_compliance = {
    'administrative_safeguards': {'score': 0, 'max_score': 10},
    'physical_safeguards': {'score': 0, 'max_score': 8},
    'technical_safeguards': {'score': 0, 'max_score': 12},
    'overall_compliance': 0
}

# Mock HIPAA compliance checks
technical_checks = [
    'Encryption in transit and at rest',
    'Access controls and authentication',
    'Audit logging enabled',
    'Data backup and recovery',
    'Network security controls'
]

for check in technical_checks:
    hipaa_compliance['technical_safeguards']['score'] += 2  # Mock passing score

total_possible = (hipaa_compliance['administrative_safeguards']['max_score'] + 
                 hipaa_compliance['physical_safeguards']['max_score'] + 
                 hipaa_compliance['technical_safeguards']['max_score'])

total_actual = (hipaa_compliance['administrative_safeguards']['score'] + 
               hipaa_compliance['physical_safeguards']['score'] + 
               hipaa_compliance['technical_safeguards']['score'])

hipaa_compliance['overall_compliance'] = (total_actual / total_possible) * 100

with open('hipaa_compliance.json', 'w') as f:
    json.dump(hipaa_compliance, f, indent=2)

print(f'HIPAA compliance validation complete:')
print(f'Overall compliance: {hipaa_compliance[\"overall_compliance\"]:.1f}%')
"
        fi
      
      # Aggregate compliance results
      - python3 -c "
import json
import sys

compliance_summary = {
    'aws_config_compliance': 0,
    'cis_benchmark_score': 0,
    'soc2_compliance': 0,
    'hipaa_compliance': 0,
    'overall_compliance_score': 0,
    'critical_violations': [],
    'compliance_frameworks': []
}

# Process AWS Config results
try:
    with open('aws_config_compliance.json', 'r') as f:
        config_data = json.load(f)
        if 'summary' in config_data:
            compliance_summary['aws_config_compliance'] = config_data['summary'].get('compliance_score', 0)
            compliance_summary['compliance_frameworks'].append('AWS Config')
except:
    pass

# Process Prowler CIS results
try:
    with open('prowler_results.json', 'r') as f:
        prowler_data = json.load(f)
        # Calculate CIS benchmark score from Prowler results
        total_checks = len(prowler_data)
        passed_checks = sum(1 for check in prowler_data if check.get('Status') == 'PASS')
        compliance_summary['cis_benchmark_score'] = (passed_checks / total_checks * 100) if total_checks > 0 else 0
        compliance_summary['compliance_frameworks'].append('CIS Benchmark')
        
        # Identify critical violations
        for check in prowler_data:
            if check.get('Status') == 'FAIL' and check.get('Severity') == 'high':
                compliance_summary['critical_violations'].append({
                    'framework': 'CIS',
                    'check': check.get('CheckID'),
                    'description': check.get('CheckTitle'),
                    'severity': check.get('Severity')
                })
except:
    pass

# Process SOC 2 results
try:
    with open('soc2_compliance.json', 'r') as f:
        soc2_data = json.load(f)
        compliance_summary['soc2_compliance'] = soc2_data.get('overall_score', 0) * 100
        compliance_summary['compliance_frameworks'].append('SOC 2 Type II')
except:
    pass

# Process HIPAA results if available
try:
    with open('hipaa_compliance.json', 'r') as f:
        hipaa_data = json.load(f)
        compliance_summary['hipaa_compliance'] = hipaa_data.get('overall_compliance', 0)
        compliance_summary['compliance_frameworks'].append('HIPAA')
except:
    pass

# Calculate overall compliance score
framework_scores = [
    compliance_summary['aws_config_compliance'],
    compliance_summary['cis_benchmark_score'],
    compliance_summary['soc2_compliance']
]

if compliance_summary['hipaa_compliance'] > 0:
    framework_scores.append(compliance_summary['hipaa_compliance'])

compliance_summary['overall_compliance_score'] = sum(framework_scores) / len(framework_scores) if framework_scores else 0

# Save summary
with open('compliance_summary.json', 'w') as f:
    json.dump(compliance_summary, f, indent=2)

print(f'Compliance validation complete:')
print(f'AWS Config compliance: {compliance_summary[\"aws_config_compliance\"]:.1f}%')
print(f'CIS benchmark score: {compliance_summary[\"cis_benchmark_score\"]:.1f}%')
print(f'SOC 2 compliance: {compliance_summary[\"soc2_compliance\"]:.1f}%')
if compliance_summary['hipaa_compliance'] > 0:
    print(f'HIPAA compliance: {compliance_summary[\"hipaa_compliance\"]:.1f}%')
print(f'Overall compliance score: {compliance_summary[\"overall_compliance_score\"]:.1f}%')
print(f'Critical violations: {len(compliance_summary[\"critical_violations\"])}')

# Compliance gate: fail if overall score is below threshold or critical violations exist
if compliance_summary['overall_compliance_score'] < 80 or len(compliance_summary['critical_violations']) > 0:
    print('COMPLIANCE GATE FAILURE: Compliance score below threshold or critical violations found')
    sys.exit(1)
"
  
  post_build:
    commands:
      - echo Compliance validation completed
      - |
        if [ $CODEBUILD_BUILD_SUCCEEDING -eq 0 ]; then
          echo "Compliance validation failed - stopping pipeline"
          aws sns publish --topic-arn arn:aws:sns:us-east-1:$AWS_ACCOUNT_ID:devsecops-alerts \
            --message "Compliance validation failed. Critical compliance violations detected."
        fi

artifacts:
  files:
    - compliance_summary.json
    - aws_config_compliance.json
    - prowler_results.json
    - soc2_compliance.json
    - hipaa_compliance.json

environment:
  variables:
    HIPAA_REQUIRED: 'false'
"""
    
    def setup_security_notifications(self, pipeline_name: str):
        """Setup SNS notifications for security events"""
        
        try:
            # Create SNS topic for DevSecOps alerts
            sns_response = self.sns.create_topic(
                Name='devsecops-alerts',
                Attributes={
                    'DisplayName': 'DevSecOps Security Alerts',
                    'Description': 'Security alerts from DevSecOps pipeline'
                }
            )
            
            topic_arn = sns_response['TopicArn']
            
            # Subscribe email endpoint (would be configured with actual email)
            self.sns.subscribe(
                TopicArn=topic_arn,
                Protocol='email',
                Endpoint='security-team@company.com'
            )
            
            return topic_arn
            
        except Exception as e:
            print(f"Failed to setup security notifications: {str(e)}")
            return None
    
    def get_account_id(self) -> str:
        """Get current AWS account ID"""
        try:
            sts = boto3.client('sts')
            return sts.get_caller_identity()['Account']
        except:
            return '123456789012'  # Fallback for demo purposes

# Example usage
pipeline_manager = SecureDevSecOpsPipeline()

Container Security Integration

EKS Security with Policy Enforcement

import boto3
import yaml
from typing import Dict, List

class EKSSecurityManager:
    def __init__(self):
        self.eks = boto3.client('eks')
        self.ec2 = boto3.client('ec2')
        self.iam = boto3.client('iam')
        
    def create_secure_eks_cluster(self, cluster_name: str, vpc_config: Dict) -> Dict:
        """Create EKS cluster with comprehensive security configurations"""
        
        try:
            # Create EKS cluster with security best practices
            cluster_response = self.eks.create_cluster(
                name=cluster_name,
                version='1.25',
                roleArn=f'arn:aws:iam::{self.get_account_id()}:role/EKSServiceRole',
                resourcesVpcConfig={
                    'subnetIds': vpc_config['private_subnets'],
                    'securityGroupIds': [vpc_config['cluster_security_group']],
                    'endpointConfigType': 'PRIVATE',
                    'endpointPrivateAccess': True,
                    'endpointPublicAccess': False,
                    'publicAccessCidrs': []
                },
                kubernetesNetworkConfig={
                    'serviceIpv4Cidr': '172.20.0.0/16'
                },
                logging={
                    'enable': True,
                    'types': ['api', 'audit', 'authenticator', 'controllerManager', 'scheduler']
                },
                encryptionConfig=[
                    {
                        'resources': ['secrets'],
                        'provider': {
                            'keyArn': vpc_config['kms_key_arn']
                        }
                    }
                ],
                tags={
                    'Environment': 'production',
                    'SecurityLevel': 'high',
                    'Compliance': 'required'
                }
            )
            
            # Wait for cluster to be active
            waiter = self.eks.get_waiter('cluster_active')
            waiter.wait(name=cluster_name)
            
            # Apply security policies
            self.apply_security_policies(cluster_name)
            
            # Configure RBAC
            self.configure_rbac(cluster_name)
            
            # Setup monitoring
            self.setup_cluster_monitoring(cluster_name)
            
            return {
                'cluster_name': cluster_name,
                'cluster_arn': cluster_response['cluster']['arn'],
                'status': 'created',
                'security_features': [
                    'Private API endpoint',
                    'Envelope encryption enabled',
                    'Comprehensive logging',
                    'Network isolation',
                    'RBAC configured',
                    'Security policies applied'
                ]
            }
            
        except Exception as e:
            return {'status': 'failed', 'error': str(e)}
    
    def apply_security_policies(self, cluster_name: str):
        """Apply Kubernetes security policies"""
        
        # Pod Security Standards
        pod_security_policy = {
            'apiVersion': 'v1',
            'kind': 'Namespace',
            'metadata': {
                'name': 'secure-namespace',
                'labels': {
                    'pod-security.kubernetes.io/enforce': 'restricted',
                    'pod-security.kubernetes.io/audit': 'restricted',
                    'pod-security.kubernetes.io/warn': 'restricted'
                }
            }
        }
        
        # Network Policy
        network_policy = {
            'apiVersion': 'networking.k8s.io/v1',
            'kind': 'NetworkPolicy',
            'metadata': {
                'name': 'default-deny-all',
                'namespace': 'secure-namespace'
            },
            'spec': {
                'podSelector': {},
                'policyTypes': ['Ingress', 'Egress']
            }
        }
        
        # Security Context Constraints
        security_context_policy = {
            'apiVersion': 'v1',
            'kind': 'SecurityContextConstraints',
            'metadata': {
                'name': 'restricted-scc'
            },
            'allowHostDirVolumePlugin': False,
            'allowHostIPC': False,
            'allowHostNetwork': False,
            'allowHostPID': False,
            'allowPrivilegedContainer': False,
            'allowedCapabilities': [],
            'defaultAddCapabilities': [],
            'requiredDropCapabilities': ['ALL'],
            'runAsUser': {
                'type': 'MustRunAsNonRoot'
            },
            'seLinuxContext': {
                'type': 'MustRunAs'
            },
            'fsGroup': {
                'type': 'MustRunAs'
            }
        }
        
        # Apply policies (in real implementation, would use kubectl or Kubernetes API)
        print(f"Applied security policies to cluster {cluster_name}")
    
    def configure_rbac(self, cluster_name: str):
        """Configure Role-Based Access Control"""
        
        # Developer role with limited permissions
        developer_role = {
            'apiVersion': 'rbac.authorization.k8s.io/v1',
            'kind': 'Role',
            'metadata': {
                'namespace': 'development',
                'name': 'developer-role'
            },
            'rules': [
                {
                    'apiGroups': [''],
                    'resources': ['pods', 'services', 'configmaps'],
                    'verbs': ['get', 'list', 'create', 'update', 'patch', 'watch']
                },
                {
                    'apiGroups': ['apps'],
                    'resources': ['deployments', 'replicasets'],
                    'verbs': ['get', 'list', 'create', 'update', 'patch', 'watch']
                }
            ]
        }
        
        # Security admin cluster role
        security_admin_role = {
            'apiVersion': 'rbac.authorization.k8s.io/v1',
            'kind': 'ClusterRole',
            'metadata': {
                'name': 'security-admin-role'
            },
            'rules': [
                {
                    'apiGroups': [''],
                    'resources': ['*'],
                    'verbs': ['*']
                },
                {
                    'apiGroups': ['rbac.authorization.k8s.io'],
                    'resources': ['*'],
                    'verbs': ['*']
                },
                {
                    'apiGroups': ['networking.k8s.io'],
                    'resources': ['networkpolicies'],
                    'verbs': ['*']
                }
            ]
        }
        
        print(f"Configured RBAC for cluster {cluster_name}")
    
    def setup_cluster_monitoring(self, cluster_name: str):
        """Setup comprehensive cluster monitoring"""
        
        # CloudWatch Container Insights
        monitoring_config = {
            'cluster_name': cluster_name,
            'cloudwatch_insights': True,
            'prometheus_metrics': True,
            'fluent_bit_logging': True,
            'security_monitoring': True
        }
        
        # Falco for runtime security monitoring
        falco_config = {
            'enabled': True,
            'rules': [
                'shell_spawned_in_container',
                'write_below_binary_dir',
                'read_sensitive_file_untrusted',
                'write_below_etc',
                'privilege_escalation'
            ]
        }
        
        print(f"Setup monitoring for cluster {cluster_name}")
        return monitoring_config

    def get_account_id(self) -> str:
        """Get current AWS account ID"""
        try:
            sts = boto3.client('sts')
            return sts.get_caller_identity()['Account']
        except:
            return '123456789012'

# Example usage
eks_security = EKSSecurityManager()

CI/CD Security Automation CloudFormation

AWSTemplateFormatVersion: '2010-09-09'
Description: 'Complete DevSecOps Pipeline with Integrated Security'

Parameters:
  RepositoryName:
    Type: String
    Description: Name of the CodeCommit repository
    Default: secure-application
  
  ArtifactBucketName:
    Type: String
    Description: S3 bucket for pipeline artifacts
    Default: devsecops-pipeline-artifacts
  
  NotificationEmail:
    Type: String
    Description: Email for security notifications
    Default: security@company.com

Resources:
  # S3 Bucket for Pipeline Artifacts
  ArtifactStore:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Sub '${ArtifactBucketName}-${AWS::AccountId}'
      BucketEncryption:
        ServerSideEncryptionConfiguration:
          - ServerSideEncryptionByDefault:
              SSEAlgorithm: aws:kms
              KMSMasterKeyID: !Ref PipelineKMSKey
      VersioningConfiguration:
        Status: Enabled
      PublicAccessBlockConfiguration:
        BlockPublicAcls: true
        BlockPublicPolicy: true
        IgnorePublicAcls: true
        RestrictPublicBuckets: true

  # KMS Key for Pipeline Encryption
  PipelineKMSKey:
    Type: AWS::KMS::Key
    Properties:
      Description: KMS Key for DevSecOps Pipeline encryption
      KeyPolicy:
        Statement:
          - Sid: Enable IAM User Permissions
            Effect: Allow
            Principal:
              AWS: !Sub 'arn:aws:iam::${AWS::AccountId}:root'
            Action: 'kms:*'
            Resource: '*'
          - Sid: Allow CodePipeline Service
            Effect: Allow
            Principal:
              Service: codepipeline.amazonaws.com
            Action:
              - kms:Decrypt
              - kms:GenerateDataKey
            Resource: '*'

  # CodeCommit Repository
  SecureRepository:
    Type: AWS::CodeCommit::Repository
    Properties:
      RepositoryName: !Ref RepositoryName
      RepositoryDescription: Secure application repository with DevSecOps integration
      Code:
        S3:
          Bucket: !Ref ArtifactStore
          Key: initial-code.zip
      Triggers:
        - Name: PipelineTrigger
          DestinationArn: !Sub 'arn:aws:sns:${AWS::Region}:${AWS::AccountId}:devsecops-alerts'
          Events:
            - createReference
            - updateReference

  # ECR Repository for Container Images
  ContainerRepository:
    Type: AWS::ECR::Repository
    Properties:
      RepositoryName: !Sub '${RepositoryName}-secure'
      ImageTagMutability: IMMUTABLE
      ImageScanningConfiguration:
        ScanOnPush: true
      EncryptionConfiguration:
        EncryptionType: KMS
        KmsKey: !Ref PipelineKMSKey
      LifecyclePolicy:
        LifecyclePolicyText: |
          {
            "rules": [
              {
                "rulePriority": 1,
                "description": "Keep last 10 production images",
                "selection": {
                  "tagStatus": "tagged",
                  "tagPrefixList": ["prod"],
                  "countType": "imageCountMoreThan",
                  "countNumber": 10
                },
                "action": {
                  "type": "expire"
                }
              },
              {
                "rulePriority": 2,
                "description": "Delete untagged images older than 1 day",
                "selection": {
                  "tagStatus": "untagged",
                  "countType": "sinceImagePushed",
                  "countUnit": "days",
                  "countNumber": 1
                },
                "action": {
                  "type": "expire"
                }
              }
            ]
          }

  # SNS Topic for Security Alerts
  SecurityAlertsTopic:
    Type: AWS::SNS::Topic
    Properties:
      TopicName: devsecops-alerts
      DisplayName: DevSecOps Security Alerts
      KmsMasterKeyId: !Ref PipelineKMSKey
      Subscription:
        - Endpoint: !Ref NotificationEmail
          Protocol: email

  # IAM Role for CodePipeline
  CodePipelineRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: DevSecOps-CodePipeline-Role
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: codepipeline.amazonaws.com
            Action: sts:AssumeRole
      Policies:
        - PolicyName: PipelineExecutionPolicy
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - s3:GetBucketVersioning
                  - s3:GetObject
                  - s3:GetObjectVersion
                  - s3:PutObject
                Resource:
                  - !Sub '${ArtifactStore}/*'
                  - !GetAtt ArtifactStore.Arn
              - Effect: Allow
                Action:
                  - codecommit:GetBranch
                  - codecommit:GetCommit
                  - codecommit:ListBranches
                  - codecommit:ListRepositories
                Resource: !GetAtt SecureRepository.Arn
              - Effect: Allow
                Action:
                  - codebuild:BatchGetBuilds
                  - codebuild:StartBuild
                Resource: '*'
              - Effect: Allow
                Action:
                  - cloudformation:CreateStack
                  - cloudformation:DeleteStack
                  - cloudformation:DescribeStacks
                  - cloudformation:UpdateStack
                  - cloudformation:CreateChangeSet
                  - cloudformation:DeleteChangeSet
                  - cloudformation:DescribeChangeSet
                  - cloudformation:ExecuteChangeSet
                Resource: '*'
              - Effect: Allow
                Action:
                  - iam:PassRole
                Resource: '*'
              - Effect: Allow
                Action:
                  - lambda:InvokeFunction
                Resource: '*'
              - Effect: Allow
                Action:
                  - kms:Decrypt
                  - kms:GenerateDataKey
                Resource: !GetAtt PipelineKMSKey.Arn

  # IAM Role for CodeBuild
  CodeBuildRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: DevSecOps-CodeBuild-Role
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: codebuild.amazonaws.com
            Action: sts:AssumeRole
      Policies:
        - PolicyName: BuildExecutionPolicy
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - logs:CreateLogGroup
                  - logs:CreateLogStream
                  - logs:PutLogEvents
                Resource: !Sub 'arn:aws:logs:${AWS::Region}:${AWS::AccountId}:*'
              - Effect: Allow
                Action:
                  - s3:GetObject
                  - s3:GetObjectVersion
                  - s3:PutObject
                Resource:
                  - !Sub '${ArtifactStore}/*'
              - Effect: Allow
                Action:
                  - ecr:BatchCheckLayerAvailability
                  - ecr:GetDownloadUrlForLayer
                  - ecr:BatchGetImage
                  - ecr:GetAuthorizationToken
                  - ecr:PutImage
                  - ecr:InitiateLayerUpload
                  - ecr:UploadLayerPart
                  - ecr:CompleteLayerUpload
                Resource: '*'
              - Effect: Allow
                Action:
                  - securityhub:BatchImportFindings
                  - securityhub:CreateInsight
                  - securityhub:GetFindings
                Resource: '*'
              - Effect: Allow
                Action:
                  - inspector2:GetFindings
                  - inspector2:ListFindings
                Resource: '*'
              - Effect: Allow
                Action:
                  - sns:Publish
                Resource: !Ref SecurityAlertsTopic
              - Effect: Allow
                Action:
                  - config:GetComplianceDetailsByConfigRule
                  - config:GetComplianceSummaryByConfigRule
                  - config:DescribeConfigRules
                Resource: '*'
              - Effect: Allow
                Action:
                  - kms:Decrypt
                  - kms:GenerateDataKey
                Resource: !GetAtt PipelineKMSKey.Arn

  # DevSecOps Pipeline
  DevSecOpsPipeline:
    Type: AWS::CodePipeline::Pipeline
    Properties:
      Name: !Sub '${RepositoryName}-devsecops-pipeline'
      RoleArn: !GetAtt CodePipelineRole.Arn
      ArtifactStore:
        Type: S3
        Location: !Ref ArtifactStore
        EncryptionKey:
          Id: !GetAtt PipelineKMSKey.Arn
          Type: KMS
      Stages:
        - Name: Source
          Actions:
            - Name: SourceAction
              ActionTypeId:
                Category: Source
                Owner: AWS
                Provider: CodeCommit
                Version: '1'
              Configuration:
                RepositoryName: !GetAtt SecureRepository.Name
                BranchName: main
                PollForSourceChanges: false
              OutputArtifacts:
                - Name: SourceOutput

        - Name: SecurityAnalysis
          Actions:
            - Name: SecretScanning
              ActionTypeId:
                Category: Build
                Owner: AWS
                Provider: CodeBuild
                Version: '1'
              Configuration:
                ProjectName: !Ref SecretScanningProject
              InputArtifacts:
                - Name: SourceOutput
              OutputArtifacts:
                - Name: SecretScanOutput
              RunOrder: 1

            - Name: DependencyScanning
              ActionTypeId:
                Category: Build
                Owner: AWS
                Provider: CodeBuild
                Version: '1'
              Configuration:
                ProjectName: !Ref DependencyScanningProject
              InputArtifacts:
                - Name: SourceOutput
              OutputArtifacts:
                - Name: DependencyScanOutput
              RunOrder: 1

            - Name: SASTAnalysis
              ActionTypeId:
                Category: Build
                Owner: AWS
                Provider: CodeBuild
                Version: '1'
              Configuration:
                ProjectName: !Ref SASTAnalysisProject
              InputArtifacts:
                - Name: SourceOutput
              OutputArtifacts:
                - Name: SASTOutput
              RunOrder: 2

        - Name: Build
          Actions:
            - Name: BuildApplication
              ActionTypeId:
                Category: Build
                Owner: AWS
                Provider: CodeBuild
                Version: '1'
              Configuration:
                ProjectName: !Ref SecureBuildProject
              InputArtifacts:
                - Name: SourceOutput
              OutputArtifacts:
                - Name: BuildOutput
              RunOrder: 1

            - Name: ContainerScanning
              ActionTypeId:
                Category: Build
                Owner: AWS
                Provider: CodeBuild
                Version: '1'
              Configuration:
                ProjectName: !Ref ContainerScanningProject
              InputArtifacts:
                - Name: BuildOutput
              OutputArtifacts:
                - Name: ContainerScanOutput
              RunOrder: 2

        - Name: SecurityTesting
          Actions:
            - Name: DASTTesting
              ActionTypeId:
                Category: Build
                Owner: AWS
                Provider: CodeBuild
                Version: '1'
              Configuration:
                ProjectName: !Ref DASTTestingProject
              InputArtifacts:
                - Name: BuildOutput
              OutputArtifacts:
                - Name: DASTOutput
              RunOrder: 1

            - Name: InfrastructureScanning
              ActionTypeId:
                Category: Build
                Owner: AWS
                Provider: CodeBuild
                Version: '1'
              Configuration:
                ProjectName: !Ref InfrastructureScanningProject
              InputArtifacts:
                - Name: BuildOutput
              OutputArtifacts:
                - Name: InfraScanOutput
              RunOrder: 1

            - Name: ComplianceValidation
              ActionTypeId:
                Category: Build
                Owner: AWS
                Provider: CodeBuild
                Version: '1'
              Configuration:
                ProjectName: !Ref ComplianceValidationProject
              InputArtifacts:
                - Name: BuildOutput
              OutputArtifacts:
                - Name: ComplianceOutput
              RunOrder: 2

        - Name: Deploy
          Actions:
            - Name: DeployToStaging
              ActionTypeId:
                Category: Deploy
                Owner: AWS
                Provider: CloudFormation
                Version: '1'
              Configuration:
                ActionMode: CREATE_UPDATE
                StackName: !Sub '${RepositoryName}-staging-stack'
                TemplatePath: BuildOutput::deployment-template.yaml
                Capabilities: CAPABILITY_IAM,CAPABILITY_NAMED_IAM
                RoleArn: !GetAtt CloudFormationRole.Arn
              InputArtifacts:
                - Name: BuildOutput
              RunOrder: 1

            - Name: SecurityValidation
              ActionTypeId:
                Category: Invoke
                Owner: AWS
                Provider: Lambda
                Version: '1'
              Configuration:
                FunctionName: !Ref SecurityValidationFunction
              RunOrder: 2

  # CodeBuild Projects (references to projects defined earlier)
  SecretScanningProject:
    Type: AWS::CodeBuild::Project
    Properties:
      Name: !Sub '${RepositoryName}-secret-scanning'
      ServiceRole: !GetAtt CodeBuildRole.Arn
      Artifacts:
        Type: CODEPIPELINE
      Environment:
        Type: LINUX_CONTAINER
        ComputeType: BUILD_GENERAL1_MEDIUM
        Image: aws/codebuild/amazonlinux2-x86_64-standard:3.0
      Source:
        Type: CODEPIPELINE
        BuildSpec: |
          version: 0.2
          phases:
            install:
              runtime-versions:
                python: 3.9
              commands:
                - pip install truffleHog3 detect-secrets
            build:
              commands:
                - echo "Running secret scanning..."
                - trufflehog3 --format json --output secret_scan_results.json .
          artifacts:
            files:
              - secret_scan_results.json

  # Additional CodeBuild projects would be defined similarly...
  DependencyScanningProject:
    Type: AWS::CodeBuild::Project
    Properties:
      Name: !Sub '${RepositoryName}-dependency-scanning'
      ServiceRole: !GetAtt CodeBuildRole.Arn
      Artifacts:
        Type: CODEPIPELINE
      Environment:
        Type: LINUX_CONTAINER
        ComputeType: BUILD_GENERAL1_MEDIUM
        Image: aws/codebuild/amazonlinux2-x86_64-standard:3.0
      Source:
        Type: CODEPIPELINE
        BuildSpec: |
          version: 0.2
          phases:
            install:
              commands:
                - pip install safety
            build:
              commands:
                - echo "Running dependency scanning..."
                - safety check --json --output dependency_scan_results.json
          artifacts:
            files:
              - dependency_scan_results.json

  # CloudFormation role for deployments
  CloudFormationRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: DevSecOps-CloudFormation-Role
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: cloudformation.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/PowerUserAccess

  # Lambda function for security validation
  SecurityValidationFunction:
    Type: AWS::Lambda::Function
    Properties:
      FunctionName: !Sub '${RepositoryName}-security-validation'
      Runtime: python3.9
      Handler: index.lambda_handler
      Role: !GetAtt SecurityValidationRole.Arn
      Code:
        ZipFile: |
          import json
          import boto3
          
          def lambda_handler(event, context):
              # Perform post-deployment security validation
              print("Running security validation...")
              
              # Example validation checks
              validation_results = {
                  'ssl_certificate_valid': True,
                  'security_headers_present': True,
                  'no_public_s3_buckets': True,
                  'encryption_enabled': True
              }
              
              # Fail if any validation fails
              if not all(validation_results.values()):
                  raise Exception("Security validation failed")
              
              return {
                  'statusCode': 200,
                  'body': json.dumps('Security validation passed')
              }

  SecurityValidationRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
      Policies:
        - PolicyName: SecurityValidationPolicy
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - s3:GetBucketPolicy
                  - s3:GetBucketAcl
                  - ec2:DescribeInstances
                  - cloudformation:DescribeStacks
                Resource: '*'

  # CloudWatch Event Rule for pipeline trigger
  PipelineTriggerRule:
    Type: AWS::Events::Rule
    Properties:
      Description: Trigger DevSecOps pipeline on code commit
      EventPattern:
        source:
          - aws.codecommit
        detail-type:
          - CodeCommit Repository State Change
        detail:
          event:
            - referenceCreated
            - referenceUpdated
          referenceName:
            - main
          repositoryName:
            - !GetAtt SecureRepository.Name
      State: ENABLED
      Targets:
        - Arn: !Sub 'arn:aws:codepipeline:${AWS::Region}:${AWS::AccountId}:pipeline/${DevSecOpsPipeline}'
          Id: DevSecOpsPipelineTarget
          RoleArn: !GetAtt PipelineTriggerRole.Arn

  PipelineTriggerRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: events.amazonaws.com
            Action: sts:AssumeRole
      Policies:
        - PolicyName: StartPipelinePolicy
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - codepipeline:StartPipelineExecution
                Resource: !Sub 'arn:aws:codepipeline:${AWS::Region}:${AWS::AccountId}:pipeline/${DevSecOpsPipeline}'

Outputs:
  PipelineName:
    Description: Name of the DevSecOps pipeline
    Value: !Ref DevSecOpsPipeline
    Export:
      Name: !Sub '${AWS::StackName}-PipelineName'
  
  RepositoryCloneUrl:
    Description: HTTPS clone URL for the repository
    Value: !GetAtt SecureRepository.CloneUrlHttp
    Export:
      Name: !Sub '${AWS::StackName}-RepositoryUrl'
  
  ContainerRepositoryUri:
    Description: ECR repository URI
    Value: !Sub '${AWS::AccountId}.dkr.ecr.${AWS::Region}.amazonaws.com/${ContainerRepository}'
    Export:
      Name: !Sub '${AWS::StackName}-ContainerRepositoryUri'
  
  SecurityAlertsTopicArn:
    Description: SNS topic ARN for security alerts
    Value: !Ref SecurityAlertsTopic
    Export:
      Name: !Sub '${AWS::StackName}-SecurityAlertsTopic'

Best Practices and Recommendations

Implementation Guidelines

• Incremental Integration: Start with basic security scanning and gradually add advanced features
• Security Gate Tuning: Calibrate security thresholds based on organizational risk tolerance
• Tool Chain Optimization: Select security tools that integrate well with existing development workflows
• Performance Monitoring: Track pipeline execution times and optimize for developer productivity
• Feedback Loop Implementation: Ensure security findings are actionable and trackable
• Training and Adoption: Provide comprehensive training on new security processes and tools

Security Considerations

Pipeline Security:

• Use IAM roles with least privilege principles for all pipeline components
• Encrypt all artifacts and logs using customer-managed KMS keys
• Implement secure credential management using AWS Secrets Manager
• Enable comprehensive audit logging for all pipeline activities
• Regularly rotate access keys and review permissions

Code Security:

• Implement branch protection rules requiring security scan approval
• Use signed commits and GPG verification for code integrity
• Implement automated dependency updates with security scanning
• Establish security coding standards and automated enforcement
• Implement secrets detection at commit time using git hooks

Infrastructure Security:

• Use Infrastructure as Code for all security configurations
• Implement drift detection and automatic remediation
• Use immutable infrastructure patterns where possible
• Implement network segmentation for pipeline components
• Regular security assessment of pipeline infrastructure

Performance Optimization

Pipeline Optimization:

# Optimize security scanning with intelligent caching
class SecurityScanOptimizer:
    def __init__(self):
        self.cache_duration = 3600  # 1 hour
        self.scan_cache = {}
        
    def should_skip_scan(self, scan_type: str, code_hash: str) -> bool:
        """Determine if scan can be skipped based on cache"""
        cache_key = f"{scan_type}:{code_hash}"
        
        if cache_key in self.scan_cache:
            cached_result = self.scan_cache[cache_key]
            age = datetime.now().timestamp() - cached_result['timestamp']
            
            # Skip scan if recent and no critical issues found
            if age < self.cache_duration and cached_result['critical_issues'] == 0:
                return True
        
        return False
    
    def cache_scan_result(self, scan_type: str, code_hash: str, results: dict):
        """Cache scan results for future reference"""
        cache_key = f"{scan_type}:{code_hash}"
        
        self.scan_cache[cache_key] = {
            'timestamp': datetime.now().timestamp(),
            'critical_issues': results.get('critical', 0),
            'results': results
        }

Resource Management:

• Use appropriate CodeBuild instance sizes based on scan complexity
• Implement parallel scanning where possible to reduce pipeline time
• Use container image caching for faster builds
• Optimize artifact transfer between pipeline stages
• Monitor and tune security tool configurations for performance

Implementation Roadmap

Phase 1: Foundation (Weeks 1-2)

• Deploy basic CI/CD pipeline with CodePipeline and CodeBuild
• Implement source code secret scanning with TruffleHog
• Add basic dependency vulnerability scanning with Safety/npm audit
• Configure SNS notifications for security findings
• Establish basic security gates for critical vulnerabilities
• Train development team on new security processes

Phase 2: Enhanced Security Scanning (Weeks 3-4)

• Implement SAST analysis with Semgrep and language-specific tools
• Add container image scanning with Trivy and ECR integration
• Configure Dockerfile security scanning with Hadolint
• Implement basic DAST testing with OWASP ZAP
• Add infrastructure scanning with Checkov and tfsec
• Establish security metrics and reporting dashboards

Phase 3: Advanced Testing and Compliance (Weeks 5-6)

• Implement comprehensive DAST testing with custom security tests
• Add compliance validation for SOC 2, HIPAA, and industry standards
• Configure AWS Config rules for continuous compliance monitoring
• Implement Prowler for CIS benchmark compliance checking
• Add runtime security monitoring with Falco for container workloads
• Establish compliance reporting and audit trails

Phase 4: Container and Kubernetes Security (Weeks 7-8)

• Deploy secure EKS cluster with comprehensive security configurations
• Implement Kubernetes security policies and RBAC
• Add network policies and pod security standards
• Configure runtime security monitoring for containerized applications
• Implement image admission controllers with policy enforcement
• Establish container security baseline and monitoring

Phase 5: Optimization and Advanced Features (Weeks 9-10)

• Implement intelligent caching and performance optimization
• Add machine learning-based anomaly detection for security events
• Configure automated incident response workflows
• Implement advanced compliance reporting and audit capabilities
• Add integration with external security tools and threat intelligence
• Conduct comprehensive security assessment and penetration testing

Related Articles

• AWS Lambda Security: Automated Threat Detection Systems
• AWS IAM Zero Trust: Identity and Network Deep Dive
• AWS Container Security: EKS and Fargate Implementation Guide
• AWS Security Hub: Centralized Security Management

Additional Resources

Official Documentation

• AWS CodePipeline User Guide - Complete CI/CD pipeline configuration and best practices
• AWS CodeBuild User Guide - Build automation and security integration
• Amazon EKS Security Best Practices - Kubernetes security implementation
• AWS Security Hub User Guide - Centralized security findings management

Tools and Frameworks

• OWASP DevSecOps Guideline - Industry standard DevSecOps practices
• Semgrep Static Analysis - Multi-language static analysis security testing
• Trivy Container Scanner - Comprehensive container vulnerability scanning
• Checkov Infrastructure Scanner - Infrastructure as Code security scanning

Industry Reports and Research

• 2025 DevSecOps Survey - Current DevSecOps adoption trends
• NIST DevSecOps Reference Architecture - Government DevSecOps guidance
• CIS Controls Implementation - Security control framework implementation
• SANS DevSecOps Survey - Industry DevSecOps practices and challenges

Community Resources

• DevSecOps Community - Community resources and best practices
• AWS DevSecOps Workshop - Hands-on DevSecOps implementation
• Cloud Security Alliance - Cloud security research and guidance
• OWASP Security Champions - Security champion program guidance

Conclusion

Implementing comprehensive security automation in AWS DevSecOps pipelines represents a fundamental shift toward proactive security that scales with development velocity. This approach transforms security from a bottleneck into an enabler, allowing teams to deliver secure software faster while maintaining strict security and compliance standards.

Key benefits of this DevSecOps implementation include:

• Early Detection: Security issues identified and resolved in minutes rather than months
• Automated Compliance: Continuous validation against industry standards and regulatory requirements
• Developer Empowerment: Self-service security capabilities that don’t impede development productivity
• Scalable Security: Automated processes that scale with organizational growth and complexity
• Risk Reduction: Comprehensive coverage across the entire software development lifecycle

The success of DevSecOps automation depends on careful tool selection, proper security gate calibration, and strong collaboration between development, security, and operations teams. Organizations must balance security requirements with development velocity while ensuring that security becomes an integral part of the development culture.

For personalized guidance on implementing DevSecOps automation in your AWS environment, connect with Jon Price on LinkedIn.