Flask API 日志过滤：通过白名单机制优化请求日志管理

聖光之護

发布时间：2025-09-28 13:31:10

587人浏览过

来源于php中文网

原创

flask api 日志过滤：通过白名单机制优化请求日志管理

本文详细介绍了如何在 Flask API 中通过实现自定义 WSGI 请求处理器，利用白名单机制过滤不必要的请求日志，从而有效应对日志被垃圾请求淹没的问题。文章着重讲解了动态获取 API 路由端点、正确配置日志过滤逻辑以及解决初始化时序问题的关键步骤，并探讨了在生产环境中可能遇到的挑战及替代方案。

引言：日志管理的挑战与白名单策略

在开发和维护 Flask API 时，请求日志是诊断问题、监控性能和理解用户行为的关键工具。然而，当 API 面临大量无效或恶意请求（如爬虫、扫描器等）时，日志文件可能会迅速膨胀，充满无用信息，严重影响日志的可读性和后续分析。为了解决这一问题，一种有效的策略是采用“白名单”机制，即只记录特定、已知的 API 端点的请求日志，而忽略其他所有请求。

本教程将指导您如何通过定制 Flask 底层的 WSGI 请求处理器，实现这一白名单日志过滤功能。

定制 WSGI 请求处理器实现日志过滤

Flask 默认使用 werkzeug.serving.WSGIRequestHandler 来处理 HTTP 请求并记录日志。我们可以通过重写其 log_request 方法来插入自定义的日志过滤逻辑。

1. 核心过滤逻辑

首先，我们需要一个函数来修改 WSGIRequestHandler.log_request 方法。这个函数将保存原始的 log_request 方法，然后用我们自己的逻辑替换它。

import re
from flask import Flask
from werkzeug import serving

# 假设您的Flask应用实例名为app
app = Flask(__name__)

def restrict_access_logs(app_instance):
    """
    修改WSGIRequestHandler的log_request方法，实现基于白名单的日志过滤。
    """
    # 保存原始的log_request方法
    parent_log_request = serving.WSGIRequestHandler.log_request

    # 动态获取所有已注册的端点名称
    # 注意：这里获取的是端点名称（endpoint），而不是完整的URL路径
    permitted_endpoints = [rule.endpoint for rule in app_instance.url_map.iter_rules()]

    def log_request(self, *args, **kwargs):
        """
        自定义的log_request方法，根据白名单判断是否记录日志。
        """
        # 检查请求路径是否匹配白名单中的任一端点
        # 假设所有API路径都以 /api/v1/ 开头，且端点名称与路径的最后一部分对应
        # 例如，如果端点是 'hello'，则匹配 '/api/v1/hello' 或 '/api/v1/hello/anything'
        is_whitelisted = False
        for endpoint in permitted_endpoints:
            # 排除Flask自带的'static'端点，通常不需要记录其日志
            if endpoint == 'static':
                continue

            # 构建正则表达式来匹配请求路径
            # 这里以 '/api/v1/' 作为前缀示例，请根据您的实际API路径结构调整
            # 确保正则表达式能正确匹配您的URL结构
            pattern = rf"/api/v1/{re.escape(endpoint)}(/.*)?$"
            if re.match(pattern, self.path):
                is_whitelisted = True
                break

        # 如果请求路径在白名单中，则调用原始的log_request方法记录日志
        if is_whitelisted:
            parent_log_request(self, *args, **kwargs)

    # 将WSGIRequestHandler的log_request方法替换为我们自定义的函数
    serving.WSGIRequestHandler.log_request = log_request

代码解析：

parent_log_request = serving.WSGIRequestHandler.log_request：保存了 Werkzeug 默认的日志记录方法，以便在白名单匹配时调用。
permitted_endpoints = [rule.endpoint for rule in app_instance.url_map.iter_rules()]：这是动态获取所有已注册 API 端点名称的关键。app.url_map.iter_rules() 会迭代所有路由规则，每个规则对象都有一个 endpoint 属性，即路由函数的名字或 @app.route 装饰器中 endpoint 参数指定的值。
log_request(self, *args, **kwargs)：这是我们自定义的日志处理逻辑。它接收 WSGIRequestHandler 实例 (self) 和其他参数。
re.match(pattern, self.path)：使用正则表达式匹配当前请求的路径 (self.path)。这里构建的模式 rf"/api/v1/{re.escape(endpoint)}(/.*)?$" 假设您的 API 路径都以 /api/v1/ 开头，并且端点名称直接跟在后面。re.escape() 用于转义端点名称中可能存在的特殊字符。(/. *)? 允许匹配端点后可能存在的子路径。请务必根据您的实际 API 路由结构调整此正则表达式。
serving.WSGIRequestHandler.log_request = log_request：将自定义的 log_request 函数赋值给 WSGIRequestHandler.log_request，从而覆盖默认行为。

2. 注册示例 API 端点

为了测试上述逻辑，我们创建一些示例 API 端点：

易网商务 Build 20030730 OEM版

优化了部分代码及一些BUG.，提高了浏览速度，可以通过会员助手自由管理各种信息，修正了反馈信息及询价订单错误，增加了自助建站系统(16种模板可选)，增加在线管理开通域名主机邮局系统，强大的备份功能可以轻松备份压缩恢复数据，后台增加验证码和日志功能，分类管理更详细，更安全默认的管理员帐户是：admin密码是：admin

下载

# 示例API路由
@app.route('/api/v1/hello', methods=['GET'])
def hello():
    return "Hello, Flask!"

@app.route('/api/v1/getEvidencesByProductID/', methods=['GET'])
def getEvidencesByProductID(product_id):
    return f"Fetching evidences for product ID: {product_id}"

@app.route('/api/v1/testpoint', methods=['GET'])
def testpoint():
    ep_list = [rule.endpoint for rule in app.url_map.iter_rules()]
    ep_str = ", ".join(ep_list)
    return f"Available Endpoints: {ep_str}"

@app.route('/api/v1/unlisted', methods=['GET'])
def unlisted_endpoint():
    return "This endpoint should not be logged."

3. 关键：函数调用时机

一个非常重要的注意事项是 restrict_access_logs() 函数的调用时机。 如果在所有路由定义之前调用它，app.url_map.iter_rules() 将无法获取到完整的端点列表，导致白名单为空或不完整。

正确的做法是，在所有 app.route 装饰器定义完毕之后，再调用 restrict_access_logs() 函数。

if __name__ == '__main__':
    # ... (上面定义的 app 实例和路由) ...

    # 在所有路由定义完成后，调用日志限制函数
    restrict_access_logs(app) 

    # 运行Flask应用
    app.run(debug=True)

通过将 restrict_access_logs(app) 放在所有 @app.route 装饰器之后，可以确保 app.url_map 包含了所有已注册的路由信息，从而动态生成的白名单是完整的。

完整代码示例

import re
from flask import Flask
from werkzeug import serving

app = Flask(__name__)

def restrict_access_logs(app_instance):
    """
    修改WSGIRequestHandler的log_request方法，实现基于白名单的日志过滤。
    """
    parent_log_request = serving.WSGIRequestHandler.log_request

    # 动态获取所有已注册的端点名称
    permitted_endpoints = [rule.endpoint for rule in app_instance.url_map.iter_rules()]
    print(f"Whitelisted Endpoints: {permitted_endpoints}") # 调试输出

    def log_request(self, *args, **kwargs):
        """
        自定义的log_request方法，根据白名单判断是否记录日志。
        """
        is_whitelisted = False
        for endpoint in permitted_endpoints:
            if endpoint == 'static': # 排除Flask自带的'static'端点
                continue

            # 根据您的API路径结构调整正则表达式
            # 例如，如果您的API前缀是/api/v1/
            pattern = rf"/api/v1/{re.escape(endpoint)}(/.*)?$"
            if re.match(pattern, self.path):
                is_whitelisted = True
                break

        if is_whitelisted:
            parent_log_request(self, *args, **kwargs)

    serving.WSGIRequestHandler.log_request = log_request

# 示例API路由定义
@app.route('/api/v1/hello', methods=['GET'])
def hello():
    return "Hello, Flask!"

@app.route('/api/v1/getEvidencesByProductID/', methods=['GET'])
def getEvidencesByProductID(product_id):
    return f"Fetching evidences for product ID: {product_id}"

@app.route('/api/v1/testpoint', methods=['GET'])
def testpoint():
    ep_list = [rule.endpoint for rule in app.url_map.iter_rules()]
    ep_str = ", ".join(ep_list)
    return f"Available Endpoints: {ep_str}"

@app.route('/api/v1/unlisted', methods=['GET'])
def unlisted_endpoint():
    return "This endpoint should not be logged."

@app.route('/no-api-prefix', methods=['GET'])
def no_api_prefix():
    return "This endpoint has no /api/v1/ prefix."

if __name__ == '__main__':
    # 确保在所有路由定义之后调用此函数
    restrict_access_logs(app) 
    app.run(debug=True)

测试方法：

运行上述 Flask 应用。
访问 http://127.0.0.1:5000/api/v1/hello：应该会在控制台看到日志输出。
访问 http://127.0.0.1:5000/api/v1/getEvidencesByProductID/123：应该会在控制台看到日志输出。
访问 http://127.0.0.1:5000/api/v1/unlisted：应该会在控制台看到日志输出（因为 unlisted_endpoint 的端点名是 unlisted_endpoint，而我们正则匹配的是 /api/v1/unlisted。这说明正则表达式和端点名称的匹配需要精确。如果端点名称是 unlisted，那么 /api/v1/unlisted 就会被匹配）。
访问 http://127.0.0.1:5000/api/v1/nonexistent：不应该在控制台看到日志输出，因为它不在白名单中。
访问 http://127.0.0.1:5000/random/path：不应该在控制台看到日志输出。
访问 http://127.0.0.1:5000/no-api-prefix：不应该在控制台看到日志输出，因为它的路径不符合 /api/v1/ 的前缀模式。

关于unlisted_endpoint的日志行为说明： 在上述示例中，@app.route('/api/v1/unlisted', methods=['GET']) 的端点名称默认为函数名 unlisted_endpoint。而我们的正则表达式 rf"/api/v1/{re.escape(endpoint)}(/.*)?$" 会尝试匹配 /api/v1/unlisted_endpoint。因此，访问 /api/v1/unlisted 将不会被匹配，从而不会记录日志。 如果希望 /api/v1/unlisted 被记录，您有两种选择：

在 app.route 装饰器中显式指定端点名：@app.route('/api/v1/unlisted', methods=['GET'], endpoint='unlisted')。
调整正则表达式，使其更灵活地匹配路径的最后一部分，或者直接使用 rule.rule （即路由路径字符串）进行匹配，而非 rule.endpoint。但使用 rule.endpoint 配合适当的正则匹配通常更稳健，因为它与业务逻辑的命名更一致。

注意事项与局限性

部署环境的差异： 这种直接修改 werkzeug.serving.WSGIRequestHandler 的方法在开发环境下（app.run(debug=True)）通常有效，因为 app.run() 内部使用了 Werkzeug 的开发服务器。然而，在生产环境中，您通常会使用 Gunicorn、uWSGI 等 WSGI 服务器，这些服务器可能有自己的请求处理机制，或者在多进程/多线程环境下，这种全局的修改可能不会按预期工作，或者每个子进程需要独立初始化。
- 排查思路： 如果在生产环境不生效，首先检查您的 WSGI 服务器是否使用了 Werkzeug 的 WSGIRequestHandler。如果不是，您可能需要查找该服务器的文档，了解如何定制其请求处理或日志行为。
正则表达式的精确性： re.match 的正则表达式 rf"/api/v1/{re.escape(endpoint)}(/.*)?$" 是一个示例。请根据您 API 的实际 URL 结构进行调整。如果您的 API 路径没有统一的前缀，或者端点名称与路径的映射关系复杂，您可能需要更复杂的正则表达式，或者在 permitted_endpoints 中存储完整的路径模式。
性能考量： 每次请求都会遍历 permitted_endpoints 列表并进行正则表达式匹配。对于拥有大量端点的应用，这可能会带来轻微的性能开销。如果性能成为瓶颈，可以考虑将正则表达式预编译，或者使用更高效的数据结构（如字典或集合）进行查找。
替代方案：
- Flask 中间件/before_request： 可以编写一个 Flask before_request 钩子函数，根据 request.path 判断是否在白名单内。如果不在，可以设置一个标志，然后自定义一个日志处理器，根据这个标志决定是否记录。
- Web 服务器层过滤： 在 Nginx、Apache 等 Web 服务器层面进行日志过滤。这种方式通常效率更高，且不占用应用服务器资源。例如，Nginx 可以配置 access_log off 或使用 map 指令根据请求路径选择性地记录日志。
- Python logging 模块的 Filter： 可以为 Python 的 logging 模块添加自定义 Filter。在 Filter 中，您可以访问日志记录 (LogRecord) 对象，其中可能包含请求相关的信息（如果您的日志记录器被配置为捕获这些信息），然后根据这些信息决定是否允许日志通过。
- 专门的日志管理工具： 使用 ELK Stack (Elasticsearch, Logstash, Kibana) 或其他日志聚合工具，在收集日志后进行过滤和分析，而不是在应用层面过滤。

总结

通过重写 werkzeug.serving.WSGIRequestHandler.log_request 方法并结合动态端点白名单，我们可以有效地过滤 Flask API 的请求日志，从而提高日志的质量和可读性。关键在于理解 app.url_map 的工作原理，并确保在所有路由注册完成后再进行日志过滤器的初始化。同时，也要充分考虑生产环境的部署差异和性能影响，并在必要时探索更适合的替代方案。正确的日志管理策略对于任何生产级应用都是至关重要的。

Python 浮点数格式化时如何控制正好显示两位小数（不四舍五入）

Python 如何在 except 块里获取当前抛出的异常对象（不写 as e）

Python httpx 如何设置全局代理 + 跳过某些域名的代理

Python 类属性和实例属性名字冲突时到底读哪个？

Python 中跨嵌套子包进行模块导入的正确实践