Build an AI Terraform Plan Reviewer with Claude API

pip install anthropic python-dotenv

export ANTHROPIC_API_KEY="sk-ant-..."

# Generate machine-readable plan
terraform plan -out=tfplan.binary
terraform show -json tfplan.binary > tfplan.json

#!/usr/bin/env python3
"""
AI Terraform Plan Reviewer using Claude API
"""
 
import json
import sys
import os
import argparse
from anthropic import Anthropic
 
client = Anthropic()
 
SYSTEM_PROMPT = """You are a senior DevOps engineer reviewing Terraform plans for production safety.
 
Analyze the terraform plan JSON and provide:
1. RISK LEVEL: HIGH / MEDIUM / LOW
2. DESTROYED RESOURCES: List any resources being destroyed (especially databases, load balancers, IAM roles)
3. SECURITY CONCERNS: IAM permission changes, security group rule changes, public access changes
4. UNEXPECTED CHANGES: Resources changing that seem unrelated to the stated purpose
5. RECOMMENDATION: APPROVE / BLOCK / REVIEW
 
Format your response as JSON:
{
  "risk_level": "HIGH|MEDIUM|LOW",
  "recommendation": "APPROVE|BLOCK|REVIEW",
  "summary": "one line summary",
  "destroyed_resources": [],
  "security_concerns": [],
  "unexpected_changes": [],
  "details": "detailed explanation"
}
 
Be conservative — if unsure, escalate the risk level."""
 
 
def load_plan(plan_file: str) -> dict:
    with open(plan_file) as f:
        return json.load(f)
 
 
def extract_key_changes(plan: dict) -> str:
    """Extract the most relevant parts of the plan to keep tokens manageable."""
    changes = plan.get("resource_changes", [])
    
    summary = {
        "terraform_version": plan.get("terraform_version"),
        "format_version": plan.get("format_version"),
        "resource_changes": []
    }
    
    for change in changes:
        actions = change.get("change", {}).get("actions", [])
        if "no-op" in actions:
            continue  # skip unchanged resources
            
        summary["resource_changes"].append({
            "address": change.get("address"),
            "type": change.get("type"),
            "name": change.get("name"),
            "actions": actions,
            "before": change.get("change", {}).get("before"),
            "after": change.get("change", {}).get("after"),
            "after_sensitive": change.get("change", {}).get("after_sensitive"),
        })
    
    return json.dumps(summary, indent=2)
 
 
def review_plan(plan_file: str, context: str = "") -> dict:
    plan = load_plan(plan_file)
    plan_summary = extract_key_changes(plan)
    
    user_message = f"""Review this Terraform plan for production safety risks.
 
Context/Purpose of this change: {context or "No context provided"}
 
Terraform Plan JSON:
{plan_summary}
 
Provide your risk assessment."""
 
    message = client.messages.create(
        model="claude-opus-4-8",
        max_tokens=2000,
        system=SYSTEM_PROMPT,
        messages=[{"role": "user", "content": user_message}]
    )
    
    response_text = message.content[0].text
    
    # Parse JSON from response
    try:
        # Find JSON in response
        start = response_text.find("{")
        end = response_text.rfind("}") + 1
        result = json.loads(response_text[start:end])
    except (json.JSONDecodeError, ValueError):
        result = {
            "risk_level": "MEDIUM",
            "recommendation": "REVIEW",
            "summary": "Could not parse structured response",
            "details": response_text
        }
    
    return result
 
 
def print_report(result: dict):
    risk = result.get("risk_level", "UNKNOWN")
    rec = result.get("recommendation", "REVIEW")
    
    colors = {"HIGH": "\033[91m", "MEDIUM": "\033[93m", "LOW": "\033[92m"}
    reset = "\033[0m"
    color = colors.get(risk, "")
    
    print("\n" + "="*60)
    print(f"  TERRAFORM PLAN AI REVIEW")
    print("="*60)
    print(f"  Risk Level:     {color}{risk}{reset}")
    print(f"  Recommendation: {color}{rec}{reset}")
    print(f"  Summary:        {result.get('summary', '')}")
    print("="*60)
    
    if result.get("destroyed_resources"):
        print(f"\n🔴 DESTROYED RESOURCES:")
        for r in result["destroyed_resources"]:
            print(f"   - {r}")
    
    if result.get("security_concerns"):
        print(f"\n⚠️  SECURITY CONCERNS:")
        for c in result["security_concerns"]:
            print(f"   - {c}")
    
    if result.get("unexpected_changes"):
        print(f"\n❓ UNEXPECTED CHANGES:")
        for u in result["unexpected_changes"]:
            print(f"   - {u}")
    
    print(f"\n📋 DETAILS:\n{result.get('details', '')}")
    print("="*60 + "\n")
 
 
def main():
    parser = argparse.ArgumentParser(description="AI Terraform Plan Reviewer")
    parser.add_argument("plan_file", help="Path to terraform plan JSON file")
    parser.add_argument("--context", "-c", help="Context/purpose of this change", default="")
    parser.add_argument("--fail-on", choices=["HIGH", "MEDIUM"], default="HIGH",
                       help="Exit with code 1 if risk level meets threshold")
    args = parser.parse_args()
    
    print(f"🔍 Reviewing Terraform plan: {args.plan_file}")
    
    result = review_plan(args.plan_file, args.context)
    print_report(result)
    
    risk = result.get("risk_level", "LOW")
    
    if args.fail_on == "HIGH" and risk == "HIGH":
        print("❌ Blocking: HIGH risk changes detected")
        sys.exit(1)
    elif args.fail_on == "MEDIUM" and risk in ["HIGH", "MEDIUM"]:
        print("❌ Blocking: MEDIUM or higher risk changes detected")
        sys.exit(1)
    
    print("✅ Plan review passed")
    sys.exit(0)
 
 
if __name__ == "__main__":
    main()

# .github/workflows/terraform-review.yml
name: Terraform Plan Review
 
on:
  pull_request:
    paths:
      - '**.tf'
      - '**.tfvars'
 
jobs:
  plan-review:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Setup Terraform
        uses: hashicorp/setup-terraform@v3
        with:
          terraform_version: "1.9.0"
      
      - name: Terraform Init
        run: terraform init
        env:
          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
      
      - name: Generate Terraform Plan JSON
        run: |
          terraform plan -out=tfplan.binary
          terraform show -json tfplan.binary > tfplan.json
        env:
          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
      
      - name: Install AI Reviewer
        run: pip install anthropic
      
      - name: AI Plan Review
        run: |
          python tf_reviewer.py tfplan.json \
            --context "PR: ${{ github.event.pull_request.title }}" \
            --fail-on HIGH
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
      
      - name: Post Review to PR
        if: always()
        uses: actions/github-script@v7
        with:
          script: |
            // Read review output and post as PR comment
            const fs = require('fs');
            // Add your comment logic here