Purge forbidden keys from a NATS JetStream KV bucket against an allowlist

id: nats-kv-forbidden-key-purge-guard
namespace: company.team
description: >-
  Diff the live keys in a NATS JetStream Key/Value bucket against an allowlist
  passed as a flow input, then delete the keys nobody owns any more. Orphaned
  feature flags, abandoned leases plus keys written by a decommissioned service
  sit in a KV bucket forever when the bucket has no TTL. A reader then cannot
  tell which key is still authoritative. The flow reads the bucket, names every
  key outside the allowlist, deletes them with io.kestra.plugin.nats.kv.Delete
  behind auto_remediate, dry_run and a Pause, then reads the bucket again and
  fails the run if a forbidden key survived or an allowlisted key went missing.
  An empty allowlist is refused before the bucket is ever read.

triggers:
  - id: hourly_allowlist_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: >-
      Audit the bucket every hour. Shipped disabled. Enable it once nats_url,
      bucket_name and allowed_keys point at a real bucket and one manual run has
      been reviewed. The trigger pins auto_remediate false and dry_run true, so
      a scheduled run reports and never deletes whatever this flow's own input
      defaults say.
    cron: "23 * * * *"
    disabled: true
    inputs:
      auto_remediate: "false"
      dry_run: "true"

inputs:
  - id: nats_url
    type: STRING
    displayName: "NATS Server URL"
    defaults: "nats://localhost:4222"
    description: >-
      Server URL passed to every io.kestra.plugin.nats task here. If your server
      requires authentication, plugin-nats also accepts username, password,
      token or creds on each NATS task. Add them as secret() properties, never
      as literals.

  - id: bucket_name
    type: STRING
    displayName: "KV Bucket Name"
    defaults: "svc_feature_flags"
    description: >-
      Existing NATS JetStream Key/Value bucket to audit. The flow reads it
      through the bucket's own backing stream subject, $KV.<bucket_name>.>, so
      the bucket has to exist. NATS bucket names cannot contain a dot, which is
      what makes the key name recoverable from the subject.

  - id: allowed_keys
    type: ARRAY
    itemType: STRING
    displayName: "Allowed Keys"
    defaults:
      - "checkout_v2_enabled"
      - "maintenance_mode"
      - "primary_region"
    description: >-
      The keys this bucket is allowed to hold. Every other live key is forbidden
      and is a deletion candidate. Keys may contain dots. An empty list is
      refused outright, because an empty allowlist means every key is forbidden
      and a blank input is far more likely to be a misconfiguration than an
      instruction to wipe the bucket.

  - id: max_purge_batch
    type: INT
    displayName: "Max Keys Per Purge"
    defaults: 25
    description: >-
      Ceiling on one purge. If more forbidden keys are found than this and the
      flow is in write mode, it fails instead of deleting. A sudden jump in
      forbidden keys usually means the allowlist is wrong, not that the bucket
      filled up with junk.

  - id: scan_max_records
    type: INT
    displayName: "Max Keys Scanned"
    defaults: 500
    description: >-
      Ceiling on the bucket scan. The scan is a JetStream consumer over the
      bucket's backing stream, so this bounds how much the flow pulls from a
      large bucket. NOTE that plugin-nats checks this limit about once a second,
      so the real stop point can overshoot by a few records.

  - id: scan_max_duration
    type: DURATION
    displayName: "Scan Timeout"
    defaults: PT20S
    description: >-
      How long the bucket scan may run before it returns what it has. The scan
      always runs to this timeout on a quiet bucket, since there is no end of
      stream to wait for.

  - id: settle_after_purge
    type: DURATION
    displayName: "Settle Before Re-read"
    defaults: PT3S
    description: >-
      Pause between the delete and the re-read. A KV delete is a stream publish,
      so a re-read in the same instant can race the tombstone. Three seconds is
      enough on a local server, raise it for a clustered one.

  - id: auto_remediate
    type: SELECT
    displayName: "Auto Remediate"
    defaults: "false"
    values:
      - "false"
      - "true"
    description: >-
      If "true", delete the forbidden keys, still gated by dry_run below and by
      the Pause. If "false", the forbidden keys are only named.

  - id: dry_run
    type: SELECT
    displayName: "Dry Run"
    defaults: "true"
    values:
      - "true"
      - "false"
    description: >-
      If "true", nothing is deleted. Must be explicitly "false" together with
      auto_remediate "true" for io.kestra.plugin.nats.kv.Delete to run at all.

tasks:
  - id: refuse_empty_allowlist
    type: io.kestra.plugin.core.flow.If
    description: >-
      Refuse an empty allowlist before the bucket is read. Every live key would
      be forbidden, so the flow would propose deleting the whole bucket. A blank
      ARRAY input is a misconfiguration, not an instruction, so this run fails
      and the errors block alerts rather than continuing into a purge plan.
    condition: "{{ (inputs.allowed_keys | length) == 0 }}"
    then:
      - id: empty_allowlist_refusal
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: >-
          Refusing to audit bucket {{ inputs.bucket_name }} with an empty
          allowed_keys list. An empty allowlist marks every live key forbidden.
          Pass the keys this bucket is allowed to hold. Disable the flow if it
          has none.

  - id: list_live_keys
    type: io.kestra.plugin.nats.core.Consume
    description: >-
      Read the key names out of the bucket. plugin-nats 1.5.1 has no list task
      for KV, so the scan goes through the bucket's own backing stream,
      KV_<bucket_name>, whose subjects are $KV.<bucket_name>.<key>.
      deliverPolicy LastPerSubject returns exactly one record per key, which is
      how a KV listing is built. A deleted key still has a subject, so its last
      record arrives as a tombstone carrying the KV-Operation header. The next
      task filters those out.
    url: "{{ inputs.nats_url }}"
    subject: "$KV.{{ inputs.bucket_name }}.>"
    deliverPolicy: LastPerSubject
    maxRecords: "{{ inputs.scan_max_records }}"
    pollDuration: PT2S
    maxDuration: "{{ inputs.scan_max_duration }}"

  - id: decode_entries
    type: io.kestra.plugin.serdes.json.IonToJson
    description: >-
      Turn the Ion records plugin-nats wrote into one JSON array, so the
      classification below is a single jq pass. newLine false is what produces
      an array rather than one JSON object per line.
    from: "{{ outputs.list_live_keys.uri }}"
    newLine: false

  - id: classify_keys
    type: io.kestra.plugin.core.output.OutputValues
    description: >-
      Build the audit report in one jq pass. live is every key whose newest
      record is not a tombstone, recovered from the subject by dropping the $KV
      token and the bucket token, then deduplicated. The dedup is load bearing:
      LastPerSubject only promises one record per subject for the backlog, so a
      key written while the scan is open arrives again as a fresh record and a
      busy bucket would otherwise report the same key many times. forbidden is
      live minus the allowlist. kept is the intersection. missing_allowed is the
      allowlist entries absent from the bucket, reported here but never created.
    values:
      report: "{{ {'entries': read(outputs.decode_entries.uri), 'allowed':
        inputs.allowed_keys} | toJson | jq('. as $r | ([($r.entries|fromjson)[]
        | select(.headers|has(\"KV-Operation\")|not) |
        (.subject|split(\".\")|.[2:]|join(\".\"))] | unique) as $live | {live:
        $live, forbidden: [$live[] | . as $k | select([$r.allowed[] | select(.
        == $k)]|length == 0)], kept: [$live[] | . as $k | select([$r.allowed[] |
        select(. == $k)]|length > 0)], missing_allowed: [$r.allowed[] | . as $k
        | select([$live[] | select(. == $k)]|length == 0)]}') | first | toJson
        }}"

  - id: report_audit
    type: io.kestra.plugin.core.log.Log
    description: >-
      One status line per run, written on every branch so the execution history
      carries the trend even when nothing is deleted.
    message: >-
      NATS KV allowlist audit: bucket={{ inputs.bucket_name }}, scanned={{
      outputs.list_live_keys.messagesCount }}, live={{
      outputs.classify_keys.values.report | jq('.live|length') | first }},
      allowlisted_present={{ outputs.classify_keys.values.report |
      jq('.kept|length') | first }}, forbidden={{
      outputs.classify_keys.values.report | jq('.forbidden|length') | first }},
      forbidden_keys={{ outputs.classify_keys.values.report |
      jq('.forbidden|join(", ")') | first }}, allowlisted_missing={{
      outputs.classify_keys.values.report | jq('.missing_allowed|join(", ")') |
      first }}, auto_remediate={{ inputs.auto_remediate }}, dry_run={{
      inputs.dry_run }}, execution={{ execution.id }}.

  - id: refuse_oversized_purge
    type: io.kestra.plugin.core.flow.If
    description: >-
      Refuse a purge wider than max_purge_batch. A forbidden count that jumps
      past the ceiling is almost always a wrong or truncated allowlist, so the
      flow fails and alerts instead of deleting. Read-only runs are unaffected.
    condition: "{{ (outputs.classify_keys.values.report | jq('.forbidden|length') |
      first) > inputs.max_purge_batch and inputs.auto_remediate == 'true' and
      inputs.dry_run == 'false' }}"
    then:
      - id: oversized_purge_refusal
        type: io.kestra.plugin.core.execution.Fail
        errorMessage: >-
          Refusing to purge bucket {{ inputs.bucket_name }}. Found {{
          outputs.classify_keys.values.report | jq('.forbidden|length') | first
          }} forbidden keys, ceiling max_purge_batch is {{
          inputs.max_purge_batch }}. Check allowed_keys, then raise the ceiling
          if the count is real.

  - id: purge_forbidden_keys
    type: io.kestra.plugin.core.flow.If
    description: >-
      The write gate. Three conditions have to hold together: forbidden keys
      exist, auto_remediate is "true", dry_run is "false". Any other combination
      takes the else branch and deletes nothing.
    condition: "{{ (outputs.classify_keys.values.report | jq('.forbidden|length') |
      first) > 0 and inputs.auto_remediate == 'true' and inputs.dry_run ==
      'false' }}"
    then:
      - id: confirm_purge
        type: io.kestra.plugin.core.flow.Pause
        description: >-
          Hold for a human. Deleting a KV key is not reversible once history
          rolls, so the operator reads the forbidden list off report_audit and
          answers approved. Killing the execution here leaves the bucket
          untouched. approved must be true, since the gate below reads
          onResume.approved with a false fallback and a resume with no answer is
          not an approval.
        onResume:
          - id: approved
            type: BOOL
            displayName: "Delete the forbidden keys"
            description: >-
              true deletes every key named in report_audit's forbidden_keys
              list. false leaves the bucket as it is. So does killing the
              execution.

      - id: apply_purge
        type: io.kestra.plugin.core.flow.If
        description: >-
          Run the delete only on an explicit true.
          io.kestra.plugin.core.flow.Pause on Kestra 2.0 has no nested tasks
          property, so the gated work sits after the Pause as a sibling and
          reads its answer.
        condition: "{{ outputs.confirm_purge.onResume.approved ?? false }}"
        then:
          - id: delete_forbidden_keys
            type: io.kestra.plugin.nats.kv.Delete
            description: >-
              Delete every forbidden key in one call. The allowlist never
              reaches this task: keys is the forbidden array computed above, so
              an allowlisted key cannot be named here even by accident. NOTE
              that deletedCount reports how many keys were submitted. A KV
              delete of a key that is already gone still succeeds, so this
              number is not proof. Only the re-read below decides.
            url: "{{ inputs.nats_url }}"
            bucketName: "{{ inputs.bucket_name }}"
            keys: "{{ outputs.classify_keys.values.report | jq('.forbidden') | first |
              toJson }}"

          - id: settle_after_purge
            type: io.kestra.plugin.core.flow.Sleep
            description: Let the tombstones land before the bucket is read again.
            duration: "{{ inputs.settle_after_purge }}"

          - id: reverify_live_keys
            type: io.kestra.plugin.nats.core.Consume
            description: >-
              Read the bucket again, the same way as the first scan. This is the
              only task whose result decides whether the purge worked.
            url: "{{ inputs.nats_url }}"
            subject: "$KV.{{ inputs.bucket_name }}.>"
            deliverPolicy: LastPerSubject
            maxRecords: "{{ inputs.scan_max_records }}"
            pollDuration: PT2S
            maxDuration: "{{ inputs.scan_max_duration }}"

          - id: reverify_decode
            type: io.kestra.plugin.serdes.json.IonToJson
            from: "{{ outputs.reverify_live_keys.uri }}"
            newLine: false

          - id: reverify_classify
            type: io.kestra.plugin.core.output.OutputValues
            description: >-
              Classify the post-purge bucket with the same jq program. A purged
              key keeps its subject and comes back as a tombstone, so without
              the KV-Operation filter every deleted key would still read as
              present and this gate would fail every successful purge.
            values:
              report: "{{ {'entries': read(outputs.reverify_decode.uri), 'allowed':
                inputs.allowed_keys} | toJson | jq('. as $r |
                ([($r.entries|fromjson)[] |
                select(.headers|has(\"KV-Operation\")|not) |
                (.subject|split(\".\")|.[2:]|join(\".\"))] | unique) as $live |
                {live: $live, forbidden: [$live[] | . as $k |
                select([$r.allowed[] | select(. == $k)]|length == 0)], kept:
                [$live[] | . as $k | select([$r.allowed[] | select(. ==
                $k)]|length > 0)], missing_allowed: [$r.allowed[] | . as $k |
                select([$live[] | select(. == $k)]|length == 0)]}') | first |
                toJson }}"

          - id: reverify_report
            type: io.kestra.plugin.core.log.Log
            message: >-
              Post-purge re-read of {{ inputs.bucket_name }}: submitted={{
              outputs.delete_forbidden_keys.deletedCount }},
              forbidden_surviving={{ outputs.reverify_classify.values.report |
              jq('.forbidden|length') | first }}, surviving_keys={{
              outputs.reverify_classify.values.report | jq('.forbidden|join(",
              ")') | first }}, allowlisted_present_before={{
              outputs.classify_keys.values.report | jq('.kept|length') | first
              }}, allowlisted_present_after={{
              outputs.reverify_classify.values.report | jq('.kept|length') |
              first }}.

          - id: fail_if_purge_did_not_land
            type: io.kestra.plugin.core.flow.If
            description: >-
              Fail the run on either outcome a purge must never produce. One is
              a forbidden key still live. The other is fewer allowlisted keys
              than before. The first catches a delete that was accepted and had
              no effect, which happens when a decommissioned writer is still
              re-creating the key. The second catches a purge that reached a key
              it was never allowed to touch.
            condition: "{{ (outputs.reverify_classify.values.report |
              jq('.forbidden|length') | first) > 0 or
              (outputs.reverify_classify.values.report | jq('.kept|length') |
              first) < (outputs.classify_keys.values.report | jq('.kept|length')
              | first) }}"
            then:
              - id: purge_did_not_land
                type: io.kestra.plugin.core.execution.Fail
                errorMessage: >-
                  Purge of {{ inputs.bucket_name }} did not land. Forbidden keys
                  still live: {{ outputs.reverify_classify.values.report |
                  jq('.forbidden|join(", ")') | first }}. Allowlisted keys
                  present before {{ outputs.classify_keys.values.report |
                  jq('.kept|length') | first }}, after {{
                  outputs.reverify_classify.values.report | jq('.kept|length') |
                  first }}. A surviving key means something is still writing it,
                  so find the writer before retrying. A drop in allowlisted keys
                  means the allowlist changed mid-run.
            else:
              - id: purge_landed
                type: io.kestra.plugin.core.log.Log
                message: >-
                  Purge of {{ inputs.bucket_name }} confirmed. Forbidden keys
                  remaining 0, allowlisted keys still present {{
                  outputs.reverify_classify.values.report | jq('.kept|length') |
                  first }}.
        else:
          - id: purge_declined
            type: io.kestra.plugin.core.log.Log
            message: >-
              Purge of {{ inputs.bucket_name }} declined at the approval gate.
              Nothing was deleted. Forbidden keys still present: {{
              outputs.classify_keys.values.report | jq('.forbidden|join(", ")')
              | first }}.
    else:
      - id: no_purge
        type: io.kestra.plugin.core.log.Log
        message: >-
          No purge this run for {{ inputs.bucket_name }}. Forbidden keys {{
          outputs.classify_keys.values.report | jq('.forbidden|length') | first
          }}, auto_remediate {{ inputs.auto_remediate }}, dry_run {{
          inputs.dry_run }}. A delete needs a forbidden key, auto_remediate
          "true", dry_run "false" and an approved Pause.

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: >-
      Alert Slack whenever this flow fails, which covers the empty-allowlist
      refusal, the oversized-purge refusal, a purge that did not land plus an
      unreachable NATS server.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "nats-kv-forbidden-key-purge-guard FAILED on bucket {{ inputs.bucket_name }} (execution {{ execution.id }}). Causes: a refused allowlist, a purge wider than max_purge_batch, a forbidden key that survived the delete, an unreachable NATS server. Check the task logs before touching the bucket by hand."
      }

outputs:
  - id: forbidden_key_count
    type: INT
    description: Number of live keys outside the allowlist at the time of the audit.
    value: "{{ (outputs.classify_keys.values.report ?? '{\"forbidden\":[]}') |
      jq('.forbidden|length') | first }}"
  - id: forbidden_keys
    type: STRING
    description: Comma separated list of those keys, empty when the bucket matches
      the allowlist.
    value: "{{ (outputs.classify_keys.values.report ?? '{\"forbidden\":[]}') |
      jq('.forbidden|join(\", \")') | first }}"