在Lambda中使用STS跨账号访问Kinesis时遭遇ExpiredTokenException问题

我之前也踩过这个坑！你遇到的ExpiredTokenException其实和Lambda执行环境的复用机制有关，而且AWS SDK并不会自动帮你重复获取令牌——除非你用对了凭证处理方式。下面给你拆解问题原因和具体解决方案：

为什么会出现这个错误？ #

Lambda会复用执行环境（也就是“冷启动”之后，后续调用会复用之前的环境）。如果你在函数handler外部初始化STS客户端或者获取AssumeRole凭证，这些凭证会被缓存到环境中。STS临时凭证默认有效期是1小时，当环境被复用超过1小时后，再用旧凭证调用Kinesis就会触发过期错误。

很多人误以为Lambda会自动刷新这些跨角色的凭证，但其实Lambda只会自动管理自身的执行角色凭证，不会帮你处理手动AssumeRole得到的凭证。

解决方案 #

1. 使用AWS SDK的自动刷新凭证机制 #

不要手动调用sts.assume_role然后缓存凭证，而是让SDK自动处理凭证的刷新。以Python的boto3为例，你可以用DeferredRefreshableCredentials来创建自动刷新的凭证提供者：

import boto3
from botocore.credentials import AssumeRoleCredentialFetcher, DeferredRefreshableCredentials
from botocore.client import Config
from botocore.session import Session

def get_auto_refresh_credentials(role_arn):
    session = Session()
    # 创建凭证获取器
    fetcher = AssumeRoleCredentialFetcher(
        source_credentials=session.get_credentials(),
        role_arn=role_arn,
        role_session_name="lambda-cross-account-kinesis",  # 自定义会话名，方便审计
        session=session,
        config=Config()
    )
    # 创建自动刷新的凭证对象
    return DeferredRefreshableCredentials(
        method="assume-role",
        refresh_using=fetcher.fetch_credentials,
        time_fetcher=lambda: session.get_credentials().current_time()
    )

def lambda_handler(event, context):
    # 账号B的角色ARN
    cross_account_role_arn = "arn:aws:iam::ACCOUNT_B_ID:role/YourCrossAccountKinesisRole"
    # 获取自动刷新的凭证
    auto_refresh_creds = get_auto_refresh_credentials(cross_account_role_arn)
    
    # 用自动刷新凭证创建Kinesis客户端
    kinesis_client = boto3.client(
        "kinesis",
        region_name="your-region",
        credentials=auto_refresh_creds
    )
    
    # 执行PutRecords操作
    put_records_response = kinesis_client.put_records(
        Records=[
            # 你的记录内容，比如{"Data": b"your-data", "PartitionKey": "key1"}
        ],
        StreamName="YourKinesisStreamName"
    )
    
    return {"statusCode": 200, "response": put_records_response}

这样配置后，SDK会自动检测凭证是否即将过期，在需要的时候自动调用STS刷新，完全不用你手动处理。

2. 确保权限配置正确 #

别忘记检查两个关键权限：

• 账号A的Lambda执行角色：需要有sts:AssumeRole权限，允许调用账号B的角色ARN。比如权限策略：

  {
      "Version": "2012-10-17",
      "Statement": [
          {
              "Effect": "Allow",
              "Action": "sts:AssumeRole",
              "Resource": "arn:aws:iam::ACCOUNT_B_ID:role/YourCrossAccountKinesisRole"
          }
      ]
  }
  

• 账号B的目标角色：信任策略必须允许账号A的Lambda执行角色来Assume它。比如信任策略：

  {
      "Version": "2012-10-17",
      "Statement": [
          {
              "Effect": "Allow",
              "Principal": {
                  "AWS": "arn:aws:iam::ACCOUNT_A_ID:role/YourLambdaExecutionRole"
              },
              "Action": "sts:AssumeRole"
          }
      ]
  }
  

3. 避免在handler外缓存凭证 #

如果你一定要手动调用sts.assume_role，那必须把获取凭证的代码放在handler函数内部，确保每次函数调用（包括环境复用的情况）都会检查凭证是否过期，必要时重新获取：

import boto3
from datetime import datetime, timedelta

sts_client = boto3.client("sts")
cached_creds = None
cred_expiry_time = None

def lambda_handler(event, context):
    global cached_creds, cred_expiry_time
    role_arn = "arn:aws:iam::ACCOUNT_B_ID:role/YourCrossAccountKinesisRole"
    
    # 检查凭证是否过期或未初始化，提前5分钟刷新避免网络延迟问题
    if not cached_creds or datetime.utcnow() >= cred_expiry_time - timedelta(minutes=5):
        response = sts_client.assume_role(
            RoleArn=role_arn,
            RoleSessionName="lambda-cross-account-kinesis"
        )
        cached_creds = response["Credentials"]
        cred_expiry_time = cached_creds["Expiration"]
    
    # 用新凭证创建Kinesis客户端
    kinesis_client = boto3.client(
        "kinesis",
        region_name="your-region",
        aws_access_key_id=cached_creds["AccessKeyId"],
        aws_secret_access_key=cached_creds["SecretAccessKey"],
        aws_session_token=cached_creds["SessionToken"]
    )
    
    # 执行PutRecords操作...

不过这种方式需要自己处理过期逻辑，不如第一种SDK自动刷新的方式可靠。

总结 #

最稳妥的方式是使用SDK的自动刷新凭证机制，同时确保跨账号的权限配置正确。这样就能彻底避免ExpiredTokenException的问题啦！

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