aws-secrets-manager – skill DevOps & Hạ tầng cho agent AI

AWS Secrets Manager

Securely store, manage, and rotate secrets in AWS.

When to Use This Skill

Use this skill when:

• Storing database credentials, API keys, or tokens in AWS
• Implementing automatic credential rotation for RDS or other services
• Replacing hardcoded secrets in application code or config files
• Integrating secrets into ECS, EKS, or Lambda workloads
• Meeting compliance requirements for secret management and rotation

Prerequisites

• AWS account with appropriate IAM permissions
• AWS CLI v2 installed and configured
• IAM policy limited to needed Secrets Manager actions and exact secret ARNs, plus required KMS permissions
• For rotation: Lambda execution role and VPC access to target services
• Python 3.9+ with boto3 for SDK examples

Secret Creation and Management

Create the referenced files privately (mode0600) using a secret-input/editor workflow, keep them out of Git and logs, and remove them according to the local secret-handling policy after use. Never put secret values directly into shell history or process arguments. get-secret-value prints plaintext: use only when explicitly needed in a private terminal; prefer consuming values inside the application. Confirm account/region/exact ARN before writes, rotation or scheduled deletion. Updating SecretString alone does not change the actual database password; use tested rotation for credentials.

# Create a secret with JSON structure
aws secretsmanager create-secret \
  --name myapp/production/database \
  --description "Production database credentials" \
  --secret-string file://database-secret.json \
  --tags '[{"Key":"Environment","Value":"production"},{"Key":"Team","Value":"platform"}]'
# Create a secret with KMS encryption (custom key)
aws secretsmanager create-secret \
  --name myapp/production/api-key \
  --description "Third-party API key" \
  --secret-string file://api-key.txt \
  --kms-key-id alias/secrets-key
# Create a binary secret (certificates, keys)
aws secretsmanager create-secret \
  --name myapp/production/tls-cert \
  --secret-binary fileb://server.pfx
# Get secret value
aws secretsmanager get-secret-value \
  --secret-id myapp/production/database \
  --query 'SecretString' --output text | jq .
# Get a specific version
aws secretsmanager get-secret-value \
  --secret-id myapp/production/database \
  --version-stage AWSPREVIOUS
# Update secret value
aws secretsmanager put-secret-value \
  --secret-id myapp/production/database \
  --secret-string file://database-secret-updated.json
# List all secrets
aws secretsmanager list-secrets \
  --filters Key=name,Values=myapp/production
# Delete secret (with recovery window)
aws secretsmanager delete-secret \
  --secret-id myapp/production/old-key \
  --recovery-window-in-days 7
# Restore a deleted secret
aws secretsmanager restore-secret \
  --secret-id myapp/production/old-key
# Tag a secret
aws secretsmanager tag-resource \
  --secret-id myapp/production/database \
  --tags '[{"Key":"RotationEnabled","Value":"true"}]'

Automatic Rotation

Enable Rotation

# Enable rotation with an existing Lambda function
aws secretsmanager rotate-secret \
  --secret-id myapp/production/database \
  --rotation-lambda-arn arn:aws:lambda:us-east-1:123456789012:function:SecretsManagerRDSPostgreSQLRotation \
  --rotation-rules '{"ScheduleExpression":"rate(30 days)"}'
# Trigger immediate rotation
aws secretsmanager rotate-secret \
  --secret-id myapp/production/database
# Check rotation status
aws secretsmanager describe-secret \
  --secret-id myapp/production/database \
  --query '{RotationEnabled:RotationEnabled,RotationLambdaARN:RotationLambdaARN,RotationRules:RotationRules,LastRotatedDate:LastRotatedDate}'

Lambda Rotation Function

"""rotation_function.py - Custom rotation Lambda for database credentials."""
import boto3
import json
import logging
import psycopg2
from psycopg2 import sql
logger = logging.getLogger()
logger.setLevel(logging.INFO)
def lambda_handler(event, context):
    """Secrets Manager rotation handler.
    The rotation process has four steps:
    1. createSecret - Generate new secret value
    2. setSecret - Apply the new secret to the target service
    3. testSecret - Verify the new secret works
    4. finishSecret - Mark rotation complete
    """
    secret_arn = event['SecretId']
    token = event['ClientRequestToken']
    step = event['Step']
    client = boto3.client('secretsmanager')
    metadata = client.describe_secret(SecretId=secret_arn)
    if not metadata.get('RotationEnabled'):
        raise ValueError(f"Secret {secret_arn} does not have rotation enabled")
    versions = metadata.get('VersionIdsToStages', {})
    if token not in versions:
        raise ValueError(f"Secret version {token} has no stage for rotation")
    if "AWSCURRENT" in versions[token]:
        return
    if "AWSPENDING" not in versions[token]:
        raise ValueError("Rotation token must carry AWSPENDING")
    if step == "createSecret":
        create_secret(client, secret_arn, token)
    elif step == "setSecret":
        set_secret(client, secret_arn, token)
    elif step == "testSecret":
        test_secret(client, secret_arn, token)
    elif step == "finishSecret":
        finish_secret(client, secret_arn, token)
    else:
        raise ValueError(f"Invalid step: {step}")


def create_secret(client, secret_arn, token):
    """Generate once per pending token; retries reuse the same value."""
    try:
        client.get_secret_value(SecretId=secret_arn, VersionId=token, VersionStage="AWSPENDING")
        return
    except client.exceptions.ResourceNotFoundException:
        pass
    current = client.get_secret_value(
        SecretId=secret_arn, VersionStage="AWSCURRENT"
    )
    current_dict = json.loads(current['SecretString'])

    new_password = client.get_random_password(
        PasswordLength=32,
        ExcludeCharacters='/@"\\',
        RequireEachIncludedType=True,
    )['RandomPassword']

    current_dict['password'] = new_password
    client.put_secret_value(
        SecretId=secret_arn,
        ClientRequestToken=token,
        SecretString=json.dumps(current_dict),
        VersionStages=['AWSPENDING'],
    )
    logger.info(f"createSecret: New secret version created for {secret_arn}")


def set_secret(client, secret_arn, token):
    """Apply the new secret to the target database."""
    pending = client.get_secret_value(
        SecretId=secret_arn, VersionId=token, VersionStage="AWSPENDING"
    )
    pending_dict = json.loads(pending['SecretString'])

    current = client.get_secret_value(
        SecretId=secret_arn, VersionStage="AWSCURRENT"
    )
    current_dict = json.loads(current['SecretString'])

    if any(current_dict.get(k) != pending_dict.get(k) for k in ("host", "username", "port", "dbname")):
        raise ValueError("Current/pending target mismatch")
    # A production handler must first test pending credentials: if they already work,
    # return successfully (setSecret retry after the database password changed).
    try:
        pending_conn = psycopg2.connect(host=pending_dict["host"], user=pending_dict["username"],
            password=pending_dict["password"], dbname=pending_dict.get("dbname", "postgres"),
            port=pending_dict.get("port", 5432), sslmode="verify-full", connect_timeout=10)
        pending_conn.close()
        return
    except psycopg2.OperationalError:
        pass
    conn = psycopg2.connect(
        sslmode="verify-full", connect_timeout=10,
        host=current_dict['host'],
        port=current_dict.get('port', 5432),
        user=current_dict['username'],
        password=current_dict['password'],
        dbname=current_dict.get('dbname', 'postgres'),
    )
    conn.autocommit = True
    with conn.cursor() as cur:
        cur.execute(
            sql.SQL("ALTER USER {} WITH PASSWORD %s").format(sql.Identifier(pending_dict["username"])),
            (pending_dict["password"],),
        )
    conn.close()
    logger.info(f"setSecret: Password updated in database for {secret_arn}")


def test_secret(client, secret_arn, token):
    """Verify the new secret works."""
    pending = client.get_secret_value(
        SecretId=secret_arn, VersionId=token, VersionStage="AWSPENDING"
    )
    pending_dict = json.loads(pending['SecretString'])

    conn = psycopg2.connect(
        sslmode="verify-full", connect_timeout=10,
        host=pending_dict['host'],
        port=pending_dict.get('port', 5432),
        user=pending_dict['username'],
        password=pending_dict['password'],
        dbname=pending_dict.get('dbname', 'postgres'),
    )
    conn.close()
    logger.info(f"testSecret: New credentials verified for {secret_arn}")


def finish_secret(client, secret_arn, token):
    """Finalize the rotation by updating version stages."""
    metadata = client.describe_secret(SecretId=secret_arn)
    versions = metadata.get('VersionIdsToStages', {})

    current_version = None
    for version_id, sta

… (nội dung đầy đủ có trong gói tải về)