KMS Key Rotation Compliance Gate with Kestra

id: kms-key-rotation-compliance-gate
namespace: company.team

description: |
  Read a customer-managed KMS key's real KeyRotationEnabled status, classify it COMPLIANT,
  CRITICAL, WARNING or PROBE_FAILED, alert Slack when rotation is disabled, optionally enable
  rotation and verify it took, and log a final audit summary either way.

inputs:
  - id: key_id
    type: STRING
    displayName: KMS key ID
    description: Customer-managed KMS key ID or ARN to audit. AWS-managed keys
      always report automatic rotation on their own schedule and are out of
      scope for this gate.
    defaults: "<your-kms-key-id>"

  - id: aws_region
    type: STRING
    displayName: AWS region
    description: Region passed to io.kestra.plugin.aws.cli.AwsCLI as AWS_DEFAULT_REGION.
    defaults: "us-east-1"

  - id: violation_severity
    type: SELECT
    displayName: Violation severity
    description: How a disabled-rotation finding is classified. critical pages
      immediately, warning logs it as a lower-urgency finding for a key with a
      known exception.
    values:
      - "critical"
      - "warning"
    defaults: "critical"

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    description: When true, together with dry_run false, enables rotation on this
      key if it is found disabled, then verifies the change took.
    defaults: false

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

  - id: use_localstack
    type: BOOL
    displayName: Use local LocalStack container
    description: When true, points the AWS CLI at a local LocalStack container
      instead of real AWS. Both GetKeyRotationStatus and EnableKeyRotation are
      implemented in LocalStack's community edition.
    defaults: true

  - id: localstack_endpoint
    type: STRING
    displayName: LocalStack endpoint
    description: Only used when use_localstack is true.
    defaults: "http://localhost:4566"

tasks:
  - id: route_connection_mode
    type: io.kestra.plugin.core.flow.If
    description: Picks the AWS CLI endpoint. The LocalStack branch sets
      endpointOverride and uses LocalStack's own fixed test credentials, the
      live branch omits endpointOverride and uses real AWS credentials.
    condition: "{{ inputs.use_localstack }}"
    then:
      - id: probe_rotation_localstack
        type: io.kestra.plugin.aws.cli.AwsCLI
        description: "Read-only. Runs kms get-key-rotation-status and captures its own
          fields, KeyId and KeyRotationEnabled among them, as named outputs via
          the AwsCLI plugin's documented ::{\"outputs\":{}}:: echo convention.
          No --query reshaping needed, the command's own JSON is already a flat
          object."
        accessKeyId: "test"
        secretKeyId: "test"
        region: "{{ inputs.aws_region }}"
        endpointOverride: "{{ inputs.localstack_endpoint }}"
        commands:
          - |-
            aws kms get-key-rotation-status --key-id {{ inputs.key_id }} --output json | tr -d '\n' | xargs -0 -I {} echo '::{"outputs":{}}::'
    else:
      - id: probe_rotation_live
        type: io.kestra.plugin.aws.cli.AwsCLI
        description: "Read-only. Same get-key-rotation-status call against real AWS."
        accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
        secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
        region: "{{ inputs.aws_region }}"
        commands:
          - |-
            aws kms get-key-rotation-status --key-id {{ inputs.key_id }} --output json | tr -d '\n' | xargs -0 -I {} echo '::{"outputs":{}}::'

  - id: classify_rotation_status
    type: io.kestra.plugin.core.debug.Return
    description: PROBE_FAILED when neither probe branch returned at all, COMPLIANT
      when KeyRotationEnabled is true, otherwise CRITICAL or WARNING per
      violation_severity.
    format: "{{ (outputs.probe_rotation_localstack is not defined and
      outputs.probe_rotation_live is not defined) ? 'PROBE_FAILED' :
      (((outputs.probe_rotation_localstack is defined ?
      outputs.probe_rotation_localstack.vars.KeyRotationEnabled :
      outputs.probe_rotation_live.vars.KeyRotationEnabled) == true) ?
      'COMPLIANT' : (inputs.violation_severity == 'critical' ? 'CRITICAL' :
      'WARNING')) }}"

  - id: route_by_rotation_status
    type: io.kestra.plugin.core.flow.Switch
    description: Branch A logs a compliant key. Branch B alerts Slack when rotation
      is disabled, or the probe itself failed.
    value: "{{ outputs.classify_rotation_status.value }}"
    cases:
      "COMPLIANT":
        - id: log_compliant
          type: io.kestra.plugin.core.log.Log
          message: "KMS key {{ inputs.key_id }} compliant. KeyRotationEnabled=true."
      "WARNING":
        - id: alert_warning
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "KMS key " ~ inputs.key_id ~ " WARNING. Automatic rotation is disabled."} | toJson }}
      "CRITICAL":
        - id: alert_critical
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "KMS key " ~ inputs.key_id ~ " CRITICAL. Automatic rotation is disabled. Data encrypted under this key has been using the same key material indefinitely."} | toJson }}
      "PROBE_FAILED":
        - id: alert_probe_failed
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "KMS key rotation probe for " ~ inputs.key_id ~ " did not return a result. Check aws_region, use_localstack, that the key exists, and AWS credentials."} | toJson }}

  - id: remediate_gate
    type: io.kestra.plugin.core.flow.If
    description: Runs only when the key was not COMPLIANT, auto_remediate is true,
      and dry_run is false.
    condition: "{{ outputs.classify_rotation_status.value != 'COMPLIANT' and
      inputs.auto_remediate and not inputs.dry_run }}"
    then:
      - id: route_remediation_connection
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.use_localstack }}"
        then:
          - id: enable_rotation_localstack
            type: io.kestra.plugin.aws.cli.AwsCLI
            description: Calls kms enable-key-rotation on this one key only, never a bulk
              operation.
            allowFailure: true
            accessKeyId: "test"
            secretKeyId: "test"
            region: "{{ inputs.aws_region }}"
            endpointOverride: "{{ inputs.localstack_endpoint }}"
            commands:
              - "aws kms enable-key-rotation --key-id {{ inputs.key_id }}"

          - id: verify_rotation_localstack
            type: io.kestra.plugin.aws.cli.AwsCLI
            allowFailure: true
            accessKeyId: "test"
            secretKeyId: "test"
            region: "{{ inputs.aws_region }}"
            endpointOverride: "{{ inputs.localstack_endpoint }}"
            commands:
              - |-
                aws kms get-key-rotation-status --key-id {{ inputs.key_id }} --output json | tr -d '\n' | xargs -0 -I {} echo '::{"outputs":{}}::'
        else:
          - id: enable_rotation_live
            type: io.kestra.plugin.aws.cli.AwsCLI
            description: Calls kms enable-key-rotation on this one key only, never a bulk
              operation.
            allowFailure: true
            accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
            secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
            region: "{{ inputs.aws_region }}"
            commands:
              - "aws kms enable-key-rotation --key-id {{ inputs.key_id }}"

          - id: verify_rotation_live
            type: io.kestra.plugin.aws.cli.AwsCLI
            allowFailure: true
            accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
            secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
            region: "{{ inputs.aws_region }}"
            commands:
              - |-
                aws kms get-key-rotation-status --key-id {{ inputs.key_id }} --output json | tr -d '\n' | xargs -0 -I {} echo '::{"outputs":{}}::'

      - id: report_remediation_result
        type: io.kestra.plugin.core.flow.If
        description: Fails the flow when the verify probe did not return at all or still
          reports rotation disabled, so a failed enable is never silently
          reported as fixed.
        condition: "{{ (outputs.verify_rotation_localstack is not defined and
          outputs.verify_rotation_live is not defined) or
          ((outputs.verify_rotation_localstack is defined ?
          outputs.verify_rotation_localstack.vars.KeyRotationEnabled :
          outputs.verify_rotation_live.vars.KeyRotationEnabled) != true) }}"
        then:
          - id: remediation_not_confirmed
            type: io.kestra.plugin.core.execution.Fail
            errorMessage: "enable-key-rotation ran for key {{ inputs.key_id }}, but rotation
              still does not show as enabled."
        else:
          - id: remediation_confirmed
            type: io.kestra.plugin.core.log.Log
            message: "Rotation enabled and verified for KMS key {{ inputs.key_id }}."

  - id: log_audit_summary
    type: io.kestra.plugin.core.log.Log
    description: Final audit log confirming the gate ran, regardless of the branch
      taken above.
    message: "KMS key rotation compliance gate completed. key={{ inputs.key_id }}
      status={{ outputs.classify_rotation_status.value }} auto_remediate={{
      inputs.auto_remediate }} dry_run={{ inputs.dry_run }}"

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {{ {"text": "KMS key rotation compliance gate flow failed (execution " ~ execution.id ~ "). Check the execution logs."} | toJson }}

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled, and always in safe mode against LocalStack by default.
    cron: "0 8 * * *"
    disabled: true
    inputs:
      key_id: "<your-kms-key-id>"
      aws_region: "us-east-1"
      violation_severity: "critical"
      auto_remediate: false
      dry_run: true
      use_localstack: true
      localstack_endpoint: "http://localhost:4566"

A customer-managed KMS key with automatic rotation disabled keeps encrypting new data under the same key material indefinitely, exactly the setting most security baselines and compliance frameworks require enabled, and nothing about the key itself looks different, encryption and decryption both keep working whether rotation is on or off. get-key-rotation-status already reports the real KeyRotationEnabled value, this blueprint is what reads it on a schedule and turns a disabled key into an alert, with a narrow, single-key remediation path instead of leaving it as a standing finding.

How it works

route_connection_mode (io.kestra.plugin.core.flow.If) picks the AWS CLI endpoint based on use_localstack. Both branches run the identical aws kms get-key-rotation-status --key-id key_id via io.kestra.plugin.aws.cli.AwsCLI, captured as named outputs through the plugin's own documented ::{"outputs":{}}:: echo convention. The command's own JSON response is already a flat object (KeyId, KeyRotationEnabled, and more), so no --query reshaping is needed this time, only the connection differs between branches.
classify_rotation_status (io.kestra.plugin.core.debug.Return) reads whichever branch's vars.KeyRotationEnabled exists (guarded with is defined/is not defined) and returns PROBE_FAILED when neither branch returned at all, COMPLIANT when it is true, CRITICAL or WARNING depending on violation_severity when it is not.
route_by_rotation_status (io.kestra.plugin.core.flow.Switch) branches on that status: COMPLIANT logs it, WARNING, CRITICAL, and PROBE_FAILED each alert Slack.
remediate_gate (io.kestra.plugin.core.flow.If) runs only when the key was not COMPLIANT, auto_remediate is true, and dry_run is false:
- route_remediation_connection picks the same connection mode again, then calls aws kms enable-key-rotation --key-id key_id on this one key, never a bulk call across every key in the account.
- verify_rotation_* re-runs get-key-rotation-status, and report_remediation_result (io.kestra.plugin.core.flow.If) fails the flow with io.kestra.plugin.core.execution.Fail if it still does not report true, so a failed enable is never silently reported as fixed.
log_audit_summary always runs last and logs the final classification.
The errors block posts to Slack if the flow itself fails outside the gate logic above.

What you get

A real answer to whether a specific KMS key actually has rotation enabled, not an assumption from whoever created it.
A Slack alert the moment a key's rotation setting changes to disabled, instead of finding out during a compliance audit months later.
A narrow, single-key remediation path that verifies the fix actually took, instead of trusting the enable call's own silence.
A fully local test path: the default inputs point at a LocalStack container with no real AWS account needed, both operations this blueprint uses are implemented in LocalStack's community edition.

Who it's for

Security and platform teams enforcing that every customer-managed KMS key rotates, a common requirement in SOC 2, PCI DSS, and similar frameworks.
Teams who want a standing, scheduled check instead of relying on a point-in-time audit to catch a key someone created, or changed, without rotation enabled.
Teams evaluating the AWS CLI plugin who want a working example of the ::{"outputs":{}}:: convention applied to a command whose own JSON output is already a flat object, no --query reshaping required.

Why orchestrate this with Kestra

get-key-rotation-status is a single point-in-time read, KMS itself has no concept of a schedule, a severity policy, or an alert built from what it finds. Kestra adds the recurring check, the severity policy per key, the Slack alert, the single-key gated remediation, and the verification that the fix worked, all as one declarative YAML file you can version and review like code, runnable against LocalStack before it ever touches a real key.

Prerequisites

For local testing, a LocalStack container, no AWS account needed:
```
docker run --rm -d --name localstack -p 4566:4566 localstack/localstack
```
Port 4566 does not collide with Kestra's own default UI port. Both GetKeyRotationStatus and EnableKeyRotation are implemented in LocalStack's community edition.
For real AWS, an IAM credential with kms:GetKeyRotationStatus permission to read, and additionally kms:EnableKeyRotation if auto_remediate will ever be used.
A Slack incoming webhook for alerts.
If Kestra itself runs in a container and use_localstack is true, point localstack_endpoint at http://host.docker.internal:4566 or connect Kestra's container to LocalStack's Docker network and use its container name instead, check the exact network name first with docker network ls.

Secrets

AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY: used by the *_live tasks only, not needed when use_localstack is true, LocalStack accepts the fixed test/test credentials this blueprint hard-codes on the LocalStack branches.
SLACK_WEBHOOK: Slack incoming webhook URL used by the alert tasks and the errors block.

In Kestra OSS, set these as base64-encoded environment variables, for example SECRET_AWS_ACCESS_KEY_ID, SECRET_SLACK_WEBHOOK.

Inputs

Input	Type	Default	Allowed values
`key_id`	STRING	`<your-kms-key-id>`	any customer-managed KMS key ID or ARN
`aws_region`	STRING	`us-east-1`	any AWS region
`violation_severity`	SELECT	`critical`	`critical`, `warning`
`auto_remediate`	BOOL	`false`	`true`, `false`
`dry_run`	BOOL	`true`	`true`, `false`
`use_localstack`	BOOL	`true`	`true`, `false`
`localstack_endpoint`	STRING	`http://localhost:4566`	any LocalStack endpoint URL

Quick start

Start LocalStack and create a test key without rotation: aws --endpoint-url=http://localhost:4566 kms create-key (note the returned KeyId).
Add the SLACK_WEBHOOK secret.
Set key_id to that key and run this flow manually, it should report CRITICAL.
Set auto_remediate: true and dry_run: false and run it again, it should enable rotation, verify it, and report remediation_confirmed.
Set use_localstack: false and add the AWS secrets to point this at a real key.
Enable the daily trigger.

How to extend

Loop this flow over a list of key IDs with io.kestra.plugin.core.flow.Loop to audit every customer-managed key in an account from one flow, instead of one key per run.
Add a second probe for describe-key and check KeyManager is CUSTOMER before proceeding, as a safety check against accidentally pointing key_id at an AWS-managed key.
Read RotationPeriodInDays and NextRotationDate, also returned by get-key-rotation-status, to alert when a key's rotation period itself was changed to something unexpectedly long, not only when rotation is fully disabled.

Pitfalls

AWS-managed keys (aws/s3, aws/rds, and similar) rotate automatically on their own schedule and cannot have rotation disabled or enabled through this API the same way, pointing key_id at one is out of scope for this gate, scope it to customer-managed keys only.
enable-key-rotation on a key pending deletion, or in certain other non-Enabled key states, fails. allowFailure: true on enable_rotation_* lets the flow continue to verify_rotation_* either way, which then correctly reports the fix did not take.
This blueprint remediates exactly one key per run, the one named in key_id, there is no bulk-enable path, intentionally, enabling rotation account-wide without reviewing which keys are AWS-managed first is not something this flow will do automatically.
KeyRotationEnabled only covers automatic rotation. RotateKeyOnDemand, a separate operation, performs a one-time rotation without changing this setting, this gate does not check for or trigger that.

KMS Key Rotation Compliance Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

KMS Key Rotation Compliance Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

Kafka Idle Consumer Partition Guard

Kafka DLQ Quarantine, Classification, and Safe Replay Gate

Kafka Producer Compression Ratio Sentinel

Kafka Consumer Lag Sentinel

Block a breaking Avro schema before a Kafka producer deploys

Kafka Schema Registry Subject Version Pruning Guard

More Cloud Blueprints

Kubernetes Orphaned Volume and Cloud Storage FinOps Reclaimer

Knative Service Revision Failure and Scale Sentinel

Kubecost Namespace Rightsizing with Approval and Chargeback

Gate Kubernetes Maintenance on PodDisruptionBudget Readiness

Block Deployments When Kubernetes Autoscaling Is Unhealthy

Kong Upstream Target Health and Balancer Sentinel

New to Kestra?