AWS Cognito、Twilio与Lex集成：认证用户权限控制技术问询

看起来你已经搭好了核心链路（API Gateway ↔ Lambda ↔ Lex），也加了Cognito授权的初步配置，接下来只需要补全用户身份关联、Twilio请求的身份验证适配和端到端流程闭环这几块，就能实现仅认证用户可交互的需求。下面是一步步的实操方案：

1. 先搞定「Twilio手机号 ↔ Cognito用户」的绑定（无服务器实现） #

因为Twilio是通过用户手机号识别会话的，你需要把用户的Twilio手机号和他们的Cognito身份关联起来，这一步用DynamoDB+Lambda就能搞定，不用单独搭后端：

• 创建一个DynamoDB表，比如叫UserPhoneBindings，主键用phone_number（格式要和Twilio的From参数一致，比如+1234567890），字段包含cognito_user_id、id_token、expires_at（记录token过期时间）。
• 写一个Lambda函数（比如BindUserPhone），用来处理绑定请求：
  • 给这个函数配一个带Cognito授权的API Gateway端点，确保只有已认证的Cognito用户能调用。
  • 函数逻辑：接收用户传入的手机号，用AWS SDK调用Cognito的GetUser接口验证当前用户的ID Token合法性，验证通过后把手机号和Cognito用户信息写入DynamoDB。

2. 调整API Gateway的授权逻辑适配Twilio #

Twilio的Webhook请求默认不会带Cognito Token，所以你原来的Cognito授权器没法直接用，建议换成自定义Lambda授权器：

• 写一个授权器Lambda函数（比如TwilioAuthChecker）：
  • 从Twilio的请求里提取From参数（用户手机号）。
  • 去DynamoDB查这个手机号对应的Cognito记录：如果存在且ID Token未过期，就返回允许访问的IAM策略；如果不存在或token过期，返回拒绝策略。
  • 别忘了加Twilio请求签名验证！用Twilio的SDK校验请求确实来自Twilio，防止恶意伪造请求。
• 在API Gateway里把原来的Cognito授权器换成这个自定义授权器，确保所有Twilio的请求都先过这层校验。

3. 完善Lex交互的Lambda函数 #

原来的Lambda只负责调用Lex，现在要加上身份关联和格式转换：

• 从Twilio请求里提取From（手机号）和Body（用户消息）。
• 从DynamoDB拿到对应的cognito_user_id，用这个ID作为Lex的userId参数调用Lex的PostText接口——这样Lex可以基于Cognito用户ID维持会话，不会串号。
• 把Lex返回的响应转换成Twilio能识别的XML格式（或者JSON，取决于你API Gateway的配置），比如：

  import boto3
  from twilio.request_validator import RequestValidator
  import os
  
  dynamodb = boto3.client('dynamodb')
  lex_client = boto3.client('lex-runtime')
  twilio_validator = RequestValidator(os.environ['TWILIO_AUTH_TOKEN'])
  
  def lambda_handler(event, context):
      # 先验证Twilio请求合法性
      if not twilio_validator.validate(
          event['headers']['X-Twilio-Signature'],
          event['requestContext']['path'],
          event['queryStringParameters']
      ):
          return {'statusCode': 403, 'body': 'Invalid request'}
  
      # 提取Twilio参数
      from_number = event['queryStringParameters']['From']
      user_message = event['queryStringParameters']['Body']
  
      # 获取绑定的Cognito用户ID
      db_response = dynamodb.get_item(
          TableName='UserPhoneBindings',
          Key={'phone_number': {'S': from_number}}
      )
      if 'Item' not in db_response:
          return {
              'statusCode': 200,
              'headers': {'Content-Type': 'application/xml'},
              'body': '<Response><Message>请先完成认证绑定后再发送消息哦～</Message></Response>'
          }
      cognito_user_id = db_response['Item']['cognito_user_id']['S']
  
      # 调用Lex机器人
      lex_response = lex_client.post_text(
          botName='YourLexBotName',
          botAlias='YourLexBotAlias',
          userId=cognito_user_id,
          inputText=user_message
      )
  
      # 返回Twilio格式的响应
      return {
          'statusCode': 200,
          'headers': {'Content-Type': 'application/xml'},
          'body': f'<Response><Message>{lex_response["message"]}</Message></Response>'
      }
  

4. 给用户搭一个简单的认证绑定入口 #

因为没有后端，你可以用S3+CloudFront搭一个静态网页，集成Cognito Hosted UI：

• 在S3上上传一个简单的静态页面，包含Cognito登录/注册按钮（用Cognito Hosted UI的链接），登录成功后调用之前的绑定API，让用户输入手机号完成绑定。
• 或者直接用Cognito的用户属性，把手机号作为必填属性，用户注册时直接填写，绑定函数自动从Cognito用户属性里取手机号，不用用户再输入。

5. 测试闭环流程 #

• 先测绑定：用Cognito注册用户，完成绑定，检查DynamoDB是否有对应记录。
• 再测交互：用绑定过的手机号发消息到Twilio号码，看是否能收到Lex的回复；用未绑定的手机号发消息，看是否被拦截并收到提示。
• 最后测过期场景：等ID Token过期后发消息，看是否会提示重新绑定。

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