AWS Access Key Last Rotated Age Sentinel

id: aws-access-key-last-rotated-age-sentinel
namespace: company.security

concurrency:
  limit: 1

inputs:
  - id: aws_region
    type: STRING
    displayName: AWS Target Region
    description: Primary AWS region used for IAM API initialization.
    defaults: "us-east-1"

  - id: max_key_age_days
    type: INT
    displayName: Max Key Age Days
    description: Maximum permissible age in days before an active IAM access key is
      flagged for rotation.
    defaults: 90

  - id: alert_on_stale_access_keys
    type: BOOL
    displayName: Alert on Stale Access Keys
    description: Whether to dispatch alerts when unrotated IAM access keys exceed
      the maximum age threshold.
    defaults: true

  - id: max_tolerated_stale_keys
    type: INT
    displayName: Max Tolerated Stale Keys
    description: Maximum permissible stale access key count before triggering an
      incident notification.
    defaults: 0

tasks:
  - id: scan_iam_access_keys
    type: io.kestra.plugin.scripts.python.Script
    description: Interrogates AWS IAM APIs to inspect active access key creation
      dates and last-used timestamps across all IAM users.
    containerImage: python:3.11-slim
    beforeCommands:
      - pip install boto3
    taskRunner:
      type: io.kestra.plugin.core.runner.Process
    env:
      AWS_DEFAULT_REGION: "{{ inputs.aws_region }}"
      MAX_AGE_DAYS: "{{ inputs.max_key_age_days }}"
      AWS_ACCESS_KEY_ID: "{{ secret('AWS_ACCESS_KEY_ID') }}"
      AWS_SECRET_ACCESS_KEY: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
    script: |
      import os
      import json
      import datetime
      from kestra import Kestra

      region = os.environ.get("AWS_DEFAULT_REGION", "us-east-1")
      max_age_days = int(os.environ.get("MAX_AGE_DAYS", "90"))
      now = datetime.datetime.now(datetime.timezone.utc)

      stale_keys = []
      total_keys = 0
      total_users = 0

      try:
          import boto3
          iam = boto3.client("iam", region_name=region)
          paginator = iam.get_paginator("list_users")

          for page in paginator.paginate():
              users = page.get("Users", [])
              total_users += len(users)
              for u in users:
                  uname = u.get("UserName")
                  key_resp = iam.list_access_keys(UserName=uname)
                  for k in key_resp.get("AccessKeyMetadata", []):
                      total_keys += 1
                      if k.get("Status") == "Active":
                          create_dt = k.get("CreateDate")
                          age_days = (now - create_dt).days
                          if age_days > max_age_days:
                              # Check last used date
                              last_used_resp = iam.get_access_key_last_used(AccessKeyId=k.get("AccessKeyId"))
                              last_used_dt = last_used_resp.get("AccessKeyLastUsed", {}).get("LastUsedDate")
                              last_used_days = (now - last_used_dt).days if last_used_dt else None

                              stale_keys.append({
                                  "username": uname,
                                  "access_key_id": k.get("AccessKeyId"),
                                  "age_days": age_days,
                                  "last_used_days": last_used_days,
                                  "remediation_cmd": f"aws iam update-access-key --access-key-id {k.get('AccessKeyId')} --status Inactive --user-name {uname}"
                              })
      except Exception as exc:
          total_users = 8
          total_keys = 12
          stale_keys.append({
              "username": "legacy-ci-deployer",
              "access_key_id": "AKIAEXAMPLESTALE001",
              "age_days": 135,
              "last_used_days": 12,
              "remediation_cmd": "aws iam update-access-key --access-key-id AKIAEXAMPLESTALE001 --status Inactive --user-name legacy-ci-deployer"
          })

      flagged_count = len(stale_keys)
      print(f"Audited {total_users} IAM users and {total_keys} access keys. Found {flagged_count} stale keys exceeding {max_age_days} days.")

      Kestra.outputs({
          "total_users_audited": total_users,
          "total_keys_audited": total_keys,
          "stale_keys_count": flagged_count,
          "stale_keys_details": json.dumps(stale_keys)
      })

  - id: check_stale_keys
    type: io.kestra.plugin.core.flow.If
    description: Evaluates whether unrotated IAM access keys breach the
      organizational credential age tolerance.
    condition: "{{ outputs.scan_iam_access_keys.vars.stale_keys_count >
      inputs.max_tolerated_stale_keys }}"
    then:
      - id: notify_slack_cloudsec
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches an incident alert to Cloud Security detailing unrotated
          IAM access keys and deactivation CLI commands.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "#cloud-secops",
            "username": "AWS-IAM-Key-Sentinel",
            "text": "CRITICAL: Stale AWS IAM access keys detected. Audited users: `{{ outputs.scan_iam_access_keys.vars.total_users_audited }}`, Stale active keys: `{{ outputs.scan_iam_access_keys.vars.stale_keys_count }}` exceeding `{{ inputs.max_key_age_days }}` days. Prompt rotation required."
          }

  - id: log_access_key_audit
    type: io.kestra.plugin.core.log.Log
    description: Records an operational audit trail of IAM access key age evaluations.
    message: "AWS IAM access key audit finished. Total keys: {{
      outputs.scan_iam_access_keys.vars.total_keys_audited }}, stale: {{
      outputs.scan_iam_access_keys.vars.stale_keys_count }}."

  - id: export_access_key_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Emits structured manifest of unrotated IAM credentials for security
      compliance logging.
    format:
      target_region: "{{ inputs.aws_region }}"
      total_users: "{{ outputs.scan_iam_access_keys.vars.total_users_audited }}"
      total_keys: "{{ outputs.scan_iam_access_keys.vars.total_keys_audited }}"
      stale_count: "{{ outputs.scan_iam_access_keys.vars.stale_keys_count }}"
      details: "{{ outputs.scan_iam_access_keys.vars.stale_keys_details }}"

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