Verification examples

import express from "express";
import crypto from "node:crypto";

const app = express();
const WEBHOOK_SECRET = process.env.KYNVA_WEBHOOK_SECRET!;

// IMPORTANT: capture the raw body, not the parsed JSON. The signature is over bytes.
app.post(
  "/webhooks/kynva",
  express.raw({ type: "application/json" }),
  (req, res) => {
    const signature = (req.header("x-renderforge-signature") ?? "").replace(/^sha256=/, "");
    const expected = crypto
      .createHmac("sha256", WEBHOOK_SECRET)
      .update(req.body)            // Buffer of raw bytes
      .digest("hex");

    const ok =
      signature.length === expected.length &&
      crypto.timingSafeEqual(
        Buffer.from(signature, "hex"),
        Buffer.from(expected, "hex"),
      );

    if (!ok) {
      return res.status(401).send("bad signature");
    }

    const payload = JSON.parse(req.body.toString("utf8"));

    // Dedupe at-least-once delivery
    const deliveryId = req.header("x-renderforge-delivery");
    // ... check Redis / DB, return 200 if already processed ...

    switch (payload.event) {
      case "render.completed":
        // do your thing
        break;
      case "render.failed":
        // alert someone
        break;
    }

    res.sendStatus(200);
  },
);

app.listen(3000);

import hmac, hashlib, json, os
from fastapi import FastAPI, Header, HTTPException, Request

app = FastAPI()
WEBHOOK_SECRET = os.environ["KYNVA_WEBHOOK_SECRET"]

@app.post("/webhooks/kynva")
async def kynva_webhook(
    request: Request,
    x_renderforge_signature: str = Header(...),
    x_renderforge_delivery: str = Header(...),
):
    raw = await request.body()  # raw bytes — required for HMAC

    signature = x_renderforge_signature.removeprefix("sha256=")
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        raw,
        hashlib.sha256,
    ).hexdigest()

    if not hmac.compare_digest(signature, expected):
        raise HTTPException(status_code=401, detail="bad signature")

    payload = json.loads(raw)

    # Dedupe at-least-once delivery on x_renderforge_delivery here.

    if payload["event"] == "render.completed":
        # do your thing
        pass

    return {"ok": True}

package main

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

var webhookSecret = os.Getenv("KYNVA_WEBHOOK_SECRET")

func handle(w http.ResponseWriter, r *http.Request) {
	body, err := io.ReadAll(r.Body)
	if err != nil {
		http.Error(w, "bad body", http.StatusBadRequest)
		return
	}

	signature := strings.TrimPrefix(r.Header.Get("X-RenderForge-Signature"), "sha256=")

	mac := hmac.New(sha256.New, []byte(webhookSecret))
	mac.Write(body)
	expected := hex.EncodeToString(mac.Sum(nil))

	if !hmac.Equal([]byte(signature), []byte(expected)) {
		http.Error(w, "bad signature", http.StatusUnauthorized)
		return
	}

	var payload struct {
		Event string `json:"event"`
	}
	_ = json.Unmarshal(body, &payload)

	// Dedupe on r.Header.Get("X-RenderForge-Delivery") here.

	w.WriteHeader(http.StatusOK)
}

func main() {
	http.HandleFunc("/webhooks/kynva", handle)
	http.ListenAndServe(":3000", nil)
}