如何为AWS S3动态限制IP地址并24小时后自动解除？

架构方案选择 #

S3原生桶策略无法实现基于请求速率的动态封禁和自动解除，推荐采用CloudFront + AWS WAF + Lambda + EventBridge的组合方案：

• CloudFront：作为S3的前端入口，统一处理所有请求，便于WAF拦截
• AWS WAF：配置速率规则识别高频请求IP，配合IP集实现封禁
• Lambda + EventBridge：完成IP的自动封禁触发和24小时后的自动解除

具体配置步骤 #

1. 将S3桶关联到CloudFront #

• 创建CloudFront分发，源设置为目标S3桶；若需兼顾安全，可使用Origin Access Control (OAC) 替代公开访问，确保仅CloudFront能访问S3
• 配置CloudFront行为，允许GET、HEAD等只读请求，匹配默认路径/*

2. 配置AWS WAF实现IP识别与封禁 #

（1）创建Web ACL并关联CloudFront

• 进入AWS WAF控制台，创建Web ACL，资源类型选择「CloudFront」，关联已创建的CloudFront分发

（2）添加速率限制规则

• 新建规则，类型选择「速率规则」
• 设置速率阈值：100次请求/秒，聚合键选择「IP地址」
• 初始阶段可将动作设为「计数」，验证规则能正确识别高频IP后，改为「阻止」（或结合后续Lambda实现动态封禁）
• 注：WAF速率规则的统计周期为5分钟，实际触发阈值为5分钟内总请求数≥30000次（100605），若需更精细的每秒限流，需结合Lambda实时统计

（3）创建IP集存储封禁IP

• 在WAF控制台创建IP集（例如命名为BlockedIPs_24h），用于存放需要临时封禁的IP地址

（4）添加IP匹配规则阻止封禁IP

• 新建规则，类型选择「IP匹配」
• 匹配条件设置为「IP地址在IP集BlockedIPs_24h中」
• 动作设置为「阻止」

3. 实现自动封禁与24小时解除自动化 #

（1）创建Lambda函数处理IP封禁

函数功能：接收WAF触发的高频IP事件，将IP添加到封禁IP集，并创建24小时后触发的解除任务

import boto3
import os
from datetime import datetime, timedelta

waf_client = boto3.client('wafv2')
eventbridge_client = boto3.client('events')

# 从环境变量读取配置
IP_SET_ARN = os.environ['IP_SET_ARN']
UNBLOCK_LAMBDA_ARN = os.environ['UNBLOCK_LAMBDA_ARN']

def lambda_handler(event, context):
    # 从WAF告警事件中提取违规IP
    ip_address = event['detail']['terminatingRule']['matchDetails'][0]['condition']['value']
    
    # 获取IP集的LockToken（更新IP集必需）
    ip_set = waf_client.get_ip_set(
        Name=IP_SET_ARN.split('/')[-3],
        Scope='CLOUDFRONT',
        Id=IP_SET_ARN.split('/')[-2]
    )
    lock_token = ip_set['LockToken']
    
    # 将IP添加到封禁IP集
    waf_client.update_ip_set(
        Name=IP_SET_ARN.split('/')[-3],
        Scope='CLOUDFRONT',
        Id=IP_SET_ARN.split('/')[-2],
        Addresses=ip_set['Addresses'] + [ip_address],
        LockToken=lock_token
    )
    
    # 计算24小时后的触发时间
    scheduled_time = (datetime.utcnow() + timedelta(hours=24)).isoformat() + 'Z'
    rule_name = f"UnblockIP_{ip_address.replace('.', '_')}"
    
    # 创建EventBridge定时规则，触发解除IP的Lambda
    eventbridge_client.put_rule(
        Name=rule_name,
        ScheduleExpression=f"at({scheduled_time})",
        State='ENABLED'
    )
    
    # 绑定解除Lambda到定时规则
    eventbridge_client.put_targets(
        Rule=rule_name,
        Targets=[
            {
                'Id': '1',
                'Arn': UNBLOCK_LAMBDA_ARN,
                'Input': f'{{"ip_address": "{ip_address}"}}'
            }
        ]
    )
    
    return {'statusCode': 200, 'body': f"IP {ip_address} blocked successfully"}

（2）创建Lambda函数处理IP解除

函数功能：从封禁IP集中移除目标IP，并清理对应的EventBridge定时规则

import boto3
import os

waf_client = boto3.client('wafv2')
eventbridge_client = boto3.client('events')

IP_SET_ARN = os.environ['IP_SET_ARN']

def lambda_handler(event, context):
    ip_address = event['ip_address']
    rule_name = f"UnblockIP_{ip_address.replace('.', '_')}"
    
    # 获取当前IP集信息
    ip_set = waf_client.get_ip_set(
        Name=IP_SET_ARN.split('/')[-3],
        Scope='CLOUDFRONT',
        Id=IP_SET_ARN.split('/')[-2]
    )
    lock_token = ip_set['LockToken']
    
    # 从IP集中移除目标IP
    updated_addresses = [ip for ip in ip_set['Addresses'] if ip != ip_address]
    waf_client.update_ip_set(
        Name=IP_SET_ARN.split('/')[-3],
        Scope='CLOUDFRONT',
        Id=IP_SET_ARN.split('/')[-2],
        Addresses=updated_addresses,
        LockToken=lock_token
    )
    
    # 删除对应的EventBridge定时规则和目标
    eventbridge_client.remove_targets(Rule=rule_name, Ids=['1'])
    eventbridge_client.delete_rule(Name=rule_name)
    
    return {'statusCode': 200, 'body': f"IP {ip_address} unblocked successfully"}

（3）配置触发与权限

• 触发方式：通过CloudWatch告警触发封禁Lambda——当WAF速率规则的计数指标超过阈值时，配置CloudWatch告警，动作选择「调用Lambda函数」
• 权限配置：
  • 给封禁Lambda添加权限：允许调用WAF的get_ip_set、update_ip_set，以及EventBridge的put_rule、put_targets
  • 给解除Lambda添加权限：允许调用WAF的get_ip_set、update_ip_set，以及EventBridge的remove_targets、delete_rule
  • 给CloudWatch添加权限：允许调用封禁Lambda

4. 测试与验证 #

• 使用压测工具（如ab、wrk）模拟每秒100+次请求，验证WAF能识别IP并触发Lambda封禁
• 等待24小时后，检查IP是否从封禁IP集中移除，请求恢复正常访问

注意事项 #

• 若需严格的每秒100次限流，可结合Lambda实时统计请求数（通过CloudWatch Logs或API Gateway日志），但会增加复杂度
• 设置IP集的最大容量，避免过多封禁IP导致性能问题
• 开启WAF日志（存储到CloudWatch Logs），便于排查误封禁情况
• 确保Lambda函数的超时时间足够（建议设置为10-15秒），避免执行失败

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