📚 课程目录第2章实现固件文件上传与校验

2. 实现固件文件上传与校验

任务说明

本案例实现一个固件上传接口，接收客户端上传的固件文件，计算其MD5值并与客户端声明的MD5比对，校验通过后保存文件，确保固件包在传输过程中未被篡改或损坏。

环境依赖

使用 Python 3.8+ 和 Flask 2.x，无需额外第三方库（MD5计算使用标准库 hashlib）。

关键完整代码

import os
import hashlib
from flask import Flask, request, jsonify

app = Flask(__name__)
UPLOAD_DIR = "./uploads"
os.makedirs(UPLOAD_DIR, exist_ok=True)

# 允许的固件扩展名
ALLOWED_EXTENSIONS = {".bin", ".hex", ".img"}

def allowed_file(filename):
    return os.path.splitext(filename)[1].lower() in ALLOWED_EXTENSIONS

def calc_md5(file_path, chunk_size=8192):
    """分块计算文件MD5，避免大文件一次性读入内存"""
    md5 = hashlib.md5()
    with open(file_path, "rb") as f:
        while chunk := f.read(chunk_size):
            md5.update(chunk)
    return md5.hexdigest()

@app.route("/api/firmware/upload", methods=["POST"])
def upload_firmware():
    # 1. 检查文件是否在请求中
    if "file" not in request.files:
        return jsonify({"code": 400, "message": "缺少file字段"}), 400

    file = request.files["file"]
    if file.filename == "":
        return jsonify({"code": 400, "message": "文件名为空"}), 400

    # 2. 检查扩展名
    if not allowed_file(file.filename):
        return jsonify({"code": 400, "message": "不支持的固件格式"}), 400

    # 3. 获取客户端声明的MD5
    client_md5 = request.form.get("md5", "").strip().lower()
    if not client_md5:
        return jsonify({"code": 400, "message": "缺少md5参数"}), 400

    # 4. 保存临时文件
    temp_path = os.path.join(UPLOAD_DIR, f"temp_{file.filename}")
    file.save(temp_path)

    try:
        # 5. 计算实际MD5并比对
        actual_md5 = calc_md5(temp_path)
        if actual_md5 != client_md5:
            os.remove(temp_path)
            return jsonify({
                "code": 400,
                "message": "MD5校验失败，文件可能已损坏",
                "expected": client_md5,
                "actual": actual_md5
            }), 400

        # 6. 校验通过，重命名为正式文件名
        final_path = os.path.join(UPLOAD_DIR, file.filename)
        os.rename(temp_path, final_path)

        return jsonify({
            "code": 200,
            "message": "上传成功，MD5校验通过",
            "filename": file.filename,
            "md5": actual_md5,
            "size": os.path.getsize(final_path)
        }), 200

    except Exception as e:
        if os.path.exists(temp_path):
            os.remove(temp_path)
        return jsonify({"code": 500, "message": f"服务器错误: {str(e)}"}), 500

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000, debug=True)

运行方式

将上述代码保存为 app.py；
安装依赖：pip install flask；
启动服务：python app.py。

验证方法

使用 curl 模拟客户端上传，先计算本地固件文件的MD5，再发起请求：

# 计算本地固件的MD5
md5sum firmware.bin

# 上传固件（将 <MD5值> 替换为上面命令的输出）
curl -X POST http://127.0.0.1:5000/api/firmware/upload \
  -F "file=@firmware.bin" \
  -F "md5=<MD5值>"

通过条件：

MD5正确时，返回 code: 200，且响应中的 md5 字段与本地计算结果一致，uploads 目录下生成固件文件；
故意传入错误MD5时，返回 code: 400，提示“MD5校验失败”，且 uploads 目录下不残留临时文件。

常见问题

大文件上传超时： Flask 默认同步处理，可配置 nginx 代理或改用异步框架；本案例聚焦校验逻辑，未做流式接收优化。
文件名冲突： 当前实现直接以原文件名保存，重复上传会覆盖。生产环境建议在文件名前加时间戳或UUID。
临时文件清理： 若进程在保存后、重命名前崩溃，temp_ 前缀文件会残留。可增加定时清理任务处理。