Receive webhooks

import { createHmac, timingSafeEqual } from 'node:crypto'
import { createServer } from 'node:http'

const SECRET = process.env.BRELIO_WEBHOOK_SECRET // whsec_…
const TOLERANCE_SECONDS = 5 * 60

/** Standard Webhooks: HMAC-SHA256 of `${id}.${timestamp}.${body}` with the base64 key after `whsec_`. */
export function verifyBrelioWebhook(secret, headers, rawBody) {
  const id = headers['webhook-id']
  const timestamp = headers['webhook-timestamp']
  const signatures = headers['webhook-signature']
  if (!id || !timestamp || !signatures) return false
  if (Math.abs(Date.now() / 1000 - Number(timestamp)) > TOLERANCE_SECONDS) return false

  const key = Buffer.from(secret.replace(/^whsec_/, ''), 'base64')
  const expected = createHmac('sha256', key).update(`${id}.${timestamp}.${rawBody}`).digest()
  // During a secret rotation Brelio signs with both secrets: `v1,<sig> v1,<sig>`.
  return signatures.split(' ').some((candidate) => {
    const [version, signature] = candidate.split(',')
    if (version !== 'v1' || !signature) return false
    const received = Buffer.from(signature, 'base64')
    return received.length === expected.length && timingSafeEqual(received, expected)
  })
}

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

createServer((req, res) => {
  let rawBody = ''
  req.on('data', (chunk) => (rawBody += chunk))
  req.on('end', () => {
    if (!verifyBrelioWebhook(SECRET, req.headers, rawBody)) {
      res.writeHead(400).end('invalid signature')
      return
    }
    const id = req.headers['webhook-id']
    if (!seen.has(id)) {
      seen.add(id)
      const event = JSON.parse(rawBody)
      console.log('received', event.type, event.id)
    }
    res.writeHead(204).end()
  })
}).listen(3000)

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

SECRET = os.environ["BRELIO_WEBHOOK_SECRET"]  # whsec_…
TOLERANCE_SECONDS = 5 * 60


def verify_brelio_webhook(secret: str, headers, raw_body: bytes) -> bool:
    """Standard Webhooks: HMAC-SHA256 of f"{id}.{timestamp}.{body}" with the base64 key after whsec_."""
    msg_id = headers.get("webhook-id")
    timestamp = headers.get("webhook-timestamp")
    signatures = headers.get("webhook-signature")
    if not msg_id or not timestamp or not signatures or not timestamp.isdigit():
        return False
    if abs(time.time() - int(timestamp)) > TOLERANCE_SECONDS:
        return False

    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()
    # During a secret rotation Brelio signs with both secrets: "v1,<sig> v1,<sig>".
    return any(
        candidate.startswith("v1,") and hmac.compare_digest(candidate[3:], expected)
        for candidate in signatures.split(" ")
    )


app = Flask(__name__)
seen: set[str] = set()  # use your database in production


@app.post("/brelio")
def brelio_webhook():
    raw_body = request.get_data()  # the exact bytes, before any JSON parsing
    if not verify_brelio_webhook(SECRET, request.headers, raw_body):
        return "invalid signature", 400
    msg_id = request.headers["webhook-id"]
    if msg_id not in seen:
        seen.add(msg_id)
        event = json.loads(raw_body)
        print("received", event["type"], event["id"])
    return "", 204

import { createServer } from 'node:http'
import { Webhook } from 'standardwebhooks'

const webhook = new Webhook(process.env.BRELIO_WEBHOOK_SECRET.replace(/^whsec_/, ''))

createServer((req, res) => {
  let rawBody = ''
  req.on('data', (chunk) => (rawBody += chunk))
  req.on('end', () => {
    try {
      const event = webhook.verify(rawBody, req.headers)
      console.log('received', event.type, event.id)
      res.writeHead(204).end()
    } catch {
      res.writeHead(400).end('invalid signature')
    }
  })
}).listen(3000)

import os
from flask import Flask, request
from standardwebhooks.webhooks import Webhook

wh = Webhook(os.environ["BRELIO_WEBHOOK_SECRET"])
app = Flask(__name__)

@app.post("/brelio")
def brelio_webhook():
    try:
        event = wh.verify(request.get_data(), dict(request.headers))
    except Exception:
        return "invalid signature", 400
    print("pylib ok", event["type"])
    return "", 204