Verifying CERTEN webhook signatures

Every webhook the gateway delivers carries a signed payload. The wire format is:

POST <your-endpoint> HTTP/1.1
Content-Type: application/json
User-Agent: Certen-Gateway/<version>
X-Certen-Event: transaction.completed
X-Certen-Delivery: 7c0c5a8c-2bdf-4f3f-8a9c-bf3a7e4d2e51   # unique per attempt
X-Certen-Timestamp: 1716624000                            # unix seconds
X-Certen-Signature: t=1716624000,v1=3a4f...e7b9            # HMAC-SHA256

{"a":1,"b":"hello"}

The HMAC input is:

<timestamp> "." <canonical-JSON of payload>

— that is, the literal timestamp from t=, a dot, then the canonical JSON of the payload (sorted keys, no whitespace, RFC-stable encoding). The body the gateway sends on the wire IS the canonical JSON, so verifiers don't need to re-canonicalize — they can hash the request body bytes directly.

What to verify

Parse X-Certen-Signature into t=<unix> and v1=<hex> parts.
Compute HMAC-SHA256(secret, t + "." + body_bytes).
Compare to the v1= value in constant time.
Reject the request if |now - t| exceeds your tolerance window (we recommend ±5 minutes). This is your replay defense.
(Optional) Deduplicate on X-Certen-Delivery so a retry doesn't apply the same business action twice — the delivery ID is stable across attempts of the same delivery row.

Reference implementations

Node.js / TypeScript

import { createHmac, timingSafeEqual } from 'crypto';

interface VerifyResult {
  ok: boolean;
  reason?: string;
}

export function verifyCertenWebhook(opts: {
  rawBody: Buffer;            // EXACT bytes you received — don't JSON.parse + re-stringify
  signatureHeader: string;    // value of X-Certen-Signature
  secret: string;             // your webhook_endpoints.secret
  toleranceSec?: number;      // default 300
}): VerifyResult {
  const tolerance = opts.toleranceSec ?? 300;
  const parts = Object.fromEntries(
    opts.signatureHeader.split(',').map((p) => {
      const eq = p.indexOf('=');
      return [p.slice(0, eq).trim(), p.slice(eq + 1).trim()];
    }),
  );
  const t = Number(parts['t']);
  const v1 = parts['v1'];
  if (!Number.isFinite(t)) return { ok: false, reason: 'bad-timestamp' };
  if (!v1 || !/^[0-9a-f]{64}$/.test(v1)) return { ok: false, reason: 'bad-v1' };
  const skew = Math.abs(Math.floor(Date.now() / 1000) - t);
  if (skew > tolerance) return { ok: false, reason: 'stale' };

  const signedString = `${t}.${opts.rawBody.toString('utf8')}`;
  const expected = createHmac('sha256', opts.secret).update(signedString).digest();
  const provided = Buffer.from(v1, 'hex');
  if (expected.length !== provided.length) return { ok: false, reason: 'length-mismatch' };
  if (!timingSafeEqual(expected, provided)) return { ok: false, reason: 'signature-mismatch' };
  return { ok: true };
}

In Express:

import express from 'express';

const app = express();

// IMPORTANT: capture the raw bytes BEFORE JSON parsing.
app.post(
  '/webhooks/certen',
  express.raw({ type: 'application/json' }),
  (req, res) => {
    const result = verifyCertenWebhook({
      rawBody: req.body,
      signatureHeader: req.header('x-certen-signature') ?? '',
      secret: process.env.CERTEN_WEBHOOK_SECRET ?? '',
    });
    if (!result.ok) {
      return res.status(400).json({ error: `signature: ${result.reason}` });
    }
    const event = JSON.parse(req.body.toString('utf8'));
    // ... handle event ...
    res.status(200).end();
  },
);

Or via the SDK:

import { verifyWebhookSignature } from '@certen.io/sdk';
// (re-exported from the gateway code; same algorithm)

Python

import hmac, hashlib, time

def verify_certen_webhook(*, raw_body: bytes, signature_header: str,
                          secret: str, tolerance_sec: int = 300) -> bool:
    parts = dict(p.strip().split("=", 1) for p in signature_header.split(","))
    try:
        t = int(parts["t"])
        v1 = parts["v1"]
    except (KeyError, ValueError):
        return False
    if abs(int(time.time()) - t) > tolerance_sec:
        return False
    signed = f"{t}.".encode("utf-8") + raw_body
    expected = hmac.new(secret.encode("utf-8"), signed, hashlib.sha256).hexdigest()
    # hmac.compare_digest is constant-time.
    return hmac.compare_digest(expected, v1)

Flask:

from flask import Flask, request

app = Flask(__name__)

@app.post("/webhooks/certen")
def webhook():
    ok = verify_certen_webhook(
        raw_body=request.get_data(),
        signature_header=request.headers.get("X-Certen-Signature", ""),
        secret=os.environ["CERTEN_WEBHOOK_SECRET"],
    )
    if not ok:
        return {"error": "signature"}, 400
    # request.get_json() is now safe to use; event payload is verified.
    event = request.get_json()
    return "", 200

Go

package webhook

import (
    "crypto/hmac"
    "crypto/sha256"
    "encoding/hex"
    "fmt"
    "strconv"
    "strings"
    "time"
)

func VerifyCertenWebhook(rawBody []byte, signatureHeader, secret string, toleranceSec int) error {
    var t int64
    var v1 string
    for _, part := range strings.Split(signatureHeader, ",") {
        eq := strings.IndexByte(part, '=')
        if eq < 0 {
            continue
        }
        k := strings.TrimSpace(part[:eq])
        v := strings.TrimSpace(part[eq+1:])
        switch k {
        case "t":
            n, err := strconv.ParseInt(v, 10, 64)
            if err == nil {
                t = n
            }
        case "v1":
            v1 = v
        }
    }
    if t == 0 || v1 == "" {
        return fmt.Errorf("malformed signature header")
    }
    if skew := time.Now().Unix() - t; skew > int64(toleranceSec) || skew < -int64(toleranceSec) {
        return fmt.Errorf("timestamp outside tolerance")
    }

    mac := hmac.New(sha256.New, []byte(secret))
    fmt.Fprintf(mac, "%d.", t)
    mac.Write(rawBody)
    expected := mac.Sum(nil)

    provided, err := hex.DecodeString(v1)
    if err != nil {
        return fmt.Errorf("bad v1 hex: %w", err)
    }
    if !hmac.Equal(expected, provided) {
        return fmt.Errorf("signature mismatch")
    }
    return nil
}

net/http:

http.HandleFunc("/webhooks/certen", func(w http.ResponseWriter, r *http.Request) {
    body, _ := io.ReadAll(r.Body)
    if err := VerifyCertenWebhook(body, r.Header.Get("X-Certen-Signature"),
        os.Getenv("CERTEN_WEBHOOK_SECRET"), 300); err != nil {
        http.Error(w, "signature: "+err.Error(), http.StatusBadRequest)
        return
    }
    // body is verified; safe to json.Unmarshal
    w.WriteHeader(http.StatusOK)
})

Rust

use hmac::{Hmac, Mac};
use sha2::Sha256;
use std::time::{SystemTime, UNIX_EPOCH};
use subtle::ConstantTimeEq;

type HmacSha256 = Hmac<Sha256>;

pub fn verify_certen_webhook(
    raw_body: &[u8],
    signature_header: &str,
    secret: &str,
    tolerance_sec: u64,
) -> Result<(), &'static str> {
    let mut t: i64 = 0;
    let mut v1_hex: &str = "";
    for part in signature_header.split(',') {
        let mut it = part.splitn(2, '=');
        match (it.next(), it.next()) {
            (Some("t"), Some(v)) => t = v.trim().parse().unwrap_or(0),
            (Some(k), Some(v)) if k.trim() == "v1" => v1_hex = v.trim(),
            _ => {}
        }
    }
    if t == 0 || v1_hex.is_empty() {
        return Err("malformed signature");
    }
    let now = SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_secs() as i64;
    if (now - t).unsigned_abs() > tolerance_sec {
        return Err("timestamp outside tolerance");
    }

    let mut mac = HmacSha256::new_from_slice(secret.as_bytes()).map_err(|_| "bad secret")?;
    mac.update(format!("{}.", t).as_bytes());
    mac.update(raw_body);
    let expected = mac.finalize().into_bytes();

    let provided = hex::decode(v1_hex).map_err(|_| "bad v1 hex")?;
    if expected.ct_eq(&provided).into() { Ok(()) } else { Err("signature mismatch") }
}

Common pitfalls

Re-stringifying the body. Frameworks like Express's express.json() middleware parse the body BEFORE your handler runs, so you can't get the original bytes back. Always use express.raw({ type: 'application/json' }) (or equivalent) for the verification path, then parse manually after the signature check passes.
Whitespace differences. The gateway sends canonical JSON ({"a":1,"b":"hello"}) — no spaces between separators. If your framework reformats the body before you hash it, the signature won't match. Hash the raw bytes, not a re-serialized object.
Comparing with ===. Use a constant-time compare (timingSafeEqual in Node, hmac.compare_digest in Python, hmac.Equal in Go, subtle::ConstantTimeEq in Rust). String equality is fast but timing-leaky.
Skipping the timestamp check. A captured request body+signature is replayable forever without the timestamp comparison. Always enforce a tolerance window (300 seconds is a good default).
Storing the secret in source. The secret should live in your secrets manager (Vault, AWS Secrets Manager, Doppler, etc), not in the repo. The gateway returns the secret exactly once at endpoint-creation time — capture it then and don't try to retrieve it later.

Rotating an endpoint's secret

The gateway doesn't currently support per-endpoint secret rotation in the same way OAuth client secrets rotate. The pragmatic pattern:

Create a NEW webhook endpoint with the same URL + event_types and a freshly-generated secret.
Wait until you see at least one delivery hit the new endpoint successfully (POST /v1/admin/webhooks/endpoints/:id/verify if you want to force-test).
Delete the old endpoint.

This gives you a clean cut-over window where both secrets are accepted by your verifier (you'd check the X-Certen-Delivery header or maintain a small webhook_endpoint_id -> secret map in your code).