Webhook Integration

from flask import Flask, request, jsonify
import hmac
import hashlib
import threading

app = Flask(__name__)

@app.route("/webhooks/yativo", methods=["POST"])
def webhook():
    raw_body = request.get_data()
    signature = request.headers.get("X-Webhook-Signature", "")

    if not verify_signature(raw_body, signature, os.environ["YATIVO_WEBHOOK_SECRET"]):
        return jsonify({"error": "Invalid signature"}), 401

    event = request.get_json(force=True)

    # Respond immediately, process in background
    thread = threading.Thread(target=handle_event, args=(event,))
    thread.daemon = True
    thread.start()

    return "", 200

<?php

$rawBody = file_get_contents("php://input");
$signature = $_SERVER["HTTP_X_WEBHOOK_SIGNATURE"] ?? "";
$secret = getenv("YATIVO_WEBHOOK_SECRET");

if (!verifySignature($rawBody, $signature, $secret)) {
    http_response_code(401);
    echo json_encode(["error" => "Invalid signature"]);
    exit;
}

// Acknowledge immediately
http_response_code(200);
echo "OK";

// Flush output and continue processing
if (function_exists("fastcgi_finish_request")) {
    fastcgi_finish_request();
}

$event = json_decode($rawBody, true);
handleEvent($event);

import hmac
import hashlib

def verify_signature(raw_body: bytes, signature_header: str, secret: str) -> bool:
    if not signature_header:
        return False

    expected = hmac.new(
        secret.encode("utf-8"),
        raw_body,
        hashlib.sha256,
    ).hexdigest()

    # Use compare_digest to prevent timing attacks
    return hmac.compare_digest(signature_header, expected)

function verifySignature(string $rawBody, string $signatureHeader, string $secret): bool {
    if (empty($signatureHeader)) {
        return false;
    }

    $expected = hash_hmac("sha256", $rawBody, $secret);

    // hash_equals is timing-safe
    return hash_equals($expected, $signatureHeader);
}