Webhooks

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

app = Flask(__name__)
secret = os.environ['QUENTLI_WEBHOOK_SECRET'].encode()

@app.post('/webhooks/quentli')
def webhook():
    timestamp = request.headers.get('X-Quentli-Timestamp')
    received = request.headers.get('X-Quentli-Signature')
    raw_body = request.get_data()
    expected = hmac.new(secret, f'{timestamp}.'.encode() + raw_body, hashlib.sha256).hexdigest()

    if not timestamp or not received or not hmac.compare_digest(received, expected):
        abort(401)

    event = json.loads(raw_body)
    # Procesa event
    return '', 200

<?php

$secret = $_ENV['QUENTLI_WEBHOOK_SECRET'];
$rawBody = file_get_contents('php://input');
$timestamp = $_SERVER['HTTP_X_QUENTLI_TIMESTAMP'] ?? '';
$received = $_SERVER['HTTP_X_QUENTLI_SIGNATURE'] ?? '';

$expected = hash_hmac('sha256', $timestamp . '.' . $rawBody, $secret);
if (!$timestamp || !$received || !hash_equals($expected, $received)) {
    http_response_code(401);
    exit;
}

$event = json_decode($rawBody, true);
// Procesa $event
http_response_code(200);

using System.Security.Cryptography;
using System.Text;
using System.Text.Json;

var app = WebApplication.Create(args);
var secret = Environment.GetEnvironmentVariable("QUENTLI_WEBHOOK_SECRET")!;

app.MapPost("/webhooks/quentli", async (HttpRequest request) =>
{
    using var reader = new StreamReader(request.Body);
    var rawBody = await reader.ReadToEndAsync();
    var timestamp = request.Headers["X-Quentli-Timestamp"].ToString();
    var received = request.Headers["X-Quentli-Signature"].ToString();

    if (string.IsNullOrEmpty(timestamp) || string.IsNullOrEmpty(received))
        return Results.Unauthorized();

    var expected = Convert.ToHexString(HMACSHA256.HashData(Encoding.UTF8.GetBytes(secret), Encoding.UTF8.GetBytes($"{timestamp}.{rawBody}"))).ToLowerInvariant();
    if (!CryptographicOperations.FixedTimeEquals(Encoding.UTF8.GetBytes(received), Encoding.UTF8.GetBytes(expected)))
        return Results.Unauthorized();

    var eventData = JsonDocument.Parse(rawBody);
    // Procesa eventData
    return Results.Ok();
});

app.Run();

import (
    "crypto/hmac"
    "crypto/sha256"
    "encoding/hex"
    "encoding/json"
    "fmt"
    "io"
    "net/http"
    "os"
)

func webhook(w http.ResponseWriter, r *http.Request) {
    rawBody, _ := io.ReadAll(r.Body)
    timestamp := r.Header.Get("X-Quentli-Timestamp")
    received := r.Header.Get("X-Quentli-Signature")

    mac := hmac.New(sha256.New, []byte(os.Getenv("QUENTLI_WEBHOOK_SECRET")))
    fmt.Fprintf(mac, "%s.%s", timestamp, rawBody)
    expected := hex.EncodeToString(mac.Sum(nil))

    if timestamp == "" || received == "" || !hmac.Equal([]byte(received), []byte(expected)) {
        http.Error(w, "invalid signature", http.StatusUnauthorized)
        return
    }

    var event map[string]any
    json.Unmarshal(rawBody, &event)
    // Procesa event
    w.WriteHeader(http.StatusOK)
}