Kafka ACL Baseline Restore Guard with Kestra

id: kafka-acl-baseline-restore-guard
namespace: company.team

description: |
  Keep a Kafka cluster holding the ACL bindings its services need. A declared baseline of
  required ALLOW grants comes in as a flow input. AclList reads the live bindings and one jq
  pass classifies every baseline entry as present, missing, blocked by a DENY that overrides it
  or refused for being too broad. AclCreate adds the missing bindings behind auto_remediate,
  dry_run and a Pause. A second AclList then re-reads the cluster, so the run fails if any
  baseline grant is still not in force.

inputs:
  - id: bootstrap_servers
    type: STRING
    displayName: Kafka bootstrap servers
    defaults: "localhost:9092"

  - id: baseline
    type: JSON
    displayName: Declared ACL baseline
    description: >-
      The ALLOW grants every service must hold. Each entry needs principal,
      resourceType, resourceName and operation. patternType defaults to LITERAL,
      host defaults to *. This flow never creates a DENY binding.
    defaults: |
      [
        {"principal": "User:orders-svc", "resourceType": "TOPIC", "resourceName": "orders", "operation": "READ"},
        {"principal": "User:orders-svc", "resourceType": "TOPIC", "resourceName": "orders", "operation": "DESCRIBE"},
        {"principal": "User:orders-svc", "resourceType": "GROUP", "resourceName": "orders-consumer", "operation": "READ"},
        {"principal": "User:billing", "resourceType": "TOPIC", "resourceName": "invoices", "operation": "WRITE"},
        {"principal": "User:billing", "resourceType": "TOPIC", "resourceName": "invoices", "operation": "DESCRIBE"},
        {"principal": "User:analytics", "resourceType": "TOPIC", "resourceName": "events.", "patternType": "PREFIXED", "operation": "READ"}
      ]

  - id: allow_broad_entries
    type: BOOL
    displayName: Allow broad baseline entries
    description: >-
      Off by default, so a baseline entry that names every resource (`*` with
      LITERAL), asks for operation ALL or uses patternType MATCH is refused
      rather than created.
    defaults: false

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    description: Allow AclCreate to add the missing baseline bindings. Needs dry_run
      false and an approved Pause as well.
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Must be false, together with auto_remediate, before any binding is created.
    defaults: true

tasks:
  - id: list_acls
    type: io.kestra.plugin.kafka.AclList
    description: Read-only. Every ACL binding the cluster holds, both ALLOW and DENY.
    properties:
      bootstrap.servers: "{{ inputs.bootstrap_servers }}"

  - id: plan
    type: io.kestra.plugin.core.output.OutputValues
    description: >-
      Classify every baseline entry against the live bindings. A live ALLOW
      counts for a baseline entry when it covers the same principal, resource
      and operation, so operation ALL, a PREFIXED pattern or the `*` literal all
      satisfy a narrower baseline grant. Read-only.
    values:
      report: >-
        {{ {'baseline': inputs.baseline, 'live': outputs.list_acls.acls ?? [],
        'broad_ok': inputs.allow_broad_entries} | toJson | jq('
          def covers($l; $b):
            ($l.operation == $b.operation or $l.operation == "ALL")
            and ($l.host == "*" or $l.host == $b.host)
            and (if $l.patternType == "LITERAL" then ($l.resourceName == "*" or ($b.patternType == "LITERAL" and $l.resourceName == $b.resourceName))
                 elif $l.patternType == "PREFIXED" then ($b.resourceName | startswith($l.resourceName))
                 else false end);
          .live as $live | .broad_ok as $broad_ok
          | [ .baseline[] | {principal: .principal, resourceType: .resourceType, resourceName: .resourceName,
                             operation: .operation, patternType: (.patternType // "LITERAL"), host: (.host // "*")} ]
          | [ .[] | . as $b
              | ([$live[] | select(.principal == $b.principal and .resourceType == $b.resourceType and .permissionType == "ALLOW") | select(covers(.; $b))] | length) as $allow
              | ([$live[] | select(.principal == $b.principal and .resourceType == $b.resourceType and .permissionType == "DENY") | select(covers(.; $b))] | length) as $deny
              | (($b.resourceName == "*" and $b.patternType == "LITERAL") or $b.operation == "ALL" or $b.patternType == "MATCH") as $broad
              | {entry: $b, state: (if ($broad and ($broad_ok | not)) then "refused" elif $deny > 0 then "blocked" elif $allow > 0 then "present" else "missing" end)} ]
          | {present: [.[] | select(.state == "present") | .entry],
             missing: [.[] | select(.state == "missing") | .entry],
             blocked: [.[] | select(.state == "blocked") | .entry],
             refused: [.[] | select(.state == "refused") | .entry]}') | first | toJson }}

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: >-
      Kafka ACL baseline on {{ inputs.bootstrap_servers }}: {{
      (outputs.list_acls.acls ?? []) | length }} live bindings, {{
      fromJson(outputs.plan.values.report).present | length }} baseline grants
      in force, {{ fromJson(outputs.plan.values.report).missing | length }}
      missing, {{ fromJson(outputs.plan.values.report).blocked | length }}
      blocked by a DENY, {{ fromJson(outputs.plan.values.report).refused |
      length }} refused as too broad. Missing: {{
      fromJson(outputs.plan.values.report).missing | jq('map("\(.principal)
      \(.operation) on \(.resourceType) \(.resourceName) (\(.patternType))") |
      join("; ") | if . == "" then "none" else . end') | first }}.

  - id: route
    type: io.kestra.plugin.core.flow.If
    condition: "{{ (fromJson(outputs.plan.values.report).missing | length) +
      (fromJson(outputs.plan.values.report).blocked | length) +
      (fromJson(outputs.plan.values.report).refused | length) > 0 }}"
    then:
      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {{ {'text': 'Kafka ACL baseline drift on ' ~ inputs.bootstrap_servers ~ ': ' ~ (fromJson(outputs.plan.values.report).missing | length) ~ ' missing, ' ~ (fromJson(outputs.plan.values.report).blocked | length) ~ ' blocked by a DENY, ' ~ (fromJson(outputs.plan.values.report).refused | length) ~ ' refused as too broad. Missing: ' ~ (fromJson(outputs.plan.values.report).missing | jq('map("\(.principal) \(.operation) on \(.resourceType) \(.resourceName)") | join("; ") | if . == "" then "none" else . end') | first) ~ '.' ~ (inputs.auto_remediate and not inputs.dry_run ? ' Asking for approval to create the missing bindings.' : ' Reporting only (auto_remediate=' ~ inputs.auto_remediate ~ ', dry_run=' ~ inputs.dry_run ~ ').')} | toJson }}

      - id: restore_gate
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.auto_remediate and not inputs.dry_run and
          (fromJson(outputs.plan.values.report).missing | length) > 0 }}"
        then:
          - id: approve
            type: io.kestra.plugin.core.flow.Pause
            description: >-
              Creating an ACL hands real access to a principal, so a human reads
              the list first. The blast radius is the reason for this gate, not
              danger: every binding below comes from the declared baseline,
              nothing is deleted, nothing is denied. Resume with approved=true
              to create exactly the bindings named in summary. Resume with no
              answer or kill the execution to leave the cluster untouched.
            onResume:
              - id: approved
                type: BOOL
                displayName: Approve creating the missing baseline ACL bindings
                required: false

          - id: apply_gate
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.approve.onResume.approved ?? false }}"
            then:
              - id: restore
                type: io.kestra.plugin.core.flow.Loop
                description: One AclCreate per missing baseline binding. Serial remediation, so
                  two iterations never write the ACL store at once.
                concurrencyLimit: 1
                values: "{{ fromJson(outputs.plan.values.report).missing | toJson }}"
                tasks:
                  - id: create_acl
                    type: io.kestra.plugin.kafka.AclCreate
                    properties:
                      bootstrap.servers: "{{ inputs.bootstrap_servers }}"
                    resourceType: "{{ fromJson(item.value).resourceType }}"
                    resourceName: "{{ fromJson(item.value).resourceName }}"
                    patternType: "{{ fromJson(item.value).patternType }}"
                    principal: "{{ fromJson(item.value).principal }}"
                    host: "{{ fromJson(item.value).host }}"
                    operation: "{{ fromJson(item.value).operation }}"
                    permissionType: ALLOW

              - id: relist
                type: io.kestra.plugin.kafka.AclList
                description: Read the ACL store again. The create going green says nothing on
                  its own.
                properties:
                  bootstrap.servers: "{{ inputs.bootstrap_servers }}"

              - id: verify
                type: io.kestra.plugin.core.output.OutputValues
                description: The same classification, against the bindings the cluster holds
                  now.
                values:
                  report: >-
                    {{ {'baseline': inputs.baseline, 'live': outputs.relist.acls
                    ?? [], 'broad_ok': inputs.allow_broad_entries} | toJson |
                    jq('
                      def covers($l; $b):
                        ($l.operation == $b.operation or $l.operation == "ALL")
                        and ($l.host == "*" or $l.host == $b.host)
                        and (if $l.patternType == "LITERAL" then ($l.resourceName == "*" or ($b.patternType == "LITERAL" and $l.resourceName == $b.resourceName))
                             elif $l.patternType == "PREFIXED" then ($b.resourceName | startswith($l.resourceName))
                             else false end);
                      .live as $live | .broad_ok as $broad_ok
                      | [ .baseline[] | {principal: .principal, resourceType: .resourceType, resourceName: .resourceName,
                                         operation: .operation, patternType: (.patternType // "LITERAL"), host: (.host // "*")} ]
                      | [ .[] | . as $b
                          | ([$live[] | select(.principal == $b.principal and .resourceType == $b.resourceType and .permissionType == "ALLOW") | select(covers(.; $b))] | length) as $allow
                          | ([$live[] | select(.principal == $b.principal and .resourceType == $b.resourceType and .permissionType == "DENY") | select(covers(.; $b))] | length) as $deny
                          | (($b.resourceName == "*" and $b.patternType == "LITERAL") or $b.operation == "ALL" or $b.patternType == "MATCH") as $broad
                          | {entry: $b, state: (if ($broad and ($broad_ok | not)) then "refused" elif $deny > 0 then "blocked" elif $allow > 0 then "present" else "missing" end)} ]
                      | {present: [.[] | select(.state == "present") | .entry],
                         missing: [.[] | select(.state == "missing") | .entry],
                         blocked: [.[] | select(.state == "blocked") | .entry],
                         refused: [.[] | select(.state == "refused") | .entry]}') | first | toJson }}

              - id: verify_gate
                type: io.kestra.plugin.core.flow.If
                description: Fail the run while any baseline grant is still not in force.
                condition: "{{ (fromJson(outputs.verify.values.report).missing | length) +
                  (fromJson(outputs.verify.values.report).blocked | length) +
                  (fromJson(outputs.verify.values.report).refused | length) > 0
                  }}"
                then:
                  - id: restore_failed
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: >-
                      AclCreate ran, the ACL store still does not satisfy the
                      baseline. Still missing: {{
                      fromJson(outputs.verify.values.report).missing |
                      jq('map("\(.principal) \(.operation) on \(.resourceType)
                      \(.resourceName)") | join("; ") | if . == "" then "none"
                      else . end') | first }}. Blocked by a DENY that overrides
                      the grant: {{
                      fromJson(outputs.verify.values.report).blocked |
                      jq('map("\(.principal) \(.operation) on \(.resourceType)
                      \(.resourceName)") | join("; ") | if . == "" then "none"
                      else . end') | first }}. Refused as too broad: {{
                      fromJson(outputs.verify.values.report).refused |
                      jq('map("\(.principal) \(.operation) on \(.resourceType)
                      \(.resourceName)") | join("; ") | if . == "" then "none"
                      else . end') | first }}.
                else:
                  - id: restored
                    type: io.kestra.plugin.core.log.Log
                    message: >-
                      Created {{ fromJson(outputs.plan.values.report).missing |
                      length }} binding(s). The cluster now holds {{
                      (outputs.relist.acls ?? []) | length }} ACLs and every one
                      of the {{ fromJson(outputs.verify.values.report).present |
                      length }} baseline grants is in force.
            else:
              - id: declined
                type: io.kestra.plugin.core.log.Log
                message: >-
                  Not approved, so no ACL was created. {{
                  fromJson(outputs.plan.values.report).missing | length }}
                  baseline binding(s) are still missing on {{
                  inputs.bootstrap_servers }}.
        else:
          - id: report_only
            type: io.kestra.plugin.core.log.Log
            message: >-
              Reporting only, nothing was created. Creatable now: {{
              fromJson(outputs.plan.values.report).missing | length }}
              binding(s), {{ fromJson(outputs.plan.values.report).missing |
              jq('map("\(.principal) \(.operation) on \(.resourceType)
              \(.resourceName) (\(.patternType), host \(.host))") | join("; ") |
              if . == "" then "none" else . end') | first }}. Needs a human
              instead: {{ fromJson(outputs.plan.values.report).blocked | length
              }} blocked by a DENY, {{
              fromJson(outputs.plan.values.report).refused | length }} refused
              as too broad. Set auto_remediate true with dry_run false, then
              approve the Pause, to create the creatable ones.
    else:
      - id: in_force
        type: io.kestra.plugin.core.log.Log
        message: >-
          Every one of the {{ fromJson(outputs.plan.values.report).present |
          length }} baseline grants is in force on {{ inputs.bootstrap_servers
          }}. Nothing to create.

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'Kafka ACL baseline guard failed on ' ~ inputs.bootstrap_servers ~ ' (execution ' ~ execution.id ~ '). The baseline is unverified. Any restore that was in flight did not complete.'} | toJson }}

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled. Always runs in safe mode.
    cron: "0 6 * * *"
    disabled: true
    inputs:
      auto_remediate: false
      dry_run: true

Kafka authorization drifts in the direction nobody watches. A cleanup script written to remove one stale grant removes three. A cluster is rebuilt and the ACL restore stops half way. An audit prunes a wildcard ALLOW that a service was quietly relying on. Nothing breaks for the operator who ran the command. It breaks for a consumer group that starts logging TopicAuthorizationException at 03:00. The only record of what the principal used to hold is in somebody's shell history.

This blueprint makes the required grants a declared input and keeps the cluster matching it. AclList reads every live binding, one pass classifies each baseline entry. AclCreate adds back only what is missing, behind auto_remediate, dry_run and a Pause. A second AclList then re-reads the store, so a run only goes green when every baseline grant is really in force.

This blueprint was created by zkasuran.

Findings

State	Meaning	Created by remediation
`present`	A live ALLOW binding covers this principal, resource and operation	No, nothing to do
`missing`	No live ALLOW covers it, no DENY stands in the way	Yes, one `AclCreate`
`blocked`	A live DENY covers it, so creating the ALLOW would not restore access	No, a human decides whether the DENY goes
`refused`	The entry itself is too broad (`*` literal, operation `ALL`, patternType `MATCH`)	No, unless `allow_broad_entries` is on

Coverage follows Kafka's own matching rules rather than string equality, so a baseline entry is present when a broader live grant already satisfies it: operation ALL, a PREFIXED pattern whose prefix the baseline name starts with, the * literal resource name or host *. That is what keeps the flow from piling a second narrow binding on top of a grant that already works.

Safety

Read-only with the shipped defaults. AclCreate needs auto_remediate: true, dry_run: false and an approved Pause, in that order.
The only write verb in the flow is AclCreate with permissionType: ALLOW. It never deletes a binding, never writes a DENY, never touches a principal that is not in the baseline.
refused exists so a baseline cannot smuggle in the grants an ACL audit would take straight back out.
The Pause is there for blast radius. A missing grant is a reversible outage rather than a security hole, so the risk being gated is handing a principal access a human has not read, not losing data.

How it works

list_acls (io.kestra.plugin.kafka.AclList, read-only) returns every binding on the cluster, ALLOW and DENY.
plan normalizes the baseline input (patternType defaults to LITERAL, host to *) and classifies each entry as present, missing, blocked or refused in one jq pass.
summary logs the four counts plus every missing grant in full.
route (io.kestra.plugin.core.flow.If):
- nothing to report: in_force logs that the baseline is satisfied.
- anything to report: alert posts the counts and the missing grants to Slack.
  - restore_gate runs only when auto_remediate is true, dry_run is false and something is missing. Otherwise report_only logs what it would have created.
  - approve (io.kestra.plugin.core.flow.Pause) waits for a resume carrying approved.
  - apply_gate reads outputs.approve.onResume.approved ?? false, so an unanswered resume is not an approval. When it is false, declined logs that nothing was created.
  - restore (io.kestra.plugin.core.flow.Loop at concurrencyLimit: 1) runs one io.kestra.plugin.kafka.AclCreate per missing binding.
  - relist reads the ACL store again, verify re-runs the same classification against it. verify_gate fails the run through io.kestra.plugin.core.execution.Fail if any baseline grant is still not in force. On a clean result restored logs the new totals.
errors posts to Slack on any failure, for example an unreachable broker or a principal that cannot alter ACLs.

What you get

The grants your services depend on, written down as data instead of living in a runbook.
A daily read-only check that an ACL cleanup or a half-finished restore has not quietly removed access.
A restore that creates only the bindings that are genuinely absent, each one matched against Kafka's real ACL semantics.
Proof the restore landed: a second AclList and a failed run if a grant is still missing.
A clear split between what this flow can fix (an absent ALLOW) and what it refuses to touch (a DENY in the way, an over-broad entry).

Who it's for

Platform teams running Kafka with an authorizer, where several services share one cluster.
Anyone who has rebuilt a cluster or restored ACLs from a backup and wants the result checked against the contract.
Teams that run ACL pruning and want a counterweight that catches an over-prune the same day.
Teams evaluating the Kafka plugin who want AclList plus AclCreate with real gates around them.

Why orchestrate this with Kestra

kafka-acls.sh --add restores a binding, but somebody has to notice it went. The baseline lives next to the flow as a typed input, so a change to it is a reviewed change. Kestra runs the diff on a schedule, keeps every run as evidence of what the cluster held that day, posts the drift to Slack, holds the write behind two inputs plus a human resume, then re-reads the cluster and marks the run red if the grants are still not there. The audit trail is the part a shell loop cannot give you.

Tested end to end

On Kestra 2.0.5 OSS against Apache Kafka 3.9.1 in KRaft mode with StandardAuthorizer, with a local HTTP sink standing in for Slack. Every binding count below was read with the cluster's own kafka-acls.sh --list, before and after each run, not from the Kestra log. The seeded cluster holds the 6 baseline grants plus one unrelated User:metrics READ on heartbeat, which no run ever touched.

Run	Cluster state and inputs	Result
1	baseline fully seeded, 7 bindings, defaults	6 present, 0 missing. Nothing created, still 7 bindings
2	3 bindings deleted (4 left), defaults	3 missing found, Slack alert sent, nothing created, still 4
3	same 4 bindings, `auto_remediate: true`, `dry_run: true`	same 3 reported as creatable, still 4 bindings
4	same 4 bindings, `auto_remediate: true`, `dry_run: false`, resumed with `approved: false`	`declined` logged, still 4 bindings
5	same 4 bindings, `auto_remediate: true`, `dry_run: false`, resumed with `approved: true`	3 `AclCreate` calls, re-list clean, 7 bindings, the `PREFIXED` grant back as `PREFIXED`
6	1 grant deleted plus a DENY seeded over a second one (6 bindings), approved	1 missing plus 1 blocked. The missing one created (7 bindings), the run then failed on the re-verify naming the DENY
7	baseline edited to ask for `*` literal, operation `ALL` and patternType `MATCH`	all 3 refused as too broad, 0 created
8	unreachable broker	`list_acls` failed, `errors` alert sent

Runs 1 and 2 are the same flow with the same inputs against two cluster states, which is what shows the check can actually fail: 0 missing on the healthy cluster, 3 missing once the bindings were pulled out from under it.

Prerequisites

A Kafka cluster with an authorizer enabled, reachable from the Kestra worker.
A principal that can DESCRIBE the cluster to list ACLs, plus ALTER on the cluster to create them. The read path needs only DESCRIBE.
Local testing: run apache/kafka:3.9.1 with KAFKA_AUTHORIZER_CLASS_NAME=org.apache.kafka.metadata.authorizer.StandardAuthorizer plus KAFKA_SUPER_USERS=User:ANONYMOUS, then seed the baseline with kafka-acls.sh --add.

Secrets

SLACK_WEBHOOK_URL: used by alert and by the errors block. In Kestra OSS, provide it as the base64-encoded environment variable SECRET_SLACK_WEBHOOK_URL.

Inputs

Input	Default	Purpose
`bootstrap_servers`	`localhost:9092`	Brokers
`baseline`	6 example grants for 3 services	The ALLOW grants that must exist
`allow_broad_entries`	`false`	Whether a wildcard or `ALL` baseline entry may be created
`auto_remediate`	`false`	Allow `AclCreate`
`dry_run`	`true`	Must be `false` for a create to run

Quick start

Add the SLACK_WEBHOOK_URL secret.
Import the flow, set bootstrap_servers and replace baseline with the grants your services need. kafka-acls.sh --list on a healthy cluster is a good first draft.
Run it with the defaults and read summary. Everything should be present.
Delete one binding you can afford to lose, run it again, then restore it with auto_remediate: true plus dry_run: false and approve the Pause.
Enable the daily trigger, which always runs in safe mode.

How to extend

Keep the baseline in Git and pass it in from a io.kestra.plugin.core.http.Request or a repository sync, so the contract has a review history.
Add a kv.Set of the last clean classification, then alert on a grant that goes missing twice in a week.
Pair it with an ACL pruning flow and run this one straight afterwards, so an over-prune is caught in the same window.
Point it at a staging cluster first and diff the two baselines to find grants production has that staging does not.
Open a ticket for each blocked entry, since removing a DENY is a decision that needs an owner.

Kafka ACL Baseline Restore Guard

Findings

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Kafka ACL Baseline Restore Guard

Findings

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Hourly OpenSearch Error Spike Gate with Discord Alerting

Search OpenAlex for Related Scholarly Works

Use OpenSearch as a Config Store with Put and Get Roundtrip

OpenSearch ISM Policy Compliance Gate

Automated OpenLineage Metadata Extraction and Marquez Governance Pipeline

OpenMetadata Catalog Health and Change-Event Sentinel

More Infrastructure Blueprints

Back Up Grafana Dashboards to Git and Post What Changed

Open a GitLab Issue Automatically on Pipeline Failure

GlusterFS Brick Split Brain and Heal Backlog Sentinel

GitLab Runner Concurrent Job Capacity Sentinel

Keep GitLab runner pools balanced so pipelines stop queuing

GKE Cluster Health and Observability Compliance Gate

New to Kestra?