Signed Webhooks: HMAC Events and Retries

GET    /v1/webhooks                        list your endpoints
POST   /v1/webhooks                        register one (secret returned once)
PATCH  /v1/webhooks/{id}                   enable/disable, change events
POST   /v1/webhooks/{id}/rotate-secret     new secret; the old one keeps signing for 24h
DELETE /v1/webhooks/{id}                   remove one

{
  "id": "<webhook id>",
  "secret": "whsec_...",
  "previousSecretValidUntil": "2026-09-25T12:00:00.000Z"
}

import crypto from "node:crypto";

// secret = the whsec_... shown once at endpoint creation. Verify the RAW
// bytes, never re-serialized JSON: in Express, mount the route with
// express.raw({ type: "application/json" }) so req.body is the exact Buffer.
//
//   app.post("/hooks/seomatic", express.raw({ type: "application/json" }),
//     (req, res) => {
//       if (!verify(req, process.env.SEOMATIC_WHSEC)) return res.sendStatus(400);
//       const event = JSON.parse(req.body.toString("utf8"));
//       res.sendStatus(200);                // answer 2xx within 10 seconds
//     });
export function verify(req, secret, toleranceSec = 300) {
  // "t=...,v1=..." - during a secret rotation there is one v1 per active
  // secret ("t=...,v1=<new>,v1=<old>"): accept the event if ANY v1 matches.
  const sig = req.headers["x-seomatic-signature"];
  const ts = Number(req.headers["x-seomatic-timestamp"]);
  const body = req.body.toString("utf8");               // the EXACT bytes received

  // 1) Reject a stale timestamp (replay guard).
  if (!ts || Math.abs(Date.now() / 1000 - ts) > toleranceSec) return false;

  // 2) Recompute HMAC-SHA256 over `${ts}.${body}` and constant-time compare.
  const expected = crypto
    .createHmac("sha256", secret)
    .update(`${ts}.${body}`)
    .digest("hex");
  const candidates = [...(sig || "").matchAll(/v1=([a-f0-9]+)/g)].map(m => m[1]);
  return candidates.some(
    got =>
      got.length === expected.length &&
      crypto.timingSafeEqual(Buffer.from(got), Buffer.from(expected))
  );
}

// Rotating the secret: POST /api/v1/webhooks/{id}/rotate-secret returns the
// new whsec_... once. For 24 hours every delivery is signed with BOTH secrets,
// so deploy the new one any time in that window without dropping events.
// Rotating again inside the window replaces the older secret immediately.

import hashlib, hmac, re, time

def verify(headers, raw_body: bytes, secret: str, tolerance_sec: int = 300) -> bool:
    sig = headers.get("X-Seomatic-Signature", "")
    try:
        ts = int(headers.get("X-Seomatic-Timestamp", ""))
    except ValueError:
        return False

    # 1) Reject a stale timestamp (replay guard).
    if abs(time.time() - ts) > tolerance_sec:
        return False

    # 2) Recompute HMAC-SHA256 over f"{ts}.{body}" and constant-time compare.
    #    During a rotation there is one v1 per active secret: accept ANY match.
    expected = hmac.new(
        secret.encode(), f"{ts}.".encode() + raw_body, hashlib.sha256
    ).hexdigest()
    return any(
        hmac.compare_digest(got, expected)
        for got in re.findall(r"v1=([a-f0-9]+)", sig)
    )

# Flask: verify(request.headers, request.get_data(), os.environ["SEOMATIC_WHSEC"])