Verify signatures

// app/api/webhooks/invoice-ai/route.ts
import { Webhooks, WebhookVerificationError } from '@horizonpay/invoice-ai'

const webhooks = new Webhooks(process.env.INVOICE_AI_WEBHOOK_SECRET)

export async function POST(req: Request) {
  let event
  try {
    event = await webhooks.constructEvent(await req.text(), req.headers)
  } catch (err) {
    if (err instanceof WebhookVerificationError) return new Response('invalid signature', { status: 400 })
    throw err
  }

  if (event.type === 'invoice.paid') {
    console.log(`${event.data.object.number} was paid`)
  }
  return new Response(null, { status: 204 })
}

import express from 'express'
import { Webhooks } from '@horizonpay/invoice-ai'

const webhooks = new Webhooks(process.env.INVOICE_AI_WEBHOOK_SECRET)
const app = express()

// express.raw keeps the body as bytes. Don't use express.json() on this route.
app.post('/webhooks/invoice-ai', express.raw({ type: 'application/json' }), async (req, res) => {
  let event
  try {
    event = await webhooks.constructEvent(req.body, req.headers)
  } catch {
    return res.sendStatus(400)
  }

  if (event.type === 'invoice.paid') console.log(`${event.data.object.number} was paid`)
  res.sendStatus(204)
})

import os

from fastapi import FastAPI, HTTPException, Request, Response
from invoice_ai import Webhook, WebhookVerificationError

app = FastAPI()
webhook = Webhook(os.environ["INVOICE_AI_WEBHOOK_SECRET"])


@app.post("/webhooks/invoice-ai")
async def invoice_ai_webhook(request: Request):
    try:
        event = webhook.verify(await request.body(), request.headers)
    except WebhookVerificationError:
        raise HTTPException(status_code=400)

    if event.type == "invoice.paid":
        print(f"{event.data.object.number} was paid")
    return Response(status_code=204)

import crypto from 'node:crypto'

export function verify(headers, rawBody, secret) {
  const id = headers['webhook-id']
  const timestamp = headers['webhook-timestamp']
  const signatures = (headers['webhook-signature'] ?? '').split(' ')
  if (!id || !timestamp) return false
  if (Math.abs(Date.now() / 1000 - Number(timestamp)) > 300) return false

  // Buffer's base64 decoder also accepts older base64url secrets.
  const key = Buffer.from(secret.replace(/^whsec_/, ''), 'base64')
  const expected = Buffer.from(
    'v1,' + crypto.createHmac('sha256', key).update(`${id}.${timestamp}.${rawBody}`).digest('base64'),
  )

  return signatures.some((candidate) => {
    const given = Buffer.from(candidate)
    return given.length === expected.length && crypto.timingSafeEqual(given, expected)
  })
}

import base64, hashlib, hmac, time


def verify(headers, raw_body: bytes, secret: str) -> bool:
    msg_id = headers.get("webhook-id")
    timestamp = headers.get("webhook-timestamp")
    signatures = headers.get("webhook-signature", "").split()
    if not msg_id or not timestamp or not timestamp.isdigit():
        return False
    if abs(time.time() - int(timestamp)) > 300:
        return False

    # Map base64url to base64 so older secrets still work.
    raw = secret.removeprefix("whsec_").replace("-", "+").replace("_", "/").rstrip("=")
    key = base64.b64decode(raw + "=" * (-len(raw) % 4))
    digest = hmac.new(key, f"{msg_id}.{timestamp}.".encode() + raw_body, hashlib.sha256).digest()
    expected = "v1," + base64.b64encode(digest).decode()

    return any(hmac.compare_digest(candidate, expected) for candidate in signatures)