如何在AWS API Gateway v2 Websocket中配置多个查询字符串授权器

API Gateway Websocket 不支持给单个路由绑定多个自定义授权器，所以你需要在现有的REQUEST类型授权器中整合多查询字符串的验证逻辑，具体实现如下：

1. 调整Terraform授权器配置 #

把需要验证的所有查询字符串参数都加入identity_sources列表，这样API Gateway会把这些参数传递给Lambda授权器。比如新增一个token查询参数的验证：

resource "aws_apigatewayv2_authorizer" "authorizer" {
  api_id                     = aws_apigatewayv2_api.api_gateway_websocket.id
  authorizer_type            = "REQUEST"
  authorizer_uri             = var.lambda_authorizer_uri
  # 列出所有需要验证的查询字符串参数
  identity_sources           = [
    "route.request.querystring.authorization", 
    "route.request.querystring.route",
    "route.request.querystring.token" # 新增的待验证参数
  ]
  name                       = var.lambda_authorizer_name
  authorizer_credentials_arn = var.authorizer_credentials_arn
}

2. 路由配置保持不变 #

原有的Connect路由只需要绑定这一个授权器即可，无需额外修改：

resource "aws_apigatewayv2_route" "ConnectRoute" {
  api_id             = aws_apigatewayv2_api.api_gateway_websocket.id
  route_key          = "$connect"
  operation_name     = "ConnectRoute"
  authorization_type = "CUSTOM"
  authorizer_id      = aws_apigatewayv2_authorizer.authorizer.id
  depends_on = [
    aws_apigatewayv2_authorizer.authorizer
  ]
}

3. 改造Lambda授权器逻辑 #

在Lambda函数中，从事件里取出所有查询参数，逐个执行验证逻辑，只要有一个验证失败就返回拒绝响应。以下是Python示例：

import json

def lambda_handler(event, context):
    # 提取所有查询字符串参数
    query_params = event.get('requestContext', {}).get('request', {}).get('queryStringParameters', {})
    auth_token = query_params.get('authorization')
    route_param = query_params.get('route')
    new_token = query_params.get('token')

    # 验证authorization参数
    if not validate_auth_token(auth_token):
        return deny_access("Invalid authorization token")
    
    # 验证route参数的合法性
    if not validate_route(route_param):
        return deny_access("Route not allowed")
    
    # 验证新增的token参数
    if not validate_new_token(new_token):
        return deny_access("Invalid additional token")
    
    # 所有验证通过，允许连接
    return allow_access(event['requestContext']['connectionId'])

# 自定义authorization验证逻辑
def validate_auth_token(token):
    return token and token.startswith('valid_jwt_')

# 自定义route验证逻辑
def validate_route(route):
    allowed_routes = ['chat', 'broadcast']
    return route in allowed_routes

# 自定义新增token的验证逻辑
def validate_new_token(token):
    return token and len(token) == 16

# 生成允许访问的响应
def allow_access(connection_id):
    return {
        'principalId': connection_id,
        'policyDocument': {
            'Version': '2012-10-17',
            'Statement': [{
                'Action': 'execute-api:Invoke',
                'Effect': 'Allow',
                'Resource': '*'
            }]
        }
    }

# 生成拒绝访问的响应
def deny_access(message):
    return {
        'principalId': '*',
        'policyDocument': {
            'Version': '2012-10-17',
            'Statement': [{
                'Action': 'execute-api:Invoke',
                'Effect': 'Deny',
                'Resource': '*'
            }]
        },
        'context': {'error': message}
    }

备选方案：拆分授权逻辑（如果需要独立维护） #

如果必须把不同的验证逻辑拆成多个独立Lambda，可以在主授权器Lambda里依次调用这些子授权Lambda，只要有一个验证失败就返回拒绝。不过这种方式会增加请求延迟，建议只在必要时使用。

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