Signature verification

import express from 'express';
import crypto from 'node:crypto';

const SECRET = process.env.KEEBAI_WEBHOOK_SECRET;
const app = express();

// Capture raw body for this route
app.post(
  '/keebai-events',
  express.raw({ type: 'application/json' }),
  (req, res) => {
    const sigHeader = req.header('X-Keebai-Signature') || '';
    const parts = Object.fromEntries(
      sigHeader.split(',').map((p) => p.split('=')),
    );
    const t = parts.t;
    const v1 = parts.v1;
    if (!t || !v1) return res.sendStatus(401);

    // Anti-replay: 5 minutes
    const skewSeconds = Math.abs(Date.now() / 1000 - Number(t));
    if (skewSeconds > 300) return res.sendStatus(401);

    const rawBody = req.body.toString('utf8');
    const expected = crypto
      .createHmac('sha256', SECRET)
      .update(`${t}.${rawBody}`)
      .digest('hex');

    const ok = crypto.timingSafeEqual(
      Buffer.from(expected, 'hex'),
      Buffer.from(v1, 'hex'),
    );
    if (!ok) return res.sendStatus(401);

    const event = JSON.parse(rawBody);
    // dedup by event.id before processing
    res.sendStatus(200);
  },
);

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

app = Flask(__name__)
SECRET = b'whsec_xxx'  # bytes

@app.post('/keebai-events')
def handler():
    sig_header = request.headers.get('X-Keebai-Signature', '')
    parts = dict(p.split('=', 1) for p in sig_header.split(','))
    t = parts.get('t')
    v1 = parts.get('v1')
    if not t or not v1:
        abort(401)

    if abs(time.time() - int(t)) > 300:
        abort(401)

    raw_body = request.get_data()
    signed = f"{t}.".encode() + raw_body
    expected = hmac.new(SECRET, signed, hashlib.sha256).hexdigest()

    if not hmac.compare_digest(expected, v1):
        abort(401)

    event = request.get_json()
    # dedup by event['id']
    return '', 200

<?php
$secret = getenv('KEEBAI_WEBHOOK_SECRET');
$rawBody = file_get_contents('php://input');
$sigHeader = $_SERVER['HTTP_X_KEEBAI_SIGNATURE'] ?? '';

$parts = [];
foreach (explode(',', $sigHeader) as $p) {
    [$k, $v] = explode('=', $p, 2);
    $parts[$k] = $v;
}
$t = $parts['t'] ?? null;
$v1 = $parts['v1'] ?? null;
if (!$t || !$v1) {
    http_response_code(401); exit;
}

if (abs(time() - (int) $t) > 300) {
    http_response_code(401); exit;
}

$expected = hash_hmac('sha256', $t . '.' . $rawBody, $secret);
if (!hash_equals($expected, $v1)) {
    http_response_code(401); exit;
}

$event = json_decode($rawBody, true);
// dedup by $event['id']
http_response_code(200);

package main

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

var secret = []byte("whsec_xxx")

func handler(w http.ResponseWriter, r *http.Request) {
	body, _ := io.ReadAll(r.Body)
	sigHeader := r.Header.Get("X-Keebai-Signature")
	parts := map[string]string{}
	for _, p := range strings.Split(sigHeader, ",") {
		kv := strings.SplitN(p, "=", 2)
		parts[kv[0]] = kv[1]
	}
	t, v1 := parts["t"], parts["v1"]
	if t == "" || v1 == "" {
		w.WriteHeader(401); return
	}
	ts, _ := strconv.ParseInt(t, 10, 64)
	if abs(time.Now().Unix()-ts) > 300 {
		w.WriteHeader(401); return
	}
	mac := hmac.New(sha256.New, secret)
	mac.Write([]byte(t + "."))
	mac.Write(body)
	expected := hex.EncodeToString(mac.Sum(nil))
	if !hmac.Equal([]byte(expected), []byte(v1)) {
		w.WriteHeader(401); return
	}
	// dedup, process...
	w.WriteHeader(200)
}

func abs(n int64) int64 { if n < 0 { return -n }; return n }