Webhooks & events

Verify webhook signatures

Check the Standard Webhooks HMAC-SHA256 signature on every delivery, in Node.js, Python or an n8n Code node.

Light AI signs deliveries with Standard Webhooks: HMAC-SHA256 over {webhook-id}.{webhook-timestamp}.{raw body}, keyed with the base64-decoded part of your whsec_… secret, sent as webhook-signature: v1,<base64>.

Read the raw request body. Parsing and re-serializing JSON changes the bytes and breaks the signature.
Reject a webhook-timestamp more than 5 minutes from now.
Compute the HMAC and compare it, in constant time, with every v1, entry (there are two during a secret rotation).
De-duplicate on webhook-id, answer 2xx, then process.

With the Standard Webhooks libraries

// npm install standardwebhooks
import { Webhook } from 'standardwebhooks';

const wh = new Webhook(process.env.LIGHT_AI_WEBHOOK_SECRET); // "whsec_…"
const event = wh.verify(rawBody, headers); // throws when invalid

# pip install standardwebhooks
import os
from standardwebhooks.webhooks import Webhook

wh = Webhook(os.environ["LIGHT_AI_WEBHOOK_SECRET"])  # "whsec_…"
event = wh.verify(raw_body, headers)  # raises when invalid

Node.js#

verify.jsJavaScript

import crypto from 'node:crypto';

// Standard Webhooks verification. rawBody must be the exact bytes received.
export function verifyLightAiWebhook(rawBody, headers, secret) {
  const id = headers['webhook-id'];
  const timestamp = headers['webhook-timestamp'];
  const signatures = headers['webhook-signature'];
  if (!id || !timestamp || !signatures) throw new Error('missing webhook headers');
  if (Math.abs(Date.now() / 1000 - Number(timestamp)) > 300) throw new Error('timestamp outside tolerance');

  const key = Buffer.from(secret.replace(/^whsec_/, ''), 'base64');
  const expected = crypto.createHmac('sha256', key)
    .update(`${id}.${timestamp}.${rawBody}`)
    .digest('base64');

  const valid = signatures.split(' ').some((entry) => {
    const [version, signature] = entry.split(',');
    return version === 'v1' && signature?.length === expected.length
      && crypto.timingSafeEqual(Buffer.from(signature), Buffer.from(expected));
  });
  if (!valid) throw new Error('invalid signature');
  return JSON.parse(rawBody);
}

ExpressJavaScript

import express from 'express';
import { verifyLightAiWebhook } from './verify.js';

const app = express();
const seen = new Set(); // use your database in production

app.post('/light-ai', express.raw({ type: 'application/json' }), (req, res) => {
  let event;
  try {
    event = verifyLightAiWebhook(req.body.toString('utf8'), req.headers, process.env.LIGHT_AI_WEBHOOK_SECRET);
  } catch {
    return res.sendStatus(400);
  }
  if (seen.has(req.headers['webhook-id'])) return res.sendStatus(200); // duplicate delivery
  seen.add(req.headers['webhook-id']);
  res.sendStatus(200); // answer within 10 seconds, then work asynchronously
  queueMicrotask(() => handle(event));
});

Python#

verify.pyPython

import base64, hashlib, hmac, json, time

def verify_light_ai_webhook(raw_body: bytes, headers, secret: str):
    msg_id = headers["webhook-id"]
    timestamp = headers["webhook-timestamp"]
    signatures = headers["webhook-signature"]
    if abs(time.time() - int(timestamp)) > 300:
        raise ValueError("timestamp outside tolerance")

    key = base64.b64decode(secret.removeprefix("whsec_"))
    signed = f"{msg_id}.{timestamp}.".encode() + raw_body
    expected = base64.b64encode(hmac.new(key, signed, hashlib.sha256).digest()).decode()

    for entry in signatures.split(" "):
        version, _, signature = entry.partition(",")
        if version == "v1" and hmac.compare_digest(signature, expected):
            return json.loads(raw_body)
    raise ValueError("invalid signature")

FlaskPython

import os
from flask import Flask, request
from verify import verify_light_ai_webhook

app = Flask(__name__)

@app.post("/light-ai")
def light_ai_webhook():
    try:
        event = verify_light_ai_webhook(request.get_data(), request.headers, os.environ["LIGHT_AI_WEBHOOK_SECRET"])
    except (KeyError, ValueError):
        return "", 400
    # de-duplicate on request.headers["webhook-id"], then handle the event asynchronously
    return "", 200

n8n#

Two options. Simple: give the endpoint a static_header (e.g. X-N8N-Auth) and check it with the Webhook node's Header Auth credential. Strict: verify the signature in a Code node:

Code nodeJavaScript

// n8n Code node ("Run Once for All Items"), right after a Webhook node
// with Options → Raw Body turned on. n8n Cloud allows the crypto module;
// self-hosted n8n needs NODE_FUNCTION_ALLOW_BUILTIN=crypto.
const crypto = require('crypto');
const secret = 'whsec_…'; // or $env.LIGHT_AI_WEBHOOK_SECRET on self-hosted n8n

const headers = $input.first().json.headers;
const raw = (await this.helpers.getBinaryDataBuffer(0, 'data')).toString('utf8');
const id = headers['webhook-id'];
const timestamp = headers['webhook-timestamp'];
if (Math.abs(Date.now() / 1000 - Number(timestamp)) > 300) throw new Error('timestamp outside tolerance');

const key = Buffer.from(secret.replace(/^whsec_/, ''), 'base64');
const expected = crypto.createHmac('sha256', key).update(`${id}.${timestamp}.${raw}`).digest('base64');
const valid = headers['webhook-signature'].split(' ')
  .some((entry) => entry === `v1,${expected}`);
if (!valid) throw new Error('invalid Light AI signature');

return [{ json: { webhook_id: id, event: JSON.parse(raw) } }];

The n8n templates include this node.

Rotating the secret#

POST /v1/webhook_endpoints/{id}/rotate_secret returns a new secret. Deliveries are signed with both secrets for 24 hours, so deploy the new one within a day.