Verification Examples

Below are minimal implementations that verify the signature, validate timestamp drift, and parse the event.

Node.js (Express)

Read raw body with express.raw({ type: "application/json" }).

import crypto from "crypto";
import express from "express";

const app = express();

app.post("/webhooks/fortisx",
  express.raw({ type: "application/json" }),
  (req, res) => {
    const secret = process.env.FORTISX_WEBHOOK_SECRET || "";
    const sigHeader = req.header("X-Fortisx-Signature") || "";
    const eventId = req.header("X-Fortisx-Event-Id") || "";

    const parts = Object.fromEntries(sigHeader.split(",").map(p => p.trim().split("=")));
    const t = Number(parts.t);
    const v1 = String(parts.v1 || "");

    const now = Math.floor(Date.now() / 1000);
    if (!t || Math.abs(now - t) > 300) return res.status(400).send("invalid timestamp");

    const raw = req.body;
    const expected = crypto.createHmac("sha256", secret).update(raw).digest("hex");

    const valid = v1.length === expected.length &&
      crypto.timingSafeEqual(Buffer.from(v1, "utf8"), Buffer.from(expected, "utf8"));
    if (!valid) return res.status(400).send("invalid signature");

    const event = JSON.parse(raw.toString("utf8"));
    res.sendStatus(200);
  }
);

app.listen(3000);

PHP

Read the body from php://input and compute HMAC on the raw bytes.

<?php
$secret = $_ENV['FORTISX_WEBHOOK_SECRET'] ?? '';
$sigHeader = $_SERVER['HTTP_X_FORTISX_SIGNATURE'] ?? '';
$eventId = $_SERVER['HTTP_X_FORTISX_EVENT_ID'] ?? '';

$parts = [];
foreach (explode(',', $sigHeader) as $pair) {
    [$k, $v] = array_map('trim', explode('=', $pair, 2));
    $parts[$k] = $v ?? '';
}
$t = isset($parts['t']) ? intval($parts['t']) : 0;
$v1 = $parts['v1'] ?? '';

if (!$t || abs(time() - $t) > 300) { http_response_code(400); echo 'invalid timestamp'; exit; }

$raw = file_get_contents('php://input');
$expected = hash_hmac('sha256', $raw, $secret);

if (!hash_equals($expected, $v1)) { http_response_code(400); echo 'invalid signature'; exit; }

$event = json_decode($raw, true);
http_response_code(200);
echo 'ok';

Python (Flask)

Get raw bytes via request.get_data() and compare with hmac.compare_digest.

import hmac, hashlib, time
from flask import Flask, request, abort

app = Flask(__name__)
SECRET = b'your-secret'

def parse_sig_header(h):
    parts = dict([p.strip().split('=', 1) for p in h.split(',') if '=' in p])
    return int(parts.get('t', '0')), parts.get('v1', '')

@app.post('/webhooks/fortisx')
def webhook():
    sig = request.headers.get('X-Fortisx-Signature', '')
    event_id = request.headers.get('X-Fortisx-Event-Id', '')
    t, v1 = parse_sig_header(sig)

    if not t or abs(int(time.time()) - t) > 300:
        abort(400, 'invalid timestamp')

    raw = request.get_data()
    expected = hmac.new(SECRET, raw, hashlib.sha256).hexdigest()

    if not hmac.compare_digest(expected, v1):
        abort(400, 'invalid signature')

    return ('ok', 200)

Go (net/http)

Read raw bytes with io.ReadAll(r.Body) and compare using subtle.ConstantTimeCompare.

package main

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

func parseSigHeader(h string) (int64, string) {
    parts := strings.Split(h, ",")
    var ts int64
    var v1 string
    for _, p := range parts {
        kv := strings.SplitN(strings.TrimSpace(p), "=", 2)
        if len(kv) != 2 { continue }
        switch kv[0] {
        case "t":
            if n, err := strconv.ParseInt(kv[1], 10, 64); err == nil { ts = n }
        case "v1":
            v1 = kv[1]
        }
    }
    return ts, v1
}

func abs64(x int64) int64 {
    if x < 0 { return -x }
    return x
}

func webhookHandler(w http.ResponseWriter, r *http.Request) {
    secret := os.Getenv("FORTISX_WEBHOOK_SECRET")
    sig := r.Header.Get("X-Fortisx-Signature")
    eventID := r.Header.Get("X-Fortisx-Event-Id")

    ts, v1 := parseSigHeader(sig)
    if ts == 0 || abs64(time.Now().Unix()-ts) > 300 {
        http.Error(w, "invalid timestamp", http.StatusBadRequest)
        return
    }

    raw, err := io.ReadAll(r.Body)
    if err != nil {
        http.Error(w, "bad request", http.StatusBadRequest)
        return
    }

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

    if len(v1) != len(expected) || subtle.ConstantTimeCompare([]byte(v1), []byte(expected)) != 1 {
        http.Error(w, "invalid signature", http.StatusBadRequest)
        return
    }

    w.WriteHeader(http.StatusOK)
}

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