如何在Lambda Serverless架构下基于AWS资源实现用户认证流程自动化

• 直接用AWS官方Python SDK boto3 操作Cognito原生接口，不需要引入第三方身份服务商，完全复用你现有AWS资源栈
• 按业务需求封装三个核心流程：用户注册、注册确认、身份认证拿令牌，全部逻辑走代码实现，无需手动执行CLI命令
• 拿到有效令牌后，直接在代码中携带令牌调用APIGateway暴露的REST API，可直接实现全流程自动化测试，也可以封装为前端可调用的后端接口给用户自主操作
• 如果需要支持用户自主完成注册确认，把你当前用的管理员确认逻辑替换为用户收取验证码后自主提交确认的逻辑即可，Cognito原生支持邮箱/短信验证码发送和校验能力

首先执行以下命令安装依赖：

pip install boto3 requests

核心实现代码如下：

import boto3
import requests
import os

# 配置参数，建议从环境变量读取，不要硬编码
REGION = os.getenv("AWS_REGION", "your-region")
COGNITO_CLIENT_ID = os.getenv("COGNITO_CLIENT_ID", "your-client-id")
COGNITO_USER_POOL_ID = os.getenv("COGNITO_USER_POOL_ID", "your-user-pool-id")
API_ENDPOINT = os.getenv("API_ENDPOINT", "your-apigw-url")

# 初始化Cognito客户端
cognito_client = boto3.client("cognito-idp", region_name=REGION)

def user_sign_up(username: str, password: str, user_attrs: list = None) -> dict:
    """用户注册"""
    try:
        response = cognito_client.sign_up(
            ClientId=COGNITO_CLIENT_ID,
            Username=username,
            Password=password,
            UserAttributes=user_attrs if user_attrs else []
        )
        return {"status": "success", "data": response}
    except Exception as e:
        return {"status": "error", "msg": str(e)}

def admin_confirm_user_sign_up(username: str) -> dict:
    """管理员确认用户注册，适用于测试场景或者后台审核注册的业务，用户自主注册可替换为验证码确认逻辑"""
    try:
        response = cognito_client.admin_confirm_sign_up(
            UserPoolId=COGNITO_USER_POOL_ID,
            Username=username
        )
        return {"status": "success", "data": response}
    except Exception as e:
        return {"status": "error", "msg": str(e)}

# 如果是用户自主注册用这个确认方法替换上面的管理员确认
# def user_confirm_sign_up(username: str, confirm_code: str) -> dict:
#     try:
#         response = cognito_client.confirm_sign_up(
#             ClientId=COGNITO_CLIENT_ID,
#             Username=username,
#             ConfirmationCode=confirm_code
#         )
#         return {"status": "success", "data": response}
#     except Exception as e:
#         return {"status": "error", "msg": str(e)}

def get_user_token(username: str, password: str) -> dict:
    """获取用户身份令牌，用户自助场景把AuthFlow改成USER_PASSWORD_AUTH即可，移除UserPoolId参数"""
    try:
        response = cognito_client.admin_initiate_auth(
            UserPoolId=COGNITO_USER_POOL_ID,
            ClientId=COGNITO_CLIENT_ID,
            AuthFlow="ADMIN_NO_SRP_AUTH",
            AuthParameters={
                "USERNAME": username,
                "PASSWORD": password
            }
        )
        token_set = response["AuthenticationResult"]
        return {
            "status": "success",
            "id_token": token_set["IdToken"],
            "access_token": token_set["AccessToken"],
            "refresh_token": token_set["RefreshToken"]
        }
    except Exception as e:
        return {"status": "error", "msg": str(e)}

def call_protected_api(token: str, method: str = "GET", payload: dict = None) -> dict:
    """携带令牌调用受保护的API接口"""
    headers = {
        "Authorization": token,
        "Content-Type": "application/json"
    }
    try:
        if method.upper() == "GET":
            response = requests.get(API_ENDPOINT, headers=headers, params=payload)
        else:
            response = requests.post(API_ENDPOINT, headers=headers, json=payload)
        return {"status": response.status_code, "data": response.json()}
    except Exception as e:
        return {"status": "error", "msg": str(e)}

# 测试全流程
if __name__ == "__main__":
    test_username = "test_user"
    test_password = "Test@123456"
    # 注册
    sign_up_res = user_sign_up(test_username, test_password)
    print("注册结果：", sign_up_res)
    if sign_up_res["status"] == "success":
        # 确认注册
        confirm_res = admin_confirm_user_sign_up(test_username)
        print("确认结果：", confirm_res)
        if confirm_res["status"] == "success":
            # 拿令牌
            token_res = get_user_token(test_username, test_password)
            print("令牌获取结果：", token_res)
            if token_res["status"] == "success":
                # 调用API
                api_res = call_protected_api(token_res["id_token"], "GET")
                print("API调用结果：", api_res)

• 生产环境禁止使用ADMIN_NO_SRP_AUTH认证流，该流程仅用于后台管理场景，普通用户自助认证请改用USER_PASSWORD_AUTH流程，调用initiate_auth方法即可，不需要传入UserPoolId参数
• 测试阶段如果需要跳过验证码确认，可在Cognito用户池配置中开启「允许管理员创建用户并立即确认」，不需要额外配置验证码发送渠道
• 所有敏感配置请勿硬编码到代码中，本地测试可通过环境变量或者本地配置文件读取，部署到Lambda时直接为Lambda执行角色绑定Cognito相关操作权限即可，不需要配置AK/SK

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