Webhooks Overview

from flask import Flask, request, abort
import hmac
import hashlib
import time
import json

app = Flask(__name__)
WEBHOOK_SECRET = "your_webhook_secret"

@app.route("/webhooks/hostpay", methods=["POST"])
def handle_webhook():
    # Read raw body before any parsing
    raw_body = request.get_data(as_text=True)

    # Extract headers
    signature_header = request.headers.get("X-HostPay-Signature", "")
    timestamp = request.headers.get("X-Webhook-Timestamp", "")

    # Reject stale webhooks (optional but recommended)
    if abs(time.time() - int(timestamp)) > 300:
        abort(400, "Webhook timestamp too old")

    # Verify the signature
    if not verify_signature(raw_body, timestamp, signature_header):
        abort(401, "Invalid signature")

    # Parse and handle the event
    event = json.loads(raw_body)
    event_type = request.headers.get("X-Webhook-Version")  # dispatched via subscription
    data = event.get("data", {})

    # Route by status in the data payload
    status = data.get("status")
    if status == "succeeded":
        handle_success(data)
    elif status == "failed":
        handle_failure(data)

    return "OK", 200


def verify_signature(raw_body: str, timestamp: str, signature_header: str) -> bool:
    # Signature header format: "v1=<hex_digest>"
    if not signature_header.startswith("v1="):
        return False
    provided_sig = signature_header[3:]

    signed_string = f"{timestamp}.{raw_body}"
    expected_sig = hmac.new(
        WEBHOOK_SECRET.encode("utf-8"),
        signed_string.encode("utf-8"),
        hashlib.sha256,
    ).hexdigest()

    return hmac.compare_digest(provided_sig, expected_sig)


def handle_success(data):
    print(f"Transaction {data['id']} succeeded: {data['amount']} {data['currency']}")


def handle_failure(data):
    print(f"Transaction {data['id']} failed")