IAM Access Key Rotation — Auto-Rotate Keys Older than 90 Days

import boto3
from datetime import datetime, timezone, timedelta
import logging
import json

logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")


def rotate_old_access_keys(
    dry_run: bool = True,
    max_age_days: int = 90,
    region: str = "us-east-1",
) -> list[dict]:
    """
    Scans all IAM users and rotates access keys older than max_age_days.

    dry_run=True (default) — prints what would happen but makes no changes.
    Always test with dry_run=True before running against production.
    """
    iam = boto3.client("iam")
    ses = boto3.client("ses", region_name=region)
    sm  = boto3.client("secretsmanager", region_name=region)
    report = []

    # ── List all IAM users (paginated) ────────────────────────────
    # list_users() is paginated — get_paginator handles NextToken automatically.
    # IAM is a global service (no region needed) but the client still works.
    users = []
    paginator = iam.get_paginator("list_users")
    for page in paginator.paginate():
        users.extend(page["Users"])
    logger.info(f"Found {len(users)} IAM users to check")

    threshold = datetime.now(timezone.utc) - timedelta(days=max_age_days)

    for user in users:
        username = user["UserName"]

        # ── List access keys for this user ─────────────────────────
        # list_access_keys() returns AccessKeyMetadata — the list of
        # key IDs, statuses, and creation dates. It does NOT return
        # the secret — secrets are shown only at creation time.
        keys = iam.list_access_keys(UserName=username)["AccessKeyMetadata"]

        for key in keys:
            if key["Status"] != "Active":
                continue   # Skip already-inactive keys

            key_age = datetime.now(timezone.utc) - key["CreateDate"]

            if key["CreateDate"] < threshold:
                logger.info(
                    f"Key {key['AccessKeyId']} for {username} is "
                    f"{key_age.days} days old (> {max_age_days}) → rotating"
                )
                action_taken = "DRY_RUN"

                if not dry_run:
                    action_taken = _rotate_key(iam, ses, sm, username, key)

                report.append({
                    "username":   username,
                    "old_key_id": key["AccessKeyId"],
                    "key_age_days": key_age.days,
                    "action":     action_taken,
                })

    # Summary
    logger.info(f"Rotation complete. {len(report)} key(s) processed.")
    return report


def _rotate_key(iam, ses, sm, username: str, old_key: dict) -> str:
    """
    Performs the actual key rotation:
    1. Create new key (AWS allows max 2 active keys per user)
    2. Store new secret in Secrets Manager (NOT in email)
    3. Deactivate old key (don't delete yet)
    4. Notify user via SES
    """
    old_key_id = old_key["AccessKeyId"]

    try:
        # Step 1 — Create new key
        # create_access_key() is the ONLY time the SecretAccessKey is visible.
        # Store it immediately in Secrets Manager — it cannot be retrieved later.
        new_key_response = iam.create_access_key(UserName=username)
        new_key = new_key_response["AccessKey"]
        new_key_id     = new_key["AccessKeyId"]
        new_key_secret = new_key["SecretAccessKey"]

        logger.info(f"  Created new key {new_key_id} for {username}")

        # Step 2 — Store in Secrets Manager (never send raw secret in email)
        secret_name = f"/iam/keys/{username}"
        secret_value = json.dumps({
            "AccessKeyId":     new_key_id,
            "SecretAccessKey": new_key_secret,
            "RotatedAt":       datetime.utcnow().isoformat() + "Z",
            "OldKeyId":        old_key_id,
        })

        try:
            # update_secret if it already exists
            sm.put_secret_value(SecretId=secret_name, SecretString=secret_value)
        except sm.exceptions.ResourceNotFoundException:
            # create_secret on first rotation for this user
            sm.create_secret(Name=secret_name, SecretString=secret_value)

        logger.info(f"  New credentials stored at Secrets Manager: {secret_name}")

        # Step 3 — Deactivate (not delete) the old key
        # update_access_key() with Status="Inactive" disables the key
        # without permanently removing it — giving a grace period to update.
        iam.update_access_key(
            UserName=username,
            AccessKeyId=old_key_id,
            Status="Inactive",
        )
        logger.info(f"  Deactivated old key {old_key_id} for {username}")

        # Step 4 — Notify user via SES
        _notify_user(ses, username, old_key_id, new_key_id, secret_name)

        return "ROTATED"

    except Exception as e:
        logger.error(f"Failed to rotate key for {username}: {e}")
        return f"FAILED: {e}"


def _notify_user(ses, username: str, old_key_id: str, new_key_id: str, secret_name: str) -> None:
    """
    Send the user a notification. We send the SECRET MANAGER PATH,
    not the raw secret, so it never appears in email logs or archives.
    """
    ses.send_email(
        Source="[email protected]",
        Destination={"ToAddresses": [f"{username}@company.com"]},
        Message={
            "Subject": {"Data": "🔑 Your AWS Access Key Has Been Rotated"},
            "Body": {
                "Text": {
                    "Data": f"""
Hi {username},

Your AWS access key {old_key_id} was {90}+ days old and has been rotated
as per our security policy (CIS AWS Benchmark 1.14).

Your new Access Key ID: {new_key_id}
Your new Secret:        Stored in AWS Secrets Manager at {secret_name}

Steps to update your credentials:
  1. aws secretsmanager get-secret-value --secret-id {secret_name}
  2. Update ~/.aws/credentials with the new key
  3. Update any CI/CD pipelines that use the old key
  4. The old key ({old_key_id}) will be DELETED in 7 days

Need help? Contact [email protected]
                    """
                }
            },
        },
    )


def delete_inactive_keys(max_age_days: int = 7, dry_run: bool = True) -> None:
    """
    Second-pass script: run 7 days after rotation to delete the old inactive keys.
    delete_access_key() permanently and irreversibly removes the key.
    """
    iam = boto3.client("iam")
    threshold = datetime.now(timezone.utc) - timedelta(days=max_age_days)

    paginator = iam.get_paginator("list_users")
    for page in paginator.paginate():
        for user in page["Users"]:
            username = user["UserName"]
            keys = iam.list_access_keys(UserName=username)["AccessKeyMetadata"]
            for key in keys:
                if key["Status"] == "Inactive" and key["CreateDate"] < threshold:
                    if dry_run:
                        print(f"[DRY-RUN] Would delete {key['AccessKeyId']} for {username}")
                    else:
                        iam.delete_access_key(
                            UserName=username, AccessKeyId=key["AccessKeyId"]
                        )
                        logger.info(f"Deleted inactive key {key['AccessKeyId']} for {username}")


if __name__ == "__main__":
    # Phase 1 — Preview
    report = rotate_old_access_keys(dry_run=True, max_age_days=90)
    print(f"\nDry-run complete: {len(report)} key(s) would be rotated")

    # Phase 2 — Apply (uncomment when ready)
    # rotate_old_access_keys(dry_run=False, max_age_days=90)

    # Phase 3 — Delete old inactive keys (run 7 days after Phase 2)
    # delete_inactive_keys(max_age_days=7, dry_run=False)

iam = boto3.client("iam")
ses = boto3.client("ses", region_name=region)
sm  = boto3.client("secretsmanager", region_name=region)

paginator = iam.get_paginator("list_users")
for page in paginator.paginate():
    users.extend(page["Users"])

threshold = datetime.now(timezone.utc) - timedelta(days=max_age_days)

keys = iam.list_access_keys(UserName=username)["AccessKeyMetadata"]

if key["Status"] != "Active":
    continue

key_age = datetime.now(timezone.utc) - key["CreateDate"]
if key["CreateDate"] < threshold:

new_key_response = iam.create_access_key(UserName=username)
new_key = new_key_response["AccessKey"]
new_key_id     = new_key["AccessKeyId"]
new_key_secret = new_key["SecretAccessKey"]

secret_value = json.dumps({
    "AccessKeyId": new_key_id,
    "SecretAccessKey": new_key_secret,
    "RotatedAt": datetime.utcnow().isoformat() + "Z",
    "OldKeyId": old_key_id,
})
sm.put_secret_value(SecretId=secret_name, SecretString=secret_value)

iam.update_access_key(
    UserName=username,
    AccessKeyId=old_key_id,
    Status="Inactive",
)

for key in keys:
    if key["Status"] == "Inactive" and key["CreateDate"] < threshold:
        iam.delete_access_key(UserName=username, AccessKeyId=key["AccessKeyId"])