Get notified with webhooks

<?php
// webhook.php: point your webhook endpoint at this file.
$secret = getenv('RASTERNIVO_WEBHOOK_SECRET'); // whsec_...
$body = file_get_contents('php://input');      // the raw body, before any JSON parsing

$timestamp = 0;
$signatures = [];
foreach (explode(',', $_SERVER['HTTP_RASTERNIVO_SIGNATURE'] ?? '') as $part) {
    [$name, $value] = array_pad(explode('=', $part, 2), 2, '');
    if ($name === 't') $timestamp = (int) $value;
    if ($name === 'v1') $signatures[] = $value;
}

$expected = hash_hmac('sha256', "$timestamp.$body", $secret);
$valid = false;
foreach ($signatures as $signature) {
    $valid = $valid || hash_equals($expected, $signature);
}
if (!$valid || abs(time() - $timestamp) > 300) { // also rejects replays older than 5 minutes
    http_response_code(400);
    exit;
}

$event = json_decode($body, true);
// De-duplicate with $_SERVER['HTTP_RASTERNIVO_EVENT_ID'], then do your work.
error_log($event['type'] . ' ' . $event['data']['conversion']['id']);
http_response_code(200);

# webhook.py: a small receiver using only the standard library.
import hashlib
import hmac
import json
import os
import time
from http.server import BaseHTTPRequestHandler, HTTPServer

SECRET = os.environ["RASTERNIVO_WEBHOOK_SECRET"]  # whsec_...


def is_valid(body: bytes, header: str) -> bool:
    parts = [p.split("=", 1) for p in header.split(",") if "=" in p]
    timestamp = next((v for k, v in parts if k == "t"), "0")
    if abs(time.time() - int(timestamp)) > 300:  # reject replays older than 5 minutes
        return False
    expected = hmac.new(SECRET.encode(), f"{timestamp}.".encode() + body, hashlib.sha256).hexdigest()
    return any(k == "v1" and hmac.compare_digest(expected, v) for k, v in parts)


class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        body = self.rfile.read(int(self.headers.get("Content-Length", 0)))  # raw body
        if not is_valid(body, self.headers.get("RasterNivo-Signature", "")):
            self.send_response(400)
            self.end_headers()
            return
        event = json.loads(body)
        # De-duplicate with the RasterNivo-Event-Id header, then do your work.
        print(event["type"], event["data"]["conversion"]["id"])
        self.send_response(200)
        self.end_headers()


HTTPServer(("", int(os.environ.get("PORT", 3000))), Handler).serve_forever()

// webhook.mjs: a small receiver using only built-in modules.
import { createHmac, timingSafeEqual } from "node:crypto";
import { createServer } from "node:http";

const secret = process.env.RASTERNIVO_WEBHOOK_SECRET; // whsec_...

function isValid(body, header = "") {
  const parts = header.split(",").map((p) => p.split("="));
  const timestamp = parts.find(([k]) => k === "t")?.[1] ?? "0";
  if (Math.abs(Date.now() / 1000 - Number(timestamp)) > 300) return false; // reject old replays
  const expected = Buffer.from(createHmac("sha256", secret).update(`${timestamp}.`).update(body).digest("hex"));
  return parts.some(([k, v]) => k === "v1" && v.length === expected.length && timingSafeEqual(Buffer.from(v), expected));
}

createServer((req, res) => {
  const chunks = [];
  req.on("data", (c) => chunks.push(c));
  req.on("end", () => {
    const body = Buffer.concat(chunks); // the raw body, before any JSON parsing
    if (!isValid(body, req.headers["rasternivo-signature"])) {
      res.writeHead(400).end();
      return;
    }
    const event = JSON.parse(body);
    // De-duplicate with the rasternivo-event-id header, then do your work.
    console.log(event.type, event.data.conversion.id);
    res.writeHead(200).end();
  });
}).listen(process.env.PORT ?? 3000);