Flask 请求体校验和模式包装 WSGI 输入流计算原始请求数据摘要【免费下载链接】flaskThe Python micro framework for building web applications.项目地址: https://gitcode.com/gh_mirrors/fl/flask本篇技术指南基于 Flask 官方文档 docs/patterns/requestchecksum.rst 讲解“Request Content Checksums请求内容校验和”这一应用模式当 API 需要验证客户端提交数据的完整性例如比对预先交换的 SHA1 摘要时如何在不改动框架的前提下通过包装 WSGI 环境的输入流wsgi.input在数据被 Flask 读取的过程中同步计算出原始请求体的校验和。读完本篇你能掌握完整的可运行代码实现、校验和流的挂接时机要求以及 Flask 请求对象延迟加载wsgi.input的底层机制从而在自己的项目中落地请求数据完整性校验。为什么需要对“原始请求体”计算校验和某些 API 场景要求服务端确认收到的请求数据与发送端一致。常见做法是客户端在发送前计算请求体的摘要如 SHA1服务端收到后再对原始字节流计算摘要并比对。问题在于在 Flask 应用中请求数据往往会先被框架“消费”掉。根据官方文档的描述各类代码会以不同路径消费请求数据JSON 数据到达request对象时已经被读取并解析完毕表单数据同样如此只是走另一条代码路径。此时如果你直接调用request.get_data()之类的方法再算摘要拿到的往往不是“原始流”——它可能已被缓存、缓冲或者顺序已经被改变。这正是文档要解决的痛点。文档给出的核心思路只有一句话包装输入流wrapping the input stream。在原始流被任何代码读取之前把它替换成一个“边读边算摘要”的装饰器流对象此后所有读取请求体的代码路径无论 JSON 解析、表单解析还是文件上传读到的数据都会自动流经摘要计算器。实现方案包装 WSGI 输入流wsgi.input在请求处理中的角色在 WSGI 规范中environ[wsgi.input]是一个类文件对象file-like object代表 HTTP 请求的原始消息体。Flask/Werkzeug 的请求对象正是从它读取字节流再交给 JSON 解析器、MultiPartParser等组件。Flask 的请求类在 src/flask/wrappers.py 中定义为 WerkzeugRequest的子类class Request(RequestBase): The request object used by default in Flask. ... The request object is a :class:~werkzeug.wrappers.Request subclass and provides all of the attributes Werkzeug defines plus a few Flask specific ones. 从源码结构看request.data、request.files、request.form等属性都是延迟加载的只有第一次被访问时才触发解析而解析的数据源就是当前environ[wsgi.input]。这就给了包装方案一个天然的切入点——只要在数据流第一次被读取之前替换environ[wsgi.input]后续所有解析路径读取到的都将是包装后的流。完整的校验和流实现官方文档给出的实现是一个装饰器类ChecksumCalcStream它持有原始流的引用和一个hashlib.sha1()摘要对象并在read与readline两个读取入口上“透传 记账”。以下代码与 docs/patterns/requestchecksum.rst 中的示例一致import hashlib class ChecksumCalcStream(object): def __init__(self, stream): self._stream stream # 原始 WSGI 输入流 self._hash hashlib.sha1() # 增量式摘要计算器 def read(self, bytes): rv self._stream.read(bytes) self._hash.update(rv) # 每读一批数据就更新摘要 return rv def readline(self, size_hint): rv self._stream.readline(size_hint) self._hash.update(rv) # 逐行读取路径同样记账 return rv def generate_checksum(request): env request.environ stream ChecksumCalcStream(env[wsgi.input]) env[wsgi.input] stream # 把包装流换入 WSGI 环境 return stream._hash # 返回同一个 hashlib 对象供后续取摘要几个实现要点值得注意同时覆盖read和readline不同解析组件读取流的方式不同JSON 解析通常整块readmultipart 解析可能逐行readline。只包装其中一个方法会导致另一种读取路径绕开摘要计算最终摘要不完整。generate_checksum返回的是hashlib对象本身而不是立即取摘要。因为此时数据还没有被读完整只能等请求体被真正消费完后调用hash.hexdigest()才能得到最终值。这个“延迟取值”的设计是该模式的关键。update(rv)接收的是实际读到的字节即使读到空字节串流末尾也无害hashlib.update对空输入是幂等的。挂接时机与完整使用示例文档强调了一条硬性约束必须在请求开始消费数据之前完成挂接。具体表述为要小心任何可能触发流读取的属性访问例如request.formbefore_request处理器尤其注意不要访问这些数据否则包装流就没机会替换原始流摘要只会覆盖后续读取的部分。在视图函数内部的用法示例如下同样来自原文档app.route(/special-api, methods[POST]) def special_api(): hash generate_checksum(request) # Accessing this parses the input stream files request.files # At this point the hash is fully constructed. checksum hash.hexdigest() return fHash was: {checksum}执行顺序拆解generate_checksum(request)把env[wsgi.input]替换为ChecksumCalcStream此时尚未读取任何字节访问request.files触发 Werkzeug 的表单/文件解析器从已包装的流中读取整个 multipart 请求体每一次read/readline都会喂给内部的 SHA1 计算器解析完成后摘要已经完整hash.hexdigest()返回与客户端预计算的摘要可比对的十六进制字符串。从 Flask 源码可以验证这个时序是可行的before_request函数由Flask.preprocess_request在视图分发之前调用见 src/flask/app.pydef preprocess_request(self, ctx: AppContext) - ft.ResponseReturnValue | None: Called before the request is dispatched. ... If any :meth:before_request handler returns a non-None value, the value is handled as if it was the return value from the view, and further request handling is stopped. 也就是说如果你选择在before_request中挂接校验和流它是完全来得及的——前提是这些处理器自身不要访问request.form、request.files、request.data或request.get_json()等任何会触发流读取的属性Flask 的表单数据加载入口_load_form_data位于 src/flask/wrappers.py其触发点在首次访问相关属性时。边界条件与实战注意事项请求体大小限制对校验和的影响Flask 支持通过MAX_CONTENT_LENGTH配置限制单个请求允许读取的最大字节数超限会抛出 413RequestEntityTooLarge。该属性在 src/flask/wrappers.py 中实现可全局配置也可按单个请求设置property def max_content_length(self) - int | None: The maximum number of bytes that will be read during this request. If this limit is exceeded, a 413 ... error is raised. ... 配置方式在 docs/patterns/fileuploads.rst 中有完整示例app.config[MAX_CONTENT_LENGTH] 16 * 1000 * 1000 # 限制最大载荷为 16 MB与校验和模式结合时要注意如果请求体超限解析在读取中途就会抛异常此时摘要对象只反映了部分数据不能用它做任何完整性结论。因此完整性校验逻辑应当放在请求成功解析之后如示例中在访问request.files之后取hexdigest并在错误处理分支中放弃摘要比对。校验和的语义边界SHA1 仅用于完整性比对不用于签名防篡改。原文档示例采用 SHA1 是沿袭一些 API 的数据完整性协议比对双方预交换的摘要。如果你的目标是防篡改或来源认证应使用 HMAC 或基于密钥的摘要方案而不是裸哈希。wsgi.input是一次性资源。一旦请求体被完整读取原始流即耗尽校验和流不会改变这一点。如果需要多次读取请求体应另用request.get_data(cacheTrue)等缓存机制而不是依赖包装流重放。包装只能拦截应用内部的读取。如果 WSGI 服务器或前置中间件在数据到达应用前已经缓冲或消费了部分流包装对象只能看到剩余字节。从源码结构看ChecksumCalcStream直接作用于request.environ中当前的流对象docs/patterns/requestchecksum.rst 中env[wsgi.input] stream一行因此它假设应用拿到的是未经篡改的原始流。在部署环境如 gunicorn、uWSGI 等见 docs/deploying中应确认中间层没有代读请求体。适配其他摘要算法ChecksumCalcStream与具体哈希算法完全解耦把hashlib.sha1()换成hashlib.sha256()即可支持其他算法若需流式输出进度也可以在同一装饰器内按read的返回字节数累计计数这正是“边读边算”模式的扩展空间。小结该模式的全部技巧可以浓缩为三点Flask/Werkzeug 的请求数据读取统一收敛到environ[wsgi.input]替换它即可拦截所有解析路径JSON、表单、文件上传装饰器流在read/readline两个入口做增量hashlib.update返回摘要对象延迟取值挂接必须发生在任何流读取动作之前视图函数开头是最稳妥的位置before_request可以用但不能自己先碰request.form/request.files。参考仓库内的原始文档 docs/patterns/requestchecksum.rst以及相关的请求与上传机制文档 docs/patterns/fileuploads.rst、请求封装源码 src/flask/wrappers.py 和请求预处理流程 src/flask/app.py即可完整复现并扩展这一模式。【免费下载链接】flaskThe Python micro framework for building web applications.项目地址: https://gitcode.com/gh_mirrors/fl/flask创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考