REST API and webhooks

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

const SECRET = process.env.AICHAT_WEBHOOK_SECRET
const TOLERANCE = 300 // seconds

const verify = (header, rawBody) => {
  const parts = Object.fromEntries(
    String(header ?? '')
      .split(',')
      .map(part => part.trim().split('='))
  )
  const timestamp = Number(parts.t)

  if (!parts.v1 || !Number.isInteger(timestamp)) return false
  if (Math.abs(Date.now() / 1000 - timestamp) > TOLERANCE) return false

  const expected = crypto
    .createHmac('sha256', SECRET)
    .update(`${parts.t}.`)
    .update(rawBody)
    .digest()
  const received = Buffer.from(parts.v1, 'hex')

  return received.length === expected.length && crypto.timingSafeEqual(received, expected)
}

const app = express()

// Keep the raw body: the signature covers the exact bytes that were sent.
app.post('/webhooks/aichat', express.raw({ type: 'application/json' }), (req, res) => {
  if (!verify(req.get('X-Aichat-Signature'), req.body)) return res.sendStatus(400)

  const event = JSON.parse(req.body.toString('utf8'))
  // Deduplicate by event.id, queue the work and answer quickly.
  res.sendStatus(200)
})

app.listen(3000)

import hashlib
import hmac
import os
import time

from flask import Flask, abort, request

SECRET = os.environ["AICHAT_WEBHOOK_SECRET"].encode()
TOLERANCE = 300  # seconds


def verify(header: str, raw_body: bytes) -> bool:
    parts = dict(item.strip().split("=", 1) for item in header.split(",") if "=" in item)
    try:
        timestamp = int(parts["t"])
    except (KeyError, ValueError):
        return False
    if abs(time.time() - timestamp) > TOLERANCE:
        return False
    signed = parts["t"].encode() + b"." + raw_body
    expected = hmac.new(SECRET, signed, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, parts.get("v1", ""))


app = Flask(__name__)


@app.post("/webhooks/aichat")
def aichat_webhook():
    # request.get_data() is the raw body: the signature covers the exact bytes that were sent.
    if not verify(request.headers.get("X-Aichat-Signature", ""), request.get_data()):
        abort(400)
    event = request.get_json()
    # Deduplicate by event["id"], queue the work and answer quickly.
    return "", 200

<?php
$secret = getenv('AICHAT_WEBHOOK_SECRET');
$tolerance = 300; // seconds

// The raw body: the signature covers the exact bytes that were sent.
$payload = file_get_contents('php://input');
$header = $_SERVER['HTTP_X_AICHAT_SIGNATURE'] ?? '';

$parts = [];
foreach (explode(',', $header) as $item) {
    $pair = explode('=', trim($item), 2);
    if (count($pair) === 2) {
        $parts[$pair[0]] = $pair[1];
    }
}

$timestamp = isset($parts['t']) ? (int) $parts['t'] : 0;
$expected = hash_hmac('sha256', ($parts['t'] ?? '') . '.' . $payload, $secret);

if (abs(time() - $timestamp) > $tolerance || !hash_equals($expected, $parts['v1'] ?? '')) {
    http_response_code(400);
    exit;
}

$event = json_decode($payload, true);
// Deduplicate by $event['id'], queue the work and answer quickly.
http_response_code(200);