DevNull 知识库

Appearance

50｜Webhook 接收与文件处理 ​

运维平台经常需要接收外部系统的推送：GitLab 在代码提交时推送事件、监控平台在告警时调用回调 URL、CI/CD 系统在构建完成后通知状态。这些都是 Webhook 场景——外部系统主动 POST 数据到本平台。同时，平台也需要处理文件上传：导入服务器清单、导出报表、用户头像。

一、接收 Webhook ​

基础实现 ​

python
from fastapi import APIRouter, Header, HTTPException, Request
import hmac
import hashlib

router = APIRouter(prefix="/webhooks", tags=["webhooks"])

WEBHOOK_SECRET = "your-webhook-secret"

@router.post("/gitlab")
async def gitlab_webhook(
    request: Request,
    x_gitlab_token: str | None = Header(None),
):
    body = await request.body()

    # 签名校验
    if x_gitlab_token:
        expected = hmac.new(
            WEBHOOK_SECRET.encode(),
            body,
            hashlib.sha256,
        ).hexdigest()
        if not hmac.compare_digest(x_gitlab_token, expected):
            raise HTTPException(status_code=401, detail="签名无效")

    data = await request.json()
    event_type = request.headers.get("X-Gitlab-Event")

    # 处理事件
    if event_type == "Push Hook":
        handle_push_event(data)
    elif event_type == "Merge Request Hook":
        handle_mr_event(data)

    return {"received": True}
安全措施作用
签名验证确认请求来自合法的 GitLab 实例
hmac.compare_digest防时序攻击的常量时间比较
IP 白名单只接受已知源的请求（可在 Nginx 层配置）

幂等处理 ​

Webhook 可能因网络超时重发，同一事件处理多次会导致数据异常：

python
processed_events = set()   # 生产环境用 Redis/数据库

@router.post("/gitlab")
async def gitlab_webhook(request: Request):
    data = await request.json()
    event_id = data.get("checkout_sha") or data.get("object_attributes", {}).get("id")

    if event_id in processed_events:
        return {"status": "already_processed"}

    process_event(data)
    processed_events.add(event_id)
    return {"status": "ok"}

二、文件上传 ​

单文件上传 ​

python
from fastapi import UploadFile, File

@router.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    content = await file.read()
    save_path = f"uploads/{file.filename}"

    with open(save_path, "wb") as f:
        f.write(content)

    return {"filename": file.filename, "size": len(content)}
属性说明
file.filename原始文件名
file.content_typeMIME 类型，如 text/csv
file.file文件对象（SpooledTemporaryFile）

多文件上传 ​

python
from fastapi import UploadFile, File
from typing import List

@router.post("/upload-multiple")
async def upload_multiple(files: List[UploadFile] = File(...)):
    results = []
    for file in files:
        content = await file.read()
        results.append({"name": file.filename, "size": len(content)})
    return results

限制文件大小和类型 ​

python
from fastapi import HTTPException

ALLOWED_TYPES = {"text/csv", "application/json"}
MAX_SIZE = 10 * 1024 * 1024   # 10MB

@router.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    if file.content_type not in ALLOWED_TYPES:
        raise HTTPException(400, f"不支持的文件类型: {file.content_type}")

    content = await file.read()
    if len(content) > MAX_SIZE:
        raise HTTPException(400, f"文件超过 {MAX_SIZE} 字节限制")

    ...

三、静态文件服务 ​

上传的文件需要能被前端访问：

python
from fastapi.staticfiles import StaticFiles

app.mount("/static", StaticFiles(directory="uploads"), name="static")

上传 uploads/report.pdf 后，通过 http://localhost:8000/static/report.pdf 访问。

四、文件下载 ​

python
from fastapi.responses import FileResponse

@router.get("/download/{filename}")
async def download_file(filename: str):
    file_path = f"uploads/{filename}"
    return FileResponse(
        path=file_path,
        filename=filename,
        media_type="application/octet-stream",
    )

五、常见错误 ​

不校验 Webhook 签名 ​

python
# 错误：任何人都可以伪造请求
@router.post("/webhook")
async def receive(data: dict):
    process(data)

# 正确：校验签名或 Token

用原始文件名直接保存 ​

python
# 错误：文件名可能包含 ../ 等路径遍历攻击
save_path = f"uploads/{file.filename}"   # file.filename = "../../../etc/passwd"

# 正确：生成安全文件名
import uuid
from pathlib import Path

safe_name = f"{uuid.uuid4()}{Path(file.filename).suffix}"
save_path = f"uploads/{safe_name}"

大文件直接读入内存 ​

python
# 错误：大文件导致内存溢出
content = await file.read()

# 正确：分块写入
with open(save_path, "wb") as f:
    while chunk := await file.read(8192):
        f.write(chunk)

Webhook 处理超时 ​

python
# 错误：Webhook 处理耗时久，发送方超时重试
@router.post("/webhook")
async def receive(data: dict):
    long_running_process(data)   # 耗时 30 秒

# 正确：立即返回，异步处理
@router.post("/webhook")
async def receive(data: dict, background_tasks: BackgroundTasks):
    background_tasks.add_task(long_running_process, data)
    return {"status": "received"}