Webhook firmati.

import hashlib
import hmac
import time


def verify_signature(secret: str, raw_body: bytes, header: str, tolerance: int = 300, now: float | None = None) -> bool:
    """True when `header` is a valid signature of `raw_body` made with `secret`, at most `tolerance` seconds old."""
    try:
        parts = dict(item.split("=", 1) for item in header.split(","))
        timestamp, received = int(parts["t"]), parts["v1"]
    except (KeyError, ValueError):
        return False
    if abs((time.time() if now is None else now) - timestamp) > tolerance:
        return False
    expected = hmac.new(secret.encode(), f"{timestamp}.".encode() + raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, received)

import json
import os

from fastapi import FastAPI, HTTPException, Request

# verify_signature is the function above (same file or imported)
app = FastAPI()
SECRET = os.environ["SEOPRIMO_WEBHOOK_SECRET"]  # whsec_...


@app.post("/webhooks/seoprimo")
async def seoprimo_webhook(request: Request):
    raw_body = await request.body()  # the bytes as received, before any JSON parsing
    if not verify_signature(SECRET, raw_body, request.headers.get("x-seoprimo-signature", "")):
        raise HTTPException(status_code=401, detail="invalid signature")
    event = json.loads(raw_body)
    # answer fast (2xx within 10 seconds) and process in the background; use event["id"] to deduplicate
    return {"received": True}

const crypto = require("node:crypto");

// rawBody: a Buffer with the bytes as received (before any JSON parsing)
function verifySignature(secret, rawBody, header, toleranceSeconds = 300, nowSeconds = Date.now() / 1000) {
  let timestamp;
  let received;
  try {
    const parts = Object.fromEntries(
      header.split(",").map((item) => {
        const i = item.indexOf("=");
        return [item.slice(0, i), item.slice(i + 1)];
      }),
    );
    timestamp = Number.parseInt(parts.t, 10);
    received = parts.v1;
    if (!Number.isFinite(timestamp) || typeof received !== "string") return false;
  } catch {
    return false;
  }
  if (Math.abs(nowSeconds - timestamp) > toleranceSeconds) return false;
  const expected = crypto
    .createHmac("sha256", secret)
    .update(`${timestamp}.`)
    .update(rawBody)
    .digest("hex");
  const a = Buffer.from(expected);
  const b = Buffer.from(received);
  return a.length === b.length && crypto.timingSafeEqual(a, b);
}

module.exports = { verifySignature };

const express = require("express");
const { verifySignature } = require("./verify");

const app = express();

// express.raw keeps the body as a Buffer: do NOT use express.json() on this route
app.post("/webhooks/seoprimo", express.raw({ type: "application/json" }), (req, res) => {
  const signature = req.get("X-SeoPrimo-Signature") || "";
  if (!verifySignature(process.env.SEOPRIMO_WEBHOOK_SECRET, req.body, signature)) {
    return res.status(401).json({ error: "invalid_signature" });
  }
  const event = JSON.parse(req.body.toString("utf8"));
  res.status(200).json({ received: true }); // answer fast (2xx within 10 seconds)
  // ... process event.type / event.data in the background; use event.id to deduplicate
});

app.listen(3000);