OPA conftest Policy Tests for Rendered Kubernetes Manifests

id: conftest-opa-manifest-tests
namespace: company.team
description: |
  Stop non-compliant Kubernetes manifests at review time by running OPA conftest
  policy tests against the rendered YAML the moment CI reports a manifest change.
  A keyed webhook starts the run, a shared working directory stages the two
  manifests and the Rego policy, conftest runs in its official container, and a
  Python step turns the JSON report into a per-failure verdict. Any failure opens
  a labeled GitHub issue and alerts Slack, while a clean run publishes pass
  outputs the merge gate can read.

triggers:
  - id: on_manifest_change
    type: io.kestra.plugin.core.trigger.Webhook
    description: CI POSTs here when rendered manifests change in a pull request, so
      the policy gate runs before merge instead of after.
    key: "{{ secret('MANIFEST_TEST_WEBHOOK_KEY') }}"
  - id: nightly_safety_net
    type: io.kestra.plugin.core.trigger.Schedule
    description: Disabled nightly net that would re-test the manifests if the policy
      drifts without a push; switch it on deliberately.
    cron: "0 6 * * *"
    disabled: true

inputs:
  - id: repository
    type: STRING
    defaults: your-org/your-repo
    description: Repository in owner/repo form that receives the GitHub issue when
      conftest reports a violation.
  - id: branch
    type: STRING
    defaults: main
    description: Branch the rendered manifests came from, quoted in the issue body
      and the Slack alert.
  - id: image
    type: STRING
    defaults: ghcr.io/acme/payments-api:1.14.2
    description: Container image rendered into the Deployment fixture, so the staged
      manifests look like a real CI artifact.

tasks:
  # --- 1. stage the manifests, run the policy tests, parse the report ---
  - id: policy_workspace
    type: io.kestra.plugin.core.flow.WorkingDirectory
    description: One shared directory so the staged fixtures, the conftest container
      and the report parser all read the same files.
    tasks:
      - id: write_fixtures
        type: io.kestra.plugin.scripts.shell.Commands
        description: Write the rendered Deployment, the Service and the Rego policy with
          heredocs, standing in for the files CI just produced.
        commands:
          - |
            mkdir -p policy
            cat > deployment.yaml <<'EOF'
            apiVersion: apps/v1
            kind: Deployment
            metadata:
              name: payments-api
              namespace: payments
              labels:
                app.kubernetes.io/name: payments-api
                app.kubernetes.io/version: 1.14.2
            spec:
              replicas: 3
              selector:
                matchLabels:
                  app.kubernetes.io/name: payments-api
              template:
                metadata:
                  labels:
                    app.kubernetes.io/name: payments-api
                spec:
                  securityContext:
                    runAsNonRoot: true
                    runAsUser: 10001
                  containers:
                    - name: api
                      image: {{ inputs.image }}
                      imagePullPolicy: IfNotPresent
                      ports:
                        - name: http
                          containerPort: 8080
                          protocol: TCP
                      readinessProbe:
                        httpGet:
                          path: /healthz
                          port: http
                        initialDelaySeconds: 5
                        periodSeconds: 10
            EOF
            cat > service.yaml <<'EOF'
            apiVersion: v1
            kind: Service
            metadata:
              name: payments-api
              namespace: payments
            spec:
              type: ClusterIP
              selector:
                app.kubernetes.io/name: payments-api
              ports:
                - name: http
                  port: 80
                  targetPort: http
                  protocol: TCP
            EOF
            cat > policy/policy.rego <<'EOF'
            package main

            import rego.v1

            # Containers must declare CPU and memory limits before they reach a cluster,
            # and nothing in a workload pod is allowed to run privileged.
            pod_containers contains container if {
              container := input.spec.template.spec.containers[_]
            }
            pod_containers contains container if {
              container := input.spec.template.spec.initContainers[_]
            }
            deny_resource_limits contains violation if {
              container := pod_containers[_]
              not container.resources.limits.cpu
              violation := {"msg": sprintf("container %s of %s must declare resources.limits.cpu", [container.name, input.metadata.name]), "rule": "resource-limits-required"}
            }
            deny_resource_limits contains violation if {
              container := pod_containers[_]
              not container.resources.limits.memory
              violation := {"msg": sprintf("container %s of %s must declare resources.limits.memory", [container.name, input.metadata.name]), "rule": "resource-limits-required"}
            }
            deny_privileged_container contains violation if {
              container := pod_containers[_]
              container.securityContext.privileged == true
              violation := {"msg": sprintf("container %s of %s must not run privileged", [container.name, input.metadata.name]), "rule": "privileged-container-forbidden"}
            }
            EOF
            ls -la policy deployment.yaml service.yaml

      - id: run_conftest
        type: io.kestra.plugin.scripts.shell.Commands
        description: Run conftest over every staged manifest in the official container;
          || true absorbs the violation exit code so the parsed report, not the
          shell, drives the gate.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: openpolicyagent/conftest:v0.56.0
        commands:
          - conftest test *.yaml --output json > conftest-report.json || true

      - id: parse_report
        type: io.kestra.plugin.scripts.python.Script
        description: Read the conftest JSON, build the failure list and pass count,
          write one line per failure, and emit the verdict through the outputs
          protocol.
        taskRunner:
          type: io.kestra.plugin.core.runner.Process
        outputFiles:
          - policy-failures.txt
        script: |
          import json
          import os

          REPORT = "conftest-report.json"
          LINES_FILE = "policy-failures.txt"

          def load_report(path):
              if not os.path.exists(path) or os.path.getsize(path) == 0:
                  return None
              try:
                  with open(path, encoding="utf-8") as handle:
                      return json.load(handle)
              except (ValueError, OSError):
                  return None

          def collect(entries):
              found = []
              passed = 0
              for entry in entries:
                  if not isinstance(entry, dict):
                      continue
                  fname = str(entry.get("filename") or "unknown-file")
                  try:
                      passed += int(entry.get("successes") or 0)
                  except (TypeError, ValueError):
                      pass
                  items = entry.get("failures")
                  if not isinstance(items, list):
                      queries = entry.get("queries")
                      if isinstance(queries, list):
                          items = queries
                  if isinstance(items, list):
                      for item in items:
                          if not isinstance(item, dict):
                              continue
                          msg = str(item.get("msg") or "").strip()
                          if not msg:
                              continue
                          meta = item.get("metadata") if isinstance(item.get("metadata"), dict) else {}
                          rule = str(meta.get("rule") or "").strip()
                          if not rule:
                              rule = str(meta.get("query") or "").rsplit(".", 1)[-1]
                          rule = rule or str(item.get("code") or "").strip() or "unnamed-rule"
                          found.append({"file": fname, "rule": rule, "msg": msg})
                      continue
                  try:
                      declared = int(items or 0)
                  except (TypeError, ValueError):
                      declared = 0
                  if declared > 0:
                      found.append({"file": fname, "rule": "undeclared-rule",
                                    "msg": "conftest reported %d failure(s) without detail" % declared})
              return found, passed

          report = load_report(REPORT)
          if isinstance(report, list) and report:
              failures, pass_count = collect(report)
          elif isinstance(report, list):
              failures = [{"file": REPORT, "rule": "no-results",
                           "msg": "conftest returned no per-file results, nothing was actually tested"}]
              pass_count = 0
          else:
              failures = [{"file": REPORT, "rule": "report-unreadable",
                           "msg": "conftest produced no parseable JSON, the policy run did not complete"}]
              pass_count = 0

          verdict = "fail" if failures else "pass"
          summary = "; ".join(
              "%s | %s | %s" % (item["file"], item["rule"], item["msg"]) for item in failures[:5]
          )
          summary = (summary.replace('"', "'").replace("\n", " ")[:500]) or "no policy failures"

          with open(LINES_FILE, "w", encoding="utf-8") as handle:
              if failures:
                  for item in failures:
                      handle.write("%s | %s | %s\n" % (item["file"], item["rule"], item["msg"]))
              else:
                  handle.write("no conftest failures\n")

          print("conftest verdict=%s failures=%d passes=%d" % (verdict, len(failures), pass_count))
          print("::" + json.dumps({
              "outputs": {
                  "verdict": verdict,
                  "failure_count": len(failures),
                  "failures": failures,
                  "pass_count": int(pass_count),
                  "summary": summary,
              }
          }, ensure_ascii=True) + "::")

  # --- 2. branch on the verdict ---
  - id: policy_gate
    type: io.kestra.plugin.core.flow.If
    description: One branch for the parsed report. Any failure escalates into the
      backlog and chat; a clean run only records itself.
    condition: "{{ outputs.parse_report.vars.failure_count > 0 }}"
    then:
      - id: open_policy_issue
        type: io.kestra.plugin.github.issues.Create
        description: Open a labeled issue carrying every failing file, rule and message
          so the violation is worked in the same backlog as the code.
        jwtToken: "{{ secret('GITHUB_TOKEN') }}"
        repository: "{{ inputs.repository }}"
        title: "conftest policy failures in {{ inputs.repository }} on {{ inputs.branch
          }}"
        body: |
          ## OPA conftest failures

          {{ outputs.parse_report.vars.failure_count }} rule(s) failed on the rendered Kubernetes manifests for branch `{{ inputs.branch }}`.

          ```json
          {{ outputs.parse_report.vars.failures | toJson }}
          ```

          Verdict `{{ outputs.parse_report.vars.verdict }}` with {{ outputs.parse_report.vars.pass_count }} passing check(s). Per-failure report {{ outputs.parse_report.outputFiles['policy-failures.txt'] }}

          Merge is blocked until the manifests satisfy the policy.
        labels:
          - kubernetes
          - policy-gate
      - id: alert_slack
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Ping the platform channel with the failure count and the first
          failing rules, plus the issue that was just opened.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "conftest blocked {{ inputs.repository }}@{{ inputs.branch }} with {{ outputs.parse_report.vars.failure_count }} policy failure(s): {{ outputs.parse_report.vars.summary }}. A GitHub issue is open for the pull request. Execution {{ execution.id }}."
          }
    else:
      - id: log_clean
        type: io.kestra.plugin.core.log.Log
        description: Record the all-clear so execution history doubles as proof the gate
          ran on this change.
        message: "conftest passed {{ inputs.repository }}@{{ inputs.branch }} with {{
          outputs.parse_report.vars.pass_count }} passing check(s)."

errors:
  - id: notify_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Escalate any failure to the ops channel with the execution url,
      because a broken policy run must never read as a clean gate.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text": "FAILED: {{ flow.id }} {{ execution.id }}"}

outputs:
  - id: verdict
    type: STRING
    description: Overall conftest outcome for this change, either pass or fail.
    value: "{{ outputs.parse_report.vars.verdict }}"
  - id: failure_count
    type: INT
    description: Number of policy violations conftest reported across the rendered
      manifests.
    value: "{{ outputs.parse_report.vars.failure_count }}"
  - id: failures
    type: ARRAY
    description: 'One entry per violation with file, rule and message, e.g.
      [{"file": "deployment.yaml", "rule": "deny_resource_limits", "msg":
      "..."}].'
    value: "{{ outputs.parse_report.vars.failures | toJson }}"
  - id: report_uri
    type: STRING
    description: Internal storage URI of the per-failure line report kept for this run.
    value: "{{ outputs.parse_report.outputFiles['policy-failures.txt'] }}"

Diagram unavailable

We could not build the topology for this blueprint. The flow itself is valid, use the YAML on the left to run it.

Most Kubernetes policy checks are written once, wired into nobody's pipeline, and quietly forgotten. This blueprint puts conftest in front of the merge: CI posts to a keyed webhook whenever rendered manifests change, the flow stages those manifests next to a Rego policy, runs conftest in its own container, and turns the JSON report into a verdict that humans and machines can both act on. Failures become a labeled GitHub issue plus a Slack ping, so the violation lands where the team already works.

How it works

The on_manifest_change trigger (io.kestra.plugin.core.trigger.Webhook) exposes a keyed URL for your CI to POST to when rendered manifests change. A disabled nightly_safety_net Schedule sits behind it as a fallback net.
policy_workspace (io.kestra.plugin.core.flow.WorkingDirectory) opens one shared filesystem so every later task sees the same files.
write_fixtures (io.kestra.plugin.scripts.shell.Commands) writes deployment.yaml (a three replica Deployment deliberately rendered with no resources.limits), service.yaml and policy/policy.rego with heredocs. The Rego defines deny_resource_limits for containers missing CPU or memory limits and deny_privileged_container for securityContext.privileged == true, covering both containers and initContainers. Every denial returns a structured violation that carries its own rule id, so the report can name the rule that fired instead of only the text it produced.
run_conftest (io.kestra.plugin.scripts.shell.Commands on the io.kestra.plugin.scripts.runner.docker.Docker runner, image openpolicyagent/conftest:v0.56.0) executes conftest test *.yaml --output json > conftest-report.json || true. The || true absorbs the violation exit code so the parsed report, not the shell, decides the outcome.
parse_report (io.kestra.plugin.scripts.python.Script) reads the report, walks the failures list of each file entry, takes the rule name from the violation's metadata.rule (falling back to metadata.query, then code), and writes policy-failures.txt with one line per violation. It emits verdict, failure_count, failures, pass_count and summary through the outputs protocol. A missing, empty or unparseable report becomes a synthetic failure, so a broken tool run can never masquerade as a pass.
policy_gate (io.kestra.plugin.core.flow.If) branches on failure_count. Over zero, open_policy_issue (io.kestra.plugin.github.issues.Create) files a labeled issue with the full failure list, then alert_slack posts the count and the first failing rules. At zero, log_clean records the all-clear.
The errors block alerts Slack when the pipeline itself breaks, because a dead policy runner looks exactly like a passing gate.

What you get

Policy failures caught before merge instead of during the next incident review.
A per-failure line report kept as a flow output (report_uri) for every execution.
Structured outputs, verdict, failure_count, failures and report_uri, ready for a status check.
Escalation where engineers already work, a labeled GitHub issue plus a Slack alert.

Who it's for

Platform and DevOps teams shipping policy-as-code with OPA but with no merge gate around it.
Security engineers who want resource limits and privilege rules enforced on rendered manifests instead of reviewed by hand.
SRE teams that need a repeatable, auditable answer to whether a change passed its policy checks.

Why orchestrate this with Kestra

conftest evaluates policy well, but it does not trigger itself, keep its reports, or route its findings. Kestra supplies the loop around it, an event entry point for CI, a containerized runner so the tool version lives in the flow, a branch that separates escalation from the all-clear, output files that survive the run, and alert routing into GitHub and Slack from one declarative file. A bare conftest line in a pipeline script gives you none of that: it prints to a log nobody reads and disappears when the job ends.

Prerequisites

A Kestra worker with Docker access for the conftest container and Python 3 for the report parser.
A GitHub token allowed to open issues in repository (classic repo scope, or fine-grained Issues: write) and a Slack incoming webhook.

Secrets

MANIFEST_TEST_WEBHOOK_KEY: key segment guarding the webhook URL, so only your CI can start a run.
GITHUB_TOKEN: token used by open_policy_issue to create the issue.
SLACK_WEBHOOK_URL: Slack incoming webhook used for the violation alert and the flow error handler.

Quick start

Add the three secrets to your Kestra namespace, then set repository, branch and image for the change you want gated.
Point your CI at /api/v1/{tenant}/executions/webhook/{namespace}/conftest-opa-manifest-tests/{MANIFEST_TEST_WEBHOOK_KEY} when manifests change, or trigger the webhook by hand.
Confirm the first run fails on the staged Deployment, read report_uri, and check that the issue plus the Slack alert landed. Then add resources.limits to the fixture and watch it go green.

How to extend

Replace the heredocs in write_fixtures with io.kestra.plugin.git.Clone so the real rendered manifests from the pull request are tested instead of fixtures.
Add a core.execution.Fail in the failing branch to turn the gate red for branch protection, and extend policy.rego with image tag, pod security or required label rules.

Test rendered Kubernetes manifests with conftest before they merge

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Test rendered Kubernetes manifests with conftest before they merge

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

ConfigJanitor - Kubernetes unused ConfigMap and Secret cleaner

Scheduled Container Registry Vulnerability Audit with Trivy and Slack Alerts

Keep CODEOWNERS-Owned Pull Requests From Merging Without Review

Lint Commit Messages Against Conventional Commits Before Every Release

AWS CodeDeploy Failed Deployment Sentinel

Contour Envoy xDS Config NACK and Sync Delay Sentinel

New to Kestra?