如何用Python结合变量正确刷新AWS凭证（解决传参报错）

问题场景 #

尝试使用botocore的RefreshableCredentials自动管理AWS凭证更新，编写了带参数的凭证获取函数，初始运行正常，但300秒后抛出参数缺失异常：

get_aws_credentials() missing 2 required positional arguments: 'aws_role_arn' and 'session_name'

初始代码如下：

import boto3
import botocore
from botocore.credentials import RefreshableCredentials
from botocore.session import get_session

def get_aws_credentials(aws_role_arn, session_name):
    sts_client = boto3.client('sts')
    assumed_role_object = sts_client.assume_role(
        RoleArn = aws_role_arn,
        RoleSessionName = session_name,
        DurationSeconds = 900
    )
    return {
        'access_key': assumed_role_object['Credentials']['AccessKeyId'],
        'secret_key': assumed_role_object['Credentials']['SecretAccessKey'],
        'token': assumed_role_object['Credentials']['SessionToken'],
        'expiry_time': assumed_role_object['Credentials']['Expiration'].isoformat()
    }

def get_aws_autorefresh_session(aws_role_arn, session_name):
    session_credentials = RefreshableCredentials.create_from_metadata(
        metadata = get_aws_credentials(aws_role_arn, session_name),
        refresh_using = get_aws_credentials,
        method = 'sts-assume-role'
    )

    session = get_session()
    session._credentials = session_credentials
    autorefresh_session = boto3.Session(botocore_session=session)

    return autorefresh_session, session_credentials

调用代码：

arn = "arn:aws:iam::123456789012:role/YourRole"
session_name = "Test"
auto_refresh_session, credentials = get_aws_autorefresh_session(arn, session_name)

问题原因 #

RefreshableCredentials的refresh_using参数要求传入的函数无额外参数。当凭证过期触发自动刷新时，botocore会直接调用该函数，不会自动传递aws_role_arn和session_name参数，导致原函数因参数缺失报错。

解决方案 #

通过闭包或functools.partial将参数绑定到凭证获取函数中，生成一个无参的包装函数供refresh_using调用。

方案1：使用闭包 #

修改get_aws_autorefresh_session函数，在内部定义绑定了外部参数的嵌套函数：

import boto3
import botocore
from botocore.credentials import RefreshableCredentials
from botocore.session import get_session

def get_aws_credentials(aws_role_arn, session_name):
    sts_client = boto3.client('sts')
    assumed_role_object = sts_client.assume_role(
        RoleArn = aws_role_arn,
        RoleSessionName = session_name,
        DurationSeconds = 900
    )
    return {
        'access_key': assumed_role_object['Credentials']['AccessKeyId'],
        'secret_key': assumed_role_object['Credentials']['SecretAccessKey'],
        'token': assumed_role_object['Credentials']['SessionToken'],
        'expiry_time': assumed_role_object['Credentials']['Expiration'].isoformat()
    }

def get_aws_autorefresh_session(aws_role_arn, session_name):
    # 定义闭包，绑定外部参数
    def refresh_credentials():
        return get_aws_credentials(aws_role_arn, session_name)
    
    session_credentials = RefreshableCredentials.create_from_metadata(
        metadata = refresh_credentials(),  # 初始调用获取凭证
        refresh_using = refresh_credentials,  # 刷新时调用闭包
        method = 'sts-assume-role'
    )

    session = get_session()
    session._credentials = session_credentials
    autorefresh_session = boto3.Session(botocore_session=session)

    return autorefresh_session, session_credentials

方案2：使用functools.partial #

利用functools.partial包装原函数，预先传入固定参数：

import boto3
import botocore
from functools import partial
from botocore.credentials import RefreshableCredentials
from botocore.session import get_session

def get_aws_credentials(aws_role_arn, session_name):
    sts_client = boto3.client('sts')
    assumed_role_object = sts_client.assume_role(
        RoleArn = aws_role_arn,
        RoleSessionName = session_name,
        DurationSeconds = 900
    )
    return {
        'access_key': assumed_role_object['Credentials']['AccessKeyId'],
        'secret_key': assumed_role_object['Credentials']['SecretAccessKey'],
        'token': assumed_role_object['Credentials']['SessionToken'],
        'expiry_time': assumed_role_object['Credentials']['Expiration'].isoformat()
    }

def get_aws_autorefresh_session(aws_role_arn, session_name):
    # 用partial绑定参数，生成无参函数
    refresh_func = partial(get_aws_credentials, aws_role_arn, session_name)
    
    session_credentials = RefreshableCredentials.create_from_metadata(
        metadata = refresh_func(),
        refresh_using = refresh_func,
        method = 'sts-assume-role'
    )

    session = get_session()
    session._credentials = session_credentials
    autorefresh_session = boto3.Session(botocore_session=session)

    return autorefresh_session, session_credentials

验证方式 #

调用方式保持不变，凭证过期后会自动调用绑定了参数的函数刷新，不会再出现参数缺失异常：

arn = "arn:aws:iam::123456789012:role/YourRole"
session_name = "Test"
auto_refresh_session, credentials = get_aws_autorefresh_session(arn, session_name)

# 使用自动刷新session调用AWS服务
s3_client = auto_refresh_session.client('s3')
print(s3_client.list_buckets())

内容的提问来源于stack exchange，提问作者espogian