Build an AI Terraform Drift Detector with Claude API

pip install anthropic python-dotenv

import subprocess
import json
import re
import os
from typing import Optional
 
 
def run_terraform_plan(
    working_dir: str,
    workspace: Optional[str] = None,
    var_file: Optional[str] = None
) -> tuple[str, str, int]:
    """
    Run terraform plan and return stdout, stderr, exit code.
    Exit code 0 = no changes, 2 = changes detected, other = error.
    """
    # Initialize if needed
    subprocess.run(
        ["terraform", "init", "-upgrade", "-no-color"],
        cwd=working_dir,
        capture_output=True
    )
 
    if workspace:
        subprocess.run(
            ["terraform", "workspace", "select", workspace],
            cwd=working_dir,
            capture_output=True
        )
 
    cmd = ["terraform", "plan", "-detailed-exitcode", "-no-color", "-json"]
    if var_file:
        cmd.extend(["-var-file", var_file])
 
    result = subprocess.run(
        cmd,
        cwd=working_dir,
        capture_output=True,
        text=True
    )
 
    return result.stdout, result.stderr, result.returncode
 
 
def parse_plan_json(plan_output: str) -> dict:
    """
    Parse terraform plan JSON output to extract changes.
    Terraform plan -json outputs one JSON object per line.
    """
    changes = {
        "resource_changes": [],
        "outputs": [],
        "drift_resources": [],
        "planned_resources": []
    }
 
    for line in plan_output.splitlines():
        if not line.strip():
            continue
        try:
            event = json.loads(line)
        except json.JSONDecodeError:
            continue
 
        event_type = event.get("type")
 
        if event_type == "resource_drift":
            # This is actual drift — resource changed outside Terraform
            changes["drift_resources"].append({
                "address": event.get("change", {}).get("resource", {}).get("addr"),
                "action": event.get("change", {}).get("action"),
                "before": event.get("change", {}).get("before"),
                "after": event.get("change", {}).get("after"),
            })
 
        elif event_type == "planned_change":
            # This is intentional planned change
            change = event.get("change", {})
            action = change.get("action")
            if action not in ["no-op", "read"]:
                changes["planned_resources"].append({
                    "address": change.get("resource", {}).get("addr"),
                    "action": action,
                    "before": change.get("before"),
                    "after": change.get("after"),
                })
 
        elif event_type == "change_summary":
            changes["summary"] = event.get("changes", {})
 
    return changes
 
 
def extract_attribute_diffs(before: dict, after: dict, path: str = "") -> list[dict]:
    """
    Find attributes that changed between before and after state.
    Returns list of {attribute, old_value, new_value}.
    """
    if not before or not after:
        return []
 
    diffs = []
    all_keys = set(list(before.keys()) + list(after.keys()))
 
    for key in all_keys:
        full_path = f"{path}.{key}" if path else key
        old_val = before.get(key)
        new_val = after.get(key)
 
        if old_val != new_val:
            if isinstance(old_val, dict) and isinstance(new_val, dict):
                diffs.extend(extract_attribute_diffs(old_val, new_val, full_path))
            else:
                diffs.append({
                    "attribute": full_path,
                    "old_value": str(old_val)[:200] if old_val else None,
                    "new_value": str(new_val)[:200] if new_val else None
                })
 
    return diffs

import anthropic
 
 
def analyze_drift_with_claude(
    drift_resources: list[dict],
    planned_resources: list[dict],
    terraform_dir: str,
    environment: str = "production"
) -> str:
    """
    Send drift analysis to Claude for risk assessment and remediation.
    """
    if not drift_resources and not planned_resources:
        return "No drift or planned changes detected. Infrastructure matches Terraform state."
 
    client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
 
    # Build detailed diff context for each drifted resource
    drift_details = []
    for resource in drift_resources:
        diffs = extract_attribute_diffs(
            resource.get("before", {}),
            resource.get("after", {})
        )
        drift_details.append({
            "resource": resource["address"],
            "action": resource["action"],
            "changed_attributes": diffs[:20]  # Limit to top 20 changes
        })
 
    prompt = f"""You are a Terraform and cloud infrastructure expert analyzing infrastructure drift.
 
## Environment
{environment}
 
## Terraform Directory
{terraform_dir}
 
## Drift Summary (Resources Changed Outside Terraform)
{json.dumps(drift_details, indent=2)}
 
## Planned Changes (Intentional Terraform Changes Pending)
{json.dumps([{
    "resource": r["address"],
    "action": r["action"],
    "changed_attributes": extract_attribute_diffs(r.get("before", {}), r.get("after", {}))[:10]
} for r in planned_resources[:10]], indent=2)}
 
---
 
Analyze this Terraform plan and provide:
 
## 1. Drift Assessment
For each drifted resource, explain:
- What changed (in plain English, not raw attribute names)
- The likely cause (manual console change, auto-scaling, AWS-managed update, etc.)
- Risk level: Critical / High / Medium / Low
- Whether this drift is dangerous in production
 
## 2. Planned Changes Assessment  
For intentional planned changes:
- What will change when terraform apply runs
- Any potential risks or unintended side effects
- Whether any changes look unexpected
 
## 3. Recommended Action
For each drifted resource, recommend one of:
- **Apply now**: Run terraform apply to revert the drift immediately
- **Update Terraform**: The change was intentional — update the Terraform code to match reality
- **Investigate first**: Need to understand why this changed before acting
- **Accept**: This is expected drift (auto-scaling, AWS-managed, etc.)
 
## 4. Priority Order
If multiple issues, list them in priority order with exact terraform commands to resolve each.
 
Keep the response practical and specific to the resource types shown."""
 
    message = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=2000,
        messages=[{"role": "user", "content": prompt}]
    )
 
    return message.content[0].text
 
 
def run_drift_analysis(
    terraform_dir: str,
    environment: str = "production",
    workspace: Optional[str] = None,
    var_file: Optional[str] = None
) -> dict:
    """Main entry point — run plan, parse, analyze, return report."""
    print(f"Running terraform plan in {terraform_dir}...")
 
    stdout, stderr, exit_code = run_terraform_plan(terraform_dir, workspace, var_file)
 
    if exit_code == 1:
        return {
            "status": "error",
            "message": f"Terraform plan failed:\n{stderr}",
            "analysis": None
        }
 
    print("Parsing plan output...")
    changes = parse_plan_json(stdout)
 
    drift_count = len(changes["drift_resources"])
    planned_count = len(changes["planned_resources"])
 
    print(f"Found {drift_count} drifted resources, {planned_count} planned changes")
 
    if drift_count == 0 and planned_count == 0:
        return {
            "status": "clean",
            "message": "No drift detected. Infrastructure matches Terraform state.",
            "analysis": None
        }
 
    print("Analyzing with Claude API...")
    analysis = analyze_drift_with_claude(
        changes["drift_resources"],
        changes["planned_resources"],
        terraform_dir,
        environment
    )
 
    return {
        "status": "drift_detected" if drift_count > 0 else "changes_pending",
        "drift_count": drift_count,
        "planned_count": planned_count,
        "analysis": analysis,
        "raw_changes": changes
    }

import sys
import urllib.request
 
 
def post_to_slack(message: str, webhook_url: str):
    """Post drift report to Slack."""
    payload = {
        "text": f":terraform: *Terraform Drift Report*",
        "blocks": [
            {"type": "header", "text": {"type": "plain_text", "text": "Terraform Drift Detected"}},
            {"type": "section", "text": {"type": "mrkdwn", "text": message[:3000]}}
        ]
    }
    data = json.dumps(payload).encode()
    req = urllib.request.Request(
        webhook_url, data=data,
        headers={"Content-Type": "application/json"}
    )
    urllib.request.urlopen(req)
 
 
if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser(description="Terraform drift detector with AI analysis")
    parser.add_argument("--dir", default=".", help="Terraform directory")
    parser.add_argument("--env", default="production", help="Environment name")
    parser.add_argument("--workspace", help="Terraform workspace")
    parser.add_argument("--var-file", help="Terraform var file")
    parser.add_argument("--slack-webhook", help="Slack webhook URL for notifications")
    args = parser.parse_args()
 
    from dotenv import load_dotenv
    load_dotenv()
 
    report = run_drift_analysis(args.dir, args.env, args.workspace, args.var_file)
 
    print("\n" + "="*60)
    print(f"STATUS: {report['status'].upper()}")
    print("="*60)
 
    if report["analysis"]:
        print(report["analysis"])
 
        if args.slack_webhook and report["status"] != "clean":
            post_to_slack(report["analysis"], args.slack_webhook)
            print("\nSlack notification sent.")
 
    # Exit 1 if critical drift detected (for CI/CD gate)
    if report["status"] == "drift_detected":
        sys.exit(1)

# .github/workflows/terraform-drift.yml
name: Terraform Drift Detection
 
on:
  schedule:
    - cron: "0 8 * * *"  # Daily at 8 AM
  workflow_dispatch:
 
jobs:
  detect-drift:
    runs-on: ubuntu-latest
    strategy:
      matrix:
        environment: [production, staging]
 
    steps:
      - uses: actions/checkout@v4
 
      - uses: hashicorp/setup-terraform@v3
        with:
          terraform_version: "1.8.0"
 
      - uses: aws-actions/configure-aws-credentials@v4
        with:
          role-to-assume: ${{ secrets.AWS_TERRAFORM_ROLE }}
          aws-region: us-east-1
 
      - name: Run drift detection
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
          SLACK_WEBHOOK: ${{ secrets.SLACK_DRIFT_WEBHOOK }}
        run: |
          pip install anthropic python-dotenv
          python drift_detector.py \
            --dir terraform/environments/${{ matrix.environment }} \
            --env ${{ matrix.environment }} \
            --workspace ${{ matrix.environment }} \
            --slack-webhook $SLACK_WEBHOOK
        continue-on-error: true  # Don't fail the workflow, just notify