Kafka ACL Overly Broad Grant Audit

id: kafka-acl-overly-broad-grant-audit
namespace: company.team

description: |
  Audit Kafka ACLs for grants that are broader than any service needs: ALLOW for every principal
  (User:*), ALLOW on every topic or group (name *), cluster-wide ALL or ALTER, operation ALL on
  a topic or group, and ACLs for principals that are not a known service account. Alert Slack,
  and optionally delete the high-severity ACLs one by one with AclDelete, then list the ACLs
  again to prove they are gone. Medium findings are reported, never deleted.

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

  - id: known_principals
    type: ARRAY
    itemType: STRING
    displayName: Known principals
    description: Service accounts expected to have ACLs. ACLs for any other
      principal are reported as unknown.
    defaults:
      - "User:orders-svc"
      - "User:billing"
      - "User:analytics"

  - id: admin_principals
    type: ARRAY
    itemType: STRING
    displayName: Admin principals
    description: Principals allowed to hold broad grants (operation ALL, cluster
      rights, wildcard resources).
    defaults:
      - "User:kafka-admin"

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    description: Delete the ACLs behind high-severity findings. Medium findings are
      only reported.
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Must be false, together with auto_remediate, for AclDelete to run.
    defaults: true

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

  - id: findings
    type: io.kestra.plugin.core.output.OutputValues
    description: One finding per ACL, with the most severe rule it breaks.
    values:
      items: >-
        {{ {'acls': outputs.list_acls.acls ?? [], 'known':
        inputs.known_principals, 'admins': inputs.admin_principals} | jq('
           .known as $known | .admins as $admins
           | [.acls[] | select(.permissionType == "ALLOW") | . as $a
              | ($admins | index($a.principal) != null) as $admin
              | [ (if .principal == "User:*" then {code: "WILDCARD_PRINCIPAL", severity: "HIGH", why: "any authenticated or anonymous client"} else empty end),
                  (if (.resourceName == "*" and .patternType == "LITERAL") and ($admin | not) then {code: "WILDCARD_RESOURCE", severity: "HIGH", why: "every \(.resourceType | ascii_downcase)"} else empty end),
                  (if .resourceType == "CLUSTER" and (.operation == "ALL" or .operation == "ALTER") and ($admin | not) then {code: "CLUSTER_ADMIN_RIGHTS", severity: "HIGH", why: "can alter the cluster, ACLs included"} else empty end),
                  (if .operation == "ALL" and .resourceType != "CLUSTER" and ($admin | not) then {code: "OPERATION_ALL", severity: "MEDIUM", why: "includes DELETE and ALTER"} else empty end),
                  (if .principal != "User:*" and ($known | index($a.principal) == null) and ($admin | not) then {code: "UNKNOWN_PRINCIPAL", severity: "MEDIUM", why: "not a known service account"} else empty end) ]
              | select(length > 0) | (sort_by(.severity) | first) + {acl: $a, codes: (map(.code) | join("+"))}]') | first | toJson }}

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: >-
      Kafka ACL audit on {{ inputs.bootstrap_servers }}: {{
      outputs.list_acls.acls | length }} ACLs, {{
      fromJson(outputs.findings.values.items) | length }} findings. {{
      fromJson(outputs.findings.values.items) | jq('map("\(.severity) \(.codes)
      \(.acl.principal) \(.acl.operation) on \(.acl.resourceType)
      \(.acl.resourceName) (\(.acl.patternType)): \(.why)") | join("; ")') |
      first }}

  - id: route
    type: io.kestra.plugin.core.flow.If
    condition: "{{ fromJson(outputs.findings.values.items) | length > 0 }}"
    then:
      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {{ {'text': 'Kafka ACL audit on ' ~ inputs.bootstrap_servers ~ ': ' ~ (fromJson(outputs.findings.values.items) | jq('map("\(.severity) \(.codes): \(.acl.principal) \(.acl.operation) on \(.acl.resourceType) \(.acl.resourceName)") | join("; ")') | first) ~ '.' ~ (inputs.auto_remediate and not inputs.dry_run ? ' Deleting the high-severity ACLs now.' : ' No change applied (auto_remediate=' ~ inputs.auto_remediate ~ ', dry_run=' ~ inputs.dry_run ~ ').')} | toJson }}

      - id: remediate_gate
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.auto_remediate and not inputs.dry_run and
          fromJson(outputs.findings.values.items) | jq('map(select(.severity ==
          \"HIGH\")) | length') | first > 0 }}"
        then:
          - id: remediate
            type: io.kestra.plugin.core.flow.Loop
            description: One AclDelete per high-severity ACL, matching every field so no
              other binding is removed.
            values: "{{ fromJson(outputs.findings.values.items) | jq('[.[] |
              select(.severity == \"HIGH\") | .acl]') | first | toJson }}"
            tasks:
              - id: delete
                type: io.kestra.plugin.kafka.AclDelete
                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: "{{ fromJson(item.value).permissionType }}"

          - id: relist
            type: io.kestra.plugin.kafka.AclList
            properties:
              bootstrap.servers: "{{ inputs.bootstrap_servers }}"

          - id: verify_result
            type: io.kestra.plugin.core.flow.If
            description: Every deleted ACL must be absent, and every other ACL must still be
              there.
            condition: >-
              {{ {'f': fromJson(outputs.findings.values.items), 'before':
              outputs.list_acls.acls, 'after': outputs.relist.acls ?? []} | jq('
                 ([.f[] | select(.severity == "HIGH") | .acl | tojson]) as $del | ([.after[] | tojson]) as $now
                 | ([$del[] | select(. as $x | $now | index($x))] | length) + ([.before[] | tojson | select(. as $x | ($del | index($x) | not) and ($now | index($x) | not))] | length)') | first > 0 }}
            then:
              - id: remediation_failed
                type: io.kestra.plugin.core.execution.Fail
                errorMessage: "AclDelete ran, but the ACL list after it does not match the plan:
                  a high-severity ACL is still there, or another ACL
                  disappeared."
            else:
              - id: remediated
                type: io.kestra.plugin.core.log.Log
                message: >-
                  Deleted and verified {{
                  fromJson(outputs.findings.values.items) |
                  jq('map(select(.severity == "HIGH")) | length') | first }}
                  ACLs, {{ outputs.relist.acls | length }} ACLs left. Medium
                  findings left for review: {{
                  fromJson(outputs.findings.values.items) |
                  jq('map(select(.severity == "MEDIUM") | "\(.acl.principal)
                  \(.acl.operation) \(.acl.resourceName)") | join(", ")') |
                  first }}.
    else:
      - id: clean
        type: io.kestra.plugin.core.log.Log
        message: "No overly broad or unknown ACL."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'Kafka ACL audit failed on ' ~ inputs.bootstrap_servers ~ ' (execution ' ~ execution.id ~ '). Treat the ACLs as unaudited.'} | toJson }}

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