AWS Lambda跨账号DynamoDB访问STS令牌过期问题排查

安全阿岳

2026-7-5

AWS Lambda跨账号DynamoDB查询的STS令牌异常问题

问题背景

我有一个使用boto3的STS客户端跨账号连接DynamoDB的AWS Lambda函数，功能是基于查询获取数据，日常运行正常，但性能测试出现异常：

测试用例1：每分钟75次API调用，持续1小时，总会在约55-57分钟时出现502错误网关，Lambda日志显示错误：[ERROR] ClientError: An error occurred (ExpiredTokenException) when calling the Query operation: The security token included in the request is expired
相同环境下，每分钟500次调用持续1小时却无任何错误。

按照建议将STS令牌创建逻辑从全局移至Lambda主函数后，问题仍未解决。

核心问题

为何低并发压测失败，而高并发压测却能正常运行？
为何代码修改后令牌仍未自动刷新？

现有代码

import json
import boto3
import os
from boto3.dynamodb.conditions import Key, Attr
from botocore.credentials import RefreshableCredentials
from decimal import Decimal
import datetime


# Constants
ROLE_ARN = os.environ['arnrole']
ROLE_SESSION_NAME = 'dynamodb-session'
EXPIRATION_THRESHOLD_SECONDS = 3580  # Setting 5 mins check below which it will refresh token


def get_sts_token():
   # Create an STS client
   sts_client = boto3.client('sts')
   # Assume the role to get temporary credentials
   response = sts_client.assume_role(
       RoleArn=ROLE_ARN,
       RoleSessionName=ROLE_SESSION_NAME
   )
   # Extract and return temporary credentials
   print("invoked sts new")
   return response['Credentials']
   
def refresh_sts_token_if_needed(current_credentials):
   # Calculate remaining time until expiration\
   print(current_credentials)
   expiration_time = current_credentials['Expiration']
   expiration_date = expiration_time.replace(tzinfo=None)
   remaining_seconds = ( expiration_date - datetime.datetime.utcnow()).total_seconds()

   print(remaining_seconds)

   if remaining_seconds < EXPIRATION_THRESHOLD_SECONDS:
       print("inside loop for new token")
       # Refresh the STS token if it's close to expiration
       return get_sts_token()
   else:
       # Return the current credentials if they are still valid
       print("else loop")
       return current_credentials


# Get or refresh STS token
current_credentials = get_sts_token()
    

""" Convert Decimal to float (test demo)
    DynamoDB stores floats as Decimals and that cannot be encoded by json.dumps()   
    the data first needs to be converted to float """
    
class DecimalEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, Decimal):
            return float(obj)
        return json.JSONEncoder.default(self, obj)

       
def get_a(event):
  return(event["pathParameters"]["a"])


def get_query_result(table, a):
    
    query_result_json = table.query(KeyConditionExpression=Key('a').eq(a))
    
    query_results = json.loads(json.dumps(query_result_json, cls=DecimalEncoder))
    
    return query_results


def define_api_response(query_results, account_number):
    
    json_body_error = {"ResponseMetadata" : {"Count" : [], "RequestId" : [] , "Message" : []}}
    
    # Define the status code and body of response based on query result
    if query_results["ResponseMetadata"]["HTTPStatusCode"] == 200 and query_results["Count"] > 0: 
        response = create_response_body(200, json.dumps(query_results["Items"][0]))
    elif query_results["ResponseMetadata"]["HTTPStatusCode"] == 200 and query_results["Count"] == 0: 

        response = create_response_body(404, json.dumps(query_results["Items"]))

    else:
        json_body_error["ResponseMetadata"]["Message"].append(query_results["errorMessage"])
        json_body_error["ResponseMetadata"]["RequestId"].append(query_results["ResponseMetadata"]["RequestId"])      
        response = create_response_body(query_results["ResponseMetadata"]["HTTPStatusCode"], json.dumps(json_body_error))
    
    return response


def create_response_body(statusCode, body):
    
    str_body = str(body)
    
    response = {
      "statusCode": statusCode,
      "body": str_body,
      "headers": {
        "Content-Type": "application/json",
        "Access-Control-Allow-Origin": "*"
      },
        }
    
    print(str_body)
    return response


def lambda_handler(event, context):

    # Use the STS token or refresh if close to expiration
    refreshed_credentials = refresh_sts_token_if_needed(current_credentials)
    
     # Dynamo client connection : 
    dynamodb = boto3.resource('dynamodb', region_name="us-east-1",aws_access_key_id=refreshed_credentials['AccessKeyId'],aws_secret_access_key=refreshed_credentials['SecretAccessKey'], aws_session_token=refreshed_credentials['SessionToken'] )

    
    # Rest of the code
    
    a = get_a(event)
    
    table = dynamodb.Table(os.environ['tablename'])
    
    query_results = get_query_result(table, account_number)
    
    response  = define_api_response(query_results, account_number)
    
    return response

问题解答

1. 低并发失败、高并发正常的原因

Lambda执行环境存在复用机制：

低并发场景下，同一个执行环境会被重复调用（最长可复用6小时），初始化时创建的STS令牌（默认有效期1小时）会随着时间推移逐渐过期，最终触发ExpiredTokenException。
高并发场景下，AWS会快速创建大量新的执行环境，每个新环境都会重新初始化并获取新的STS令牌，而测试周期仅1小时，这些令牌在环境生命周期内不会过期，因此无错误。

2. 代码修改后令牌仍未自动刷新的原因

代码存在两个核心缺陷：

全局凭证未更新：refresh_sts_token_if_needed返回的新凭证仅赋值给局部变量refreshed_credentials，全局变量current_credentials始终保留初始化时的旧令牌。下一次调用同一个执行环境时，依然用旧令牌判断有效期，无法触发有效刷新。
阈值逻辑无效：EXPIRATION_THRESHOLD_SECONDS = 3580（仅比1小时少20秒），导致令牌创建后仅20秒就会触发刷新，但刷新后的新凭证未替换全局旧凭证，后续调用还是基于旧令牌判断，最终旧令牌到期后报错。

修复方案

使用boto3官方推荐的RefreshableCredentials实现自动凭证刷新，无需手动判断有效期：

import json
import boto3
import os
from boto3.dynamodb.conditions import Key
from botocore.credentials import RefreshableCredentials
from botocore.session import Session
from decimal import Decimal


# Constants
ROLE_ARN = os.environ['arnrole']
ROLE_SESSION_NAME = 'dynamodb-session'

def get_sts_token():
    sts_client = boto3.client('sts')
    response = sts_client.assume_role(
        RoleArn=ROLE_ARN,
        RoleSessionName=ROLE_SESSION_NAME
    )
    credentials = response['Credentials']
    return {
        'access_key': credentials['AccessKeyId'],
        'secret_key': credentials['SecretAccessKey'],
        'token': credentials['SessionToken'],
        'expiry_time': credentials['Expiration'].isoformat()
    }

# 初始化可自动刷新的凭证
refreshable_credentials = RefreshableCredentials.create_from_metadata(
    metadata=get_sts_token(),
    refresh_using=get_sts_token,
    method='sts-assume-role'
)

# 创建带自动刷新凭证的Session
session = Session()
session._credentials = refreshable_credentials
dynamodb = session.resource('dynamodb', region_name="us-east-1")

class DecimalEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, Decimal):
            return float(obj)
        return json.JSONEncoder.default(self, obj)

def get_a(event):
    return event["pathParameters"]["a"]

def get_query_result(table, a):
    query_result_json = table.query(KeyConditionExpression=Key('a').eq(a))
    return json.loads(json.dumps(query_result_json, cls=DecimalEncoder))

def define_api_response(query_results):
    json_body_error = {"ResponseMetadata" : {"Count" : [], "RequestId" : [] , "Message" : []}}
    
    if query_results["ResponseMetadata"]["HTTPStatusCode"] == 200 and query_results["Count"] > 0: 
        response = create_response_body(200, json.dumps(query_results["Items"][0]))
    elif query_results["ResponseMetadata"]["HTTPStatusCode"] == 200 and query_results["Count"] == 0: 
        response = create_response_body(404, json.dumps(query_results["Items"]))
    else:
        json_body_error["ResponseMetadata"]["Message"].append(query_results["errorMessage"])
        json_body_error["ResponseMetadata"]["RequestId"].append(query_results["ResponseMetadata"]["RequestId"])      
        response = create_response_body(query_results["ResponseMetadata"]["HTTPStatusCode"], json.dumps(json_body_error))
    
    return response

def create_response_body(statusCode, body):
    response = {
        "statusCode": statusCode,
        "body": str(body),
        "headers": {
            "Content-Type": "application/json",
            "Access-Control-Allow-Origin": "*"
        },
    }
    print(body)
    return response

def lambda_handler(event, context):
    a = get_a(event)
    table = dynamodb.Table(os.environ['tablename'])
    query_results = get_query_result(table, a)
    response = define_api_response(query_results)
    return response

关键修复点

自动凭证管理：RefreshableCredentials会在凭证即将过期时自动调用get_sts_token获取新凭证，无需手动判断有效期。
全局资源复用：基于带刷新凭证的Session创建全局DynamoDB资源，确保每次请求都使用有效凭证。
修复代码错误：修正了原代码中account_number未定义的逻辑错误。

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

AWS Lambda跨账号DynamoDB访问STS令牌过期问题排查

问题背景 #

核心问题 #

现有代码 #

问题解答 #

1. 低并发失败、高并发正常的原因 #

2. 代码修改后令牌仍未自动刷新的原因 #

修复方案 #

关键修复点 #