如何基于AWS Lambda实现HTTP API Gateway异常IP自动拉黑？

针对HTTP API Gateway + Python 3.9 Lambda的IP频次检测与自动拉黑需求，可通过DynamoDB存储+Lambda授权器+定时清理的组合方案实现，具体如下：

1. 核心组件选型 #

• DynamoDB：存储IP调用频次数据和黑名单，无服务器架构适配性强，读写性能满足高频请求场景。
• Lambda授权器：作为HTTP API Gateway的前置校验节点，负责IP黑名单检查、调用频次统计与阈值判断。
• CloudWatch Events + Lambda：定时清理过期的频次数据，避免存储冗余。

2. 具体实现步骤 #

2.1 部署DynamoDB表 #

创建两张表（或单表用不同分区键区分）：

• IPCallFrequency：分区键ip，排序键function_name，字段包含call_count（调用次数）、last_call_time（最后调用时间戳）。
• IPBlacklist：分区键ip，字段包含blocked_time（拉黑时间戳）、expire_time（自动解封时间戳）。

2.2 改造Lambda授权器 #

授权器需完成三个核心逻辑：提取客户端IP、黑名单校验、频次统计与阈值触发拉黑。以下是Python实现代码：

import boto3
from datetime import datetime, timedelta

dynamodb = boto3.resource('dynamodb')
freq_table = dynamodb.Table('IPCallFrequency')
blacklist_table = dynamodb.Table('IPBlacklist')

# 可配置参数
CALL_THRESHOLD = 100  # 时间窗口内允许的最大调用次数
TIME_WINDOW = 300     # 统计时间窗口（单位：秒，示例为5分钟）
BLOCK_DURATION = 3600 # IP拉黑时长（单位：秒，示例为1小时）

def lambda_handler(event, context):
    # 从HTTP API请求上下文提取客户端IP和目标Lambda标识
    source_ip = event['requestContext']['http']['sourceIp']
    # 此处根据你的路由规则提取关联的Lambda名称，示例为从路由键末尾截取
    function_name = event['requestContext']['routeKey'].split('/')[-1]

    # 第一步：检查IP是否在有效黑名单内
    blacklist_item = blacklist_table.get_item(Key={'ip': source_ip}).get('Item')
    if blacklist_item:
        if datetime.now().timestamp() < blacklist_item['expire_time']:
            # IP仍在拉黑期，返回拒绝授权
            return {
                'isAuthorized': False,
                'context': {'message': 'IP blocked due to excessive requests'}
            }
        else:
            # 黑名单已过期，移除记录
            blacklist_table.delete_item(Key={'ip': source_ip})

    # 第二步：更新调用频次并检查阈值
    now_ts = datetime.now().timestamp()
    cutoff_ts = now_ts - TIME_WINDOW

    # 原子更新频次数据，避免并发计数错误
    update_result = freq_table.update_item(
        Key={'ip': source_ip, 'function_name': function_name},
        UpdateExpression='SET call_count = if_not_exists(call_count, :zero) + :inc, last_call_time = :now',
        ConditionExpression='last_call_time > :cutoff OR attribute_not_exists(last_call_time)',
        ExpressionAttributeValues={
            ':inc': 1,
            ':zero': 0,
            ':now': now_ts,
            ':cutoff': cutoff_ts
        },
        ReturnValues='UPDATED_NEW'
    )

    # 若未触发条件（超过时间窗口），重置计数为1
    if 'Attributes' not in update_result:
        freq_table.put_item(Item={
            'ip': source_ip,
            'function_name': function_name,
            'call_count': 1,
            'last_call_time': now_ts
        })
        current_count = 1
    else:
        current_count = update_result['Attributes']['call_count']

    # 第三步：判断是否触发拉黑条件
    if current_count >= CALL_THRESHOLD:
        expire_ts = now_ts + BLOCK_DURATION
        blacklist_table.put_item(Item={
            'ip': source_ip,
            'blocked_time': now_ts,
            'expire_time': expire_ts
        })
        return {
            'isAuthorized': False,
            'context': {'message': 'IP blocked due to excessive requests'}
        }

    # 正常请求，返回授权通过
    return {
        'isAuthorized': True,
        'context': {'current_call_count': current_count}
    }

2.3 定时清理过期频次数据 #

创建CloudWatch Events定时规则（例如每小时执行一次），触发以下Lambda清理IPCallFrequency中超过时间窗口的旧数据：

import boto3
from datetime import datetime

dynamodb = boto3.resource('dynamodb')
freq_table = dynamodb.Table('IPCallFrequency')
TIME_WINDOW = 300  # 与授权器的时间窗口保持一致

def lambda_handler(event, context):
    cutoff_ts = datetime.now().timestamp() - TIME_WINDOW

    # 扫描获取过期的频次记录
    scan_response = freq_table.scan(
        FilterExpression='last_call_time < :cutoff',
        ExpressionAttributeValues={':cutoff': cutoff_ts},
        ProjectionExpression='ip, function_name'
    )

    # 批量删除过期数据
    with freq_table.batch_writer() as batch:
        for item in scan_response['Items']:
            batch.delete_item(Key={'ip': item['ip'], 'function_name': item['function_name']})

    return {'deleted_records_count': len(scan_response['Items'])}

3. 关键注意事项 #

• 并发安全：依赖DynamoDB的update_item原子操作，避免多请求同时更新导致计数错误。
• IP准确性：若API Gateway前有CloudFront等代理，需确保sourceIp为真实客户端IP，可通过配置X-Forwarded-For头处理。
• 性能优化：DynamoDB开启按需读写模式，适配流量波动；给频次表添加全局二级索引优化查询。
• 阈值调整：根据业务实际流量调整CALL_THRESHOLD和TIME_WINDOW，避免误拉黑正常用户。

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