Terraform Plan Parser and Risk Scorer

Problem Statement

Infrastructure as Code (IaC) pipelines often execute terraform apply automatically or require human approval during Pull Request (PR) reviews. However, inspecting raw Terraform plans or multi-megabyte JSON outputs is tedious and error-prone. A developer might accidentally approve a plan that destroys an production database instance, deletes a core S3 bucket, or replaces critical IAM policies alongside hundreds of benign updates.

This script parses the structured JSON output generated by terraform show -json <plan-file> and computes an objective Risk Score for the proposed changes. By categorizing operations (create, update, delete, replace) and cross-referencing resource types against configurable criticality rules, it acts as an automated CI/CD guardrail that blocks dangerous applies or highlights high-risk infrastructure changes before human review.

How It Works

The tool follows a linear processing pipeline using Python's standard library:

1. Ingestion & Parsing: Reads the Terraform plan JSON file and extracts the resource_changes array, which details every resource's prior state, planned state, and pending actions.
2. Action Classification: Normalizes Terraform actions into standardized operations: no-op, create, update, delete, or replace (destroy-then-create).
3. Risk Matrix Assessment: Evaluates each change against a rule engine. Higher base risk scores are assigned to destructive operations (delete, replace), which are further multiplied if the resource type belongs to critical infrastructure classes (e.g., databases, IAM rules, object storage, KMS keys).
4. Aggregation & Thresholding: Aggregates individual resource scores into an overall plan score and assigns a plan-level risk classification (LOW, MEDIUM, HIGH, CRITICAL). If the plan risk exceeds the target --risk-threshold (or TF_RISK_THRESHOLD), the script exits with a non-zero exit code to fail CI/CD build pipelines.

Full Script

#!/usr/bin/env python3
"""
Terraform Plan Parser and Risk Scorer

Parses a Terraform JSON plan file, evaluates resource changes against a risk matrix,
and outputs risk metrics. Exits with non-zero status if risk exceeds threshold.
"""

import argparse
from dataclasses import dataclass, field
from enum import IntEnum
import json
import os
import sys
from typing import Any

# --- Configuration defaults from Environment ---
ENV_RISK_THRESHOLD = os.environ.get("TF_RISK_THRESHOLD", "medium").lower()
ENV_STRICT_MODE = os.environ.get("TF_STRICT_MODE", "false").lower() in ("true", "1", "yes")
ENV_CUSTOM_CRITICAL_RESOURCES = os.environ.get("TF_CRITICAL_RESOURCES", "")


class RiskLevel(IntEnum):
    LOW = 1
    MEDIUM = 2
    HIGH = 3
    CRITICAL = 4

    @classmethod
    def from_str(cls, label: str) -> "RiskLevel":
        try:
            return cls[label.upper()]
        except KeyError:
            valid = ", ".join([e.name.lower() for e in cls])
            raise ValueError(f"Invalid risk level '{label}'. Must be one of: {valid}")

    def __str__(self) -> str:
        return self.name


# High-risk resource type patterns (AWS, Azure, GCP common defaults)
DEFAULT_CRITICAL_PATTERNS = {
    "db_",
    "database",
    "rds",
    "s3_bucket",
    "storage_account",
    "iam_",
    "kms_",
    "key_vault",
    "security_group",
    "firewall",
    "vpc",
}


@dataclass
class ResourceAssessment:
    address: str
    resource_type: str
    actions: list[str]
    is_critical_resource: bool
    score: int
    risk_level: RiskLevel
    reason: str


@dataclass
class PlanSummary:
    total_resources: int = 0
    creates: int = 0
    updates: int = 0
    deletes: int = 0
    replaces: int = 0
    no_ops: int = 0
    total_score: int = 0
    max_risk_level: RiskLevel = RiskLevel.LOW
    assessments: list[ResourceAssessment] = field(default_factory=list)


class TerraformRiskAnalyzer:
    def __init__(self, custom_critical_patterns: list[str] | None = None):
        self.critical_patterns = set(DEFAULT_CRITICAL_PATTERNS)
        if custom_critical_patterns:
            self.critical_patterns.update(custom_critical_patterns)

    def _is_critical(self, resource_type: str) -> bool:
        return any(pattern in resource_type for pattern in self.critical_patterns)

    def _normalize_actions(self, actions: list[str]) -> str:
        if actions == ["no-op"]:
            return "no-op"
        if actions == ["create"]:
            return "create"
        if actions == ["update"]:
            return "update"
        if actions == ["delete"]:
            return "delete"
        if set(actions) == {"create", "delete"}:
            return "replace"
        if actions == ["read"]:
            return "no-op"
        return "update"

    def evaluate_resource(self, change: dict[str, Any]) -> ResourceAssessment:
        address = change.get("address", "unknown")
        resource_type = change.get("type", "unknown")
        change_detail = change.get("change", {})
        raw_actions = change_detail.get("actions", ["no-op"])
        
        action = self._normalize_actions(raw_actions)
        is_crit = self._is_critical(resource_type)

        # Base scoring matrix based on operation safety
        # Action -> (Base Score, Non-Critical Level, Critical Level)
        scoring_matrix = {
            "no-op": (0, RiskLevel.LOW, RiskLevel.LOW),
            "create": (1, RiskLevel.LOW, RiskLevel.MEDIUM),
            "update": (2, RiskLevel.LOW, RiskLevel.MEDIUM),
            "delete": (5, RiskLevel.HIGH, RiskLevel.CRITICAL),
            "replace": (8, RiskLevel.HIGH, RiskLevel.CRITICAL),
        }

        base_score, std_level, crit_level = scoring_matrix.get(
            action, (2, RiskLevel.MEDIUM, RiskLevel.HIGH)
        )

        multiplier = 2.5 if is_crit else 1.0
        final_score = int(base_score * multiplier)
        assigned_level = crit_level if is_crit else std_level

        # Reason generator
        crit_tag = " [CRITICAL RESOURCE]" if is_crit else ""
        reason = f"Operation '{action}' on '{resource_type}'{crit_tag}"

        return ResourceAssessment(
            address=address,
            resource_type=resource_type,
            actions=raw_actions,
            is_critical_resource=is_crit,
            score=final_score,
            risk_level=assigned_level,
            reason=reason,
        )

    def analyze_plan(self, plan_data: dict[str, Any]) -> PlanSummary:
        summary = PlanSummary()
        resource_changes = plan_data.get("resource_changes", [])
        summary.total_resources = len(resource_changes)

        for change in resource_changes:
            # Skip resources with no actual changes planned
            actions = change.get("change", {}).get("actions", [])
            if actions == ["no-op"]:
                summary.no_ops += 1
                continue

            assessment = self.evaluate_resource(change)
            summary.assessments.append(assessment)
            summary.total_score += assessment.score

            # Track action metrics
            norm_action = self._normalize_actions(actions)
            if norm_action == "create":
                summary.creates += 1
            elif norm_action == "update":
                summary.updates += 1
            elif norm_action == "delete":
                summary.deletes += 1
            elif norm_action == "replace":
                summary.replaces += 1

            # Elevate maximum overall plan risk
            if assessment.risk_level > summary.max_risk_level:
                summary.max_risk_level = assessment.risk_level

        return summary


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Parse Terraform JSON plans and assess execution risk."
    )
    parser.add_argument(
        "-p", "--plan", required=True, help="Path to the Terraform plan JSON file"
    )
    parser.add_argument(
        "-r",
        "--risk-threshold",
        default=ENV_RISK_THRESHOLD,
        help="Maximum allowable risk level (low, medium, high, critical). Default: medium",
    )
    parser.add_argument(
        "--strict",
        action="store_true",
        default=ENV_STRICT_MODE,
        help="Fail if any destructive changes (delete/replace) exist regardless of score.",
    )
    parser.add_argument(
        "--json-output",
        action="store_true",
        help="Format output as JSON instead of human-readable text.",
    )
    return parser.parse_args()


def format_text_report(summary: PlanSummary, threshold: RiskLevel) -> str:
    lines = []
    lines.append("=" * 60)
    lines.append("          TERRAFORM PLAN RISK ASSESSMENT REPORT          ")
    lines.append("=" * 60)
    lines.append(f" Total Changes Analyzed : {len(summary.assessments)}")
    lines.append(f"  [+] Creates           : {summary.creates}")
    lines.append(f"  [~] Updates           : {summary.updates}")
    lines.append(f"  [-] Deletes           : {summary.deletes}")
    lines.append(f"  [±] Replaces          : {summary.replaces}")
    lines.append("-" * 60)
    lines.append(f" Calculated Risk Score : {summary.total_score}")
    lines.append(f" Overall Risk Level    : {summary.max_risk_level.name}")
    lines.append(f" Configured Threshold  : {threshold.name}")
    lines.append("-" * 60)

    if summary.assessments:
        lines.append(" Detailed High Risk Changes:")
        for item in summary.assessments:
            if item.risk_level >= RiskLevel.MEDIUM:
                lines.append(
                    f"  - [{item.risk_level.name}] {item.address} ({item.reason})"
                )
    else:
        lines.append(" No impactful resource changes detected.")

    lines.append("=" * 60)
    return "\n".join(lines)


def main() -> None:
    args = parse_args()

    try:
        target_threshold = RiskLevel.from_str(args.risk_threshold)
    except ValueError as err:
        print(f"Configuration Error: {err}", file=sys.stderr)
        sys.exit(2)

    if not os.path.exists(args.plan):
        print(f"Error: Plan file '{args.plan}' not found.", file=sys.stderr)
        sys.exit(2)

    try:
        with open(args.plan, "r", encoding="utf-8") as f:
            plan_data = json.load(f)
    except json.JSONDecodeError as err:
        print(f"Error: Invalid JSON format in '{args.plan}': {err}", file=sys.stderr)
        sys.exit(2)
    except Exception as err:
        print(f"Error reading file '{args.plan}': {err}", file=sys.stderr)
        sys.exit(2)

    # Custom resources from environment
    custom_patterns = [
        p.strip() for p in ENV_CUSTOM_CRITICAL_RESOURCES.split(",") if p.strip()
    ]

    analyzer = TerraformRiskAnalyzer(custom_critical_patterns=custom_patterns)
    summary = analyzer.analyze_plan(plan_data)

    if args.json_output:
        report_data = {
            "total_score": summary.total_score,
            "max_risk_level": summary.max_risk_level.name,
            "threshold": target_threshold.name,
            "threshold_exceeded": summary.max_risk_level > target_threshold,
            "changes": {
                "creates": summary.creates,
                "updates": summary.updates,
                "deletes": summary.deletes,
                "replaces": summary.replaces,
            },
            "assessments": [
                {
                    "address": a.address,
                    "type": a.resource_type,
                    "risk_level": a.risk_level.name,
                    "score": a.score,
                    "reason": a.reason,
                }
                for a in summary.assessments
            ],
        }
        print(json.dumps(report_data, indent=2))
    else:
        print(format_text_report(summary, target_threshold))

    # Evaluate execution results against policies
    failed = False
    if summary.max_risk_level > target_threshold:
        print(
            f"POLICY FAILURE: Calculated risk level ({summary.max_risk_level.name}) "
            f"exceeds allowable threshold ({target_threshold.name}).",
            file=sys.stderr,
        )
        failed = True

    if args.strict and (summary.deletes > 0 or summary.replaces > 0):
        print(
            "POLICY FAILURE: Strict mode enabled and plan contains destructive changes (deletes/replaces).",
            file=sys.stderr,
        )
        failed = True

    if failed:
        sys.exit(1)

    print("SUCCESS: Terraform plan passed risk policy evaluation.")
    sys.exit(0)


if __name__ == "__main__":
    main()

Configuration

The script can be configured via environment variables or CLI flags. Environment variables serve as defaults, which CLI flags override.

Usage Examples

1. Basic Local Scan

Generate a plan JSON using Terraform, then score it against the default medium threshold:

# Export Terraform plan to JSON format
terraform plan -out=tfplan.binary
terraform show -json tfplan.binary > tfplan.json

# Run the risk scorer
python3 tf_plan.py --plan tfplan.json

Output:

============================================================
          TERRAFORM PLAN RISK ASSESSMENT REPORT          
============================================================
 Total Changes Analyzed : 3
  [+] Creates           : 1
  [~] Updates           : 1
  [-] Deletes           : 1
  [±] Replaces          : 0
------------------------------------------------------------
 Calculated Risk Score : 8
 Overall Risk Level    : HIGH
 Configured Threshold  : MEDIUM
------------------------------------------------------------
 Detailed High Risk Changes:
  - [HIGH] aws_s3_bucket.data_archive (Operation 'delete' on 'aws_s3_bucket' [CRITICAL RESOURCE])
  - [MEDIUM] aws_instance.web_app (Operation 'update' on 'aws_instance')
============================================================
POLICY FAILURE: Calculated risk level (HIGH) exceeds allowable threshold (MEDIUM).

2. CI/CD Automated Guardrail with Environment Variables

Configure strict execution in a pipeline (GitHub Actions, GitLab CI, or Jenkins) where no destructive actions are permitted and extra critical resource patterns are monitored:

export TF_RISK_THRESHOLD="low"
export TF_STRICT_MODE="true"
export TF_CRITICAL_RESOURCES="helm_release,vault_"

python3 tf_plan.py --plan tfplan.json --json-output

Output:

{
  "total_score": 12,
  "max_risk_level": "CRITICAL",
  "threshold": "LOW",
  "threshold_exceeded": true,
  "changes": {
    "creates": 0,
    "updates": 1,
    "deletes": 0,
    "replaces": 1
  },
  "assessments": [
    {
      "address": "helm_release.ingress",
      "type": "helm_release",
      "risk_level": "CRITICAL",
      "score": 12,
      "reason": "Operation 'replace' on 'helm_release' [CRITICAL RESOURCE]"
    }
  ]
}

Error Handling

The script includes explicit error handling for edge cases common in automated pipelines:

• File Not Found: Exits with code 2 if