如何用Python Boto3分别认证AWS两账号的Secret Manager与Athena
问题
本地开发环境中,我想用Boto3给Python应用做顺序独立认证,实现从aws-account-1的AWS Secrets Manager拉取密钥,同时在aws-account-2的AWS Athena执行查询。
当前遇到的问题:
- 执行
aws sso login --profile aws-account-1认证账号1后,Secret Manager客户端能正常拉取密钥,但Athena查询因未认证到正确配置文件失败; - 执行
aws sso login --profile aws-account-2认证账号2后,Athena查询正常执行,但拉取密钥因权限不足失败。
不想用跨账号角色切换(assumeRole)的方案,目前代码大致如下:
def __local_authentication(self, account, resource): try: session = boto3.Session(profile_name=account) client = self.__session.client(resource, region_name = self.__region) return (session, client) except Exception as e: raise "Authentication error."
# 拉取密钥的代码 session, client = __local_authentication(self,'aws-account-1', 'secretmanager') response = client.get_secret_value(SecretId=secret_name)
# Athena查询代码 session, client = __local_authentication(self,'aws-account-2', 'athena') resultset = wr.athena.query("", session=session)
请问有没有不用assumeRole的方法,能让代码同时完成拉取Secret Manager密钥和执行Athena查询的认证?
解决方案
以下是几种无需使用assumeRole的可行方案:
1. 同时保持两个SSO会话有效
AWS SSO支持同时登录多个配置文件,只需分别执行两次登录命令:
aws sso login --profile aws-account-1 aws sso login --profile aws-account-2
执行完成后,本地会同时保存两个账号的SSO会话凭证。此时代码可正常通过指定profile_name创建对应账号的Boto3 Session,无需切换会话。
注意修正代码错误:原方法中误用了self.__session.client,应该使用刚创建的session对象,修正后代码:
def __local_authentication(self, account, resource): try: session = boto3.Session(profile_name=account) client = session.client(resource, region_name=self.__region) return (session, client) except Exception as e: raise Exception(f"Authentication error: {str(e)}")
2. 使用独立的凭证文件路径(可选)
如果遇到SSO会话冲突,可为每个账号指定独立的凭证文件:
- 执行登录时分别指定不同的凭证文件:
aws sso login --profile aws-account-1 --credentials-file ~/.aws/credentials-account1 aws sso login --profile aws-account-2 --credentials-file ~/.aws/credentials-account2
- 在创建Boto3 Session时指定对应凭证文件:
import os def __local_authentication(self, account, resource): try: creds_path = "~/.aws/credentials-account1" if account == 'aws-account-1' else "~/.aws/credentials-account2" session = boto3.Session( profile_name=account, credentials_file=os.path.expanduser(creds_path) ) client = session.client(resource, region_name=self.__region) return (session, client) except Exception as e: raise Exception(f"Authentication error: {str(e)}")
3. 利用环境变量临时切换(适合脚本场景)
在代码执行不同操作前,临时设置对应账号的AWS_PROFILE环境变量:
import os # 拉取密钥前切换到账号1 os.environ['AWS_PROFILE'] = 'aws-account-1' session1 = boto3.Session() sm_client = session1.client('secretmanager', region_name=self.__region) response = sm_client.get_secret_value(SecretId=secret_name) # 执行Athena查询前切换到账号2 os.environ['AWS_PROFILE'] = 'aws-account-2' session2 = boto3.Session() resultset = wr.athena.query("", session=session2)
这种方式无需在Session中指定profile_name,直接读取环境变量对应的配置。
内容的提问来源于stack exchange,提问作者thebackendmonk
相关产品推荐
相关产品推荐

