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")})。覆盖类型非法、超大、路径穿越三类恶意用例。
官方参考
本文基于官方文档整理,未进行运行时或性能测试。示例中的业务函数、数据模型和部署地址需结合项目补全;局部片段不等同于完整生产应用。