21. 反爬工程师都会用的手段，IP限制反爬

21. 反爬工程师都会用的手段，IP限制反爬 - 爬虫训练场

news2024/9/21 16:31:27

本篇博客我们实现的案例是 IP 限制反爬，翻译过来就是每个 IP 在规定时间内限制访问次数。
例如，可以限制单 IP 每秒访问 5 次，超过之后就会返回 403 错误。

Flask 实现 IP 限制

- 使用 Flask 插件
- 自定义中间件限制 IP
- 自定义请求钩子

使用 Flask 插件

实战中可以使用第三方模块 Flask-Limiter（Flask 限流器）来实现 IP 限制。在 PyCharm 中直接安装即可。

21. 反爬工程师都会用的手段，IP限制反爬 - 爬虫训练场
使用命令行如下所示：

pip install Flask-Limiter

然后就可以在 Flask 应用中使用 Flask-Limiter 插件，提前建立相关视图函数。

视图函数在 app/school/index.py 文件中；
前端模板文件复制 templates/school/ajax_list.html 文件为 ajax_list3.html 即可。

原接口请求地址是 /ss/api2，现在修改为 /ss/api3，同时在 index.py 中复制原接口函数。

@s.route('api3')
def school_api3():
    page = int(request.args.get("page", 1))

    pagination = pagination_object(page)

    return jsonify(pagination)

后续我们的核心逻辑都将在上述函数进行改写，导入 Flask-Limiter 插件。

from flask_limiter import Limiter
from flask_limiter.util import get_remote_address,get_ipaddr

上述代码在 app/__init__.py 中进行实现，实例化操作也在该文件创建。

limiter = Limiter(app, key_func=get_ipaddr)

使用的时候，只需要引入该对象，然后添加装饰器即可。

# 从 app 中导入 limiter 对象
from app import limiter

@s.route('api3')
@limiter.limit("3/second")
def school_api3():
    page = int(request.args.get("page", 1))

    pagination = pagination_object(page)

    return jsonify(pagination)

上述代码 @limiter.limit("3/second") 表示限制每秒3次请求，有两种方式表示速率限制，如下所示。

“100 per day”、“50 per hour”、“20 per minute”、“1 per second”
“100/day”、“50/hour”、“20/minute”、“1/second”

装饰器可以单一修饰，也可以多个修饰，分别如下所示。

单一修饰：限制字符串可以是单个限制，也可以是定界符分隔的字符串。

@app.route("api3")
@limiter.limit("100/day;10/hour;1/minute")
def my_route()

多个装饰器：限制字符串可以是单个限制，也可以是定界符分隔的字符串，也可以是两者的组合。

@app.route("api3")
@limiter.limit("100/day")
@limiter.limit("10/hour")
@limiter.limit("1/minute")
def my_route():

除此之外，在实例化限速器对象的时候，还用到了限制域内容，特指根据什么进行限制。

get_remote_address：根据请求的 remote_address；
get_ipaddr：根据请求 X-Forwarded-For 标头中的最后IP地址。

接下来我们需要拿到生成环境做一下测试，查看一下捕获到的 IP 是否正确，在爬虫训练冲项目中，可以正确获取到 IP 值，如果你使用了反向代理，那么 Flask-Limiter 将使用默认的 key_func，即 get_remote_address 来获取客户端的 IP 地址。在这种情况下，Flask-Limiter 将会看到反向代理的 IP 地址，而不是实际客户端的 IP 地址。

为了解决这个问题，你可以使用自定义的 key_func 函数，用来检索客户端的真实 IP 地址。例如：

from flask import request

def get_real_ip():
    if request.headers.getlist("X-Forwarded-For"):
        return request.headers.getlist("X-Forwarded-For")[0]
    return request.remote_addr

limiter = Limiter(app, key_func=get_real_ip)

这样，Flask-Limiter 就会使用客户端的真实 IP 地址来进行限制，而不是反向代理的 IP 地址。

请注意，如果你使用了多个反向代理，那么上述 get_real_ip() 函数可能无法获取客户端的真实 IP 地址。在这种情况下，你可能需要检查所有的 “X-Forwarded-For” 头，以找到客户端的真实 IP 地址。例如：

def get_real_ip():
    x_forwarded_for = request.headers.getlist("X-Forwarded-For")
    if x_forwarded_for:
        return x_forwarded_for[-1]
    return request.remote_addr

这样，你就可以使用 Flask-Limiter 来限制客户端的真实 IP 地址，即使在反向代理的情况下也是如此。

实战中有时候需要记录 IP 访问此处，判断该 IP 访问次数是否达到上线，Limiter 默认使用的是内存记录，这在生产环境中显然是不现实的，所以这里补充一个小知识点：Flask-Limiter 使用 redis 作为数据储存容易。

自定义中间件限制 IP

你也可以自己实现一个中间件来实现 IP 限制。

中间件是一个可以在请求和响应之间插入的代码，通常用于实现某些功能，比如认证、IP 限制等。

在 Flask 中，中间件可以通过装饰器的方式使用，例如：

def ip_limiter(app):
    def middleware(next):
        def wrapper(*args, **kwargs):
            # 在这里处理 IP 限制逻辑
            return next(*args, **kwargs)
        return wrapper
    return middleware

app.register_middleware(ip_limiter)

后续操作是在 middleware() 函数中实现 IP 限制逻辑即可。

自定义请求钩子

Flask 还提供了一种在请求之前插入代码的机制，叫做请求钩子。使用方法如下：

@app.before_request
def ip_limiter():
    # 在这里处理 IP 限制逻辑
    pass

在 ip_limiter() 函数中，你可以获取请求的 IP 地址，然后根据你的限制规则来决定是否允许这个请求继续执行。

例如，你可以使用 Flask 提供的 request 对象来获取请求的 IP 地址：

@app.before_request
def ip_limiter():
    ip = request.remote_addr
    # 在这里处理 IP 限制逻辑
    pass

注意，如果你的应用部署在有反向代理的环境中，那么 request.remote_addr 可能不是真实的客户端 IP，而是反向代理的 IP。在这种情况下，你需要使用特定的方法来获取客户端的真实 IP。

最后，如果你希望限制某个 IP 访问次数，你可以使用一个字典来记录每个 IP 的访问次数，每次请求时递增计数器，然后判断是否超过限制。

ip_counter = {}

@app.before_request
def ip_limiter():
    ip = request.remote_addr
    if ip not in ip_counter:
        ip_counter[ip] = 0
    ip_counter[ip] += 1
    if ip_counter[ip] > 10:
        # 超过限制，拒绝请求
        abort(429)