AWS IAM Credential Hygiene Audit with Kestra

id: aws-iam-credential-hygiene-audit
namespace: company.team
description: |
  Audit IAM credential hygiene from the AWS credential report: keys older than
  the rotation policy, credentials nobody has used, console passwords without
  MFA - one Slack advisory with every finding, weekly or on demand.

inputs:
  - id: aws_region
    type: STRING
    displayName: AWS region
    description: IAM is a global service - any valid region works for the API call.
    defaults: us-east-1

  - id: max_key_age_days
    type: INT
    displayName: Maximum key age (days)
    description: Access keys and passwords older than this are reported as overdue
      for rotation.
    defaults: 90

  - id: max_unused_days
    type: INT
    displayName: Maximum unused days (days)
    description: Active credentials unused for longer than this are reported, as is
      any never used.
    defaults: 45

  - id: require_mfa
    type: BOOL
    displayName: Require MFA
    description: Report console users without MFA active when true.
    defaults: true

tasks:
  - id: generate_report
    type: io.kestra.plugin.aws.cli.AwsCLI
    description: Build the account credential report - AWS generates it
      asynchronously, so this is a separate step.
    accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
    secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
    region: "{{ inputs.aws_region }}"
    commands:
      - aws iam generate-credential-report

  - id: fetch_report
    type: io.kestra.plugin.aws.cli.AwsCLI
    description: Pull the base64-encoded CSV of every user's credential state.
    accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
    secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
    region: "{{ inputs.aws_region }}"
    commands:
      - aws iam get-credential-report --query Content --output text

  - id: analyze_credentials
    type: io.kestra.plugin.scripts.python.Script
    description: Apply the rotation, usage, and MFA rules to every row of the
      credential report.
    taskRunner:
      type: io.kestra.plugin.core.runner.Process
    env:
      REPORT_B64: "{{ outputs.fetch_report.stdOut }}"
    script: |
      import base64
      import csv
      import io
      import json
      import os
      from datetime import date

      MAX_KEY_AGE = int({{ inputs.max_key_age_days }})
      MAX_UNUSED = int({{ inputs.max_unused_days }})
      REQUIRE_MFA = "{{ inputs.require_mfa }}" == "true"

      today = date.today()
      findings = []
      scanned = 0


      def parse_date(value):
          if not value or value in ("N/A", "no_information", "not_supported"):
              return None
          try:
              return date.fromisoformat(value[:10])
          except ValueError:
              return None


      def age_days(value):
          parsed = parse_date(value)
          return None if parsed is None else (today - parsed).days


      def add(user, kind, detail):
          findings.append({"user": user, "kind": kind, "detail": detail})


      raw = base64.b64decode(os.environ["REPORT_B64"].strip()).decode("utf-8")
      for row in csv.DictReader(io.StringIO(raw)):
          user = row.get("user", "")
          scanned += 1

          for slot in ("1", "2"):
              active = row.get("access_key_%s_active" % slot) == "true"
              if not active:
                  continue
              rotated = age_days(row.get("access_key_%s_last_rotated" % slot))
              if rotated is not None and rotated > MAX_KEY_AGE:
                  add(user, "stale_key", "access key %s is %d days old (policy %d)" % (slot, rotated, MAX_KEY_AGE))
              used = age_days(row.get("access_key_%s_last_used_date" % slot))
              if used is None:
                  add(user, "unused_key", "access key %s has never been used" % slot)
              elif used > MAX_UNUSED:
                  add(user, "unused_key", "access key %s unused for %d days (policy %d)" % (slot, used, MAX_UNUSED))

          password_on = row.get("password_enabled") == "true"
          if password_on:
              changed = age_days(row.get("password_last_changed"))
              if changed is not None and changed > MAX_KEY_AGE:
                  add(user, "stale_password", "console password unchanged for %d days (policy %d)" % (changed, MAX_KEY_AGE))
              if REQUIRE_MFA and row.get("mfa_active") != "true":
                  add(user, "missing_mfa", "console password enabled without MFA")

      print("Scanned %d user(s), %d finding(s)" % (scanned, len(findings)))
      print("::" + json.dumps({"outputs": {
          "finding_count": len(findings),
          "scanned_users": scanned,
          "findings": findings[:25],
      }}) + "::")

  - id: credential_gate
    type: io.kestra.plugin.core.flow.If
    description: Alert Slack with every finding, log a quiet all-clear when the
      account is clean.
    condition: "{{ outputs.analyze_credentials.vars.finding_count > 0 }}"
    then:
      - id: alert_slack
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          :key: IAM credential audit found {{ outputs.analyze_credentials.vars.finding_count }} finding(s) across {{ outputs.analyze_credentials.vars.scanned_users }} user(s).
          {% for f in outputs.analyze_credentials.vars.findings %}- {{ f.user }}: {{ f.kind }} - {{ f.detail }}
          {% endfor %}
          Rotate or delete the flagged credentials in the IAM console.
    else:
      - id: log_clean
        type: io.kestra.plugin.core.log.Log
        message: "IAM credential audit clean: {{
          outputs.analyze_credentials.vars.scanned_users }} user(s) scanned, 0
          findings."

outputs:
  - id: finding_count
    type: INT
    value: "{{ outputs.analyze_credentials.vars.finding_count }}"
  - id: scanned_users
    type: INT
    value: "{{ outputs.analyze_credentials.vars.scanned_users }}"
  - id: findings
    type: JSON
    value: "{{ outputs.analyze_credentials.vars.findings | json }}"

errors:
  - id: alert_on_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: ":rotating_light: IAM credential audit {{ flow.id }} FAILED
      (execution {{ execution.id }}). The account was NOT checked - verify IAM
      read permissions (iam:GenerateCredentialReport, iam:GetCredentialReport)
      and the credentials. Error: {{ error.message }}"

triggers:
  - id: weekly
    type: io.kestra.plugin.core.trigger.Schedule
    description: Weekly hygiene audit on Monday morning. Enable when ready.
    cron: "0 7 * * 1"
    disabled: true

  - id: on_demand
    type: io.kestra.plugin.core.trigger.Webhook
    description: Audit after an offboarding or incident. Replace the key with your
      own value.
    key: replace-with-a-random-key