通过 Lambda edge/Dynamodb 实现网站的访问限速

背景

客户是一家做网上直播/点播/游戏相关的服务公司，网站服务于全球各个国家的终端用户。由于历史原因，客户的有些 URL 后端对应比较复杂的数据库操作，每分钟 10 次以内的调用就可以给后台造成非常大的压力。最近运营人员发现时常有攻击者低频率调用这些 URL 来绕过 WAF 限速机制引起数据库过载。对于这些访问，客户希望做到更低频率的限速，而且延时不能太大。

具体需求

客户目前使用的是 AWS CloudFront，源站为本地数据中心服务器，访问者遍布全球各地。
客户要求针对预设置的 URL 和限制（每条 URL 对应不同的限速）进行限速（多条 URL 要求访问不超过 5 次/10 秒/IP）。
客户要求限速在终端客户访问后尽快的进行，否则几秒钟网站访问就会受到影响。

需求分析

根据客户要求，我们的思路如下：

AWS WAF 可以添加 CloudFront 分配，然后对其进行限速，但是目前支持是最低 5 分钟 100 次的速度限制，检测窗口是 30 秒，每隔 30 秒统计前 5 分钟的访问次数，这种方案可以挡住 HTTP 泛洪类的请求。初步将 DDoS 级别的 HTTP 泛洪请求限制到可控的水平。作为前端的第一道大容量，粗粒度的防线。

所以我们可以采用 Lambda Edge 的方法来实现，思路如下：

整体采用 Lambda@Edge + DynamoDB 来实现。
采用 Lambda@Edge，每次访问都需要验证此次访问是否访问需要限制的 URL，如果是就进行计数或者判断。
这里的问题在于，客户端可能会访问任意一个 CloudFront 的 POP 点，所以为了降低访问延时，我们需要在多个物理位置维护一份数据并且保证同步，这里我们就会使用的 Amazon DynamoDB，DynamoDB 的全局表很好的契合了这个需求，可以在毫秒级实现全球数据同步。
DynamoDB 需要维护两张全局表来作为计数的基表，供 Lambda Edge 访问，一张存储必要的访问日志，一张存储被封禁的 IP 列表。

方案拓扑

具体实施步骤

步骤一：创建 Lambda 函数

AWS Lambda@Edge 是一项云计算服务，它允许您在全球范围内的 AWS 内容分发网络（CDN）节点上运行您的代码。您可以使用 Lambda@Edge 来响应用户请求，修改请求和响应，并将结果返回给用户。Lambda@Edge 可以帮助您提高网站的性能和可扩展性，并且您只需为实际使用支付费用。

1. 首先，登录到 AWS 控制台，选择 us-east-1 区域，在服务搜索里面中找到“Lambda”，然后点击进入“Lambda”控制台。

2. 在左侧踩点中点击函数，在 Lambda 控制台的顶部，点击“创建函数”按钮。

3. 在“创建函数”页面中，选择“从头开始制作”选项，输入自定的函数名称，然后选择“Python 3.8”，选择“x86_64”构架，点击“创建函数”。

4. 在接下来函数的具体显示页面，代码源部分，粘贴复制下面的代码，记住输入完毕之后一定要点击“Deploy”：

# 导入 各种 库
import boto3
import time
import datetime

# 创建 需要限速的URL和次数字典以及限速窗口
rate_limits = {
    "/a/a.html": 5,
    "/b/b.html": 10,
}
DURATION = 10

# 设置变量
ACCESSLOG = 'access_logs' #存放访问日志
BANNED_IPS_TABLE = 'banned_ips' #存放封禁的IP清单


# 创建 DynamoDB 客户端
dynamodb = boto3.client("dynamodb")

def lambda_handler(event, context):
    # 获取客户端 IP 和请求的 URL
    request = event["Records"][0]["cf"]["request"]
    uri = request["uri"]
    client_ip = request["clientIp"]

    # 检查客户端是否有权进行请求
    if is_rate_limited(client_ip,uri):
        # 如果客户端被禁止，则返回 HTTP 状态码 429
        return {
            "status": "429",
            "body": "Too Many Requests",
        }

    # 返回请求
    return request


# 读取限制速率的 URL 和次数
def get_rate_limit(uri):
    return rate_limits.get(uri, 0)

# 检查客户端是否有权进行请求
def is_rate_limited(client_ip,uri):
    rate_limit = get_rate_limit(uri)
    # 检查客户端 IP 是否已经被禁止
    response = dynamodb.get_item(
        TableName=BANNED_IPS_TABLE,
        Key={"ip": {"S": client_ip}
            }
    )
    if "Item" in response:
        return True
    else:
        # 获取当前请求的 URL 的限制速率
        # 如果没有限制，则判断失败
        if rate_limit == 0:
           return False
        # 如果有限制，则进行计数
        else:
            # 添加一条访问记录并计算规定时间内的访问次数
            tsnow = int(datetime.datetime.utcnow().timestamp())
            response = dynamodb.put_item(
                TableName=ACCESSLOG,
                Item={"ip_uri": {"S": client_ip+'_'+uri},
                    "ts": {"N": str(tsnow)}
                    }
            )
            timestamp_start = tsnow - DURATION
            timestamp_end = tsnow
            num_requests = dynamodb.query(
                TableName=ACCESSLOG,
                KeyConditionExpression="ip_uri = :pk and ts BETWEEN :start_timestamp AND :end_timestamp",
                ExpressionAttributeValues={
                    ":pk": {"S": client_ip+'_'+uri},
                    ":start_timestamp": {"N": str(timestamp_start)},
                    ":end_timestamp": {"N": str(timestamp_end)},
                }
            )["Count"]
            # 如果请求次数大于限制速率，则将客户端 IP 添加到禁止列表中
            if num_requests > rate_limit:
                dynamodb.put_item(
                    TableName=BANNED_IPS_TABLE,
                    Item={"ip": {"S": client_ip}, "ts": {"N": str(tsnow)}}
                )
            return num_requests > rate_limit

上述代码就实现了限速的逻辑，每次访问请求都会进行验证，查询请求的 IP 在访问产生时的前 10 秒内对需要限速的 URL 的访问次数是否符合规定，如果超限制则进行封禁，如果没超限制则放行。

在上面的代码中，我们定义了一个名为 is_rate_limited 的函数，这个函数可以用来检查客户端是否有权进行请求。函数的参数包括客户端的 IP 地址和请求的 URL。

在函数内部，我们首先通过调用 get_rate_limit 函数来获取客户端请求的 URL 的限制速率。这个是通过查询预定义的词典 rate_limits 来获得相应的结果，统计窗口为 DURATION 参数，这里设置的是 10 秒。

然后，我们使用 DynamoDB 的 get_item 方法来查询封禁 IP 的数据表 banned_ips 检查客户端 IP 是否已经被禁止。如果客户端被禁止，则直接返回 True。

如果客户端没有被禁止，则继续检查客户端请求的 URL 是否有限制速率。如果没有限制，则直接返回 False。如果有限制，则使用 DynamoDB 的 put_item 方法来添加一条访问记录，并使用 query 方法来计算规定时间内的访问次数。

如果访问次数超过限制速率，则使用 put_item 方法来将客户端的 IP 加入封禁 IP 的表中，并返回 True。如果访问次数没有超过限制速率，则直接返回 False。

需要注意的是，如果需要更强的易用性，可以将环境变量配置到 Lambda 函数的环境变量里面，这样可以随时修改。这里为了性能，所以将限速的 URL 和次数字典直接写到 Lambda 函数内部，也可以通过外部来实现。

Lambda 函数的执行角色的权限要注意，首先在信任关系中按如下内容编辑：

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "Service": [
                    "lambda.amazonaws.com",
                    "edgelambda.amazonaws.com"
                ]
            },
            "Action": "sts:AssumeRole"
        }
    ]
}

其次 Lambda 函数的执行角色的需要对 Dynamodb 的表要有访问权限：

点击 Lambda 函数的主界面的“配置”，然后选择权限
可以看到对应的角色名称，点击角色名称链接，进入 IAM 配置界面
添加对应的 Dynamodb 的访问权限即可

步骤二：建立 DynamoDB 的表

这里需要建立两张全局表，在上面的 Lambda 函数中也有描述：

ACCESSLOG = 'access_logs' #存放访问日志
BANNED_IPS_TABLE = 'banned_ips' #存放封禁的IP清单

相关的建表语句（AWS CLI）如下（这里只选择了美东 1 和新加坡的两个副本，具体要按照实际情况建立）：

# 创建 banned_ips 表，用于存放封禁IP清单
aws dynamodb create-table \
--table-name banned_ips \
--attribute-definitions AttributeName=ip,AttributeType=S \
--key-schema AttributeName=ip,KeyType=HASH \
--provisioned-throughput ReadCapacityUnits=5,WriteCapacityUnits=5 \
--stream-specification StreamEnabled=true,StreamViewType=NEW_AND_OLD_IMAGES \
--region us-east-1

aws dynamodb create-table \
--table-name banned_ips \
--attribute-definitions AttributeName=ip,AttributeType=S \
--key-schema AttributeName=ip,KeyType=HASH \
--provisioned-throughput ReadCapacityUnits=5,WriteCapacityUnits=5 \
--stream-specification StreamEnabled=true,StreamViewType=NEW_AND_OLD_IMAGES \
--region ap-southeast-1

aws dynamodb create-global-table \
    --global-table-name banned_ips \
    --replication-group RegionName=ap-southeast-1 RegionName=us-east-1
# 创建 access_logs 表，用于存放必要的访问记录
aws dynamodb create-table \
    --table-name access_logs \--attribute-definitions \
        AttributeName=ip_uri,AttributeType=S \
        AttributeName=ts,AttributeType=N \
    --key-schema AttributeName=ip_uri,KeyType=HASH AttributeName=ts,KeyType=RANGE \
    --provisioned-throughput ReadCapacityUnits=5,WriteCapacityUnits=5 \
    --stream-specification StreamEnabled=true,StreamViewType=NEW_AND_OLD_IMAGES \
    --region us-east-1

aws dynamodb create-table \
    --table-name access_logs \--attribute-definitions \
        AttributeName=ip_uri,AttributeType=S \
        AttributeName=ts,AttributeType=N \
    --key-schema AttributeName=ip_uri,KeyType=HASH AttributeName=ts,KeyType=RANGE \
    --provisioned-throughput ReadCapacityUnits=5,WriteCapacityUnits=5 \
    --stream-specification StreamEnabled=true,StreamViewType=NEW_AND_OLD_IMAGES \
    --region ap-southeast-1  
    
aws dynamodb create-global-table \
    --global-table-name access_logs \
    --replication-group RegionName=ap-southeast-1 RegionName=us-east-1

上述是使用 AWS CLI 建立对应的 Dynamodb 的方式，当然也可以采用控制台的方式创建，具体可以参考 AWS 官方文档，链接如下：https://docs.thinkwithwp.com/zh_cn/amazondynamodb/latest/developerguide/V2globaltables.tutorial.html。

这里为什么需要需要 Dynamodb 全局表，原因如下：

利用了 Amazon DynamoDB 的全球覆盖范围构建，和完全托管的多区域、多主控数据库特性，可以为全局应用程序提供快速的本地读写性能。
Lambda@Edge 是会在 CloudFront 上引用，所以为了降低访问延时，需要就近访问对应的表数据。
上面只是示例，在两个区域进行了部署，有需要的话需要在其他的 region 建立对应的全局表。

步骤三：在 CloudFront 上面部署 Lambda@Edge

1. 首先，登录到 AWS 控制台，选择 us-east-1 区域，在服务搜索里面中找到“Lambda”，然后点击进入“Lambda”控制台。

2. 左侧菜单选择“函数”，点击刚才创建的 Lambda 函数名称，进入函数的主页面。

3. 点击“版本”，发布一个新的版本，并且进入此版本。

4. 在此函数的版本的主页面，点击“添加触发器”的按钮：

a. 选择一个源的部分，下拉菜单选择 CloudFront
b. 选择 Configure new CloudFront trigger
c. 选择需要集成的分配（Distribution）
d. 选择对应的行为（Cache behavior）
e. CloudFront Event 选择查看器请求（Viewer request）
f. 选中“Confirm deploy to Lambda@Edge”
g. 然后点击“添加”

步骤四：客户端验证

此次 POC 中，我们选择的两个页面为/a/a.html和/b/b.html，我们可以先选择访问其他的页面，比如/c/c.html。

可以看到，无论访问多少次，都不会又问题，页面正常显示。

然后我们访问/a/a.html，开始访问没有任何问题，如下图所示：

然后我们快速刷新访问/a/a.html，第五次之后就会出现如下界面信息：

可以看到，过于频繁的访问已经被拒绝了，我们可以查看 Dynamodb 的表的信息，如下所示：

access_logs 表：

banned_ips 表：

可以看到，访问超过限制次数之后，IP 已经被加到封禁列表里，禁止访问。

到此，整个验证完毕，可以看到实现了客户的功能。

Takeaway

此方案优势

1. 响应非常迅速，因为所有的访问都会经过 Lambda@Edge 验证，所以可以实现快速限速和封禁。响应速度一般可以达到超过访问限制后 1~2 秒就能开始阻断。而且统计周期可根据实际调整，上述示例中就是统计访问产生时前 10 秒内对应的访问次数进行限速。

2. 无服务构架，客户不需要事先投入对应的资源而产生费用，在项目初期或者中小流量下成本优势明显。

方案注意事项/优化事项

1. Lambda@Edge 的费用需要注意，如果访问量很大，会产生较高费用。

如果只是针对某些 URL 限速，可以在 CloudFront 分配里面针对不同的 URL 设置不同的行为，只在对应的行为里面启用 Lambda@Edge，而不对所有的访问调用 Lambda@Edge，降低其使用量和成本，如下图所示：

比如上图分配的行为页中，我们可以只对第一个行为配置 Lambda@Edge，那么只有相应的访问才会调用 Lambda，节省成本。

如果采用部分行为才调用 Lambda@Edge，那么实际上 client 只有访问对应 URL 才会进行阻止，但是有些情况下会认为触发了对这些 URL 的访问限制即为攻击者，需要对其进行全局封禁，那么这种情况下还需要结合 WAF 的黑名单，更新封禁 IP 列表的时候，直接更新 WAF 的黑名单即可，更新黑名单的方式可以通过 EventBridge 定时任务读取 banned_ips 表，将 IP 写入预定义的 WAF 黑名单 IP set 即可。结合 WAF 黑名单可以利用 WAF 封禁攻击者 IP 大大减少 Lambda@Edge 的调用次数从而节省成本。

2. Lambda@Edge 的介入会造成访问的延时增加，需要特别注意，具体需要实测看是否满足需求。

3. 应当结合 WAF rate limit 先行做限速，通过低限速的请求再用 Lambda@Edge 做进一步限速，节约费用。

4. 此方案中，可以在不同的判断的给予不同的返回信息，如果需要，可以在 Lambda 函数中自行添加对应的消息。

5. 此方案客户要求进入黑名单之后，先不考虑从黑名单中移除，如果有周期性移除需求，比如要求几个小时之后没有超限制的 IP 可以从 banned_ips 表中移除，实现解封，可以使用 Eventbridge 定时任务，调用 lambda 来实现，这里不做赘述。

6. 方案中由于 POC 阶段访问客户端相对固定，Dynamodb 全局表的使用局限于某些 region，后期需要在多个 region 部署全局表实现全球访问。

亚马逊AWS官方博客