FastAPI 文件上传完全指南:安全的文件处理

FastAPI 文件上传实践与生产边界:单文件与多文件上传、类型与大小校验、防路径穿越、流式处理大文件与安全存储,避开常见安全坑。

最佳实践
并发任务的多路径协同插画

直接回答:FastAPI 用 UploadFile 优雅处理文件上传,但生产级上传系统必须自己把好三道关:类型校验(不只信扩展名)、大小限制(流式读取防撑爆内存)、路径净化(防目录穿越)——做对了是功能,做错了是漏洞。

最小可用上传

pip install "fastapi[standard]" python-multipart
from fastapi import FastAPI, UploadFile

app = FastAPI()

@app.post("/upload")
async def upload(file: UploadFile):
    content = await file.read()
    return {"filename": file.filename, "size": len(content)}

python-multipart 是表单上传的必备依赖。UploadFile 使用可溢写到磁盘的临时文件,超过阈值才落盘;上面无参数的 await file.read() 仍会把全部内容读入内存,只适合受限小文件。multipart 通常在进入路由前已解析,因此路由中的大小检查不能替代入口请求体限制。

完整校验体系

四道防线缺一不可:

from pathlib import Path
from uuid import uuid4
import magic  # pip install python-magic;还需系统 libmagic
from fastapi import HTTPException

MIME_SUFFIX = {"image/jpeg": ".jpg", "image/png": ".png", "application/pdf": ".pdf"}
MAX_SIZE = 10 * 1024 * 1024
UPLOAD_DIR = Path("uploads")  # 私有隔离目录,不直接通过静态服务公开
UPLOAD_DIR.mkdir(mode=0o700, parents=True, exist_ok=True)

def save_one(file: UploadFile):
    path = None
    try:
        head = file.file.read(2048)
        mime = magic.from_buffer(head, mime=True)
        if mime not in MIME_SUFFIX:
            raise HTTPException(415, "不支持的文件类型")
        file.file.seek(0)
        safe_name = f"{uuid4().hex}{MIME_SUFFIX[mime]}"
        total = 0
        # x 模式拒绝覆盖;只有成功创建的文件才由本函数清理
        candidate = UPLOAD_DIR / safe_name
        with candidate.open("xb") as output:
            path = candidate
            while chunk := file.file.read(1024 * 1024):
                total += len(chunk)
                if total > MAX_SIZE:
                    raise HTTPException(413, "文件过大")
                output.write(chunk)
        return {"saved": safe_name, "size": total}
    except BaseException:
        if path is not None:
            path.unlink(missing_ok=True)
        raise
    finally:
        file.file.close()

# 替换上面的演示端点,不要在同一个 app 重复注册 POST /upload
@app.post("/upload")
def upload_secure(file: UploadFile):
    return save_one(file)

同步 def 路由由 FastAPI 放到线程池运行,避免同步 libmagic/磁盘写操作直接阻塞事件循环。MIME 检测仅是筛选信号,不证明内容安全;此示例也未包含用户鉴权、配额、病毒扫描和下载授权,不能直接视为完整生产上传系统。

3. 文件名净化:用户传来的 ../../etc/passwd 是经典路径穿越——永远用服务端生成的文件名(UUID),原始文件名只作展示。

4. 存储隔离:上传文件不要授予执行权限,不放在可执行或公开 Web 根目录。使用私有对象存储时同样需要授权、生命周期管理、扫描与安全下载响应头。

多文件上传

@app.post("/upload/batch")
def upload_batch(files: list[UploadFile]):
    if not 1 <= len(files) <= 5:
        for file in files:
            file.file.close()
        raise HTTPException(400, "每批限 1 至 5 个文件")
    results = []
    try:
        for file in files:
            results.append(save_one(file))
        return {"saved": results}
    except BaseException:
        for result in results:
            (UPLOAD_DIR / result["saved"]).unlink(missing_ok=True)
        raise
    finally:
        for file in files:
            file.file.close()

本例最多 5 个文件、每个最多 10 MiB,因此文件内容合计最多 50 MiB;请求体还包含 multipart 开销。网关应另设总请求大小上限,应用还需磁盘配额和并发限额。

生产加分项

  • 进度反馈:大文件上传配前端进度条(XHR 的 progress 事件),应区分浏览器上传进度与服务端解析/扫描/持久化进度;
  • 异步处理:扫描/转码/缩略图放后台任务,接口立即返回"已受理";
  • 防恶意文件:图片在受限资源下解码、检查尺寸并重新编码,可减少部分风险,但不能保证移除所有恶意内容;文档类按策略隔离扫描,扫描通过前不对外可用;
  • 限流:上传接口加频率限制,防刷爆磁盘与带宽。

验证上传限制时,分别提交正常文件、超限文件和中断请求,记录结果、字节数与耗时。若已用观测云采集应用日志,可通过文件采集配置接入这些记录,按结果检查拒绝和失败原因;日志中保留文件标识即可,避免记录上传内容或访问凭据。

常见问题(FAQ)

Q:client_max_body_size 之类的前置限制还要吗?
A:要。反向代理/网关层的限制是第一道闸(直接拒于门外),应用内校验是第二道。纵深防御,各管一层。

Q:UploadFile 的临时文件要清理吗?
A:请求结束后 FastAPI 自动清理;但你自己落盘的文件要管生命周期——定期清理孤儿文件(上传中断的残留)是常被遗忘的运维任务。

Q:怎么测文件上传接口?
A:TestClient 直接传:client.post("/upload", files={"file": ("a.png", io.BytesIO(png_bytes), "image/png")})。覆盖类型非法、超大、路径穿越三类恶意用例。

官方参考

本文基于官方文档整理,未进行运行时或性能测试。示例中的业务函数、数据模型和部署地址需结合项目补全;局部片段不等同于完整生产应用。

延伸阅读

获取专属方案

联系我们

加入社区

微信扫码
加入官方交流群

立即体验

在线开通,按量计费,真正的云服务!

立即开始

选择观测云版本

代码托管平台