|
| 1 | +import json |
| 2 | +import os |
| 3 | +import sys |
| 4 | +from typing import List |
| 5 | + |
| 6 | +from google import genai |
| 7 | +from google.genai import types |
| 8 | +from pydantic import BaseModel |
| 9 | + |
| 10 | + |
| 11 | +# 1. Define the exact JSON structure we want Gemini to return |
| 12 | +class SecurityFinding(BaseModel): |
| 13 | + resource_name: str |
| 14 | + severity: str # CRITICAL, WARNING, or INFO |
| 15 | + vulnerability: str |
| 16 | + remediation_steps: str |
| 17 | + |
| 18 | + |
| 19 | +class SecurityReport(BaseModel): |
| 20 | + status: str # MUST be "PASS" or "FAIL" |
| 21 | + summary: str |
| 22 | + findings: List[SecurityFinding] |
| 23 | + |
| 24 | + |
| 25 | +def audit_terraform_plan(): |
| 26 | + # Verify API key |
| 27 | + if not os.environ.get("GEMINI_API_KEY"): |
| 28 | + print("❌ Error: GEMINI_API_KEY environment variable is not set.") |
| 29 | + sys.exit(1) |
| 30 | + |
| 31 | + # Load the real plan.json generated by the user |
| 32 | + plan_path = "plan.json" |
| 33 | + if not os.path.exists(plan_path): |
| 34 | + print( |
| 35 | + f"❌ Error: {plan_path} not found. Run 'terraform show -json tfplan.binary > plan.json' first." |
| 36 | + ) |
| 37 | + sys.exit(1) |
| 38 | + |
| 39 | + print(f"📖 Reading {plan_path}...") |
| 40 | + with open(plan_path, "r") as file: |
| 41 | + plan_data = json.load(file) |
| 42 | + |
| 43 | + # Clean the plan slightly to save token space (focusing on resource changes) |
| 44 | + resource_changes = plan_data.get("resource_changes", []) |
| 45 | + |
| 46 | + print(f"Securing {len(resource_changes)} planned resource modifications...") |
| 47 | + |
| 48 | + # 2. Design the prompt |
| 49 | + prompt = f""" |
| 50 | + You are an automated DevSecOps Cloud Security Auditor. Your job is to analyze the following |
| 51 | + Terraform execution plan JSON data and evaluate it against enterprise security best practices. |
| 52 | +
|
| 53 | + Look specifically for: |
| 54 | + - Publicly accessible storage (S3 buckets, Azure Blobs, etc.) |
| 55 | + - Overly permissive networking rules (e.g., Ingress from 0.0.0.0/0 on sensitive ports like 22, 3389, 80, 443 unless justified) |
| 56 | + - Unencrypted resources or databases |
| 57 | + - Cleartext credentials/secrets |
| 58 | +
|
| 59 | + If you find ANY CRITICAL security violations, you MUST set the status to "FAIL". |
| 60 | + If the infrastructure is safe or contains only minor issues, set the status to "PASS". |
| 61 | +
|
| 62 | + RAW TERRAFORM PLAN RESOURCE CHANGES: |
| 63 | + {json.dumps(resource_changes, indent=2)} |
| 64 | + """ |
| 65 | + |
| 66 | + print("🤖 Analyzing plan with Gemini Security Architect...") |
| 67 | + client = genai.Client() |
| 68 | + |
| 69 | + # 3. Request structured content matching our Pydantic model |
| 70 | + response = client.models.generate_content( |
| 71 | + model="gemini-2.5-flash", |
| 72 | + contents=prompt, |
| 73 | + config=types.GenerateContentConfig( |
| 74 | + response_mime_type="application/json", |
| 75 | + response_schema=SecurityReport, |
| 76 | + temperature=0.1, # Low temperature ensures strict, analytical deterministic results |
| 77 | + ), |
| 78 | + ) |
| 79 | + |
| 80 | + # 4. Parse the structured output |
| 81 | + report = SecurityReport.model_validate_json(response.text) |
| 82 | + |
| 83 | + # 5. Print a beautiful console report |
| 84 | + print("\n" + "=" * 50) |
| 85 | + print(f"🛡️ DEVSECOPS SECURITY VERDICT: {report.status}") |
| 86 | + print("=" * 50) |
| 87 | + print(f"Summary: {report.summary}\n") |
| 88 | + |
| 89 | + if report.findings: |
| 90 | + print(f"⚠️ Found {len(report.findings)} Security Vulnerabilities:") |
| 91 | + for idx, finding in enumerate(report.findings, 1): |
| 92 | + print(f"\n [{idx}] Resource: {finding.resource_name}") |
| 93 | + print(f" Severity: {finding.severity}") |
| 94 | + print(f" Vulnerability: {finding.vulnerability}") |
| 95 | + print(f" Fix: {finding.remediation_steps}") |
| 96 | + else: |
| 97 | + print("✅ No critical security violations detected.") |
| 98 | + |
| 99 | + print("=" * 50) |
| 100 | + |
| 101 | + # 6. ENFORCE THE GATEKEEPER RULE (Crucial for CI/CD pipelines!) |
| 102 | + if report.status == "FAIL": |
| 103 | + print("❌ Deployment Blocked: Critical security flaws must be resolved.") |
| 104 | + sys.exit(1) # Forces automated tools like GitHub Actions to halt immediately |
| 105 | + else: |
| 106 | + print("🚀 Deployment Approved: Infrastructure compliant.") |
| 107 | + sys.exit(0) |
| 108 | + |
| 109 | + |
| 110 | +if __name__ == "__main__": |
| 111 | + audit_terraform_plan() |
0 commit comments