Verifying signatures

import { createServer } from 'node:http'
import { verifyWebhook, WebhookSignatureError, type WebhookEventBase } from '@isnap/sdk'

const secret = process.env.ISNAP_WEBHOOK_SECRET
if (!secret) throw new Error('ISNAP_WEBHOOK_SECRET is not set')

createServer((req, res) => {
  const chunks: Buffer[] = []
  req.on('data', (chunk: Buffer) => chunks.push(chunk))
  req.on('end', () => {
    // The raw body, exactly as received, as a string. Do not parse it before you verify it.
    const rawBody = Buffer.concat(chunks).toString('utf8')
    let event: WebhookEventBase
    try {
      event = verifyWebhook(rawBody, req.headers, secret, { tolerance: 300 })
    } catch (err) {
      if (!(err instanceof WebhookSignatureError)) throw err
      res.writeHead(400).end()
      return
    }
    // Respond first, process later.
    res.writeHead(204).end()
    setImmediate(() => handle(event))
  })
}).listen(Number(process.env.PORT ?? 3000))

function handle (event: WebhookEventBase) {
  // Process the event unless its event_id is recorded, then record its event_id.
  console.log(event.event_id, event.event_type)
}

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

const TOLERANCE_SECONDS = 300

/**
 * Verifies one delivery and returns the parsed event. Throws otherwise.
 *   rawBody  the request body exactly as received, as a Buffer or a string
 *   headers  Node's `req.headers`: header names in lowercase
 *   secret   the signing secret, or an array of secrets during a rotation
 *   now      the current time in Unix seconds
 */
export function verifyDelivery (rawBody, headers, secret, now = Math.floor(Date.now() / 1000)) {
  const id = headers['webhook-id']
  const timestamp = headers['webhook-timestamp']
  const signatures = headers['webhook-signature']
  if (!id || !timestamp || !signatures) throw new Error('missing signature header')

  if (!/^\d+$/.test(timestamp) || Math.abs(now - Number(timestamp)) > TOLERANCE_SECONDS) {
    throw new Error('timestamp outside the tolerance')
  }

  const given = signatures
    .split(' ')
    .filter((entry) => entry.startsWith('v1,'))
    .map((entry) => Buffer.from(entry.slice(3), 'base64'))

  for (const candidate of [secret].flat()) {
    // The key is the base64 decoding of the secret without its prefix, not its hexadecimal decoding.
    const key = Buffer.from(candidate.slice('whsec_'.length), 'base64')
    const expected = createHmac('sha256', key).update(`${id}.${timestamp}.`).update(rawBody).digest()
    // Constant-time comparison.
    if (given.some((signature) => signature.length === expected.length && timingSafeEqual(signature, expected))) {
      return JSON.parse(rawBody)
    }
  }
  throw new Error('signature mismatch')
}

import { createServer } from 'node:http'
import { verifyDelivery } from './verify.mjs'

const secret = process.env.ISNAP_WEBHOOK_SECRET
if (!secret) throw new Error('ISNAP_WEBHOOK_SECRET is not set')

createServer((req, res) => {
  const chunks = []
  req.on('data', (chunk) => chunks.push(chunk))
  req.on('end', () => {
    let event
    try {
      // The raw body, exactly as received. Do not parse it before you verify it.
      event = verifyDelivery(Buffer.concat(chunks), req.headers, secret)
    } catch {
      res.writeHead(400).end()
      return
    }
    // Respond first, process later.
    res.writeHead(204).end()
    setImmediate(() => handle(event))
  })
}).listen(Number(process.env.PORT ?? 3000))

function handle (event) {
  // Process the event unless its event_id is recorded, then record its event_id.
  console.log(event.event_id, event.event_type)
}

import base64
import hashlib
import hmac
import json
import time

TOLERANCE_SECONDS = 300


def verify_delivery(raw_body: bytes, headers, secret, now=None) -> dict:
    """Verifies one delivery and returns the parsed event. Raises ValueError otherwise.

    raw_body  the request body exactly as received, as bytes
    headers   a mapping of header names in lowercase to their values
    secret    the signing secret, or a list of secrets during a rotation
    now       the current time in Unix seconds
    """
    message_id = headers.get("webhook-id")
    timestamp = headers.get("webhook-timestamp")
    signatures = headers.get("webhook-signature")
    if not message_id or not timestamp or not signatures:
        raise ValueError("missing signature header")

    now = time.time() if now is None else now
    if not (timestamp.isascii() and timestamp.isdigit()):
        raise ValueError("malformed timestamp")
    if abs(now - int(timestamp)) > TOLERANCE_SECONDS:
        raise ValueError("timestamp outside the tolerance")

    signed = f"{message_id}.{timestamp}.".encode() + raw_body
    given = [entry[3:] for entry in signatures.split(" ") if entry.startswith("v1,")]

    for candidate in [secret] if isinstance(secret, str) else secret:
        # The key is the base64 decoding of the secret without its prefix, not its hexadecimal decoding.
        key = base64.b64decode(candidate.removeprefix("whsec_"))
        expected = base64.b64encode(hmac.new(key, signed, hashlib.sha256).digest()).decode()
        # Constant-time comparison.
        if any(hmac.compare_digest(expected.encode(), signature.encode()) for signature in given):
            return json.loads(raw_body)
    raise ValueError("signature mismatch")

import os
from http.server import BaseHTTPRequestHandler, HTTPServer

from verify import verify_delivery

SECRET = os.environ["ISNAP_WEBHOOK_SECRET"]


class Receiver(BaseHTTPRequestHandler):
    def do_POST(self):
        # The raw body, exactly as received. Do not parse it before you verify it.
        raw_body = self.rfile.read(int(self.headers.get("Content-Length", "0")))
        headers = {name.lower(): value for name, value in self.headers.items()}
        try:
            event = verify_delivery(raw_body, headers, SECRET)
        except ValueError:
            self.send_response(400)
            self.end_headers()
            return
        # Respond first, process later.
        self.send_response(204)
        self.end_headers()
        self.wfile.flush()
        handle(event)


def handle(event):
    # Process the event unless its event_id is recorded, then record its event_id.
    print(event["event_id"], event["event_type"])


HTTPServer(("", int(os.environ.get("PORT", "3000"))), Receiver).serve_forever()