Webhooks

import crypto from 'node:crypto';
import express from 'express';

const app = express();

app.post(
    '/webhooks/pulsesend',
    express.raw({ type: 'application/json' }),
    (req, res) => {
        const header = req.get('X-PulseSend-Signature') ?? '';
        const parts = Object.fromEntries(
            header.split(',').map((p) => p.trim().split('=')),
        );
        const t = Number(parts.t);

        if (!t || Math.abs(Date.now() / 1000 - t) > 300)
            return res.sendStatus(400);

        const expected = crypto
            .createHmac('sha256', process.env.PULSESEND_WEBHOOK_SECRET)
            .update(`${t}.${req.body}`)
            .digest('hex');

        const valid =
            parts.v1?.length === expected.length &&
            crypto.timingSafeEqual(
                Buffer.from(parts.v1),
                Buffer.from(expected),
            );

        if (!valid) return res.sendStatus(400);

        const event = JSON.parse(req.body);
        // handle event.type ...
        res.sendStatus(200);
    },
);

use Illuminate\Http\Request;

Route::post('/webhooks/pulsesend', function (Request $request) {
    $payload = $request->getContent();
    parse_str(str_replace(',', '&', (string) $request->header('X-PulseSend-Signature')), $parts);

    $t = (int) ($parts['t'] ?? 0);
    $expected = hash_hmac('sha256', $t.'.'.$payload, config('services.pulsesend.webhook_secret'));

    abort_if(abs(time() - $t) > 300, 400);
    abort_unless(hash_equals($expected, (string) ($parts['v1'] ?? '')), 400);

    $event = json_decode($payload, true);
    // handle $event['type'] ...

    return response()->noContent();
});

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

app = Flask(__name__)

@app.post("/webhooks/pulsesend")
def pulsesend_webhook():
    payload = request.get_data()
    parts = dict(p.strip().split("=", 1) for p in request.headers.get("X-PulseSend-Signature", "").split(",") if "=" in p)
    t = int(parts.get("t", 0))

    expected = hmac.new(os.environ["PULSESEND_WEBHOOK_SECRET"].encode(), f"{t}.".encode() + payload, hashlib.sha256).hexdigest()

    if abs(time.time() - t) > 300 or not hmac.compare_digest(expected, parts.get("v1", "")):
        abort(400)

    event = request.get_json()
    # handle event["type"] ...
    return "", 204