typeform_webhook_fastapi.py

from fastapi import FastAPI,Request,HTTPException

import hashlib
import hmac
import json
import base64
import os

app = FastAPI()

@app.get("/")
def read_root():
    return {"message": "Point webhook to /hook of this application."}
  
@app.get("/hook")
def getHook():
  return {"message": "Use to POST route to send webhook."}

@app.post("/hook")
async def recWebHook(req: Request):
    body = await req.json()
    raw = await req.body()
    
    receivedSignature = req.headers.get("typeform-signature")
    
    if receivedSignature is None:
      return HTTPException(403, detail="Permission denied.")
      
    sha_name, signature = receivedSignature.split('=', 1)
    if sha_name != 'sha256':
      return HTTPException(501, detail="Operation not supported.")
    
    is_valid = verifySignature(signature, raw)
    
    if(is_valid != True):
      return HTTPException(403, detail="Invalid signature. Permission Denied.")
    
    # Do something with the payload received
    return {"Message": "Webhook well received"}
    
  
  
def verifySignature(receivedSignature: str, payload):
    
    WEBHOOK_SECRET = os.environ.get('TYPEFORM_SECRET_KEY')
    digest = hmac.new(WEBHOOK_SECRET.encode('utf-8'), payload, hashlib.sha256).digest()
    
    e = base64.b64encode(digest).decode()
    
    if(e == receivedSignature):
      return True
    return False