Signature verification

import crypto from "node:crypto";

const TOLERANCE_SECONDS = 300; // reject events more than 5 minutes off

/**
 * Verify an X-Certivant-Signature-V2 header ("t=<unix>,v1=<hex>[,v1=<hex>]").
 * `secrets` may hold two values while you rotate (old + new).
 */
export function verifyCertivantSignature(
  rawBody: string,
  signatureHeader: string | undefined | null,
  secrets: string | string[],
  toleranceSeconds = TOLERANCE_SECONDS,
): boolean {
  if (!signatureHeader) return false;

  // 1. Parse "t=...,v1=...,v1=..."
  let t = "";
  const candidates: string[] = [];
  for (const part of signatureHeader.split(",")) {
    const i = part.indexOf("=");
    if (i < 0) continue;
    const key = part.slice(0, i).trim();
    const value = part.slice(i + 1).trim();
    if (key === "t") t = value;
    else if (key === "v1") candidates.push(value);
  }
  if (!/^\d+$/.test(t) || candidates.length === 0) return false;

  // 2. Reject stale / future-dated events (replay protection)
  if (Math.abs(Date.now() / 1000 - Number(t)) > toleranceSeconds) return false;

  // 3. Recompute HMAC over `${t}.${rawBody}` — the timestamp is part of the
  //    signed message — and compare in constant time against every v1.
  for (const secret of Array.isArray(secrets) ? secrets : [secrets]) {
    const expected = crypto
      .createHmac("sha256", secret)
      .update(`${t}.${rawBody}`)
      .digest("hex");
    for (const candidate of candidates) {
      const a = Buffer.from(expected);
      const b = Buffer.from(candidate);
      if (a.length === b.length && crypto.timingSafeEqual(a, b)) return true;
    }
  }
  return false;
}

// Hono usage
app.post("/webhooks/certivant", async (c) => {
  const rawBody = await c.req.text();   // <-- raw, not c.req.json()
  const sig = c.req.header("X-Certivant-Signature-V2");
  if (!verifyCertivantSignature(rawBody, sig, process.env.CERTIVANT_WEBHOOK_SECRET!)) {
    return c.json({ error: "Invalid signature" }, 401);
  }
  const payload = JSON.parse(rawBody);
  // De-duplicate on c.req.header("X-Certivant-Event-Id"), then handle payload
  return c.json({ received: true });
});

// Express usage — needs express.raw() to keep the raw body
import express from "express";
const app = express();
app.post(
  "/webhooks/certivant",
  express.raw({ type: "application/json" }),
  (req, res) => {
    const rawBody = req.body.toString();  // Buffer -> string
    const sig = req.header("X-Certivant-Signature-V2");
    if (!verifyCertivantSignature(rawBody, sig, process.env.CERTIVANT_WEBHOOK_SECRET!)) {
      return res.status(401).json({ error: "Invalid signature" });
    }
    const payload = JSON.parse(rawBody);
    res.json({ received: true });
  },
);

import hashlib
import hmac
import re
import time

TOLERANCE_SECONDS = 300  # reject events more than 5 minutes off


def verify_certivant_signature(
    raw_body: bytes,
    signature_header: str | None,
    secrets: str | list[str],
    tolerance_seconds: int = TOLERANCE_SECONDS,
) -> bool:
    """Verify X-Certivant-Signature-V2 ("t=<unix>,v1=<hex>[,v1=<hex>]").

    `secrets` may hold two values while you rotate (old + new)."""
    if not signature_header:
        return False

    # 1. Parse "t=...,v1=...,v1=..."
    t = ""
    candidates: list[str] = []
    for part in signature_header.split(","):
        key, sep, value = part.partition("=")
        if not sep:
            continue
        key, value = key.strip(), value.strip()
        if key == "t":
            t = value
        elif key == "v1":
            candidates.append(value)
    if not re.fullmatch(r"\d+", t) or not candidates:
        return False

    # 2. Reject stale / future-dated events (replay protection)
    if abs(time.time() - int(t)) > tolerance_seconds:
        return False

    # 3. Recompute HMAC over `${t}.${rawBody}`. We feed bytes so the body
    #    is never accidentally re-encoded; compare in constant time.
    signed = t.encode("ascii") + b"." + raw_body
    for secret in [secrets] if isinstance(secrets, str) else secrets:
        expected = hmac.new(secret.encode("utf-8"), signed, hashlib.sha256).hexdigest()
        if any(hmac.compare_digest(expected, c) for c in candidates):
            return True
    return False


# FastAPI
from fastapi import FastAPI, HTTPException, Request
import os

app = FastAPI()

@app.post("/webhooks/certivant")
async def handle(request: Request):
    raw_body = await request.body()          # bytes, untouched
    sig = request.headers.get("x-certivant-signature-v2")
    if not verify_certivant_signature(raw_body, sig, os.environ["CERTIVANT_WEBHOOK_SECRET"]):
        raise HTTPException(status_code=401, detail="Invalid signature")
    payload = await request.json()
    return {"received": True}

# Flask
from flask import Flask, abort, request

app = Flask(__name__)

@app.post("/webhooks/certivant")
def handle():
    raw_body = request.get_data()
    sig = request.headers.get("X-Certivant-Signature-V2")
    if not verify_certivant_signature(raw_body, sig, os.environ["CERTIVANT_WEBHOOK_SECRET"]):
        abort(401)
    payload = request.get_json()
    return {"received": True}

package webhook

import (
    "crypto/hmac"
    "crypto/sha256"
    "encoding/hex"
    "io"
    "math"
    "net/http"
    "os"
    "strconv"
    "strings"
    "time"
)

const toleranceSeconds = 300

// Verifies X-Certivant-Signature-V2 ("t=<unix>,v1=<hex>[,v1=<hex>]").
// Pass two secrets while rotating (old + new).
func verifyCertivantSignature(rawBody []byte, header string, secrets ...string) bool {
    var t string
    var candidates []string
    for _, part := range strings.Split(header, ",") {
        kv := strings.SplitN(strings.TrimSpace(part), "=", 2)
        if len(kv) != 2 {
            continue
        }
        switch kv[0] {
        case "t":
            t = kv[1]
        case "v1":
            candidates = append(candidates, kv[1])
        }
    }
    ts, err := strconv.ParseInt(t, 10, 64)
    if err != nil || len(candidates) == 0 {
        return false
    }
    if math.Abs(float64(time.Now().Unix()-ts)) > toleranceSeconds {
        return false // replay protection
    }
    for _, secret := range secrets {
        mac := hmac.New(sha256.New, []byte(secret))
        mac.Write([]byte(t + "."))
        mac.Write(rawBody)
        expected := hex.EncodeToString(mac.Sum(nil))
        for _, c := range candidates {
            if hmac.Equal([]byte(expected), []byte(c)) {
                return true
            }
        }
    }
    return false
}

func Handler(w http.ResponseWriter, r *http.Request) {
    rawBody, err := io.ReadAll(r.Body)
    if err != nil {
        http.Error(w, "read body", http.StatusBadRequest)
        return
    }
    sig := r.Header.Get("X-Certivant-Signature-V2")
    if !verifyCertivantSignature(rawBody, sig, os.Getenv("CERTIVANT_WEBHOOK_SECRET")) {
        http.Error(w, "invalid signature", http.StatusUnauthorized)
        return
    }
    // unmarshal rawBody into your struct and handle...
    w.WriteHeader(http.StatusOK)
}

require "openssl"

TOLERANCE_SECONDS = 300

# Verifies X-Certivant-Signature-V2 ("t=<unix>,v1=<hex>[,v1=<hex>]").
# `secrets` may hold two values while you rotate (old + new).
def verify_certivant_signature(raw_body, signature_header, secrets)
  return false if signature_header.nil? || signature_header.empty?
  parts = signature_header.split(",").map { |p| p.strip.split("=", 2) }
  t = parts.find { |k, _| k == "t" }&.last.to_s
  candidates = parts.select { |k, _| k == "v1" }.map(&:last)
  return false unless t.match?(/\A\d+\z/) && !candidates.empty?
  return false if (Time.now.to_i - t.to_i).abs > TOLERANCE_SECONDS # replay protection

  Array(secrets).any? do |secret|
    expected = OpenSSL::HMAC.hexdigest("SHA256", secret, "#{t}.#{raw_body}")
    candidates.any? { |c| Rack::Utils.secure_compare(expected, c) }
  end
end

# Rails
class WebhooksController < ApplicationController
  skip_before_action :verify_authenticity_token, only: [:certivant]

  def certivant
    raw_body = request.raw_post
    sig = request.headers["X-Certivant-Signature-V2"]
    unless verify_certivant_signature(raw_body, sig, ENV["CERTIVANT_WEBHOOK_SECRET"])
      head :unauthorized and return
    end
    payload = JSON.parse(raw_body)
    # ... handle
    render json: { received: true }
  end
end

<?php
const CERTIVANT_TOLERANCE_SECONDS = 300;

// Verifies X-Certivant-Signature-V2 ("t=<unix>,v1=<hex>[,v1=<hex>]").
// $secrets may hold two values while you rotate (old + new).
function verify_certivant_signature(string $rawBody, ?string $signatureHeader, string|array $secrets): bool {
    if ($signatureHeader === null || $signatureHeader === "") return false;
    $t = "";
    $candidates = [];
    foreach (explode(",", $signatureHeader) as $part) {
        $kv = explode("=", trim($part), 2);
        if (count($kv) !== 2) continue;
        if ($kv[0] === "t") $t = $kv[1];
        elseif ($kv[0] === "v1") $candidates[] = $kv[1];
    }
    if (!ctype_digit($t) || count($candidates) === 0) return false;
    if (abs(time() - (int) $t) > CERTIVANT_TOLERANCE_SECONDS) return false; // replay protection

    foreach ((array) $secrets as $secret) {
        $expected = hash_hmac("sha256", $t . "." . $rawBody, $secret);
        foreach ($candidates as $candidate) {
            if (hash_equals($expected, $candidate)) return true;
        }
    }
    return false;
}

// Plain PHP
$rawBody = file_get_contents("php://input");
$sig = $_SERVER["HTTP_X_CERTIVANT_SIGNATURE_V2"] ?? null;
if (!verify_certivant_signature($rawBody, $sig, getenv("CERTIVANT_WEBHOOK_SECRET"))) {
    http_response_code(401);
    echo json_encode(["error" => "Invalid signature"]);
    exit;
}
$payload = json_decode($rawBody, true);
// ... handle
echo json_encode(["received" => true]);

// Laravel
public function certivant(\Illuminate\Http\Request $request) {
    $rawBody = $request->getContent();
    $sig = $request->header("X-Certivant-Signature-V2");
    if (!verify_certivant_signature($rawBody, $sig, env("CERTIVANT_WEBHOOK_SECRET"))) {
        abort(401);
    }
    $payload = json_decode($rawBody, true);
    return response()->json(["received" => true]);
}