import json
import os
import hmac
import hashlib
import re
from flask import Flask, request
app = Flask(__name__)
def verify_webhook_signature(raw_body: bytes | str, signature_header: str, secret: str) -> bool:
if not secret or not signature_header or raw_body is None:
return False
m = re.match(r"t=(\d+),v1=([a-f0-9]+)", signature_header.strip())
if not m:
return False
t, v1 = m.group(1), m.group(2)
body_str = raw_body.decode("utf-8") if isinstance(raw_body, bytes) else raw_body
signed_payload = f"{t}.{body_str}".encode("utf-8")
expected = hmac.new(
secret.encode("utf-8"),
signed_payload,
hashlib.sha256,
).hexdigest()
return hmac.compare_digest(v1, expected)
@app.route("/contiguity/webhook", methods=["POST"])
def webhook():
raw_body = request.get_data(as_text=True)
sig = request.headers.get("Contiguity-Signature")
secret = os.environ.get("WEBHOOK_SECRET")
if not verify_webhook_signature(raw_body, sig, secret):
return "", 401
event = json.loads(raw_body)
# handle event...
return "OK", 200