This question already has answers here: Using FastAPI in a sync way, how can I get the raw body of a POST request? (2 answers) Closed 2 months ago. I have a FastAPI backend application that should handle webhook and OAuth with Shopify. computing hmac for the OAuth is working as intended, but not the webhook version of it that require to access the raw body. In Flask for example, the computed hmac is not the same if we use request.get_data() or request.data (the later being the correct one). My question is, in FastAPI, what is the equivalent of request.data ? Because it seems like await request.body() doesn't do the work here. And I'am sure the secret key is the correct one since it does work for 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) I have tried using hex instead of base64, tried with API KEY instead of SECRET. I also tried to use hardcoded secret, but that's not changing anything. The problem surely lies in await request.body() Here is a working Flask example from another thread, where request.get_data() was leading to a wrong hmac and the correct one was from 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...') Thank you in advance, Cheers
Jese Leos
August 19, 2024
Verified user
I had a similar case while validating Paddle webhooks. I've done it in the following way: 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 Hope it'll help!