OPA-Gated IaC Applies with Human Approval

id: iac-opa-policy-gate-human-approval
namespace: company.team

description: |
  Gate IaC applies through OPA policy with three outcomes: hard violations stop
  the run immediately, soft violations route to a group-attributed human
  approval, and clean manifests proceed straight to kubectl apply - so policy
  decides the fast paths and people only review the borderline cases.

triggers:
  - id: ci_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Called by CI with the plan or manifest set to gate before apply.

  - id: nightly_recheck
    type: io.kestra.plugin.core.trigger.Schedule
    description: Periodic re-evaluation of the standing manifest set (enable as needed).
    cron: "0 4 * * *"
    disabled: true

inputs:
  - id: requester
    type: STRING
    displayName: Requester
    description: Who or which CI job is asking for this apply - recorded in the OPA input.
    defaults: platform-engineer

  - id: manifests
    type: JSON
    displayName: Manifests to gate
    description: Deployment-shaped resources evaluated by policy and applied after passing.
    defaults: |
      [
        {"kind": "Deployment", "name": "web", "namespace": "staging", "image": "registry.example.com/web:1.4.2", "privileged": false, "replicas": 2},
        {"kind": "Deployment", "name": "metrics-exporter", "namespace": "staging", "image": "docker.io/library/busybox:1.36", "privileged": false, "replicas": 1}
      ]

  - id: apply_namespace
    type: STRING
    displayName: Fallback namespace
    description: Namespace used when a manifest does not carry its own.
    defaults: staging

  - id: opa_policy_path
    type: STRING
    displayName: OPA policy path
    description: Policy document path served by the OPA instance.
    defaults: iac/gate

  - id: approval_timeout
    type: STRING
    displayName: Approval timeout
    description: ISO-8601 duration before the human approval expires and the run fails.
    defaults: P2D

tasks:
  - id: build_policy_input
    type: io.kestra.plugin.scripts.python.Script
    description: Normalize the manifest set into the resource facts OPA evaluates.
    env:
      MANIFESTS: "{{ inputs.manifests | toJson }}"
      APPLY_NAMESPACE: "{{ inputs.apply_namespace }}"
      REQUESTER: "{{ inputs.requester }}"
      EXECUTION_ID: "{{ execution.id }}"
    script: |
      import json
      import os

      try:
          raw = json.loads(os.environ.get("MANIFESTS") or "[]")
      except Exception:
          raw = []
      if not isinstance(raw, list):
          raw = [raw]

      fallback_ns = os.environ.get("APPLY_NAMESPACE") or "default"
      resources = []
      for item in raw:
          if not isinstance(item, dict):
              continue
          resources.append({
              "kind": str(item.get("kind") or "Deployment"),
              "name": str(item.get("name") or "unnamed"),
              "namespace": str(item.get("namespace") or fallback_ns),
              "image": str(item.get("image") or ""),
              "privileged": bool(item.get("privileged")),
              "replicas": int(item.get("replicas") or 1),
          })

      policy_input = {
          "requester": os.environ.get("REQUESTER") or "unknown",
          "execution_id": os.environ.get("EXECUTION_ID") or "",
          "resources": resources,
      }

      print("gating {} resources for {}".format(len(resources), policy_input["requester"]))
      print(
          "::"
          + json.dumps(
              {
                  "outputs": {
                      "resources": resources,
                      "policy_input": policy_input,
                      "manifest_count": len(resources),
                  }
              }
          )
          + "::"
      )

  - id: evaluate
    type: io.kestra.plugin.ee.opa.policy.Evaluate
    description: Evaluate the normalized resources against the OPA gate policy.
    url: "{{ secret('OPA_URL') }}"
    token: "{{ secret('OPA_TOKEN') }}"
    policyPath: "{{ inputs.opa_policy_path }}"
    input:
      requester: "{{ outputs.build_policy_input.policy_input.requester }}"
      execution_id: "{{ outputs.build_policy_input.policy_input.execution_id }}"
      resources: "{{ outputs.build_policy_input.policy_input.resources }}"

  - id: classify_decision
    type: io.kestra.plugin.scripts.python.Script
    description: Split OPA violations into hard denials that stop the run and soft
      items that need a human.
    env:
      VIOLATIONS: "{{ outputs.evaluate.result | toJson }}"
      DEFINED: "{{ outputs.evaluate.defined }}"
      DECISION_ID: "{{ outputs.evaluate.decisionId | default('') }}"
    script: |
      import json
      import os

      try:
          violations = json.loads(os.environ.get("VIOLATIONS") or "[]")
      except Exception:
          violations = []
      if violations is None:
          violations = []
      if not isinstance(violations, list):
          violations = [violations]

      defined = str(os.environ.get("DEFINED") or "false").lower() == "true"
      decision_id = os.environ.get("DECISION_ID") or ""

      auto_deny = []
      review = []
      for entry in violations:
          if isinstance(entry, dict):
              severity = str(entry.get("severity") or "hard").lower()
              rule = str(entry.get("rule") or entry.get("msg") or entry.get("message") or "unnamed rule")
              resource = str(entry.get("resource") or "")
              item = {"rule": rule, "resource": resource, "severity": severity}
          else:
              item = {"rule": str(entry), "resource": "", "severity": "hard"}
          if severity in ("soft", "warning", "review"):
              review.append(item)
          else:
              auto_deny.append(item)

      if not defined:
          auto_deny.append({
              "rule": "policy evaluation unavailable or undefined",
              "resource": "",
              "severity": "hard",
          })

      print(
          "decision {}: {} hard, {} need review".format(
              decision_id or "no-decision", len(auto_deny), len(review)
          )
      )
      print(
          "::"
          + json.dumps(
              {
                  "outputs": {
                      "auto_deny": auto_deny,
                      "review": review,
                      "auto_deny_count": len(auto_deny),
                      "review_count": len(review),
                      "decision_id": decision_id,
                  }
              }
          )
          + "::"
      )

  - id: hard_deny_gate
    type: io.kestra.plugin.core.flow.If
    description: Hard violations stop the run before any human sees it - policy
      forbids the change outright.
    condition: "{{ outputs.classify_decision.auto_deny_count > 0 }}"
    then:
      - id: notify_hard_deny
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Post the blocking violations with the OPA decision id.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": ":no_entry: *IaC apply blocked by OPA* (decision {{ outputs.classify_decision.decision_id }})\n{% for v in outputs.classify_decision.auto_deny %}• [{{ v.severity }}] {{ v.rule }}{{ v.resource ? (' - ' ~ v.resource) : '' }}\n{% endfor %}Requested by {{ inputs.requester }}. Fix the plan and re-run; nothing was applied."
          }

      - id: stop_on_hard_deny
        type: io.kestra.plugin.core.execution.Fail
        description: Fail the execution so CI stops before apply.
        errorMessage: "OPA hard-deny: {{ outputs.classify_decision.auto_deny_count }}
          violation(s) - see execution logs and Slack."

  - id: review_gate
    type: io.kestra.plugin.core.flow.If
    description: Soft violations route to a group-attributed human approval; clean
      sets skip straight to apply.
    condition: "{{ outputs.classify_decision.review_count > 0 }}"
    then:
      - id: notify_review
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Tell reviewers what is waiting and why.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": ":eyes: *IaC apply needs review* (decision {{ outputs.classify_decision.decision_id }})\n{% for v in outputs.classify_decision.review %}• [{{ v.severity }}] {{ v.rule }}{{ v.resource ? (' - ' ~ v.resource) : '' }}\n{% endfor %}Requested by {{ inputs.requester }}. Approve in Kestra to continue to apply."
          }

      - id: human_approval
        type: io.kestra.plugin.ee.flow.HumanTask
        description: Route the borderline set to the platform approvers - only that
          group can resume it.
        assignment:
          groups:
            - platform-approvers
        pauseDuration: "{{ inputs.approval_timeout }}"
        behavior: FAIL
        onResume:
          - id: decision
            type: SELECT
            values:
              - approve
              - deny
            defaults: deny
            description: Allow or refuse the apply.
          - id: reviewer_note
            type: STRING
            defaults: ""
            description: Rationale stored with the execution for auditors.

      - id: approval_gate
        type: io.kestra.plugin.core.flow.If
        description: Continue to apply only on an explicit approval from the group.
        condition: "{{ (outputs.human_approval.onResume is defined) and
          outputs.human_approval.onResume.decision == 'approve' }}"
        then:
          - id: log_approved
            type: io.kestra.plugin.core.log.Log
            description: Record the approval so the apply is attributable.
            message: "Approved by platform-approvers: {{
              outputs.human_approval.onResume.reviewer_note ?? 'no note' }}"
        else:
          - id: notify_denied
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Record the denial with the reviewer note.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": ":x: *IaC apply denied by platform-approvers.* Note: {{ outputs.human_approval.onResume.reviewer_note ?? 'none given' }}. Nothing was applied."
              }

          - id: stop_on_denial
            type: io.kestra.plugin.core.execution.Fail
            description: Fail the execution so CI stops before apply.
            errorMessage: "Apply denied by human review - see reviewer note in execution
              logs."

    else:
      - id: log_auto_pass
        type: io.kestra.plugin.core.log.Log
        description: No soft violations - policy cleared the set without a human.
        message: "OPA cleared {{ outputs.build_policy_input.manifest_count }}
          manifest(s) with no review items (decision {{
          outputs.classify_decision.decision_id }})."

  - id: apply_manifests
    type: io.kestra.plugin.core.flow.Loop
    description: Apply each gated manifest - reached only when policy or a human
      cleared the set.
    values: "{{ outputs.build_policy_input.resources }}"
    tasks:
      - id: apply_item
        type: io.kestra.plugin.kubernetes.kubectl.Apply
        description: Create or update the Deployment with its gated spec.
        connection:
          masterUrl: "{{ secret('K8S_MASTER_URL') }}"
          oauthToken: "{{ secret('K8S_TOKEN') }}"
          trustCerts: true
        namespace: "{{ item.value.namespace }}"
        spec: |
          apiVersion: apps/v1
          kind: Deployment
          metadata:
            name: {{ item.value.name }}
            namespace: {{ item.value.namespace }}
          spec:
            replicas: {{ item.value.replicas }}
            selector:
              matchLabels:
                app: {{ item.value.name }}
            template:
              metadata:
                labels:
                  app: {{ item.value.name }}
            spec:
              containers:
                - name: {{ item.value.name }}
                  image: {{ item.value.image }}
                  securityContext:
                    privileged: {{ item.value.privileged }}

  - id: announce_applied
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: "Confirm the apply with the path taken - straight through or
      human-approved."
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": ":white_check_mark: *IaC apply complete* - {{ outputs.build_policy_input.manifest_count }} manifest(s) applied for {{ inputs.requester }}. Review path: {{ outputs.human_approval is defined ? ('human approval (' ~ (outputs.human_approval.onResume.reviewer_note ?? 'no note') ~ ')') : 'auto (no soft violations)' }}. OPA decision {{ outputs.classify_decision.decision_id }}."
      }

outputs:
  - id: manifest_count
    type: INT
    description: Number of resources gated and applied.
    value: "{{ outputs.build_policy_input.manifest_count }}"

  - id: resources
    type: JSON
    description: Normalized resource facts that OPA evaluated.
    value: "{{ outputs.build_policy_input.resources }}"

  - id: opa_decision_id
    type: STRING
    description: OPA decision id for auditing which policy version answered.
    value: "{{ outputs.classify_decision.decision_id }}"

  - id: review_items
    type: JSON
    description: Soft violations that were routed to human approval.
    value: "{{ outputs.classify_decision.review }}"

  - id: human_decision
    type: STRING
    description: Reviewer verdict when a human approval was required.
    value: "{{ outputs.human_approval is defined ?
      outputs.human_approval.onResume.decision : 'not-required' }}"

errors:
  - id: alert_on_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert when the gate itself errors so a broken OPA or cluster
      connection is never mistaken for a denial.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "iac-opa-policy-gate-human-approval ERRORED in flow {{ flow.id }} (execution {{ execution.id }}) - check the execution logs."
      }