FastAPI中request.data等效方案及Shopify Webhook HMAC验证失败问题

我开发了一个FastAPI后端应用，用于处理Shopify的Webhook与OAuth功能。OAuth的HMAC验证功能正常，但Webhook的HMAC验证始终失败——该验证需要获取原始请求体。例如在Flask中，request.get_data()和request.data计算出的HMAC结果不同，其中request.data是正确的。

我的问题是：在FastAPI中，Flask的request.data的等效方法是什么？我尝试使用await request.body()但无法解决问题，且确认密钥正确（因为OAuth验证可用）。我已尝试改用十六进制替代Base64、使用API密钥替代密钥、硬编码密钥等方法，但均无效，问题肯定出在请求体获取环节。

FastAPI代码示例 #

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

def verify_webhook(data, hmac_header):
    digest = hmac.new(SHOPIFY_SECRET.encode('utf-8'), data, digestmod=hashlib.sha256).digest()
    computed_hmac = base64.b64encode(digest)
    print(f"Computed HMAC: {computed_hmac}")
    print(f"Received HMAC: {hmac_header.encode('utf-8')}")
    return hmac.compare_digest(computed_hmac, hmac_header.encode('utf-8'))

@app.post('/webhook/customer/data_request')
async def customer_data_request_webhook(request: Request):
    try:
        data = await request.body()
        headers = dict(request.headers)
        print('RAW DATA:', data)
        print('HEADERS:', headers)

        hmac_header = headers.get('x-shopify-hmac-sha256')
        if not hmac_header:
            raise HTTPException(status_code=HTTP_401_UNAUTHORIZED, detail="HMAC header not found")
        print("HMAC HEADER:", hmac_header)

        verified = verify_webhook(data, hmac_header)
        if not verified:
            raise HTTPException(status_code=HTTP_401_UNAUTHORIZED, detail="HMAC verification failed")

        print("Received customer data request webhook:", data)
        return Response(status_code=200)

    except Exception as e:
        print("Error processing customer data request webhook:", e)
        return Response(status_code=HTTP_500_INTERNAL_SERVER_ERROR)

Flask对比示例 #

可正常运行的版本 #

from flask import Flask, request, abort
import hmac
import hashlib
import base64

app = Flask(__name__)

SECRET = '...'

def verify_webhook(data, hmac_header):    
    digest = hmac.new(SECRET.encode('utf-8'), data, hashlib.sha256).digest()
    genHmac = base64.b64encode(digest)

    return hmac.compare_digest(genHmac, hmac_header.encode('utf-8'))

@app.route('/', methods=['POST'])
def hello_world(request):
    print('Received Webhook...')

    data = request.data # NOT request.get_data() !!!!!
    hmac_header = request.headers.get('X-Shopify-Hmac-SHA256')
    verified = verify_webhook(data, hmac_header)
    
    if not verified:
        return 'Integrity of request compromised...', 401
    
    print('Verified request...')

无法运行的版本 #

from flask import Flask, request, abort
import hmac
import hashlib
import base64

app = Flask(__name__)

SECRET = '...'

def verify_webhook(data, hmac_header):    
    digest = hmac.new(SECRET.encode('utf-8'), data, hashlib.sha256).digest()
    genHmac = base64.b64encode(digest)

    return hmac.compare_digest(genHmac, hmac_header.encode('utf-8'))

@app.route('/', methods=['POST'])
def hello_world(request):
    print('Received Webhook...')

    data = request.get_data()
    hmac_header = request.headers.get('X-Shopify-Hmac-SHA256')
    verified = verify_webhook(data, hmac_header)
    
    if not verified:
        return 'Integrity of request compromised...', 401
    
    print('Verified request...')

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