ECS任务中通过Boto3/Botocore代入IAM角色免本地凭证访问DynamoDB咨询
无本地AWS凭证的botocore临时会话实现DynamoDB访问
本方案全程不读取~/.aws/credentials路径下的本地长期凭证,基于STS角色代入获取临时访问凭证,适配生产环境部署要求,不会因凭证硬编码/本地存储触发AWS凭证扫描风险。
前置准备
- 安装依赖包:执行命令
pip install botocore - 提前获取目标IAM角色的ARN,格式参考
arn:aws-cn:iam::123456789012:role/YourDynamoDBAccessRole - 给服务部署资源(EC2实例、ECS任务、EKS服务账户、Lambda执行角色)绑定基础IAM身份,确保该基础身份拥有目标角色的
sts:AssumeRole权限,同时目标角色配置对应信任关系,允许该基础身份代入,并按需分配DynamoDB最小操作权限。
实现逻辑说明
初始化botocore会话时,显式从凭证提供链中移除本地共享凭证文件、本地AWS配置文件的读取模块,从根源上跳过~/.aws/credentials的加载流程,仅从运行时云环境身份(实例/任务/Pod自带的IAM角色)、临时注入的环境变量获取基础身份,再通过STS代入目标角色拿到短期临时凭证,最终初始化DynamoDB客户端。
完整示例代码
import botocore.session from botocore.config import Config from botocore.exceptions import ClientError def init_dynamodb_client( target_role_arn: str, region: str = "cn-north-1", session_duration: int = 3600, session_name: str = "prod-dynamodb-access" ): # 初始化自定义botocore会话 boto_session = botocore.session.Session() # 调整凭证提供链,移除本地凭证文件读取逻辑 credential_providers = boto_session.get_component("credential_provider") credential_providers.remove("shared-credentials-file") # 禁用~/.aws/credentials读取 credential_providers.remove("config-file") # 禁用~/.aws/config读取 # 初始化STS客户端 base_config = Config( region_name=region, signature_version="v4", retries={"max_attempts": 3, "mode": "standard"} ) sts_client = boto_session.create_client("sts", config=base_config) # 代入目标角色获取临时凭证 try: assume_role_resp = sts_client.assume_role( RoleArn=target_role_arn, RoleSessionName=session_name, DurationSeconds=session_duration ) temp_creds = assume_role_resp["Credentials"] except ClientError as e: raise RuntimeError(f"STS角色代入失败: {e.response['Error']['Message']}") from e # 基于临时凭证初始化DynamoDB客户端 dynamodb_client = boto_session.create_client( "dynamodb", aws_access_key_id=temp_creds["AccessKeyId"], aws_secret_access_key=temp_creds["SecretAccessKey"], aws_session_token=temp_creds["SessionToken"], config=base_config ) return dynamodb_client if __name__ == "__main__": # 替换为实际目标角色ARN TARGET_ROLE_ARN = "arn:aws-cn:iam::123456789012:role/YourDynamoDBAccessRole" ddb_client = init_dynamodb_client(target_role_arn=TARGET_ROLE_ARN) # 测试连通性 try: table_list = ddb_client.list_tables()["TableNames"] print(f"DynamoDB连接成功,当前账号下表清单:{table_list}") except ClientError as e: print(f"DynamoDB调用失败: {e.response['Error']['Message']}")
生产落地注意事项
- 临时凭证有效期可按需配置,取值范围为900秒(15分钟)到43200秒(12小时),凭证到期前需要重新调用STS接口刷新,避免服务调用中断。
- 全程不要在代码、配置文件、环境变量中写入长期AK/SK,运行时身份统一使用云资源绑定的IAM角色,实现凭证全托管,无人工维护成本。
- IAM权限遵循最小化原则:基础身份仅保留目标角色的
sts:AssumeRole权限,目标角色仅分配业务必需的DynamoDB操作权限,禁止配置全资源全操作的宽松权限。 - 本地开发测试时,可通过临时凭证生成工具获取短期开发凭证注入环境变量,代码会自动读取环境变量中的临时凭证,不需要写入本地
~/.aws/credentials文件。
内容的提问来源于stack exchange,提问作者karkir subu
相关产品推荐
相关产品推荐

