pip install checkov boto3 pyyaml jinja2 requests

#!/usr/bin/env python3
"""
Security Audit Orchestrator
Scans IaC, evaluates OPA policies, and generates compliance reports.
"""
import json
import os
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path
import yaml
import checkov
from checkov.main import Checkov
from checkov.runner_filter import RunnerFilter
import boto3

class SecurityAuditOrchestrator:
    def __init__(self, config_path: str = "audit_config.yaml"):
        with open(config_path, "r") as f:
            self.config = yaml.safe_load(f)
        self.timestamp = datetime.now(timezone.utc).isoformat()
        self.evidence_store = Path(self.config["output"]["evidence_dir"])
        self.evidence_store.mkdir(parents=True, exist_ok=True)

    def run_iac_scan(self, terraform_dir: str) -> dict:
        """Execute Checkov scan on Terraform configurations."""
        runner_filter = RunnerFilter(
            framework=["terraform"],
            skip_check=self.config.get("skip_checks", []),
            quiet=True
        )
        checkov_runner = Checkov()
        report = checkov_runner.run(root_folder=terraform_dir, runner_filter=runner_filter)
        
        scan_result = {
            "timestamp": self.timestamp,
            "tool": "checkov",
            "target": terraform_dir,
            "passed": report.passed_checks,
            "failed": report.failed_checks,
            "skipped": report.skipped_checks,
            "summary": {
                "passed": len(report.passed_checks),
                "failed": len(report.failed_checks),
                "skipped": len(report.skipped_checks)
            }
        }
        return scan_result

    def evaluate_opa_policy(self, resource_state: dict, policy_path: str) -> dict:
        """Evaluate OPA policy against cloud resource state."""
        # In production, use `opa eval` or `conftest` via subprocess
        # This example demonstrates the integration pattern
        eval_cmd = [
            "opa", "eval",
            "--data", policy_path,
            "--input", "-",
            "data.security_audit.deny"
        ]
        process = subprocess.run(
            eval_cmd,
            input=json.dumps(resource_state),
            capture_output=True,
            text=True
        )
        return {
            "tool": "opa",
            "policy": policy_path,
            "decision": json.loads(process.stdout) if process.stdout.strip() else [],
            "exit_code": process.returncode
        }

    def generate_compliance_report(self, scans: list) -> dict:
        """Aggregate scans into a control-mapped compliance report."""
        report = {
            "audit_id": f"AUDIT-{self.timestamp.replace(':', '').replace('-', '')}",
            "generated_at": self.timestamp,
            "framework": self.

config["compliance_framework"], "controls": {}, "summary": {"total_controls": 0, "compliant": 0, "non_compliant": 0, "partial": 0} }

    for scan in scans:
        for check in scan.get("failed", []):
            control_id = check.check_id
            report["controls"][control_id] = {
                "status": "non_compliant",
                "evidence": check.resource,
                "severity": check.severity,
                "remediation": check.guideline or "Review control documentation"
            }
            report["summary"]["non_compliant"] += 1

        for check in scan.get("passed", []):
            control_id = check.check_id
            report["controls"][control_id] = {
                "status": "compliant",
                "evidence": check.resource,
                "severity": "info"
            }
            report["summary"]["compliant"] += 1

    report["summary"]["total_controls"] = len(report["controls"])
    return report

def persist_evidence(self, report: dict):
    """Store evidence with deterministic naming and optional signing."""
    evidence_file = self.evidence_store / f"audit_report_{self.timestamp.replace(':', '')}.json"
    with open(evidence_file, "w") as f:
        json.dump(report, f, indent=2)
    print(f"[✓] Evidence persisted: {evidence_file}")
    return str(evidence_file)

def execute(self, terraform_dir: str, opa_policy_dir: str = "policies/"):
    """Main execution flow."""
    print(f"[*] Starting audit at {self.timestamp}")
    
    # 1. IaC Scan
    print("[*] Running IaC security scan...")
    iac_scan = self.run_iac_scan(terraform_dir)
    
    # 2. OPA Evaluation (mock resource state for demonstration)
    print("[*] Evaluating OPA policies...")
    mock_state = {
        "aws_s3_bucket": {
            "id": "prod-data-logs",
            "server_side_encryption": False,
            "public_access_block": False
        }
    }
    opa_eval = self.evaluate_opa_policy(mock_state, os.path.join(opa_policy_dir, "s3_policy.rego"))
    
    # 3. Generate Report
    print("[*] Generating compliance report...")
    report = self.generate_compliance_report([iac_scan, opa_eval])
    
    # 4. Persist Evidence
    evidence_path = self.persist_evidence(report)
    
    print(f"[✓] Audit complete. Report: {evidence_path}")
    return report

if name == "main": orchestrator = SecurityAuditOrchestrator() orchestrator.execute(terraform_dir="infrastructure/", opa_policy_dir="policies/")

#### OPA Policy Example (`policies/s3_policy.rego`)
```rego
package security_audit

deny[msg] {
    input.aws_s3_bucket.server_side_encryption == false
    msg := "S3 bucket lacks server-side encryption. Control: AWS-CIS-1.4.1"
}

deny[msg] {
    input.aws_s3_bucket.public_access_block == false
    msg := "S3 bucket missing public access block. Control: AWS-CIS-1.4.2"
}

CI/CD Integration (GitHub Actions)

name: Security Audit Pipeline
on:
  pull_request:
    paths: ["infrastructure/**"]
  schedule:
    - cron: "0 6 * * 1" # Weekly baseline audit

jobs:
  audit:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - name: Install Dependencies
        run: pip install checkov boto3 pyyaml jinja2
      - name: Run Audit Orchestrator
        run: python audit_orchestrator.py
      - name: Upload Evidence
        uses: actions/upload-artifact@v4
        with:
          name: security-audit-evidence
          path: evidence/
          retention-days: 90

This architecture transforms auditing from a manual exercise into a deterministic, version-controlled, and continuously validated process. The orchestrator can be extended with cloud provider SDKs, SIEM integrations, or Jira automation to close the remediation loop.

🚧 Pitfall Guide (5-7)

1. Automating Without Governance

Risk: Uncontrolled automation generates noise, blocks legitimate changes, or creates policy drift. Mitigation: Implement a policy review board, version control all policies, and require peer approval for control changes. Use git blame and PR templates to track who modified what and why.

2. Ignoring False Positive/Negative Rates

Risk: High false positives cause alert fatigue; false negatives create compliance blind spots. Mitigation: Establish a baseline audit, track precision/recall metrics, and implement a feedback loop where analysts can flag misclassifications. Retrain or adjust policies quarterly based on empirical data.

3. Hardcoding Secrets in Automation Pipelines

Risk: CI/CD scripts or IaC scans may accidentally expose API keys, tokens, or credentials. Mitigation: Use secret managers (AWS Secrets Manager, HashiCorp Vault, GitHub Actions Secrets). Never echo secrets in logs. Implement pre-commit hooks with gitleaks or trufflehog to catch accidental commits.

4. Lack of Audit Trail for Policy Changes

Risk: Auditors will reject automated reports if they cannot verify policy integrity or version history. Mitigation: Store policies in Git with signed commits. Generate cryptographic hashes of policy files at runtime. Include policy version and hash in every audit report. Use immutable storage for evidence.

5. Skipping Human-in-the-Loop for Critical Controls

Risk: Over-automation of high-risk controls (e.g., data classification, access provisioning) can bypass necessary business judgment. Mitigation: Implement a tiered automation model. Tier 1 (technical controls) fully automated. Tier 2 (business-aligned controls) automated with manual approval gates. Tier 3 (strategic controls) human-driven with automated evidence collection.

6. Inadequate Runtime vs. Build-Time Coverage

Risk: Scanning IaC at commit time misses drift, ephemeral resources, and runtime misconfigurations. Mitigation: Combine pre-merge IaC scanning with continuous cloud configuration monitoring (AWS Config, Azure Policy, GCP Security Command Center). Reconcile drift daily and trigger automated remediation or alerts.

7. Compliance Framework Misalignment

Risk: Mapping controls to outdated or incorrect compliance frameworks leads to audit failures and wasted effort. Mitigation: Maintain a control mapping matrix that explicitly links each automated check to specific framework requirements (e.g., SOC 2 CC6.1, ISO 27001 A.9.2.1). Update mappings quarterly and validate with compliance stakeholders.

📦 Production Bundle

✅ Security Audit Automation Checklist

Pre-Automation

• Inventory existing controls and map to compliance frameworks
• Define policy-as-code repository structure and ownership model
• Select scanning tools aligned with tech stack (IaC, containers, cloud, SAST/DAST)
• Establish evidence retention policy (duration, storage, access controls)
• Create rollback and incident response procedures for automation failures

Execution

• Deploy orchestrator in staging environment
• Run baseline audit against current infrastructure
• Validate false positive/negative rates and adjust thresholds
• Integrate with CI/CD pipelines (PR checks, scheduled runs)
• Configure alerting and ticketing integrations (Slack, Jira, ServiceNow)
• Implement evidence signing and immutable storage

Post-Implementation

• Generate first automated compliance report
• Conduct dry-run with internal audit/compliance team
• Document SOPs for policy updates, tool upgrades, and exception handling
• Schedule quarterly policy review and metric analysis
• Measure MTTR, audit cycle time, and cost savings

📊 Decision Matrix: Tool Selection Criteria

Criteria	Weight	Checkov	Trivy	TFSec	OPA/Conftest	AWS Config
IaC Coverage	25%	★★★★★	★★★☆☆	★★★★★	★★★★☆	★★☆☆☆
Cloud Runtime	20%	★★☆☆☆	★★☆☆☆	★★☆☆☆	★★★☆☆	★★★★★
CI/CD Integration	15%	★★★★★	★★★★★	★★★★☆	★★★★★	★★★☆☆
Policy Flexibility	15%	★★★☆☆	★★☆☆☆	★★★☆☆	★★★★★	★★☆☆☆
Compliance Mapping	10%	★★★★☆	★★★☆☆	★★★☆☆	★★★☆☆	★★★★☆
Community/Support	10%	★★★★★	★★★★★	★★★★☆	★★★★☆	★★★★★
Cost	5%	Free	Free	Free	Free	Pay-as-you-go
Weighted Score	100%	4.35	3.55	4.15	4.45	3.80

Scoring: 1-5 stars. Adjust weights based on organizational priorities. Recommended stack: OPA (policy) + Checkov (IaC) + Cloud-native config (runtime).

⚙️ Config Template

audit_config.yaml

compliance_framework: "SOC2-2023"
output:
  evidence_dir: "./evidence"
  report_format: "json"
  retention_days: 90
skip_checks:
  - CKV_AWS_20  # Example: Skip if business-approved exception exists
  - CKV_AWS_21
alerting:
  slack_webhook: "${SLACK_WEBHOOK_URL}"
  severity_threshold: "HIGH"
remediation:
  auto_apply: false
  ticket_system: "jira"
  project_key: "SEC"
policy_version: "1.2.0"
last_updated: "2024-06-15T10:00:00Z"

github_actions_workflow.yaml (Snippet)

jobs:
  security-audit:
    runs-on: ubuntu-latest
    env:
      CHECKOV_LOG_LEVEL: "WARNING"
      OPA_STRICT: "true"
    steps:
      - name: Checkout
        uses: actions/checkout@v4
      - name: Run IaC Scan
        run: |
          pip install checkov
          checkov -d infrastructure/ --output json --output-file-path reports/iac.json
      - name: Evaluate Policies
        run: |
          conftest test policies/ -d reports/iac.json --output json > reports/opa.json
      - name: Generate Report
        run: python scripts/audit_orchestrator.py
      - name: Upload Artifacts
        uses: actions/upload-artifact@v4
        with:
          name: compliance-evidence
          path: evidence/

🚀 Quick Start: Deploy in <30 Minutes

1. Initialize Repository

   mkdir security-audit-automation && cd security-audit-automation
   git init
   mkdir -p infrastructure policies evidence scripts
   

2. Add Configuration

   • Save audit_config.yaml to root
   • Save s3_policy.rego to policies/
   • Save audit_orchestrator.py to scripts/

3. Install Dependencies

   pip install checkov boto3 pyyaml jinja2 conftest
   

4. Run Baseline Audit

   python scripts/audit_orchestrator.py
   # Verify evidence/ directory contains timestamped JSON report
   

5. Integrate with CI/CD

   • Add GitHub Actions workflow to .github/workflows/security-audit.yml
   • Configure repository secrets for webhooks and cloud credentials
   • Commit and push to trigger first automated run

6. Validate & Iterate

   • Review evidence/ report against compliance requirements
   • Adjust skip_checks and policy thresholds
   • Schedule weekly runs and set up Slack/Jira alerts
   • Document SOPs and hand off to security engineering team

7. Scale

   • Add container scanning (Trivy)
   • Integrate runtime drift detection
   • Implement automated remediation for low-risk controls
   • Establish quarterly policy review cadence

---

Security audit automation is not a tool installation project; it is a security engineering discipline. By treating policies as code, evidence as immutable artifacts, and validation as continuous, organizations transform compliance from a cost center into a velocity multiplier. The strategies outlined here provide a production-ready foundation. Implement iteratively, measure rigorously, and align automation with business risk to achieve sustainable, auditable security posture at scale.