求AWS User Pools自定义认证流Python示例及触发器编写方法

一、Python版本的User Pools触发器基础写法 #

JavaScript里用context.done来结束触发器并返回结果，但Python的Lambda触发器逻辑更直接：你只需要接收event参数，根据触发器类型处理后返回修改后的event对象即可。

举个最常见的前置认证（PreAuthentication）触发器示例，比如在用户登录前检查自定义权限：

def lambda_handler(event, context):
    # 从event中提取核心信息
    user_name = event['userName']
    user_pool_id = event['userPoolId']
    
    # 自定义逻辑：比如检查用户是否在允许登录的白名单内
    allowed_users = ["alice", "bob"]
    if user_name not in allowed_users:
        # 抛出异常会直接终止认证流程，Cognito会返回对应错误给用户
        raise Exception("User is not authorized to access this resource")
    
    # 处理完成后直接返回event，Cognito会自动继续后续流程
    return event

不同类型的触发器（比如PostAuthentication、PreSignUp等）的event结构会略有差异，你可以在Lambda控制台的测试事件模板里选择对应Cognito触发器类型，查看完整的字段结构。核心逻辑始终是：处理event，返回event。

二、自定义认证流的Python完整示例 #

自定义认证流需要搭配三个核心触发器：DefineAuthChallenge、CreateAuthChallenge、VerifyAuthChallengeResponse，下面是一个基于短信验证码的完整示例：

1. DefineAuthChallenge 触发器 #

这个函数用来决定当前认证步骤的挑战类型，比如首次请求发起验证码挑战，验证通过后发放令牌：

def lambda_handler(event, context):
    session = event['request']['session']
    
    if len(session) == 0:
        # 首次请求，启动验证码挑战
        event['response']['challengeName'] = 'CUSTOM_CHALLENGE'
        event['response']['issueTokens'] = False
        event['response']['failAuthentication'] = False
    else:
        # 检查上一次挑战的结果
        last_challenge = session[-1]
        if last_challenge['challengeName'] == 'CUSTOM_CHALLENGE' and last_challenge['challengeResult']:
            # 验证码验证通过，发放身份令牌
            event['response']['issueTokens'] = True
            event['response']['failAuthentication'] = False
        else:
            # 验证失败，终止认证流程
            event['response']['issueTokens'] = False
            event['response']['failAuthentication'] = True
    
    return event

2. CreateAuthChallenge 触发器 #

这个函数生成挑战内容（比如6位验证码），并存储到私有参数中（仅后端可访问）：

import random

def lambda_handler(event, context):
    if event['request']['challengeName'] == 'CUSTOM_CHALLENGE':
        # 生成6位随机验证码
        verification_code = str(random.randint(100000, 999999))
        
        # 这里可以添加发送验证码的逻辑，比如调用AWS SNS发送短信
        # send_sms(event['userName'], verification_code)
        
        # 将验证码存入私有参数，供后续验证使用
        event['response']['privateChallengeParameters'] = {
            'verificationCode': verification_code
        }
        # 添加元数据标识挑战类型
        event['response']['challengeMetadata'] = 'SMS_VERIFICATION_CHALLENGE'
    
    return event

3. VerifyAuthChallengeResponse 触发器 #

这个函数验证用户输入的答案是否与存储的验证码匹配：

def lambda_handler(event, context):
    # 获取用户输入的验证码
    user_input = event['request']['challengeAnswer']
    # 取出之前存储的验证码
    stored_code = event['request']['privateChallengeParameters']['verificationCode']
    
    if user_input == stored_code:
        # 验证通过
        event['response']['answerCorrect'] = True
    else:
        # 验证失败
        event['response']['answerCorrect'] = False
    
    return event

注意事项 #

• 要确保这三个触发器都正确关联到你的Cognito User Pool的自定义认证流设置中
• 发送验证码的逻辑需要自行实现（比如用AWS SES发邮件、SNS发短信）
• 需要在User Pool的「认证流程」设置里开启「自定义认证」选项

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