Verifying signatures

import express from 'express'
import { createHmac, timingSafeEqual } from 'node:crypto'

const SECRET = process.env.WEBHOOK_SECRET
const TOLERANCE = 300

function verify(rawBody, timestamp, signature) {
  if (Math.abs(Date.now() / 1000 - Number(timestamp)) > TOLERANCE) return false

  const expected =
    'v1=' +
    createHmac('sha256', SECRET).update(`${timestamp}.${rawBody}`, 'utf8').digest('hex')

  const a = Buffer.from(expected)
  const b = Buffer.from(signature ?? '')
  return a.length === b.length && timingSafeEqual(a, b)
}

const app = express()

// express.raw keeps the bytes intact — express.json would destroy them.
app.post('/webhook', express.raw({ type: '*/*' }), (req, res) => {
  const rawBody = req.body.toString('utf8')

  if (!verify(rawBody, req.header('X-MyAIChatbot-Timestamp'), req.header('X-MyAIChatbot-Signature'))) {
    return res.status(401).end()
  }

  const event = JSON.parse(rawBody)
  // handle it
  res.status(200).end()
})

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

SECRET = os.environ["WEBHOOK_SECRET"].encode()
TOLERANCE = 300

def verify(raw_body: bytes, timestamp: str, signature: str) -> bool:
    try:
        if abs(time.time() - int(timestamp)) > TOLERANCE:
            return False
    except (TypeError, ValueError):
        return False

    signed = timestamp.encode() + b"." + raw_body
    expected = "v1=" + hmac.new(SECRET, signed, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature or "")

app = Flask(__name__)

@app.post("/webhook")
def webhook():
    # request.get_data() returns the untouched bytes
    if not verify(
        request.get_data(),
        request.headers.get("X-MyAIChatbot-Timestamp"),
        request.headers.get("X-MyAIChatbot-Signature"),
    ):
        return "", 401

    event = request.get_json()
    return "", 200

<?php
$secret    = getenv('WEBHOOK_SECRET');
$tolerance = 300;

$rawBody   = file_get_contents('php://input');
$timestamp = $_SERVER['HTTP_X_MYAICHATBOT_TIMESTAMP'] ?? '';
$signature = $_SERVER['HTTP_X_MYAICHATBOT_SIGNATURE'] ?? '';

if (abs(time() - (int) $timestamp) > $tolerance) {
    http_response_code(401);
    exit;
}

$expected = 'v1=' . hash_hmac('sha256', $timestamp . '.' . $rawBody, $secret);

if (!hash_equals($expected, $signature)) {
    http_response_code(401);
    exit;
}

$event = json_decode($rawBody, true);
http_response_code(200);

package main

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

const tolerance = 300

func verify(rawBody []byte, timestamp, signature string) bool {
    ts, err := strconv.ParseInt(timestamp, 10, 64)
    if err != nil || math.Abs(float64(time.Now().Unix()-ts)) > tolerance {
        return false
    }

    mac := hmac.New(sha256.New, []byte(os.Getenv("WEBHOOK_SECRET")))
    mac.Write([]byte(timestamp + "."))
    mac.Write(rawBody)
    expected := "v1=" + hex.EncodeToString(mac.Sum(nil))

    return hmac.Equal([]byte(expected), []byte(signature))
}

func webhook(w http.ResponseWriter, r *http.Request) {
    rawBody, _ := io.ReadAll(r.Body)

    if !verify(rawBody,
        r.Header.Get("X-MyAIChatbot-Timestamp"),
        r.Header.Get("X-MyAIChatbot-Signature")) {
        w.WriteHeader(http.StatusUnauthorized)
        return
    }

    w.WriteHeader(http.StatusOK)
}

# Recompute a captured signature by hand (debugging only).
TIMESTAMP="1752721512"
RAW_BODY='{"event":"message.created"}'

printf '%s.%s' "$TIMESTAMP" "$RAW_BODY" \
  | openssl dgst -sha256 -hmac "$WEBHOOK_SECRET" -hex \
  | sed 's/^.* /v1=/'

# Compare with the X-MyAIChatbot-Signature header. Use printf, not echo:
# echo adds a newline and changes the result.