实现botocore.credentials的RefreshableCredentials后脚本1小时仍失效

核心问题 #

你的代码存在两个关键问题，导致凭证过期后仍触发报错：

1. 静态环境变量未同步刷新后的凭证 #

你将初始获取的临时凭证直接赋值给env_var，但这个变量是静态的，不会随着RefreshableCredentials的自动刷新而更新。而grab_pod_information中调用K8s相关操作（比如通过aws eks get-token生成kubeconfig凭证）时，使用的是这个静态环境变量里的过期凭证，因此会触发ExpiredToken错误。

2. get_target_account_session函数的潜在问题 #

• 函数内sts = client("sts")未正确导入client（应为boto3.client或基于源账号session创建），若后续刷新时使用过期临时凭证调用assume_role，会直接失败。
• 未导入ClientError，异常处理会触发未定义错误。

修复后的代码 #

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

def get_target_account_session(target_account_id: str, region: str):
    # 使用源账号默认session创建STS客户端，确保刷新时依赖有效源凭证
    sts = boto3.client("sts", region_name=region)
    try:
        response = sts.assume_role(
            RoleArn=f"arn:aws:iam::{target_account_id}:role/<role_name>",
            RoleSessionName="<session_name>"
        )
        if "Credentials" in response:
            creds = response["Credentials"]
            return {
                "access_key": creds["AccessKeyId"],
                "secret_key": creds["SecretAccessKey"],
                "token": creds["SessionToken"],
                "expiry_time": creds["Expiration"].isoformat(),
                "region_name": region
            }
    except ClientError as err:
        print(f"AssumeRole调用失败: {err}")
        exit(1)
    return None

def get_aws_autorefresh_session(target_account_id, region):
    session_credentials = RefreshableCredentials.create_from_metadata(
        metadata=get_target_account_session(target_account_id, region),
        refresh_using=partial(get_target_account_session, target_account_id, region),
        method="sts-assume-role"
    )

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

    return session_credentials, autorefresh_session

# 获取自动刷新的session
creds, ses = get_aws_autorefresh_session(value[0], region)
eks = ses.client("eks", region_name=region)
clusters = list_clusters_boto3(eks)

# 修改grab_pod_information，动态获取最新凭证
def grab_pod_information(target_account_id, clusters, region, boto3_session, param):
    # 每次执行K8s操作前，动态拉取当前有效凭证
    current_creds = boto3_session.get_credentials().get_frozen_credentials()
    env_var = {
        "AWS_ACCESS_KEY_ID": current_creds.access_key,
        "AWS_SECRET_ACCESS_KEY": current_creds.secret_key,
        "AWS_SESSION_TOKEN": current_creds.token,
        "KUBECONFIG": "/root/.kube/<kube_config>"
    }
    # 后续K8s操作逻辑使用这个动态生成的env_var
    # ... 你的原有业务代码 ...

# 传入自动刷新的session调用函数
grab_pod_information(value[0], clusters, region, ses, value[1])

关键修复点说明 #

• 动态获取凭证：在每次执行K8s操作前，通过boto3_session.get_credentials().get_frozen_credentials()获取最新有效凭证，确保环境变量始终使用未过期的凭证。
• 修复STS客户端初始化：使用源账号默认session创建STS客户端，避免刷新时依赖过期临时凭证。
• 补全依赖导入：添加ClientError的导入，确保异常处理正常工作。
• 废弃静态凭证传递：不再将初始凭证赋值给静态变量，而是通过boto3 session动态同步最新凭证状态。

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