AWS KMS Customer Managed Key (CMK) Automatic Rotation Drift Sentinel

id: aws-kms-customer-managed-key-rotation-sentinel
namespace: company.cloud

inputs:
  - id: aws_region
    type: STRING
    defaults: "us-east-1"
    description: AWS Region to audit.

tasks:
  - id: audit_kms_keys
    type: io.kestra.plugin.aws.cli.AwsCLI
    accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
    secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
    region: "{{ inputs.aws_region }}"
    commands:
      - |
        # List all customer managed KMS keys
        aws kms list-keys --query "Keys[*].KeyId" --output text > /tmp/kms_keys.txt

        # Check key rotation status for each key
        python3 -c "import subprocess; keys=open('/tmp/kms_keys.txt').read().split(); print(len([k for k in keys if subprocess.run(['aws','kms','get-key-rotation-status','--key-id',k,'--query','KeyRotationEnabled','--output','text'],capture_output=True,text=True).stdout.strip()=='False']))" > /tmp/disabled_count.txt

        DISABLED_COUNT=$(cat /tmp/disabled_count.txt | tr -d '[:space:]')
        echo "::set-output name=disabled_count::${DISABLED_COUNT:-0}::"

  - id: evaluate_rotation_compliance
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.audit_kms_keys.vars.disabled_count | int > 0 }}"
    then:
      - id: notify_security_team
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK') }}"
        payload: |
          {
            "channel": "#security-ops",
            "attachments": [
              {
                "color": "warning",
                "title": "🔐 AWS KMS Key Rotation Compliance Warning",
                "text": "AWS Region `{{ inputs.aws_region }}` contains *{{ outputs.audit_kms_keys.vars.disabled_count }} Customer Managed Key(s) with automatic rotation DISABLED*! Cryptographic compliance standards (PCI-DSS, SOC2, HIPAA) mandate annual automatic key rotation.",
                "footer": "Kestra Cloud Sentinel • AWS KMS"
              }
            ]
          }

errors:
  - id: alert_pipeline_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {
        "channel": "#security-ops",
        "attachments": [
          {
            "color": "danger",
            "title": "⚠️ AWS KMS Rotation Sentinel Failure",
            "text": "Failed to audit AWS KMS keys. Verify IAM permissions for kms:ListKeys and kms:GetKeyRotationStatus.",
            "footer": "Kestra Workflow Failure Sentinel"
          }
        ]
      }

triggers:
  - id: periodic_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 8 * * 1"
    disabled: true