Terraform Plan Risk & Blast-Radius Policy Gate on Kestra

id: terraform-plan-risk-policy-gate
namespace: company.team
description: |
  Evaluate Terraform execution plans before deployment to classify infrastructure blast
  radius and detect destructive or stateful changes. Automatically post risk summaries
  to Slack and enforce policy gates to block high-risk plans from reaching production.

triggers:
  - id: on_iac_plan_request
    type: io.kestra.plugin.core.trigger.Webhook
    description: Webhook endpoint triggered by CI/CD pipelines (e.g., GitHub
      Actions, GitLab CI) on pull requests to evaluate Terraform plan risk.
    key: "{{ secret('TERRAFORM_PLAN_WEBHOOK_KEY') }}"

inputs:
  - id: environment
    type: STRING
    defaults: production
    description: Target deployment environment (e.g., production, staging, development).

  - id: plan_file
    type: FILE
    required: false
    description: Optional pre-generated Terraform JSON plan artifact (`terraform
      show -json tfplan > plan.json`). If provided, this artifact is directly
      evaluated without executing Terraform commands.

  - id: terraform_code
    type: STRING
    defaults: |
      terraform {
        required_providers {
          local = {
            source  = "hashicorp/local"
            version = "2.5.2"
          }
          random = {
            source  = "hashicorp/random"
            version = "3.6.3"
          }
        }
      }

      resource "random_id" "server_id" {
        byte_length = 8
      }

      resource "local_file" "app_manifest" {
        filename = "app_manifest.json"
        content = jsonencode({
          service     = "order-processing"
          environment = "production"
          version     = "1.2.0"
        })
      }
    description: Terraform HCL configuration to initialize, plan, and evaluate if
      `plan_file` is omitted. In CI/CD pipelines, this can be injected
      dynamically or mounted from a checked-out repository.

  - id: max_allowed_deletions
    type: INT
    defaults: 0
    description: Maximum number of resource deletions (DELETE actions) permitted
      before escalating policy risk to CRITICAL and REJECTED.

  - id: blast_radius_warning_threshold
    type: INT
    defaults: 25
    description: Blast radius score (0-100) at or above which the plan triggers a
      WARNING or requires explicit approval.

  - id: blast_radius_rejection_threshold
    type: INT
    defaults: 60
    description: Blast radius score (0-100) at or above which the plan is rejected
      by policy.

  - id: fail_on_rejection
    type: BOOL
    defaults: false
    description: Whether to fail pipeline execution when policy verdict is REJECTED
      or APPROVAL_REQUIRED (automated deployment blocking gate). When false
      (default for safe out-of-the-box audit and demo runs), rejected plans log
      a warning without failing the flow execution. Set to true in production
      CI/CD pipelines to enforce automated deployment blocking.

  - id: require_approval_on_warning
    type: BOOL
    defaults: false
    description: Whether to escalate WARNING verdicts to APPROVAL_REQUIRED when the
      plan exceeds warning thresholds or touches operational resources.

  - id: notify_slack
    type: BOOL
    defaults: false
    description: Whether to send a structured risk assessment alert to Slack when
      risk level is CRITICAL or HIGH.

tasks:
  - id: terraform_workspace
    type: io.kestra.plugin.core.flow.WorkingDirectory
    description: Isolated workspace containing Terraform configuration, uploaded
      plans, and generated risk artifacts.
    inputFiles:
      main.tf: "{{ inputs.terraform_code ?? '' }}"
      uploaded_plan.json: "{{ inputs.plan_file }}"
    tasks:
      - id: resolve_plan
        type: io.kestra.plugin.terraform.cli.TerraformCLI
        description: Check for an uploaded plan artifact or initialize Terraform and
          generate execution plan in binary and JSON formats.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: hashicorp/terraform:1.9.8
        commands:
          - |
            if [ -f "uploaded_plan.json" ] && [ -s "uploaded_plan.json" ]; then
              echo "[INFO] Using user-provided Terraform JSON plan artifact."
              cp uploaded_plan.json plan.json
            else
              echo "[INFO] No uploaded plan artifact found; planning infrastructure from terraform_code."
              terraform init -input=false -no-color
              terraform plan -input=false -no-color -out=tfplan
              terraform show -json tfplan > plan.json
            fi
        outputFiles:
          - plan.json

      - id: analyze_plan_risk
        type: io.kestra.plugin.scripts.python.Script
        description: Parse plan.json to deterministically evaluate blast radius,
          stateful destruction, and policy gates.
        taskRunner:
          type: io.kestra.plugin.core.runner.Process
        env:
          PLAN_FILE: plan.json
          MAX_ALLOWED_DELETIONS: "{{ inputs.max_allowed_deletions }}"
          WARNING_THRESHOLD: "{{ inputs.blast_radius_warning_threshold }}"
          REJECTION_THRESHOLD: "{{ inputs.blast_radius_rejection_threshold }}"
          REQUIRE_APPROVAL_ON_WARNING: "{{ inputs.require_approval_on_warning }}"
          ENVIRONMENT: "{{ inputs.environment }}"
        script: |
          import json
          import os
          import sys
          from datetime import datetime, timezone

          STATEFUL_KEYWORDS = [
              "db_instance", "rds_cluster", "dynamodb_table", "sql_database",
              "database_instance", "cosmosdb_", "mongodb_", "redis_instance",
              "elasticache_cluster", "s3_bucket", "storage_bucket", "storage_container",
              "gcs_bucket", "ebs_volume", "persistent_disk", "compute_disk",
              "managed_disk", "persistent_volume", "postgresql_server", "mysql_server",
              "mssql_database", "mssql_server", "mariadb_server", "kafka_cluster",
              "sqs_queue", "eventhub"
          ]

          MAJOR_INFRA_TYPES = {
              "aws_instance", "aws_eks_cluster", "aws_ecs_cluster",
              "google_compute_instance", "google_container_cluster",
              "azurerm_linux_virtual_machine", "azurerm_windows_virtual_machine",
              "azurerm_kubernetes_cluster", "aws_vpc", "aws_subnet",
              "aws_nat_gateway", "aws_internet_gateway", "aws_route_table",
              "aws_security_group", "google_compute_network", "google_compute_subnetwork",
              "azurerm_virtual_network", "azurerm_subnet", "azurerm_network_security_group"
          }

          def is_stateful(res_type: str) -> bool:
              res_lower = res_type.lower()
              return any(kw in res_lower for kw in STATEFUL_KEYWORDS)

          def is_major_infra(res_type: str) -> bool:
              return res_type in MAJOR_INFRA_TYPES

          plan_file = os.environ.get("PLAN_FILE", "plan.json")
          if not os.path.exists(plan_file):
              print(f"[ERROR] Plan artifact '{plan_file}' not found.", file=sys.stderr)
              sys.exit(1)

          if os.path.getsize(plan_file) == 0:
              print(f"[ERROR] Plan artifact '{plan_file}' is empty (0 bytes).", file=sys.stderr)
              sys.exit(1)

          try:
              with open(plan_file, "r", encoding="utf-8") as f:
                  plan_data = json.load(f)
          except Exception as exc:
              print(f"[ERROR] Failed to parse {plan_file} as valid JSON: {exc}", file=sys.stderr)
              sys.exit(1)

          if not isinstance(plan_data, dict):
              print("[ERROR] Invalid plan format: root JSON structure must be an object/dict.", file=sys.stderr)
              sys.exit(1)

          if "format_version" not in plan_data:
              print("[ERROR] Invalid plan format: missing required 'format_version' metadata.", file=sys.stderr)
              sys.exit(1)

          if not isinstance(plan_data["format_version"], str):
              print(f"[ERROR] Invalid plan format: 'format_version' must be a string, got {type(plan_data['format_version']).__name__}.", file=sys.stderr)
              sys.exit(1)

          if "resource_changes" not in plan_data:
              print("[ERROR] Invalid plan format: missing required 'resource_changes' property.", file=sys.stderr)
              sys.exit(1)

          resource_changes = plan_data.get("resource_changes")
          if not isinstance(resource_changes, list):
              print(f"[ERROR] Invalid plan format: 'resource_changes' must be a list, got {type(resource_changes).__name__}.", file=sys.stderr)
              sys.exit(1)

          max_allowed_deletions = int(os.environ.get("MAX_ALLOWED_DELETIONS", 0))
          warning_threshold = int(os.environ.get("WARNING_THRESHOLD", 25))
          rejection_threshold = int(os.environ.get("REJECTION_THRESHOLD", 60))
          require_approval_on_warning = str(os.environ.get("REQUIRE_APPROVAL_ON_WARNING", "false")).lower() in ("true", "1", "yes")
          environment = os.environ.get("ENVIRONMENT", "production")

          VALID_ACTIONS = {"no-op", "create", "read", "update", "delete"}
          VALID_ACTION_SETS = {
              ("no-op",),
              ("read",),
              ("create",),
              ("update",),
              ("delete",),
              ("delete", "create"),
              ("create", "delete")
          }

          created_count = 0
          updated_count = 0
          deleted_count = 0
          replaced_count = 0
          no_op_count = 0
          read_count = 0
          stateful_deletions = 0
          stateful_replacements = 0
          major_infra_deletions = 0
          major_infra_replacements = 0

          affected_resources = []
          raw_impact_points = 0

          for idx, rc in enumerate(resource_changes):
              if not isinstance(rc, dict):
                  print(f"[ERROR] Malformed resource change entry at index {idx}: expected dict, got {type(rc).__name__}.", file=sys.stderr)
                  sys.exit(1)

              addr = str(rc.get("address", f"unknown_resource_{idx}"))
              res_type = str(rc.get("type", "unknown"))

              change = rc.get("change")
              if not isinstance(change, dict):
                  print(f"[ERROR] Malformed resource change for '{addr}': missing or non-dict 'change' object.", file=sys.stderr)
                  sys.exit(1)

              actions = change.get("actions")
              if not isinstance(actions, list) or len(actions) == 0:
                  print(f"[ERROR] Malformed resource change for '{addr}': 'actions' must be a non-empty list.", file=sys.stderr)
                  sys.exit(1)

              for act in actions:
                  if not isinstance(act, str) or act not in VALID_ACTIONS:
                      print(f"[ERROR] Unsupported or invalid action '{act}' for '{addr}'.", file=sys.stderr)
                      sys.exit(1)

              actions_tuple = tuple(actions)
              if actions_tuple not in VALID_ACTION_SETS:
                  print(f"[ERROR] Unsupported action combination {actions} for '{addr}'.", file=sys.stderr)
                  sys.exit(1)

              is_st = is_stateful(res_type)
              is_mj = is_major_infra(res_type)
              classification = "UNKNOWN"
              item_impact = 0

              if actions_tuple in (("no-op",), ("read",)):
                  if actions_tuple == ("no-op",):
                      no_op_count += 1
                      classification = "NO-OP"
                  else:
                      read_count += 1
                      classification = "READ"
                  item_impact = 0
              elif "delete" in actions and "create" in actions:
                  replaced_count += 1
                  if is_st:
                      stateful_replacements += 1
                      classification = "STATEFUL REPLACEMENT"
                      item_impact = 45
                  elif is_mj:
                      major_infra_replacements += 1
                      classification = "MAJOR INFRA REPLACEMENT"
                      item_impact = 30
                  else:
                      classification = "REPLACEMENT"
                      item_impact = 20
              elif "delete" in actions:
                  deleted_count += 1
                  if is_st:
                      stateful_deletions += 1
                      classification = "STATEFUL DELETION"
                      item_impact = 50
                  elif is_mj:
                      major_infra_deletions += 1
                      classification = "MAJOR INFRA DELETION"
                      item_impact = 35
                  else:
                      classification = "DELETION"
                      item_impact = 15
              elif "update" in actions:
                  updated_count += 1
                  classification = "IN-PLACE UPDATE"
                  item_impact = 3
              elif "create" in actions:
                  created_count += 1
                  classification = "ADDITION"
                  item_impact = 1

              raw_impact_points += item_impact

              if classification not in ("NO-OP", "READ"):
                  affected_resources.append({
                      "address": addr,
                      "type": res_type,
                      "actions": list(actions),
                      "classification": classification,
                      "is_stateful": is_st,
                      "is_major_infra": is_mj,
                      "impact_points": item_impact
                  })

          total_destructive_changes = deleted_count + replaced_count
          total_changes = created_count + updated_count + deleted_count + replaced_count
          blast_radius_score = min(100, raw_impact_points)

          if stateful_deletions > 0 or stateful_replacements > 0 or total_destructive_changes >= 5 or blast_radius_score >= 70:
              blast_radius = "CRITICAL"
          elif major_infra_deletions > 0 or major_infra_replacements > 0 or total_destructive_changes >= 3 or blast_radius_score >= 40:
              blast_radius = "HIGH"
          elif updated_count > 0 or (created_count + updated_count) >= 3 or total_destructive_changes > 0 or blast_radius_score >= 15:
              blast_radius = "MEDIUM"
          elif created_count > 0 or blast_radius_score > 0:
              blast_radius = "LOW"
          else:
              blast_radius = "NONE"

          policy_violations = []
          if stateful_deletions > 0:
              policy_violations.append(f"Stateful resource deletion detected: {stateful_deletions} resource(s) scheduled for destruction.")
          if stateful_replacements > 0:
              policy_violations.append(f"Stateful resource replacement detected: {stateful_replacements} resource(s) scheduled for recreate cycle.")
          if deleted_count > max_allowed_deletions:
              policy_violations.append(f"Resource deletions ({deleted_count}) exceed max_allowed_deletions threshold ({max_allowed_deletions}).")
          if blast_radius_score >= rejection_threshold:
              policy_violations.append(f"Blast radius score ({blast_radius_score}) meets or exceeds rejection threshold ({rejection_threshold}).")

          if len(policy_violations) > 0:
              verdict = "REJECTED"
              risk_level = "CRITICAL"
              reason = "; ".join(policy_violations)
          elif blast_radius_score >= warning_threshold or blast_radius in ("HIGH", "MEDIUM"):
              if require_approval_on_warning:
                  verdict = "APPROVAL_REQUIRED"
                  risk_level = "HIGH" if blast_radius == "HIGH" else "MEDIUM"
                  reason = f"Blast radius score ({blast_radius_score}) exceeds warning threshold ({warning_threshold}) and requires explicit approval."
              else:
                  verdict = "WARNING"
                  risk_level = "HIGH" if blast_radius == "HIGH" else "MEDIUM"
                  reason = f"Blast radius score ({blast_radius_score}) exceeds warning threshold ({warning_threshold}) or contains operational modifications."
          elif created_count > 0:
              verdict = "APPROVED"
              risk_level = "LOW"
              reason = f"Plan contains {created_count} additive resource creation(s) within safe operational boundaries."
          else:
              verdict = "APPROVED"
              risk_level = "LOW"
              reason = "Plan proposes zero infrastructure mutations (clean or no-op)."

          summary = f"Verdict: {verdict} | Risk: {risk_level} | Blast Radius Score: {blast_radius_score}/100 ({blast_radius}) | {reason}"

          report_payload = {
              "timestamp": datetime.now(timezone.utc).isoformat(),
              "environment": environment,
              "verdict": verdict,
              "risk_level": risk_level,
              "blast_radius": {
                  "score": blast_radius_score,
                  "rating": blast_radius,
                  "raw_impact_points": raw_impact_points,
                  "warning_threshold": warning_threshold,
                  "rejection_threshold": rejection_threshold
              },
              "policy_violations": policy_violations,
              "measured_facts": {
                  "total_changes": total_changes,
                  "created_count": created_count,
                  "updated_count": updated_count,
                  "deleted_count": deleted_count,
                  "replaced_count": replaced_count,
                  "no_op_count": no_op_count,
                  "read_count": read_count,
                  "total_destructive_changes": total_destructive_changes,
                  "stateful_deletions": stateful_deletions,
                  "stateful_replacements": stateful_replacements,
                  "major_infra_deletions": major_infra_deletions,
                  "major_infra_replacements": major_infra_replacements
              },
              "affected_resources": affected_resources,
              "summary": summary
          }

          with open("risk_report.json", "w", encoding="utf-8") as f_json:
              json.dump(report_payload, f_json, indent=2)

          icon = "🔴" if verdict == "REJECTED" else ("🟠" if verdict == "APPROVAL_REQUIRED" else ("🟡" if verdict == "WARNING" else "🟢"))
          md = []
          md.append(f"# {icon} Terraform Plan Risk & Blast-Radius Assessment Report")
          md.append("")
          md.append(f"**Target Environment:** `{environment}` | **Evaluation Timestamp:** `{report_payload['timestamp']}`")
          md.append("")
          md.append("## Executive Summary")
          md.append("")
          md.append("| Metric | Evaluated Value | Assessment / Boundary |")
          md.append("| :--- | :--- | :--- |")
          md.append(f"| **Policy Gate Verdict** | `{verdict}` | {'⛔ Deployment BLOCKED' if verdict == 'REJECTED' else ('⚠️ Manual Approval Required' if verdict == 'APPROVAL_REQUIRED' else ('⚠️ Warning Flagged' if verdict == 'WARNING' else '✅ Plan Approved'))} |")
          md.append(f"| **Policy Risk Level** | `{risk_level}` | Impact Severity |")
          md.append(f"| **Blast Radius Score** | **`{blast_radius_score} / 100`** | Rating: `{blast_radius}` (Warning: `{warning_threshold}`, Rejection: `{rejection_threshold}`) |")
          md.append(f"| **Total Changes** | `{total_changes}` | Planned mutations |")
          md.append(f"| **Destructive Changes** | `{total_destructive_changes}` | Deletions: `{deleted_count}`, Replacements: `{replaced_count}` |")
          md.append(f"| **Stateful Touchpoints** | `{stateful_deletions + stateful_replacements}` | Stateful Del: `{stateful_deletions}`, Stateful Rep: `{stateful_replacements}` |")
          md.append("")

          if policy_violations:
              md.append("## ⚠️ Policy Violations Detected")
              md.append("")
              for v in policy_violations:
                  md.append(f"- ❌ **{v}**")
              md.append("")

          md.append("## Measured Facts vs. Heuristic Risk Assessment")
          md.append("")
          md.append("### Grounded Plan Metrics (Measured Facts)")
          md.append("")
          md.append("| Operation Type | Resource Count | Measured Risk Category |")
          md.append("| :--- | :--- | :--- |")
          md.append(f"| Additions (`create`) | `{created_count}` | Additive / Non-destructive |")
          md.append(f"| In-Place Updates (`update`) | `{updated_count}` | Operational Modification |")
          md.append(f"| Standard Deletions (`delete`) | `{deleted_count}` | Destructive Action |")
          md.append(f"| Replacements (`destroy and create`) | `{replaced_count}` | Recreate Cycle (Service Disruption Risk) |")
          md.append(f"| Stateful Resource Deletions | `{stateful_deletions}` | High Data Loss Risk |")
          md.append(f"| Stateful Resource Replacements | `{stateful_replacements}` | High Data Loss Risk |")
          md.append(f"| Major Infrastructure Deletions | `{major_infra_deletions}` | High Service Disruption Risk |")
          md.append(f"| Major Infrastructure Replacements | `{major_infra_replacements}` | High Service Disruption Risk |")
          md.append("")

          if affected_resources:
              md.append("## Affected Resources Inventory")
              md.append("")
              md.append("| Resource Address | Type | Action(s) | Classification | Impact Points |")
              md.append("| :--- | :--- | :--- | :--- | :--- |")
              for r in affected_resources:
                  act_str = ", ".join(r["actions"])
                  md.append(f"| `{r['address']}` | `{r['type']}` | `{act_str}` | `{r['classification']}` | `+{r['impact_points']}` |")
              md.append("")

          md.append("## Remediation & Operational Guidance")
          md.append("")
          if verdict == "REJECTED":
              md.append("1. **Do not force apply.** The plan violates organizational safety gates and contains destructive operations exceeding policy thresholds.")
              md.append("2. **Inspect stateful changes.** Ensure resources scheduled for deletion or replacement have explicit backups, lifecycle protection (`lifecycle { prevent_destroy = true }`), or migration plans.")
              md.append("3. **Reduce change scope.** Split large changes into smaller, purely additive or non-destructive pull requests.")
          elif verdict == "APPROVAL_REQUIRED":
              md.append("1. **Request peer review.** The calculated blast radius score warrants senior engineer or CAB review before proceeding.")
              md.append("2. **Verify target environment.** Confirm that proposed updates target the intended infrastructure tier.")
          else:
              md.append("1. **Plan approved for automated or manual execution.** Changes remain within verified policy thresholds.")
          md.append("")

          with open("risk_report.md", "w", encoding="utf-8") as f_md:
              f_md.write("\n".join(md))

          print(f"[INFO] Plan Risk Analysis Complete for '{environment}':")
          print(f"  Risk Level: {risk_level} | Verdict: {verdict} | Blast Radius Score: {blast_radius_score}/100 ({blast_radius})")
          print(f"  Created: {created_count} | Updated: {updated_count} | Deleted: {deleted_count} | Replaced: {replaced_count}")
          print(f"  Stateful Deletions: {stateful_deletions} | Stateful Replacements: {stateful_replacements}")

          output_payload = {
              "verdict": verdict,
              "risk_level": risk_level,
              "blast_radius_score": blast_radius_score,
              "blast_radius_rating": blast_radius,
              "total_destructive_changes": total_destructive_changes,
              "total_changes": total_changes,
              "created_count": created_count,
              "updated_count": updated_count,
              "deleted_count": deleted_count,
              "replaced_count": replaced_count,
              "stateful_deletions": stateful_deletions,
              "stateful_replacements": stateful_replacements,
              "violations_count": len(policy_violations),
              "summary": summary
          }

          print("::" + json.dumps({"outputs": output_payload}) + "::")
        outputFiles:
          - risk_report.json
          - risk_report.md

  - id: check_slack_alert
    type: io.kestra.plugin.core.flow.If
    description: Dispatch structured Slack risk scorecard when notify_slack is
      enabled and risk is elevated.
    condition: "{{ inputs.notify_slack and outputs.analyze_plan_risk.vars.risk_level
      in ['CRITICAL', 'HIGH'] }}"
    then:
      - id: send_slack_alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Send structured risk assessment scorecard to Slack.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": ":warning: *Terraform Plan Risk Policy Alert - {{ inputs.environment }}*\n• *Verdict:* `{{ outputs.analyze_plan_risk.vars.verdict }}`\n• *Risk Level:* `{{ outputs.analyze_plan_risk.vars.risk_level }}`\n• *Blast Radius Score:* `{{ outputs.analyze_plan_risk.vars.blast_radius_score }}/100` (`{{ outputs.analyze_plan_risk.vars.blast_radius_rating }}`)\n• *Destructive Changes:* `{{ outputs.analyze_plan_risk.vars.total_destructive_changes }}` (Deleted: {{ outputs.analyze_plan_risk.vars.deleted_count }}, Replaced: {{ outputs.analyze_plan_risk.vars.replaced_count }})\n• *Stateful Deletions:* `{{ outputs.analyze_plan_risk.vars.stateful_deletions }}`\n• *Summary:* {{ outputs.analyze_plan_risk.vars.summary }}\n• *Execution ID:* `{{ execution.id }}`"
          }

  - id: enforce_policy_gate
    type: io.kestra.plugin.core.flow.If
    description: Evaluate risk classification against policy gate thresholds,
      blocking rejected deployments.
    condition: "{{ outputs.analyze_plan_risk.vars.verdict in ['REJECTED',
      'APPROVAL_REQUIRED'] }}"
    then:
      - id: check_fail_gate
        type: io.kestra.plugin.core.flow.If
        description: Terminate pipeline execution with error when fail_on_rejection is
          enabled.
        condition: "{{ inputs.fail_on_rejection }}"
        then:
          - id: fail_gate
            type: io.kestra.plugin.core.execution.Fail
            description: Halt pipeline execution to block high-risk or policy-violating
              infrastructure deployment.
            errorMessage: >-
              Terraform Plan Risk Policy Gate FAILED for '{{ inputs.environment
              }}': Verdict {{ outputs.analyze_plan_risk.vars.verdict }} (Risk:
              {{ outputs.analyze_plan_risk.vars.risk_level }}, Blast Radius
              Score: {{ outputs.analyze_plan_risk.vars.blast_radius_score
              }}/100) violates safety thresholds. {{
              outputs.analyze_plan_risk.vars.summary }}
        else:
          - id: log_rejection_advisory
            type: io.kestra.plugin.core.log.Log
            description: Log high-risk advisory warning when fail_on_rejection is disabled.
            message: >-
              [WARN] Plan rejected by policy for '{{ inputs.environment }}'
              (Verdict: {{ outputs.analyze_plan_risk.vars.verdict }}, Score: {{
              outputs.analyze_plan_risk.vars.blast_radius_score }}/100, Risk: {{
              outputs.analyze_plan_risk.vars.risk_level }}), but
              fail_on_rejection is disabled. Continuing pipeline execution in
              advisory mode.

    else:
      - id: log_plan_approved
        type: io.kestra.plugin.core.log.Log
        description: Log plan approval when risk remains within acceptable thresholds.
        message: >-
          [SUCCESS] Terraform plan risk policy gate PASSED for '{{
          inputs.environment }}'. Verdict: {{
          outputs.analyze_plan_risk.vars.verdict }}, Risk: {{
          outputs.analyze_plan_risk.vars.risk_level }}, Blast Radius Score: {{
          outputs.analyze_plan_risk.vars.blast_radius_score }}/100 ({{
          outputs.analyze_plan_risk.vars.blast_radius_rating }}). Created: {{
          outputs.analyze_plan_risk.vars.created_count }}, Updated: {{
          outputs.analyze_plan_risk.vars.updated_count }}, Deleted: {{
          outputs.analyze_plan_risk.vars.deleted_count }}, Replaced: {{
          outputs.analyze_plan_risk.vars.replaced_count }}.

outputs:
  - id: verdict
    type: STRING
    description: Policy gate decision verdict (APPROVED, WARNING, APPROVAL_REQUIRED,
      REJECTED).
    value: "{{ outputs.analyze_plan_risk.vars.verdict }}"

  - id: risk_level
    type: STRING
    description: Policy severity classification level (CRITICAL, HIGH, MEDIUM, LOW).
    value: "{{ outputs.analyze_plan_risk.vars.risk_level }}"

  - id: blast_radius_score
    type: INT
    description: Grounded numerical blast radius impact score (0 to 100).
    value: "{{ outputs.analyze_plan_risk.vars.blast_radius_score }}"

  - id: blast_radius_rating
    type: STRING
    description: Estimated infrastructure impact scope category (CRITICAL, HIGH,
      MEDIUM, LOW, NONE).
    value: "{{ outputs.analyze_plan_risk.vars.blast_radius_rating }}"

  - id: created_count
    type: INT
    description: Count of resources to be newly created.
    value: "{{ outputs.analyze_plan_risk.vars.created_count }}"

  - id: updated_count
    type: INT
    description: Count of resources to be updated in-place.
    value: "{{ outputs.analyze_plan_risk.vars.updated_count }}"

  - id: deleted_count
    type: INT
    description: Count of resources to be destroyed.
    value: "{{ outputs.analyze_plan_risk.vars.deleted_count }}"

  - id: replaced_count
    type: INT
    description: Count of resources to be destroyed and recreated (replacement).
    value: "{{ outputs.analyze_plan_risk.vars.replaced_count }}"

  - id: stateful_deletions
    type: INT
    description: Count of stateful resources (databases, storage, persistent disks)
      scheduled for deletion.
    value: "{{ outputs.analyze_plan_risk.vars.stateful_deletions }}"

  - id: stateful_replacements
    type: INT
    description: Count of stateful resources scheduled for replacement.
    value: "{{ outputs.analyze_plan_risk.vars.stateful_replacements }}"

  - id: total_destructive_changes
    type: INT
    description: Total count of destructive operations (deletions plus replacements).
    value: "{{ outputs.analyze_plan_risk.vars.total_destructive_changes }}"

  - id: total_changes
    type: INT
    description: Total count of proposed infrastructure mutations.
    value: "{{ outputs.analyze_plan_risk.vars.total_changes }}"

  - id: violations_count
    type: INT
    description: Total count of detected policy violations.
    value: "{{ outputs.analyze_plan_risk.vars.violations_count }}"

  - id: summary
    type: STRING
    description: Sanitized audit summary of the plan risk assessment.
    value: "{{ outputs.analyze_plan_risk.vars.summary }}"

  - id: risk_report_json
    type: FILE
    description: Machine-readable comprehensive risk evaluation report artifact in
      JSON format.
    value: "{{ outputs.analyze_plan_risk.outputFiles['risk_report.json'] }}"

  - id: risk_report_md
    type: FILE
    description: Human-readable executive summary scorecard and affected resources
      inventory in Markdown format.
    value: "{{ outputs.analyze_plan_risk.outputFiles['risk_report.md'] }}"

Accidental destruction of stateful infrastructure (such as production databases, persistent volumes, or object storage) is one of the most catastrophic incidents that can occur in cloud operations. While CI/CD pipelines commonly automate terraform plan, raw plan text often spans thousands of lines, burying destructive actions (["delete"] or ["delete", "create"]) beneath harmless property additions or tag changes.

Unlike manual approval pauses that halt workflows for human review or Infracost gates that only calculate dollar expenditure, this blueprint automatically evaluates the actual Terraform execution plan (terraform show -json). It deterministically inspects planned resource operations, flags stateful infrastructure destruction, computes a grounded blast-radius score (0-100), delivers actionable Slack notifications, generates auditable Markdown and JSON report artifacts, and enforces an automated deployment gate before changes reach production.

Architecture

Webhook Trigger / Direct Run
       ↓
WorkingDirectory (Terraform Workspace)
       ↓
TerraformCLI (Evaluate Uploaded Plan OR init & plan & show -json)
       ↓
Python Risk Analyzer (Blast-Radius Scoring & Stateful Classification)
       ↓
If [Elevated Risk & notify_slack]: Slack Webhook Notification
       ↓
If [REJECTED / APPROVAL_REQUIRED]
    ├── If [fail_on_rejection: true]: Fail Task (Halt Deployment)
    └── If [fail_on_rejection: false]: Log Warning (Advisory Mode)
       ↓
Outputs & Artifacts (Score, Verdict, JSON Report, Markdown Scorecard)

Grounded Facts vs. Heuristic Risk Assessment

The risk analyzer separates verifiable plan measurements from heuristic scoring:

Measured Grounded Facts: Exact counts extracted directly from HashiCorp Terraform 1.0+ JSON specification (resource_changes array). Quantifies created, updated, deleted, and replaced resources, as well as exact keyword and provider matches for stateful and major infrastructure resources.
Heuristic Blast-Radius Score (0-100): A transparent mathematical calculation reflecting total physical operational footprint across the target cloud estate. Stateful destructions (+50 pts) and replacements (+45 pts) carry heavy impact, followed by major infrastructure touches (+35/+30 pts), standard deletions (+15 pts), in-place updates (+3 pts), and additions (+1 pt).
Policy Gate Verdict:
- REJECTED: Stateful destruction, deletions exceeding max_allowed_deletions, or blast radius score $\ge$ blast_radius_rejection_threshold.
- APPROVAL_REQUIRED: Elevated blast radius score exceeding blast_radius_warning_threshold when require_approval_on_warning is enabled.
- WARNING: Score exceeds warning threshold but remains within advisory limits.
- APPROVED: Clean additive changes or no-op plans within safe boundaries.

Deployment Gate & Advisory Modes

The flow supports two operational modes controlled by fail_on_rejection:

Advisory & Audit Mode (fail_on_rejection: false, default): When policy violations or high blast radius are detected, the flow logs a high-severity warning and dispatches Slack notifications without halting execution. This safe default allows non-blocking ad-hoc evaluation, demos, and visibility without prematurely failing unconfigured pipelines.
Enforcing Policy Gate Mode (fail_on_rejection: true): Any plan evaluated as REJECTED or APPROVAL_REQUIRED triggers the fail_gate task (io.kestra.plugin.core.execution.Fail), immediately terminating execution with a non-zero exit code to halt automated CI/CD deployment pipelines.

Ingestion Flexibility: Uploaded Artifact vs. HCL Planning

The blueprint supports two ingestion pathways:

Direct JSON Plan Artifact (plan_file): Upload or pass an already-generated Terraform plan in JSON format (terraform show -json tfplan > plan.json). The blueprint evaluates the artifact directly without requiring cloud credentials or Terraform initialization.
Inline HCL Configuration (terraform_code): Provide Terraform configuration code. The workspace initializes Terraform with the pinned hashicorp/terraform:1.9.8 Docker container and generates the plan automatically.

Actions and Replacements

Terraform plans express actions as arrays:

["create"]: New resource addition.
["update"]: In-place modification.
["delete"]: Resource destruction.
["delete", "create"] or ["create", "delete"]: Resource replacement (destroy then recreate). Replacements of stateful resources are categorized as CRITICAL risk because data loss can occur during recreate cycles.

Security & Sensitive Data Handling

Terraform execution plans can contain sensitive state attributes, environment variables, or secrets. To adhere to security best practices:

Raw plan.json files are never dumped into execution logs or Slack notifications.
The Python analyzer extracts only safe metadata (type, address, actions, classification).
Slack notifications and Markdown scorecards contain only sanitized risk metrics and aggregated summary tables.

CI/CD Pipeline Integration

Trigger this blueprint from GitHub Actions, GitLab CI, or Jenkins upon pull request creation:

curl -X POST "https://<your-kestra-host>/api/v1/executions/webhook/company.team/terraform-plan-risk-policy-gate/<TERRAFORM_PLAN_WEBHOOK_KEY>" \
  -H "Content-Type: application/json" \
  -d '{
    "inputs": {
      "environment": "production",
      "max_allowed_deletions": 0,
      "blast_radius_rejection_threshold": 60,
      "fail_on_rejection": true,
      "notify_slack": true
    }
  }'

Prerequisites & Secrets

Container Engine: Docker or container runner accessible to Kestra when generating plans from terraform_code using hashicorp/terraform:1.9.8.
TERRAFORM_PLAN_WEBHOOK_KEY (required for Webhook trigger): Secret key securing the on_iac_plan_request Webhook trigger.
SLACK_WEBHOOK_URL (optional): Required only when notify_slack: true is enabled.
Cloud Provider Credentials (optional): When evaluating cloud infrastructure configurations via terraform_code, provide provider environment variables (such as AWS_ACCESS_KEY_ID, GOOGLE_APPLICATION_CREDENTIALS).