FastAPI 相当于 request.data，使用 shopify webhook 进行 hmac 验证错误

Question

我有一个 FastAPI 后端应用程序，应该使用 Shopify 处理 webhook 和 OAuth。 OAuth 的计算 hmac 正在按预期工作，但需要访问原始正文的 Webhook 版本则不然。例如，在 Flask 中，如果我们使用 request.get_data() 或 request.data（后者是正确的），计算出的 hmac 是不一样的。

我的问题是，在FastAPI中，request.data相当于什么？因为看起来await request.body() 并没有在这里完成工作。我确信密钥是正确的，因为它确实适用于 OAuth。

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)

我尝试使用十六进制而不是base64，尝试使用API KEY而不是SECRET。我还尝试使用硬编码的秘密，但这并没有改变任何东西。问题肯定出在await request.body()

这是来自另一个线程的一个工作 Flask 示例，其中 request.get_data() 导致错误的 hmac，而正确的 hmac 来自 request.data

###########

THIS IS WORKING

###########

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...')


###########

THIS IS NOT WORKING

###########

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...')

预先感谢您， 干杯

Answer 1

我在验证 Paddle webhooks 时遇到了类似的情况。我已经按照以下方式完成了：

signature: str | None = request.headers.get("Paddle-Signature")
raw_body: bytes = await request.body()

ts, h1 = ... # Parse signature string to extract ts and h1

body = raw_body.decode()
signed_payload = f"{ts}:{body}"

h1_built = hmac.new(
    key=self._webhook_secret_key.encode("utf-8"),
    msg=signed_payload.encode(),
    digestmod=hashlib.sha256,
)

if not h1_built.hexdigest() == h1:
    # Raise error if not equal

希望对你有帮助！