Webhooks

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

export function verify(rawBody, header, secret, toleranceMs = 5 * 60 * 1000) {
  const parts = Object.fromEntries(
    header.split(',').map((p) => {
      const i = p.indexOf('=')
      return [p.slice(0, i).trim(), p.slice(i + 1).trim()]
    }),
  )
  const t = Number(parts.t)
  if (!Number.isFinite(t) || !parts.v1) return false
  if (Math.abs(Date.now() - t) > toleranceMs) return false

  const expected = createHmac('sha256', secret).update(`${t}.${rawBody}`).digest('hex')
  const a = Buffer.from(expected)
  const b = Buffer.from(parts.v1)
  // Constant time: a normal === leaks the correct prefix byte by byte.
  return a.length === b.length && timingSafeEqual(a, b)
}

import express from 'express'

const app = express()

// express.json() would consume the stream and leave you unable to verify.
app.post('/agrihub', express.raw({ type: 'application/json' }), (req, res) => {
  const raw = req.body.toString('utf8')
  if (!verify(raw, req.get('X-Agrihub-Signature') ?? '', process.env.WEBHOOK_SECRET)) {
    return res.status(401).send('bad signature')
  }
  const event = JSON.parse(raw)

  // Acknowledge FIRST, then do the work. We time out after 10 seconds and
  // retry, so slow processing turns into duplicate deliveries.
  res.status(200).json({ received: true })
  void handle(event)
})

app.listen(3000)

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

SECRET = os.environ["WEBHOOK_SECRET"].encode()
app = Flask(__name__)

def verify(raw: bytes, header: str, tolerance_s: int = 300) -> bool:
    try:
        parts = dict(p.strip().split("=", 1) for p in header.split(","))
        t = int(parts["t"])
        v1 = parts["v1"]
    except (ValueError, KeyError):
        return False
    if abs(time.time() * 1000 - t) > tolerance_s * 1000:
        return False
    expected = hmac.new(SECRET, f"{t}.".encode() + raw, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, v1)

@app.post("/agrihub")
def receive():
    # request.get_data() is the raw body; request.json would re-serialise.
    raw = request.get_data()
    if not verify(raw, request.headers.get("X-Agrihub-Signature", "")):
        abort(401)
    event = json.loads(raw)
    print(event["type"], event["id"])
    return {"received": True}, 200