Webhook Integration

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

app = Flask(__name__)

@app.route("/webhooks/yativo", methods=["POST"])
def webhook():
    signature = request.headers.get("X-Yativo-Signature", "")
    timestamp = request.headers.get("X-Yativo-Timestamp", "")
    payload = request.get_json(force=True)

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

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

    return "", 200

<?php

$payload = json_decode(file_get_contents("php://input"), true);
$signature = $_SERVER["HTTP_X_YATIVO_SIGNATURE"] ?? "";
$timestamp = $_SERVER["HTTP_X_YATIVO_TIMESTAMP"] ?? "";
$secret = getenv("YATIVO_WEBHOOK_SECRET");

if (!verifyWebhookSignature($payload, $signature, $timestamp, $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();
}

handleEvent($payload);

import hmac
import hashlib
import json
import time

def verify_webhook_signature(payload: dict, signature_header: str, timestamp: str, secret: str) -> bool:
    if not signature_header or not timestamp:
        return False

    # Reject replays older than 5 minutes
    if abs(time.time() - int(timestamp)) > 300:
        return False

    signed_payload = f"{timestamp}.{json.dumps(payload, separators=(',', ':'))}"
    expected = "sha256=" + hmac.new(
        secret.encode("utf-8"),
        signed_payload.encode("utf-8"),
        hashlib.sha256,
    ).hexdigest()

    return hmac.compare_digest(signature_header, expected)

function verifyWebhookSignature(array $payload, string $signatureHeader, string $timestamp, string $secret): bool {
    if (empty($signatureHeader) || empty($timestamp)) {
        return false;
    }

    // Reject replays older than 5 minutes
    if (abs(time() - intval($timestamp)) > 300) {
        return false;
    }

    $signedPayload = $timestamp . '.' . json_encode($payload);
    $expected = 'sha256=' . hash_hmac('sha256', $signedPayload, $secret);

    return hash_equals($expected, $signatureHeader);
}