Terraform Plan Parser and Notifier

Terraform Plan Parser and Notifier

Problem Statement

Terraform's plan output is verbose and easy to miss critical changes buried in hundreds of lines of JSON. In CI/CD pipelines, a terraform plan that destroys a production database looks identical to one that just adds a tag — until someone actually reads the output. Teams need automated triage: parse the plan, classify changes by risk (create/update/replace/destroy), and route notifications to the right channel with enough context for a human to approve or block.

This script wraps python-terraform, generates a plan file, parses the resulting JSON, summarizes resource actions (especially destructive ones), and posts a formatted message to Slack, Discord, or stdout. It can also auto-approve when the plan is empty, which is handy for gating pipelines without human intervention.

How It Works

The script uses python-terraform's Terraform class to run terraform init (optional) and terraform plan -out=tfplan, then terraform show -json tfplan to get a machine-readable plan. It parses resource_changes from the JSON, grouping each resource by the action set (create, update, delete, replace). A risk score is computed based on how many destructive actions exist, and the notifier dispatches a colored message to the chosen backend.

Design decisions: (1) The plan file is written to a temp directory so we never clobber the user's working tree. (2) Notifiers are pluggable — each is a small class implementing send(summary). (3) Exit codes are meaningful: 0 = no changes (and optionally auto-approved), 1 = changes present, 2 = error — so pipelines can branch on the result.

Full Script

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

Runs `terraform plan`, parses the resulting JSON, summarizes resource changes
(create/update/replace/destroy), and notifies Slack/Discord/stdout. Optionally
auto-approves when there are no changes.

Exit codes:
    0 -> no changes detected (or --approve-on-no-changes and no changes)
    1 -> changes detected
    2 -> error during plan or parsing
"""

from __future__ import annotations

import argparse
import json
import os
import sys
import tempfile
import textwrap
import urllib.error
import urllib.request
from dataclasses import dataclass, field
from typing import Iterable

try:
    from python_terraform import Terraform, IsFlagged
except ImportError:  # pragma: no cover
    print("ERROR: python-terraform not installed. Run: pip install python-terraform",
          file=sys.stderr)
    sys.exit(2)


# ---------------------------------------------------------------------------
# Configuration (env vars with sensible defaults)
# ---------------------------------------------------------------------------
SLACK_WEBHOOK = os.environ.get("SLACK_WEBHOOK_URL", "")
DISCORD_WEBHOOK = os.environ.get("DISCORD_WEBHOOK_URL", "")
TF_BIN = os.environ.get("TF_BIN", "terraform")
DEFAULT_TIMEOUT = int(os.environ.get("TF_PLAN_TIMEOUT", "600"))


# ---------------------------------------------------------------------------
# Data model
# ---------------------------------------------------------------------------
@dataclass
class ResourceChange:
    address: str
    actions: tuple[str, ...]
    resource_type: str = ""

    @property
    def is_destructive(self) -> bool:
        return "delete" in self.actions

    @property
    def is_replace(self) -> bool:
        return set(self.actions) == {"delete", "create"}

    def symbol(self) -> str:
        if self.is_replace:
            return "🔁 replace"
        if self.is_destructive:
            return "❌ destroy"
        if "create" in self.actions:
            return "➕ create"
        if "update" in self.actions:
            return "✏️  update"
        return "❓ " + ",".join(self.actions)


@dataclass
class PlanSummary:
    creates: list[ResourceChange] = field(default_factory=list)
    updates: list[ResourceChange] = field(default_factory=list)
    destroys: list[ResourceChange] = field(default_factory=list)
    replaces: list[ResourceChange] = field(default_factory=list)
    no_op_count: int = 0

    @property
    def total_changes(self) -> int:
        return len(self.creates) + len(self.updates) + len(self.destroys) + len(self.replaces)

    @property
    def has_destructive(self) -> bool:
        return bool(self.destroys or self.replaces)

    @property
    def risk_level(self) -> str:
        if self.destroys:
            return "HIGH"
        if self.replaces:
            return "MEDIUM"
        if self.updates:
            return "LOW"
        if self.creates:
            return "LOW"
        return "NONE"

    def render_text(self, directory: str) -> str:
        lines = [
            f"*Terraform Plan* — `{directory}`",
            f"Risk: *{self.risk_level}* | Changes: {self.total_changes} "
            f"(+{len(self.creates)} ~{len(self.updates)} -{len(self.destroys)} 🔁{len(self.replaces)})",
        ]
        if self.total_changes == 0:
            lines.append("✅ No changes. Infrastructure matches configuration.")
            return "\n".join(lines)

        for label, items in (
            ("Destroy", self.destroys),
            ("Replace", self.replaces),
            ("Update", self.updates),
            ("Create", self.creates),
        ):
            if not items:
                continue
            lines.append(f"\n*{label}* ({len(items)}):")
            for rc in items[:20]:  # cap to keep messages readable
                lines.append(f"  • {rc.symbol()} `{rc.address}`")
            if len(items) > 20:
                lines.append(f"  … and {len(items) - 20} more")
        return "\n".join(lines)


# ---------------------------------------------------------------------------
# Terraform execution + parsing
# ---------------------------------------------------------------------------
class TerraformPlanError(RuntimeError):
    pass


def run_plan(directory: str, init: bool = False, var_file: str | None = None) -> PlanSummary:
    """Run terraform plan and return a parsed summary."""
    tf = Terraform(working_dir=directory, terraform_bin_path=TF_BIN)

    if init:
        rc, _, err = tf.init(capture_output=True, no_color=IsFlagged)
        if rc != 0:
            raise TerraformPlanError(f"terraform init failed: {err.strip()}")

    with tempfile.TemporaryDirectory(prefix="tfplan-") as tmpdir:
        plan_file = os.path.join(tmpdir, "tfplan.bin")
        plan_kwargs = {
            "out": plan_file,
            "capture_output": True,
            "no_color": IsFlagged,
            "detailed_exitcode": IsFlagged,  # 2 = changes present
        }
        if var_file:
            plan_kwargs["var_file"] = var_file

        rc, _, err = tf.plan(**plan_kwargs)
        # python-terraform returns 0 or 2 depending on version; treat 2 as "changes"
        if rc not in (0, 2):
            raise TerraformPlanError(f"terraform plan failed (rc={rc}): {err.strip()}")

        # Show the plan as JSON
        rc, out, err = tf.show(json=IsFlagged, plan_file=plan_file, capture_output=True)
        if rc != 0:
            raise TerraformPlanError(f"terraform show failed: {err.strip()}")

        try:
            data = json.loads(out)
        except json.JSONDecodeError as exc:
            raise TerraformPlanError(f"Could not parse plan JSON: {exc}") from exc

    return parse_plan(data)


def parse_plan(data: dict) -> PlanSummary:
    summary = PlanSummary()
    for change in data.get("resource_changes", []):
        actions = tuple(change.get("change", {}).get("actions", []))
        if actions == ("no-op",) or not actions:
            summary.no_op_count += 1
            continue
        rc = ResourceChange(
            address=change.get("address", "unknown"),
            actions=actions,
            resource_type=change.get("type", ""),
        )
        if rc.is_replace:
            summary.replaces.append(rc)
        elif rc.is_destructive:
            summary.destroys.append(rc)
        elif "create" in actions:
            summary.creates.append(rc)
        elif "update" in actions:
            summary.updates.append(rc)
    return summary


# ---------------------------------------------------------------------------
# Notifiers
# ---------------------------------------------------------------------------
class Notifier:
    def send(self, message: str, risk: str) -> None:
        raise NotImplementedError


class StdoutNotifier(Notifier):
    def send(self, message: str, risk: str) -> None:
        border = "=" * 70
        print(border)
        print(f" RISK: {risk}")
        print(border)
        # Strip Slack-style *bold* markers for plain output
        print(message.replace("*", ""))


class SlackNotifier(Notifier):
    COLORS = {"HIGH": "danger", "MEDIUM": "warning", "LOW": "good", "NONE": "good"}

    def __init__(self, webhook: str, channel: str | None = None) -> None:
        if not webhook:
            raise ValueError("SLACK_WEBHOOK_URL is not set")
        self.webhook = webhook
        self.channel = channel

    def send(self, message: str, risk: str) -> None:
        payload = {
            "attachments": [{
                "color": self.COLORS.get(risk, "#888888"),
                "text": message,
                "mrkdwn_in": ["text"],
            }]
        }
        if self.channel:
            payload["channel"] = self.channel
        self._post(payload)

    def _post(self, payload: dict) -> None:
        req = urllib.request.Request(
            self.webhook,
            data=json.dumps(payload).encode(),
            headers={"Content-Type": "application/json"},
        )
        try:
            with urllib.request.urlopen(req, timeout=15) as resp:
                if resp.status >= 300:
                    raise RuntimeError(f"Slack webhook returned {resp.status}")
        except urllib.error.URLError as exc:
            raise RuntimeError(f"Failed to notify Slack: {exc}") from exc


class DiscordNotifier(Notifier):
    COLORS = {"HIGH": 0xE74C3C, "MEDIUM": 0xF1C40F, "LOW": 0x2ECC71, "NONE": 0x2ECC71}

    def __init__(self, webhook: str) -> None:
        if not webhook:
            raise ValueError("DISCORD_WEBHOOK_URL is not set")
        self.webhook = webhook

    def send(self, message: str, risk: str) -> None:
        payload = {
            "embeds": [{
                "title": f"Terraform Plan ({risk} risk)",
                "description": message.replace("*", "**"),
                "color": self.COLORS.get(risk, 0x888888),
            }]
        }
        req = urllib.request.Request(
            self.webhook,
            data=json.dumps(payload).encode(),
            headers={"Content-Type": "application/json"},
        )
        try:
            with urllib.request.urlopen(req, timeout=15) as resp:
                if resp.status >= 300:
                    raise RuntimeError(f"Discord webhook returned {resp.status}")
        except urllib.error.URLError as exc:
            raise RuntimeError(f"Failed to notify Discord: {exc}") from exc


def build_notifier(name: str, channel: str | None) -> Notifier:
    if name == "stdout":
        return StdoutNotifier()
    if name == "slack":
        return SlackNotifier(SLACK_WEBHOOK, channel)
    if name == "discord":
        return DiscordNotifier(DISCORD_WEBHOOK)
    raise ValueError(f"Unknown notifier: {name}")


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def parse_args(argv: Iterable[str] | None = None) -> argparse.Namespace:
    p = argparse.ArgumentParser(description="Parse a Terraform plan and notify.")
    p.add_argument("--dir", default=".", help="Terraform working directory")
    p.add_argument("--notify", choices=["stdout", "slack", "discord"], default="stdout")
    p.add_argument("--channel", default=None, help="Channel override (Slack only)")
    p.add_argument("--init", action="store_true", help="Run terraform init first")
    p.add_argument("--var-file", default=None, help="Terraform var file")
    p.add_argument(
        "--approve-on-no-changes",
        action="store_true",
        help="Exit 0 when no changes are detected (for pipeline gating)",
    )
    return p.parse_args(argv)


def main(argv: Iterable[str] | None = None) -> int:
    args = parse_args(argv)
    directory = os.path.abspath(args.dir)

    if not os.path.isdir(directory):
        print(f"ERROR: directory does not exist: {directory}", file=sys.stderr)
        return 2

    try:
        summary = run_plan(directory, init=args.init, var_file=args.var_file)
    except TerraformPlanError as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 2

    message = summary.render_text(directory)

    try:
        notifier = build_notifier(args.notify, args.channel)
        notifier.send(message, summary.risk_level)
    except (ValueError, RuntimeError) as exc:
        # Notification failure shouldn't hide the plan result — print and continue
        print(f"WARNING: notification failed: {exc}", file=sys.stderr)
        print(textwrap.dedent(message.replace("*", "")), file=sys.stderr)

    if summary.total_changes == 0:
        return 0
    if args.approve_on_no_changes and summary.total_changes == 0:
        return 0
    return 1


if __name__ == "__main__":
    sys.exit(main())

Configuration

CLI flags: --dir (working directory), --notify (stdout/slack/discord), --channel (Slack channel override), --init (run terraform init first), --var-file, and --approve-on-no-changes.

Usage Examples

1. Local dry-run, print plan summary to stdout:

$ python3 tf_plan.py --dir ./infra
======================================================================
 RISK: MEDIUM
======================================================================
Terraform Plan — /home/dev/infra
Risk: MEDIUM | Changes: 3 (+1 ~1 -0 🔁1)

Replace (1):
  • 🔁 replace `aws_db_instance.primary`
Update (1):
  • ✏️  update `aws_security_group.web`
Create (1):
  • ➕ create `aws_lb_listener.https`

2. Post to Slack with a channel override:

$ SLACK_WEBHOOK_URL=https://hooks.slack.com/services/XXX \
  python3 tf_plan.py --dir ./infra --notify slack --channel '#infra-alerts'
# Exit code 1 (changes present)

3. Auto-approve in CI when nothing changed:

$ python3 tf_plan.py --dir ./infra --approve-on-no-changes --init
✅ No changes. Infrastructure matches configuration.
$ echo $?
0

Error Handling

• Terraform binary missing / wrong version → python-terraform raises; caught and returned as exit code 2 with a clear message.
• terraform init failure (bad provider, network) → TerraformPlanError with the captured stderr.
• terraform plan failure (syntax error, missing var) → propagated with the underlying stderr.
• Malformed JSON from terraform show → wraps json.JSONDecodeError in TerraformPlanError.
• Webhook unreachable or non-2xx → logged as a warning; the plan result is still printed to stderr and the process returns the correct exit code, so a transient Slack outage never masks a destructive change.
• Missing webhook env var → ValueError from the notifier constructor, surfaced as a warning (plan output still printed).

Extending Ideas

• Diff detail in notifications: include change.before / change.after for updated resources so reviewers see exactly which attributes change (e.g., instance_type: t3.small → t3.large).
• Policy gate: add an --forbid-destroy-of flag taking a regex; if any matching resource appears in destroys/replaces, exit 3 to hard-block the pipeline.
• Multi-workspace fan-out: iterate over terraform workspace list and produce one summary per workspace, tagging the notification title accordingly.
• Prometheus metrics: emit tf_plan_changes_total{action="destroy"} to a textfile collector so you can alert on destructive-change frequency over time.