文件上传规范

类型、大小、病毒扫描

类型校验

白名单机制

[Requirement] 文件上传采用扩展名白名单 + MIME 类型校验 + 文件头魔数校验三重校验，三者一致才放行。

[Requirement] 禁止使用黑名单（禁止 .exe、.php 等），因为黑名单容易被绕过（如 .phtml、.php5）。

业务场景	允许扩展名	MIME 类型
头像	jpg、jpeg、png、gif、webp	image/jpeg、image/png、image/gif、image/webp
文档	pdf、doc、docx、xls、xlsx	application/pdf、application/msword、…
压缩包	zip	application/zip
视频	mp4	video/mp4

魔数校验

[Requirement] 读取文件头字节（魔数）判断真实类型，防止改扩展名绕过。

import struct
 
# 常见文件魔数
MAGIC_NUMBERS = {
    b"\xff\xd8\xff": "image/jpeg",
    b"\x89PNG\r\n\x1a\n": "image/png",
    b"GIF87a": "image/gif",
    b"GIF89a": "image/gif",
    b"%PDF-": "application/pdf",
    b"PK\x03\x04": "application/zip",  # zip / docx / xlsx
    b"\x00\x00\x00 ftyp": "video/mp4",
}
 
def detect_mime(file_path: str) -> str | None:
    with open(file_path, "rb") as f:
        header = f.read(16)
    for magic, mime in MAGIC_NUMBERS.items():
        if header.startswith(magic):
            return mime
    return None

文件名处理

[Requirement] 上传文件名必须重新生成，禁止使用用户上传的原始文件名（防止路径遍历、XSS、特殊字符）。

[Standard] 新文件名格式：{uuid}.{安全扩展名}，扩展名从白名单中取，不使用用户提供的扩展名。

import uuid
 
def generate_filename(original_name: str, allowed_ext: set[str]) -> str:
    ext = original_name.rsplit(".", 1)[-1].lower() if "." in original_name else ""
    if ext not in allowed_ext:
        raise ValueError("不支持的文件类型")
    return f"{uuid.uuid4().hex}.{ext}"

大小限制

单文件与总量限制

[Requirement] 必须限制单文件大小和单次请求总大小，防止磁盘耗尽和 DoS。

场景	单文件上限	单次请求上限
头像	2 MB	10 MB
文档	20 MB	50 MB
压缩包	50 MB	100 MB
视频	200 MB	500 MB

多层限制

[Requirement] 大小限制在网关、应用框架、业务代码三层同时设置，任一层超限即拒绝。

# Nginx 层
client_max_body_size 50m;

# FastAPI 层
from fastapi import UploadFile, File, HTTPException
 
MAX_SIZE = 20 * 1024 * 1024  # 20 MB
 
@app.post("/upload")
async def upload(file: UploadFile = File(...)):
    # 1. 先校验 Content-Length
    if file.size and file.size > MAX_SIZE:
        raise HTTPException(413, "文件过大")
    # 2. 流式读取时再校验实际大小
    content = await file.read()
    if len(content) > MAX_SIZE:
        raise HTTPException(413, "文件过大")

[Caution] 禁止信任 Content-Length 头（可伪造），必须在读取后校验实际字节数。

压缩包炸弹防护

[Standard] 上传 zip 等压缩包时，限制解压后文件数量和总大小，防止压缩炸弹（1MB 压缩包解压出 100GB）。

import zipfile
 
def safe_extract(zip_path: str, dest: str, max_files: int = 1000, max_size: int = 1024 * 1024 * 1024):
    with zipfile.ZipFile(zip_path) as z:
        infos = z.infolist()
        if len(infos) > max_files:
            raise ValueError("压缩包文件过多")
        total = sum(i.file_size for i in infos)
        if total > max_size:
            raise ValueError("解压后总大小超限")
        z.extractall(dest)

病毒扫描

扫描要求

[Requirement] 所有上传文件必须经过病毒扫描，扫描通过后才能存储和分发。

[Standard] 使用 ClamAV（开源）或商业杀毒引擎（如 VirusTotal API、云厂商杀毒服务）。

扫描流程

上传 → 临时存储 → 病毒扫描 → 通过 → 转存正式存储 → 返回 URL
                              → 未通过 → 删除临时文件 → 返回拒绝

import pyclamd  # ClamAV Python 客户端
 
def scan_file(file_path: str) -> bool:
    try:
        cd = pyclamd.ClamdAgnostic()
        result = cd.scan_file(file_path)
        if result is None:
            return True  # 无病毒
        # result 非 None 表示发现病毒
        logger.warning("病毒扫描发现威胁: %s", result)
        return False
    except Exception as e:
        logger.error("病毒扫描服务异常", exc_info=True)
        return False  # 扫描失败默认拒绝（fail-close）

[Requirement] 病毒扫描失败时采用 fail-close（默认拒绝），禁止 fail-open（扫描失败也放行）。

异步扫描

[Default] 大文件或高并发场景采用异步扫描：上传后先存隔离区，扫描通过后移到正式区；扫描未通过的文件状态标记为”隔离”，不可访问。

存储路径

隔离存储

[Requirement] 上传文件必须存储在独立目录 / 独立存储桶，与应用代码、系统文件隔离，禁止存到 Web 可直接执行的目录（如 /var/www/html）。

[Standard] 生产环境使用对象存储（OSS / S3 / COS），不存本地磁盘；对象存储开启版本控制和服务端加密。

路径安全

[Requirement] 文件路径由服务端拼接，禁止包含用户输入的路径片段，防止路径遍历。

# 正确：服务端生成路径
key = f"uploads/{date.today().isoformat()}/{uuid.uuid4().hex}.{ext}"
oss.put_object(BUCKET, key, content)
 
# 错误：使用用户输入拼路径
key = f"uploads/{user_input_path}"  # 可能含 ../../etc/passwd

存储桶权限

[Requirement] 对象存储桶默认私有，文件通过签名 URL（预签名 URL，有效期短）访问，禁止公开读写。

# 生成预签名 URL（有效期 1 小时）
url = oss.generate_presigned_url(
    "get_object",
    Params={"Bucket": BUCKET, "Key": key},
    ExpiresIn=3600,
)

访问控制

权限校验

[Requirement] 文件访问必须校验当前用户的权限，禁止仅凭 URL 访问（防止 IDOR / 越权下载）。

[Standard] 文件 URL 不暴露存储原始路径，通过 /files/{file_id} 接口代理访问，服务端校验权限后返回文件流或重定向到签名 URL。

@app.get("/files/{file_id}")
async def download_file(file_id: str, current_user: User = Depends(get_current_user)):
    file_meta = file_repo.get(file_id)
    if not file_meta:
        raise NotFoundError("文件不存在")
    # 权限校验：文件归属或共享权限
    if file_meta.owner_id != current_user.id and not has_share_permission(file_meta, current_user):
        raise PermissionDeniedError("无权限访问该文件")
    # 返回签名 URL 或代理流
    url = oss.generate_presigned_url("get_object", ...)
    return RedirectResponse(url)

图片二次处理

[Standard] 用户上传的图片在存储前做二次处理（重新编码、去除 EXIF 元数据、压缩），防止图片中嵌入恶意脚本或泄露位置信息。

from PIL import Image
import io
 
def sanitize_image(content: bytes, max_size: tuple = (2048, 2048)) -> bytes:
    img = Image.open(io.BytesIO(content))
    img.thumbnail(max_size)
    # 去除 EXIF，重新编码
    output = io.BytesIO()
    img.save(output, format=img.format, quality=85)
    return output.getvalue()

防盗链

[Standard] 公开访问的资源（如头像）通过 Referer 白名单或签名 URL 防盗链，防止被外部站点滥用流量。

删除策略

[Standard] 用户删除文件时，标记为”已删除”并设置过期时间（如 30 天后物理删除），支持误删恢复；过期后从存储中彻底删除。