如何通过Lambda触发器实现Amazon Cognito多自定义认证挑战（OTP选择+验证）并解决当前异常

我仔细看了你的代码和问题描述，咱们一步步梳理问题根源，然后修正代码解决这些异常。

问题根源分析 #

你遇到的流程卡住、Lambda输出错误等问题，主要是因为三个Lambda的职责边界混淆、Metadata字符串不匹配、会话状态判断逻辑不严谨导致的：

1. Metadata不匹配：VerifyAuth里判断第一步挑战用的是"CHOOSE_OTP_DELIVERY"，但Define/CreateAuth里用的是"SELECT_DELIVERY_METHOD"，导致第一步验证失败，流程卡住。
2. 职责越界：你在DefineAuth里直接设置挑战的具体内容（比如publicChallengeParameters），这其实是CreateAuthChallenge的工作，Cognito流程会因为职责混乱无法正确触发后续Lambda。
3. 分支覆盖不全：部分代码分支没有正确设置响应字段，导致Cognito识别为“unrecognized Lambda output”。

修正后的Lambda代码 #

1. DefineAuthChallenge Lambda（仅负责决定当前需要什么挑战）

这个Lambda的核心是根据会话状态判断下一步要发起的挑战类型，不负责生成挑战内容：

import json

def lambda_handler(event, context):
    print("EVENT request:", event["request"])
    session = event["request"]["session"]
    response = event["response"]

    # 情况1：没有已完成的挑战 → 发起第一步：选择OTP发送方式
    if len(session) == 0:
        response["challengeName"] = "CUSTOM_CHALLENGE"
        response["challengeMetadata"] = "SELECT_DELIVERY_METHOD"
        response["issueTokens"] = False
        response["failAuthentication"] = False
    # 情况2：已完成1次挑战（选择了发送方式且验证通过）→ 发起第二步：验证OTP
    elif len(session) == 1 and session[0]["challengeResult"] is True:
        response["challengeName"] = "CUSTOM_CHALLENGE"
        response["challengeMetadata"] = "VERIFY_OTP"
        response["issueTokens"] = False
        response["failAuthentication"] = False
    # 情况3：已完成2次挑战（OTP验证通过）→ 颁发令牌
    elif len(session) == 2 and session[1]["challengeResult"] is True:
        response["issueTokens"] = True
        response["failAuthentication"] = False
    # 所有其他情况 → 认证失败
    else:
        response["issueTokens"] = False
        response["failAuthentication"] = True

    print("EVENT response:", response)
    return event

2. CreateAuthChallenge Lambda（负责生成挑战的具体内容）

这个Lambda根据DefineAuth指定的挑战类型，生成对应的挑战提示、私有参数等：

import json
import random

def lambda_handler(event, context):
    print("EVENT request:", event["request"])
    session = event["request"]["session"]
    challenge_metadata = event["request"]["challengeMetadata"]
    response = event["response"]

    # 第一步挑战：生成选择OTP发送方式的提示
    if challenge_metadata == "SELECT_DELIVERY_METHOD":
        response["challengeName"] = "CUSTOM_CHALLENGE"
        response["challengeMetadata"] = "SELECT_DELIVERY_METHOD"
        response["publicChallengeParameters"] = {
            "question": "How would you like to receive the OTP? (email or sms)"
        }
        response["issueTokens"] = False
        response["failAuthentication"] = False
    # 第二步挑战：生成OTP并存储，模拟发送
    elif challenge_metadata == "VERIFY_OTP":
        # 获取第一步用户选择的发送方式（从会话中取上一次的验证结果）
        selected_method = session[0]["challengeAnswer"]
        # 生成8位OTP
        otp = str(random.randint(10000000, 99999999))
        
        response["challengeName"] = "CUSTOM_CHALLENGE"
        response["challengeMetadata"] = "VERIFY_OTP"
        response["publicChallengeParameters"] = {
            "otpPrompt": f"Enter the OTP sent to your {selected_method}"
        }
        # 存储OTP和发送方式（仅Cognito和VerifyAuth可见）
        response["privateChallengeParameters"] = {
            "otp": otp,
            "deliveryMethod": selected_method
        }
        response["issueTokens"] = False
        response["failAuthentication"] = False
        
        # 模拟发送OTP（实际项目中替换为SES/SNS调用）
        print(f"Sending OTP {otp} via {selected_method}")
    # 其他情况 → 认证失败
    else:
        response["issueTokens"] = False
        response["failAuthentication"] = True

    print("EVENT response:", response)
    return event

3. VerifyAuthChallenge Lambda（负责验证用户输入是否正确）

修正Metadata匹配问题，确保每个挑战的验证逻辑正确：

def lambda_handler(event, context):
    print("EVENT request:", event["request"])
    challenge_metadata = event["request"]["challengeMetadata"]
    user_answer = event["request"]["challengeAnswer"]
    private_params = event["request"].get("privateChallengeParameters", {})
    response = event["response"]

    # 第一步验证：检查用户选择的发送方式是否合法
    if challenge_metadata == "SELECT_DELIVERY_METHOD":
        # 验证输入是否是允许的选项
        if user_answer.lower() in ["email", "sms"]:
            response["answerCorrect"] = True
        else:
            response["answerCorrect"] = False
    # 第二步验证：检查OTP是否匹配
    elif challenge_metadata == "VERIFY_OTP":
        expected_otp = private_params.get("otp")
        response["answerCorrect"] = (user_answer == expected_otp)
    # 默认情况：验证失败
    else:
        response["answerCorrect"] = False

    print("EVENT response:", response)
    return event

测试与注意事项 #

1. Cognito配置检查：
   • 确保用户池已启用「自定义认证」，并正确关联了三个Lambda触发器
   • 目标用户的属性中已填写email或phone_number（对应OTP发送方式）
2. 前端流程配合：
   • 第一步：收到SELECT_DELIVERY_METHOD的metadata后，展示选择界面，让用户输入email或sms
   • 第二步：收到VERIFY_OTP的metadata后，展示OTP输入框
   • 确保前端正确维护Cognito会话ID，不要在流程中丢失（解决“Local storage is missing an ID Token”的错误）
3. 日志排查：开启Lambda的CloudWatch日志，每次请求后查看日志，确认event的输入输出是否符合Cognito的格式要求。

备注：内容来源于stack exchange，提问作者burns0907