检索存储在 Python Web 应用程序 Secrets Manager 中的轮换 RDS 凭证的推荐方法？

Question

我有一个

flask

应用程序在

ECS

上运行，其中

EC2 instance type

可以访问

RDS

数据库。我将数据库凭据存储在

Secrets Manager

中。我想启用

Secrets Manager

提供的内置旋转，但我不确定实现它的最佳方法。

由于成本原因，我不想每次需要提出请求时都轮询

Secrets Manager

来获取凭据。我也无法通过一次调用

Secrets Manager

来硬编码凭证以用于应用程序的生命周期，因为凭证最终将被轮换。

我的想法是使用缓存机制和后台任务，以固定的时间间隔轮询

Secrets Manager

以获取

Secrets Manager

凭证的更改。不过，这对我来说似乎仍然代价高昂，因为你不断地进行投票

Secrets Manager

。另一个问题是，如果凭据在固定间隔之间轮换，那么它们将无效，直到后台任务再次运行。然后我还必须担心缓存中凭证的安全。

我的另一个想法是在

Secrets Manager

轮换凭证时绑定一个 CloudWatch 事件。但是，我很难在

flask

上运行的

ECS

应用程序中触发事件来更新

Secrets Manager

的凭据。

Answer 1

使用 CloudWatch 事件在 Secrets Manager 中轮换凭证时触发凭证更新似乎是您的情况的解决方案。

您可以创建一个 Lambda 函数来侦听 Secrets Manager 轮换事件，然后更新 Flask 应用程序中的缓存凭证。例如，您可以使用 Redis 缓存应用程序的凭据。

您的 Lambda 中将包含以下内容：

import boto3
import redis
import json


def lambda_handler(event, context):
    secret_arn = event['detail']['SecretId']
    if 'previousSecretVersionId' in event['detail']:
        previous_secret_version_id = event['detail']['previousSecretVersionId']
    else:
        previous_secret_version_id = None

    if 'versionId' in event['detail']:
        current_secret_version_id = event['detail']['versionId']
    else:
        current_secret_version_id = None

    if previous_secret_version_id and current_secret_version_id:
        client = boto3.client('secretsmanager')
        response = client.get_secret_value(SecretId=secret_arn, VersionId=current_secret_version_id)
        secrets = json.loads(response['SecretString'])
        
        # Update the cached credentials here
        redis_client = redis.Redis(host='<redis_endpoint>', port=6379, db=0, password='<redis_password>')
        redis_client.set('DB_USERNAME', secrets['username'])
        redis_client.set('DB_PASSWORD', secrets['password'])
    else:
        print('Rotation event is missing `previousSecretVersionId` or current `versionId` ID.')

更新： 您不应创建 Lambda 来侦听凭证轮换，而应创建一个根据特定计划处理轮换的 Lambda。在 Flask 应用程序中，您应该从 AWS

SecretCache

(https://github.com/aws/aws-secretsmanager-caching-python) 检索凭证，以减少 Secrets Manager API 请求的成本：

import botocore 
import botocore.session 
from aws_secretsmanager_caching import SecretCache, SecretCacheConfig 

client = botocore.session.get_session().create_client('secretsmanager')
cache_config = SecretCacheConfig()
cache = SecretCache(config = cache_config, client = client)

secret = cache.get_secret_string('mysecret')

旋转 Lambda 将执行以下操作：

1. 创建新版本的秘密
2. 更改数据库或服务中的凭据
3. 测试新的秘密版本
4. 完成旋转

这里用于更新 RDS PostgreSQL 用户凭证的 Lambda 函数模板。

参考资料：

• https://docs.aws.amazon.com/secretsmanager/latest/userguide/rotating-secrets.html
• https://docs.aws.amazon.com/secretsmanager/latest/userguide/retrieve-secrets.html
• https://docs.aws.amazon.com/secretsmanager/latest/userguide/retriving-secrets_cache-python.html

Answer 2

1. 为 RDS 密钥启用密码轮换
2. 创建 EventBridge 规则以捕获 RotationSucceed 事件并调用 AWS lambda
3. 调用的AWS lambda将调用ECS强制新部署，以便Flask应用程序容器可以通过ENV读取新的RDS密钥