FastAPI中request.get_data()的等效方法及Shopify Webhook验证

获取原始请求字节（等效Flask的request.get_data()） #

在FastAPI里，你只需在路由函数中注入Request对象，通过await request.body()就能拿到原始请求体的字节数据——这和Flask的request.get_data()功能完全一致，返回的是bytes类型的原始未解析数据。

完整的Shopify请求验证代码示例 #

from fastapi import FastAPI, Request, HTTPException
import hmac
import hashlib

app = FastAPI()
# 替换为你在Shopify后台获取的Webhook密钥
SHOPIFY_WEBHOOK_SECRET = "your-shopify-webhook-secret"

@app.post("/shopify/webhook")
async def process_shopify_webhook(request: Request):
    # 提取Shopify发送的HMAC签名头
    received_hmac = request.headers.get("X-Shopify-Hmac-SHA256")
    if not received_hmac:
        raise HTTPException(status_code=401, detail="Missing HMAC authentication header")
    
    # 获取原始请求字节串
    raw_request_body = await request.body()
    
    # 计算本地HMAC签名
    computed_hmac = hmac.new(
        SHOPIFY_WEBHOOK_SECRET.encode("utf-8"),
        raw_request_body,
        digestmod=hashlib.sha256
    ).digest().hex()
    
    # 安全对比签名（防止时序攻击）
    if not hmac.compare_digest(computed_hmac, received_hmac):
        raise HTTPException(status_code=401, detail="Invalid HMAC signature")
    
    # 验证通过后再按需解析请求内容
    webhook_payload = await request.json()
    return {"status": "validated", "message": "Webhook processed successfully", "data": webhook_payload}

重要注意事项 #

• 优先获取原始字节：绝对不能先调用request.json()或用Pydantic模型接收参数，因为解析过程会调整数据格式（比如JSON键排序、空格压缩等），导致HMAC计算结果与Shopify发送的签名不匹配。
• 使用安全对比方法：必须用hmac.compare_digest()而非直接用==比较签名，这个方法能避免时序攻击，提升验证安全性。
• 区分密钥类型：这里使用的是Shopify后台的Webhook密钥，不是API密钥，二者不可混淆。

内容的提问来源于stack exchange，提问作者savdbroek