Lambda函数在Secrets扩展初始化前调用引发400错误求助

我严格按照AWS官方文档实现了用AWS Secrets Lambda Extension缓存Secrets Manager中的密钥。未添加缓存层时能正常获取密钥，但添加扩展后请求返回400错误。

代码实现 #

def get_secrets():
    headers = {"X-Aws-Parameters-Secrets-Token": os.environ.get('AWS_SESSION_TOKEN')}
    secrets_extension_http_port = "2773"
    secrets_extension_endpoint = "http://localhost:" + \
    secrets_extension_http_port + \
    "/secretsmanager/get?secretId=" + \
    "my-secrets"
try:
    r = requests.get(url="http://localhost:2773/secretsmanager/get?secretId=sf-secrets", headers=headers)
    secret = json.loads(r.text)["SecretString"]  
    return secret
except Exception as e:
    print(f"""error::::{e}""")

日志信息 #

[AWS参数与密钥Lambda扩展] 2023/08/11 07:29:53 PARAMETERS_SECRETS_EXTENSION_LOG_LEVEL 设为debug，日志级别已设置为debug。
[AWS参数与密钥Lambda扩展] 2023/08/11 07:29:53 调试：未设置PARAMETERS_SECRETS_EXTENSION_CACHE_ENABLED，默认启用缓存。
[AWS参数与密钥Lambda扩展] 2023/08/11 07:29:53 调试：未设置PARAMETERS_SECRETS_EXTENSION_CACHE_SIZE，使用默认缓存大小：1000个对象。
[AWS参数与密钥Lambda扩展] 2023/08/11 07:29:53 调试：未设置SECRETS_MANAGER_TTL，设置默认存活时间：5分钟。
[AWS参数与密钥Lambda扩展] 2023/08/11 07:29:53 调试：未设置SSM_PARAMETER_STORE_TTL，设置默认存活时间：5分钟。
[AWS参数与密钥Lambda扩展] 2023/08/11 07:29:53 调试：未设置SECRETS_MANAGER_TIMEOUT_MILLIS，设置默认超时：0秒。
[AWS参数与密钥Lambda扩展] 2023/08/11 07:29:53 调试：未设置SSM_PARAMETER_STORE_TIMEOUT_MILLIS，设置默认超时：0秒。
[AWS参数与密钥Lambda扩展] 2023/08/11 07:29:53 调试：未设置PARAMETERS_SECRETS_EXTENSION_MAX_CONNECTIONS，设置默认值：3。
[AWS参数与密钥Lambda扩展] 2023/08/11 07:29:53 调试：未设置PARAMETERS_SECRETS_EXTENSION_HTTP_PORT，设置默认端口：2773。
[AWS参数与密钥Lambda扩展] 2023/08/11 07:29:53 信息：Systems Manager参数存储与Secrets Manager Lambda扩展版本1.0.103
[AWS参数与密钥Lambda扩展] 2023/08/11 07:29:53 调试：创建大小为1000的新缓存
[AWS参数与密钥Lambda扩展] 2023/08/11 07:29:53 信息：在端口2773提供服务
响应码400
尚未准备好处理请求，请稍候
期望获取值：第1行第1列（字符0）

[错误] TypeError: 'NoneType'对象不可下标
回溯（最近的调用最后）：
文件 "/var/lang/lib/python3.8/imp.py", 第234行, 在load_module中
返回 load_source(name, filename, file)
文件 "/var/lang/lib/python3.8/imp.py", 第171行, 在load_source中
module = _load(spec)
文件 "<frozen importlib._bootstrap>", 第702行, 在_load中
文件 "<frozen importlib._bootstrap>", 第671行, 在_load_unlocked中
文件 "<frozen importlib._bootstrap_external>", 第843行, 在exec_module中
文件 "<frozen importlib._bootstrap>", 第219行, 在_call_with_frames_removed中
文件 "/var/task/app/main.py", 第3行, 在<module>中
from app.routers import router as api_router
文件 "/var/task/app/routers/router.py", 第2行, 在<module>中
import app.constants as constants
文件 "/var/task/app/constants.py", 第3行, 在<module>中
DATABASE_HOST = str(secrets.get_database_host())
文件 "/var/task/app/secrets.py", 第79行, 在get_database_host中
return PASSWORDS['username']
扩展 名称: AWSParametersAndSecretsLambdaExtension     
状态: 就绪    事件: [INVOKE, SHUTDOWN]
启动 RequestId: 15572c5d-9c3c-4f17-92c8-6b17d854ac93 版本: 
$LATEST
发生未知应用错误
Runtime.Unknown
结束 RequestId: 15572c5d-9c3c-4f17-92c8-6b17d854ac93
报告 RequestId: 15572c5d-9c3c-4f17-92c8-6b17d854ac93   
持续时间: 10126.93 ms   计费持续时间: 10127 ms   内存
大小: 128 MB    最大内存使用: 58 MB

已尝试操作 #

• 将secretId参数替换为密钥名称和ARN，错误依旧
• Lambda函数已关联VPC，并通过VPC终端节点连接Secrets Manager

Lambda执行角色权限 #

"secretsmanager:GetSecretValue",
"secretsmanager:DescribeSecret",
"secretsmanager:ListSecrets",
"kms:Decrypt"

按文档实现看似无问题，请问是否遗漏了配置步骤？

复现步骤 #

运行FastAPI应用：

1. 创建app目录并添加app.py文件：

from fastapi import FastAPI
from mangum import Mangum
from app.routers import router as api_router

app = FastAPI()
handler = Mangum(app)

app.include_router(
    api_router.router,
    prefix="/api/v2"
)

2. 创建router目录并添加router.py文件：

import app.secrets as secrets
from fastapi import APIRouter

router = APIRouter()

@router.get("/secrets")
async def get_secrets():
    db_host = secrets.get_database_host
    print(db_host)
    return db_host

3. 在app目录下添加secrets.py文件：

import json
import requests
import os

def get_secrets():
    headers = {"X-Aws-Parameters-Secrets-Token": os.environ.get('AWS_SESSION_TOKEN')}
    try:
        r = requests.get(url="http://localhost:2773/secretsmanager/get?secretId=sf-secrets", headers=headers)
        print(f"""response code {r.status_code}""")
        print(r.headers)
        secret = json.loads(r.text)["SecretString"]  
        return secret
    except Exception as e:
        print(f"""error::::{e}""")

PASSWORDS = get_secrets()
print(f""""Response from secrets manager {PASSWORDS}""")

def get_database_host() -> str:
    return PASSWORDS['username']

def get_database_password() -> str:
    return PASSWORDS['password']

添加requirements.txt：

fastapi==0.95.0
uvicorn==0.17.6
mangum==0.17.0
boto3==1.28.3
requests==2.31.0

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