aws-secrets-manager

AWS Secrets Manager

在 AWS 中安全地存储、管理和轮换机密。

何时使用此技能

适用于以下情况：

• 在 AWS 中存储数据库凭据、API 密钥或令牌
• 为 RDS 或其他服务实现自动凭据轮换
• 替换应用程序代码或配置文件中硬编码的机密
• 将机密集成到 ECS、EKS 或 Lambda 工作负载中
• 满足机密管理和轮换方面的合规要求

前提条件

• 具有适当 IAM 权限的 AWS 账户
• 已安装并配置 AWS CLI v2
• 允许执行 secretsmanager:* 操作的 IAM 策略（或范围受限的权限）
• 对于轮换：Lambda 执行角色，以及对目标服务的 VPC 访问权限
• 用于 SDK 示例的 Python 3.9+ 和 boto3

机密的创建与管理

# Create a secret with JSON structure
aws secretsmanager create-secret \
  --name myapp/production/database \
  --description "Production database credentials" \
  --secret-string '{"username":"dbadmin","password":"S3cur3P@ssw0rd!","engine":"postgres","host":"db.internal.example.com","port":5432,"dbname":"myapp"}' \
  --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 "ak_live_xxxxxxxxxxxx" \
  --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 '{"username":"dbadmin","password":"N3wS3cur3P@ss!","engine":"postgres","host":"db.internal.example.com","port":5432,"dbname":"myapp"}'

# 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"}]'

自动轮换

启用轮换

# 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 '{"AutomaticallyAfterDays":30,"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.py - Custom rotation Lambda for database credentials."""

import boto3
import json
import logging
import psycopg2

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 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 a new secret value."""
    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'])

    conn = psycopg2.connect(
        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(
            "ALTER USER %s WITH PASSWORD %s",
            (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(
        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, stages in versions.items():
        if "AWSCURRENT" in stages:
            if version_id == token:
                logger.info("finishSecret: Version already marked AWSCURRENT")
                return
            current_version = version_id
            break

    client.update_secret_version_stage(
        SecretId=secret_arn,
        VersionStage="AWSCURRENT",
        MoveToVersionId=token,
        RemoveFromVersionId=current_version,
    )
    logger.info(f"finishSecret: Rotation complete for {secret_arn}")

轮换 Lambda Terraform 配置

resource "aws_lambda_function" "rotation" {
  filename         = "rotation_function.zip"
  function_name    = "secrets-rotation-postgresql"
  role             = aws_iam_role.rotation.arn
  handler          = "rotation_function.lambda_handler"
  runtime          = "python3.11"
  timeout          = 60

  vpc_config {
    subnet_ids         = var.private_subnet_ids
    security_group_ids = [aws_security_group.rotation.id]
  }

  environment {
    variables = {
      SECRETS_MANAGER_ENDPOINT = "https://secretsmanager.${var.region}.amazonaws.com"
    }
  }
}

resource "aws_lambda_permission" "secrets_manager" {
  action        = "lambda:InvokeFunction"
  function_name = aws_lambda_function.rotation.function_name
  principal     = "secretsmanager.amazonaws.com"
  statement_id  = "AllowSecretsManager"
}

resource "aws_secretsmanager_secret_rotation" "db" {
  secret_id           = aws_secretsmanager_secret.db.id
  rotation_lambda_arn = aws_lambda_function.rotation.arn
  rotation_rules {
    automatically_after_days = 30
  }
}

应用程序集成

Python SDK

import boto3
import json
from functools import lru_cache

def get_secret(secret_name: str, region: str = "us-east-1") -> dict:
    """Retrieve and parse a secret from AWS Secrets Manager."""
    client = boto3.client("secretsmanager", region_name=region)
    response = client.get_secret_value(SecretId=secret_name)
    if "SecretString" in response:
        return json.loads(response["SecretString"])
    else:
        import base64
        return base64.b64decode(response["SecretBinary"])

@lru_cache(maxsize=32)
def get_cached_secret(secret_name: str) -> dict:
    """Cached secret retrieval. Clear cache on rotation events."""
    return get_secret(secret_name)

# Usage
creds = get_secret("myapp/production/database")
connection_string = (
    f"postgresql://{creds['username']}:{creds['password']}"
    f"@{creds['host']}:{creds['port']}/{creds['dbname']}"
)

ECS 任务定义

{
  "containerDefinitions": [
    {
      "name": "myapp",
      "image": "ghcr.io/acme/myapp:v1.0.0",
      "secrets": [
        {
          "name": "DB_USERNAME",
          "valueFrom": "arn:aws:secretsmanager:us-east-1:123456789:secret:myapp/production/database:username::"
        },
        {
          "name": "DB_PASSWORD",
          "valueFrom": "arn:aws:secretsmanager:us-east-1:123456789:secret:myapp/production/database:password::"
        },
        {
          "name": "API_KEY",
          "valueFrom": "arn:aws:secretsmanager:us-east-1:123456789:secret:myapp/production/api-key"
        }
      ]
    }
  ],
  "executionRoleArn": "arn:aws:iam::123456789:role/ecsTaskExecutionRole"
}

使用 External Secrets Operator 的 EKS

apiVersion: external-secrets.io/v1beta1
kind: SecretStore
metadata:
  name: aws-secrets-manager
  namespace: production
spec:
  provider:
    aws:
      service: SecretsManager
      region: us-east-1
      auth:
        jwt:
          serviceAccountRef:
            name: external-secrets-sa
---
apiVersion: external-secrets.io/v1beta1
kind: ExternalSecret
metadata:
  name: db-credentials
  namespace: production
spec:
  refreshInterval: 1h
  secretStoreRef:
    name: aws-secrets-manager
    kind: SecretStore
  target:
    name: db-credentials
    creationPolicy: Owner
  data:
    - secretKey: username
      remoteRef:
        key: myapp/production/database
        property: username
    - secretKey: password
      remoteRef:
        key: myapp/production/database
        property: password

基于资源的策略

# Restrict secret access to specific roles
aws secretsmanager put-resource-policy \
  --secret-id myapp/production/database \
  --resource-policy '{
    "Version": "2012-10-17",
    "Statement": [
      {
        "Effect": "Allow",
        "Principal": {
          "AWS": [
            "arn:aws:iam::123456789:role/myapp-ecs-task-role",
            "arn:aws:iam::123456789:role/myapp-lambda-role"
          ]
        },
        "Action": [
          "secretsmanager:GetSecretValue"
        ],
        "Resource": "*",
        "Condition": {
          "StringEquals": {
            "aws:RequestedRegion": "us-east-1"
          }
        }
      },
      {
        "Effect": "Deny",
        "Principal": "*",
        "Action": "secretsmanager:GetSecretValue",
        "Resource": "*",
        "Condition": {
          "StringNotEquals": {
            "aws:PrincipalAccount": "123456789012"
          }
        }
      }
    ]
  }'

故障排除

问题	原因	解决方案
调用 GetSecretValue 时出现 AccessDeniedException	IAM 策略缺少权限	为角色添加 secretsmanager:GetSecretValue；检查基于资源的策略
轮换因 Lambda 超时而失败	Lambda 无法连接数据库	确保 Lambda 位于同一 VPC 中，且有通往数据库的路由；检查安全组
轮换后机密值为空	createSecret 步骤失败	检查 Lambda CloudWatch 日志；确认随机密码生成正常
ECS 容器无法启动	机密 ARN 格式不正确	使用包含 :: 的完整 ARN 提取 JSON 键；确认机密存在
轮换后应用程序仍使用旧凭据	客户端缓存了过期值	在轮换时实现缓存失效；缩短缓存 TTL
轮换 Lambda 权限错误	缺少 lambda:InvokeFunction 权限	为主体 secretsmanager.amazonaws.com 添加 aws_lambda_permission
KMS 解密失败	机密的 KMS 密钥策略未包含该角色	将访问该机密的角色添加到 KMS 密钥策略的 kms:Decrypt 主体中

最佳实践

• 启用自动轮换，间隔至少为 30 天
• 除 IAM 策略外，还应使用基于资源的策略（纵深防御）
• 使用客户管理的 KMS 密钥加密机密（不要使用默认密钥）
• 实施最小权限访问控制（仅向需要各机密的角色授予访问权限）
• 使用机密版本控制，以便在轮换出现问题时安全回滚
• 使用 CloudTrail 监控机密访问，并对异常模式发出警报
• 按层级组织机密名称：{app}/{env}/{secret-type}
• 切勿记录机密值；仅记录机密 ARN 和访问元数据
• 在生产环境启用轮换前，先在预发布环境测试
• 设置 CloudWatch 警报，以便在轮换失败时收到通知

相关技能

• hashicorp-vault - 多云机密管理
• aws-iam - IAM 策略
• azure-keyvault - Azure 机密管理
• gcp-secret-manager - GCP 机密管理